From f956f0a260f210b5a2625f475d5956c9febd3b80 Mon Sep 17 00:00:00 2001 From: Roberto Musso Date: Sat, 11 Apr 2026 18:48:58 +0200 Subject: [PATCH] feat: Brevo double opt-in + contact sync - Add brevo.py: transactional email sending + contact list sync via Brevo API - Add token.py: stateless HMAC-signed confirmation tokens (no DB migration needed) - Update routes.py: POST /waitlist sends confirmation email, GET /waitlist/confirm verifies token - Update config.py: Brevo + confirmation settings (gracefully disabled when BREVO_API_KEY is empty) - Update .env.example with new Brevo and confirmation variables - Add httpx dependency - Add 8 new tests (token roundtrip/expiry/tamper, confirm endpoint, Brevo mock) --- .env.example | 11 ++ app/brevo.py | 224 +++++++++++++++++++++++++++++++++++++++++ app/config.py | 17 ++++ app/routes.py | 100 +++++++++++++++++- app/token.py | 60 +++++++++++ requirements.txt | 1 + tests/test_waitlist.py | 119 +++++++++++++++++++++- 7 files changed, 528 insertions(+), 4 deletions(-) create mode 100644 app/brevo.py create mode 100644 app/token.py diff --git a/.env.example b/.env.example index 6f72742..f280375 100644 --- a/.env.example +++ b/.env.example @@ -9,3 +9,14 @@ RATE_LIMIT_PER_MINUTE=5 # Set to "production" in prod to enforce strict origin checks ENVIRONMENT=development + +# Brevo (email) — leave BREVO_API_KEY empty to disable email features +BREVO_API_KEY= +BREVO_SENDER_EMAIL=noreply@adiuvai.com +BREVO_SENDER_NAME=adiuvAI +BREVO_LIST_ID=0 + +# Confirmation link +CONFIRM_SECRET=replace-with-a-long-random-string +CONFIRM_BASE_URL=https://waitlist.adiuvai.com +CONFIRM_TOKEN_EXPIRY_HOURS=48 diff --git a/app/brevo.py b/app/brevo.py new file mode 100644 index 0000000..aa7a57c --- /dev/null +++ b/app/brevo.py @@ -0,0 +1,224 @@ +""" +Brevo (ex-Sendinblue) integration. + +- Send transactional confirmation emails +- Sync confirmed contacts to a Brevo list +""" + +import logging + +import httpx + +from app.config import settings + +logger = logging.getLogger(__name__) + +BREVO_API = "https://api.brevo.com/v3" + + +def _headers() -> dict[str, str]: + return { + "api-key": settings.BREVO_API_KEY, + "Content-Type": "application/json", + "Accept": "application/json", + } + + +async def send_confirmation_email(email: str, confirm_url: str) -> bool: + """Send a double opt-in confirmation email. Returns True on success.""" + if not settings.brevo_configured: + logger.warning("Brevo not configured — skipping confirmation email for %s***", email[:3]) + return False + + payload = { + "sender": { + "name": settings.BREVO_SENDER_NAME, + "email": settings.BREVO_SENDER_EMAIL, + }, + "to": [{"email": email}], + "subject": "Confirm your spot on the adiuvAI waitlist", + "htmlContent": _confirmation_html(confirm_url), + } + + try: + async with httpx.AsyncClient(timeout=10) as client: + resp = await client.post(f"{BREVO_API}/smtp/email", headers=_headers(), json=payload) + resp.raise_for_status() + logger.info("Confirmation email sent to %s***", email[:3]) + return True + except httpx.HTTPError: + logger.exception("Failed to send confirmation email to %s***", email[:3]) + return False + + +async def add_contact_to_list(email: str) -> bool: + """Add a confirmed contact to the Brevo waitlist list. Returns True on success.""" + if not settings.brevo_configured: + logger.warning("Brevo not configured — skipping contact sync for %s***", email[:3]) + return False + + if settings.BREVO_LIST_ID == 0: + logger.warning("BREVO_LIST_ID not set — skipping contact sync") + return False + + payload = { + "email": email, + "listIds": [settings.BREVO_LIST_ID], + "updateEnabled": True, + } + + try: + async with httpx.AsyncClient(timeout=10) as client: + resp = await client.post(f"{BREVO_API}/contacts", headers=_headers(), json=payload) + resp.raise_for_status() + logger.info("Contact synced to Brevo list %d: %s***", settings.BREVO_LIST_ID, email[:3]) + return True + except httpx.HTTPError: + logger.exception("Failed to sync contact to Brevo: %s***", email[:3]) + return False + + +def _confirmation_html(confirm_url: str) -> str: + """Email template aligned with the adiuvAI landing page brand.""" + return f"""\ + + + + + + + + + Confirm your spot — adiuvAI + + + + + +
+ Confirm your email to secure your early access spot on the adiuvAI waitlist. +
+ + + + +
+ + + + + + + + + + + + + + + + + + + +
 
