chainsentry.py

#!/usr/bin/env python3
"""
╔═══════════════════════════════════════════════════════════════╗
║              ChainSentry — Bitcoin Forensics Tool             ║
║                                                               ║
║  Traces Bitcoin wallet activity and fund flows using the      ║
║  free BlockCypher public API (no API key required).           ║
║                                                               ║
║  Features:                                                     ║
║  • Fetches wallet balance and full transaction history         ║
║  • Recursively follows output addresses up to N hops deep     ║
║  • Visited-set deduplication to prevent infinite loops        ║
║  • Clean hierarchical transaction-tree report                  ║
╚═══════════════════════════════════════════════════════════════╝
"""

import sys
import time
import argparse
import textwrap
from typing import Optional
from dataclasses import dataclass, field

import requests

# ──────────────────────────────────────────────────────────────
# CONSTANTS
# ──────────────────────────────────────────────────────────────

BASE_URL   = "https://api.blockcypher.com/v1/btc/main"
MAX_HOPS   = 3          # Default recursion depth
RATE_DELAY = 0.4        # Seconds between API calls (respects free-tier limits)
REQUEST_TIMEOUT = 15    # HTTP timeout in seconds

# ANSI colour codes for terminal output
RESET  = "\033[0m"
BOLD   = "\033[1m"
CYAN   = "\033[96m"
GREEN  = "\033[92m"
YELLOW = "\033[93m"
RED    = "\033[91m"
GREY   = "\033[90m"
BLUE   = "\033[94m"
MAGENTA= "\033[95m"


# ──────────────────────────────────────────────────────────────
# DATA CLASSES
# ──────────────────────────────────────────────────────────────

@dataclass
class TxOutput:
    """Represents a single output entry within a Bitcoin transaction."""
    addresses: list[str]       # Bitcoin addresses receiving funds
    value_satoshi: int         # Amount in satoshis


@dataclass
class Transaction:
    """Lightweight representation of a Bitcoin transaction."""
    txid: str
    confirmed: Optional[str]   # ISO-8601 timestamp (None if unconfirmed)
    total_input_sat: int
    total_output_sat: int
    fee_sat: int
    outputs: list[TxOutput]


@dataclass
class WalletInfo:
    """Aggregated balance and transaction list for a Bitcoin address."""
    address: str
    balance_sat: int
    total_received_sat: int
    total_sent_sat: int
    n_tx: int
    transactions: list[Transaction] = field(default_factory=list)


@dataclass
class TreeNode:
    """Node in the fund-flow tree, representing one traced address."""
    address: str
    depth: int
    wallet: Optional[WalletInfo] = None
    children: list["TreeNode"] = field(default_factory=list)


# ──────────────────────────────────────────────────────────────
# API LAYER
# ──────────────────────────────────────────────────────────────

class BlockCypherClient:
    """
    Thin wrapper around the BlockCypher REST API.

    All methods raise RuntimeError on non-2xx responses so
    callers can handle errors cleanly without inspecting HTTP codes.
    """

    def __init__(self, base_url: str = BASE_URL):
        self.base_url = base_url
        self.session = requests.Session()
        self.session.headers.update({"User-Agent": "ChainSentry/1.0"})

    def _get(self, path: str, params: Optional[dict] = None) -> dict:
        """
        Issue a GET request and return the parsed JSON body.

        Automatically rests RATE_DELAY seconds between calls to stay
        within BlockCypher's free-tier limit of ~3 req/s.
        """
        url = f"{self.base_url}{path}"
        time.sleep(RATE_DELAY)
        try:
            resp = self.session.get(url, params=params, timeout=REQUEST_TIMEOUT)
            resp.raise_for_status()
            return resp.json()
        except requests.exceptions.HTTPError as exc:
            raise RuntimeError(f"HTTP {exc.response.status_code} for {url}") from exc
        except requests.exceptions.RequestException as exc:
            raise RuntimeError(f"Network error: {exc}") from exc

    # ----------------------------------------------------------
    # Public helpers
    # ----------------------------------------------------------

    def fetch_address(self, address: str) -> dict:
        """
        Fetch the full address resource including embedded transactions.

        BlockCypher returns up to 50 txrefs by default; we request a
        higher limit and ask for full transaction objects via the
        `includeHex=false` shorthand query parameters.

        Docs: https://www.blockcypher.com/dev/bitcoin/#address-full-endpoint
        """
        return self._get(
            f"/addrs/{address}/full",
            params={"limit": 50, "txlimit": 20, "includeHex": "false"},
        )

    def fetch_transaction(self, txid: str) -> dict:
        """
        Fetch a single transaction by its hash.

        Docs: https://www.blockcypher.com/dev/bitcoin/#transaction-hash-endpoint
        """
        return self._get(f"/txs/{txid}")


