"""Regression gate: /models-catalog 200 must never mean WORKING without key attribution (garbage-key negative control). CDN edges serve byte-identical cached public catalogs (observed: beijing models=519 for 161 unrelated keys). Offline: all HTTP faked. Run: python tests/test_probe_authblind.py DB-level contract: an authblind verdict DEMOTES a stale WORKING row to NEW (the insert path only demotes DEAD/ERR, so the branch must write), leaves an already-NEW row untouched, and never writes DEAD/WORKING/LIMITED (LIMITED is terminal for non-ZAI/KIMI families per revalidate_scanner.py). Rows are seeded as WORKING so the demotion assertion is not vacuous; paired reject->DEAD controls prove the write path is live. """ import os, sys, tempfile _HAS_E = False _fd, _TMPDB = tempfile.mkstemp(prefix="kf_authblind_", suffix=".db") os.close(_fd) os.remove(_TMPDB) # sqlite recreates; unique per run = no lock clash os.environ["KF_DB"] = _TMPDB # explicit, BEFORE any hunt-module import _HERE = os.path.dirname(os.path.abspath(__file__)) ROOT = os.path.dirname(_HERE) if os.path.basename(_HERE) == "tests" else _HERE sys.path.insert(0, ROOT) sys.path.insert(0, os.path.join(ROOT, "src")) # bootstrap: DH/UH import builds PATPool from data/gh_tokens.txt — tree may lack # it (same idiom as test_universal_hunt.py). Dummy file, removed at exit. _DATA = os.path.join(ROOT, "data") os.makedirs(_DATA, exist_ok=True) _TOK = os.path.join(_DATA, "gh_tokens.txt") _CREATED_DUMMY = False if not os.path.exists(_TOK) or os.path.getsize(_TOK) == 0: with open(_TOK, "w") as _f: _f.write("ghp_" + "DUMMYFORTESTS" * 3 + "\n") _CREATED_DUMMY = True import atexit def _cleanup(): if _CREATED_DUMMY and os.path.exists(_TOK): os.remove(_TOK) for suffix in ("", "-wal", "-shm"): try: os.remove(_TMPDB + suffix) except OSError: pass atexit.register(_cleanup) GARB = "PLACEHOLDER-reassigned-after-import" CATALOG = {"data": [{"id": f"m{i}"} for i in range(11)], "models": [{"name": f"m{i}"} for i in range(11)]} class FakeResp: def __init__(self, code, payload=None, text=""): self.status_code = code self.text = text or ("" if payload is None else "json") self._p = payload def json(self): if self._p is None: raise ValueError("no json") return self._p FAILS = [] def check(name, cond, got=None): if cond: print(f" PASS {name}") else: print(f" FAIL {name} got={got!r}") FAILS.append(name) # ---- dashscope_hunt ---- import dashscope_hunt as DH GARB = DH.GARBAGE_KEY # single source of truth def blind_map(url, hdr, params, method="GET"): if method == "POST" or "/models" not in url: return FakeResp(401, None, "invalid") # chat/balance always auth-check return FakeResp(200, CATALOG) # cached catalog served to EVERYONE def authed_map(url, hdr, params, method="GET"): auth = hdr.get("Authorization", "") if GARB in auth: return FakeResp(401, None, '{"error":"invalid"}') return FakeResp(200, CATALOG) def dead_map(url, hdr, params, method="GET"): return FakeResp(401, None, '{"error":"invalid"}') class FakeRequests: def __init__(self, m): self.m = m def get(self, url, headers=None, timeout=None, **kw): return self.m(url, headers or {}, None, "GET") def post(self, url, headers=None, json=None, timeout=None, **kw): return self.m(url, headers or {}, None, "POST") print("== dashscope_hunt ==") DH.requests = FakeRequests(blind_map) st, info = DH.probe_zai("483981f89cf44b66a641bd1c0e87eeee.Ve5Xabc123456") check("probe_zai authblind -> not WORKING", st != "WORKING", (st, info)) st, info = DH.probe_bigmodel("483981f89cf44b66a641bd1c0e87eeee.Ve5Xabc123456") check("probe_bigmodel authblind -> not WORKING", st != "WORKING", (st, info)) st, info = DH.probe_siliconflow("sk-00000000000000000000000000000000") check("probe_siliconflow authblind -> not WORKING", st != "WORKING", (st, info)) st, info = DH.probe_dashscope("sk-00000000000000000000000000000000") check("probe_dashscope authblind -> not WORKING", st != "WORKING", (st, info)) DH.requests = FakeRequests(authed_map) st, info = DH.probe_zai("483981f89cf44b66a641bd1c0e87eeee.Ve5Xabc123456") check("probe_zai attributed 200 -> WORKING", st == "WORKING", (st, info)) st, info = DH.probe_dashscope("sk-00000000000000000000000000000000") check("probe_dashscope attributed 200 -> WORKING", st == "WORKING", (st, info)) DH.requests = FakeRequests(dead_map) st, info = DH.probe_zai("483981f89cf44b66a641bd1c0e87eeee.Ve5Xabc123456") check("probe_zai 401 -> reject/DEAD path (not WORKING)", st != "WORKING", (st, info)) DH.requests = __import__("requests") # ---- universal_hunt ---- print("== universal_hunt ==") import universal_hunt as UH class FakeS: def __init__(self, m): self.m = m def get(self, url, headers=None, params=None, timeout=None, **kw): return self.m(url, headers or {}, params, "GET") def post(self, url, headers=None, json=None, timeout=None, **kw): return self.m(url, headers or {}, None, "POST") def uh_blind(url, hdr, params, method="GET"): if method == "POST" or "/models" not in url: return FakeResp(401, None, "invalid") return FakeResp(200, CATALOG) def uh_authed(url, hdr, params, method="GET"): key = hdr.get("Authorization", "") + hdr.get("x-api-key", "") + str(params or {}) if GARB in key or "AUTHBLIND" in key: return FakeResp(401, None, "invalid") return FakeResp(200, CATALOG) real_s = UH.S UH.S = FakeS(uh_blind) for name, fn, arg in (("probe_openai", UH.probe_openai, "sk-proj-abc123"), ("probe_anthropic", UH.probe_anthropic, "sk-ant-api03-abc"), ("probe_anthropic_oat", UH.probe_anthropic_oat, "sk-ant-oat01-abc"), ("probe_gemini", UH.probe_gemini, "AIzaSyTest1234567890abcdefghij"), ("_models_probe", lambda k: UH._models_probe("test", "https://api.test.dev/v1/models", k), "sk-test-123")): st, info = fn(arg) check(f"{name} authblind -> not WORKING", st != "WORKING", (st, info)) UH.S = FakeS(uh_authed) st, info = UH.probe_anthropic("sk-ant-api03-abc") check("probe_anthropic attributed -> WORKING", st == "WORKING", (st, info)) st, info = UH.probe_gemini("AIzaSyTest1234567890abcdefghij") check("probe_gemini attributed -> WORKING", st == "WORKING", (st, info)) UH.S = real_s # ---- eternal_v10 _v_default loose branches (skip if module absent) ---- try: import eternal_v10 as E _HAS_E = True except ImportError: pass print("== eternal_v10 ==" if _HAS_E else "== eternal_v10 == SKIP (no module)") if _HAS_E: class FakeHTTP: class _sess: def post(self, url, headers=None, json=None, timeout=None, **kw): return FakeResp(200, None, "WAF intercept page") @staticmethod def session(): return FakeHTTP._sess() real_http = E.HTTP E.HTTP = FakeHTTP prov = {"test_url": "https://api.test/v1/chat", "test_model": "m1"} res = E._v_default(prov, "sk-x", "KIMI") check("_v_default non-JSON 200 -> not WORKING", res[0] != "WORKING", res) class FakeHTTPAlt: calls = [0] class _sess: def post(self, url, headers=None, json=None, timeout=None, **kw): FakeHTTPAlt.calls[0] += 1 if FakeHTTPAlt.calls[0] == 1: return FakeResp(401, None, "invalid") return FakeResp(200, None, "not json") @staticmethod def session(): return FakeHTTPAlt._sess() E.HTTP = FakeHTTPAlt prov2 = {"test_url": "https://api.test/v1/chat", "test_model": "m1", "alt_urls": [("ALT", "https://alt.test/v1/chat", "m2")]} res = E._v_default(prov2, "sk-x", "DASHSCOPE") check("_v_default alt non-JSON 200 -> not WORKING", res[0] != "WORKING", res) E.HTTP = real_http # ---- DB-level: validate_key must PRESERVE authblind rows (never DEAD/WORKING) ---- print("== DB-level validate_key ==") import sqlite3 as _sq for _mod in (UH, DH): _mod.conn = _sq.connect(_TMPDB, timeout=30, check_same_thread=False, isolation_level=None) _dl = _mod.conn.execute("PRAGMA database_list").fetchall() _paths = [str(r[2]).replace("\\", "/") for r in _dl] assert any(_TMPDB.replace("\\", "/") == p for p in _paths), \ f"WRONG DB TARGET: {_dl}" print(f" {_mod.__name__} conn -> {_paths}") SCHEMA = ("CREATE TABLE IF NOT EXISTS keys (hash TEXT PRIMARY KEY, key TEXT, " "prov TEXT, status TEXT DEFAULT 'NEW', plan TEXT, remaining TEXT, repo TEXT)") for mod, tag in ((UH, "t_authblind_uh"), (DH, "t_authblind_dh")): mod.conn.execute(SCHEMA) mod.conn.execute("DELETE FROM keys WHERE hash=?", (tag,)) uh_key = "sk-test-groq-key-1234567890" UH.conn.execute("INSERT INTO keys (hash, key, prov, status, repo) VALUES " "('t_authblind_uh',?,'GROQ','WORKING','o/r')", (uh_key,)) UH.S = FakeS(uh_blind) res = UH.validate_key("t_authblind_uh", uh_key, "GROQ", "o/r") row = UH.conn.execute("SELECT status FROM keys WHERE hash='t_authblind_uh'").fetchone() check("UH authblind DEMOTES stale WORKING -> NEW", res[0] == "NEW" and str(res[1]).startswith("authblind:") and row and row[0] == "NEW", (res, row)) res = UH.validate_key("t_authblind_uh", uh_key, "GROQ", "o/r") row = UH.conn.execute("SELECT status FROM keys WHERE hash='t_authblind_uh'").fetchone() check("UH authblind second run: already-NEW row untouched (no-op)", res[0] == "NEW" and row and row[0] == "NEW", (res, row)) UH.S = real_s dh_key = "sk-00000000000000000000000000000000" DH.conn.execute("INSERT INTO keys (hash, key, prov, status, repo) VALUES " "('t_authblind_dh',?,'DASHSCOPE','WORKING','o/r')", (dh_key,)) DH.requests = FakeRequests(blind_map) res = DH.validate_key("t_authblind_dh", dh_key, "DASHSCOPE", "o/r") row = DH.conn.execute("SELECT status FROM keys WHERE hash='t_authblind_dh'").fetchone() check("DH authblind DEMOTES stale WORKING -> NEW", res == ("NEW", "retry:UNVERIFIED_AUTHBLIND") and row and row[0] == "NEW", (res, row)) res = DH.validate_key("t_authblind_dh", dh_key, "DASHSCOPE", "o/r") row = DH.conn.execute("SELECT status FROM keys WHERE hash='t_authblind_dh'").fetchone() check("DH authblind second run: already-NEW row untouched (no-op)", res[0] == "NEW" and row and row[0] == "NEW", (res, row)) DH.requests = __import__("requests") # ---- balance probes: WORKING only on parsed positive number ---- print("== balance probes ==") def bal_map(empty=True, zero=False): def m(url, hdr, params, method="GET"): if "balance" not in url: return FakeResp(401, None, "invalid") if empty: return FakeResp(200, {}, "") v = 0 if zero else 12.5 if "moonshot" in url: return FakeResp(200, {"data": {"balance": v}}) return FakeResp(200, {"balance_infos": [{"total_balance": v}]}) return m for empty, zero, want_dead in ((True, False, True), (False, True, True), (False, False, False)): DH.requests = FakeRequests(bal_map(empty, zero)) st1, i1 = DH.probe_moonshot("sk-97e78a69f0d14b68a1d7c77a4e2d1b29") st2, i2 = DH.probe_deepseek("sk-97e78a69f0d14b68a1d7c77a4e2d1b29") label = "unparsed" if empty else ("zero" if zero else "positive") if want_dead: check(f"probe_moonshot {label} -> not WORKING", st1 != "WORKING", (st1, i1)) check(f"probe_deepseek {label} -> not WORKING", st2 != "WORKING", (st2, i2)) else: check("probe_moonshot positive-balance -> WORKING", st1 == "WORKING", (st1, i1)) check("probe_deepseek positive-balance -> WORKING", st2 == "WORKING", (st2, i2)) DH.requests = __import__("requests") # ---- paired controls: rejected keys must still go DEAD (non-vacuous gate) ---- print("== paired DEAD controls ==") UH.S = FakeS(lambda url, hdr, params, method="GET": FakeResp(401, None, "invalid")) res = UH.validate_key("t_authblind_uh", uh_key, "GROQ", "o/r") row = UH.conn.execute("SELECT status FROM keys WHERE hash='t_authblind_uh'").fetchone() check("UH validate_key rejected -> DEAD row (control)", res[0] == "DEAD" and row and row[0] == "DEAD", (res, row)) UH.S = real_s DH.requests = FakeRequests(dead_map) res = DH.validate_key("t_authblind_dh", dh_key, "DASHSCOPE", "o/r") row = DH.conn.execute("SELECT status FROM keys WHERE hash='t_authblind_dh'").fetchone() check("DH validate_key rejected -> DEAD row (control)", res[0] == "DEAD" and row and row[0] == "DEAD", (res, row)) DH.requests = __import__("requests") print("RESULT:", "PASS all" if not FAILS else f"FAIL {len(FAILS)}: {FAILS}") sys.exit(1 if FAILS else 0)