+ + + + + + +
+ + + + + + + + + + adiuvAI +
+
+ + + + +
+ ● One more step +
+ + +

+ Confirm your spot on
the waitlist +

+ + +

+ Thanks for signing up! Please confirm your email address + so we can keep you in the loop when adiuvAI launches. +

+ + + + +
+ + Confirm my email + +
+ + +

+ This link expires in {settings.CONFIRM_TOKEN_EXPIRY_HOURS} hours. +

+ +
+ + +
 
+
+

+ If you didn't sign up, simply ignore this email. +

+

+ adiuvai.com +

+
+ + +
+ + +""" diff --git a/app/config.py b/app/config.py index 8e97382..9ae62d7 100644 --- a/app/config.py +++ b/app/config.py @@ -1,3 +1,5 @@ +import secrets + from pydantic_settings import BaseSettings @@ -7,6 +9,17 @@ class Settings(BaseSettings): RATE_LIMIT_PER_MINUTE: int = 5 ENVIRONMENT: str = "development" + # Brevo (email) + BREVO_API_KEY: str = "" + BREVO_SENDER_EMAIL: str = "noreply@adiuvai.com" + BREVO_SENDER_NAME: str = "adiuvAI" + BREVO_LIST_ID: int = 0 # Brevo contact list ID for waitlist subscribers + + # Confirmation link + CONFIRM_SECRET: str = secrets.token_hex(32) # override in production .env + CONFIRM_BASE_URL: str = "https://waitlist.adiuvai.com" + CONFIRM_TOKEN_EXPIRY_HOURS: int = 48 + model_config = {"env_file": ".env", "env_file_encoding": "utf-8"} @property @@ -17,5 +30,9 @@ class Settings(BaseSettings): def sync_database_url(self) -> str: return self.DATABASE_URL.replace("+asyncpg", "+psycopg2") + @property + def brevo_configured(self) -> bool: + return bool(self.BREVO_API_KEY) + settings = Settings() diff --git a/app/routes.py b/app/routes.py index 279f15e..4b2e5e9 100644 --- a/app/routes.py +++ b/app/routes.py @@ -1,13 +1,19 @@ +import asyncio import logging +from urllib.parse import urlencode from fastapi import APIRouter, Depends, Request +from fastapi.responses import HTMLResponse from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession +from app.brevo import add_contact_to_list, send_confirmation_email +from app.config import settings from app.db import get_db from app.rate_limit import _get_client_ip from app.schemas import WaitlistRequest, WaitlistResponse from app.models import WaitlistEntry +from app.token import generate_token, verify_token logger = logging.getLogger(__name__) router = APIRouter() @@ -25,6 +31,7 @@ async def join_waitlist( - Honeypot: if `website` field is non-empty, silently succeed (bot trap). - Duplicate emails: idempotent — returns success without error. - Stores the Cloudflare-resolved client IP for analytics (not exposed). + - Sends a confirmation email via Brevo (fire-and-forget). """ # Honeypot — bots fill hidden fields; silently "succeed" if body.website: @@ -35,14 +42,101 @@ async def join_waitlist( # Check for existing entry — idempotent existing = await db.execute( - select(WaitlistEntry.id).where(WaitlistEntry.email == email) + select(WaitlistEntry).where(WaitlistEntry.email == email) ) - if existing.scalar_one_or_none() is not None: + entry = existing.scalar_one_or_none() + + if entry is not None: + # Re-send confirmation if not yet confirmed + if not entry.confirmed and settings.brevo_configured: + token = generate_token(email) + confirm_url = f"{settings.CONFIRM_BASE_URL}/api/v1/waitlist/confirm?token={token}" + asyncio.create_task(send_confirmation_email(email, confirm_url)) return WaitlistResponse() entry = WaitlistEntry(email=email, ip_address=ip, source="website") db.add(entry) await db.commit() - logger.info("New waitlist signup: %s", email[:3] + "***") + logger.info("New waitlist signup: %s***", email[:3]) + + # Fire-and-forget: send confirmation email via Brevo + if settings.brevo_configured: + token = generate_token(email) + confirm_url = f"{settings.CONFIRM_BASE_URL}/api/v1/waitlist/confirm?token={token}" + asyncio.create_task(send_confirmation_email(email, confirm_url)) + return WaitlistResponse() + + +@router.get("/waitlist/confirm", response_class=HTMLResponse) +async def confirm_email( + token: str, + db: AsyncSession = Depends(get_db), +) -> HTMLResponse: + """ + Double opt-in confirmation endpoint. + + Verifies the HMAC token, marks the entry as confirmed, + and syncs the contact to Brevo. + """ + email = verify_token(token) + if email is None: + return HTMLResponse(content=_result_page(success=False), status_code=400) + + result = await db.execute( + select(WaitlistEntry).where(WaitlistEntry.email == email) + ) + entry = result.scalar_one_or_none() + if entry is None: + return HTMLResponse(content=_result_page(success=False), status_code=400) + + if not entry.confirmed: + entry.confirmed = True + await db.commit() + logger.info("Email confirmed: %s***", email[:3]) + + # Sync confirmed contact to Brevo list (fire-and-forget) + if settings.brevo_configured: + asyncio.create_task(add_contact_to_list(email)) + + return HTMLResponse(content=_result_page(success=True)) + + +def _result_page(*, success: bool) -> str: + """Minimal HTML response for confirmation result.""" + if success: + title = "You're confirmed!" + message = "Your email has been verified. We'll notify you when adiuvAI is ready." + color = "#16a34a" + else: + title = "Invalid or expired link" + message = "This confirmation link is no longer valid. Please sign up again." + color = "#dc2626" + + return f"""\ + + + + + + {title} — adiuvAI + + +
+

