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)
This commit is contained in:
60
app/token.py
Normal file
60
app/token.py
Normal file
@@ -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
|
||||
Reference in New Issue
Block a user