osint-investigation

1#!/usr/bin/env python32"""Fetch OFAC SDN list (CSV format) and normalize.3 4Public endpoint: https://www.treasury.gov/ofac/downloads/sdn.csv5Format reference: https://ofac.treasury.gov/specially-designated-nationals-and-blocked-persons-list-sdn-human-readable-lists6 7The SDN CSV uses a specific 12-column format with no header row:8    ent_num, sdn_name, sdn_type, program, title, call_sign, vess_type,9    tonnage, grt, vess_flag, vess_owner, remarks10Address and AKA records live in separate files. We fetch all three and join.11"""12from __future__ import annotations13 14import argparse15import csv16import io17import sys18from collections import defaultdict19from pathlib import Path20 21sys.path.insert(0, str(Path(__file__).parent))22from _http import get  # noqa: E40223 24SDN_URL = "https://www.treasury.gov/ofac/downloads/sdn.csv"25ADD_URL = "https://www.treasury.gov/ofac/downloads/add.csv"26ALT_URL = "https://www.treasury.gov/ofac/downloads/alt.csv"27 28SDN_COLS = [29    "ent_num", "sdn_name", "sdn_type", "program", "title",30    "call_sign", "vess_type", "tonnage", "grt", "vess_flag",31    "vess_owner", "remarks",32]33ADD_COLS = [34    "ent_num", "add_num", "address", "city_state_zip", "country", "add_remarks",35]36ALT_COLS = [37    "ent_num", "alt_num", "alt_type", "alt_name", "alt_remarks",38]39 40COLUMNS = [41    "entity_id",42    "name",43    "entity_type",44    "program_list",45    "title",46    "nationalities",47    "aka_list",48    "addresses",49    "dob",50    "pob",51    "remarks",52    "last_updated",53]54 55_TYPE_MAP = {56    "individual": "individual",57    "entity": "entity",58    "vessel": "vessel",59    "aircraft": "aircraft",60}61 62 63def _read_csv(url: str, columns: list[str]) -> list[dict[str, str]]:64    body = get(url, timeout=60).decode("latin-1", errors="replace")65    reader = csv.reader(io.StringIO(body))66    out = []67    for row in reader:68        if not row:69            continue70        # Pad/truncate to expected width.71        row = row[: len(columns)] + [""] * (len(columns) - len(row))72        out.append(dict(zip(columns, row)))73    return out74 75 76def _strip_quotes(s: str) -> str:77    s = s.strip()78    if s.startswith('"') and s.endswith('"'):79        s = s[1:-1]80    if s == "-0-":81        return ""82    return s83 84 85def fetch(86    program: str | None,87    entity_type: str | None,88    out_path: str,89) -> int:90    sdn = _read_csv(SDN_URL, SDN_COLS)91    addresses = _read_csv(ADD_URL, ADD_COLS)92    akas = _read_csv(ALT_URL, ALT_COLS)93 94    addr_by_ent: dict[str, list[str]] = defaultdict(list)95    for a in addresses:96        ent = _strip_quotes(a["ent_num"])97        parts = [98            _strip_quotes(a[c])99            for c in ("address", "city_state_zip", "country")100            if _strip_quotes(a[c])101        ]102        if parts:103            addr_by_ent[ent].append(", ".join(parts))104 105    aka_by_ent: dict[str, list[str]] = defaultdict(list)106    for k in akas:107        ent = _strip_quotes(k["ent_num"])108        name = _strip_quotes(k["alt_name"])109        if name:110            aka_by_ent[ent].append(name)111 112    rows: list[dict[str, str]] = []113    for r in sdn:114        ent_num = _strip_quotes(r["ent_num"])115        if not ent_num:116            continue117        sdn_type = _TYPE_MAP.get(_strip_quotes(r["sdn_type"]).lower(), _strip_quotes(r["sdn_type"]))118        if entity_type and sdn_type != entity_type:119            continue120        progs = _strip_quotes(r["program"])121        if program and program.upper() not in progs.upper().split(";"):122            continue123        remarks = _strip_quotes(r["remarks"])124        # DOB / POB are commonly embedded in remarks for individuals.125        dob = ""126        pob = ""127        if sdn_type == "individual" and remarks:128            for chunk in remarks.split(";"):129                ch = chunk.strip()130                if ch.upper().startswith("DOB"):131                    dob = ch.split(maxsplit=1)[1] if " " in ch else ""132                elif ch.upper().startswith("POB"):133                    pob = ch.split(maxsplit=1)[1] if " " in ch else ""134        rows.append(135            {136                "entity_id": ent_num,137                "name": _strip_quotes(r["sdn_name"]),138                "entity_type": sdn_type,139                "program_list": "; ".join(p.strip() for p in progs.split(";") if p.strip()),140                "title": _strip_quotes(r["title"]),141                "nationalities": "",  # not in this CSV; available in XML format142                "aka_list": "; ".join(aka_by_ent.get(ent_num, [])),143                "addresses": "; ".join(addr_by_ent.get(ent_num, [])),144                "dob": dob,145                "pob": pob,146                "remarks": remarks,147                "last_updated": "",148            }149        )150 151    Path(out_path).parent.mkdir(parents=True, exist_ok=True)152    with open(out_path, "w", newline="", encoding="utf-8") as fh:153        w = csv.DictWriter(fh, fieldnames=COLUMNS)154        w.writeheader()155        w.writerows(rows)156    return len(rows)157 158 159def main() -> int:160    p = argparse.ArgumentParser(description=__doc__)161    p.add_argument("--program", help="Filter to specific sanctions program (e.g. SDGT, IRAN)")162    p.add_argument(163        "--entity-type",164        choices=["individual", "entity", "vessel", "aircraft"],165        help="Filter to a specific entity type",166    )167    p.add_argument("--out", required=True)168    a = p.parse_args()169    n = fetch(program=a.program, entity_type=a.entity_type, out_path=a.out)170    print(f"Wrote {n} OFAC SDN rows to {a.out}")171    return 0172 173 174if __name__ == "__main__":175    raise SystemExit(main())176