hydrogen/modules/systemd/systemd_security.py

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)