"""ETERNAL Common Crawl: CDX index walk for archived config files.

Walks the Common Crawl CDX index (index.commoncrawl.org) with host queries over a
seed domain list, filters for config-like URL paths, then fetches exactly the
archived WARC record from data.commoncrawl.org with an HTTP Range request,
gunzips it, strips WARC+HTTP headers and scans the body for keys.

Different hosts from the wayback lane (index./data.commoncrawl.org vs
web.archive.org) -> separate quota pool, no collision.

Endpoint(s): https://index.commoncrawl.org/collinfo.json  (crawl listing)
             https://index.commoncrawl.org/<CC-MAIN-week>-index?output=json  (CDX)
             https://data.commoncrawl.org/<filename>  (WARC, Range)
Auth: none
"""
import os as _os
BASE = _os.path.dirname(_os.path.abspath(__file__))
_os.makedirs(_os.path.join(BASE, "data"), exist_ok=True)

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

S = requests.Session()
S.trust_env = False  # contract 4: bypass local dashscope proxy

DB = _os.environ.get("KF_DB") or _os.path.join(BASE, "data", "keys.db")
LOG = _os.path.join(BASE, "data", "commoncrawl_eternal.log")
CURSOR_FILE = _os.path.join(BASE, "data", "commoncrawl_cursor.txt")
SEEN_FILE = _os.path.join(BASE, "data", "commoncrawl_seen.txt")
N_THREADS = 1
SLEEP = 120
MAX_ITEMS = 200
CDX_LIMIT = 1000
WARC_RANGE = 400_000      # bytes pulled per record: config files are small
FILE_CAP = 500_000
BACKOFF = (2, 4, 8)

from key_patterns import scan_text

GARBAGE = ("xxxx", "your", "YOUR", "EXAMPLE", "example", "n0tr3al",
           "placeholder", "CHANGEME", "changeme", "dummy", "DUMMY",
           "abcdefghijklmnopqrst", "1234567890abcdef")

UA = {"User-Agent": "keyfarm-commoncrawl/1.0"}

# Hosts whose captures CC indexes in bulk and that historically serve
# accidental config dumps. Anonymous per-IP quota on the CDX gateway.
SEED_DOMAINS = [
    "raw.githubusercontent.com",
    "gist.githubusercontent.com",
    "pastebin.com",
    "gitlab.com",
    "bitbucket.org",
    "codeberg.org",
    "sourceforge.net",
    "s3.amazonaws.com",
    "storage.googleapis.com",
    "jsdelivr.net",
    "unpkg.com",
]

# Path suffixes that harbour secrets / tokens.
_CONFIG_RE = re.compile(
    r'/(?:'
    r'\.env(?:\.[\w-]+)?|'
    r'credentials(?:\.[\w-]+)?|'
    r'config\.(?:js|json|php)|'
    r'configuration\.php|'
    r'wp-config\.php|'
    r'application\.properties|'
    r'\.npmrc|\.pypirc|'
    r'id_(?:rsa|ed25519|ecdsa|dsa)(?:\.pub)?|'
    r'secrets?(?:\.[\w-]+)?|'
    r'\.netrc|\.git-credentials|\.dockercfg|'
    r'setup\.config|web\.config|'
    r'database\.ya?ml|config\.ya?ml'
    r')(?:[?#].*)?$',
    re.IGNORECASE)


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)


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
    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", "commoncrawl", (repo or "")[:300]))
        total += 1


def load_cursor():
    """Returns (crawl_idx, domain_idx, page)."""
    try:
        with open(CURSOR_FILE, encoding="utf-8") as f:
            p = f.read().split()
        if len(p) == 3:
            return int(p[0]), int(p[1]), int(p[2])
    except (OSError, ValueError):
        pass
    return 0, 0, 0


