| """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 |
| |
| 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: |
| 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: |
| |
| |
| 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() |
|
|
|
|
| |
| 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() |
|
|
