Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
40 changes: 31 additions & 9 deletions app/verify/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:")
Expand Down Expand Up @@ -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("| --- | ---: |")
Expand All @@ -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
Expand All @@ -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:
Expand Down
68 changes: 58 additions & 10 deletions app/verify/http_check.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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) -----------------------------------

Expand Down Expand Up @@ -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:
Expand All @@ -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)
Expand All @@ -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(
Expand All @@ -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)


Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -292,14 +340,14 @@ 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)
if e is None:
continue
if e.get("alive"):
live += 1
else:
elif not is_automation_challenge(e):
dead += 1
return live, dead
39 changes: 23 additions & 16 deletions app/verify/promote.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand All @@ -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

Expand All @@ -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")


Expand Down
39 changes: 39 additions & 0 deletions tests/verify/test_http_check.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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."""

Expand Down Expand Up @@ -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"
Expand Down
Loading
Loading