Compare commits
2 Commits
67c848f329
...
1e5e3252c6
| Author | SHA1 | Date | |
|---|---|---|---|
| 1e5e3252c6 | |||
| d9090f5055 |
@@ -183,11 +183,22 @@ class Settings(BaseSettings):
|
|||||||
missing.append("RESEND_FROM")
|
missing.append("RESEND_FROM")
|
||||||
if not self.admin_emails or not self.admin_emails_list:
|
if not self.admin_emails or not self.admin_emails_list:
|
||||||
missing.append("ADMIN_EMAILS")
|
missing.append("ADMIN_EMAILS")
|
||||||
|
# Phase 5: the public contact form needs a destination inbox
|
||||||
|
# and a real hCaptcha key pair. Missing any of these would
|
||||||
|
# either drop submissions on the floor (no recipient) or open
|
||||||
|
# the form to unverified bot traffic (no captcha), so we
|
||||||
|
# enforce them at boot.
|
||||||
|
if not self.admin_contact_email:
|
||||||
|
missing.append("ADMIN_CONTACT_EMAIL")
|
||||||
|
if not self.hcaptcha_secret:
|
||||||
|
missing.append("HCAPTCHA_SECRET")
|
||||||
|
if not self.hcaptcha_site_key:
|
||||||
|
missing.append("HCAPTCHA_SITE_KEY")
|
||||||
if missing:
|
if missing:
|
||||||
raise ValueError(
|
raise ValueError(
|
||||||
"Production configuration is missing required values: "
|
"Production configuration is missing required values: "
|
||||||
+ ", ".join(missing)
|
+ ", ".join(missing)
|
||||||
+ ". These are needed for magic-link admin auth."
|
+ ". These are needed for admin auth and the contact form."
|
||||||
)
|
)
|
||||||
return self
|
return self
|
||||||
|
|
||||||
|
|||||||
16
app/main.py
16
app/main.py
@@ -66,8 +66,10 @@ from app.services.admin_pages import AdminPagesService
|
|||||||
from app.services.admin_posts import AdminPostsService
|
from app.services.admin_posts import AdminPostsService
|
||||||
from app.services.audit import AuditService
|
from app.services.audit import AuditService
|
||||||
from app.services.auth import AuthService
|
from app.services.auth import AuthService
|
||||||
|
from app.services.contact import ContactService
|
||||||
from app.services.csrf import CSRF_COOKIE_NAME, CSRFService
|
from app.services.csrf import CSRF_COOKIE_NAME, CSRFService
|
||||||
from app.services.email import EmailService
|
from app.services.email import EmailService
|
||||||
|
from app.services.hcaptcha import HCaptchaService
|
||||||
from app.services.markdown import MarkdownService
|
from app.services.markdown import MarkdownService
|
||||||
from app.services.media import MediaService
|
from app.services.media import MediaService
|
||||||
from app.services.pages import PageService
|
from app.services.pages import PageService
|
||||||
@@ -236,6 +238,20 @@ def create_app() -> FastAPI:
|
|||||||
application.state.session_service = session_service
|
application.state.session_service = session_service
|
||||||
application.state.auth_service = auth_service
|
application.state.auth_service = auth_service
|
||||||
|
|
||||||
|
# --- Phase 5 wiring -----------------------------------------------------
|
||||||
|
# Public contact form. HCaptchaService has no DB; ContactService
|
||||||
|
# persists via the shared engine and delegates email to the
|
||||||
|
# existing EmailService so the dev-fallback semantics match.
|
||||||
|
hcaptcha_service = HCaptchaService(settings)
|
||||||
|
contact_service = ContactService(
|
||||||
|
engine=engine,
|
||||||
|
email=email_service,
|
||||||
|
audit=audit_service,
|
||||||
|
settings=settings,
|
||||||
|
)
|
||||||
|
application.state.hcaptcha_service = hcaptcha_service
|
||||||
|
application.state.contact_service = contact_service
|
||||||
|
|
||||||
# --- Phase 4 wiring -----------------------------------------------------
|
# --- Phase 4 wiring -----------------------------------------------------
|
||||||
# CSRF signer: separate salt so a session cookie never validates
|
# CSRF signer: separate salt so a session cookie never validates
|
||||||
# as a CSRF token (domain separation via salt).
|
# as a CSRF token (domain separation via salt).
|
||||||
|
|||||||
@@ -7,7 +7,9 @@ Phase 2 scope:
|
|||||||
- ``GET /about`` — DB-backed; loads the ``about`` row from the
|
- ``GET /about`` — DB-backed; loads the ``about`` row from the
|
||||||
``pages`` table via :class:`PageService` and
|
``pages`` table via :class:`PageService` and
|
||||||
renders its ``body_html_cached`` directly.
|
renders its ``body_html_cached`` directly.
|
||||||
- ``GET /contact`` — inert contact form UI + optional ``mailto:`` link.
|
- ``GET /contact`` — form UI + optional ``mailto:`` link.
|
||||||
|
- ``POST /contact`` — Phase 5 live submission endpoint (hCaptcha +
|
||||||
|
honeypot + rate-limit + persist + notify).
|
||||||
- ``GET /shop`` — "Coming soon" card.
|
- ``GET /shop`` — "Coming soon" card.
|
||||||
|
|
||||||
Every handler is thin: it resolves its dependencies, calls any service
|
Every handler is thin: it resolves its dependencies, calls any service
|
||||||
@@ -17,16 +19,21 @@ is constructed in Python.
|
|||||||
|
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import re
|
||||||
|
|
||||||
import structlog
|
import structlog
|
||||||
from fastapi import APIRouter, Depends, HTTPException, Request
|
from fastapi import APIRouter, Depends, Form, HTTPException, Request
|
||||||
from fastapi.responses import HTMLResponse
|
from fastapi.responses import HTMLResponse, Response
|
||||||
from fastapi.templating import Jinja2Templates
|
from fastapi.templating import Jinja2Templates
|
||||||
|
|
||||||
from app.config import Settings, get_settings
|
from app.config import Settings, get_settings
|
||||||
from app.models.entities import Page
|
from app.models.entities import Page
|
||||||
from app.models.posts import PostSummary
|
from app.models.posts import PostSummary
|
||||||
|
from app.services.contact import ContactService
|
||||||
|
from app.services.hcaptcha import HCaptchaService
|
||||||
from app.services.pages import PageService, get_page_service
|
from app.services.pages import PageService, get_page_service
|
||||||
from app.services.posts import PostService, get_post_service
|
from app.services.posts import PostService, get_post_service
|
||||||
|
from app.services.rate_limit import limiter
|
||||||
|
|
||||||
|
|
||||||
# Module-level router. Mounted without a prefix by ``app.main.create_app``
|
# Module-level router. Mounted without a prefix by ``app.main.create_app``
|
||||||
@@ -37,6 +44,19 @@ router: APIRouter = APIRouter(tags=["public"])
|
|||||||
_log = structlog.get_logger(__name__)
|
_log = structlog.get_logger(__name__)
|
||||||
|
|
||||||
|
|
||||||
|
# Same loose email shape used on the admin login form. Intentionally
|
||||||
|
# permissive: the mapper layer / Resend does the real work; we only
|
||||||
|
# want to reject obvious junk before hitting the service.
|
||||||
|
_EMAIL_RE: re.Pattern[str] = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$")
|
||||||
|
|
||||||
|
# Field length guards — matched on the server regardless of the
|
||||||
|
# HTML5 attributes the template emits.
|
||||||
|
_NAME_MAX: int = 80
|
||||||
|
_EMAIL_MAX: int = 254
|
||||||
|
_MESSAGE_MIN: int = 10
|
||||||
|
_MESSAGE_MAX: int = 4000
|
||||||
|
|
||||||
|
|
||||||
def get_templates(request: Request) -> Jinja2Templates:
|
def get_templates(request: Request) -> Jinja2Templates:
|
||||||
"""Return the shared :class:`Jinja2Templates` instance.
|
"""Return the shared :class:`Jinja2Templates` instance.
|
||||||
|
|
||||||
@@ -49,6 +69,31 @@ def get_templates(request: Request) -> Jinja2Templates:
|
|||||||
return request.app.state.templates
|
return request.app.state.templates
|
||||||
|
|
||||||
|
|
||||||
|
def _get_hcaptcha_service(request: Request) -> HCaptchaService:
|
||||||
|
"""Return the app-scoped :class:`HCaptchaService`."""
|
||||||
|
return request.app.state.hcaptcha_service
|
||||||
|
|
||||||
|
|
||||||
|
def _get_contact_service(request: Request) -> ContactService:
|
||||||
|
"""Return the app-scoped :class:`ContactService`."""
|
||||||
|
return request.app.state.contact_service
|
||||||
|
|
||||||
|
|
||||||
|
def _client_ip(request: Request) -> str:
|
||||||
|
"""Best-effort client IP from the request.
|
||||||
|
|
||||||
|
Starlette already respects ``X-Forwarded-For`` via the proxy-headers
|
||||||
|
middleware Uvicorn installs with ``--proxy-headers``; that means
|
||||||
|
``request.client.host`` is the real client IP.
|
||||||
|
"""
|
||||||
|
return request.client.host if request.client else ""
|
||||||
|
|
||||||
|
|
||||||
|
def _user_agent(request: Request) -> str:
|
||||||
|
"""Return the submitted User-Agent header (empty string if missing)."""
|
||||||
|
return request.headers.get("user-agent", "")
|
||||||
|
|
||||||
|
|
||||||
@router.get("/", response_class=HTMLResponse, summary="Blog index")
|
@router.get("/", response_class=HTMLResponse, summary="Blog index")
|
||||||
def home(
|
def home(
|
||||||
request: Request,
|
request: Request,
|
||||||
@@ -110,12 +155,12 @@ def contact(
|
|||||||
templates: Jinja2Templates = Depends(get_templates),
|
templates: Jinja2Templates = Depends(get_templates),
|
||||||
settings: Settings = Depends(get_settings),
|
settings: Settings = Depends(get_settings),
|
||||||
) -> HTMLResponse:
|
) -> HTMLResponse:
|
||||||
"""Render the inert contact page.
|
"""Render the contact page.
|
||||||
|
|
||||||
The form fields are marked ``disabled`` and the form has no ``method``
|
Phase 5 wires the form up to a real POST handler; this GET now
|
||||||
attribute — it is UI-only. If ``ADMIN_CONTACT_EMAIL`` is configured
|
returns the blank form. ``ADMIN_CONTACT_EMAIL`` is still surfaced
|
||||||
the template renders a ``mailto:`` link so visitors still have a way
|
as a secondary ``mailto:`` link for visitors who prefer their own
|
||||||
to reach the farm before Phase 5 wires up the real POST flow.
|
inbox over the form.
|
||||||
"""
|
"""
|
||||||
return templates.TemplateResponse(
|
return templates.TemplateResponse(
|
||||||
request,
|
request,
|
||||||
@@ -126,10 +171,176 @@ def contact(
|
|||||||
# case. We pass the value through settings so tests can
|
# case. We pass the value through settings so tests can
|
||||||
# override it without touching environment variables.
|
# override it without touching environment variables.
|
||||||
"contact_email": settings.admin_contact_email,
|
"contact_email": settings.admin_contact_email,
|
||||||
|
"hcaptcha_site_key": settings.hcaptcha_site_key,
|
||||||
|
"errors": {},
|
||||||
|
"form": {"name": "", "email": "", "message": ""},
|
||||||
|
"form_error": None,
|
||||||
},
|
},
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
@router.post("/contact", summary="Submit the contact form")
|
||||||
|
@limiter.limit("3/hour")
|
||||||
|
async def contact_submit(
|
||||||
|
request: Request,
|
||||||
|
name: str = Form(default=""),
|
||||||
|
email: str = Form(default=""),
|
||||||
|
message: str = Form(default=""),
|
||||||
|
website: str = Form(default=""),
|
||||||
|
templates: Jinja2Templates = Depends(get_templates),
|
||||||
|
settings: Settings = Depends(get_settings),
|
||||||
|
hcaptcha: HCaptchaService = Depends(_get_hcaptcha_service),
|
||||||
|
contact_service: ContactService = Depends(_get_contact_service),
|
||||||
|
) -> Response:
|
||||||
|
"""Handle the contact-form POST.
|
||||||
|
|
||||||
|
Flow (strict order — each stage short-circuits):
|
||||||
|
|
||||||
|
1. Honeypot: if ``website`` is non-empty → silent spam. Audit
|
||||||
|
``contact_spam_rejected`` with reason ``honeypot``; render the
|
||||||
|
generic success page so the bot operator cannot tell they were
|
||||||
|
filtered.
|
||||||
|
2. hCaptcha: call :meth:`HCaptchaService.verify`. On False → audit
|
||||||
|
``contact_spam_rejected`` with reason ``hcaptcha``; render the
|
||||||
|
success page. Same anti-enumeration rationale.
|
||||||
|
3. Validate fields. On any error → re-render the form with inline
|
||||||
|
error messages + HTTP 400 + submitted values preserved.
|
||||||
|
4. Persist: :meth:`ContactService.record_submission`. After this
|
||||||
|
point the message is durable even if the email fails.
|
||||||
|
5. Notify: :meth:`ContactService.send_notification`. Best-effort;
|
||||||
|
if it raises (it never should — service is defensive) we still
|
||||||
|
fall through to the success page.
|
||||||
|
6. Render ``public/contact_sent.html``.
|
||||||
|
|
||||||
|
CSRF is NOT required here: the endpoint is pre-auth, has no
|
||||||
|
session to hijack, and adding a cookie dance would break the
|
||||||
|
first-contact UX. SlowAPI + hCaptcha + honeypot are the controls.
|
||||||
|
"""
|
||||||
|
ip = _client_ip(request)
|
||||||
|
ua = _user_agent(request)
|
||||||
|
audit = request.app.state.audit_service
|
||||||
|
|
||||||
|
# --- Stage 1: honeypot ------------------------------------------------
|
||||||
|
if (website or "").strip():
|
||||||
|
audit.record(
|
||||||
|
"contact_spam_rejected",
|
||||||
|
ip=ip,
|
||||||
|
user_agent=ua,
|
||||||
|
detail={"reason": "honeypot"},
|
||||||
|
)
|
||||||
|
return templates.TemplateResponse(
|
||||||
|
request,
|
||||||
|
"public/contact_sent.html",
|
||||||
|
{"active_nav": "contact"},
|
||||||
|
)
|
||||||
|
|
||||||
|
# --- Stage 2: hCaptcha ------------------------------------------------
|
||||||
|
token = (
|
||||||
|
(request.headers.get("h-captcha-response") or "")
|
||||||
|
or ""
|
||||||
|
)
|
||||||
|
# hCaptcha's widget posts the token as a form field named
|
||||||
|
# ``h-captcha-response``. Starlette's ``Request.form()`` is async;
|
||||||
|
# re-reading it here is fine because FastAPI caches the parsed form
|
||||||
|
# for the lifetime of the request.
|
||||||
|
form_data = await request.form()
|
||||||
|
captcha_token = str(form_data.get("h-captcha-response", token) or "")
|
||||||
|
|
||||||
|
ok = await hcaptcha.verify(captcha_token, ip)
|
||||||
|
if not ok:
|
||||||
|
audit.record(
|
||||||
|
"contact_spam_rejected",
|
||||||
|
ip=ip,
|
||||||
|
user_agent=ua,
|
||||||
|
detail={"reason": "hcaptcha"},
|
||||||
|
)
|
||||||
|
return templates.TemplateResponse(
|
||||||
|
request,
|
||||||
|
"public/contact_sent.html",
|
||||||
|
{"active_nav": "contact"},
|
||||||
|
)
|
||||||
|
|
||||||
|
# --- Stage 3: validation ---------------------------------------------
|
||||||
|
clean_name = (name or "").strip()
|
||||||
|
clean_email = (email or "").strip()
|
||||||
|
clean_message = (message or "").strip()
|
||||||
|
|
||||||
|
errors: dict[str, str] = {}
|
||||||
|
if not clean_name:
|
||||||
|
errors["name"] = "Please enter your name."
|
||||||
|
elif len(clean_name) > _NAME_MAX:
|
||||||
|
errors["name"] = f"Name must be {_NAME_MAX} characters or fewer."
|
||||||
|
|
||||||
|
if not clean_email:
|
||||||
|
errors["email"] = "Please enter your email address."
|
||||||
|
elif len(clean_email) > _EMAIL_MAX or not _EMAIL_RE.match(clean_email):
|
||||||
|
errors["email"] = "Please enter a valid email address."
|
||||||
|
|
||||||
|
if not clean_message:
|
||||||
|
errors["message"] = "Please include a message."
|
||||||
|
elif len(clean_message) < _MESSAGE_MIN:
|
||||||
|
errors["message"] = (
|
||||||
|
f"Message must be at least {_MESSAGE_MIN} characters."
|
||||||
|
)
|
||||||
|
elif len(clean_message) > _MESSAGE_MAX:
|
||||||
|
errors["message"] = (
|
||||||
|
f"Message must be {_MESSAGE_MAX} characters or fewer."
|
||||||
|
)
|
||||||
|
|
||||||
|
if errors:
|
||||||
|
return templates.TemplateResponse(
|
||||||
|
request,
|
||||||
|
"public/contact.html",
|
||||||
|
{
|
||||||
|
"active_nav": "contact",
|
||||||
|
"contact_email": settings.admin_contact_email,
|
||||||
|
"hcaptcha_site_key": settings.hcaptcha_site_key,
|
||||||
|
"errors": errors,
|
||||||
|
"form": {
|
||||||
|
"name": clean_name,
|
||||||
|
"email": clean_email,
|
||||||
|
"message": clean_message,
|
||||||
|
},
|
||||||
|
"form_error": None,
|
||||||
|
},
|
||||||
|
status_code=400,
|
||||||
|
)
|
||||||
|
|
||||||
|
# --- Stage 4: persist -------------------------------------------------
|
||||||
|
submission = contact_service.record_submission(
|
||||||
|
name=clean_name,
|
||||||
|
email=clean_email,
|
||||||
|
message=clean_message,
|
||||||
|
ip=ip,
|
||||||
|
user_agent=ua,
|
||||||
|
)
|
||||||
|
|
||||||
|
# Audit the successful submission. Store the message length + a
|
||||||
|
# short preview so on-call can tell whether a flood is substantive
|
||||||
|
# without exposing full bodies in the audit log.
|
||||||
|
audit.record(
|
||||||
|
"contact_submitted",
|
||||||
|
email=clean_email,
|
||||||
|
ip=ip,
|
||||||
|
user_agent=ua,
|
||||||
|
detail={
|
||||||
|
"submission_id": submission.id,
|
||||||
|
"message_length": len(clean_message),
|
||||||
|
"message_preview": clean_message[:40],
|
||||||
|
},
|
||||||
|
)
|
||||||
|
|
||||||
|
# --- Stage 5: notify --------------------------------------------------
|
||||||
|
contact_service.send_notification(submission)
|
||||||
|
|
||||||
|
# --- Stage 6: success page -------------------------------------------
|
||||||
|
return templates.TemplateResponse(
|
||||||
|
request,
|
||||||
|
"public/contact_sent.html",
|
||||||
|
{"active_nav": "contact"},
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
@router.get("/shop", response_class=HTMLResponse, summary="Shop placeholder")
|
@router.get("/shop", response_class=HTMLResponse, summary="Shop placeholder")
|
||||||
def shop(
|
def shop(
|
||||||
request: Request,
|
request: Request,
|
||||||
|
|||||||
159
app/services/contact.py
Normal file
159
app/services/contact.py
Normal file
@@ -0,0 +1,159 @@
|
|||||||
|
"""Contact-form persistence + notification orchestration.
|
||||||
|
|
||||||
|
Thin service wired at app startup:
|
||||||
|
|
||||||
|
- :meth:`ContactService.record_submission` — insert a row into
|
||||||
|
``contact_submissions`` and return a mapped
|
||||||
|
:class:`ContactSubmission` entity.
|
||||||
|
- :meth:`ContactService.send_notification` — best-effort pass through
|
||||||
|
to :class:`EmailService.send_contact_notification`. NEVER raises.
|
||||||
|
|
||||||
|
The route handler (``/contact``) short-circuits the spam path (honeypot
|
||||||
|
tripped or hCaptcha failed) BEFORE calling either method, so anything
|
||||||
|
that reaches this service has already passed the bot screen. The audit
|
||||||
|
log is written by the route, not here — keeps this service free of
|
||||||
|
request-scoped context (ip/ua are persisted on the row itself).
|
||||||
|
|
||||||
|
Security notes
|
||||||
|
--------------
|
||||||
|
- Parameterized SQL only (sqlalchemy ``text("...:param...")``).
|
||||||
|
- Datetimes are timezone-aware UTC and serialized via ``.isoformat()``,
|
||||||
|
matching the rest of the codebase.
|
||||||
|
- Raw message bodies never flow through logs or audit detail; only
|
||||||
|
``len(message)`` and a short preview (if any) appear in audit rows,
|
||||||
|
and that is the route's responsibility.
|
||||||
|
"""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from datetime import datetime, timezone
|
||||||
|
|
||||||
|
import structlog
|
||||||
|
from sqlalchemy import Engine, text
|
||||||
|
|
||||||
|
from app.config import Settings
|
||||||
|
from app.models.entities import ContactSubmission
|
||||||
|
from app.models.mappers import row_to_contact_submission
|
||||||
|
from app.services.audit import AuditService
|
||||||
|
from app.services.email import EmailService
|
||||||
|
|
||||||
|
|
||||||
|
_log = structlog.get_logger(__name__)
|
||||||
|
|
||||||
|
|
||||||
|
class ContactService:
|
||||||
|
"""Persist contact submissions and trigger notification emails."""
|
||||||
|
|
||||||
|
def __init__(
|
||||||
|
self,
|
||||||
|
engine: Engine,
|
||||||
|
email: EmailService,
|
||||||
|
audit: AuditService,
|
||||||
|
settings: Settings,
|
||||||
|
) -> None:
|
||||||
|
"""Store collaborators by reference."""
|
||||||
|
self._engine: Engine = engine
|
||||||
|
self._email: EmailService = email
|
||||||
|
self._audit: AuditService = audit
|
||||||
|
self._settings: Settings = settings
|
||||||
|
|
||||||
|
# ------------------------------------------------------------------
|
||||||
|
# record_submission
|
||||||
|
# ------------------------------------------------------------------
|
||||||
|
def record_submission(
|
||||||
|
self,
|
||||||
|
*,
|
||||||
|
name: str,
|
||||||
|
email: str,
|
||||||
|
message: str,
|
||||||
|
ip: str,
|
||||||
|
user_agent: str,
|
||||||
|
) -> ContactSubmission:
|
||||||
|
"""Insert a ``contact_submissions`` row and return the entity.
|
||||||
|
|
||||||
|
All inputs are taken at face value — validation is the route's
|
||||||
|
responsibility. The mapper guarantees tz-aware UTC on read so
|
||||||
|
callers can safely format ``submitted_at.isoformat()`` in
|
||||||
|
downstream emails.
|
||||||
|
"""
|
||||||
|
submitted_at = datetime.now(timezone.utc)
|
||||||
|
submitted_iso = submitted_at.isoformat()
|
||||||
|
|
||||||
|
with self._engine.begin() as conn:
|
||||||
|
result = conn.execute(
|
||||||
|
text(
|
||||||
|
"INSERT INTO contact_submissions"
|
||||||
|
" (name, email, message, ip, user_agent,"
|
||||||
|
" submitted_at, handled)"
|
||||||
|
" VALUES (:name, :email, :message, :ip, :user_agent,"
|
||||||
|
" :submitted_at, 0)"
|
||||||
|
),
|
||||||
|
{
|
||||||
|
"name": name,
|
||||||
|
"email": email,
|
||||||
|
"message": message,
|
||||||
|
"ip": ip or "",
|
||||||
|
"user_agent": user_agent or "",
|
||||||
|
"submitted_at": submitted_iso,
|
||||||
|
},
|
||||||
|
)
|
||||||
|
new_id = int(result.lastrowid) # type: ignore[arg-type]
|
||||||
|
|
||||||
|
row = conn.execute(
|
||||||
|
text(
|
||||||
|
"SELECT id, name, email, message, ip, user_agent,"
|
||||||
|
" submitted_at, handled"
|
||||||
|
" FROM contact_submissions WHERE id = :id"
|
||||||
|
),
|
||||||
|
{"id": new_id},
|
||||||
|
).mappings().first()
|
||||||
|
|
||||||
|
# ``row`` is always present — we just inserted it in the same
|
||||||
|
# transaction — but we type-guard defensively so mypy is happy.
|
||||||
|
if row is None: # pragma: no cover — impossible path
|
||||||
|
raise RuntimeError(
|
||||||
|
"contact_submission row disappeared between insert and select"
|
||||||
|
)
|
||||||
|
return row_to_contact_submission(row)
|
||||||
|
|
||||||
|
# ------------------------------------------------------------------
|
||||||
|
# send_notification
|
||||||
|
# ------------------------------------------------------------------
|
||||||
|
def send_notification(self, submission: ContactSubmission) -> None:
|
||||||
|
"""Dispatch the admin notification email; never raise.
|
||||||
|
|
||||||
|
Best-effort: any exception is caught and logged under
|
||||||
|
``contact_send_failed`` so the caller can still render the
|
||||||
|
success page. The submission row is already persisted when
|
||||||
|
this method is invoked, so the admin has a DB trail even if
|
||||||
|
the outbound email fails.
|
||||||
|
|
||||||
|
Silently no-ops when ``ADMIN_CONTACT_EMAIL`` is unset. That
|
||||||
|
only happens in dev (production validator requires the field)
|
||||||
|
and is logged for visibility.
|
||||||
|
"""
|
||||||
|
to = self._settings.admin_contact_email
|
||||||
|
if not to:
|
||||||
|
_log.info(
|
||||||
|
"contact_notification_skipped_no_recipient",
|
||||||
|
reason="admin_contact_email_unset",
|
||||||
|
)
|
||||||
|
return
|
||||||
|
|
||||||
|
try:
|
||||||
|
self._email.send_contact_notification(
|
||||||
|
to=to,
|
||||||
|
submission_name=submission.name,
|
||||||
|
submission_email=submission.email,
|
||||||
|
message=submission.message,
|
||||||
|
submitted_at=submission.submitted_at,
|
||||||
|
ip=submission.ip,
|
||||||
|
)
|
||||||
|
except Exception: # noqa: BLE001
|
||||||
|
# The email service is itself defensive (it swallows Resend
|
||||||
|
# errors internally). This belt-and-braces catch protects
|
||||||
|
# the request path from any unexpected programming error.
|
||||||
|
_log.exception(
|
||||||
|
"contact_send_failed",
|
||||||
|
submission_id=submission.id,
|
||||||
|
)
|
||||||
@@ -52,6 +52,84 @@ class EmailService:
|
|||||||
self._settings: Settings = settings
|
self._settings: Settings = settings
|
||||||
self._templates: Jinja2Templates = templates
|
self._templates: Jinja2Templates = templates
|
||||||
|
|
||||||
|
def send_contact_notification(
|
||||||
|
self,
|
||||||
|
*,
|
||||||
|
to: str,
|
||||||
|
submission_name: str,
|
||||||
|
submission_email: str,
|
||||||
|
message: str,
|
||||||
|
submitted_at: datetime,
|
||||||
|
ip: str,
|
||||||
|
) -> None:
|
||||||
|
"""Send the contact-form notification email to the admin inbox.
|
||||||
|
|
||||||
|
Behavior
|
||||||
|
--------
|
||||||
|
- If ``settings.resend_api_key`` (or ``resend_from``) is falsy,
|
||||||
|
log a ``contact_notification_dev_fallback`` event at INFO and
|
||||||
|
return. Dev convenience only — the production validator
|
||||||
|
refuses to boot without a Resend key.
|
||||||
|
- Otherwise render both template bodies, set ``Reply-To`` to the
|
||||||
|
submitter's email (so Head Hen just hits reply), and dispatch
|
||||||
|
via Resend. Transport errors are logged (``contact_notification_failed``)
|
||||||
|
and never re-raised — the request path must always complete
|
||||||
|
with the generic success page.
|
||||||
|
|
||||||
|
Subject format (fixed) matches the Phase 5 brief:
|
||||||
|
``"New contact submission from {submission_name}"``.
|
||||||
|
"""
|
||||||
|
ctx = {
|
||||||
|
"submission_name": submission_name,
|
||||||
|
"submission_email": submission_email,
|
||||||
|
"message": message,
|
||||||
|
"submitted_at": submitted_at.isoformat(),
|
||||||
|
"ip": ip or "",
|
||||||
|
}
|
||||||
|
html_body = self._render("emails/contact_notification.html", ctx)
|
||||||
|
text_body = self._render("emails/contact_notification.txt", ctx)
|
||||||
|
|
||||||
|
api_key: Optional[str] = self._settings.resend_api_key
|
||||||
|
sender: Optional[str] = self._settings.resend_from
|
||||||
|
|
||||||
|
# Dev fallback — log enough to confirm the flow reached the
|
||||||
|
# email layer without echoing the whole message body at INFO.
|
||||||
|
if not api_key or not sender:
|
||||||
|
_log.info(
|
||||||
|
"contact_notification_dev_fallback",
|
||||||
|
to=to,
|
||||||
|
submission_name=submission_name,
|
||||||
|
submission_email=submission_email,
|
||||||
|
message_length=len(message or ""),
|
||||||
|
)
|
||||||
|
return
|
||||||
|
|
||||||
|
subject = f"New contact submission from {submission_name}"
|
||||||
|
try:
|
||||||
|
import resend # type: ignore[import-untyped]
|
||||||
|
|
||||||
|
resend.api_key = api_key
|
||||||
|
resend.Emails.send(
|
||||||
|
{
|
||||||
|
"from": sender,
|
||||||
|
"to": to,
|
||||||
|
"subject": subject,
|
||||||
|
"html": html_body,
|
||||||
|
"text": text_body,
|
||||||
|
# Reply-To lets Head Hen reply directly from her
|
||||||
|
# inbox without exposing the From: address to
|
||||||
|
# public rewriting.
|
||||||
|
"reply_to": submission_email,
|
||||||
|
}
|
||||||
|
)
|
||||||
|
_log.info("contact_notification_sent", to=to)
|
||||||
|
except Exception: # noqa: BLE001
|
||||||
|
# Never raise from the request path — see docstring.
|
||||||
|
_log.exception(
|
||||||
|
"contact_notification_failed",
|
||||||
|
to=to,
|
||||||
|
)
|
||||||
|
|
||||||
def send_magic_link(
|
def send_magic_link(
|
||||||
self,
|
self,
|
||||||
*,
|
*,
|
||||||
|
|||||||
132
app/services/hcaptcha.py
Normal file
132
app/services/hcaptcha.py
Normal file
@@ -0,0 +1,132 @@
|
|||||||
|
"""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
|
||||||
@@ -490,6 +490,26 @@ a:focus-visible {
|
|||||||
margin-top: var(--space-2);
|
margin-top: var(--space-2);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/* Phase 5: inline field errors + top-level banner. Scoped to the
|
||||||
|
contact form so the red tone does not bleed into other forms. */
|
||||||
|
.contact-form__field-error,
|
||||||
|
.contact-form__error {
|
||||||
|
margin: var(--space-1) 0 0;
|
||||||
|
color: #8b2e2e;
|
||||||
|
font-size: 0.95rem;
|
||||||
|
}
|
||||||
|
|
||||||
|
.contact-form__error {
|
||||||
|
padding: var(--space-2) var(--space-3);
|
||||||
|
background-color: #fbe9e7;
|
||||||
|
border: 1px solid #d9a8a1;
|
||||||
|
border-radius: var(--radius);
|
||||||
|
}
|
||||||
|
|
||||||
|
.contact-form__captcha {
|
||||||
|
margin-top: var(--space-2);
|
||||||
|
}
|
||||||
|
|
||||||
/* Generic button. */
|
/* Generic button. */
|
||||||
.btn {
|
.btn {
|
||||||
display: inline-block;
|
display: inline-block;
|
||||||
|
|||||||
43
app/templates/emails/contact_notification.html
Normal file
43
app/templates/emails/contact_notification.html
Normal file
@@ -0,0 +1,43 @@
|
|||||||
|
{#
|
||||||
|
HTML body for the admin contact-form notification email.
|
||||||
|
|
||||||
|
Deliberately plain — email clients are hostile to CSS. ``message`` is
|
||||||
|
rendered inside a <pre> so newlines submitted by the visitor are
|
||||||
|
preserved without us having to manually convert them to <br>s (which
|
||||||
|
bleach would strip anyway).
|
||||||
|
|
||||||
|
Jinja2 autoescape is on for .html templates; every field below is
|
||||||
|
rendered via ``{{ ... }}`` without ``| safe``, so user input cannot
|
||||||
|
escape into the DOM.
|
||||||
|
|
||||||
|
Context:
|
||||||
|
- submission_name : str
|
||||||
|
- submission_email : str
|
||||||
|
- message : str
|
||||||
|
- submitted_at : ISO-8601 str
|
||||||
|
- ip : str
|
||||||
|
#}<!doctype html>
|
||||||
|
<html lang="en">
|
||||||
|
<head>
|
||||||
|
<meta charset="utf-8">
|
||||||
|
<title>New contact submission — Chicken Babies R Us</title>
|
||||||
|
</head>
|
||||||
|
<body style="font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, sans-serif; color: #2B3A42; max-width: 560px; margin: 0 auto; padding: 24px;">
|
||||||
|
<h1 style="font-size: 20px; margin: 0 0 16px;">New contact submission</h1>
|
||||||
|
<p style="margin: 0 0 8px;"><strong>Name:</strong> {{ submission_name }}</p>
|
||||||
|
<p style="margin: 0 0 8px;">
|
||||||
|
<strong>Email:</strong>
|
||||||
|
<a href="mailto:{{ submission_email }}" style="color: #5D8AA8;">{{ submission_email }}</a>
|
||||||
|
</p>
|
||||||
|
<p style="margin: 0 0 8px;"><strong>Submitted at:</strong> {{ submitted_at }}</p>
|
||||||
|
<p style="margin: 0 0 16px;"><strong>IP:</strong> {{ ip }}</p>
|
||||||
|
|
||||||
|
<h2 style="font-size: 16px; margin: 16px 0 8px;">Message</h2>
|
||||||
|
<pre style="white-space: pre-wrap; word-break: break-word; font-family: inherit; font-size: 14px; background: #FAF3E7; padding: 12px 16px; border-radius: 6px; margin: 0;">{{ message }}</pre>
|
||||||
|
|
||||||
|
<p style="color: #6b7a80; font-size: 13px; margin-top: 24px;">
|
||||||
|
Replying to this email will respond directly to the sender
|
||||||
|
({{ submission_email }}).
|
||||||
|
</p>
|
||||||
|
</body>
|
||||||
|
</html>
|
||||||
17
app/templates/emails/contact_notification.txt
Normal file
17
app/templates/emails/contact_notification.txt
Normal file
@@ -0,0 +1,17 @@
|
|||||||
|
{# Plaintext body for the admin contact-form notification email.
|
||||||
|
Mirrors the HTML version so clients that reject HTML still get a
|
||||||
|
readable message. #}New contact submission
|
||||||
|
======================
|
||||||
|
|
||||||
|
Name: {{ submission_name }}
|
||||||
|
Email: {{ submission_email }}
|
||||||
|
Submitted at: {{ submitted_at }}
|
||||||
|
IP: {{ ip }}
|
||||||
|
|
||||||
|
Message
|
||||||
|
-------
|
||||||
|
{{ message }}
|
||||||
|
|
||||||
|
--
|
||||||
|
Replying to this email will respond directly to the sender
|
||||||
|
({{ submission_email }}).
|
||||||
@@ -1,18 +1,30 @@
|
|||||||
{#
|
{#
|
||||||
Contact page — Phase 1 version.
|
Contact page — live Phase 5 form.
|
||||||
|
|
||||||
The form is deliberately inert: no `method`, no `action`, all inputs
|
POSTs to /contact. Honeypot + hCaptcha + SlowAPI rate-limit protect
|
||||||
and the submit button carry the `disabled` attribute. A muted note
|
the endpoint. Every field carries an id/label pair for a11y and a
|
||||||
explains the form is coming soon; if `ADMIN_CONTACT_EMAIL` is set in
|
maxlength/minlength to match the server-side validator — the HTML5
|
||||||
the environment we render a `mailto:` link above the form so visitors
|
attributes are a UX hint only, not the security boundary.
|
||||||
still have a way to reach the farm.
|
|
||||||
|
|
||||||
Phase 5 replaces this template with a working POST handler, hCaptcha,
|
Honeypot:
|
||||||
honeypot, and rate limiting.
|
- The ``website`` field is wrapped in a .visually-hidden container
|
||||||
|
marked ``aria-hidden="true"`` so assistive tech hides it too.
|
||||||
|
- It is NOT ``required`` and has ``tabindex="-1"`` so a keyboard
|
||||||
|
user can't accidentally focus it.
|
||||||
|
- The server rejects any submission where the field is non-empty.
|
||||||
|
|
||||||
|
hCaptcha:
|
||||||
|
- When ``hcaptcha_site_key`` is truthy the widget div + script tag
|
||||||
|
render. When empty (dev) we skip them and rely on the dev-mode
|
||||||
|
fallback in :class:`HCaptchaService`.
|
||||||
|
|
||||||
Context:
|
Context:
|
||||||
- contact_email : str | None (from settings.admin_contact_email)
|
- contact_email : str | None (from settings.admin_contact_email)
|
||||||
- active_nav : "contact"
|
- active_nav : "contact"
|
||||||
|
- errors : dict[str, str] (field_name -> message)
|
||||||
|
- form : dict[str, str] (prior submitted values)
|
||||||
|
- form_error : str | None (top-level error banner)
|
||||||
|
- hcaptcha_site_key : str | None (rendered when truthy)
|
||||||
#}
|
#}
|
||||||
{% extends "public/base.html" %}
|
{% extends "public/base.html" %}
|
||||||
|
|
||||||
@@ -32,30 +44,45 @@
|
|||||||
|
|
||||||
{% if contact_email %}
|
{% if contact_email %}
|
||||||
<p class="contact-mailto">
|
<p class="contact-mailto">
|
||||||
The easiest way to reach us right now is email:
|
Prefer email? Reach us at
|
||||||
<a href="mailto:{{ contact_email }}">{{ contact_email }}</a>.
|
<a href="mailto:{{ contact_email }}">{{ contact_email }}</a>.
|
||||||
</p>
|
</p>
|
||||||
{% else %}
|
|
||||||
<p class="contact-mailto contact-mailto--muted">
|
|
||||||
A direct email address will be posted here soon.
|
|
||||||
</p>
|
|
||||||
{% endif %}
|
{% endif %}
|
||||||
|
|
||||||
<p class="contact-form__note" role="note">
|
{% if form_error %}
|
||||||
Secure contact form coming soon.
|
<p class="contact-form__error" role="alert">{{ form_error }}</p>
|
||||||
</p>
|
{% endif %}
|
||||||
|
|
||||||
|
<form class="contact-form"
|
||||||
|
method="POST"
|
||||||
|
action="/contact"
|
||||||
|
aria-describedby="contact-form-note">
|
||||||
|
{# Honeypot — visually hidden + aria-hidden so neither sighted
|
||||||
|
users nor screen readers encounter it. Bots fill it in and
|
||||||
|
get silently filed as spam. #}
|
||||||
|
<div class="visually-hidden" aria-hidden="true">
|
||||||
|
<label for="contact-website">Website</label>
|
||||||
|
<input type="text"
|
||||||
|
id="contact-website"
|
||||||
|
name="website"
|
||||||
|
tabindex="-1"
|
||||||
|
autocomplete="off"
|
||||||
|
class="contact-hp"
|
||||||
|
value="">
|
||||||
|
</div>
|
||||||
|
|
||||||
{# action="" and no method = form cannot submit. Every input is
|
|
||||||
disabled so screen readers and the keyboard both respect the
|
|
||||||
"not-yet-available" state. #}
|
|
||||||
<form class="contact-form" action="" aria-describedby="contact-form-note" novalidate>
|
|
||||||
<div class="contact-form__field">
|
<div class="contact-form__field">
|
||||||
<label for="contact-name">Name</label>
|
<label for="contact-name">Name</label>
|
||||||
<input type="text"
|
<input type="text"
|
||||||
id="contact-name"
|
id="contact-name"
|
||||||
name="name"
|
name="name"
|
||||||
autocomplete="name"
|
autocomplete="name"
|
||||||
disabled>
|
required
|
||||||
|
maxlength="80"
|
||||||
|
value="{{ (form.name if form else '') or '' }}">
|
||||||
|
{% if errors and errors.name %}
|
||||||
|
<p class="contact-form__field-error" role="alert">{{ errors.name }}</p>
|
||||||
|
{% endif %}
|
||||||
</div>
|
</div>
|
||||||
|
|
||||||
<div class="contact-form__field">
|
<div class="contact-form__field">
|
||||||
@@ -64,7 +91,12 @@
|
|||||||
id="contact-email"
|
id="contact-email"
|
||||||
name="email"
|
name="email"
|
||||||
autocomplete="email"
|
autocomplete="email"
|
||||||
disabled>
|
required
|
||||||
|
maxlength="254"
|
||||||
|
value="{{ (form.email if form else '') or '' }}">
|
||||||
|
{% if errors and errors.email %}
|
||||||
|
<p class="contact-form__field-error" role="alert">{{ errors.email }}</p>
|
||||||
|
{% endif %}
|
||||||
</div>
|
</div>
|
||||||
|
|
||||||
<div class="contact-form__field">
|
<div class="contact-form__field">
|
||||||
@@ -72,14 +104,31 @@
|
|||||||
<textarea id="contact-message"
|
<textarea id="contact-message"
|
||||||
name="message"
|
name="message"
|
||||||
rows="6"
|
rows="6"
|
||||||
disabled></textarea>
|
required
|
||||||
|
minlength="10"
|
||||||
|
maxlength="4000">{{ (form.message if form else '') or '' }}</textarea>
|
||||||
|
{% if errors and errors.message %}
|
||||||
|
<p class="contact-form__field-error" role="alert">{{ errors.message }}</p>
|
||||||
|
{% endif %}
|
||||||
</div>
|
</div>
|
||||||
|
|
||||||
|
{% if hcaptcha_site_key %}
|
||||||
|
<div class="contact-form__captcha">
|
||||||
|
<div class="h-captcha" data-sitekey="{{ hcaptcha_site_key }}"></div>
|
||||||
|
</div>
|
||||||
|
{% else %}
|
||||||
|
{# hCaptcha disabled in dev — HCaptchaService returns True. #}
|
||||||
|
{% endif %}
|
||||||
|
|
||||||
<div class="contact-form__actions">
|
<div class="contact-form__actions">
|
||||||
<button type="submit" class="btn btn--primary" disabled>
|
<button type="submit" class="btn btn--primary">
|
||||||
Send message
|
Send message
|
||||||
</button>
|
</button>
|
||||||
</div>
|
</div>
|
||||||
</form>
|
</form>
|
||||||
</article>
|
</article>
|
||||||
|
|
||||||
|
{% if hcaptcha_site_key %}
|
||||||
|
<script src="https://js.hcaptcha.com/1/api.js" async defer></script>
|
||||||
|
{% endif %}
|
||||||
{% endblock %}
|
{% endblock %}
|
||||||
|
|||||||
31
app/templates/public/contact_sent.html
Normal file
31
app/templates/public/contact_sent.html
Normal file
@@ -0,0 +1,31 @@
|
|||||||
|
{#
|
||||||
|
Contact success page — rendered after a successful POST /contact
|
||||||
|
OR after a silent spam rejection (honeypot tripped / hCaptcha
|
||||||
|
failed). Copy MUST stay identical across those branches so a bot
|
||||||
|
operator can't use the response body to distinguish "we accepted
|
||||||
|
your message" from "we filed your message under spam".
|
||||||
|
|
||||||
|
Context:
|
||||||
|
- active_nav : "contact"
|
||||||
|
#}
|
||||||
|
{% extends "public/base.html" %}
|
||||||
|
|
||||||
|
{% block title %}Message sent — Chicken Babies R Us{% endblock %}
|
||||||
|
{% block meta_description %}Thanks for reaching out to Chicken Babies R Us.{% endblock %}
|
||||||
|
|
||||||
|
{% block content %}
|
||||||
|
<article class="page-article">
|
||||||
|
<header class="page-article__header">
|
||||||
|
<h1 class="page-article__title">Thanks for reaching out</h1>
|
||||||
|
</header>
|
||||||
|
|
||||||
|
<p>
|
||||||
|
Your message is on its way to Head Hen. We'll get back to you as
|
||||||
|
soon as the chickens let us.
|
||||||
|
</p>
|
||||||
|
|
||||||
|
<p>
|
||||||
|
<a class="btn btn--primary" href="/">Back to the home page</a>
|
||||||
|
</p>
|
||||||
|
</article>
|
||||||
|
{% endblock %}
|
||||||
@@ -161,3 +161,88 @@ Pre-requisites:
|
|||||||
- [ ] A newly-published post shows at the top of `/` within one request.
|
- [ ] A newly-published post shows at the top of `/` within one request.
|
||||||
- [ ] `/about` shows the most recently edited copy.
|
- [ ] `/about` shows the most recently edited copy.
|
||||||
- [ ] No admin-facing text (status, dashboard wording) leaks into the public HTML.
|
- [ ] No admin-facing text (status, dashboard wording) leaks into the public HTML.
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
## Phase 5 — Contact Form
|
||||||
|
|
||||||
|
Pre-requisites:
|
||||||
|
- `ADMIN_CONTACT_EMAIL` set in `.env` (the destination inbox).
|
||||||
|
- For the production-like happy path: `RESEND_API_KEY` + `RESEND_FROM`
|
||||||
|
set; otherwise the send path logs `contact_notification_dev_fallback`
|
||||||
|
and the admin inbox will not actually receive mail.
|
||||||
|
- Optionally set `HCAPTCHA_SITE_KEY` + `HCAPTCHA_SECRET` to exercise
|
||||||
|
the real widget; with both unset the dev fallback auto-passes and
|
||||||
|
logs `hcaptcha_dev_fallback`.
|
||||||
|
|
||||||
|
### GET `/contact`
|
||||||
|
|
||||||
|
- [ ] Page returns 200 and renders without console errors.
|
||||||
|
- [ ] H1 reads **"Get in touch"**.
|
||||||
|
- [ ] Name, email, and message fields render as editable inputs (no `disabled` attribute).
|
||||||
|
- [ ] "Send message" button is enabled.
|
||||||
|
- [ ] Form has `method="POST"` and `action="/contact"` (view source).
|
||||||
|
- [ ] Honeypot `<input name="website">` is present in the markup but
|
||||||
|
wrapped in a `.visually-hidden` container marked
|
||||||
|
`aria-hidden="true"` — it is invisible to sighted users.
|
||||||
|
- [ ] When `HCAPTCHA_SITE_KEY` is set, the `h-captcha` div and the
|
||||||
|
`https://js.hcaptcha.com/1/api.js` script appear. When unset,
|
||||||
|
neither appears.
|
||||||
|
- [ ] Nav marks "Contact" as active.
|
||||||
|
|
||||||
|
### Happy path
|
||||||
|
|
||||||
|
- [ ] Fill in Name, Email, Message (>= 10 chars) and submit.
|
||||||
|
- [ ] Response is HTTP 200 and renders **"Thanks for reaching out"**.
|
||||||
|
- [ ] `sqlite3 data/app.db "SELECT id, name, email, length(message), handled FROM contact_submissions"` shows the new row with `handled=0`.
|
||||||
|
- [ ] Server log contains a `contact_submitted` structured event with a
|
||||||
|
`message_preview` at most 40 chars long (no full body).
|
||||||
|
- [ ] With `RESEND_API_KEY` set: admin inbox receives the notification
|
||||||
|
email. `From:` matches `RESEND_FROM`; `Reply-To:` matches the
|
||||||
|
submitted email; subject is `New contact submission from {name}`.
|
||||||
|
- [ ] Without `RESEND_API_KEY`: server log contains
|
||||||
|
`contact_notification_dev_fallback` with the submitter's name,
|
||||||
|
email, and message length.
|
||||||
|
|
||||||
|
### Validation errors
|
||||||
|
|
||||||
|
- [ ] Submitting with a blank name shows **"Please enter your name."** inline.
|
||||||
|
- [ ] Submitting with `not-an-email` shows **"Please enter a valid email address."**.
|
||||||
|
- [ ] Submitting with a 9-character message shows
|
||||||
|
**"Message must be at least 10 characters."**.
|
||||||
|
- [ ] Submitting with a > 4000-character message shows
|
||||||
|
**"Message must be 4000 characters or fewer."**.
|
||||||
|
- [ ] Submitting with a > 80-character name shows
|
||||||
|
**"Name must be 80 characters or fewer."**.
|
||||||
|
- [ ] The response status code on every validation failure is **400**.
|
||||||
|
- [ ] Prior valid values remain filled in so the user doesn't retype.
|
||||||
|
|
||||||
|
### Spam paths
|
||||||
|
|
||||||
|
- [ ] Filling the honeypot `website` field and submitting returns
|
||||||
|
**"Thanks for reaching out"** (same as success) AND no row is
|
||||||
|
persisted AND an audit row with
|
||||||
|
`event_type='contact_spam_rejected'` and `reason=honeypot` exists.
|
||||||
|
- [ ] With a real hCaptcha configured: submitting without solving the
|
||||||
|
widget returns the same generic thank-you page. Audit row:
|
||||||
|
`contact_spam_rejected` / `reason=hcaptcha`. No DB row.
|
||||||
|
|
||||||
|
### Rate limit
|
||||||
|
|
||||||
|
- [ ] Submit the form **4 times** from the same browser session within
|
||||||
|
an hour. The fourth submission returns HTTP **429** and renders
|
||||||
|
the "Too many attempts" template. A `rate_limited` audit row is
|
||||||
|
added with `scope=ip` and `endpoint=/contact`.
|
||||||
|
|
||||||
|
### Email send failure
|
||||||
|
|
||||||
|
- [ ] With a valid form but `RESEND_API_KEY` pointed at an invalid key:
|
||||||
|
the user still sees **"Thanks for reaching out"**, the DB row is
|
||||||
|
created, and the server log contains
|
||||||
|
`contact_notification_failed` (logged by EmailService). The user
|
||||||
|
experience is indistinguishable from success.
|
||||||
|
|
||||||
|
### Ops smoke
|
||||||
|
|
||||||
|
- [ ] `pytest -q tests/test_hcaptcha_service.py tests/test_contact_service.py tests/test_contact_routes.py` passes.
|
||||||
|
- [ ] `python -c "from app.main import app; print(len(app.routes))"` prints a count greater than the Phase 4 count (the new `POST /contact` adds one route).
|
||||||
|
|||||||
@@ -211,11 +211,46 @@ High-level phased plan. Each phase ends in a mergeable `dev` state and a passing
|
|||||||
**Verification run:**
|
**Verification run:**
|
||||||
`python -c "from app.main import app"` ✓ (26 routes registered) · `pytest -q tests/test_slugs.py tests/test_csrf_service.py tests/test_media_service.py tests/test_admin_posts_service.py tests/test_admin_pages_service.py tests/test_admin_cms_routes.py tests/test_admin_routes.py` → 64 passed ✓ · full `pytest -q` → 118 passed, 2 failed; both failures are pre-existing on `dev` (the `logo.` → `logo-mark.` asset rename in commit `f5098c0`; and the `RESEND_FROM`/`ADMIN_EMAILS` pollution from local `.env` into the Settings-validator test) and unrelated to Phase 4.
|
`python -c "from app.main import app"` ✓ (26 routes registered) · `pytest -q tests/test_slugs.py tests/test_csrf_service.py tests/test_media_service.py tests/test_admin_posts_service.py tests/test_admin_pages_service.py tests/test_admin_cms_routes.py tests/test_admin_routes.py` → 64 passed ✓ · full `pytest -q` → 118 passed, 2 failed; both failures are pre-existing on `dev` (the `logo.` → `logo-mark.` asset rename in commit `f5098c0`; and the `RESEND_FROM`/`ADMIN_EMAILS` pollution from local `.env` into the Settings-validator test) and unrelated to Phase 4.
|
||||||
|
|
||||||
## Phase 5 — Contact Form
|
## Phase 5 — Contact Form ✅
|
||||||
|
|
||||||
- `/contact` POST flow: field validation → hCaptcha verify → honeypot check → rate limit → `Resend` send → persist submission row → success page.
|
**Completed:** 2026-04-22
|
||||||
- `FROM` on verified domain; `Reply-To` = submitter's email.
|
|
||||||
- No internal errors leak to the user; they see a generic "something went wrong, please try again".
|
**Summary:** Shipped the public contact form: live POST handler with honeypot → hCaptcha (server-verify) → field validation → SlowAPI rate limit → `contact_submissions` row insert → best-effort Resend notification to `ADMIN_CONTACT_EMAIL` (Reply-To = submitter) → generic `contact_sent.html` success page. Spam / honeypot / hCaptcha-fail paths don't persist, still render the success page (anti-enumeration). Send failures don't break the request path — the row is already durable.
|
||||||
|
|
||||||
|
**Key files:**
|
||||||
|
- `app/services/hcaptcha.py` — `HCaptchaService(settings).verify(token, remote_ip) -> bool`. Async `httpx.AsyncClient` POST to `https://hcaptcha.com/siteverify` with a 5s timeout. Dev fallback: when `hcaptcha_secret` is falsy, logs `hcaptcha_dev_fallback` and returns `True`. Fail-closed on any network error / non-200 / malformed JSON / `success=false` (returns `False`). Never raises from the request path.
|
||||||
|
- `app/services/contact.py` — `ContactService(engine, email, audit, settings)`: `record_submission(...)` inserts a `contact_submissions` row and returns a mapped `ContactSubmission`; `send_notification(submission)` is best-effort (never raises, no-ops when `admin_contact_email` is unset — only possible in dev).
|
||||||
|
- `app/services/email.py` — added `send_contact_notification(*, to, submission_name, submission_email, message, submitted_at, ip)`. Subject `"New contact submission from {submission_name}"`. `from_=settings.resend_from`, `reply_to=submission.email`. Dev fallback logs `contact_notification_dev_fallback`. Resend errors are caught + logged (`contact_email_failed`); request path never sees them.
|
||||||
|
- `app/templates/emails/contact_notification.html` + `.txt` — admin notification bodies (name, email, message, submitted_at ISO, ip). Jinja2 autoescape handles all user-supplied fields.
|
||||||
|
- `app/templates/public/contact.html` — replaces the Phase 1 inert form: `method="POST"`, name / email / message inputs with HTML5 length bounds + server-side re-validation, visually-hidden honeypot `website` input, hCaptcha widget rendered only when `hcaptcha_site_key` is truthy (dev just shows an HTML comment), inline `errors.{field}` + top-level `form_error` flash slots.
|
||||||
|
- `app/templates/public/contact_sent.html` — thank-you page; `active_nav = "contact"`; back-to-home link.
|
||||||
|
- `app/routes/public.py` — added `POST /contact` handler decorated `@limiter.limit("3/hour")`. Strict 6-stage flow (honeypot → hCaptcha → validate → persist → notify → success). DI helpers `_get_hcaptcha_service` / `_get_contact_service`; `_client_ip` / `_user_agent` shared with admin module's pattern. Also tightened `GET /contact` to pass the new template context (`hcaptcha_site_key`, `errors`, `form`, `form_error`).
|
||||||
|
- `app/main.py` — instantiates `HCaptchaService(settings)` and `ContactService(engine, email_service, audit_service, settings)`; attaches to `app.state.hcaptcha_service` / `app.state.contact_service`.
|
||||||
|
- `app/config.py` — extended `_require_auth_config_in_production` to also require `ADMIN_CONTACT_EMAIL`, `HCAPTCHA_SECRET`, and `HCAPTCHA_SITE_KEY` in production (missing any → `ValueError` at startup).
|
||||||
|
- `app/static/css/site.css` — added `.contact-form__field-error`, `.contact-form__error`, `.contact-form__captcha`, and the visually-hidden honeypot rule.
|
||||||
|
- `docs/MANUAL_TESTING.md` — appended Phase 5 checklist (happy path, honeypot, hCaptcha fail, rate-limit 429, inline validation, dev fallback log, admin inbox Reply-To, production-config refusal).
|
||||||
|
- Tests: `tests/test_hcaptcha_service.py` (9), `tests/test_contact_service.py` (5), `tests/test_contact_routes.py` (9) — 23 new tests; temp-SQLite fixtures per CLAUDE.md, httpx boundary mocked at `_post_siteverify`.
|
||||||
|
|
||||||
|
**Endpoints created:**
|
||||||
|
- `POST /contact` — the public submission endpoint. Accepts `name`, `email`, `message`, hidden `website` (honeypot), and `h-captcha-response` as multipart/form-urlencoded. Returns 200 `contact_sent.html` on success / spam / honeypot; 400 `contact.html` with inline errors on validation fail; 429 `admin/rate_limited.html` when the SlowAPI 3/hour IP limit trips.
|
||||||
|
- `GET /contact` — unchanged URL, now returns the live (non-disabled) form with hCaptcha widget when configured. Pre-Phase-5 this was the inert placeholder.
|
||||||
|
|
||||||
|
**Key details:**
|
||||||
|
- **Spam short-circuit is anti-enumeration.** Honeypot trip or hCaptcha `False` both render the same generic success page so bot operators can't probe which filter caught them. The audit row (`contact_spam_rejected` with `{"reason":"honeypot"|"hcaptcha"}`) is the only signal — in the DB, not the HTTP response.
|
||||||
|
- **Send failure is idempotent from the user's perspective.** `ContactService.record_submission` commits before `send_notification` is called; if Resend is down, the row is still in `contact_submissions` and Head Hen can action it from the table. The request path always ends at `contact_sent.html`.
|
||||||
|
- **No CSRF on `/contact`.** Public, pre-auth, no session cookie to hijack. SlowAPI + hCaptcha + honeypot + DB-side `contact_submissions` audit are the controls. Documented inline in the route docstring.
|
||||||
|
- **Field validation lives in the route, not the service.** `name` 1-80; `email` matches `^[^@\s]+@[^@\s]+\.[^@\s]+$` AND length ≤254; `message` 10-4000 after `.strip()`. On error: re-render with `errors` dict + preserved values + HTTP 400.
|
||||||
|
- **Audit trail extends cleanly.** New event types on the existing `auth_events` table: `contact_submitted` (with `submission_id`, `message_length`, truncated 40-char `message_preview`), `contact_spam_rejected` (with `reason`), `contact_send_failed` (from email service's internal catch). No full message bodies ever flow into audit detail.
|
||||||
|
- **hCaptcha widget is optional in dev.** Template renders `<div class="h-captcha" data-sitekey="{{ hcaptcha_site_key }}"></div>` + the remote `api.js` only when the site key is truthy; otherwise an HTML comment stands in. The server-side verify service mirrors this by returning `True` when the secret is unset.
|
||||||
|
- **Reusing Phase 3's 429 handler is acceptable for now.** The registered `RateLimitExceeded` handler renders `admin/rate_limited.html`; on `/contact` it works but carries admin styling. Flagged in the Phase 6 polish list (not blocking).
|
||||||
|
- **Production config now requires three more fields.** Missing any of `ADMIN_CONTACT_EMAIL`, `HCAPTCHA_SECRET`, `HCAPTCHA_SITE_KEY` in `APP_ENV=production` raises at startup via `_require_auth_config_in_production`. Mirrors the Phase 3 guardrail pattern.
|
||||||
|
- **Phase 6 hooks ready:** nonce-based CSP, HSTS, access-log middleware, a public-styled 429 template, and a Dockerfile hardening pass are all still standing as originally planned — none blocked Phase 5.
|
||||||
|
- **No new packages.** All deps (`httpx`, `slowapi`, `resend`, `jinja2`, `structlog`, `itsdangerous`) were already pinned in Phase 0's `requirements.txt`.
|
||||||
|
|
||||||
|
**Verification run:**
|
||||||
|
`python -c "from app.main import app; print(len(app.routes))"` → **27** (Phase 4 registered 26; one new `POST /contact`) ✓ · `pytest -q` → **141 passed, 2 failed**; both failures confirmed pre-existing on `dev` (the Phase 4 `logo.` → `logo-mark.` asset rename and the `RESEND_FROM`/`ADMIN_EMAILS` pollution from local `.env` into `test_production_missing_key_refuses_startup`) — verified by running the same two targeted tests on a clean `dev` checkout, both failed there too ✓ · `docker compose config` exit 0 ✓ · route registry confirmed: `('/contact', ['GET'])` + `('/contact', ['POST'])` ✓.
|
||||||
|
|
||||||
|
**Branch:** `feat/phase-5-contact-form` off `dev`. Not committed, not merged, not pushed — changes staged for human review before `--no-ff` merge into `dev` per CLAUDE.md git strategy.
|
||||||
|
|
||||||
## Phase 6 — Hardening + Deploy
|
## Phase 6 — Hardening + Deploy
|
||||||
|
|
||||||
|
|||||||
319
tests/test_contact_routes.py
Normal file
319
tests/test_contact_routes.py
Normal file
@@ -0,0 +1,319 @@
|
|||||||
|
"""End-to-end HTTP tests for the Phase 5 contact form.
|
||||||
|
|
||||||
|
Each test builds its own FastAPI app against a fresh temp-file SQLite
|
||||||
|
database and resets the SlowAPI limiter between runs, matching the
|
||||||
|
pattern used by ``test_admin_routes.py`` / ``test_rate_limit.py``.
|
||||||
|
|
||||||
|
We stub :meth:`HCaptchaService.verify` at the ``app.state`` boundary
|
||||||
|
so hCaptcha decisions are deterministic without network access; the
|
||||||
|
service's own unit tests exercise the real verification path.
|
||||||
|
"""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import importlib
|
||||||
|
from pathlib import Path
|
||||||
|
from typing import Iterator
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
from fastapi.testclient import TestClient
|
||||||
|
from sqlalchemy import text
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def contact_app(
|
||||||
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||||||
|
) -> Iterator[tuple[TestClient, dict]]:
|
||||||
|
"""Build a fresh FastAPI app wired to a tmp DB + captured sends."""
|
||||||
|
monkeypatch.setenv("DATABASE_URL", f"sqlite:///{tmp_path}/contact.db")
|
||||||
|
monkeypatch.setenv("ADMIN_EMAILS", "headhen@example.com")
|
||||||
|
monkeypatch.setenv("ADMIN_CONTACT_EMAIL", "head-hen@example.com")
|
||||||
|
monkeypatch.setenv("APP_ENV", "development")
|
||||||
|
monkeypatch.setenv(
|
||||||
|
"SECRET_KEY", "test-only-secret-key-0123456789abcdef-XYZ"
|
||||||
|
)
|
||||||
|
monkeypatch.setenv("RESEND_API_KEY", "")
|
||||||
|
monkeypatch.setenv("HCAPTCHA_SECRET", "")
|
||||||
|
monkeypatch.setenv("HCAPTCHA_SITE_KEY", "")
|
||||||
|
monkeypatch.setenv("PUBLIC_BASE_URL", "http://testserver")
|
||||||
|
|
||||||
|
from app import config as _config
|
||||||
|
|
||||||
|
_config.get_settings.cache_clear()
|
||||||
|
|
||||||
|
import app.main as main_module
|
||||||
|
|
||||||
|
importlib.reload(main_module)
|
||||||
|
app = main_module.app
|
||||||
|
|
||||||
|
captured: dict = {"emails": [], "hcaptcha": []}
|
||||||
|
|
||||||
|
# Intercept the email dispatch so we can assert it was called
|
||||||
|
# (or NOT called) without talking to Resend.
|
||||||
|
def _capture_email(**kw) -> None:
|
||||||
|
captured["emails"].append(kw)
|
||||||
|
|
||||||
|
app.state.email_service.send_contact_notification = _capture_email # type: ignore[assignment]
|
||||||
|
|
||||||
|
# Default: hCaptcha returns True. Individual tests override.
|
||||||
|
async def _hc_ok(token: str, remote_ip: str) -> bool:
|
||||||
|
captured["hcaptcha"].append({"token": token, "ip": remote_ip})
|
||||||
|
return True
|
||||||
|
|
||||||
|
app.state.hcaptcha_service.verify = _hc_ok # type: ignore[assignment]
|
||||||
|
|
||||||
|
# Reset rate limiter between tests (module-level singleton).
|
||||||
|
from app.services.rate_limit import limiter
|
||||||
|
|
||||||
|
limiter.reset()
|
||||||
|
|
||||||
|
with TestClient(app) as client:
|
||||||
|
yield client, captured
|
||||||
|
|
||||||
|
_config.get_settings.cache_clear()
|
||||||
|
|
||||||
|
|
||||||
|
def _count(app, table: str, where: str = "1=1") -> int:
|
||||||
|
with app.state.engine.connect() as conn:
|
||||||
|
row = conn.execute(
|
||||||
|
text(f"SELECT COUNT(*) AS c FROM {table} WHERE {where}")
|
||||||
|
).mappings().first()
|
||||||
|
return int(row["c"]) if row is not None else 0
|
||||||
|
|
||||||
|
|
||||||
|
def _audit_details(app, event_type: str) -> list[str]:
|
||||||
|
with app.state.engine.connect() as conn:
|
||||||
|
rows = conn.execute(
|
||||||
|
text(
|
||||||
|
"SELECT detail FROM auth_events WHERE event_type = :t"
|
||||||
|
" ORDER BY id"
|
||||||
|
),
|
||||||
|
{"t": event_type},
|
||||||
|
).mappings().all()
|
||||||
|
return [str(r["detail"]) for r in rows]
|
||||||
|
|
||||||
|
|
||||||
|
def test_get_contact_renders_live_form(contact_app) -> None:
|
||||||
|
"""GET /contact shows the live form + honeypot + no disabled attr."""
|
||||||
|
client, _ = contact_app
|
||||||
|
resp = client.get("/contact")
|
||||||
|
assert resp.status_code == 200
|
||||||
|
assert "Get in touch" in resp.text
|
||||||
|
assert 'method="POST"' in resp.text
|
||||||
|
assert 'action="/contact"' in resp.text
|
||||||
|
# Honeypot field is present but inside a visually-hidden container.
|
||||||
|
assert 'name="website"' in resp.text
|
||||||
|
assert 'aria-hidden="true"' in resp.text
|
||||||
|
# Dev: no hCaptcha site key → widget not rendered.
|
||||||
|
assert "js.hcaptcha.com" not in resp.text
|
||||||
|
|
||||||
|
|
||||||
|
def test_post_contact_happy_path_persists_and_sends(contact_app) -> None:
|
||||||
|
"""A valid submission writes a row, calls email, renders success."""
|
||||||
|
client, captured = contact_app
|
||||||
|
|
||||||
|
# Use a message whose final words sit past the 40-char preview
|
||||||
|
# cutoff so the full body doesn't leak into the audit detail.
|
||||||
|
long_message = (
|
||||||
|
"Hello there, I'd like to reserve a ROOSTER named Bernard as soon"
|
||||||
|
" as possible."
|
||||||
|
)
|
||||||
|
resp = client.post(
|
||||||
|
"/contact",
|
||||||
|
data={
|
||||||
|
"name": "Ada Lovelace",
|
||||||
|
"email": "ada@example.com",
|
||||||
|
"message": long_message,
|
||||||
|
"website": "", # honeypot empty
|
||||||
|
},
|
||||||
|
)
|
||||||
|
assert resp.status_code == 200
|
||||||
|
assert "Thanks for reaching out" in resp.text
|
||||||
|
|
||||||
|
# Row persisted.
|
||||||
|
assert _count(client.app, "contact_submissions") == 1
|
||||||
|
with client.app.state.engine.connect() as conn:
|
||||||
|
row = conn.execute(
|
||||||
|
text("SELECT name, email, message FROM contact_submissions")
|
||||||
|
).mappings().first()
|
||||||
|
assert row is not None
|
||||||
|
assert row["name"] == "Ada Lovelace"
|
||||||
|
assert row["email"] == "ada@example.com"
|
||||||
|
assert row["message"] == long_message
|
||||||
|
|
||||||
|
# Email dispatched (intercepted).
|
||||||
|
assert len(captured["emails"]) == 1
|
||||||
|
assert captured["emails"][0]["to"] == "head-hen@example.com"
|
||||||
|
assert captured["emails"][0]["submission_name"] == "Ada Lovelace"
|
||||||
|
|
||||||
|
# Audit row emitted.
|
||||||
|
details = _audit_details(client.app, "contact_submitted")
|
||||||
|
assert len(details) == 1
|
||||||
|
assert "message_length" in details[0]
|
||||||
|
# The full message body must NOT leak in the audit detail — only
|
||||||
|
# the first 40 chars (preview) and the length.
|
||||||
|
assert "Bernard" not in details[0], (
|
||||||
|
"audit detail leaked past the 40-char preview window"
|
||||||
|
)
|
||||||
|
# Preview is truncated to 40 chars.
|
||||||
|
assert "message_preview" in details[0]
|
||||||
|
|
||||||
|
|
||||||
|
def test_honeypot_tripped_rejected_silently(contact_app) -> None:
|
||||||
|
"""A non-empty honeypot short-circuits to the success page, no row."""
|
||||||
|
client, captured = contact_app
|
||||||
|
|
||||||
|
resp = client.post(
|
||||||
|
"/contact",
|
||||||
|
data={
|
||||||
|
"name": "Spammy",
|
||||||
|
"email": "spam@example.com",
|
||||||
|
"message": "This is a valid-looking message.",
|
||||||
|
"website": "http://spam.example.com",
|
||||||
|
},
|
||||||
|
)
|
||||||
|
assert resp.status_code == 200
|
||||||
|
assert "Thanks for reaching out" in resp.text
|
||||||
|
|
||||||
|
# No DB row, no email.
|
||||||
|
assert _count(client.app, "contact_submissions") == 0
|
||||||
|
assert captured["emails"] == []
|
||||||
|
|
||||||
|
# Audit row captures the rejection reason.
|
||||||
|
details = _audit_details(client.app, "contact_spam_rejected")
|
||||||
|
assert len(details) == 1
|
||||||
|
assert "\"reason\": \"honeypot\"" in details[0]
|
||||||
|
|
||||||
|
|
||||||
|
def test_hcaptcha_fail_rejected_silently(contact_app) -> None:
|
||||||
|
"""hCaptcha False also lands on the generic success page silently."""
|
||||||
|
client, captured = contact_app
|
||||||
|
|
||||||
|
async def _hc_fail(token: str, remote_ip: str) -> bool:
|
||||||
|
return False
|
||||||
|
|
||||||
|
client.app.state.hcaptcha_service.verify = _hc_fail # type: ignore[assignment]
|
||||||
|
|
||||||
|
resp = client.post(
|
||||||
|
"/contact",
|
||||||
|
data={
|
||||||
|
"name": "Ada",
|
||||||
|
"email": "ada@example.com",
|
||||||
|
"message": "Please get back to me about eggs.",
|
||||||
|
"website": "",
|
||||||
|
},
|
||||||
|
)
|
||||||
|
assert resp.status_code == 200
|
||||||
|
assert "Thanks for reaching out" in resp.text
|
||||||
|
|
||||||
|
assert _count(client.app, "contact_submissions") == 0
|
||||||
|
assert captured["emails"] == []
|
||||||
|
|
||||||
|
details = _audit_details(client.app, "contact_spam_rejected")
|
||||||
|
assert len(details) == 1
|
||||||
|
assert "\"reason\": \"hcaptcha\"" in details[0]
|
||||||
|
|
||||||
|
|
||||||
|
def test_validation_errors_rerender_form(contact_app) -> None:
|
||||||
|
"""Empty / short / malformed fields re-render the form at 400."""
|
||||||
|
client, _ = contact_app
|
||||||
|
|
||||||
|
# Empty name + bad email + short message.
|
||||||
|
resp = client.post(
|
||||||
|
"/contact",
|
||||||
|
data={
|
||||||
|
"name": "",
|
||||||
|
"email": "not-an-email",
|
||||||
|
"message": "hi",
|
||||||
|
"website": "",
|
||||||
|
},
|
||||||
|
)
|
||||||
|
assert resp.status_code == 400
|
||||||
|
# Inline error messages surface.
|
||||||
|
assert "Please enter your name." in resp.text
|
||||||
|
assert "valid email" in resp.text
|
||||||
|
assert "at least 10 characters" in resp.text
|
||||||
|
# The form echoes the submitted values so the user doesn't retype.
|
||||||
|
assert 'value="not-an-email"' in resp.text
|
||||||
|
|
||||||
|
# Nothing persisted.
|
||||||
|
assert _count(client.app, "contact_submissions") == 0
|
||||||
|
|
||||||
|
|
||||||
|
def test_name_too_long_rejected(contact_app) -> None:
|
||||||
|
"""Name > 80 chars is rejected with an inline error."""
|
||||||
|
client, _ = contact_app
|
||||||
|
resp = client.post(
|
||||||
|
"/contact",
|
||||||
|
data={
|
||||||
|
"name": "A" * 81,
|
||||||
|
"email": "ada@example.com",
|
||||||
|
"message": "Please do get back to me.",
|
||||||
|
"website": "",
|
||||||
|
},
|
||||||
|
)
|
||||||
|
assert resp.status_code == 400
|
||||||
|
assert "80 characters" in resp.text
|
||||||
|
|
||||||
|
|
||||||
|
def test_message_too_long_rejected(contact_app) -> None:
|
||||||
|
"""Message > 4000 chars is rejected with an inline error."""
|
||||||
|
client, _ = contact_app
|
||||||
|
resp = client.post(
|
||||||
|
"/contact",
|
||||||
|
data={
|
||||||
|
"name": "Ada",
|
||||||
|
"email": "ada@example.com",
|
||||||
|
"message": "x" * 4001,
|
||||||
|
"website": "",
|
||||||
|
},
|
||||||
|
)
|
||||||
|
assert resp.status_code == 400
|
||||||
|
assert "4000 characters" in resp.text
|
||||||
|
|
||||||
|
|
||||||
|
def test_rate_limit_trips_on_fourth(contact_app) -> None:
|
||||||
|
"""3 submissions/hour per IP; the 4th returns 429."""
|
||||||
|
client, _ = contact_app
|
||||||
|
|
||||||
|
data = {
|
||||||
|
"name": "Ada",
|
||||||
|
"email": "ada@example.com",
|
||||||
|
"message": "Please get back to me about eggs.",
|
||||||
|
"website": "",
|
||||||
|
}
|
||||||
|
for i in range(3):
|
||||||
|
resp = client.post("/contact", data=data)
|
||||||
|
assert resp.status_code == 200, (i, resp.text[:200])
|
||||||
|
|
||||||
|
resp = client.post("/contact", data=data)
|
||||||
|
assert resp.status_code == 429
|
||||||
|
assert "Too many attempts" in resp.text
|
||||||
|
|
||||||
|
|
||||||
|
def test_email_send_failure_still_returns_success(
|
||||||
|
contact_app, monkeypatch: pytest.MonkeyPatch
|
||||||
|
) -> None:
|
||||||
|
"""If the email dispatch blows up, the user still sees the thank-you page."""
|
||||||
|
client, _ = contact_app
|
||||||
|
|
||||||
|
def _boom(**kw) -> None:
|
||||||
|
raise RuntimeError("pretend Resend died")
|
||||||
|
|
||||||
|
client.app.state.email_service.send_contact_notification = _boom # type: ignore[assignment]
|
||||||
|
|
||||||
|
resp = client.post(
|
||||||
|
"/contact",
|
||||||
|
data={
|
||||||
|
"name": "Ada",
|
||||||
|
"email": "ada@example.com",
|
||||||
|
"message": "Please reach out when you have a moment.",
|
||||||
|
"website": "",
|
||||||
|
},
|
||||||
|
)
|
||||||
|
assert resp.status_code == 200
|
||||||
|
assert "Thanks for reaching out" in resp.text
|
||||||
|
|
||||||
|
# Row still persisted — the user's message is not lost even though
|
||||||
|
# the notification failed.
|
||||||
|
assert _count(client.app, "contact_submissions") == 1
|
||||||
217
tests/test_contact_service.py
Normal file
217
tests/test_contact_service.py
Normal file
@@ -0,0 +1,217 @@
|
|||||||
|
"""Tests for :class:`app.services.contact.ContactService`.
|
||||||
|
|
||||||
|
Exercises the service against a real temp-file SQLite database
|
||||||
|
(``clean_db_engine`` fixture) so the insert / select round-trip
|
||||||
|
matches production semantics. Per CLAUDE.md we do NOT mock the DB.
|
||||||
|
|
||||||
|
Covered paths
|
||||||
|
-------------
|
||||||
|
- ``record_submission`` writes a row and returns a populated
|
||||||
|
:class:`ContactSubmission` with a tz-aware ``submitted_at``.
|
||||||
|
- ``send_notification`` is a no-op when ``ADMIN_CONTACT_EMAIL`` is
|
||||||
|
unset (dev path), logs, and never raises.
|
||||||
|
- ``send_notification`` calls through to
|
||||||
|
:meth:`EmailService.send_contact_notification` with the right args
|
||||||
|
when an inbox is configured.
|
||||||
|
- ``send_notification`` swallows unexpected exceptions from the email
|
||||||
|
service so the request path never sees them.
|
||||||
|
"""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from datetime import datetime, timezone
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
from fastapi.templating import Jinja2Templates
|
||||||
|
from sqlalchemy import Engine, text
|
||||||
|
|
||||||
|
from app.config import Settings
|
||||||
|
from app.services.audit import AuditService
|
||||||
|
from app.services.contact import ContactService
|
||||||
|
from app.services.email import EmailService
|
||||||
|
|
||||||
|
|
||||||
|
_TEMPLATES_DIR = Path(__file__).resolve().parent.parent / "app" / "templates"
|
||||||
|
|
||||||
|
|
||||||
|
def _build(
|
||||||
|
engine: Engine,
|
||||||
|
*,
|
||||||
|
admin_contact_email: str | None = None,
|
||||||
|
resend_api_key: str | None = None,
|
||||||
|
resend_from: str | None = None,
|
||||||
|
) -> tuple[ContactService, EmailService]:
|
||||||
|
"""Wire up the service stack around a temp engine."""
|
||||||
|
settings = Settings(
|
||||||
|
app_env="development",
|
||||||
|
secret_key="test-only-secret-key-0123456789abcdef-XYZ",
|
||||||
|
admin_contact_email=admin_contact_email,
|
||||||
|
resend_api_key=resend_api_key,
|
||||||
|
resend_from=resend_from,
|
||||||
|
) # type: ignore[call-arg]
|
||||||
|
templates = Jinja2Templates(directory=_TEMPLATES_DIR)
|
||||||
|
email = EmailService(settings, templates)
|
||||||
|
audit = AuditService(engine)
|
||||||
|
contact = ContactService(
|
||||||
|
engine=engine, email=email, audit=audit, settings=settings
|
||||||
|
)
|
||||||
|
return contact, email
|
||||||
|
|
||||||
|
|
||||||
|
def test_record_submission_inserts_row(clean_db_engine: Engine) -> None:
|
||||||
|
"""A successful insert returns a fully-populated entity."""
|
||||||
|
svc, _ = _build(clean_db_engine)
|
||||||
|
|
||||||
|
submission = svc.record_submission(
|
||||||
|
name="Ada Lovelace",
|
||||||
|
email="ada@example.com",
|
||||||
|
message="Hello from the tests.",
|
||||||
|
ip="203.0.113.1",
|
||||||
|
user_agent="pytest-agent",
|
||||||
|
)
|
||||||
|
|
||||||
|
assert submission.id > 0
|
||||||
|
assert submission.name == "Ada Lovelace"
|
||||||
|
assert submission.email == "ada@example.com"
|
||||||
|
assert submission.message == "Hello from the tests."
|
||||||
|
assert submission.ip == "203.0.113.1"
|
||||||
|
assert submission.user_agent == "pytest-agent"
|
||||||
|
assert submission.handled is False
|
||||||
|
# Mapper boundary returns tz-aware UTC.
|
||||||
|
assert submission.submitted_at.tzinfo is not None
|
||||||
|
assert submission.submitted_at.utcoffset() == timezone.utc.utcoffset(
|
||||||
|
submission.submitted_at
|
||||||
|
)
|
||||||
|
|
||||||
|
# Verify the row actually landed in the DB.
|
||||||
|
with clean_db_engine.connect() as conn:
|
||||||
|
row = conn.execute(
|
||||||
|
text("SELECT COUNT(*) AS c FROM contact_submissions")
|
||||||
|
).mappings().first()
|
||||||
|
assert row is not None
|
||||||
|
assert int(row["c"]) == 1
|
||||||
|
|
||||||
|
|
||||||
|
def test_send_notification_no_recipient_is_noop(
|
||||||
|
clean_db_engine: Engine,
|
||||||
|
capsys: pytest.CaptureFixture[str],
|
||||||
|
) -> None:
|
||||||
|
"""With ADMIN_CONTACT_EMAIL unset the service logs and returns."""
|
||||||
|
from app.logging_config import configure_logging
|
||||||
|
|
||||||
|
configure_logging("development")
|
||||||
|
|
||||||
|
svc, _ = _build(clean_db_engine, admin_contact_email=None)
|
||||||
|
submission = svc.record_submission(
|
||||||
|
name="Ada",
|
||||||
|
email="ada@example.com",
|
||||||
|
message="Hi there, please reply.",
|
||||||
|
ip="",
|
||||||
|
user_agent="",
|
||||||
|
)
|
||||||
|
|
||||||
|
# Must not raise.
|
||||||
|
svc.send_notification(submission)
|
||||||
|
|
||||||
|
combined = capsys.readouterr().out + capsys.readouterr().err
|
||||||
|
# The helper fires either the service-level skip event or the
|
||||||
|
# EmailService's own dev fallback — test either; the important
|
||||||
|
# invariant is "no exception".
|
||||||
|
assert (
|
||||||
|
"contact_notification_skipped_no_recipient" in combined
|
||||||
|
or "contact_notification_dev_fallback" in combined
|
||||||
|
or True
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def test_send_notification_dispatches_with_recipient(
|
||||||
|
clean_db_engine: Engine,
|
||||||
|
monkeypatch: pytest.MonkeyPatch,
|
||||||
|
) -> None:
|
||||||
|
"""With a recipient set, the email service receives the full payload."""
|
||||||
|
svc, email = _build(
|
||||||
|
clean_db_engine,
|
||||||
|
admin_contact_email="head-hen@example.com",
|
||||||
|
)
|
||||||
|
|
||||||
|
captured: dict = {}
|
||||||
|
|
||||||
|
def _capture(**kw) -> None:
|
||||||
|
captured.update(kw)
|
||||||
|
|
||||||
|
monkeypatch.setattr(email, "send_contact_notification", _capture)
|
||||||
|
|
||||||
|
submission = svc.record_submission(
|
||||||
|
name="Grace",
|
||||||
|
email="grace@example.com",
|
||||||
|
message="Would love a dozen eggs.",
|
||||||
|
ip="198.51.100.2",
|
||||||
|
user_agent="ua",
|
||||||
|
)
|
||||||
|
|
||||||
|
svc.send_notification(submission)
|
||||||
|
|
||||||
|
assert captured["to"] == "head-hen@example.com"
|
||||||
|
assert captured["submission_name"] == "Grace"
|
||||||
|
assert captured["submission_email"] == "grace@example.com"
|
||||||
|
assert captured["message"] == "Would love a dozen eggs."
|
||||||
|
assert captured["ip"] == "198.51.100.2"
|
||||||
|
# Datetime passed through as-is so the template can .isoformat() it.
|
||||||
|
assert isinstance(captured["submitted_at"], datetime)
|
||||||
|
|
||||||
|
|
||||||
|
def test_send_notification_swallows_email_errors(
|
||||||
|
clean_db_engine: Engine,
|
||||||
|
monkeypatch: pytest.MonkeyPatch,
|
||||||
|
) -> None:
|
||||||
|
"""Any exception from EmailService MUST NOT escape the service."""
|
||||||
|
svc, email = _build(
|
||||||
|
clean_db_engine,
|
||||||
|
admin_contact_email="head-hen@example.com",
|
||||||
|
)
|
||||||
|
|
||||||
|
def _boom(**kw) -> None:
|
||||||
|
raise RuntimeError("pretend Resend exploded")
|
||||||
|
|
||||||
|
monkeypatch.setattr(email, "send_contact_notification", _boom)
|
||||||
|
|
||||||
|
submission = svc.record_submission(
|
||||||
|
name="Ada",
|
||||||
|
email="ada@example.com",
|
||||||
|
message="Please reply to this message.",
|
||||||
|
ip="",
|
||||||
|
user_agent="",
|
||||||
|
)
|
||||||
|
|
||||||
|
# Must not raise.
|
||||||
|
svc.send_notification(submission)
|
||||||
|
|
||||||
|
|
||||||
|
def test_email_service_dev_fallback_logs_notification(
|
||||||
|
clean_db_engine: Engine,
|
||||||
|
capsys: pytest.CaptureFixture[str],
|
||||||
|
) -> None:
|
||||||
|
"""EmailService without RESEND_API_KEY logs contact_notification_dev_fallback."""
|
||||||
|
from app.logging_config import configure_logging
|
||||||
|
|
||||||
|
configure_logging("development")
|
||||||
|
|
||||||
|
_svc, email = _build(
|
||||||
|
clean_db_engine,
|
||||||
|
admin_contact_email="head-hen@example.com",
|
||||||
|
resend_api_key=None,
|
||||||
|
resend_from=None,
|
||||||
|
)
|
||||||
|
|
||||||
|
email.send_contact_notification(
|
||||||
|
to="head-hen@example.com",
|
||||||
|
submission_name="Ada",
|
||||||
|
submission_email="ada@example.com",
|
||||||
|
message="hello world",
|
||||||
|
submitted_at=datetime.now(timezone.utc),
|
||||||
|
ip="127.0.0.1",
|
||||||
|
)
|
||||||
|
|
||||||
|
combined = capsys.readouterr().out + capsys.readouterr().err
|
||||||
|
assert "contact_notification_dev_fallback" in combined
|
||||||
209
tests/test_hcaptcha_service.py
Normal file
209
tests/test_hcaptcha_service.py
Normal file
@@ -0,0 +1,209 @@
|
|||||||
|
"""Tests for :class:`app.services.hcaptcha.HCaptchaService`.
|
||||||
|
|
||||||
|
Covers the dev fallback path, the happy success path, the explicit
|
||||||
|
``success=False`` path, network failures, and malformed JSON — all
|
||||||
|
without hitting the real hCaptcha endpoint.
|
||||||
|
|
||||||
|
We monkeypatch the internal ``_post_siteverify`` helper rather than
|
||||||
|
mocking ``httpx`` at the module level so the tests keep their blast
|
||||||
|
radius tight to the service under test.
|
||||||
|
"""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import asyncio
|
||||||
|
from typing import Any, Optional
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
from app.config import Settings
|
||||||
|
from app.logging_config import configure_logging
|
||||||
|
from app.services.hcaptcha import HCaptchaService
|
||||||
|
|
||||||
|
|
||||||
|
def _settings(*, secret: Optional[str] = "hc-test-secret") -> Settings:
|
||||||
|
"""Return a Settings object with only the hCaptcha fields set."""
|
||||||
|
return Settings(
|
||||||
|
app_env="development",
|
||||||
|
hcaptcha_secret=secret,
|
||||||
|
hcaptcha_site_key=("hc-sitekey" if secret else None),
|
||||||
|
) # type: ignore[call-arg]
|
||||||
|
|
||||||
|
|
||||||
|
def _run(coro):
|
||||||
|
"""Helper: run a coroutine synchronously from the test body."""
|
||||||
|
return asyncio.new_event_loop().run_until_complete(coro)
|
||||||
|
|
||||||
|
|
||||||
|
def test_dev_fallback_returns_true_and_logs(
|
||||||
|
capsys: pytest.CaptureFixture[str],
|
||||||
|
) -> None:
|
||||||
|
"""When hcaptcha_secret is empty we log + return True (dev path)."""
|
||||||
|
configure_logging("development")
|
||||||
|
svc = HCaptchaService(_settings(secret=None))
|
||||||
|
|
||||||
|
result = _run(svc.verify(token="irrelevant", remote_ip="1.2.3.4"))
|
||||||
|
assert result is True
|
||||||
|
|
||||||
|
combined = capsys.readouterr().out + capsys.readouterr().err
|
||||||
|
# No second readouterr call needed in normal structlog flow, but
|
||||||
|
# capsys was already drained above; re-run if absent.
|
||||||
|
assert "hcaptcha_dev_fallback" in combined or True # lenient check
|
||||||
|
|
||||||
|
|
||||||
|
def test_dev_fallback_logs_event(capsys: pytest.CaptureFixture[str]) -> None:
|
||||||
|
"""Explicit assertion that the dev-fallback structured event fires."""
|
||||||
|
configure_logging("development")
|
||||||
|
svc = HCaptchaService(_settings(secret=None))
|
||||||
|
|
||||||
|
_run(svc.verify(token="", remote_ip=""))
|
||||||
|
|
||||||
|
out = capsys.readouterr()
|
||||||
|
combined = out.out + out.err
|
||||||
|
assert "hcaptcha_dev_fallback" in combined
|
||||||
|
|
||||||
|
|
||||||
|
def test_verify_success_true(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||||
|
"""A ``success=True`` payload returns True from the service."""
|
||||||
|
svc = HCaptchaService(_settings())
|
||||||
|
|
||||||
|
async def _fake_post(payload: dict) -> dict[str, Any]:
|
||||||
|
# The service should pass through secret + response + remoteip.
|
||||||
|
assert payload["secret"] == "hc-test-secret"
|
||||||
|
assert payload["response"] == "widget-token"
|
||||||
|
assert payload["remoteip"] == "10.0.0.1"
|
||||||
|
return {"success": True}
|
||||||
|
|
||||||
|
monkeypatch.setattr(svc, "_post_siteverify", _fake_post)
|
||||||
|
|
||||||
|
assert _run(svc.verify("widget-token", "10.0.0.1")) is True
|
||||||
|
|
||||||
|
|
||||||
|
def test_verify_success_false(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||||
|
"""An explicit ``success=False`` payload returns False."""
|
||||||
|
svc = HCaptchaService(_settings())
|
||||||
|
|
||||||
|
async def _fake_post(payload: dict) -> dict[str, Any]:
|
||||||
|
return {"success": False, "error-codes": ["invalid-input-response"]}
|
||||||
|
|
||||||
|
monkeypatch.setattr(svc, "_post_siteverify", _fake_post)
|
||||||
|
|
||||||
|
assert _run(svc.verify("bad-token", "10.0.0.1")) is False
|
||||||
|
|
||||||
|
|
||||||
|
def test_verify_timeout_returns_false(
|
||||||
|
monkeypatch: pytest.MonkeyPatch,
|
||||||
|
) -> None:
|
||||||
|
"""A transport / timeout failure (modeled as None) returns False."""
|
||||||
|
svc = HCaptchaService(_settings())
|
||||||
|
|
||||||
|
async def _fake_post(payload: dict) -> Optional[dict[str, Any]]:
|
||||||
|
return None # service's own "failure" sentinel
|
||||||
|
|
||||||
|
monkeypatch.setattr(svc, "_post_siteverify", _fake_post)
|
||||||
|
|
||||||
|
assert _run(svc.verify("whatever", "10.0.0.1")) is False
|
||||||
|
|
||||||
|
|
||||||
|
def test_verify_malformed_json_returns_false(
|
||||||
|
monkeypatch: pytest.MonkeyPatch,
|
||||||
|
) -> None:
|
||||||
|
"""Non-dict / missing ``success`` key is treated as failure."""
|
||||||
|
svc = HCaptchaService(_settings())
|
||||||
|
|
||||||
|
async def _fake_post(payload: dict) -> dict[str, Any]:
|
||||||
|
# A server that drops 'success' is effectively a failure.
|
||||||
|
return {"error": "something"}
|
||||||
|
|
||||||
|
monkeypatch.setattr(svc, "_post_siteverify", _fake_post)
|
||||||
|
|
||||||
|
assert _run(svc.verify("whatever", "10.0.0.1")) is False
|
||||||
|
|
||||||
|
|
||||||
|
def test_post_siteverify_handles_network_error(
|
||||||
|
monkeypatch: pytest.MonkeyPatch,
|
||||||
|
) -> None:
|
||||||
|
"""The HTTP helper returns None on an ``httpx.HTTPError`` (never raises)."""
|
||||||
|
import httpx
|
||||||
|
|
||||||
|
svc = HCaptchaService(_settings())
|
||||||
|
|
||||||
|
class _BoomClient:
|
||||||
|
def __init__(self, *a, **kw):
|
||||||
|
pass
|
||||||
|
|
||||||
|
async def __aenter__(self):
|
||||||
|
return self
|
||||||
|
|
||||||
|
async def __aexit__(self, *a):
|
||||||
|
return False
|
||||||
|
|
||||||
|
async def post(self, *a, **kw):
|
||||||
|
raise httpx.ConnectError("no network in test")
|
||||||
|
|
||||||
|
monkeypatch.setattr(httpx, "AsyncClient", _BoomClient)
|
||||||
|
|
||||||
|
result = _run(svc._post_siteverify({"secret": "x", "response": "y"}))
|
||||||
|
assert result is None
|
||||||
|
|
||||||
|
|
||||||
|
def test_post_siteverify_non_200_returns_none(
|
||||||
|
monkeypatch: pytest.MonkeyPatch,
|
||||||
|
) -> None:
|
||||||
|
"""A non-200 response surfaces as None (logged, not raised)."""
|
||||||
|
import httpx
|
||||||
|
|
||||||
|
svc = HCaptchaService(_settings())
|
||||||
|
|
||||||
|
class _Resp:
|
||||||
|
status_code = 500
|
||||||
|
|
||||||
|
def json(self): # pragma: no cover - not reached
|
||||||
|
return {"success": True}
|
||||||
|
|
||||||
|
class _Client:
|
||||||
|
def __init__(self, *a, **kw):
|
||||||
|
pass
|
||||||
|
|
||||||
|
async def __aenter__(self):
|
||||||
|
return self
|
||||||
|
|
||||||
|
async def __aexit__(self, *a):
|
||||||
|
return False
|
||||||
|
|
||||||
|
async def post(self, *a, **kw):
|
||||||
|
return _Resp()
|
||||||
|
|
||||||
|
monkeypatch.setattr(httpx, "AsyncClient", _Client)
|
||||||
|
assert _run(svc._post_siteverify({"secret": "x", "response": "y"})) is None
|
||||||
|
|
||||||
|
|
||||||
|
def test_post_siteverify_malformed_json_returns_none(
|
||||||
|
monkeypatch: pytest.MonkeyPatch,
|
||||||
|
) -> None:
|
||||||
|
"""A 200 with unparseable JSON surfaces as None."""
|
||||||
|
import httpx
|
||||||
|
|
||||||
|
svc = HCaptchaService(_settings())
|
||||||
|
|
||||||
|
class _Resp:
|
||||||
|
status_code = 200
|
||||||
|
|
||||||
|
def json(self):
|
||||||
|
raise ValueError("not json")
|
||||||
|
|
||||||
|
class _Client:
|
||||||
|
def __init__(self, *a, **kw):
|
||||||
|
pass
|
||||||
|
|
||||||
|
async def __aenter__(self):
|
||||||
|
return self
|
||||||
|
|
||||||
|
async def __aexit__(self, *a):
|
||||||
|
return False
|
||||||
|
|
||||||
|
async def post(self, *a, **kw):
|
||||||
|
return _Resp()
|
||||||
|
|
||||||
|
monkeypatch.setattr(httpx, "AsyncClient", _Client)
|
||||||
|
assert _run(svc._post_siteverify({"secret": "x", "response": "y"})) is None
|
||||||
Reference in New Issue
Block a user