Class ComplianceReporter

compliance_report.py:32–339

Builds CIS and PCI compliance views from data Ragnar already collects.

class ComplianceReporter:
    """Builds CIS and PCI compliance views from data Ragnar already collects."""

    def __init__(self, shared_data=None, db=None):
        self.shared_data = shared_data
        self.db = db or getattr(shared_data, "db", None)
        self.mappings = self._load_mappings()

    def _load_mappings(self) -> Dict[str, Any]:
        try:
            with open(_MAPPINGS_FILE, "r", encoding="utf-8") as handle:
                return json.load(handle)
        except Exception as exc:
            logger.error(f"Could not load compliance mappings: {exc}")
            return {"cis": {"lynis_to_cis": {}, "prefix_to_cis": {}}, "pci": {"requirements": []}}

    # ------------------------------------------------------------------
    # Data collection
    # ------------------------------------------------------------------

    def _vuln_dir(self) -> Optional[str]:
        return getattr(self.shared_data, "vulnerabilities_dir", None)

    def _collect_lynis(self, host: Optional[str] = None) -> List[Dict[str, Any]]:
        records = []
        base = self._vuln_dir()
        if not base or not os.path.isdir(base):
            return records

        for root, _dirs, files in os.walk(base):
            for filename in files:
                match = _LYNIS_DAT_RE.match(filename)
                if not match:
                    continue
                rec_host = match.group("host")
                if host and rec_host != host:
                    continue
                path = os.path.join(root, filename)
                try:
                    with open(path, "r", encoding="utf-8", errors="ignore") as handle:
                        parsed = parse_lynis_dat(handle.read()) or {}
                except Exception as exc:
                    logger.warning(f"Failed to parse Lynis dat {filename}: {exc}")
                    continue

                metadata = parsed.get("metadata", {}) if isinstance(parsed, dict) else {}
                scan_date = datetime.fromtimestamp(os.path.getmtime(path)).strftime("%Y-%m-%d %H:%M:%S")
                records.append({
                    "host": rec_host,
                    "scan_date": scan_date,
                    "hardening_index": metadata.get("hardening_index"),
                    "warnings": parsed.get("warnings", []),
                    "suggestions": parsed.get("suggestions", []),
                    "vulnerable_packages": parsed.get("vulnerable_packages", []),
                })

        records.sort(key=lambda r: r["scan_date"], reverse=True)
        seen = set()