# ──────────────────────────────────────────────────────────────
# PARSING LAYER
# ──────────────────────────────────────────────────────────────

def parse_tx_output(raw_output: dict) -> TxOutput:
    """Convert a raw BlockCypher output dict into a TxOutput dataclass."""
    return TxOutput(
        addresses=raw_output.get("addresses") or [],
        value_satoshi=raw_output.get("value", 0),
    )


def parse_transaction(raw_tx: dict) -> Transaction:
    """Convert a raw BlockCypher transaction dict into a Transaction dataclass."""
    fee = raw_tx.get("fees", 0) or 0

    # BlockCypher surfaces totals as total_input and total_output inside
    # the full-address endpoint response; fall back to summing outputs.
    total_out = raw_tx.get("total", 0) or sum(
        o.get("value", 0) for o in raw_tx.get("outputs", [])
    )
    total_in = total_out + fee

    outputs = [parse_tx_output(o) for o in raw_tx.get("outputs", [])]

    return Transaction(
        txid=raw_tx.get("hash", "unknown"),
        confirmed=raw_tx.get("confirmed"),          # None if mempool
        total_input_sat=total_in,
        total_output_sat=total_out,
        fee_sat=fee,
        outputs=outputs,
    )


def parse_wallet_info(address: str, raw: dict) -> WalletInfo:
    """Build a WalletInfo from the BlockCypher full-address response."""
    txs = [parse_transaction(tx) for tx in raw.get("txs", [])]
    return WalletInfo(
        address=address,
        balance_sat=raw.get("balance", 0),
        total_received_sat=raw.get("total_received", 0),
        total_sent_sat=raw.get("total_sent", 0),
        n_tx=raw.get("n_tx", 0),
        transactions=txs,
    )


# ──────────────────────────────────────────────────────────────
# CORE FORENSICS LOGIC
# ──────────────────────────────────────────────────────────────

def collect_output_addresses(wallet: WalletInfo, self_address: str) -> list[str]:
    """
    Extract unique addresses that *received* funds from this wallet.

    We skip:
      • The source address itself (change outputs).
      • Addresses already seen by the caller (handled in visited set).

    Returns a deduplicated list preserving first-seen order.
    """
    seen: set[str] = set()
    result: list[str] = []
    for tx in wallet.transactions:
        for output in tx.outputs:
            for addr in output.addresses:
                if addr != self_address and addr not in seen:
                    seen.add(addr)
                    result.append(addr)
    return result


