"""A one-page PDF case report for a single entity. ReportLab, drawing the subgraph straight onto the canvas rather than rendering an image with a plotting library — one fewer dependency, and vector output that stays readable when the page is printed for a file. The report is a summary of what the system found, not a conclusion. Every number on it is a lead; the caveat at the bottom says so, and stays on the page. """ from __future__ import annotations from datetime import datetime from io import BytesIO from pathlib import Path import networkx as nx from reportlab.lib.colors import HexColor from reportlab.lib.pagesizes import A4 from reportlab.lib.units import mm from reportlab.lib.utils import simpleSplit from reportlab.pdfgen import canvas as pdf_canvas from .format_id import format_id MARGIN = 20 * mm WIDTH, HEIGHT = A4 INK = HexColor("#111318") MUTED = HexColor("#5b6472") RULE = HexColor("#c8cdd6") NODE_COLOURS = {"wallet": HexColor("#2f6f4f"), "transaction": HexColor("#3a5ba0"), "ip": HexColor("#8a5a2b")} FOCUS = HexColor("#b3261e") CAVEAT = ("Generated by btc-intel from synthetic/offline data. Risk scores rank leads " "for a human analyst; they are not evidence of wrongdoing and no entity here " "has been attributed to a person. An IP candidate marked an anonymized entry " "point is where a broadcast entered the network, not who sent it.") def _wrap(c, text: str, x: float, y: float, width: float, size: float = 9.5, leading: float = 12.5, colour=INK, font: str = "Helvetica") -> float: c.setFont(font, size) c.setFillColor(colour) for line in simpleSplit(str(text), font, size, width): c.drawString(x, y, line) y -= leading return y def _heading(c, text: str, y: float) -> float: c.setFont("Helvetica-Bold", 10) c.setFillColor(MUTED) c.drawString(MARGIN, y, text.upper()) c.setStrokeColor(RULE) c.line(MARGIN, y - 4, WIDTH - MARGIN, y - 4) return y - 16 def _layout(nodes: list[dict], edges: list[dict], box: tuple[float, float, float, float], given: dict[str, dict] | None = None) -> dict[str, tuple[float, float]]: """Positions from networkx's spring layout, rescaled into the box. `given` short-circuits it: when the analyst saved an investigation, the report draws the arrangement they built rather than a fresh one, which is the difference between a report of the case and a picture of the data. """ x0, y0, w, h = box if given: points = {n["data"]["id"]: given[n["data"]["id"]] for n in nodes if n["data"]["id"] in given} if points: xs = [p["x"] for p in points.values()] ys = [p["y"] for p in points.values()] span_x = (max(xs) - min(xs)) or 1.0 span_y = (max(ys) - min(ys)) or 1.0 # PDF y grows upward, the screen's grows down — flip it. return {node: (x0 + (p["x"] - min(xs)) / span_x * w, y0 + h - (p["y"] - min(ys)) / span_y * h) for node, p in points.items()} g = nx.Graph() g.add_nodes_from(n["data"]["id"] for n in nodes) g.add_edges_from((e["data"]["source"], e["data"]["target"]) for e in edges) if not g: return {} pos = nx.spring_layout(g, seed=0) xs = [p[0] for p in pos.values()] ys = [p[1] for p in pos.values()] span_x = (max(xs) - min(xs)) or 1.0 span_y = (max(ys) - min(ys)) or 1.0 return {node: (x0 + (p[0] - min(xs)) / span_x * w, y0 + (p[1] - min(ys)) / span_y * h) for node, p in pos.items()} def _draw_graph(c, graph: dict, y_top: float, height: float) -> None: nodes = graph["elements"]["nodes"][:60] positions = graph.get("positions") keep = {n["data"]["id"] for n in nodes} edges = [e for e in graph["elements"]["edges"] if e["data"]["source"] in keep and e["data"]["target"] in keep] box = (MARGIN + 6, y_top - height + 6, WIDTH - 2 * MARGIN - 12, height - 12) c.setStrokeColor(RULE) c.rect(MARGIN, y_top - height, WIDTH - 2 * MARGIN, height) pos = _layout(nodes, edges, box, positions) c.setLineWidth(0.4) c.setStrokeColor(RULE) for edge in edges: a, b = pos.get(edge["data"]["source"]), pos.get(edge["data"]["target"]) if a and b: c.line(a[0], a[1], b[0], b[1]) for node in nodes: data = node["data"] placed = pos.get(data["id"]) if placed is None: # a saved view can hold a node the layout skipped continue x, y = placed focus = data.get("is_focus") c.setFillColor(FOCUS if focus else NODE_COLOURS.get(data["type"], MUTED)) radius = 3.6 if focus else 2.4 c.circle(x, y, radius, stroke=0, fill=1) c.setFont("Helvetica", 7.5) c.setFillColor(MUTED) legend = ((f"{graph.get('source_label')} · " if graph.get("source_label") else "") + f"green wallets {graph['counts']['wallets']} · " f"blue transactions {graph['counts']['transactions']} · " f"brown IPs {graph['counts']['ips']} · red = this entity · " f"within {graph['hops']} hops" + (" (truncated)" if graph.get("truncated") else "")) c.drawString(MARGIN + 4, y_top - height + 4, legend) def _seal_lines(custody: dict) -> list[str]: """The seal, as short fixed-width lines a person can read out or retype. Hashes are printed in full: half a hash verifies nothing, and this is the one place on the page where truncating an identifier would be a mistake. """ lines = [f"ledger head {custody.get('head', '—')}"] for sealed in custody.get("files", [])[:4]: name = Path(sealed["path"]).name lines.append(f"{name:<24} {sealed['sha256']}") lines.append("verify with: python -m custody verify") return lines def render_pdf(detail: dict, graph: dict, now: datetime, custody: dict | None = None) -> bytes: """One page: who, how risky, why, what the evidence is, and the neighbourhood. `custody` is the evidence seal — the hashes of the data this report was produced from and the head of the custody ledger at the moment it was produced. Printed on the page so the paper can be checked against the system later; the report's own hash goes into the ledger, which is the other half of the same check. """ buffer = BytesIO() c = pdf_canvas.Canvas(buffer, pagesize=A4) c.setTitle(f"btc-intel case report — {detail['entity_id']}") width = WIDTH - 2 * MARGIN y = HEIGHT - MARGIN c.setFont("Helvetica-Bold", 16) c.setFillColor(INK) c.drawString(MARGIN, y, "btc-intel case report") c.setFont("Helvetica", 9) c.setFillColor(MUTED) c.drawRightString(WIDTH - MARGIN, y, now.strftime("%Y-%m-%d %H:%M:%S UTC")) y -= 22 score = detail["scores"].get("risk_score") c.setFont("Helvetica-Bold", 11) c.setFillColor(INK) c.drawString(MARGIN, y, f"Entity {format_id(detail['entity_id'], head=20, tail=8)}") c.drawRightString(WIDTH - MARGIN, y, f"risk score {score:.3f}" if isinstance(score, (int, float)) else "not alerted") y -= 14 c.setFont("Helvetica", 9) c.setFillColor(MUTED) queue = detail.get("queue_position") or {} rank = (f" · queue rank {queue['rank']} of {queue['of']}" if queue.get("rank") else "") c.drawString(MARGIN, y, f"{detail['entity_type']} · {len(detail['wallets'])} wallet(s)" + (f" · top signal {detail['scores']['top_signal']}" if detail["scores"].get("top_signal") else "") + (f" · {', '.join(detail['pattern_types'])}" if detail.get("pattern_types") else "") + rank) y -= 22 y = _heading(c, "reason", y) y = _wrap(c, detail.get("reason") or "No alert was raised for this entity.", MARGIN, y, width) - 10 y = _heading(c, "evidence", y) evidence = list(detail.get("evidence") or [])[:10] if not evidence: y = _wrap(c, "—", MARGIN, y, width, colour=MUTED) for item in evidence: # Shortened exactly as the console shortened it, so a reader can match # a line in the report to a row on the screen. y = _wrap(c, f"• {format_id(item, head=18, tail=8)}", MARGIN, y, width, size=8.5, leading=11) y -= 10 if detail.get("taint_path"): y = _heading(c, "taint path", y) y = _wrap(c, " → ".join(format_id(step) for step in detail["taint_path"]), MARGIN, y, width, size=8.5) - 10 leads = detail.get("leads") or [] if leads: y = _heading(c, "attribution leads", y) for lead in leads[:3]: y = _wrap(c, f"• {lead['ip']} ({lead.get('label', lead.get('ip_class'))}, " f"confidence {lead.get('confidence')})", MARGIN, y, width, size=8.5, leading=11) y -= 10 graph_height = max(y - MARGIN - 52, 90) y = _heading(c, "neighbourhood", y) _draw_graph(c, graph, y + 2, min(graph_height, y - MARGIN - 46)) y = y + 2 - min(graph_height, y - MARGIN - 46) - 14 if custody: y = max(y, MARGIN + 46) c.setFont("Helvetica-Bold", 7.5) c.setFillColor(MUTED) c.drawString(MARGIN, y, "EVIDENCE SEAL") c.setFont("Courier", 6.5) for line in _seal_lines(custody): y -= 8 c.drawString(MARGIN, y, line) y -= 10 _wrap(c, CAVEAT, MARGIN, max(y, MARGIN + 24), width, size=7.5, leading=9.5, colour=MUTED, font="Helvetica-Oblique") c.showPage() c.save() return buffer.getvalue() # --- exit-point investigator packet (analysis/exit_point.py) ------------------- SEALED_NOT_SIGNED = ("The tag bundles behind these candidates are sealed (tamper-evident): " "their files are unchanged since sealing. They are not signed, so the " "seal does not prove who built them.") def packet_disclaimers(result: dict) -> list[str]: """What the first page states before anything else. Kept here so the PDF and its tests read the same sentences.""" lines = [result["statement"]] lines.append("Simulated tags are simulated: at least one tag bundle behind this packet is " "derived from generator ground truth and is not intelligence." if result.get("simulated_tags") else "No simulated tag bundle is loaded; every tag names its real source.") lines.append(SEALED_NOT_SIGNED) return lines def render_packet(result: dict, now: datetime, custody: dict | None = None) -> bytes: """The investigator packet for one exit-point trace: a disclaimer page, then every candidate with its tag, receiving addresses, time window, traced amounts and hop-by-hop paths, then the untagged sinks and the seal.""" buffer = BytesIO() c = pdf_canvas.Canvas(buffer, pagesize=A4) c.setTitle(f"btc-intel exit-point packet — {result['subject']}") width = WIDTH - 2 * MARGIN state = {"y": HEIGHT - MARGIN} def room(needed: float) -> None: if state["y"] - needed < MARGIN: c.showPage() state["y"] = HEIGHT - MARGIN def line(text: str, size: float = 8.5, colour=INK, font: str = "Helvetica") -> None: room(12) state["y"] = _wrap(c, text, MARGIN, state["y"], width, size=size, leading=size + 2.5, colour=colour, font=font) def heading(text: str) -> None: room(40) state["y"] = _heading(c, text, state["y"]) c.setFont("Helvetica-Bold", 16) c.setFillColor(INK) c.drawString(MARGIN, state["y"], "btc-intel exit-point packet") c.setFont("Helvetica", 9) c.setFillColor(MUTED) c.drawRightString(WIDTH - MARGIN, state["y"], now.strftime("%Y-%m-%d %H:%M:%S UTC")) state["y"] -= 22 line(f"Seed: {result['kind']} {result['subject']}", size=10, font="Helvetica-Bold") if result.get("note"): line(result["note"], colour=MUTED) state["y"] -= 6 heading("read this first") for text in packet_disclaimers(result): line(f"• {text}", size=9.5, font="Helvetica-Bold") cfg = result.get("config", {}) line(f"Taint models: haircut (proportional) and poison (any contact). Depth limit " f"{cfg.get('max_hops')} hops; an entity receiving under {cfg.get('min_share', 0):.1%} of " f"the seed's outflow (haircut) or {cfg.get('min_value_btc')} BTC (poison) is not " "followed further.", colour=MUTED) for trace in result["traces"]: c.showPage() state["y"] = HEIGHT - MARGIN heading(f"trace from cluster {trace['seed']}") line(f"Seed outflow {trace['seed_amount']:.8f} BTC. Still in flight at the depth " f"limit: haircut {trace['at_depth_limit']['haircut']:.8f}, " f"poison {trace['at_depth_limit']['poison']:.8f} BTC.") for model, mixes in trace["mixes"].items(): for mix in mixes: line(f"{model}: funds entered a CoinJoin from {mix['from']} " f"({mix['value']:.8f} BTC, tx {mix['txids'][0]}). Not traced past the mix.", colour=MUTED) if not trace["candidates"]: line("No cluster tagged exchange/VASP was reached.", colour=MUTED) for cand in trace["candidates"]: heading(f"candidate {cand['rank']}: cluster {cand['entity_id']}") for tag in cand["tags"]: line(f"Tag: {tag['label']} ({tag['category']}), source {tag['source']}, " f"collected {tag['collected']}, bundle {tag['bundle']}" + (" — SIMULATED, not intelligence" if tag["simulated"] else ""), font="Helvetica-Bold") if cand["conflict"]: line("Conflicting tags on this cluster; all are listed, none preferred.") line("Receiving addresses: " + (", ".join(cand["receiving_addresses"]) or "—")) line(f"Time window: {cand['time_window'][0]} to {cand['time_window'][1]}") for model, m in cand["models"].items(): line(f"{model}: {m['amount']:.8f} BTC traced ({m['share']:.2%} of the seed's " f"outflow), best path confidence {m['path_confidence']:.4f}") for model, m in cand["models"].items(): for i, path in enumerate(m["paths"], 1): line(f"{model} path {i} (confidence {path['confidence']:.4f}):", font="Helvetica-Bold") for n, hop in enumerate(path["hops"], 1): line(f" hop {n}: {hop['reason']}. tx {', '.join(hop['txids'][:3])}" + (" …" if len(hop["txids"]) > 3 else ""), size=7.5) line(f"Evidence sha256 (this candidate's record): {cand['evidence_sha256']}", size=7, font="Courier") if trace["sinks"]: heading("untagged sinks (clusters that kept traced value; not services)") for sink in trace["sinks"]: line(f"• cluster {sink['entity_id']}: {sink['label']}, kept " f"{sink['share_kept']:.2%} under haircut") heading("evidence seal") for bundle in result.get("tag_bundles", []): line(f"tag bundle {bundle['name']}: manifest {bundle.get('manifest_hash')}" + (" (simulated)" if bundle.get("simulated") else ""), size=7, font="Courier") for text in _seal_lines(custody or {}): line(text, size=7, font="Courier") state["y"] -= 6 line(CAVEAT, size=7.5, colour=MUTED, font="Helvetica-Oblique") c.showPage() c.save() return buffer.getvalue()