feat: added built-in systemd security plugin

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):

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)

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)

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)