feat(auth): add email verification and resend verification flow

.env.example

@@ -7,6 +7,7 @@ API_V1_STR="/api/v1"
 SECRET_KEY="your-super-secret-key-here"
 ACCESS_TOKEN_EXPIRE_MINUTES=15
 REFRESH_TOKEN_EXPIRE_DAYS=7
+VERIFICATION_TOKEN_EXPIRE_HOURS=24
 
 # Database (PostgreSQL)
 POSTGRES_SERVER=localhost

README.md

@@ -180,6 +180,18 @@ ruff .
 python -m ruff check .
 ```
 
+---
+
+## Logging Safety
+
+To prevent log injection and control-character pollution from user-provided inputs,
+FluentMeet sanitizes dynamic log arguments with `app/core/sanitize.py`.
+
+- Use `sanitize_for_log(value)` for a single value.
+- Use `sanitize_log_args(*values)` when logging multiple placeholders.
+- Newline, carriage return, tab, and other control characters are escaped or replaced.
+- Long values are truncated with `...<truncated>`.
+
 ## 🤝 Contributing
 
 We welcome contributions! Please follow these steps:

alembic/versions/19dc9714d9ea_add_email_verification_model.py

@@ -0,0 +1,57 @@
+"""Add email verification model
+
+Revision ID: 19dc9714d9ea
+Revises: 11781e907181
+Create Date: 2026-03-19 12:22:16.244801
+
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+# revision identifiers, used by Alembic.
+revision: str = "19dc9714d9ea"
+down_revision: Union[str, Sequence[str], None] = "11781e907181"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Upgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table(
+        "verification_tokens",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("user_id", sa.Integer(), nullable=False),
+        sa.Column("token", sa.String(length=36), nullable=False),
+        sa.Column("expires_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+        sa.ForeignKeyConstraint(
+            ["user_id"],
+            ["users.id"],
+        ),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    op.create_index(
+        op.f("ix_verification_tokens_id"), "verification_tokens", ["id"], unique=False
+    )
+    op.create_index(
+        op.f("ix_verification_tokens_token"),
+        "verification_tokens",
+        ["token"],
+        unique=True,
+    )
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    """Downgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_index(
+        op.f("ix_verification_tokens_token"), table_name="verification_tokens"
+    )
+    op.drop_index(op.f("ix_verification_tokens_id"), table_name="verification_tokens")
+    op.drop_table("verification_tokens")
+    # ### end Alembic commands ###

app/api/v1/endpoints/auth.py

@@ -1,25 +1,34 @@
 import logging
 from uuid import uuid4
 
-from fastapi import APIRouter, Depends, status
+from fastapi import APIRouter, Depends, Query, Request, status
 from sqlalchemy.orm import Session
 
 from app.core.config import settings
+from app.core.rate_limiter import limiter
+from app.core.sanitize import sanitize_log_args
 from app.crud.user.user import create_user, get_user_by_email
 from app.db.session import get_db
 from app.schemas.auth import (
     ActionAcknowledgement,
     ForgotPasswordRequest,
+    ResendVerificationRequest,
     SignupResponse,
+    VerifyEmailResponse,
 )
 from app.schemas.user import UserCreate
+from app.services.auth_verification import (
+    AuthVerificationService,
+    get_auth_verification_service,
+)
 from app.services.email_producer import EmailProducerService, get_email_producer_service
 
 logger = logging.getLogger(__name__)
 
 router = APIRouter(prefix="/auth", tags=["auth"])
 DB_SESSION_DEPENDENCY = Depends(get_db)
 EMAIL_PRODUCER_DEPENDENCY = Depends(get_email_producer_service)
+AUTH_VERIFICATION_SERVICE_DEPENDENCY = Depends(get_auth_verification_service)
 
 
 @router.post(
@@ -31,11 +40,18 @@ async def signup(
     user_in: UserCreate,
     db: Session = DB_SESSION_DEPENDENCY,
     email_producer: EmailProducerService = EMAIL_PRODUCER_DEPENDENCY,
+    auth_verification_service: AuthVerificationService = (
+        AUTH_VERIFICATION_SERVICE_DEPENDENCY
+    ),
 ) -> SignupResponse:
     user = create_user(db=db, user_in=user_in)
+    verification_token = auth_verification_service.create_verification_token(
+        db=db,
+        user_id=user.id,
+    )
 
     verification_link = (
-        f"{settings.FRONTEND_BASE_URL}/verify-email?user={user.id}&token={uuid4()}"
+        f"{settings.FRONTEND_BASE_URL}/verify-email?token={verification_token.token}"
     )
     try:
         await email_producer.send_email(
@@ -47,8 +63,11 @@ async def signup(
         )
     except Exception as exc:
         # Signup should succeed even if email queueing fails.
+        user_id_safe, exc_safe = sanitize_log_args(user.id, exc)
         logger.warning(
-            "Failed to enqueue verification email for user %s: %s", user.id, exc
+            "Failed to enqueue verification email for user %s: %s",
+            user_id_safe,
+            exc_safe,
         )
 
     return SignupResponse.model_validate(user)
@@ -81,8 +100,11 @@ async def forgot_password(
                 template="password_reset",
             )
         except Exception as exc:
+            email_safe, exc_safe = sanitize_log_args(user.email, exc)
             logger.warning(
-                "Failed to enqueue password reset email for %s: %s", user.email, exc
+                "Failed to enqueue password reset email for %s: %s",
+                email_safe,
+                exc_safe,
             )
 
     return ActionAcknowledgement(
@@ -91,3 +113,106 @@ async def forgot_password(
             "password reset instructions."
         )
     )
+
+
+@router.get(
+    "/verify-email",
+    response_model=VerifyEmailResponse,
+    status_code=status.HTTP_200_OK,
+    summary="Verify user email address",
+    description=(
+        "Validates an email verification token, activates the user account, "
+        "and invalidates the token."
+    ),
+    responses={
+        400: {
+            "description": "Missing, invalid, or expired token",
+            "content": {
+                "application/json": {
+                    "examples": {
+                        "missing": {
+                            "value": {
+                                "status": "error",
+                                "code": "MISSING_TOKEN",
+                                "message": "Verification token is required.",
+                                "details": [],
+                            }
+                        },
+                        "invalid": {
+                            "value": {
+                                "status": "error",
+                                "code": "INVALID_TOKEN",
+                                "message": "Verification token is invalid.",
+                                "details": [],
+                            }
+                        },
+                        "expired": {
+                            "value": {
+                                "status": "error",
+                                "code": "TOKEN_EXPIRED",
+                                "message": (
+                                    "Verification token has expired. "
+                                    "Please request a new one."
+                                ),
+                                "details": [],
+                            }
+                        },
+                    }
+                }
+            },
+        }
+    },
+)
+def verify_email(
+    token: str | None = Query(default=None),
+    db: Session = DB_SESSION_DEPENDENCY,
+    auth_verification_service: AuthVerificationService = (
+        AUTH_VERIFICATION_SERVICE_DEPENDENCY
+    ),
+) -> VerifyEmailResponse:
+    auth_verification_service.verify_email(db=db, token=token)
+    return VerifyEmailResponse(
+        message="Email successfully verified. You can now log in.",
+    )
+
+
+@router.post(
+    "/resend-verification",
+    response_model=ActionAcknowledgement,
+    status_code=status.HTTP_200_OK,
+    summary="Resend email verification link",
+    description=(
+        "Queues a new verification email when the account exists and is not "
+        "verified. Always returns a generic response to prevent user enumeration."
+    ),
+)
+@limiter.limit("3/minute")
+async def resend_verification(
+    request: Request,
+    payload: ResendVerificationRequest,
+    db: Session = DB_SESSION_DEPENDENCY,
+    email_producer: EmailProducerService = EMAIL_PRODUCER_DEPENDENCY,
+    auth_verification_service: AuthVerificationService = (
+        AUTH_VERIFICATION_SERVICE_DEPENDENCY
+    ),
+) -> ActionAcknowledgement:
+    del request
+    try:
+        await auth_verification_service.resend_verification_email(
+            db=db,
+            email=str(payload.email),
+            email_producer=email_producer,
+        )
+    except Exception as exc:
+        email_safe, exc_safe = sanitize_log_args(payload.email, exc)
+        logger.warning(
+            "Failed to enqueue verification resend for %s: %s",
+            email_safe,
+            exc_safe,
+        )
+
+    return ActionAcknowledgement(
+        message=(
+            "If an account with that email exists, we have sent a verification email."
+        )
+    )

app/core/config.py

@@ -22,6 +22,7 @@ class Settings(BaseSettings):
     SECRET_KEY: str = "placeholder_secret_key"
     ACCESS_TOKEN_EXPIRE_MINUTES: int = 15
     REFRESH_TOKEN_EXPIRE_DAYS: int = 7
+    VERIFICATION_TOKEN_EXPIRE_HOURS: int = 24
 
     # Database
     POSTGRES_SERVER: str = "localhost"

app/core/exception_handlers.py

@@ -8,6 +8,7 @@
 
 from app.core.error_responses import create_error_response
 from app.core.exceptions import FluentMeetException
+from app.core.sanitize import sanitize_for_log
 
 logger = logging.getLogger(__name__)
 
@@ -62,7 +63,7 @@ async def unhandled_exception_handler(
     """
     Handler for all other unhandled exceptions (500).
     """
-    logger.exception("Unhandled exception occurred: %s", str(exc))
+    logger.exception("Unhandled exception occurred: %s", sanitize_for_log(exc))
     return create_error_response(
         status_code=500,
         code="INTERNAL_SERVER_ERROR",

app/core/rate_limiter.py

@@ -0,0 +1,20 @@
+from fastapi import Request
+from fastapi.responses import JSONResponse
+from slowapi import Limiter
+from slowapi.errors import RateLimitExceeded
+from slowapi.util import get_remote_address
+
+from app.core.error_responses import create_error_response
+
+limiter = Limiter(key_func=get_remote_address)
+
+
+async def rate_limit_exception_handler(
+    _request: Request,
+    _exc: RateLimitExceeded,
+) -> JSONResponse:
+    return create_error_response(
+        status_code=429,
+        code="RATE_LIMIT_EXCEEDED",
+        message="Too many requests. Please try again later.",
+    )

app/core/sanitize.py

@@ -0,0 +1,34 @@
+import re
+from collections.abc import Iterable
+
+
+class LogSanitizer:
+    """Sanitizes user-controlled values before they are written to logs."""
+
+    _CONTROL_CHARS = re.compile(r"[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]")
+
+    def __init__(self, max_length: int = 256) -> None:
+        self._max_length = max_length
+
+    def sanitize(self, value: object) -> str:
+        text = str(value)
+        text = text.replace("\r", r"\r").replace("\n", r"\n").replace("\t", r"\t")
+        text = self._CONTROL_CHARS.sub("?", text)
+        if len(text) <= self._max_length:
+            return text
+
+        return f"{text[: self._max_length]}...<truncated>"
+
+    def sanitize_many(self, values: Iterable[object]) -> tuple[str, ...]:
+        return tuple(self.sanitize(value) for value in values)
+
+
+log_sanitizer = LogSanitizer()
+
+
+def sanitize_for_log(value: object) -> str:
+    return log_sanitizer.sanitize(value)
+
+
+def sanitize_log_args(*values: object) -> tuple[str, ...]:
+    return log_sanitizer.sanitize_many(values)