osint-investigation

1#!/usr/bin/env python32"""Build a structured findings.json with evidence chains (stdlib-only).3 4Aggregates cross_links.csv (entity_resolution output) and an optional5timing.json (timing_analysis output) into a single evidence-chain document.6 7Output structure:8    {9      "metadata": {...},10      "findings": [11        {12          "id": "F0001",13          "title": "...",14          "severity": "HIGH|MEDIUM|LOW",15          "confidence": "high|medium|low",16          "summary": "...",17          "evidence": [18            {"source": "cross_links.csv", "row": 12, "fields": {...}},19            ...20          ],21          "sources": ["cross_links.csv", "timing.json"]22        }23      ]24    }25 26Every finding traces to specific source rows. No naked claims.27"""28from __future__ import annotations29 30import argparse31import csv32import json33from collections import defaultdict34from pathlib import Path35 36CONFIDENCE_ORDER = {"high": 0, "medium": 1, "low": 2}37SEVERITY_ORDER = {"HIGH": 0, "MEDIUM": 1, "LOW": 2}38 39 40def _read_cross_links(path: str) -> list[dict[str, str]]:41    with open(path, newline="", encoding="utf-8") as fh:42        return list(csv.DictReader(fh))43 44 45def build_findings(46    cross_links_path: str,47    timing_path: str | None = None,48    out_path: str = "findings.json",49    bundled_threshold: int = 3,50) -> dict:51    findings: list[dict] = []52    next_id = 153 54    # 1. Match-based findings, grouped by (left_normalized, right_normalized).55    matches = _read_cross_links(cross_links_path)56    grouped: dict[tuple[str, str], list[dict[str, str]]] = defaultdict(list)57    for i, row in enumerate(matches):58        row["__row__"] = str(i)59        grouped[(row.get("left_normalized", ""), row.get("right_normalized", ""))].append(row)60 61    for (left_norm, right_norm), rows in grouped.items():62        if not left_norm or not right_norm:63            continue64        # Use the highest-confidence match for the finding's overall confidence.65        best = min(rows, key=lambda r: CONFIDENCE_ORDER.get(r.get("confidence", "low"), 2))66        finding_id = f"F{next_id:04d}"67        next_id += 168        evidence = [69            {70                "source": "cross_links.csv",71                "row": int(r["__row__"]),72                "fields": {73                    "match_type": r.get("match_type", ""),74                    "confidence": r.get("confidence", ""),75                    "left_name": r.get("left_name", ""),76                    "right_name": r.get("right_name", ""),77                    "overlap_ratio": r.get("overlap_ratio", ""),78                    "shared_tokens": r.get("shared_tokens", ""),79                },80            }81            for r in rows82        ]83        findings.append(84            {85                "id": finding_id,86                "title": f"Entity match: {best.get('left_name', '')} ↔ {best.get('right_name', '')}",87                "severity": "MEDIUM" if best.get("confidence") == "high" else "LOW",88                "confidence": best.get("confidence", "low"),89                "summary": (90                    f"{len(rows)} cross-link record(s) tie "91                    f"'{best.get('left_name', '')}' to "92                    f"'{best.get('right_name', '')}' "93                    f"(best tier: {best.get('match_type', '')})."94                ),95                "evidence": evidence,96                "sources": ["cross_links.csv"],97            }98        )99 100    # 2. Bundled-donations findings (if cross_links carries donor↔candidate pattern).101    #    Heuristic: many distinct left names sharing the same right name.102    by_right: dict[str, set[str]] = defaultdict(set)103    by_right_rows: dict[str, list[dict[str, str]]] = defaultdict(list)104    for r in matches:105        right = r.get("right_normalized", "")106        left_raw = r.get("left_name", "").strip()107        if right and left_raw:108            by_right[right].add(left_raw)109            by_right_rows[right].append(r)110    for right_norm, lefts in by_right.items():111        if len(lefts) < bundled_threshold:112            continue113        rows = by_right_rows[right_norm]114        right_raw = rows[0].get("right_name", "")115        findings.append(116            {117                "id": f"F{next_id:04d}",118                "title": f"Bundled cross-links: {len(lefts)} distinct left entities ↔ '{right_raw}'",119                "severity": "HIGH",120                "confidence": "medium",121                "summary": (122                    f"{len(lefts)} distinct left-side entities link to "123                    f"'{right_raw}'. Pattern suggests coordinated relationship "124                    f"(e.g. bundled donations, multi-vendor employer)."125                ),126                "evidence": [127                    {128                        "source": "cross_links.csv",129                        "row": int(r.get("__row__", "0")),130                        "fields": {131                            "left_name": r.get("left_name", ""),132                            "match_type": r.get("match_type", ""),133                        },134                    }135                    for r in rows136                ],137                "sources": ["cross_links.csv"],138            }139        )140        next_id += 1141 142    # 3. Timing-based findings.143    if timing_path and Path(timing_path).exists():144        timing = json.loads(Path(timing_path).read_text(encoding="utf-8"))145        for r in timing.get("results", []):146            if not r.get("significant"):147                continue148            findings.append(149                {150                    "id": f"F{next_id:04d}",151                    "title": (152                        f"Donation timing significantly clusters near awards: "153                        f"{r['donor']} ↔ {r['recipient']}"154                    ),155                    "severity": "HIGH" if r["p_value"] < 0.01 else "MEDIUM",156                    "confidence": "medium",157                    "summary": (158                        f"Mean nearest-award distance {r['observed_mean_days']} days "159                        f"(null {r['null_mean_days']} days). p={r['p_value']}, "160                        f"effect size {r['effect_size_sd']} SD. "161                        f"{r['n_donations']} donations, {r['n_award_dates']} awards."162                    ),163                    "evidence": [164                        {165                            "source": "timing.json",166                            "row": None,167                            "fields": r,168                        }169                    ],170                    "sources": ["timing.json"],171                }172            )173            next_id += 1174 175    # Sort: severity → confidence → id.176    findings.sort(177        key=lambda f: (178            SEVERITY_ORDER.get(f["severity"], 3),179            CONFIDENCE_ORDER.get(f["confidence"], 3),180            f["id"],181        )182    )183 184    payload = {185        "metadata": {186            "n_findings": len(findings),187            "cross_links_path": cross_links_path,188            "timing_path": timing_path,189            "bundled_threshold": bundled_threshold,190        },191        "findings": findings,192    }193    Path(out_path).write_text(json.dumps(payload, indent=2), encoding="utf-8")194    return payload195 196 197def main() -> int:198    p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)199    p.add_argument("--cross-links", required=True)200    p.add_argument("--timing", help="Optional timing.json from timing_analysis.py")201    p.add_argument("--out", default="findings.json")202    p.add_argument(203        "--bundled-threshold",204        type=int,205        default=3,206        help="Minimum distinct left entities to flag as bundled (default 3)",207    )208    a = p.parse_args()209 210    payload = build_findings(211        cross_links_path=a.cross_links,212        timing_path=a.timing,213        out_path=a.out,214        bundled_threshold=a.bundled_threshold,215    )216    print(f"Wrote {payload['metadata']['n_findings']} findings to {a.out}")217    return 0218 219 220if __name__ == "__main__":221    raise SystemExit(main())222