add TOTP two-factor authentication with recovery codes

.env.example

@@ -12,3 +12,7 @@ SECRET_KEY=change-this-to-a-secure-random-string-in-production
 # Admin user (created on first run)
 ADMIN_EMAIL=admin@example.com
 ADMIN_PASSWORD=changeme
+
+# TOTP 2FA encryption key (generate with: python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())")
+# Leave empty to disable TOTP support
+TOTP_ENCRYPTION_KEY=

backend/.env.example

@@ -13,3 +13,7 @@ ADMIN_NAME=Admin
 
 # CORS (JSON array)
 CORS_ORIGINS=["http://localhost:5173", "http://localhost:3000"]
+
+# TOTP 2FA encryption key (generate with: python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())")
+# Leave empty to disable TOTP support
+TOTP_ENCRYPTION_KEY=

backend/app/api/admin.py

@@ -9,6 +9,7 @@
 )
 from app.api.deps import AdminUser
 from app.services.partitions import create_team_partition, drop_team_partition
+from app.services.totp import TOTPService
 
 router = APIRouter()
 
@@ -93,6 +94,20 @@ async def delete_user(admin: AdminUser, user_id: UUID):
     return {"message": "User deleted"}
 
 
+@router.post("/users/{user_id}/reset-totp")
+async def reset_user_totp(admin: AdminUser, user_id: UUID):
+    """Reset TOTP for a user (admin only)."""
+    user = await User.filter(id=user_id).first()
+    if user is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
+
+    if not user.totp_enabled:
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="TOTP is not enabled for this user")
+
+    await TOTPService.disable(user)
+    return {"message": "TOTP reset successfully"}
+
+
 # ============== Teams ==============
 
 async def _team_response(team: Team) -> TeamResponse:

backend/app/api/auth.py

@@ -1,15 +1,37 @@
 from fastapi import APIRouter, HTTPException, status
 from app.models import User
-from app.schemas import Token, LoginRequest, RefreshRequest, UserResponse
+from app.models.totp_attempt import TotpAttempt
+from app.schemas import (
+    Token, LoginRequest, RefreshRequest, UserResponse,
+    LoginResponse, TOTPVerifyLoginRequest, TOTPSetupResponse,
+    TOTPVerifySetupRequest, TOTPDisableRequest,
+)
 from app.services.auth import AuthService
+from app.services.totp import TOTPService
 from app.api.deps import CurrentUser
 
 router = APIRouter()
 
+_TOTP_MAX_ATTEMPTS = 5
 
-@router.post("/login", response_model=Token)
+
+async def _check_totp_attempts(jti: str) -> None:
+    """Raise 401 if this token has exceeded the attempt limit."""
+    count = await TotpAttempt.filter(jti=jti).count()
+    if count >= _TOTP_MAX_ATTEMPTS:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Too many TOTP attempts, please log in again",
+        )
+
+
+async def _record_totp_failure(jti: str) -> None:
+    await TotpAttempt.create(jti=jti)
+
+
+@router.post("/login", response_model=LoginResponse)
 async def login(request: LoginRequest):
-    """Authenticate user and return tokens."""
+    """Authenticate user and return tokens, or TOTP challenge if 2FA is enabled."""
     user = await User.filter(email=request.email, is_active=True).first()
 
     if user is None or not user.verify_password(request.password):
@@ -18,14 +40,105 @@ async def login(request: LoginRequest):
             detail="Invalid email or password",
         )
 
-    access_token, refresh_token = AuthService.create_tokens(str(user.id))
+    if user.totp_enabled:
+        totp_token = AuthService.create_totp_token(str(user.id))
+        return LoginResponse(
+            totp_required=True,
+            totp_token=totp_token,
+        )
 