def trace_funds(
    client: BlockCypherClient,
    address: str,
    visited: set[str],
    depth: int,
    max_depth: int,
) -> TreeNode:
    """
    Recursively build a fund-flow tree rooted at `address`.

    Algorithm:
      1. Add `address` to `visited` immediately to prevent loops.
      2. Fetch balance + transactions from BlockCypher.
      3. Collect all output addresses not yet visited.
      4. For each candidate, recurse with depth+1 until max_depth.

    Parameters
    ----------
    client    : BlockCypherClient instance
    address   : Bitcoin address to analyse at this node
    visited   : Global set of already-processed addresses
    depth     : Current recursion depth (0 = root / seed address)
    max_depth : Maximum depth to recurse to

    Returns
    -------
    TreeNode with fully populated children.
    """
    node = TreeNode(address=address, depth=depth)

    # Guard: mark visited before any recursive call to avoid re-entry
    visited.add(address)

    indent = "  " * depth
    print(f"{GREY}{indent}[depth {depth}] Fetching {address[:20]}...{RESET}")

    try:
        raw = client.fetch_address(address)
        wallet = parse_wallet_info(address, raw)
        node.wallet = wallet
    except RuntimeError as exc:
        print(f"{RED}{indent}  ✗ Could not fetch {address}: {exc}{RESET}")
        return node  # Return partial node; children remain empty

    # Only recurse if we haven't hit the depth ceiling
    if depth < max_depth:
        output_addrs = collect_output_addresses(wallet, address)
        for dest_addr in output_addrs:
            if dest_addr in visited:
                continue   # Already processed — skip to prevent loops
            child_node = trace_funds(client, dest_addr, visited, depth + 1, max_depth)
            node.children.append(child_node)

    return node


# ──────────────────────────────────────────────────────────────
# REPORTING LAYER
# ──────────────────────────────────────────────────────────────

def sat_to_btc(satoshis: int) -> str:
    """Format a satoshi integer as a human-readable BTC string."""
    return f"{satoshis / 1e8:.8f} BTC"


def format_txid(txid: str, width: int = 16) -> str:
    """Truncate a txid to `width` chars + ellipsis for display."""
    return txid[:width] + "…" if len(txid) > width else txid


def print_header(address: str) -> None:
    """Print the top-level report banner."""
    line = "═" * 62
    print(f"\n{CYAN}{BOLD}╔{line}╗")
    print(f"║{'ChainSentry — Blockchain Forensics Report':^62}║")
    print(f"╠{line}╣")
    print(f"║  Seed Address: {YELLOW}{address}{CYAN}{'':>{max(0, 44 - len(address))}}║")
    print(f"╚{line}╝{RESET}\n")


def print_wallet_summary(wallet: WalletInfo, depth: int) -> None:
    """Print the balance/stats block for a single wallet node."""
    pad = "  " * depth
    col = [CYAN, GREEN, YELLOW, MAGENTA, BLUE][min(depth, 4)]

    print(f"{col}{pad}┌─ Address: {BOLD}{wallet.address}{RESET}")
    print(f"{col}{pad}│  Balance       : {sat_to_btc(wallet.balance_sat)}{RESET}")
    print(f"{col}{pad}│  Total Received: {sat_to_btc(wallet.total_received_sat)}{RESET}")
    print(f"{col}{pad}│  Total Sent    : {sat_to_btc(wallet.total_sent_sat)}{RESET}")
    print(f"{col}{pad}│  Transactions  : {wallet.n_tx} on-chain{RESET}")

    # Print the most recent transactions (cap at 5 for readability)
    recent = wallet.transactions[:5]
    if recent:
        print(f"{col}{pad}│  Recent Txs:{RESET}")
        for tx in recent:
            status = "✓ confirmed" if tx.confirmed else "⏳ unconfirmed"
            date   = tx.confirmed[:10] if tx.confirmed else "pending"
            print(
                f"{GREY}{pad}│    [{format_txid(tx.txid)}] "
                f"{date}  fee={sat_to_btc(tx.fee_sat)}  "
                f"out={sat_to_btc(tx.total_output_sat)}  {status}{RESET}"
            )