def save_cursor(crawl_idx, domain_idx, page):
    with open(CURSOR_FILE, "w", encoding="utf-8") as f:
        f.write(f"{crawl_idx} {domain_idx} {page}")


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 OSError:
        pass
    return s


seen = load_seen()


def mark_seen(url):
    seen.add(url)
    with open(SEEN_FILE, "a", encoding="utf-8") as f:
        f.write(url + "\n")


def is_junk(v):
    """Contract 5: placeholder / low-entropy noise filter."""
    if any(g in v for g in GARBAGE):
        return True
    if len(set(v)) < 6:
        return True
    if len(v) < 30 and not any(c.isdigit() for c in v):
        return True
    return False


def is_config_url(url):
    """True when the path's last segment looks like a secret-bearing config."""
    return bool(_CONFIG_RE.search(url.split("#")[0]))


def get(url, hdrs=None, timeout=30, tries=3):
    """GET with backoff, honouring Retry-After on 429/403. -> resp | None."""
    h = dict(UA)
    if hdrs:
        h.update(hdrs)
    for a in range(tries):
        try:
            r = S.get(url, headers=h, timeout=timeout)
        except requests.RequestException as ex:
            log(f"  net err {type(ex).__name__} on {url[:120]}")
            time.sleep(BACKOFF[min(a, len(BACKOFF) - 1)])
            continue
        if r.status_code in (429, 403):
            ra = r.headers.get("Retry-After")
            try:
                nap = min(int(ra), 60)
            except (TypeError, ValueError):
                nap = BACKOFF[min(a, len(BACKOFF) - 1)]
            log(f"  {r.status_code} throttle, sleep {nap}")
            time.sleep(nap)
            continue
        if r.status_code >= 500:          # CC CDX gateway 504s regularly
            time.sleep(BACKOFF[min(a, len(BACKOFF) - 1)])
            continue
        return r
    return None


def fetch_crawls():
    """collinfo.json -> [crawl-id, ...] newest first."""
    r = get("https://index.commoncrawl.org/collinfo.json", timeout=25)
    if r is None or r.status_code != 200:
        return []
    try:
        return [d["id"] for d in r.json()]
    except (ValueError, KeyError, TypeError):
        return []


def parse_cdx_rows(text):
    """NDJSON CDX body -> [row dict, ...]; ignores non-JSON/HTML noise."""
    rows = []
    for line in text.split("\n"):
        line = line.strip()
        if not line.startswith("{"):
            continue
        try:
            j = json.loads(line)
        except ValueError:
            continue
        if isinstance(j, dict):
            rows.append(j)
    return rows


def fetch_cdx(crawl, domain, page, limit):
    """One CDX page -> [row, ...]. Empty list == no more data / hard error."""
    url = (f"https://index.commoncrawl.org/{crawl}-index"
           f"?output=json&fl=urlkey,timestamp,url,status,length,offset,filename"
           f"&matchType=host&url={domain}&limit={limit}&page={page}")
    r = get(url, timeout=45)
    if r is None or r.status_code != 200:
        return []
    return parse_cdx_rows(r.text)


def _inflate(blob):
    """gunzip a byte range that may hold >1 member; return decompressed bytes.

    CC warc.gz files are concatenated gzip members, one per record. The range
    may end mid-member, so read incrementally and keep what decoded.
    """
    g = gzip.GzipFile(fileobj=io.BytesIO(blob))
    out = bytearray()
    try:
        while True:
            chunk = g.read(1 << 16)
            if not chunk:
                break
            out += chunk
    except (OSError, EOFError, gzip.BadGzipFile):
        pass
    return bytes(out)


def warc_http_body(rec):
    """Extract the HTTP response body from one WARC/1.x record (bytes->bytes).

    Record = WARC headers, CRLF CRLF, HTTP status line + headers, CRLF CRLF, body.
    """
    i = rec.find(b"\r\n\r\n")
    if i < 0:
        return b""
    http = rec[i + 4:]
    j = http.find(b"\r\n\r\n")
    if j < 0:
        return b""
    hdrs = http[:j].decode("latin-1", "replace")
    body = http[j + 4:]
    m = re.search(r"^Content-Length:\s*(\d+)", hdrs, re.M | re.I)
    if m:
        n = int(m.group(1))
        if 0 < n <= len(body):
            body = body[:n]
    return body