-    return Token(
+    access_token, refresh_token = AuthService.create_tokens(str(user.id))
+    return LoginResponse(
+        totp_required=False,
         access_token=access_token,
         refresh_token=refresh_token,
     )
 
 
+@router.post("/verify-totp", response_model=Token)
+async def verify_totp(request: TOTPVerifyLoginRequest):
+    """Verify TOTP code and issue real JWT tokens."""
+    payload = AuthService.decode_token(request.totp_token)
+
+    if payload is None or payload.type != "totp_required" or not payload.jti:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Invalid or expired TOTP token",
+        )
+
+    await _check_totp_attempts(payload.jti)
+
+    user = await User.filter(id=payload.sub, is_active=True).first()
+    if user is None:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="User not found or inactive",
+        )
+
+    if not await TOTPService.verify_login_code(user, request.code):
+        await _record_totp_failure(payload.jti)
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Invalid TOTP code",
+        )
+
+    # Success — clean up attempt records for this token
+    await TotpAttempt.filter(jti=payload.jti).delete()
+
+    access_token, refresh_token = AuthService.create_tokens(str(user.id))
+    return Token(access_token=access_token, refresh_token=refresh_token)
+
+
+@router.post("/totp/setup", response_model=TOTPSetupResponse)
+async def totp_setup(user: CurrentUser):
+    """Begin TOTP setup: generate secret, QR code, and recovery codes."""
+    if user.totp_enabled:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="TOTP is already enabled",
+        )
+
+    result = await TOTPService.setup_begin(user)
+    return TOTPSetupResponse(**result)
+
+
+@router.post("/totp/setup/verify")
+async def totp_setup_verify(user: CurrentUser, request: TOTPVerifySetupRequest):
+    """Verify the first TOTP code to complete setup."""
+    if user.totp_enabled:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="TOTP is already enabled",
+        )
+
+    if not await TOTPService.setup_verify(user, request.code):
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="Invalid TOTP code",
+        )
+
+    return {"message": "TOTP enabled successfully"}
+
+
+@router.post("/totp/disable")
+async def totp_disable(user: CurrentUser, request: TOTPDisableRequest):
+    """Disable TOTP (requires password confirmation)."""
+    if not user.totp_enabled:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="TOTP is not enabled",
+        )
+
+    if not user.verify_password(request.password):
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Invalid password",
+        )
+
+    await TOTPService.disable(user)
+    return {"message": "TOTP disabled successfully"}
+
+
 @router.post("/refresh", response_model=Token)
 async def refresh(request: RefreshRequest):
     """Get new access token using refresh token."""
@@ -61,6 +174,4 @@ async def get_current_user_info(user: CurrentUser):
 @router.post("/logout")
 async def logout(user: CurrentUser):
     """Logout user (client should discard tokens)."""
-    # JWT tokens are stateless, so we just return success
-    # Client is responsible for discarding the tokens
     return {"message": "Logged out successfully"}

backend/app/config.py

@@ -17,6 +17,9 @@ class Settings(BaseSettings):
     admin_password: str = "changeme"
     admin_name: str = "Admin"
 
+    # TOTP
+    totp_encryption_key: str = ""
+
     # CORS
     cors_origins: list[str] = ["http://localhost:5173", "http://localhost:3000"]
 

backend/app/models/__init__.py

@@ -2,5 +2,7 @@
 from app.models.team import Team, TeamMembership, TeamRole
 from app.models.api_key import ApiKey
 from app.models.log import Log, LogLevel
+from app.models.recovery_code import RecoveryCode
+from app.models.totp_attempt import TotpAttempt
 
-__all__ = ["User", "Team", "TeamMembership", "TeamRole", "ApiKey", "Log", "LogLevel"]
+__all__ = ["User", "Team", "TeamMembership", "TeamRole", "ApiKey", "Log", "LogLevel", "RecoveryCode", "TotpAttempt"]

backend/app/models/recovery_code.py

@@ -0,0 +1,18 @@
+import bcrypt
+from tortoise import fields
+from tortoise.models import Model
+
+
+class RecoveryCode(Model):
+    id = fields.UUIDField(pk=True)
+    user = fields.ForeignKeyField("models.User", related_name="recovery_codes", on_delete=fields.CASCADE)
+    code_hash = fields.CharField(max_length=255)
+    used = fields.BooleanField(default=False)
+    created_at = fields.DatetimeField(auto_now_add=True)
+
+    class Meta:
+        table = "recovery_codes"
+
+    def verify_code(self, code: str) -> bool:
+        """Check a plaintext recovery code against the stored hash."""
+        return bcrypt.checkpw(code.encode(), self.code_hash.encode())

backend/app/models/totp_attempt.py

@@ -0,0 +1,11 @@
+from tortoise import fields
+from tortoise.models import Model
+
+
+class TotpAttempt(Model):
+    id = fields.UUIDField(pk=True)
+    jti = fields.CharField(max_length=36, index=True)
+    created_at = fields.DatetimeField(auto_now_add=True)
+
+    class Meta:
+        table = "totp_attempts"

