adiuvAI/waitlist
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: GDPR compliance — anonymization, unsubscribe, consent tracking

Browse Source 
- Add consent_given_at and anonymized_at fields + Alembic migration (002)
- Add GET /waitlist/unsubscribe endpoint (HMAC token, anonymizes PII)
- Add cleanup.py: cron-able script to anonymize unconfirmed entries after 48h
- Clear IP address on email confirmation (no longer needed)
- Add unsubscribe link in confirmation email footer
- Record consent timestamp on signup
- Add 4 new tests (unsubscribe, consent timestamp)
- Update .env.example, schemas
This commit is contained in:
Roberto Musso
2026-04-11 19:41:27 +02:00
parent 5f79ce87f9
commit 352e25d651
6 changed files with 232 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

81
app/routes.py
View File

@@ -1,4 +1,5 @@
import asyncio
import datetime
import logging
from fastapi import APIRouter, Depends, Request
@@ -31,6 +32,7 @@ async def join_waitlist(
- 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).
- Records consent timestamp (GDPR Art. 7).
"""
# Honeypot — bots fill hidden fields; silently "succeed"
if body.website:
@@ -38,6 +40,7 @@ async def join_waitlist(
email = body.email.lower().strip()
ip = _get_client_ip(request)
now = datetime.datetime.now(datetime.timezone.utc)
# Check for existing entry — idempotent
existing = await db.execute(
@@ -46,14 +49,20 @@ async def join_waitlist(
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:
# Re-send confirmation if not yet confirmed (and not anonymized)
if not entry.confirmed and not entry.anonymized_at 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))
unsub_url = f"{settings.CONFIRM_BASE_URL}/api/v1/waitlist/unsubscribe?token={token}"
asyncio.create_task(send_confirmation_email(email, confirm_url, unsub_url))
return WaitlistResponse()
entry = WaitlistEntry(email=email, ip_address=ip, source="website")
entry = WaitlistEntry(
email=email,
ip_address=ip,
source="website",
consent_given_at=now,
)
db.add(entry)
await db.commit()
@@ -63,7 +72,8 @@ async def join_waitlist(
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))
unsub_url = f"{settings.CONFIRM_BASE_URL}/api/v1/waitlist/unsubscribe?token={token}"
asyncio.create_task(send_confirmation_email(email, confirm_url, unsub_url))
return WaitlistResponse()
@@ -92,6 +102,8 @@ async def confirm_email(
if not entry.confirmed:
entry.confirmed = True
# Clear IP now that consent is confirmed — no longer needed
entry.ip_address = None
await db.commit()
logger.info("Email confirmed: %s***", email[:3])
@@ -102,9 +114,62 @@ async def confirm_email(
return HTMLResponse(content=_result_page(success=True))
def _result_page(*, success: bool) -> str:
"""Branded HTML response for confirmation result, matching the adiuvAI landing page."""
if success:
@router.get("/waitlist/unsubscribe", response_class=HTMLResponse)
async def unsubscribe(
token: str,
db: AsyncSession = Depends(get_db),
) -> HTMLResponse:
"""
GDPR erasure (Art. 17) — anonymize the entry.
Uses the same HMAC token system as confirmation.
"""
email = verify_token(token)
if email is None:
return HTMLResponse(
content=_result_page(success=False, variant="unsubscribe"),
status_code=400,
)
result = await db.execute(
select(WaitlistEntry).where(WaitlistEntry.email == email)
)
entry = result.scalar_one_or_none()
if entry is None:
# Already gone — show success anyway (idempotent)
return HTMLResponse(content=_result_page(success=True, variant="unsubscribe"))
_anonymize_entry(entry)
await db.commit()
logger.info("Waitlist entry anonymized (unsubscribe)")
return HTMLResponse(content=_result_page(success=True, variant="unsubscribe"))
def _anonymize_entry(entry: WaitlistEntry) -> None:
"""Strip all PII from a waitlist entry, keeping only anonymous analytics."""
now = datetime.datetime.now(datetime.timezone.utc)
entry.email = f"anon-{entry.id}@removed.invalid"
entry.ip_address = None
entry.consent_given_at = None
entry.anonymized_at = now
def _result_page(*, success: bool, variant: str = "confirm") -> str:
"""Branded HTML response for confirmation/unsubscribe result."""
if variant == "unsubscribe":
if success:
title = "Data removed"
message = "Your personal data has been anonymized. You will not receive any further emails from us."
icon_bg = "rgba(251,200,129,0.12)"
icon_border = "rgba(251,200,129,0.25)"
icon_svg = '<svg viewBox="0 0 24 24" width="24" height="24" fill="none" stroke="#e5a94e" stroke-width="2.5" stroke-linecap="round" stroke-linejoin="round"><polyline points="20 6 9 17 4 12"/></svg>'
else:
title = "Invalid or expired link"
message = "This unsubscribe link is no longer valid. Contact privacy@adiuvai.com if you need help."
icon_bg = "rgba(220,38,38,0.08)"
icon_border = "rgba(220,38,38,0.18)"
icon_svg = '<svg viewBox="0 0 24 24" width="24" height="24" fill="none" stroke="#dc2626" stroke-width="2.5" stroke-linecap="round" stroke-linejoin="round"><line x1="18" y1="6" x2="6" y2="18"/><line x1="6" y1="6" x2="18" y2="18"/></svg>'
elif success:
title = "You&#8217;re confirmed!"
message = "Your email has been verified. We&#8217;ll notify you when adiuvAI is ready."
icon_bg = "rgba(251,200,129,0.12)"