diff --git a/app/verify/cli.py b/app/verify/cli.py index 2235da7..8c6a6a0 100644 --- a/app/verify/cli.py +++ b/app/verify/cli.py @@ -453,11 +453,22 @@ def cmd_check_urls(args: argparse.Namespace) -> int: def _summarize_cache(cache: dict[str, dict[str, Any]], targets: list[str]) -> None: from collections import Counter alive = sum(1 for u in targets if cache.get(u, {}).get("alive")) - dead = sum(1 for u in targets if u in cache and not cache[u].get("alive")) - print(f"\nliveness over {len(targets)} targeted URL(s): {alive} alive, {dead} dead") + indeterminate = sum( + 1 for u in targets if http_check.is_automation_challenge(cache.get(u, {})) + ) + dead = sum( + 1 for u in targets + if u in cache and not cache[u].get("alive") + and not http_check.is_automation_challenge(cache[u]) + ) + print( + f"\nliveness over {len(targets)} targeted URL(s): {alive} alive, {dead} dead, " + f"{indeterminate} automation-challenged" + ) reasons = Counter( cache[u].get("reason") for u in targets if u in cache and not cache[u].get("alive") + and not http_check.is_automation_challenge(cache[u]) ) if reasons: print("dead reasons:") @@ -619,10 +630,17 @@ def cmd_pr(args: argparse.Namespace) -> int: except Exception as exc: # network hiccup must not sink the report print(f"_Tier 1 skipped: {exc}_\n") alive = sum(1 for e in url_cache.values() if e.get("alive")) - dead = len(url_cache) - alive + indeterminate = sum(1 for e in url_cache.values() if http_check.is_automation_challenge(e)) + dead = len(url_cache) - alive - indeterminate print("### Tier 1 — source-URL liveness (changed)\n") - print(f"Checked **{len(url_cache)}** unique URL(s): **{alive} alive**, **{dead} dead**.\n") - dead_reasons = Counter(e["reason"] for e in url_cache.values() if not e.get("alive")) + print( + f"Checked **{len(url_cache)}** unique URL(s): **{alive} alive**, " + f"**{dead} dead**, **{indeterminate} automation-challenged**.\n" + ) + dead_reasons = Counter( + e["reason"] for e in url_cache.values() + if not e.get("alive") and not http_check.is_automation_challenge(e) + ) if dead_reasons: print("| Dead reason | Count |") print("| --- | ---: |") @@ -632,13 +650,13 @@ def cmd_pr(args: argparse.Namespace) -> int: # Tier 2 — external cross-reference (network, exact-heading only). fetcher = crossref.WikidataFetcher() - xref: dict[str, str] = {} + xref: dict[tuple[str, str], str] = {} decisions: Counter[str] = Counter() for r, _ in scored: try: xres = crossref.crossref_record(r.data, fetcher) if r.slug: - xref[r.slug] = xres.decision + xref[(r.category, r.slug)] = xres.decision decisions[xres.decision] += 1 except Exception: decisions["error"] += 1 @@ -655,8 +673,12 @@ def cmd_pr(args: argparse.Namespace) -> int: hold = 0 for r, s in scored: urls_r = [u for u in r.data.get("source_urls", []) if isinstance(u, str)] - dec = promote.decide(band=s.band, source_urls=urls_r, url_cache=url_cache, - crossref_decision=xref.get(r.slug) if r.slug else None) + dec = promote.decide( + band=s.band, + source_urls=urls_r, + url_cache=url_cache, + crossref_decision=xref.get((r.category, r.slug)) if r.slug else None, + ) if dec.promote: promote_rows.append((r, dec.reason)) else: diff --git a/app/verify/http_check.py b/app/verify/http_check.py index b0a8d60..271a98a 100644 --- a/app/verify/http_check.py +++ b/app/verify/http_check.py @@ -36,6 +36,7 @@ # nothing about whether the page exists, so these answers must never be cached as # a verdict — otherwise one impatient run marks a whole host dead for a TTL. TRANSIENT_STATUSES = frozenset({429, 503}) +AUTOMATION_CHALLENGE_REASON = "automation-challenge" RETRY_ATTEMPTS = 3 RETRY_BACKOFF_S = (2.0, 6.0) MAX_RETRY_AFTER_S = 15.0 @@ -58,6 +59,11 @@ class CheckResult(NamedTuple): def transient(self) -> bool: return self.status in TRANSIENT_STATUSES + @property + def indeterminate(self) -> bool: + """True when the endpoint reached an anti-bot challenge, not a page verdict.""" + return self.reason == AUTOMATION_CHALLENGE_REASON + # --- opener abstraction (injectable for tests) ----------------------------------- @@ -102,7 +108,15 @@ def _is_homepage_redirect(original: str, final: str) -> bool: return _path_depth(original) >= 1 and _path_depth(final) == 0 -def classify(original_url: str, status: int | None, final_url: str | None) -> tuple[bool, str]: +def classify( + original_url: str, + status: int | None, + final_url: str | None, + *, + automation_challenge: bool = False, +) -> tuple[bool, str]: + if automation_challenge: + return False, AUTOMATION_CHALLENGE_REASON if status is None: return False, "error" if status >= 400: @@ -122,11 +136,34 @@ def _retry_after_seconds(exc: Exception) -> float | None: return None # HTTP-date form; fall back to our own backoff -def _attempt(url: str, opener: Any) -> tuple[int | None, str | None, float | None]: - """One HEAD-then-GET pass. Returns (status, final_url, retry_after).""" +def _is_automation_challenge(exc: Exception, code: int) -> bool: + """Recognize an explicit anti-bot challenge without weakening generic 403s. + + A September 2026 probe with this module's User-Agent found + ``browser.geekbench.com`` returning ``403`` with + ``CF-Mitigated: challenge`` and Cloudflare's challenge page. That proves the + request reached the cited host but not whether an unauthenticated client may + inspect the page. Keep ordinary 401/403 responses as dead: an HTTP status + alone cannot distinguish a missing or forbidden page from bot protection. + """ + if code != 403: + return False + headers = getattr(exc, "headers", None) + mitigated = headers.get("CF-Mitigated", "") if headers is not None else "" + return isinstance(mitigated, str) and mitigated.lower() == "challenge" + + +def _attempt(url: str, opener: Any) -> tuple[int | None, str | None, float | None, bool]: + """One HEAD-then-GET pass. + + Returns ``(status, final_url, retry_after, automation_challenge)``. The last + value is intentionally based on an explicit gateway signal, never on a bare + 403, so unavailable pages cannot become promotable merely by returning 403. + """ status: int | None = None final: str | None = None retry_after: float | None = None + automation_challenge = False for method in ("HEAD", "GET"): try: status, final = opener.open(url, method) @@ -138,11 +175,12 @@ def _attempt(url: str, opener: Any) -> tuple[int | None, str | None, float | Non if isinstance(code, int): status, final = code, getattr(exc, "url", None) or url retry_after = _retry_after_seconds(exc) + automation_challenge = _is_automation_challenge(exc, code) if method == "HEAD" and code in (400, 403, 405, 501): continue break - status, final = None, None - return status, final, retry_after + status, final, automation_challenge = None, None, False + return status, final, retry_after, automation_challenge def check_one( @@ -154,15 +192,18 @@ def check_one( to wait, not that the page is gone. """ status = final = retry_after = None + automation_challenge = False for attempt in range(RETRY_ATTEMPTS): - status, final, retry_after = _attempt(url, opener) + status, final, retry_after, automation_challenge = _attempt(url, opener) if status not in TRANSIENT_STATUSES: break if on_rate_limit is not None: on_rate_limit(host_of(url)) if attempt < RETRY_ATTEMPTS - 1: time.sleep(retry_after if retry_after is not None else RETRY_BACKOFF_S[attempt]) - alive, reason = classify(url, status, final) + alive, reason = classify( + url, status, final, automation_challenge=automation_challenge + ) return CheckResult(url, status, final, alive, reason) @@ -247,13 +288,20 @@ def _task(url: str) -> CheckResult: # --- cache ----------------------------------------------------------------------- +def is_automation_challenge(entry: dict[str, Any]) -> bool: + """Whether a cached result reached an explicit anti-bot challenge.""" + return entry.get("reason") == AUTOMATION_CHALLENGE_REASON + + def load_cache(path: Path = URL_CACHE_PATH) -> dict[str, dict[str, Any]]: """Load the cache, dropping rate-limit answers written by older runs. A 429/503 is not a verdict, so an entry holding one is not a cache hit — it is a URL we still have to check. Filtering on load heals a cache that a previous run poisoned (3,998 GSMArena pages were parked as dead this way, - all of which answer 200 when asked at a civil pace). + all of which answer 200 when asked at a civil pace). Explicit anti-bot + challenges stay cached as *indeterminate*: the promotion layer can use that + reachability signal only for an already-authoritative cited host. """ return { e["url"]: e @@ -292,7 +340,7 @@ def result_to_entry(r: CheckResult, ts: str) -> dict[str, Any]: def record_liveness(source_urls: list[str], cache: dict[str, dict[str, Any]]) -> tuple[int, int]: - """(#live, #dead) for a record's URLs that are present in the cache.""" + """(#live, #dead) for cached URLs, excluding anti-bot indeterminate results.""" live = dead = 0 for u in source_urls: e = cache.get(u) @@ -300,6 +348,6 @@ def record_liveness(source_urls: list[str], cache: dict[str, dict[str, Any]]) -> continue if e.get("alive"): live += 1 - else: + elif not is_automation_challenge(e): dead += 1 return live, dead diff --git a/app/verify/promote.py b/app/verify/promote.py index 1466916..81723a6 100644 --- a/app/verify/promote.py +++ b/app/verify/promote.py @@ -19,7 +19,7 @@ from pathlib import Path from typing import Any, NamedTuple -from . import hosts +from . import hosts, http_check from .common import STATE_DIR CROSSREF_CACHE_PATH = STATE_DIR / "crossref_cache.jsonl" @@ -36,16 +36,27 @@ class PromotionDecision(NamedTuple): def has_live_authoritative_source( source_urls: list[str], url_cache: dict[str, dict[str, Any]] ) -> bool: - """True if some cited URL is an authoritative host (Tier 1 *or* Tier 2) AND - confirmed alive. The green band already requires a T1/T2 source + completeness - + consistency; this just adds "and that source actually resolves". Requiring a - *manufacturer/encyclopaedia* (T1 only) was too strict — it never promoted the - many green records sourced from reputable spec/benchmark DBs (gsmarena, - cpubenchmark, ...), so verified never moved off its floor. + """True if an authoritative cited URL has a usable liveness signal. + + A normal alive response always qualifies. An explicit anti-bot challenge also + qualifies, but only for an already classified Tier-1/Tier-2 host: it proves + the request reached that host while the gateway withheld page inspection. + This preserves the live-source gate without treating every 403 as a live + record. Generic errors, ordinary 401/403, redirects to a homepage, and + unclassified hosts remain insufficient for promotion. + + This distinction is deliberate. The September 2026 probe found Geekbench's + Cloudflare challenge under the verifier User-Agent, while several other + Tier-2 databases returned normal 200 responses. Treating challenge responses + as dead made a host-wide automation policy look like thousands of dead + citations; dropping the liveness gate altogether made unchecked citations + promotable. This narrow fallback avoids both failure modes. """ for u in source_urls: entry = url_cache.get(u) - if entry and entry.get("alive") and hosts.tier_of_host(hosts.host_of(u)) in (1, 2): + if not entry or hosts.tier_of_host(hosts.host_of(u)) not in (1, 2): + continue + if entry.get("alive") or http_check.is_automation_challenge(entry): return True return False @@ -66,14 +77,10 @@ def decide( # Reality confirm: external source agrees -> strongest promotion. if crossref_decision == "confirm": return PromotionDecision(True, "crossref-confirm") - # green = passed the offline inspection: an authoritative source (T1/T2) IS - # cited, cross-field consistency holds, the record is complete, and there are - # no hard violations. That is the verification result, so promote it (unless - # crossref contradicted above). The earlier live-HTTP source re-check was - # dropped: most green records cite spec DBs (phonedb/cpubenchmark) that block - # automated requests, which kept verified stuck far below the green band. - if band == "green": - return PromotionDecision(True, "green") + # green is only an offline candidate. Promotion still requires a cited + # authoritative source to have been confirmed alive by Tier 1. + if band == "green" and has_live_authoritative_source(source_urls, url_cache): + return PromotionDecision(True, "green-live-source") return PromotionDecision(False, "needs-confirmation") diff --git a/tests/verify/test_http_check.py b/tests/verify/test_http_check.py index 6ce05ad..6f3debb 100644 --- a/tests/verify/test_http_check.py +++ b/tests/verify/test_http_check.py @@ -92,6 +92,14 @@ def test_record_liveness(): assert http_check.record_liveness(["a", "b", "c", "missing"], cache) == (2, 1) +def test_record_liveness_excludes_automation_challenge(): + cache = { + "blocked": {"alive": False, "reason": http_check.AUTOMATION_CHALLENGE_REASON}, + "dead": {"alive": False, "reason": "http-404"}, + } + assert http_check.record_liveness(["blocked", "dead"], cache) == (0, 1) + + def test_cache_roundtrip(): # tmp_path fixture is unreliable on this Windows runner; use a local scratch file. from pathlib import Path @@ -115,6 +123,16 @@ def __init__(self, url, retry_after=None): self.headers = {"Retry-After": retry_after} if retry_after else {} +class _Http403Challenge(Exception): + """urllib-shaped Cloudflare bot challenge observed from Geekbench.""" + + def __init__(self, url): + super().__init__("Forbidden") + self.code = 403 + self.url = url + self.headers = {"CF-Mitigated": "challenge"} + + class FlakyOpener(FakeOpener): """Rate-limits the first ``fail_times`` calls, then answers normally.""" @@ -146,6 +164,27 @@ def test_persistent_rate_limit_is_transient_not_dead(monkeypatch): assert res.status == 429 and res.transient # caller must not cache this as a verdict +def test_cloudflare_bot_challenge_is_indeterminate_not_dead(): + url = "https://browser.geekbench.com/v6/cpu/1" + op = FakeOpener({url: _Http403Challenge(url)}) + res = http_check.check_one(url, op) + assert res.status == 403 + assert not res.alive + assert res.indeterminate + assert not res.transient + assert res.reason == http_check.AUTOMATION_CHALLENGE_REASON + assert [method for _url, method in op.calls] == ["HEAD", "GET"] + + +def test_generic_403_stays_dead(): + err = type("E", (Exception,), {"code": 403, "url": None, "headers": {}})() + url = "https://example.com/private" + res = http_check.check_one(url, FakeOpener({url: err})) + assert not res.alive + assert not res.indeterminate + assert res.reason == "http-403" + + def test_rate_limit_slows_the_host_down(monkeypatch): monkeypatch.setattr(http_check.time, "sleep", lambda _s: None) url = "https://www.gsmarena.com/z-3.php" diff --git a/tests/verify/test_promote_crossref.py b/tests/verify/test_promote_crossref.py index fa155c7..fed71b5 100644 --- a/tests/verify/test_promote_crossref.py +++ b/tests/verify/test_promote_crossref.py @@ -113,21 +113,66 @@ def test_write_back_atomic_lf_preserved(): # --- promotion decision ---------------------------------------------------------- -def test_green_promotes(): - # green = passed the offline inspection -> verified. The live-source state no - # longer gates (green already requires a cited authoritative source). - d = promote.decide(band="green", source_urls=[], url_cache={}, crossref_decision=None) - assert d.promote and d.reason == "green" +def test_green_requires_live_authoritative_source(): + d = promote.decide( + band="green", + source_urls=["https://www.intel.com/x"], + url_cache={}, + crossref_decision=None, + ) + assert not d.promote + + +def test_green_promotes_with_live_authoritative_source(): + cache = {"https://www.phonedb.net/x": {"alive": True}} + d = promote.decide( + band="green", source_urls=["https://www.phonedb.net/x"], + url_cache=cache, crossref_decision=None, + ) + assert d.promote and d.reason == "green-live-source" -def test_green_promotes_regardless_of_source_liveness(): - # Even with a dead / unchecked source, a green record promotes. +def test_green_does_not_promote_with_dead_authoritative_source(): cache = {"https://www.phonedb.net/x": {"alive": False}} d = promote.decide( band="green", source_urls=["https://www.phonedb.net/x"], url_cache=cache, crossref_decision=None, ) - assert d.promote and d.reason == "green" + assert not d.promote + + +def test_green_promotes_with_authoritative_automation_challenge(): + url = "https://browser.geekbench.com/v6/cpu/1" + d = promote.decide( + band="green", + source_urls=[url], + url_cache={ + url: { + "alive": False, + "status": 403, + "reason": "automation-challenge", + } + }, + crossref_decision=None, + ) + assert d.promote and d.reason == "green-live-source" + + +def test_green_does_not_promote_with_unclassified_automation_challenge(): + url = "https://example.com/specs/1" + d = promote.decide( + band="green", + source_urls=[url], + url_cache={ + url: { + "alive": False, + "status": 403, + "reason": "automation-challenge", + } + }, + crossref_decision=None, + ) + assert not d.promote def test_yellow_without_confirm_holds():