From 5f4fb52424658be16ca3e62ebab6527b13d6c763 Mon Sep 17 00:00:00 2001 From: ek0ms savi0r Date: Sun, 21 Jun 2026 17:45:32 +0000 Subject: [PATCH] Upload files to "redflare" --- redflare/__main__.py | 4 + redflare/cli.py | 181 ++++++++++++++++++++++++ redflare/interactive.py | 176 +++++++++++++++++++++++ redflare/profiles.py | 25 ++++ redflare/visualize.py | 305 ++++++++++++++++++++++++++++++++++++++++ 5 files changed, 691 insertions(+) create mode 100644 redflare/__main__.py create mode 100644 redflare/cli.py create mode 100644 redflare/interactive.py create mode 100644 redflare/profiles.py create mode 100644 redflare/visualize.py diff --git a/redflare/__main__.py b/redflare/__main__.py new file mode 100644 index 0000000..9ae637f --- /dev/null +++ b/redflare/__main__.py @@ -0,0 +1,4 @@ +from .cli import main + +if __name__ == "__main__": + main() diff --git a/redflare/cli.py b/redflare/cli.py new file mode 100644 index 0000000..9802ad9 --- /dev/null +++ b/redflare/cli.py @@ -0,0 +1,181 @@ +from __future__ import annotations + +import argparse +import json +import os +import sys +from pathlib import Path + +from . import __version__ +from .core.scope import ScopeError, ScopePolicy, normalize_target +from .core.standards import registry_document +from .core.runner import Runner +from .core.storage import RunStore, target_run_id +from .modules.repository import run_repository_intelligence +from .modules.base import ModuleContext +from .profiles import PROFILES, build_modules +from .interactive import interactive_arguments, yes_no +from .ui import LiveConsole +from .visualize import VisualServer + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(prog="redflare", description="Authorized web assessment orchestrator") + parser.add_argument("--version", action="version", version=f"REDflare {__version__}") + sub = parser.add_subparsers(dest="command", required=True) + + sub.add_parser("modules", help="List profiles and modules") + sub.add_parser("doctor", help="Check native REDflare capabilities") + sub.add_parser("tests", help="List stable test IDs and standards mappings") + visualize = sub.add_parser("visualize", help="Open the local visual investigation console for a run") + visualize.add_argument("run_directory", help="REDflare run directory or local file:/// URL") + visualize.add_argument("--port", type=int, default=8765, help="Loopback port (0 chooses an available port)") + visualize.add_argument("--no-browser", action="store_true", help="Do not open the default browser") + + intel = sub.add_parser("intel", help="Run native REDflare repository secret intelligence") + intel.add_argument("--repo", action="append", required=True, help="Authorized GitHub repository URL; repeatable") + intel.add_argument("--authorized", action="store_true", help="Acknowledge authorization for every repository") + default_runs = str(Path(os.environ.get("REDFLARE_HOME", Path.cwd())) / "runs") + intel.add_argument("--output", default=default_runs, help="Base run output directory") + + scan = sub.add_parser("scan", help="Run an authorized assessment") + scan.add_argument("targets", nargs="*", help="HTTP(S) URLs or hostnames") + scan.add_argument("--targets-file", help="Targets file, one per line") + scan.add_argument("--scope", help="JSON scope policy with allowed_hosts") + scan.add_argument("--authorized", action="store_true", help="Acknowledge explicit authorization for every target") + scan.add_argument("--allow-public", action="store_true", help="Permit authorized public targets") + scan.add_argument("--profile", choices=sorted(PROFILES), default="quick") + scan.add_argument("--output", default=default_runs, help="Base run output directory") + scan.add_argument("--wordlist", help="Path wordlist for the web profile") + scan.add_argument("--max-paths", type=int, default=100) + scan.add_argument("--max-crawl-pages", type=int, default=30) + scan.add_argument("--max-crawl-depth", type=int, default=2) + scan.add_argument("--max-scripts", type=int, default=20) + scan.add_argument("--max-schema-documents", type=int, default=8) + scan.add_argument("--max-exposure-endpoints", type=int, default=75) + scan.add_argument("--max-exposure-findings", type=int, default=100) + scan.add_argument("--max-exposure-body-bytes", type=int, default=2_000_000) + scan.add_argument("--max-cve-products", type=int, default=12, help="Maximum exact product/version fingerprints to correlate with NVD") + scan.add_argument("--max-cves-per-product", type=int, default=100, help="Maximum NVD CVEs retained per fingerprint") + scan.add_argument( + "--graphql-introspection", + action="store_true", + help="Explicitly permit bounded GraphQL schema introspection on in-scope endpoints", + ) + scan.add_argument("--rate", type=float, default=2.0, help="Path requests per second") + scan.add_argument("--timeout", type=float, default=8.0) + scan.add_argument("--workers", type=int, default=2) + scan.add_argument("--github-repo", action="append", default=[], help="Associated authorized GitHub repository URL; repeatable") + return parser + + +def main(argv: list[str] | None = None) -> int: + guided = argv is None and len(sys.argv) == 1 + if guided: + argv = interactive_arguments() + if argv is None: + return 0 + args = build_parser().parse_args(argv) + if args.command == "modules": + for name, classes in PROFILES.items(): + print(f"{name:5} " + ", ".join(module.name for module in classes)) + return 0 + if args.command == "doctor": + try: + import playwright # noqa: F401 + browser = "native-playwright" + except ImportError: + browser = "native-http-fallback" + print(json.dumps({"standalone": True, "external_tools_required": False, + "capabilities": {"browser_runtime": browser, "unauthenticated_surface": "native", + "repository_intelligence": "native"}}, indent=2)) + return 0 + if args.command == "tests": + print(json.dumps(registry_document(), indent=2)) + return 0 + if args.command == "visualize": + try: + server = VisualServer(args.run_directory, max(0, args.port)) + except (OSError, ValueError, json.JSONDecodeError) as exc: + print(f"Visualization error: {exc}", file=sys.stderr) + return 2 + server.serve(open_browser=not args.no_browser) + return 0 + if args.command == "intel": + if not args.authorized: + print("Refusing to run intelligence collection without --authorized.", file=sys.stderr) + return 2 + store = RunStore(args.output) + result = run_repository_intelligence(args.repo, store.artifacts / "repository_intelligence") + store.write_manifest( + {"run_id": store.run_id, "kind": "repository-intelligence", "repositories": args.repo} + ) + (store.root / "summary.json").write_text(json.dumps(result, indent=2), encoding="utf-8") + print(json.dumps(result, indent=2)) + return 0 if result["status"] == "completed" else 1 + if not args.authorized: + print("Refusing to scan without --authorized acknowledgement.", file=sys.stderr) + return 2 + + values = list(args.targets) + if args.targets_file: + values.extend( + line.strip() + for line in Path(args.targets_file).read_text(encoding="utf-8").splitlines() + if line.strip() and not line.lstrip().startswith("#") + ) + if not values: + print("Provide at least one target or --targets-file.", file=sys.stderr) + return 2 + + try: + policy = ScopePolicy.from_file(args.scope, allow_public=args.allow_public) + targets = [] + seen = set() + for value in values: + target = normalize_target(value) + policy.validate(target) + if target.url not in seen: + targets.append(target) + seen.add(target.url) + except (ScopeError, OSError, json.JSONDecodeError) as exc: + print(f"Scope error: {exc}", file=sys.stderr) + return 2 + + store = RunStore(args.output, target_run_id(targets)) + console = LiveConsole() + context = ModuleContext( + run_id=store.run_id, + artifact_dir=store.artifacts, + timeout=args.timeout, + rate=args.rate, + wordlist=args.wordlist, + max_paths=max(1, args.max_paths), + max_crawl_pages=max(1, args.max_crawl_pages), + max_crawl_depth=max(0, args.max_crawl_depth), + max_scripts=max(0, args.max_scripts), + max_schema_documents=max(0, args.max_schema_documents), + max_exposure_endpoints=max(1, args.max_exposure_endpoints), + max_exposure_findings=max(1, args.max_exposure_findings), + max_exposure_body_bytes=max(1_024, args.max_exposure_body_bytes), + max_cve_products=max(1, args.max_cve_products), + max_cves_per_product=max(1, args.max_cves_per_product), + graphql_introspection=args.graphql_introspection, + allow_public=args.allow_public, + reporter=console.emit, + ) + print(f"REDflare {__version__} | run={store.run_id} | profile={args.profile}") + print(f"Evidence: {store.root}") + results, summary = Runner(store, build_modules(args.profile), context, args.workers).run(targets) + if args.github_repo: + intel = run_repository_intelligence(args.github_repo, store.artifacts / "repository_intelligence") + summary["repository_intelligence"] = intel + (store.root / "summary.json").write_text(json.dumps(summary, indent=2), encoding="utf-8") + console.final_report(summary, results, store.root) + intel_ok = not args.github_repo or summary["repository_intelligence"]["status"] == "completed" + if guided and yes_no(f"Open this run in the visual console now ({store.root.name})", True): + try: + VisualServer(store.root, 8765).serve(open_browser=True) + except (OSError, ValueError, json.JSONDecodeError) as exc: + print(f"Visualization error: {exc}", file=sys.stderr) + return 0 if summary["errors"] == 0 and intel_ok else 1 diff --git a/redflare/interactive.py b/redflare/interactive.py new file mode 100644 index 0000000..fee24d9 --- /dev/null +++ b/redflare/interactive.py @@ -0,0 +1,176 @@ +from __future__ import annotations + +from pathlib import Path +import json +import os +from datetime import datetime + + +BANNER = r""" + ██████╗ ███████╗██████╗ ███████╗██╗ █████╗ ██████╗ ███████╗ + ██╔══██╗██╔════╝██╔══██╗██╔════╝██║ ██╔══██╗██╔══██╗██╔════╝ + ██████╔╝█████╗ ██║ ██║█████╗ ██║ ███████║██████╔╝█████╗ + ██╔══██╗██╔══╝ ██║ ██║██╔══╝ ██║ ██╔══██║██╔══██╗██╔══╝ + ██║ ██║███████╗██████╔╝██║ ███████╗██║ ██║██║ ██║███████╗ + ╚═╝ ╚═╝╚══════╝╚═════╝ ╚═╝ ╚══════╝╚═╝ ╚═╝╚═╝ ╚═╝╚══════╝ + Authorized Web Assessment Framework +""" + + +def prompt(text: str, default: str = "") -> str: + suffix = f" [{default}]" if default else "" + value = input(f"{text}{suffix}: ").strip() + return value or default + + +def yes_no(text: str, default: bool = False) -> bool: + marker = "Y/n" if default else "y/N" + value = input(f"{text} [{marker}]: ").strip().lower() + if not value: + return default + return value in {"y", "yes"} + + +def recent_runs(limit: int = 10) -> list[Path]: + roots = [Path(os.environ.get("REDFLARE_HOME", Path.cwd())) / "runs"] + found: dict[Path, float] = {} + for root in roots: + if not root.is_dir(): + continue + for candidate in root.iterdir(): + if candidate.is_dir() and ((candidate / "attack_surface.json").exists() or (candidate / "findings.jsonl").exists()): + resolved = candidate.resolve() + found[resolved] = max(found.get(resolved, 0), candidate.stat().st_mtime) + return [path for path, _ in sorted(found.items(), key=lambda item: item[1], reverse=True)[:limit]] + + +def run_label(run: Path) -> str: + try: + manifest = json.loads((run / "manifest.json").read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + manifest = {} + targets = manifest.get("targets") or [] + names = [str(item.get("host") or item.get("url") or "") for item in targets if isinstance(item, dict)] + target = names[0] if names else run.name + if len(names) > 1: + target += f" +{len(names) - 1}" + created = str(manifest.get("created_at") or "") + try: + when = datetime.fromisoformat(created).astimezone().strftime("%b %d %H:%M") + except ValueError: + when = datetime.fromtimestamp(run.stat().st_mtime).strftime("%b %d %H:%M") + try: + summary = json.loads((run / "summary.json").read_text(encoding="utf-8")) + findings = f" · {int(summary.get('findings', 0))} findings" + except (OSError, json.JSONDecodeError, TypeError, ValueError): + findings = "" + return f"{target} · {when}{findings}" + + +def visualize_arguments() -> list[str] | None: + runs = recent_runs() + print("\nVisual investigation console") + if runs: + print("Recent REDflare runs:") + for index, run in enumerate(runs, start=1): + print(f"{index}) {run_label(run)}") + print(f" {run}") + selection = prompt("Choose a run number or paste a run path / file URL", "1") + run = str(runs[int(selection) - 1]) if selection.isdigit() and 0 < int(selection) <= len(runs) else selection + else: + run = prompt("Run directory or file:/// URL") + if not run: + print("No run selected.") + return None + arguments = ["visualize", run, "--port", prompt("Local visualizer port", "8765")] + if not yes_no("Open the visualizer in your browser", True): + arguments.append("--no-browser") + print("\nOpening the REDflare visual investigation console...\n") + return arguments + + +def interactive_arguments() -> list[str] | None: + print(BANNER) + print("1) Full assessment pipeline (recommended)") + print("2) Web assessment (native modules)") + print("3) Quick reconnaissance") + print("4) Visualize a completed run") + print("5) Native capability health check") + print("6) Exit") + choice = prompt("Select an option", "1") + if choice == "4": + return visualize_arguments() + if choice == "5": + return ["doctor"] + if choice == "6": + return None + profile = {"1": "full", "2": "web", "3": "quick"}.get(choice) + if not profile: + print("Unknown menu choice.") + return None + + print("\nTarget input") + print("1) Enter one or more targets") + print("2) Load targets from a file") + target_mode = prompt("Select target input", "1") + arguments = ["scan"] + if target_mode == "2": + arguments.extend(["--targets-file", prompt("Path to target file")]) + else: + values = prompt("Target URL(s), comma-separated") + arguments.extend(value.strip() for value in values.split(",") if value.strip()) + + scope = prompt("Optional JSON scope file (blank to use entered targets)") + if scope: + arguments.extend(["--scope", scope]) + + print("\nAuthorization gate") + print("Only continue for systems covered by explicit written authorization.") + if not yes_no("I confirm every entered target is authorized", False): + print("Authorization was not confirmed. Scan cancelled.") + return None + arguments.append("--authorized") + if yes_no("Does this scope include public internet hosts", True): + arguments.append("--allow-public") + + if profile == "full" and not yes_no( + "Full mode includes browser interaction and focused unauthenticated service probing. Is that permitted", + False, + ): + print("Falling back to native web assessment mode.") + profile = "web" + arguments.extend(["--profile", profile]) + + output = prompt("Base output directory", str(Path(os.environ.get("REDFLARE_HOME", Path.cwd())) / "runs")) + arguments.extend(["--output", output]) + arguments.extend(["--workers", prompt("Targets to process concurrently", "1")]) + arguments.extend(["--timeout", prompt("Request timeout in seconds", "10")]) + + if profile in {"web", "full"}: + wordlist = prompt("Optional path wordlist") + if wordlist: + arguments.extend(["--wordlist", wordlist]) + arguments.extend(["--rate", prompt("Path requests per second", "1")]) + arguments.extend(["--max-paths", prompt("Maximum paths per target", "100")]) + arguments.extend(["--max-crawl-pages", prompt("Maximum pages to map per target", "30")]) + arguments.extend(["--max-crawl-depth", prompt("Maximum crawler depth", "2")]) + if yes_no("Permit bounded GraphQL schema introspection on in-scope endpoints", False): + arguments.append("--graphql-introspection") + arguments.extend(["--max-exposure-endpoints", prompt("Maximum responses to inspect for sensitive exposure", "75")]) + + if profile == "full": + repositories = prompt( + "Optional associated GitHub repositories as owner/repository or URLs (blank if none)" + ) + for repository in repositories.split(","): + value = repository.strip() + if not value: + continue + bare = value.removesuffix(".git").removeprefix("https://github.com/").removeprefix("http://github.com/") + if len([part for part in bare.split("/") if part]) != 2: + print(f"Skipping invalid repository {value!r}; expected owner/repository or a GitHub URL.") + continue + arguments.extend(["--github-repo", value]) + + print("\nConfiguration complete. Starting REDflare pipeline...\n") + return arguments diff --git a/redflare/profiles.py b/redflare/profiles.py new file mode 100644 index 0000000..8f45ec9 --- /dev/null +++ b/redflare/profiles.py @@ -0,0 +1,25 @@ +from redflare.modules import ( + ApplicationMappingModule, + HeaderModule, + PassiveReconModule, + PathDiscoveryModule, + SensitiveExposureModule, + SurfaceAnalysisModule, + CVEIntelligenceModule, + NativeBrowserRuntimeModule, + NativeNoAuthModule, +) + + +PROFILES = { + "quick": [PassiveReconModule, HeaderModule, CVEIntelligenceModule, SensitiveExposureModule], + "web": [PassiveReconModule, HeaderModule, SurfaceAnalysisModule, ApplicationMappingModule, PathDiscoveryModule, CVEIntelligenceModule, SensitiveExposureModule], + "full": [PassiveReconModule, HeaderModule, SurfaceAnalysisModule, ApplicationMappingModule, PathDiscoveryModule, NativeBrowserRuntimeModule, NativeNoAuthModule, CVEIntelligenceModule, SensitiveExposureModule], +} + + +def build_modules(profile: str): + try: + return [module() for module in PROFILES[profile]] + except KeyError as exc: + raise ValueError(f"unknown profile: {profile}") from exc diff --git a/redflare/visualize.py b/redflare/visualize.py new file mode 100644 index 0000000..65f4c52 --- /dev/null +++ b/redflare/visualize.py @@ -0,0 +1,305 @@ +from __future__ import annotations + +import hashlib +import json +import mimetypes +import threading +import webbrowser +from dataclasses import dataclass +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from importlib.resources import files +from pathlib import Path +from typing import Any +from urllib.parse import unquote, urlsplit + + +def _stable_id(kind: str, value: str) -> str: + digest = hashlib.sha256(value.encode("utf-8", errors="replace")).hexdigest()[:16] + return f"{kind}:{digest}" + + +def _read_json(path: Path, default: Any) -> Any: + if not path.exists(): + return default + try: + return json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + return default + + +def _read_jsonl(path: Path) -> list[dict[str, Any]]: + if not path.exists(): + return [] + values = [] + for line in path.read_text(encoding="utf-8").splitlines(): + try: + value = json.loads(line) + except json.JSONDecodeError: + continue + if isinstance(value, dict): + values.append(value) + return values + + +def _canonical_url(value: str) -> str: + parsed = urlsplit(value) + return f"{parsed.scheme.lower()}://{parsed.netloc.lower()}{parsed.path or '/'}" + + +def resolve_run_directory(value: str | Path) -> Path: + raw = str(value).strip().strip('"\'') + parsed = urlsplit(raw) + if parsed.scheme == "file": + if parsed.netloc not in {"", "localhost"}: + raise ValueError("remote file URLs are not supported; use a local run directory") + raw = unquote(parsed.path) + return Path(raw).expanduser().resolve() + + +def build_visual_graph(run_directory: str | Path) -> dict[str, Any]: + root = resolve_run_directory(run_directory) + if not root.is_dir(): + raise ValueError(f"run directory does not exist: {root}") + if not (root / "attack_surface.json").exists() and not (root / "findings.jsonl").exists(): + raise ValueError(f"not a REDflare run directory: {root}") + surface = _read_json(root / "attack_surface.json", {"targets": {}, "summary": {}}) + findings = _read_jsonl(root / "findings.jsonl") + module_results = [ + value + for path in sorted((root / "modules").glob("*.json")) + if isinstance((value := _read_json(path, {})), dict) and value + ] if (root / "modules").is_dir() else [] + summary = _read_json(root / "summary.json", {}) + manifest = _read_json(root / "manifest.json", {}) + + run_id = str(summary.get("run_id") or manifest.get("run_id") or root.name) + run_node_id = _stable_id("run", run_id) + nodes: dict[str, dict[str, Any]] = { + run_node_id: { + "id": run_node_id, + "label": run_id, + "type": "run", + "info": {"run_directory": str(root), "summary": summary}, + } + } + edges: dict[tuple[str, str, str], dict[str, Any]] = {} + target_ids: dict[str, str] = {} + endpoint_ids: dict[tuple[str, str], str] = {} + module_ids: dict[tuple[str, str], str] = {} + + def add_node(node: dict[str, Any]) -> str: + nodes.setdefault(node["id"], node) + return node["id"] + + def add_edge(source: str, target: str, relation: str, label: str = "") -> None: + if source not in nodes or target not in nodes: + return + key = (source, target, relation) + edges.setdefault( + key, + {"id": _stable_id("edge", "\x00".join(key)), "source": source, "target": target, "type": relation, "label": label}, + ) + + for target, data in sorted((surface.get("targets") or {}).items()): + target_id = add_node( + { + "id": _stable_id("target", target), + "label": urlsplit(target).netloc or target, + "type": "target", + "info": {"url": target}, + } + ) + target_ids[target] = target_id + add_edge(run_node_id, target_id, "contains") + + for endpoint in data.get("endpoints", []): + url = str(endpoint.get("url") or "") + if not url: + continue + endpoint_id = add_node( + { + "id": _stable_id("endpoint", target + "\x00" + url), + "label": urlsplit(url).path or "/", + "type": "endpoint", + "info": endpoint, + } + ) + endpoint_ids[(target, _canonical_url(url))] = endpoint_id + add_edge(target_id, endpoint_id, "serves") + for parameter in endpoint.get("parameters", []): + name = str(parameter.get("name") or "parameter") + location = str(parameter.get("location") or "unknown") + parameter_id = add_node( + { + "id": _stable_id("parameter", endpoint_id + "\x00" + location + "\x00" + name), + "label": name, + "type": "parameter", + "info": parameter, + } + ) + add_edge(endpoint_id, parameter_id, "accepts", location) + + for document in data.get("documents", []): + identity = json.dumps(document, sort_keys=True, default=str) + document_id = add_node( + { + "id": _stable_id("document", target + "\x00" + identity), + "label": str(document.get("title") or document.get("kind") or "schema"), + "type": "document", + "info": document, + } + ) + add_edge(target_id, document_id, "documents") + + for edge in data.get("edges", []): + source = endpoint_ids.get((target, _canonical_url(str(edge.get("source") or "")))) + destination = endpoint_ids.get((target, _canonical_url(str(edge.get("destination") or "")))) + if source and destination: + add_edge(source, destination, str(edge.get("relation") or "links")) + + for result in module_results: + target = str(result.get("target") or "") + module = str(result.get("module") or "module") + module_id = add_node( + { + "id": _stable_id("module", target + "\x00" + module), + "label": module.replace("_", " "), + "type": "module", + "info": result, + } + ) + module_ids[(target, module)] = module_id + add_edge(target_ids.get(target, run_node_id), module_id, "executed") + + standard_nodes: dict[tuple[str, str], str] = {} + for finding in findings: + target = str(finding.get("target") or "") + target_id = target_ids.get(target, run_node_id) + category = str(finding.get("category") or "finding") + node_type = "exposure" if category == "sensitive-data-exposure" else "finding" + raw_finding_id = str(finding.get("id") or hashlib.sha256(json.dumps(finding, sort_keys=True).encode()).hexdigest()[:16]) + finding_id = add_node( + { + "id": "finding:" + raw_finding_id, + "label": str(finding.get("title") or category), + "type": node_type, + "severity": str(finding.get("severity") or "info").lower(), + "info": finding, + } + ) + evidence_url = str((finding.get("evidence") or {}).get("url") or "") + endpoint_id = endpoint_ids.get((target, _canonical_url(evidence_url))) if evidence_url else None + add_edge(endpoint_id or target_id, finding_id, "exposes" if node_type == "exposure" else "has_finding") + module_id = module_ids.get((target, str(finding.get("module") or ""))) + if module_id: + add_edge(module_id, finding_id, "reported") + + for family, references in (finding.get("standards") or {}).items(): + for reference in references or []: + identifier = str(reference.get("id") or "") + if not identifier: + continue + key = (family, identifier) + standard_id = standard_nodes.get(key) + if not standard_id: + standard_id = add_node( + { + "id": _stable_id("standard", family + "\x00" + identifier), + "label": identifier, + "type": "cve" if family == "CVE" else "standard", + "info": {"family": family, **reference}, + } + ) + standard_nodes[key] = standard_id + add_edge(finding_id, standard_id, "maps_to", family) + + node_list = list(nodes.values()) + edge_list = list(edges.values()) + type_counts: dict[str, int] = {} + severity_counts: dict[str, int] = {} + for node in node_list: + type_counts[node["type"]] = type_counts.get(node["type"], 0) + 1 + if node.get("severity"): + severity = node["severity"] + severity_counts[severity] = severity_counts.get(severity, 0) + 1 + return { + "schema_version": "1.0", + "metadata": { + "run_id": run_id, + "run_directory": str(root), + "nodes": len(node_list), + "edges": len(edge_list), + "type_counts": dict(sorted(type_counts.items())), + "severity_counts": dict(sorted(severity_counts.items())), + }, + "nodes": node_list, + "edges": edge_list, + } + + +@dataclass +class VisualServer: + run_directory: str | Path + port: int = 8765 + + def __post_init__(self) -> None: + self.run_directory = resolve_run_directory(self.run_directory) + self.graph = build_visual_graph(self.run_directory) + self.assets = files("redflare.web") + handler = self._handler() + self.httpd = ThreadingHTTPServer(("127.0.0.1", self.port), handler) + self.port = int(self.httpd.server_address[1]) + + @property + def url(self) -> str: + return f"http://127.0.0.1:{self.port}/" + + def _handler(self): + graph = json.dumps(self.graph).encode("utf-8") + assets = self.assets + + class Handler(BaseHTTPRequestHandler): + def do_GET(self) -> None: + path = self.path.split("?", 1)[0] + if path == "/api/graph": + return self._send(graph, "application/json; charset=utf-8") + asset_name = "index.html" if path == "/" else path.lstrip("/") + if asset_name not in {"index.html", "app.js", "styles.css"}: + self.send_error(404) + return + resource = assets.joinpath(asset_name) + try: + data = resource.read_bytes() + except (FileNotFoundError, OSError): + self.send_error(404) + return + content_type = mimetypes.guess_type(asset_name)[0] or "application/octet-stream" + self._send(data, content_type + ("; charset=utf-8" if content_type.startswith("text/") or content_type.endswith("javascript") else "")) + + def _send(self, data: bytes, content_type: str) -> None: + self.send_response(200) + self.send_header("Content-Type", content_type) + self.send_header("Content-Length", str(len(data))) + self.send_header("Cache-Control", "no-store") + self.send_header("X-Content-Type-Options", "nosniff") + self.send_header("Content-Security-Policy", "default-src 'self'; script-src 'self'; style-src 'self' 'unsafe-inline'; img-src 'self' data:; connect-src 'self'") + self.end_headers() + self.wfile.write(data) + + def log_message(self, *_args) -> None: + return + + return Handler + + def serve(self, *, open_browser: bool = True) -> None: + print(f"REDflare visual console: {self.url}") + print(f"Run: {self.run_directory}") + print("Press Ctrl+C to stop.") + if open_browser: + threading.Timer(0.2, lambda: webbrowser.open(self.url)).start() + try: + self.httpd.serve_forever() + except KeyboardInterrupt: + pass + finally: + self.httpd.server_close()