Upload files to "redflare"
This commit is contained in:
+221
@@ -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)
|
||||
Reference in New Issue
Block a user