Files
chicken_babies_site/app/services/hcaptcha.py
Phillip Tarrant d9090f5055 feat: phase 5 contact form — hCaptcha, honeypot, rate limit, notify
Working /contact POST flow: honeypot → hCaptcha server-verify →
field validation → SlowAPI 3/hr IP rate limit → contact_submissions
row → best-effort Resend notification (Reply-To = submitter) →
generic success page. Spam paths don't persist and render the same
success page (anti-enumeration). Send failures don't break the
request path — the row is already durable.

New services: HCaptchaService (async httpx + dev fallback),
ContactService. EmailService gains send_contact_notification.
Production config validator now requires ADMIN_CONTACT_EMAIL,
HCAPTCHA_SECRET, HCAPTCHA_SITE_KEY. 23 new tests, all green.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-04-22 06:47:06 -05:00

133 lines
4.8 KiB
Python

"""hCaptcha server-side verification.
Wraps the hCaptcha ``siteverify`` endpoint with a dev-mode bypass so
local work does not require a real site-key / secret pair.
Security and UX rules
---------------------
- **Never raise from the request path.** The caller (``/contact``)
treats a ``False`` return as "reject the submission as spam" and a
``True`` return as "continue to validation". Any network hiccup /
non-200 / malformed-JSON surfaces as ``False`` — it is safer to drop
a legitimate visitor's submission than to accept a spam flood if
hCaptcha is temporarily unavailable.
- **Dev fallback:** when ``settings.hcaptcha_secret`` is falsy we log a
structured ``hcaptcha_dev_fallback`` event at INFO and return
``True``. The production config validator refuses to boot without a
secret, so this path only runs locally.
- Raw tokens are single-use, short-lived, and bound to the submitter's
session — we do not persist or log them.
"""
from __future__ import annotations
from typing import Any, Optional
import httpx
import structlog
from app.config import Settings
_log = structlog.get_logger(__name__)
# hCaptcha's server-side verify endpoint. Documented at
# https://docs.hcaptcha.com/#verify-the-user-response-server-side.
_SITEVERIFY_URL: str = "https://hcaptcha.com/siteverify"
# Bounded network timeout. hCaptcha's own advice is 5s; longer would
# block the request path unnecessarily under an incident.
_TIMEOUT_SECONDS: float = 5.0
class HCaptchaService:
"""Verify hCaptcha responses server-side.
Usage
-----
``await hcaptcha_service.verify(token, remote_ip)`` returns a bool.
``True`` means "treat the request as human"; ``False`` means "reject".
"""
def __init__(self, settings: Settings) -> None:
"""Store settings by reference so rotation at runtime works."""
self._settings: Settings = settings
async def verify(self, token: str, remote_ip: str) -> bool:
"""Verify ``token`` against hCaptcha; return True on success.
Behavior
--------
- If ``settings.hcaptcha_secret`` is falsy → log
``hcaptcha_dev_fallback`` and return ``True`` (dev).
- Otherwise POST to ``/siteverify`` with a 5s timeout.
- Non-200, network error, malformed JSON, or ``success=False``
all return ``False``.
The method never raises — errors are logged and converted to
``False`` so the caller can render the generic thank-you page
without leaking internal state.
"""
secret: Optional[str] = self._settings.hcaptcha_secret
if not secret:
# Dev bypass. The config validator forbids this in
# production, so reaching this branch is always a dev/test
# condition and safe to trust.
_log.info("hcaptcha_dev_fallback", remote_ip=remote_ip or "")
return True
payload = {
"secret": secret,
"response": token or "",
"remoteip": remote_ip or "",
}
data = await self._post_siteverify(payload)
if data is None:
# Timeout / transport error / non-200 / parse failure —
# already logged by the helper. Treat as spam.
return False
success = bool(data.get("success", False))
if not success:
# Log error codes if present so operators can diagnose
# misconfigured keys without exposing the token.
_log.info(
"hcaptcha_verify_failed",
error_codes=list(data.get("error-codes") or []),
)
return success
# ------------------------------------------------------------------
# Internals
# ------------------------------------------------------------------
async def _post_siteverify(self, payload: dict) -> Optional[dict[str, Any]]:
"""POST to hCaptcha's verify endpoint and return parsed JSON.
Returns ``None`` on any failure (timeout, non-200, transport
error, malformed JSON). Kept separate from :meth:`verify` so
tests can monkeypatch the HTTP boundary without touching the
decision logic.
"""
try:
async with httpx.AsyncClient(timeout=_TIMEOUT_SECONDS) as client:
resp = await client.post(_SITEVERIFY_URL, data=payload)
except httpx.HTTPError:
# Network error / timeout. Do not raise; do not log the
# payload (contains the secret).
_log.exception("hcaptcha_request_failed")
return None
if resp.status_code != 200:
_log.info(
"hcaptcha_non_200",
status_code=resp.status_code,
)
return None
try:
return resp.json()
except ValueError:
_log.exception("hcaptcha_malformed_json")
return None