backend/app/models/user.py

@@ -10,11 +10,14 @@ class User(Model):
     name = fields.CharField(max_length=255)
     is_admin = fields.BooleanField(default=False)
     is_active = fields.BooleanField(default=True)
+    totp_secret_encrypted = fields.TextField(null=True, default=None)
+    totp_enabled = fields.BooleanField(default=False)
     created_at = fields.DatetimeField(auto_now_add=True)
     updated_at = fields.DatetimeField(auto_now=True)
 
     # Reverse relations
     team_memberships: fields.ReverseRelation["TeamMembership"]
+    recovery_codes: fields.ReverseRelation["RecoveryCode"]
 
     class Meta:
         table = "users"

backend/app/schemas/__init__.py

@@ -1,4 +1,8 @@
-from app.schemas.auth import Token, TokenPayload, LoginRequest, RefreshRequest
+from app.schemas.auth import (
+    Token, TokenPayload, LoginRequest, RefreshRequest,
+    LoginResponse, TOTPVerifyLoginRequest, TOTPSetupResponse,
+    TOTPVerifySetupRequest, TOTPDisableRequest,
+)
 from app.schemas.user import UserCreate, UserUpdate, UserResponse
 from app.schemas.team import (
     TeamCreate, TeamUpdate, TeamResponse, TeamCreateResponse,
@@ -10,6 +14,8 @@
 
 __all__ = [
     "Token", "TokenPayload", "LoginRequest", "RefreshRequest",
+    "LoginResponse", "TOTPVerifyLoginRequest", "TOTPSetupResponse",
+    "TOTPVerifySetupRequest", "TOTPDisableRequest",
     "UserCreate", "UserUpdate", "UserResponse",
     "TeamCreate", "TeamUpdate", "TeamResponse", "TeamCreateResponse",
     "ApiKeyCreate", "ApiKeyResponse", "ApiKeyWithSecret",

backend/app/schemas/auth.py

@@ -18,4 +18,34 @@ class Token(BaseModel):
 
 class TokenPayload(BaseModel):
     sub: str  # user id
-    type: str  # "access" or "refresh"
+    type: str  # "access", "refresh", or "totp_required"
+    jti: str | None = None  # unique token id (used for totp_required tokens)
+
+
+class LoginResponse(BaseModel):
+    totp_required: bool = False
+    # Present when totp_required is False (normal login)
+    access_token: str | None = None
+    refresh_token: str | None = None
+    token_type: str = "bearer"
+    # Present when totp_required is True
+    totp_token: str | None = None
+
+
+class TOTPVerifyLoginRequest(BaseModel):
+    totp_token: str
+    code: str
+
+
+class TOTPSetupResponse(BaseModel):
+    secret: str
+    qr_code: str
+    recovery_codes: list[str]
+
+
+class TOTPVerifySetupRequest(BaseModel):
+    code: str
+
+
+class TOTPDisableRequest(BaseModel):
+    password: str

backend/app/schemas/user.py

@@ -24,6 +24,7 @@ class UserResponse(BaseModel):
     name: str
     is_admin: bool
     is_active: bool
+    totp_enabled: bool = False
     created_at: datetime
     updated_at: datetime
 

backend/app/services/auth.py

@@ -1,3 +1,4 @@
+import uuid
 from datetime import datetime, timedelta, timezone
 from jose import jwt, JWTError
 from app.config import get_settings
@@ -27,11 +28,22 @@ def create_refresh_token(user_id: str) -> str:
         }
         return jwt.encode(payload, settings.secret_key, algorithm=settings.algorithm)
 
+    @staticmethod
+    def create_totp_token(user_id: str) -> str:
+        expire = datetime.now(timezone.utc) + timedelta(minutes=5)
+        payload = {
+            "sub": user_id,
+            "type": "totp_required",
+            "jti": str(uuid.uuid4()),
+            "exp": expire,
+        }
+        return jwt.encode(payload, settings.secret_key, algorithm=settings.algorithm)
+
     @staticmethod
     def decode_token(token: str) -> TokenPayload | None:
         try:
             payload = jwt.decode(token, settings.secret_key, algorithms=[settings.algorithm])
-            return TokenPayload(sub=payload["sub"], type=payload["type"])
+            return TokenPayload(sub=payload["sub"], type=payload["type"], jti=payload.get("jti"))
         except JWTError:
             return None