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
attack_surface.jsontest_registry.json
Consolidated findings
| Severity | Test ID | Module | Target | Finding | Description |
{''.join(rows)}
"""
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 ""