diff --git a/redflare/ui.py b/redflare/ui.py new file mode 100644 index 0000000..90ba7af --- /dev/null +++ b/redflare/ui.py @@ -0,0 +1,221 @@ +from __future__ import annotations + +import sys +import threading +from datetime import datetime +from pathlib import Path +from typing import Any + +from redflare.core.models import ModuleResult +from redflare.core.storage import deduplicate_findings + + +class LiveConsole: + COLORS = { + "start": "\033[96m", + "progress": "\033[90m", + "info": "\033[94m", + "finding": "\033[93m", + "success": "\033[92m", + "error": "\033[91m", + "skipped": "\033[95m", + } + ICONS = { + "start": "▶", + "progress": "·", + "info": "i", + "finding": "!", + "success": "✓", + "error": "✗", + "skipped": "↷", + } + + def __init__(self, stream=None): + self.stream = stream or sys.stdout + self.color = hasattr(self.stream, "isatty") and self.stream.isatty() + self.lock = threading.Lock() + + def emit(self, target: str, module: str, kind: str, message: str) -> None: + timestamp = datetime.now().strftime("%H:%M:%S") + icon = self.ICONS.get(kind, "·") + prefix = f"[{timestamp}] {icon} [{short_target(target)}] {module}" + line = f"{prefix} — {message}" + if self.color: + line = f"{self.COLORS.get(kind, '')}{line}\033[0m" + with self.lock: + print(line, file=self.stream, flush=True) + + def final_report( + self, + summary: dict, + results: list[ModuleResult], + run_directory: Path, + ) -> None: + findings = deduplicate_findings( + [finding for result in results for finding in result.findings] + ) + severity_order = {"critical": 0, "high": 1, "medium": 2, "low": 3, "info": 4} + findings.sort(key=lambda item: (severity_order.get(item.severity, 9), item.target, item.title)) + width = 88 + with self.lock: + print("\n" + "═" * width, file=self.stream) + print("REDFLARE FINAL ASSESSMENT REPORT", file=self.stream) + print("═" * width, file=self.stream) + print(f"Run: {summary['run_id']}", file=self.stream) + print(f"Evidence: {run_directory}", file=self.stream) + print( + f"Modules: {summary['completed']} completed / {summary['errors']} errors", + file=self.stream, + ) + print(f"Findings: {summary['findings']} {summary.get('by_severity', {})}", file=self.stream) + if summary.get("sensitive_exposures"): + print( + f"Sensitive exposures: {summary['sensitive_exposures']}", + file=self.stream, + ) + + print("\nMODULE EXECUTION", file=self.stream) + print("─" * width, file=self.stream) + for result in sorted(results, key=lambda item: (item.target, item.module)): + print( + f"{result.status.upper():9} {result.duration_seconds:7.2f}s " + f"{result.module:20} {short_target(result.target)}", + file=self.stream, + ) + + for result in sorted(results, key=lambda item: (item.target, item.module)): + self._module_assessment(result, width) + + print("\nCONSOLIDATED FINDINGS", file=self.stream) + print("─" * width, file=self.stream) + if not findings: + print("No findings were produced by the configured modules.", file=self.stream) + for index, finding in enumerate(findings, start=1): + print( + f"{index:>3}. [{finding.severity.upper():8}] {finding.title}", + file=self.stream, + ) + print(f" Target: {finding.target}", file=self.stream) + print(f" Module: {finding.module} | Confidence: {finding.confidence:.2f}", file=self.stream) + if finding.test_id: + print(f" Test ID: {finding.test_id}", file=self.stream) + print(f" {finding.description}", file=self.stream) + if finding.evidence: + print(" Evidence:", file=self.stream) + self._print_value(finding.evidence, indent=7) + if finding.remediation: + print(f" Remediation: {finding.remediation}", file=self.stream) + + intel = summary.get("repository_intelligence") + if intel: + print("\nREPOSITORY INTELLIGENCE", file=self.stream) + print("─" * width, file=self.stream) + print(f"Status: {intel.get('status')} | Output: {intel.get('output', 'n/a')}", file=self.stream) + + print("\nREPORT FILES", file=self.stream) + print("─" * width, file=self.stream) + for name in ( + "report.html", + "summary.json", + "findings.jsonl", + "attack_surface.json", + "test_registry.json", + "manifest.json", + ): + print(f"{run_directory / name}", file=self.stream) + print("═" * width, file=self.stream, flush=True) + + def _module_assessment(self, result: ModuleResult, width: int) -> None: + title = result.module.replace("_", " ").upper() + print(f"\n{title} ASSESSMENT — {result.target}", file=self.stream) + print("─" * width, file=self.stream) + print( + f"Status: {result.status.upper()} | Duration: {result.duration_seconds:.2f}s | " + f"Findings: {len(result.findings)}", + file=self.stream, + ) + + print("\nObservations:", file=self.stream) + if result.observations: + self._print_value(result.observations, indent=2) + else: + print(" None recorded.", file=self.stream) + + print("\nModule findings:", file=self.stream) + if not result.findings: + print(" No module findings.", file=self.stream) + for index, finding in enumerate(result.findings, start=1): + print( + f" {index}. [{finding.severity.upper()}] {finding.title} " + f"(confidence {finding.confidence:.2f})", + file=self.stream, + ) + if finding.test_id: + print(f" Test ID: {finding.test_id}", file=self.stream) + print(f" {finding.description}", file=self.stream) + if finding.evidence: + print(" Evidence:", file=self.stream) + self._print_value(finding.evidence, indent=7) + if finding.remediation: + print(f" Remediation: {finding.remediation}", file=self.stream) + + print("\nErrors:", file=self.stream) + if result.errors: + for error in result.errors: + print(f" - {error}", file=self.stream) + else: + print(" None.", file=self.stream) + + print("Artifacts:", file=self.stream) + if result.artifacts: + for artifact in result.artifacts: + print(f" - {artifact}", file=self.stream) + else: + print(" None.", file=self.stream) + + def _print_value(self, value: Any, indent: int = 0, label: str | None = None) -> None: + prefix = " " * indent + if isinstance(value, dict): + if label is not None: + print(f"{prefix}{human_label(label)}:", file=self.stream) + indent += 2 + if not value: + print(f"{' ' * indent}{{}}", file=self.stream) + for key, item in value.items(): + self._print_value(item, indent, str(key)) + return + if isinstance(value, list): + if label is not None: + print(f"{prefix}{human_label(label)} ({len(value)}):", file=self.stream) + indent += 2 + if not value: + print(f"{' ' * indent}None", file=self.stream) + for index, item in enumerate(value, start=1): + marker = f"[{index}]" + if isinstance(item, (dict, list)): + print(f"{' ' * indent}{marker}", file=self.stream) + self._print_value(item, indent + 2) + else: + print(f"{' ' * indent}{marker} {display_scalar(item)}", file=self.stream) + return + if label is None: + print(f"{prefix}{display_scalar(value)}", file=self.stream) + else: + print(f"{prefix}{human_label(label)}: {display_scalar(value)}", file=self.stream) + + +def short_target(target: str) -> str: + value = target.replace("https://", "").replace("http://", "") + return value[:42] + ("…" if len(value) > 42 else "") + + +def human_label(value: str) -> str: + return value.replace("_", " ").strip().capitalize() + + +def display_scalar(value: Any) -> str: + if value is None: + return "null" + if isinstance(value, bool): + return "yes" if value else "no" + return str(value)