saksham
1a4c80958f
FastAPI + Celery + Next.js + Postgres/Redis app with company monitoring, source collection, LLM-based change analysis, enrichment, and account security (Turnstile, escalating lockout, email verification).
72 lines
3.0 KiB
Python
72 lines
3.0 KiB
Python
"""Cloudflare Turnstile server-side verification. Canonical siteverify
|
|
contract per Cloudflare's own reference: POST {secret, response, remoteip}
|
|
to the fixed challenges.cloudflare.com endpoint, check `success === true`.
|
|
Fails closed on any network error or non-2xx response - a Cloudflare outage
|
|
must never silently let requests through unverified.
|
|
|
|
One deliberate exception: a misconfigured *secret* (typo'd/invalid, as
|
|
opposed to a genuinely bad/expired user token) fails open instead. Cloudflare
|
|
reports this distinctly via `error-codes` (`invalid-input-secret` /
|
|
`missing-input-secret`) rather than as an ambiguous non-2xx/network failure,
|
|
so it's a real, detectable "the admin's config is broken" signal, not "we
|
|
couldn't tell if this passed." Locking out every real register/login/
|
|
password-reset attempt because of an admin's own copy-paste mistake is a
|
|
worse outcome than briefly running with reduced bot protection - especially
|
|
since Turnstile is one layer among several here (see SECURITY.md's IP
|
|
throttle/ban and account-lockout layers, which stay fully active either
|
|
way).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import httpx
|
|
|
|
from app.core.config import Settings
|
|
from app.core.logging import get_logger
|
|
|
|
logger = get_logger(__name__)
|
|
|
|
_SITEVERIFY_URL = "https://challenges.cloudflare.com/turnstile/v0/siteverify"
|
|
|
|
# Cloudflare's own error-code vocabulary for a broken *secret* - distinct
|
|
# from user-token-level codes like invalid-input-response/timeout-or-
|
|
# duplicate/missing-input-response, which are legitimate rejections and
|
|
# must keep failing closed.
|
|
_SECRET_MISCONFIGURED_CODES = {"invalid-input-secret", "missing-input-secret"}
|
|
|
|
|
|
async def verify_turnstile(token: str, remote_ip: str, settings: Settings) -> bool:
|
|
try:
|
|
async with httpx.AsyncClient(timeout=10) as client:
|
|
response = await client.post(
|
|
_SITEVERIFY_URL,
|
|
data={
|
|
"secret": settings.turnstile_secret,
|
|
"response": token,
|
|
"remoteip": remote_ip,
|
|
},
|
|
)
|
|
response.raise_for_status()
|
|
data = response.json()
|
|
if data.get("success") is True:
|
|
return True
|
|
|
|
error_codes = set(data.get("error-codes") or [])
|
|
if error_codes & _SECRET_MISCONFIGURED_CODES:
|
|
logger.error(
|
|
"turnstile_secret_misconfigured",
|
|
error_codes=sorted(error_codes),
|
|
)
|
|
return True # fail open - this is a config problem, not the caller's
|
|
return False
|
|
except Exception as exc: # noqa: BLE001 - fail closed on any error
|
|
logger.warning("turnstile_verify_failed", error=str(exc))
|
|
return False
|
|
|
|
|
|
def turnstile_required(is_localhost: bool, settings: Settings) -> bool:
|
|
"""Skipped entirely for a loopback caller, or when no secret is
|
|
configured at all (matches this app's usual optional-provider
|
|
convention - e.g. NinjaPear/USPTO/Brave all no-op when unset)."""
|
|
return not is_localhost and bool(settings.turnstile_secret)
|