Handling CAPTCHA and Anti-Bot Measures on State Portals #

import hashlib
import json
import logging
import time
from dataclasses import asdict, dataclass
from enum import Enum
from typing import Callable

import httpx
from playwright.sync_api import Page, sync_playwright

logging.basicConfig(level=logging.INFO, format="%(message)s")
logger = logging.getLogger("captcha_handler")


class ChallengeType(str, Enum):
    NONE = "none"
    HTTP_403 = "http_403"
    HTTP_429 = "http_429"
    DOM_RECAPTCHA = "dom_recaptcha"
    DOM_HCAPTCHA = "dom_hcaptcha"
    BEHAVIORAL_BLOCK = "behavioral_block"


class AntiBotChallenge(Exception):
    """Terminal: an interactive challenge was reached. Never solved in-band."""

    def __init__(self, entity_id: str, kind: ChallengeType) -> None:
        self.entity_id, self.kind = entity_id, kind
        super().__init__(f"{kind.value} on {entity_id}")


@dataclass(frozen=True)
class AuditRecord:
    entity_id: str
    jurisdiction: str
    timestamp: float
    challenge_type: ChallengeType
    resolution_method: str   # direct | headless | escalated_human_review | cache
    response_hash: str
    status_code: int


# Fingerprint consistency: the TLS client and these headers must agree, or a WAF
# injects a silent challenge before the first byte of HTML is parsed.
_HEADERS = {
    "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
    "Accept-Language": "en-US,en;q=0.9",
    "Sec-CH-UA": '"Chromium";v="147", "Not_A Brand";v="8", "Google Chrome";v="147"',
    "Sec-CH-UA-Platform": '"macOS"',
    "User-Agent": (
        "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 "
        "(KHTML, like Gecko) Chrome/147.0.0.0 Safari/537.36"
    ),
}


def _hash(entity_id: str, jurisdiction: str, payload: bytes) -> str:
    # Canonical, tamper-evident provenance for every browser- or API-sourced read.
    return hashlib.sha256(
        f"{entity_id}:{jurisdiction}:{payload.hex()}".encode()
    ).hexdigest()


def _classify(resp: httpx.Response) -> ChallengeType:
    if resp.status_code == 403:
        return ChallengeType.HTTP_403
    if resp.status_code == 429:
        return ChallengeType.HTTP_429
    body = resp.text.lower()
    if "g-recaptcha" in body or "recaptcha" in body:
        return ChallengeType.DOM_RECAPTCHA
    if "h-captcha" in body or "hcaptcha" in body:
        return ChallengeType.DOM_HCAPTCHA
    return ChallengeType.NONE


class CaptchaAwareClient:
    def __init__(
        self,
        jurisdiction: str,
        base_url: str,
        escalate: Callable[[str, ChallengeType], None],
        cache_ttl: int = 300,
    ) -> None:
        self.jurisdiction = jurisdiction
        self.base_url = base_url.rstrip("/")
        self.escalate = escalate          # hand-off to the human-review queue
        self.cache_ttl = cache_ttl
        self._http = httpx.Client(http2=True, timeout=15.0, headers=_HEADERS)
        self._cache: dict[str, tuple[float, bytes]] = {}

    def _log(self, event: str, **fields: object) -> None:
        logger.info(json.dumps({"event": event, "jurisdiction": self.jurisdiction, **fields}))

    def _headless_read(self, url: str, entity_id: str) -> bytes:
        with sync_playwright() as p:
            browser = p.chromium.launch(
                headless=True,
                # Surface automation honestly; do NOT spoof input telemetry to pass scoring.
                args=["--disable-blink-features=AutomationControlled"],
            )
            ctx = browser.new_context(
                viewport={"width": 1280, "height": 800},
                user_agent=_HEADERS["User-Agent"],
                locale="en-US",
                timezone_id="America/New_York",
                ignore_https_errors=False,   # invalid government TLS must fail loudly
            )
            page: Page = ctx.new_page()
            try:
                page.goto(url, wait_until="networkidle", timeout=20_000)
                page.wait_for_timeout(2_000)  # allow deferred challenge injection
                # If an interactive challenge rendered, this is terminal — stop here.
                if page.query_selector("iframe[src*='recaptcha'], .h-captcha, #challenge-running"):
                    raise AntiBotChallenge(entity_id, ChallengeType.BEHAVIORAL_BLOCK)
                return page.content().encode("utf-8")
            finally:
                ctx.close()
                browser.close()

    def fetch(self, entity_id: str) -> AuditRecord:
        url = f"{self.base_url}/entity/{entity_id}"
        cached = self._cache.get(entity_id)
        if cached and time.time() - cached[0] < self.cache_ttl:
            self._log("cache_hit", entity_id=entity_id)
            return AuditRecord(entity_id, self.jurisdiction, time.time(),
                               ChallengeType.NONE, "cache",
                               _hash(entity_id, self.jurisdiction, cached[1]), 200)

        resp = self._http.get(url)
        kind = _classify(resp)

        if kind is ChallengeType.NONE:
            payload, method = resp.content, "direct"
        else:
            # Any anti-bot signal invalidates a cached read and escalates the tier.
            self._cache.pop(entity_id, None)
            self._log("challenge_detected", entity_id=entity_id,
                      challenge_type=kind.value, status_code=resp.status_code)
            try:
                payload, method = self._headless_read(url, entity_id), "headless"
            except AntiBotChallenge as exc:
                # Terminal: log, escalate to human review, never solve in-band.
                self._log("captcha_terminal", entity_id=entity_id, challenge_type=exc.kind.value)
                self.escalate(entity_id, exc.kind)
                return AuditRecord(entity_id, self.jurisdiction, time.time(),
                                   exc.kind, "escalated_human_review",
                                   _hash(entity_id, self.jurisdiction, b""), 511)

        self._cache[entity_id] = (time.time(), payload)
        record = AuditRecord(entity_id, self.jurisdiction, time.time(), kind, method,
                             _hash(entity_id, self.jurisdiction, payload), 200)
        self._log("audit_record", **{k: (v.value if isinstance(v, Enum) else v)
                                      for k, v in asdict(record).items()})
        return record