adiuvAI

+
+ {'✓' if success else '✕'} +
+

{title}

+

{message}

+ + Go to adiuvai.com + +
+ +""" diff --git a/app/token.py b/app/token.py new file mode 100644 index 0000000..1496c9c --- /dev/null +++ b/app/token.py @@ -0,0 +1,60 @@ +""" +Stateless HMAC-signed confirmation tokens. + +Token format: base64url( email : unix_timestamp : hmac_sha256 ) +No database storage needed — the signature proves authenticity. +""" + +import base64 +import hashlib +import hmac +import time + +from app.config import settings + + +def generate_token(email: str) -> str: + """Create a URL-safe confirmation token for the given email.""" + timestamp = str(int(time.time())) + payload = f"{email}:{timestamp}" + sig = hmac.new( + settings.CONFIRM_SECRET.encode(), + payload.encode(), + hashlib.sha256, + ).hexdigest() + return base64.urlsafe_b64encode(f"{payload}:{sig}".encode()).decode() + + +def verify_token(token: str) -> str | None: + """ + Verify a confirmation token. Returns the email if valid, None otherwise. + Checks both HMAC signature and expiry. + """ + try: + decoded = base64.urlsafe_b64decode(token.encode()).decode() + # rsplit from the right: email may contain colons (unlikely but safe) + parts = decoded.rsplit(":", 2) + if len(parts) != 3: + return None + + email, timestamp, sig = parts + + # Verify HMAC + expected = hmac.new( + settings.CONFIRM_SECRET.encode(), + f"{email}:{timestamp}".encode(), + hashlib.sha256, + ).hexdigest() + if not hmac.compare_digest(sig, expected): + return None + + # Check expiry + age_seconds = time.time() - int(timestamp) + if age_seconds > settings.CONFIRM_TOKEN_EXPIRY_HOURS * 3600: + return None + if age_seconds < 0: + return None # future timestamp — tampered + + return email + except Exception: + return None diff --git a/requirements.txt b/requirements.txt index 2e8d622..5d4ecde 100644 --- a/requirements.txt +++ b/requirements.txt @@ -8,3 +8,4 @@ pydantic>=2.0,<3.0 pydantic-settings>=2.0,<3.0 email-validator>=2.0,<3.0 python-dotenv>=1.0,<2.0 +httpx>=0.27,<1.0 diff --git a/tests/test_waitlist.py b/tests/test_waitlist.py index 8d9a70f..8d95f6e 100644 --- a/tests/test_waitlist.py +++ b/tests/test_waitlist.py @@ -1,12 +1,17 @@ +import time +from unittest.mock import AsyncMock, patch + import pytest import pytest_asyncio from httpx import ASGITransport, AsyncClient +from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine from app.main import app -from app.models import Base +from app.models import Base, WaitlistEntry from app.db import get_db from app.rate_limit import _hits_store +from app.token import generate_token, verify_token # Use SQLite for tests (no Postgres dependency) TEST_DB_URL = "sqlite+aiosqlite:///./test_waitlist.db" @@ -107,3 +112,115 @@ async def test_rate_limit(client): # The 6th request should be rate-limited (limit is 5) assert resp.status_code == 429 assert "Retry-After" in resp.headers + + +# ── Confirmation token tests ───────────────────────────────────────── + + +@pytest.mark.asyncio +async def test_token_roundtrip(): + """A generated token should verify back to the same email.""" + email = "token@example.com" + token = generate_token(email) + assert verify_token(token) == email + + +@pytest.mark.asyncio +async def test_token_expired(): + """An expired token should return None.""" + email = "expired@example.com" + with patch("app.token.time") as mock_time: + # Generate token "49 hours ago" + past = time.time() - 49 * 3600 + mock_time.time.return_value = past + token = generate_token(email) + + # Now verify with real time — should be expired (>48h) + assert verify_token(token) is None + + +@pytest.mark.asyncio +async def test_token_tampered(): + """A tampered token should return None.""" + token = generate_token("legit@example.com") + # Flip a character in the token + tampered = token[:-1] + ("A" if token[-1] != "A" else "B") + assert verify_token(tampered) is None + + +# ── Confirm endpoint tests ─────────────────────────────────────────── + + +@pytest.mark.asyncio +async def test_confirm_valid_token(client, db_session): + """GET /confirm with valid token marks email as confirmed.""" + # Seed an unconfirmed entry + entry = WaitlistEntry(email="confirm@example.com", source="website") + db_session.add(entry) + await db_session.commit() + + token = generate_token("confirm@example.com") + resp = await client.get(f"/api/v1/waitlist/confirm?token={token}") + assert resp.status_code == 200 + assert "confirmed" in resp.text.lower() or "verified" in resp.text.lower() + + # Verify DB state + result = await db_session.execute( + select(WaitlistEntry).where(WaitlistEntry.email == "confirm@example.com") + ) + assert result.scalar_one().confirmed is True + + +@pytest.mark.asyncio +async def test_confirm_invalid_token(client): + """GET /confirm with invalid token returns 400.""" + resp = await client.get("/api/v1/waitlist/confirm?token=garbage") + assert resp.status_code == 400 + assert "invalid" in resp.text.lower() or "expired" in resp.text.lower() + + +@pytest.mark.asyncio +async def test_confirm_idempotent(client, db_session): + """Confirming an already confirmed email returns 200 (idempotent).""" + entry = WaitlistEntry(email="idem@example.com", source="website", confirmed=True) + db_session.add(entry) + await db_session.commit() + + token = generate_token("idem@example.com") + resp = await client.get(f"/api/v1/waitlist/confirm?token={token}") + assert resp.status_code == 200 + + +@pytest.mark.asyncio +async def test_confirm_unknown_email(client): + """Token for a non-existent email returns 400.""" + token = generate_token("unknown@example.com") + resp = await client.get(f"/api/v1/waitlist/confirm?token={token}") + assert resp.status_code == 400 + + +# ── Brevo integration tests (mocked) ──────────────────────────────── + + +@pytest.mark.asyncio +async def test_signup_triggers_confirmation_email(client, db_session): + """When Brevo is configured, signup sends a confirmation email.""" + with patch("app.routes.settings") as mock_settings, \ + patch("app.routes.send_confirmation_email", new_callable=AsyncMock) as mock_send: + mock_settings.brevo_configured = True + mock_settings.CONFIRM_BASE_URL = "http://test" + + resp = await client.post( + "/api/v1/waitlist", + json={"email": "brevo@example.com"}, + ) + assert resp.status_code == 200 + + # Wait for fire-and-forget task + import asyncio + await asyncio.sleep(0.1) + + mock_send.assert_called_once() + call_args = mock_send.call_args + assert call_args[0][0] == "brevo@example.com" + assert "confirm" in call_args[0][1]