"""searchcode.com eternal lane — MCP dossier API (api.searchcode.com/v1/mcp). Old REST /api/codesearch_I/ is DEAD (404, verified 2026-10-19); searchcode relaunched as per-repo "dossier" service: no global code search, only repository-scoped analysis. Their scanner reports secret locations (security_summary.findings: file+line+type+confidence, capped at 5/repo, dossiers cached per commit_sha → 0.1s for popular repos, on-demand clone otherwise). Pipeline: known_repos pool (keys.db, 226k slugs) → code_analyze → findings → code_get_files (whole flagged files) → key_patterns.scan_text → keys.db. Stdlib+requests only, same conventions as other *_eternal.py. """ import os KF_BASE = os.path.dirname(os.path.abspath(__file__)) os.makedirs(os.path.join(KF_BASE, "data"), exist_ok=True) import re import json import sqlite3 import sys import time import hashlib import requests from key_patterns import scan_text DB = os.environ.get("KF_DB") or os.path.join(KF_BASE, "data", "keys.db") MCP = "https://api.searchcode.com/v1/mcp" UA = {"User-Agent": "keyfarm-searchcode/2.0", "Content-Type": "application/json", "Accept": "application/json, text/event-stream"} REPOS_FILE = os.path.join(KF_BASE, "data", "searchcode_repos.txt") CURSOR_FILE = os.path.join(KF_BASE, "data", "searchcode_cursor.txt") REPOS_PER_CYCLE = 3 SLEEP = 3 # politeness between repos (beta API, limits unknown) CALL_TIMEOUT = 240 # uncached repos are cloned on demand → can be slow # bare "org/repo" slugs only; anything with a foreign prefix is skipped _SLUG_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]*/[A-Za-z0-9][A-Za-z0-9._-]*$") db = sqlite3.connect(DB, timeout=120, check_same_thread=False, isolation_level=None) db.execute("PRAGMA busy_timeout=120000") S = requests.Session() S.trust_env = False # bypass dashscope proxy like every other lane S.headers.update(UA) def log(msg): print(time.strftime("[%H:%M:%S] ") + msg, flush=True) def mcp_call(name, args, timeout=CALL_TIMEOUT) -> "dict | None": """One MCP tools/call. Returns parsed JSON object, or None on transport failure. Tool-level errors come back as {'code':..., 'message':...}.""" payload = {"jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": {"name": name, "arguments": args}} for attempt in (1, 2): try: r = S.post(MCP, json=payload, timeout=timeout) if r.status_code == 429: wait = int(r.headers.get("Retry-After", 60)) log(f" mcp 429, sleeping {wait}s") time.sleep(wait) continue if r.status_code != 200: log(f" mcp http {r.status_code}") return None d = r.json() if "error" in d: log(f" mcp rpc-error {str(d['error'])[:100]}") return None content = d["result"]["content"] try: parsed = json.loads(content[0]["text"]) return parsed if isinstance(parsed, dict) else {"raw": parsed} except (ValueError, KeyError, IndexError, TypeError): return {"raw": str(content)[:200]} except (requests.RequestException, ValueError) as e: if attempt == 2: log(f" mcp {name} fail {type(e).__name__} {str(e)[:60]}") return None time.sleep(10) return None def mine_pool(): """known_repos → cleaned github slug pool file.""" seen = set() out = [] cur = db.execute("SELECT repo FROM known_repos") for (raw,) in cur: if not raw: continue slug = raw.strip() # tolerate the few full-URL / prefixed rows m = re.search(r"github\.com/([^/\s]+/[^/\s#?]+)", slug) if m: slug = m.group(1) slug = slug.rstrip("/") if not _SLUG_RE.match(slug) or slug in seen: continue seen.add(slug) out.append(slug) tmp = REPOS_FILE + ".tmp" with open(tmp, "w", encoding="utf-8") as f: f.write("\n".join(out) + "\n") os.replace(tmp, REPOS_FILE) log(f"pool mined: {len(out)} repos") if len(out) < 1000: raise RuntimeError(f"pool too small ({len(out)}) — known_repos shape changed?") return out def load_pool(): if not os.path.exists(REPOS_FILE): return mine_pool() with open(REPOS_FILE, encoding="utf-8") as f: return [ln.strip() for ln in f if ln.strip()] def load_cursor(): try: with open(CURSOR_FILE) as f: return int(f.read().strip() or 0) except (OSError, ValueError): return 0 def save_cursor(v): tmp = CURSOR_FILE + ".tmp" with open(tmp, "w") as f: f.write(str(v)) os.replace(tmp, CURSOR_FILE) def insert_key(key, prov, repo): h = hashlib.sha1(key.encode()).hexdigest()[:16] try: db.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')", (h, key, prov, "NEW", "searchcode", repo)) except sqlite3.Error: pass def check_repo(slug, seen): """Analyze one repo; fetch flagged files; scan. Returns new-key count.""" repo = f"github.com/{slug}" a = mcp_call("code_analyze", {"repository": repo}) if a is None: return -1 # transport failure if a.get("code"): log(f" {slug}: {a['code']}") return 0 ss = a.get("security_summary") or {} findings = [f for f in ss.get("findings", []) if not f.get("test") and f.get("file")] total = ss.get("total_findings", 0) if not findings: log(f" {slug}: secrets={total} listed=0") return 0 paths = [{"path": f["file"]} for f in findings[:10]] g = mcp_call("code_get_files", {"repository": repo, "paths": paths, "max_lines": 2000}) if not isinstance(g, dict) or not g.get("files"): log(f" {slug}: findings={len(findings)} files=0") return 0 n = 0 for x in g["files"]: text = x.get("content") or "" for prov, key in scan_text(text): key = key.strip().strip('"\'`.,:') if not key or len(key) < 20 or key in seen: continue insert_key(key, prov, repo) seen.add(key) n += 1 log(f" +{prov} {key[:12]}… {slug} {x.get('path','')}") log(f" {slug}: secrets={total} files={len(g['files'])} +{n}") return n def cycle(pool): idx = load_cursor() seen = {r[0] for r in db.execute("SELECT val FROM keys")} ok = fail = new = 0 for i in range(REPOS_PER_CYCLE): slug = pool[(idx + i) % len(pool)] r = check_repo(slug, seen) if r < 0: fail += 1 else: ok += 1 new += r time.sleep(SLEEP) save_cursor(idx + REPOS_PER_CYCLE) log(f"cycle idx={idx} ok={ok} fail={fail} +{new} keys") def main(): log("=== searchcode eternal (MCP dossiers) START ===") if "--remine" in sys.argv: try: os.remove(REPOS_FILE) except OSError: pass pool = load_pool() log(f"pool: {len(pool)} repos, cursor={load_cursor()}") if not pool: log("empty pool, abort") return if "--once" in sys.argv: cycle(pool) return while True: try: cycle(pool) except Exception as e: log(f"cycle-exception {type(e).__name__} {e}") time.sleep(SLEEP) if __name__ == "__main__": main()