"""ETERNAL Sourcegraph: public code search via SSE stream for API keys/tokens. Queries sourcegraph.com's streaming search endpoint with high-signal dorks, parses the SSE event stream, extracts snippet text from lineMatches, and feeds it to scan_text for key extraction. Endpoint(s): https://sourcegraph.com/.api/search/stream (SSE, unauth) Auth: none (Cloudflare bypassed with Chrome User-Agent) Quota: per-IP, no explicit rate-limit header observed; 10k result cap per query Collision: no (unique host) """ import os BASE = os.path.dirname(os.path.abspath(__file__)) import hashlib, json, re, sqlite3, sys, threading, time import requests S = requests.Session() S.trust_env = False os.makedirs(os.path.join(BASE, "data"), exist_ok=True) DB = os.environ.get("KF_DB") or os.path.join(BASE, "data", "keys.db") LOG = os.path.join(BASE, "data", "sourcegraph_eternal.log") SEEN_FILE = os.path.join(BASE, "data", "sourcegraph_seen.txt") N_THREADS = 1 SLEEP = 90 MAX_ITEMS = 200 from key_patterns import scan_text GARBAGE = ("xxxx", "your", "YOUR", "EXAMPLE", "example", "n0tr3al", "placeholder", "CHANGEME", "changeme", "dummy", "DUMMY", "abcdefghijklmnopqrst", "1234567890abcdef") UA = { "User-Agent": ( "Mozilla/5.0 (X11; Linux x86_64) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/120.0.0.0 Safari/537.36" ), "Accept": "text/event-stream", } # Dorks that pass Cloudflare (probe-verified). Format: (query, label). DORKS = [ ("AIzaSy", "GOOGLE_KEY"), ("AKIA", "AWS_KEY"), ("ghp_", "GITHUB_PAT"), ("nvapi-", "NVIDIA_KEY"), ("xai-", "XAI_KEY"), ("gsk_", "GROQ_KEY"), ("DATABASE_URL", "DB_URL"), ] # Dorks that hit Cloudflare WAF — logged once, never retried. BLOCKED = ("sk-", "sk-proj-", ".env") API_URL = "https://sourcegraph.com/.api/search/stream" conn = sqlite3.connect(DB, timeout=120, check_same_thread=False, isolation_level=None) conn.execute(""" CREATE TABLE IF NOT EXISTS keys ( id INTEGER PRIMARY KEY AUTOINCREMENT, hash TEXT UNIQUE, val TEXT, prov TEXT, status TEXT DEFAULT "NEW", plan TEXT, price TEXT, remaining TEXT, found TEXT, repo TEXT) """) lock = threading.Lock() total = 0 def log(msg): line = f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] {msg}\n" with open(LOG, "a", encoding="utf-8") as f: f.write(line) def db_add(key, prov, repo): global total with lock: conn.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", "sourcegraph", repo[:300])) total += 1 def load_seen(): s = set() try: with open(SEEN_FILE, encoding="utf-8") as f: for line in f: line = line.strip() if line: s.add(line) except FileNotFoundError: pass return s seen = load_seen() def mark_seen(sig): seen.add(sig) with open(SEEN_FILE, "a", encoding="utf-8") as f: f.write(sig + "\n") def is_junk(v): if any(g in v for g in GARBAGE): return True if len(set(v)) < 6: return True return False # --------------------------------------------------------------------------- # SSE parser # --------------------------------------------------------------------------- def parse_sse_events(text): """Yields (event_type, data_str) tuples from an SSE response body.""" event = None data_lines = [] for line in text.split("\n"): if line.startswith("event: "): event = line[7:] elif line.startswith("data: "): data_lines.append(line[6:]) elif line == "" and data_lines: if event: yield event, "\n".join(data_lines) event = None data_lines = [] # flush trailing event if no trailing blank line if event and data_lines: yield event, "\n".join(data_lines) # How many results per dork query (overridden by --once) DORKS_PER_QUERY = 100 # --------------------------------------------------------------------------- # Core: fetch one dork, extract snippets, scan, db_add # --------------------------------------------------------------------------- def fetch_and_scan(dork, label, max_items=DORKS_PER_QUERY): """Query sourcegraph with one dork, return number of keys found. Streams SSE via iter_lines to avoid waiting for the full 10k-result response. Stops after processing `max_items` distinct files. """ params = {"q": dork, "v": "V3", "t": "keyword", "count": max_items} n = 0 for attempt in (1, 2, 3): try: r = S.get(API_URL, params=params, headers=UA, timeout=30, stream=True) if r.status_code == 429: retry_s = int(r.headers.get("Retry-After", 60)) r.close() log(f" sg 429 {label}, retry-after {retry_s}s") time.sleep(retry_s) continue if r.status_code == 403: body = r.text[:200].lower() r.close() if "just a moment" in body or "cloudflare" in body: log(f" sg CF-blocked {label}, skipping") return 0 log(f" sg 403 {label}: {body[:100]}") return 0 if r.status_code != 200: r.close() log(f" sg {r.status_code} {label}, attempt {attempt}") if attempt == 3: return 0 time.sleep(8 * attempt) continue # Streaming SSE parse: process events as they arrive, stop early event = None data_lines = [] items_processed = 0 for raw_line in r.iter_lines(): if not raw_line: # Empty line = event boundary; process accumulated event if event and data_lines: data_str = "".join(data_lines) if event == "matches": matches = json.loads(data_str) if isinstance(matches, list): for m in matches: if items_processed >= max_items: break repo = m.get("repository", "") path = m.get("path", "") commit = m.get("commit", "") sig = f"{repo}:{path}:{commit}" if sig in seen: continue snippets = [ lm["line"] for lm in m.get("lineMatches", []) if lm.get("line") ] mark_seen(sig) if snippets: text_block = "\n".join(snippets) for prov, key in scan_text(text_block): if is_junk(key) or len(key) < 30: continue db_add(key, prov, f"{repo}:{path}") n += 1 items_processed += 1 elif event == "progress": # Log matchCount for info try: p = json.loads(data_str) if p.get("done"): pass # stream will end naturally except json.JSONDecodeError: pass # Reset for next event event = None data_lines = [] if items_processed >= max_items: # We have enough; stop reading r.close() return n continue line = raw_line.decode("utf-8", "replace") if line.startswith("event: "): event = line[7:] elif line.startswith("data: "): data_lines.append(line[6:]) r.close() return n except requests.RequestException as e: # If stream was cut short after some items, return partial result if n > 0: return n if attempt == 3: log(f" sg-err {label} (final): {e}") return 0 time.sleep(8 * attempt) except Exception as e: if n > 0: return n if attempt == 3: log(f" sg-exc {label} (final): {type(e).__name__} {e}") return 0 time.sleep(8) return 0 # --------------------------------------------------------------------------- # Cycle + main # --------------------------------------------------------------------------- def cycle(max_items=DORKS_PER_QUERY): """Run one cycle: query all dorks.""" global total total = 0 once = "--once" in sys.argv per_dork = min(max_items, 5) if once else max_items log(f"cycle start (per_dork={per_dork})") n_queries = 0 for dork, label in DORKS: before = total fetch_and_scan(dork, label, per_dork) n_queries += 1 log(f" {label}: +{total - before} keys") time.sleep(0.5) log(f"cycle done: {n_queries} queries, +{total} keys") def main(): log("sourcegraph_eternal started") # Log blocked dorks once at startup log(f" blocked by CF: {', '.join(BLOCKED)}") once = "--once" in sys.argv if once: cycle() log("sourcegraph_eternal --once done") return while True: try: cycle() except Exception as e: log(f"cycle-exception {type(e).__name__} {e}") time.sleep(SLEEP) if __name__ == "__main__": main()