#!/usr/bin/env python3 from __future__ import annotations import csv from collections import Counter, defaultdict from dataclasses import dataclass from pathlib import Path from urllib.parse import urlsplit ROOT = Path(__file__).resolve().parents[1] VAULT_CREDS = ROOT / "findings" / "data" / "vault_creds.csv" OUT_GRAPH = ROOT / "findings" / "data" / "password_reset_chain_graph.csv" OUT_SUMMARY = ROOT / "findings" / "data" / "password_reset_chain_summary.md" EMAIL_PROVIDER_PATTERNS = { "microsoft": ["login.live.com", "outlook.", "hotmail.", "office.com", "microsoftonline.com"], "google": ["accounts.google.com", "mail.google.com", "gmail."], "proton": ["proton.me", "protonmail.com", "account.proton.me"], "mailru": ["mail.ru"], "yandex": ["yandex."], "rambler": ["rambler."], "yahoo": ["yahoo."], } CATEGORY_PATTERNS = { "banking": ["chase.com", "wellsfargo.com", "usbank.com", "navyfederal.org", "santander", "bank", "telebank", "lloydsbank"], "financial": ["paypal.com", "venmo.com", "blockchain.com", "fidelity.com", "creditkarma.com"], "developer": ["github.com", "gitlab", "docker.com", "vercel.com", "atlassian.net", "jira", "jenkins"], "email": ["outlook.", "hotmail.", "gmail.", "proton", "mail.ru", "yandex.", "rambler.", "yahoo."], "social": ["discord.com", "facebook.com", "instagram.com", "twitter.com", "x.com", "reddit.com"], "shopping": ["amazon.", "ebay.", "aliexpress.", "wayfair.", "temu.", "walmart."], "streaming": ["netflix.com", "spotify.com", "disneyplus", "disney.", "hulu.com"], "gaming": ["steam", "epicgames.com", "roblox.com", "riotgames.com", "minecraft", "rockstargames.com"], } @dataclass(frozen=True) class VaultRecord: dump_id: str browser: str url: str url_norm: str username: str username_norm: str password: str def normalize_url(value: str) -> str: value = (value or "").strip() if not value: return "" if value.startswith("android://"): return value.lower() try: parts = urlsplit(value) scheme = parts.scheme.lower() netloc = parts.netloc.lower() path = parts.path.rstrip("/") or "/" return f"{scheme}://{netloc}{path}" except ValueError: return value.lower().rstrip("/") def normalize_username(value: str) -> str: value = (value or "").strip() if not value: return "" return value.lower() if "@" in value else value def classify_email_provider(url_norm: str) -> str: for provider, patterns in EMAIL_PROVIDER_PATTERNS.items(): if any(pattern in url_norm for pattern in patterns): return provider return "" def classify_target(url_norm: str) -> str: for category, patterns in CATEGORY_PATTERNS.items(): if any(pattern in url_norm for pattern in patterns): return category return "other" def load_vault_records() -> list[VaultRecord]: records: list[VaultRecord] = [] with VAULT_CREDS.open(newline="", errors="ignore") as handle: reader = csv.DictReader(handle) for row in reader: username = (row["username"] or "").strip() password = (row["password"] or "").strip() if not username or not password: continue records.append( VaultRecord( dump_id=row["dump_id"], browser=row["browser"], url=row["url"], url_norm=normalize_url(row["url"]), username=username, username_norm=normalize_username(username), password=password, ) ) return records def build_edges(records: list[VaultRecord]) -> list[dict[str, str]]: by_dump: dict[str, list[VaultRecord]] = defaultdict(list) for record in records: by_dump[record.dump_id].append(record) seen: set[tuple[str, str, str, str, str]] = set() edges: list[dict[str, str]] = [] for dump_id, dump_records in by_dump.items(): email_records = [] for record in dump_records: provider = classify_email_provider(record.url_norm) if provider and "@" in record.username_norm: email_records.append((record, provider)) if not email_records: continue for email_record, provider in email_records: for target in dump_records: if target is email_record: continue target_category = classify_target(target.url_norm) if target_category == "email" and target.username_norm == email_record.username_norm: continue if target.username_norm == email_record.username_norm: inference_type = "same_username" confidence = "high" else: inference_type = "co_resident_dump" confidence = "medium" key = ( dump_id, email_record.username_norm, target.url_norm, target.username_norm, inference_type, ) if key in seen: continue seen.add(key) edges.append( { "dump_id": dump_id, "email_provider": provider, "email_account": email_record.username, "email_account_norm": email_record.username_norm, "email_url": email_record.url, "target_category": target_category, "target_url": target.url, "target_url_norm": target.url_norm, "target_username": target.username, "target_username_norm": target.username_norm, "inference_type": inference_type, "confidence": confidence, } ) return edges def write_graph(edges: list[dict[str, str]]) -> None: fieldnames = [ "dump_id", "email_provider", "email_account", "email_account_norm", "email_url", "target_category", "target_url", "target_url_norm", "target_username", "target_username_norm", "inference_type", "confidence", ] with OUT_GRAPH.open("w", newline="") as handle: writer = csv.DictWriter(handle, fieldnames=fieldnames) writer.writeheader() writer.writerows(edges) def write_summary(edges: list[dict[str, str]]) -> None: unique_email_nodes = sorted({edge["email_account_norm"] for edge in edges}) by_email: dict[str, set[str]] = defaultdict(set) by_category = Counter() by_provider = Counter() by_confidence = Counter() for edge in edges: by_email[edge["email_account_norm"]].add(edge["target_url_norm"] or edge["target_username_norm"]) by_category[edge["target_category"]] += 1 by_provider[edge["email_provider"]] += 1 by_confidence[edge["confidence"]] += 1 top_email_nodes = sorted(by_email.items(), key=lambda item: (-len(item[1]), item[0]))[:25] with OUT_SUMMARY.open("w") as handle: handle.write("# Password Reset Chain Summary\n\n") handle.write("## Coverage\n\n") handle.write(f"- Email-driven inferred edges: **{len(edges)}**\n") handle.write(f"- Unique email nodes: **{len(unique_email_nodes)}**\n") handle.write(f"- `high` confidence edges: **{by_confidence['high']}**\n") handle.write(f"- `medium` confidence edges: **{by_confidence['medium']}**\n\n") handle.write("## Email Provider Breakdown\n\n") for provider, count in by_provider.most_common(): handle.write(f"- `{provider}`: {count}\n") handle.write("\n## Target Category Breakdown\n\n") for category, count in by_category.most_common(): handle.write(f"- `{category}`: {count}\n") handle.write("\n## Top Email Nodes By Reachable Targets\n\n") handle.write("| Email | Reachable Targets |\n") handle.write("|-------|-------------------|\n") for email, targets in top_email_nodes: handle.write(f"| {email} | {len(targets)} |\n") handle.write("\n## Method Note\n\n") handle.write( "This graph is an inference layer, not proof of live recovery bindings. " "Edges are created from co-resident credentials within the same victim profile, " "with `high` confidence when the target username matches the email account exactly " "and `medium` confidence when the email and target service are only co-resident in the same dump.\n" ) def main() -> int: records = load_vault_records() edges = build_edges(records) write_graph(edges) write_summary(edges) print(f"vault_records={len(records)}") print(f"reset_edges={len(edges)}") print(f"unique_email_nodes={len({edge['email_account_norm'] for edge in edges})}") return 0 if __name__ == "__main__": raise SystemExit(main())