from __future__ import annotations import html import json import re from dataclasses import replace from datetime import datetime from pathlib import Path from typing import Iterable from .models import Finding, ModuleResult def target_run_id(targets: Iterable[object]) -> str: values = list(targets) first = values[0] if values else None authority = str(getattr(first, "authority", None) or getattr(first, "host", None) or "assessment") slug = re.sub(r"[^A-Za-z0-9._-]+", "_", authority).strip("._-") or "assessment" extra = f"_plus_{len(values) - 1}" if len(values) > 1 else "" stamp = datetime.now().strftime("%Y%m%d_%H%M%S") return f"scan_{slug}{extra}_{stamp}" class RunStore: def __init__(self, base: str = "runs", run_id: str | None = None): stamp = datetime.now().strftime("%Y%m%d_%H%M%S") self.run_id = run_id or f"run_{stamp}" root = Path(base) candidate = root / self.run_id sequence = 1 while candidate.exists(): candidate = root / f"{self.run_id}_{sequence:02d}" sequence += 1 self.root = candidate self.modules = self.root / "modules" self.artifacts = self.root / "artifacts" self.modules.mkdir(parents=True) self.artifacts.mkdir(parents=True) self.findings_file = self.root / "findings.jsonl" self.findings_file.touch() def write_manifest(self, manifest: dict) -> None: self._write_json(self.root / "manifest.json", manifest) def write_result(self, result: ModuleResult) -> None: safe_target = result.target.replace("://", "_").replace("/", "_").replace(":", "_") self._write_json(self.modules / f"{safe_target}__{result.module}.json", result.to_dict()) with self.findings_file.open("a", encoding="utf-8") as handle: for finding in result.findings: handle.write(json.dumps(finding.to_dict(), sort_keys=True) + "\n") def write_surface_graph(self, graph: dict) -> None: self._write_json(self.root / "attack_surface.json", graph) def write_test_registry(self, registry: dict) -> None: self._write_json(self.root / "test_registry.json", registry) def finalize(self, results: Iterable[ModuleResult], surface_graph: dict | None = None) -> dict: results = list(results) findings = deduplicate_findings( [finding for result in results for finding in result.findings] ) with self.findings_file.open("w", encoding="utf-8") as handle: for finding in findings: handle.write(json.dumps(finding.to_dict(), sort_keys=True) + "\n") summary = { "run_id": self.run_id, "module_results": len(results), "completed": sum(result.status == "completed" for result in results), "errors": sum(result.status == "error" for result in results), "findings": len(findings), "by_severity": count_by(finding.severity for finding in findings), "by_module": count_by(finding.module for finding in findings), "by_category": count_by(finding.category for finding in findings), "sensitive_exposures": sum( finding.category == "sensitive-data-exposure" for finding in findings ), "known_cves": sum( finding.category == "known-vulnerable-component" for finding in findings ), "attack_surface": (surface_graph or {}).get("summary", {}), } self._write_json(self.root / "summary.json", summary) (self.root / "report.html").write_text( render_html(summary, findings, results), encoding="utf-8" ) return summary @staticmethod def _write_json(path: Path, value: dict) -> None: path.write_text(json.dumps(value, indent=2, sort_keys=True), encoding="utf-8") def count_by(values: Iterable[str]) -> dict[str, int]: counts: dict[str, int] = {} for value in values: counts[value] = counts.get(value, 0) + 1 return dict(sorted(counts.items())) def deduplicate_findings(findings: list[Finding]) -> list[Finding]: canonical_categories = { "browser-security-headers": "security-headers", } selected: dict[tuple[str, str], Finding] = {} sources: dict[tuple[str, str], set[str]] = {} for finding in findings: category = canonical_categories.get(finding.category, finding.category) key = ( finding.target, category if category == "security-headers" else finding.id, ) sources.setdefault(key, set()).add(finding.module) current = selected.get(key) if current is None or finding.confidence > current.confidence: selected[key] = replace( finding, category=category, evidence=dict(finding.evidence), ) for key, finding in selected.items(): modules = sorted(sources[key]) if len(modules) > 1: finding.evidence["corroborated_by"] = modules finding.tags = sorted(set(finding.tags + ["corroborated"])) return list(selected.values()) def render_html( summary: dict, findings: list[Finding], results: list[ModuleResult] | None = None, ) -> str: rows = [] for finding in sorted(findings, key=lambda item: (item.severity, item.target), reverse=True): rows.append( "" f"{html.escape(finding.severity)}" f"{html.escape(finding.test_id)}" f"{html.escape(finding.module)}" f"{html.escape(finding.target)}" f"{html.escape(finding.title)}" f"{html.escape(finding.description)}" "" ) assessments = [] for result in sorted(results or [], key=lambda item: (item.target, item.module)): module_findings = "".join( "
" f"

[{html.escape(finding.severity.upper())}] {html.escape(finding.title)}

" f"

Test ID: {html.escape(finding.test_id or 'unmapped')}

" f"

{html.escape(finding.description)}

" f"{render_reference_links(finding)}" f"
{html.escape(json.dumps(finding.evidence, indent=2, default=str))}
" f"

Remediation: {html.escape(finding.remediation or 'Review and remediate based on verified impact.')}

" "
" for finding in result.findings ) or "

No module findings.

" errors = "".join(f"
  • {html.escape(error)}
  • " for error in result.errors) or "
  • None
  • " artifacts = "".join( f"
  • {html.escape(artifact)}
  • " for artifact in result.artifacts ) or "
  • None
  • " assessments.append( "
    " f"

    {html.escape(result.module.replace('_', ' ').upper())} ASSESSMENT

    " f"

    Target: {html.escape(result.target)}
    " f"Status: {html.escape(result.status.upper())}   " f"Duration: {result.duration_seconds:.2f}s   " f"Findings: {len(result.findings)}

    " "

    Complete observations

    " f"
    {html.escape(json.dumps(result.observations, indent=2, default=str))}
    " f"

    Module findings

    {module_findings}" f"

    Errors

    " f"

    Artifacts

    " "
    " ) return f"""REDflare {html.escape(summary['run_id'])}

    REDflare Final Assessment Report

    Run summary

    {html.escape(json.dumps(summary, indent=2))}
    {''.join(assessments)}

    Assessment artifacts

    Consolidated findings

    {''.join(rows)}
    SeverityTest IDModuleTargetFindingDescription
    """ def render_reference_links(finding: Finding) -> str: links = [] seen = set() for family, references in finding.standards.items(): for reference in references: url = str(reference.get("url") or "") if not url or url in seen: continue seen.add(url) label = str(reference.get("id") or family) links.append( f'' f'{html.escape(label)}' ) return f"

    References: {' · '.join(links)}

    " if links else ""