def process_record(row, repo_base):
    """Range-fetch one WARC record, scan body. -> (scanned, keys) or (0,0)."""
    fname, off, ln = row.get("filename"), row.get("offset"), row.get("length")
    url = row.get("url") or ""
    if not fname or off is None or ln is None:
        return 0, 0
    try:
        off, ln = int(off), int(ln)
    except (TypeError, ValueError):
        return 0, 0
    if ln <= 0 or ln > FILE_CAP:
        return 0, 0
    span = max(WARC_RANGE, min(ln * 3 + 4096, FILE_CAP + 65536))
    r = get("https://data.commoncrawl.org/" + fname,
            {"Range": f"bytes={off}-{off + span - 1}"}, timeout=40)
    if r is None or r.status_code not in (200, 206) or not r.content:
        return 0, 0
    blob = r.content
    rec = _inflate(blob)
    if not rec.startswith(b"WARC/"):
        # Some ranges land on an uncompressed record; try raw too.
        rec = blob
    if not rec.startswith(b"WARC/"):
        return 0, 0
    body = warc_http_body(rec)
    if not body:
        return 0, 0
    text = body.decode("utf-8", "replace")[:FILE_CAP]
    n = 0
    for prov, key in scan_text(text):
        if is_junk(key):
            continue
        db_add(key, prov, f"{url} @ {fname}:{off}")
        n += 1
    return 1, n


def cycle(max_items=MAX_ITEMS):
    global total, seen
    base = total
    crawls = fetch_crawls()
    if not crawls:
        log("cycle: no crawl list (CDX gateway down?), skip")
        return
    ci, di, page = load_cursor()
    ci %= len(crawls)
    crawl = crawls[ci]
    domain = SEED_DOMAINS[di % len(SEED_DOMAINS)]
    rows = fetch_cdx(crawl, domain, page, CDX_LIMIT)
    if not rows:
        # advance to next domain (and next crawl when domains are exhausted)
        di += 1
        if di >= len(SEED_DOMAINS):
            di = 0
            ci = (ci + 1) % len(crawls)
        save_cursor(ci, di, 0)
        log(f"cycle: {crawl} {domain} p{page} -> 0 rows, advance")
        return
    cfg = [r for r in rows
           if str(r.get("status")) == "200" and is_config_url(r.get("url", ""))]
    processed = found = 0
    for row in cfg[:max_items]:
        u = row.get("url", "")
        ck = f"{row.get('timestamp')}|{u}"
        if ck in seen:
            continue
        mark_seen(ck)
        try:
            p, f = process_record(row, u)
        except Exception as ex:
            log(f"  rec err {type(ex).__name__}: {ex}")
            p, f = 0, 0
        processed += p
        found += f
        if processed >= max_items:
            break
    page += 1
    if len(rows) < CDX_LIMIT:
        page = 0
        di += 1
        if di >= len(SEED_DOMAINS):
            di = 0
            ci = (ci + 1) % len(crawls)
    save_cursor(ci, di, page)
    log(f"cycle: {crawl} {domain} rows={len(rows)} cfg={len(cfg)} "
        f"scanned={processed} keys=+{found} total={total} -> p{page} d{di}")


def main():
    log("commoncrawl_eternal started")
    once = "--once" in sys.argv
    while True:
        try:
            cycle(5 if once else MAX_ITEMS)
        except Exception as ex:
            log(f"cycle fail {type(ex).__name__}: {ex}")
        if once:
            log(f"--once done total={total}")
            return


if __name__ == "__main__":
    main()