Repo hardening: fix install instructions, dead imports, hygiene

README.md

@@ -16,10 +16,16 @@
 </div>
 
 ```bash
-pip install cognis-cryptotrace
+pip install "git+https://github.com/cognis-digital/cryptotrace.git"
 cryptotrace scan .            # → prioritized findings in seconds
 ```
 
+<!-- cognis:layman:start -->
+## What is this?
+
+CryptoTrace is a free command-line tool that checks cryptocurrency wallet addresses against the US government's official sanctions list (OFAC), so you can see at a glance whether any Bitcoin or Ethereum address in a transaction set belongs to a sanctioned group like Lazarus Group, Tornado Cash, or a darknet exchange. It also groups addresses that appear to be controlled by the same person or organization based on how funds move between them. You give it a list of transactions in JSON format, and it produces a prioritized report of risks — direct hits, addresses one or two steps removed from a sanctioned wallet, and suspicious wallet clusters — either as a readable table or machine-readable JSON for use in automated pipelines. It is aimed at compliance teams, security researchers, and developers building anti-money-laundering checks into their own tools.
+<!-- cognis:layman:end -->
+
 ## Contents
 
 - [Why cryptotrace?](#why) · [Features](#features) · [Quick start](#quick-start) · [Example](#example) · [Architecture](#architecture) · [AI stack](#ai-stack) · [How it compares](#how-it-compares) · [Integrations](#integrations) · [Install anywhere](#install-anywhere) · [Related](#related) · [Contributing](#contributing)
@@ -47,10 +53,46 @@ Free-tier blockchain investigator — ETH/BTC clustering + sanctions xref — wi
 <div align="right"><a href="#top">↑ back to top</a></div>
 
 <a name="quick-start"></a>
+<!-- cognis:install:start -->
+## Install
+
+`cryptotrace` is source-available (not published to PyPI) — every method below installs
+straight from GitHub. Pick whichever you prefer; the one-line scripts auto-detect
+the best tool available on your machine.
+
+**One-liner (Linux / macOS):**
+```sh
+curl -fsSL https://raw.githubusercontent.com/cognis-digital/cryptotrace/HEAD/install.sh | sh
+```
+
+**One-liner (Windows PowerShell):**
+```powershell
+irm https://raw.githubusercontent.com/cognis-digital/cryptotrace/HEAD/install.ps1 | iex
+```
+
+**Or install manually — any one of:**
+```sh
+pipx install "git+https://github.com/cognis-digital/cryptotrace.git"     # isolated (recommended)
+uv tool install "git+https://github.com/cognis-digital/cryptotrace.git"  # uv
+pip install "git+https://github.com/cognis-digital/cryptotrace.git"      # pip
+```
+
+**From source:**
+```sh
+git clone https://github.com/cognis-digital/cryptotrace.git
+cd cryptotrace && pip install .
+```
+
+Then run:
+```sh
+cryptotrace --help
+```
+<!-- cognis:install:end -->
+
 ## Quick start
 
 ```bash
-pip install cognis-cryptotrace
+pip install "git+https://github.com/cognis-digital/cryptotrace.git"
 cryptotrace --version
 cryptotrace scan .                       # scan current project
 cryptotrace scan . --format json         # machine-readable

cryptotrace/__init__.py

@@ -15,11 +15,15 @@
     Finding,
     Cluster,
     TraceResult,
+    Transfer,
     parse_txs,
     analyze,
     cluster_addresses,
     is_sanctioned,
     ofac_entries,
+    classify_address,
+    sanctions_xref,
+    investigate,
 )
 
 __all__ = [
@@ -30,9 +34,13 @@
     "Finding",
     "Cluster",
     "TraceResult",
+    "Transfer",
     "parse_txs",
     "analyze",
     "cluster_addresses",
     "is_sanctioned",
     "ofac_entries",
+    "classify_address",
+    "sanctions_xref",
+    "investigate",
 ]

cryptotrace/cli.py

@@ -7,13 +7,16 @@
 
 from . import TOOL_NAME, TOOL_VERSION
 from .core import (
-    SEVERITY_ORDER,
     TraceResult,
+    Transfer,
     analyze,
+    classify_address,
     cluster_addresses,
+    investigate,
     is_sanctioned,
     ofac_entries,
     parse_txs,
+    sanctions_xref,
 )
 
 
@@ -70,13 +73,24 @@ def _render_table(res: TraceResult) -> str:
 
 
 def _cmd_screen(args: argparse.Namespace) -> int:
+    if args.max_hops < 0:
+        print("error: --max-hops must be >= 0", file=sys.stderr)
+        return 2
     try:
         text = _read(args.txfile)
     except OSError as exc:
         print(f"error: cannot read tx file: {exc}", file=sys.stderr)
         return 2
-    txs = parse_txs(text)
-    res = analyze(txs, max_hops=args.max_hops)
+    try:
+        txs = parse_txs(text)
+    except Exception as exc:
+        print(f"error: failed to parse tx file: {exc}", file=sys.stderr)
+        return 2
+    try:
+        res = analyze(txs, max_hops=args.max_hops)
+    except Exception as exc:
+        print(f"error: analysis failed: {exc}", file=sys.stderr)
+        return 2
 
     if args.format == "json":
         out = json.dumps(res.to_dict(), indent=2)
@@ -105,7 +119,11 @@ def _cmd_cluster(args: argparse.Namespace) -> int:
     except OSError as exc:
         print(f"error: cannot read tx file: {exc}", file=sys.stderr)
         return 2
-    clusters = cluster_addresses(parse_txs(text))
+    try:
+        clusters = cluster_addresses(parse_txs(text))
+    except Exception as exc:
+        print(f"error: clustering failed: {exc}", file=sys.stderr)
+        return 2
     if args.format == "json":
         print(json.dumps([c.to_dict() for c in clusters], indent=2))
     else:
@@ -149,6 +167,80 @@ def _cmd_sdn(args: argparse.Namespace) -> int:
     return 0
 
 
+def _cmd_investigate(args: argparse.Namespace) -> int:
+    """``investigate`` subcommand — high-level Transfer-based investigation."""
+    path = args.txfile
+    try:
+        with open(path, "r", encoding="utf-8") as fh:
+            raw = fh.read()
+    except OSError as exc:
+        print(f"error: cannot read file: {exc}", file=sys.stderr)
+        return 1
+    try:
+        rows = json.loads(raw)
+    except json.JSONDecodeError as exc:
+        print(f"error: invalid JSON: {exc}", file=sys.stderr)
+        return 1
+    if not isinstance(rows, list):
+        print("error: expected a JSON array of transfer records", file=sys.stderr)
+        return 1
+    transfers = [
+        Transfer(
+            src=r.get("src", r.get("from", "")),
+            dst=r.get("dst", r.get("to", "")),
+            value=float(r.get("value", 0) or 0),
+            inputs=r.get("inputs", []),
+            asset=str(r.get("asset", "ETH")),
+            txid=str(r.get("txid", "")),
+        )
+        for r in rows if isinstance(r, dict)
+    ]
+    report = investigate(transfers)
+    fmt = getattr(args, "format", "table")
+    if fmt == "json":
+        print(json.dumps(report, indent=2))
+    else:
+        s = report["summary"]
+        print(f"CRYPTOTRACE investigate report")
+        print(f"  Transfers        : {s['total_transfers']}")
+        print(f"  Addresses        : {s['total_addresses']}")
+        print(f"  Flagged          : {s['flagged_addresses']}")
+        print(f"  Sanctioned clusters: {s['sanctioned_clusters']}")
+        print(f"  Highest severity : {s['max_severity'].upper()}")
+        for f in report["findings"]:
+            print(f"  [{f['severity'].upper():8}] {f['kind']:24} {f['address']}")
+    return 0
+
+
+def _cmd_xref(args: argparse.Namespace) -> int:
+    """``xref`` subcommand — OFAC xref for a single address; exit 2 on hit."""
+    hits = sanctions_xref([args.address])
+    fmt = getattr(args, "format", "table")
+    if fmt == "json":
+        print(json.dumps(hits, indent=2))
+    else:
+        if hits:
+            h = hits[0]
+            print(f"SANCTIONED: {h['address']}")
+            print(f"  entity  : {h['entity']}")
+            print(f"  program : {h['program']}")
+            print(f"  listed  : {h['added']}")
+        else:
+            print(f"clean: {args.address} is not on the bundled OFAC SDN list")
+    return 2 if hits else 0
+
+
+def _cmd_classify(args: argparse.Namespace) -> int:
+    """``classify`` subcommand — classify an address type; exit 1 if invalid."""
+    kind = classify_address(args.address)
+    fmt = getattr(args, "format", "table")
+    if fmt == "json":
+        print(json.dumps({"address": args.address, "type": kind}, indent=2))
+    else:
+        print(f"{args.address}  →  {kind}")
+    return 1 if kind == "invalid" else 0
+
+
 def _build_parser() -> argparse.ArgumentParser:
     p = argparse.ArgumentParser(
         prog=TOOL_NAME,
@@ -157,8 +249,11 @@ def _build_parser() -> argparse.ArgumentParser:
     )
     p.add_argument("--version", action="version",
                    version=f"{TOOL_NAME} {TOOL_VERSION}")
+    # Global --format so callers can place it before OR after the subcommand.
+    p.add_argument("--format", choices=("table", "json"), default="table",
+                   help="output format (default: table)")
 
-    # Shared parent so every subcommand accepts --format {table,json}.
+    # Shared parent so every subcommand also accepts --format locally.
     fmt = argparse.ArgumentParser(add_help=False)
     fmt.add_argument("--format", choices=("table", "json"), default="table",
                      help="output format")
@@ -187,13 +282,35 @@ def _build_parser() -> argparse.ArgumentParser:
                          help="list the bundled OFAC SDN crypto addresses")
     sdn.set_defaults(func=_cmd_sdn)
 
+    inv = sub.add_parser("investigate", parents=[fmt],
+                         help="full investigation over a Transfer JSON list")
+    inv.add_argument("txfile", help="JSON file with transfer records")
+    inv.set_defaults(func=_cmd_investigate)
+
+    xr = sub.add_parser("xref", parents=[fmt],
+                        help="OFAC xref for a single address (exit 2 on hit)")
+    xr.add_argument("address", help="address to xref")
+    xr.set_defaults(func=_cmd_xref)
+
+    cl = sub.add_parser("classify", parents=[fmt],
+                        help="classify an address type (exit 1 if invalid)")
+    cl.add_argument("address", help="address to classify")
+    cl.set_defaults(func=_cmd_classify)
+
     return p
 
 
 def main(argv: list[str] | None = None) -> int:
     parser = _build_parser()
     args = parser.parse_args(argv)
-    return args.func(args)
+    try:
+        return args.func(args)
+    except KeyboardInterrupt:
+        print("\ninterrupted", file=sys.stderr)
+        return 130
+    except Exception as exc:  # noqa: BLE001
+        print(f"error: unexpected failure: {exc}", file=sys.stderr)
+        return 2
 
 
 if __name__ == "__main__":