306 lines
12 KiB
Python
306 lines
12 KiB
Python
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()
|