diff --git a/core/base.py b/core/base.py index e6c724a..1c718c3 100644 --- a/core/base.py +++ b/core/base.py @@ -59,7 +59,7 @@ class BaseTransport(ABC): class TypedTransport(BaseTransport, Generic[TReportTransport], ABC): - config_model: type[TReportTransport] + config_model: type[TReportTransport] # type: ignore def publish(self, rendered_report: RenderedReport, transport: ResolvedPluginConfig) -> str: if not isinstance(transport.config, self.config_model): diff --git a/modules/systemd/__init__.py b/modules/systemd/__init__.py new file mode 100644 index 0000000..aad3be9 --- /dev/null +++ b/modules/systemd/__init__.py @@ -0,0 +1,20 @@ +from core.base import BaseWorker +from core.schemas import ModuleManifest + +from .config import SystemdModuleConfig +from .systemd_security import SystemdWorker + +MANIFEST = ModuleManifest( + identifier="systemd", + name="Systemd Security Audit", + category="systemd", + version="0.1.0", + api_version="1", + description="Audits systemd service and timer configuration.", +) + +CONFIG_MODEL = SystemdModuleConfig + + +def build_worker(config: SystemdModuleConfig) -> BaseWorker: + return SystemdWorker(config) diff --git a/modules/systemd/config.py b/modules/systemd/config.py new file mode 100644 index 0000000..8c937a7 --- /dev/null +++ b/modules/systemd/config.py @@ -0,0 +1,8 @@ +from pydantic import BaseModel, Field + + +class SystemdModuleConfig(BaseModel): + enabled: bool = Field(True) + check_masked_services: bool = Field(True) + check_failed_services: bool = Field(True) + check_timers: bool = Field(True) diff --git a/modules/systemd/systemd_security.py b/modules/systemd/systemd_security.py new file mode 100644 index 0000000..a4efc2b --- /dev/null +++ b/modules/systemd/systemd_security.py @@ -0,0 +1,178 @@ +import os +import subprocess +from pathlib import Path + +from core.base import BaseWorker +from core.schemas.results import AuditFindings, AuditResults +from core.schemas.status import AuditSeverity, AuditStatus + +from .config import SystemdModuleConfig + +SERVICE_DIRS = [ + Path("/etc/systemd/system"), + Path("/usr/lib/systemd/system"), + Path("/run/systemd/system"), +] + + +class SystemdWorker(BaseWorker): + config_model = SystemdModuleConfig + + def __init__(self, config: SystemdModuleConfig | None = None) -> None: + super().__init__(config) + self.module_config = config or SystemdModuleConfig() + + def run(self) -> AuditResults: + if not self.module_config.enabled: + return AuditResults( + status=AuditStatus.SKIPPED, + findings=[], + risk_level=0.0, + ) + + if os.name != "posix": + return AuditResults( + status=AuditStatus.SKIPPED, + findings=[], + risk_level=0.0, + ) + + findings: list[AuditFindings] = [] + + if self.module_config.check_failed_services: + findings.extend(self._check_failed_services()) + + if self.module_config.check_masked_services: + findings.extend(self._check_masked_services()) + + if self.module_config.check_timers: + findings.extend(self._check_timers()) + + findings.extend(self._check_service_permissions()) + + return AuditResults( + status=AuditStatus.PASS if not findings else AuditStatus.FAIL, + findings=findings, + risk_level=self._calculate_risk(findings), + ) + + def _run_systemctl(self, *args: str) -> list[str]: + try: + result = subprocess.run( + ["systemctl", *args], + capture_output=True, + text=True, + timeout=30, + ) + return result.stdout.splitlines() + except (FileNotFoundError, subprocess.TimeoutExpired): + return [] + + def _check_failed_services(self) -> list[AuditFindings]: + findings: list[AuditFindings] = [] + output = self._run_systemctl("--failed", "--no-legend", "--no-pager") + + for line in output: + parts = line.strip().split() + if len(parts) >= 3 and parts[1] == "failed": + findings.append( + AuditFindings( + name=f"failed_service_{parts[0]}", + description=f"Systemd service {parts[0]} is in failed state.", + severity=AuditSeverity.MEDIUM, + ) + ) + + return findings + + def _check_masked_services(self) -> list[AuditFindings]: + findings: list[AuditFindings] = [] + output = self._run_systemctl("list-unit-files", "--no-legend", "--no-pager") + + for line in output: + parts = line.strip().split() + if len(parts) >= 2 and parts[1] == "masked": + findings.append( + AuditFindings( + name=f"masked_service_{parts[0]}", + description=f"Service {parts[0]} is masked and cannot be started.", + severity=AuditSeverity.LOW, + ) + ) + + return findings + + def _check_timers(self) -> list[AuditFindings]: + findings: list[AuditFindings] = [] + output = self._run_systemctl("list-timers", "--no-legend", "--no-pager") + + for line in output: + if "n/a" in line and "n/a" in line.split()[:3]: + parts = line.strip().split() + if parts: + findings.append( + AuditFindings( + name=f"missed_timer_{parts[-1]}", + description=f"Timer {parts[-1]} has missed its scheduled trigger.", + severity=AuditSeverity.MEDIUM, + ) + ) + + return findings + + def _check_service_permissions(self) -> list[AuditFindings]: + findings: list[AuditFindings] = [] + + for service_dir in SERVICE_DIRS: + if not service_dir.exists(): + continue + + if not os.access(str(service_dir), os.R_OK | os.X_OK): + findings.append( + AuditFindings( + name=f"restricted_service_dir_{service_dir.name}", + description=f"Service directory {service_dir} is not accessible.", + severity=AuditSeverity.HIGH, + ) + ) + continue + + for unit in sorted(service_dir.glob("*.service")): + self._check_unit_permissions(unit, findings) + + return findings + + def _check_unit_permissions( + self, unit: Path, findings: list[AuditFindings] + ) -> None: + stat = unit.stat() + if stat.st_uid != 0: + findings.append( + AuditFindings( + name=f"{unit.stem}_not_owned_by_root", + description=f"Unit file {unit} should be owned by root.", + severity=AuditSeverity.HIGH, + ) + ) + + if stat.st_mode & 0o022: + findings.append( + AuditFindings( + name=f"{unit.stem}_world_writable", + description=f"Unit file {unit} has overly permissive permissions.", + severity=AuditSeverity.CRITICAL, + ) + ) + + def _calculate_risk(self, findings: list[AuditFindings]) -> float: + if not findings: + return 0.0 + + severity_scores = { + AuditSeverity.CRITICAL: 0.45, + AuditSeverity.HIGH: 0.3, + AuditSeverity.MEDIUM: 0.2, + AuditSeverity.LOW: 0.1, + } + risk = sum(severity_scores[finding.severity] for finding in findings) + return min(1.0, risk)