def print_tree(node: TreeNode) -> None:
    """
    Recursively print the transaction tree in a readable indented format.

    Depth is conveyed visually via indentation and colour cycling so
    analysts can quickly spot the hop level for each address.
    """
    if node.wallet:
        print_wallet_summary(node.wallet, node.depth)
    else:
        pad = "  " * node.depth
        print(f"{RED}{pad}┌─ Address: {node.address}  (no data retrieved){RESET}")

    for child in node.children:
        print()   # Blank line between sibling subtrees
        print_tree(child)

    # Close the visual bracket for non-leaf nodes
    if node.children:
        pad = "  " * node.depth
        col = [CYAN, GREEN, YELLOW, MAGENTA, BLUE][min(node.depth, 4)]
        print(f"{col}{pad}└─ (end of hop {node.depth}){RESET}")


def print_summary_stats(root: TreeNode, visited: set[str]) -> None:
    """Print aggregate statistics for the entire trace session."""
    # Walk the tree to count non-empty nodes and total BTC seen
    total_received = 0
    non_empty = 0

    def _walk(n: TreeNode) -> None:
        nonlocal total_received, non_empty
        if n.wallet:
            non_empty += 1
            total_received += n.wallet.total_received_sat
        for c in n.children:
            _walk(c)

    _walk(root)

    line = "─" * 62
    print(f"\n{BOLD}{CYAN}{'─' * 62}{RESET}")
    print(f"{BOLD}  Trace Summary{RESET}")
    print(f"{CYAN}{line}{RESET}")
    print(f"  Addresses traced : {BOLD}{len(visited)}{RESET}")
    print(f"  Nodes with data  : {BOLD}{non_empty}{RESET}")
    print(f"  Total BTC seen   : {BOLD}{sat_to_btc(total_received)}{RESET}")
    print(f"{CYAN}{line}{RESET}\n")


# ──────────────────────────────────────────────────────────────
# CLI ENTRY POINT
# ──────────────────────────────────────────────────────────────

def build_arg_parser() -> argparse.ArgumentParser:
    """Define and return the CLI argument parser."""
    parser = argparse.ArgumentParser(
        prog="chainsentry",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        description=textwrap.dedent("""\
            ChainSentry — Bitcoin Forensics Tool
            ─────────────────────────────────────
            Traces fund flows from a seed Bitcoin address up to
            MAX_HOPS hops deep using the free BlockCypher API.
        """),
        epilog=textwrap.dedent("""\
            Examples:
              python chainsentry.py 1A1zP1eP5QGefi2DMPTfTL5SLmv7Divf1V
              python chainsentry.py <address> --hops 2
        """),
    )
    parser.add_argument(
        "address",
        help="Seed Bitcoin address to start tracing from.",
    )
    parser.add_argument(
        "--hops",
        type=int,
        default=MAX_HOPS,
        metavar="N",
        help=f"Maximum recursive depth (default: {MAX_HOPS}).",
    )
    return parser


def main() -> None:
    parser = build_arg_parser()
    args = parser.parse_args()

    address  = args.address.strip()
    max_hops = max(0, args.hops)   # Clamp to non-negative

    # Basic sanity check — Bitcoin addresses are 25-34 chars
    if len(address) < 25 or len(address) > 62:
        print(f"{RED}✗ '{address}' does not look like a valid Bitcoin address.{RESET}")
        sys.exit(1)

    print_header(address)
    print(f"{BOLD}Starting trace  ·  max depth = {max_hops}{RESET}\n")

    client  = BlockCypherClient()
    visited: set[str] = set()

    try:
        root = trace_funds(
            client=client,
            address=address,
            visited=visited,
            depth=0,
            max_depth=max_hops,
        )
    except KeyboardInterrupt:
        print(f"\n{YELLOW}⚠  Trace interrupted by user.{RESET}\n")
        sys.exit(0)

    # ── Print the full hierarchical report ──────────────────────
    print(f"\n{BOLD}{CYAN}{'═' * 62}")
    print(f"{'  TRANSACTION TREE REPORT':^62}")
    print(f"{'═' * 62}{RESET}\n")

    print_tree(root)
    print_summary_stats(root, visited)


if __name__ == "__main__":
    main()