From ee3031013edf76e6685811fd385e03325f5f4536 Mon Sep 17 00:00:00 2001 From: Tianyu Liu Date: Sun, 21 Jun 2026 21:55:17 +0200 Subject: [PATCH] M4-T05: add TOTP service and setup/enable/disable/status API --- app/api/routes/api/session.py | 113 ++++++++ app/config.py | 7 + app/schemas/totp.py | 59 +++++ app/services/totp.py | 221 ++++++++++++++++ docs/design/m4-login-hardening.md | 2 +- openapi/openapi.json | 262 +++++++++++++++++++ openapi/openapi.yaml | 216 +++++++++++++++ requirements.in | 1 + requirements.txt | 2 + tests/test_api_totp.py | 419 ++++++++++++++++++++++++++++++ 10 files changed, 1301 insertions(+), 1 deletion(-) create mode 100644 app/schemas/totp.py create mode 100644 app/services/totp.py create mode 100644 tests/test_api_totp.py diff --git a/app/api/routes/api/session.py b/app/api/routes/api/session.py index 58c1ecf..25c72ec 100644 --- a/app/api/routes/api/session.py +++ b/app/api/routes/api/session.py @@ -14,6 +14,12 @@ from app.schemas.session import ( SessionResponse, SessionUser, ) +from app.schemas.totp import ( + TotpDisableRequest, + TotpEnableRequest, + TotpSetupResponse, + TotpStatusResponse, +) from app.services.auth import ( AuthPasswordChangeError, AuthenticatedSession, @@ -23,6 +29,7 @@ from app.services.auth import ( revoke_session, ) from app.services import login_throttle +from app.services import totp as totp_service logger = logging.getLogger(__name__) @@ -179,3 +186,109 @@ def post_change_password( logger.info("Password updated for user '%s'", auth.user.username) return Response(status_code=status.HTTP_204_NO_CONTENT) + + +# --------------------------------------------------------------------------- +# TOTP endpoints (M4-T05) +# --------------------------------------------------------------------------- + + +@router.post("/auth/totp/setup", response_model=TotpSetupResponse) +def post_totp_setup( + db: Session = Depends(get_db), + settings: Settings = Depends(get_app_settings), + auth: AuthenticatedSession = Depends(require_session), + _csrf: None = Depends(require_csrf), +) -> TotpSetupResponse: + """ + Generate a new pending TOTP secret, otpauth URI, and one-time recovery codes. + + The secret is stored in the DB but TOTP is NOT yet enabled (totp_enabled stays + False until the user confirms with POST /api/auth/totp/enable). + + Recovery codes are returned here as plaintext exactly once; their Argon2 hashes + are persisted immediately so enable only needs to flip the enabled flag. + + Repeating this call replaces any prior pending secret and regenerates codes. + + Requires: session cookie + X-CSRF-Token. + """ + secret, otpauth_uri, recovery_codes = totp_service.setup( + db, + user=auth.user, + issuer=settings.effective_totp_issuer, + ) + return TotpSetupResponse( + secret=secret, + otpauth_uri=otpauth_uri, + recovery_codes=recovery_codes, + ) + + +@router.post("/auth/totp/enable", status_code=status.HTTP_204_NO_CONTENT) +def post_totp_enable( + body: TotpEnableRequest, + db: Session = Depends(get_db), + auth: AuthenticatedSession = Depends(require_session), + _csrf: None = Depends(require_csrf), +) -> Response: + """ + Enable TOTP by confirming with the current 6-digit code from the authenticator app. + + Requires a prior call to POST /api/auth/totp/setup (so that a pending secret + exists). On success, totp_enabled becomes True. + + Returns 400 if the code is wrong or there is no pending secret. + Requires: session cookie + X-CSRF-Token. + """ + ok = totp_service.enable(db, user=auth.user, code=body.code) + if not ok: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="invalid TOTP code or no pending setup", + ) + return Response(status_code=status.HTTP_204_NO_CONTENT) + + +@router.post("/auth/totp/disable", status_code=status.HTTP_204_NO_CONTENT) +def post_totp_disable( + body: TotpDisableRequest, + db: Session = Depends(get_db), + auth: AuthenticatedSession = Depends(require_session), + _csrf: None = Depends(require_csrf), +) -> Response: + """ + Disable TOTP. The caller must provide exactly one of: + - ``password``: the user's current login password, OR + - ``code``: the current 6-digit TOTP code. + + On success: totp_enabled=False, totp_secret cleared, all recovery codes deleted. + Returns 400 if neither credential matches or neither is provided. + Requires: session cookie + X-CSRF-Token. + """ + ok = totp_service.disable( + db, + user=auth.user, + password=body.password, + code=body.code, + ) + if not ok: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="invalid credential; provide a valid password or TOTP code", + ) + return Response(status_code=status.HTTP_204_NO_CONTENT) + + +@router.get("/auth/totp", response_model=TotpStatusResponse) +def get_totp_status( + auth: AuthenticatedSession = Depends(require_session), +) -> TotpStatusResponse: + """ + Return the current TOTP status for the authenticated user. + + Response contains only ``{"enabled": bool}``. + Secret and recovery codes are NEVER returned here. + Requires: session cookie only (no CSRF — read-only). + """ + return TotpStatusResponse(enabled=auth.user.totp_enabled) diff --git a/app/config.py b/app/config.py index 61fbbe9..24c5b74 100644 --- a/app/config.py +++ b/app/config.py @@ -39,6 +39,7 @@ class Settings(BaseSettings): auth_cookie_secure_override: bool | None = True auth_login_throttle_enabled: bool = True auth_trust_forwarded_for: bool = False + auth_totp_issuer: str = "" # defaults to app_name when empty model_config = SettingsConfigDict( env_file=".env", @@ -88,6 +89,12 @@ class Settings(BaseSettings): return self.auth_cookie_secure_override return not self.is_development + @computed_field + @property + def effective_totp_issuer(self) -> str: + """The issuer label shown in Authenticator apps. Falls back to app_name.""" + return self.auth_totp_issuer.strip() or self.app_name + @lru_cache def get_settings() -> Settings: diff --git a/app/schemas/totp.py b/app/schemas/totp.py new file mode 100644 index 0000000..69a7073 --- /dev/null +++ b/app/schemas/totp.py @@ -0,0 +1,59 @@ +"""Pydantic schemas for TOTP setup / enable / disable / status endpoints (M4-T05).""" +from __future__ import annotations + +from pydantic import BaseModel + + +# --------------------------------------------------------------------------- +# Setup (POST /api/auth/totp/setup) — response +# --------------------------------------------------------------------------- + + +class TotpSetupResponse(BaseModel): + """Returned once after a setup call. + + ``secret`` and ``recovery_codes`` are **one-time plaintext values**. + They are never returned again by any subsequent API call. + The frontend must display and instruct the user to save them before confirming. + """ + + secret: str + otpauth_uri: str + recovery_codes: list[str] + + +# --------------------------------------------------------------------------- +# Enable (POST /api/auth/totp/enable) — request +# --------------------------------------------------------------------------- + + +class TotpEnableRequest(BaseModel): + """The user confirms setup by providing the 6-digit TOTP code.""" + + code: str + + +# --------------------------------------------------------------------------- +# Disable (POST /api/auth/totp/disable) — request +# --------------------------------------------------------------------------- + + +class TotpDisableRequest(BaseModel): + """Disable TOTP by proving identity. + + Exactly one of ``password`` or ``code`` must be provided. + """ + + password: str | None = None + code: str | None = None + + +# --------------------------------------------------------------------------- +# Status (GET /api/auth/totp) — response +# --------------------------------------------------------------------------- + + +class TotpStatusResponse(BaseModel): + """Minimal status response — never exposes secret or recovery codes.""" + + enabled: bool diff --git a/app/services/totp.py b/app/services/totp.py new file mode 100644 index 0000000..464421b --- /dev/null +++ b/app/services/totp.py @@ -0,0 +1,221 @@ +"""TOTP service: secret management, recovery-code lifecycle, setup/enable/disable (M4-T05). + +State-machine summary +--------------------- +DISABLED (default) + totp_secret=None, totp_enabled=False, no recovery codes + +PENDING (after setup, before enable) + totp_secret=, totp_enabled=False, recovery codes stored as hashes + - Re-calling setup replaces the pending secret and regenerates recovery codes. + +ENABLED (after enable) + totp_secret=, totp_enabled=True, recovery codes stored as hashes + +After disable: + totp_secret=None, totp_enabled=False, recovery codes deleted → back to DISABLED + + +Recovery-code timing +-------------------- +1. setup → generate 10 plaintext codes, persist their Argon2 hashes immediately, + return plaintext to caller (ONLY time). +2. enable → verify 6-digit TOTP code; if ok, flip totp_enabled=True. + Recovery codes are already in the DB from step 1. +3. disable → clear secret, clear enabled flag, delete all recovery codes. + +Security note: ``totp_secret`` is stored as plaintext (same posture as other +secrets in this project: protected by file-system permissions on the SQLite +database). Recovery codes are stored as Argon2 hashes only. +""" +from __future__ import annotations + +import logging +import secrets +import string + +import pyotp +from sqlalchemy import delete, select +from sqlalchemy.orm import Session + +from app.models.auth import AuthUser, RecoveryCode +from app.services.auth import hash_password, verify_password + +logger = logging.getLogger(__name__) + +# Number of recovery codes to generate per setup +_RECOVERY_CODE_COUNT = 10 +# Alphabet for each 4-character segment: lowercase letters + digits, no ambiguous chars +_CODE_ALPHABET = string.ascii_lowercase + string.digits + + +# --------------------------------------------------------------------------- +# Internal helpers +# --------------------------------------------------------------------------- + + +def _generate_recovery_code() -> str: + """Return a single recovery code in ``xxxx-xxxx`` format.""" + segment = lambda: "".join(secrets.choice(_CODE_ALPHABET) for _ in range(4)) # noqa: E731 + return f"{segment()}-{segment()}" + + +def _verify_totp_code(secret: str, code: str) -> bool: + """Verify a 6-digit TOTP code against the given base-32 secret (±1 window).""" + return pyotp.TOTP(secret).verify(code, valid_window=1) + + +def _delete_recovery_codes(db: Session, *, user_id: int) -> None: + """Delete ALL recovery codes for a user (called on setup re-run and disable).""" + db.execute(delete(RecoveryCode).where(RecoveryCode.user_id == user_id)) + + +# --------------------------------------------------------------------------- +# Public API +# --------------------------------------------------------------------------- + + +def setup( + db: Session, + *, + user: AuthUser, + issuer: str, +) -> tuple[str, str, list[str]]: + """Generate a new pending TOTP secret and recovery codes. + + The secret is stored in ``user.totp_secret`` (``totp_enabled`` stays + ``False``). Any pre-existing pending secret and recovery codes are + replaced atomically. + + Returns + ------- + (secret, otpauth_uri, plaintext_recovery_codes) + + ``plaintext_recovery_codes`` are returned **once** here and MUST NOT be + stored or returned elsewhere. + """ + # Generate new TOTP secret + new_secret = pyotp.random_base32() + otpauth_uri = pyotp.TOTP(new_secret).provisioning_uri( + name=user.username, + issuer_name=issuer, + ) + + # Generate plaintext recovery codes + plaintext_codes = [_generate_recovery_code() for _ in range(_RECOVERY_CODE_COUNT)] + + # --- Persist atomically --- + # 1. Delete any old pending recovery codes (idempotent on re-setup) + assert user.id is not None # mypy guard; always set after DB insertion + _delete_recovery_codes(db, user_id=user.id) + + # 2. Update the secret (pending — totp_enabled stays False) + user.totp_secret = new_secret + + # 3. Persist new recovery codes as Argon2 hashes + for plaintext in plaintext_codes: + db.add(RecoveryCode(user_id=user.id, code_hash=hash_password(plaintext))) + + db.commit() + db.refresh(user) + + logger.info("TOTP setup (pending) for user '%s'; %d recovery codes generated.", user.username, _RECOVERY_CODE_COUNT) + return new_secret, otpauth_uri, plaintext_codes + + +def enable(db: Session, *, user: AuthUser, code: str) -> bool: + """Enable TOTP after the user proves they can generate the correct code. + + Parameters + ---------- + code: + The current 6-digit TOTP code from the user's authenticator app. + + Returns + ------- + True on success; False if the code is invalid or no secret is pending. + """ + if not user.totp_secret: + logger.info("TOTP enable rejected for '%s': no pending secret.", user.username) + return False + + if not _verify_totp_code(user.totp_secret, code): + logger.info("TOTP enable rejected for '%s': wrong code.", user.username) + return False + + user.totp_enabled = True + db.commit() + db.refresh(user) + logger.info("TOTP enabled for user '%s'.", user.username) + return True + + +def disable( + db: Session, + *, + user: AuthUser, + password: str | None = None, + code: str | None = None, +) -> bool: + """Disable TOTP. Caller must supply exactly one of ``password`` or ``code``. + + On success: ``totp_enabled=False``, ``totp_secret=None``, all recovery + codes deleted. + + Returns + ------- + True on success; False if neither credential is valid. + """ + if not password and not code: + logger.info("TOTP disable rejected for '%s': no credential provided.", user.username) + return False + + if password: + if not verify_password(password, user.password_hash): + logger.info("TOTP disable rejected for '%s': wrong password.", user.username) + return False + elif code: + if not user.totp_secret or not _verify_totp_code(user.totp_secret, code): + logger.info("TOTP disable rejected for '%s': wrong TOTP code.", user.username) + return False + + # Clear TOTP state + assert user.id is not None + _delete_recovery_codes(db, user_id=user.id) + user.totp_enabled = False + user.totp_secret = None + db.commit() + db.refresh(user) + logger.info("TOTP disabled for user '%s'.", user.username) + return True + + +def verify_recovery_code(db: Session, *, user: AuthUser, code: str) -> bool: + """Verify and consume a one-time recovery code. + + Finds the first unused recovery code whose hash matches ``code``, marks it + as consumed (sets ``used_at``), and returns ``True``. Returns ``False`` if + no matching unused code exists. + + This function is provided for T06 (login two-factor) but lives here so the + TOTP service owns all recovery-code logic. + """ + from datetime import UTC, datetime + + unused = db.execute( + select(RecoveryCode).where( + RecoveryCode.user_id == user.id, + RecoveryCode.used_at.is_(None), + ) + ).scalars().all() + + for rc in unused: + if verify_password(code, rc.code_hash): + rc.used_at = datetime.now(UTC) + db.commit() + logger.info( + "Recovery code consumed for user '%s' (id=%d).", user.username, rc.id + ) + return True + + return False diff --git a/docs/design/m4-login-hardening.md b/docs/design/m4-login-hardening.md index 721cd01..e831804 100644 --- a/docs/design/m4-login-hardening.md +++ b/docs/design/m4-login-hardening.md @@ -199,7 +199,7 @@ Phase B（TOTP，可选配） - **Reviewer checklist**: 现有用户迁移后默认未启用 TOTP（不破坏现有登录）。 ### M4-T05 — TOTP service + setup/enable/disable/status API -- **Status**: `todo` · **Depends**: M4-T04 +- **Status**: `done` · **Depends**: M4-T04 - **Files**: `create app/services/totp.py`、`app/schemas/totp.py`；`modify app/api/routes/api/session.py`（或新 `auth_totp.py` 路由）、`app/config.py`(+`auth_totp_issuer`)；`modify requirements.in/.txt`(+`pyotp`)；`create tests/test_api_totp.py` - **Steps**: pyotp 生成 secret + `otpauth://` URI（issuer=`auth_totp_issuer`）；恢复码生成（N=10、`xxxx-xxxx`，Argon2 哈希存、明文仅 setup 返回一次）；`setup`（pending，不启用）/`enable`（带当前码确认）/`disable`（带密码或当前码）/`GET status`；全部 session+CSRF。 - **Out of scope / 不要碰**: 不改 login 校验（T06）。 diff --git a/openapi/openapi.json b/openapi/openapi.json index 94981ce..95f3c5f 100644 --- a/openapi/openapi.json +++ b/openapi/openapi.json @@ -795,6 +795,184 @@ } } }, + "/api/auth/totp/setup": { + "post": { + "tags": [ + "api-session" + ], + "summary": "Post Totp Setup", + "description": "Generate a new pending TOTP secret, otpauth URI, and one-time recovery codes.\n\nThe secret is stored in the DB but TOTP is NOT yet enabled (totp_enabled stays\nFalse until the user confirms with POST /api/auth/totp/enable).\n\nRecovery codes are returned here as plaintext exactly once; their Argon2 hashes\nare persisted immediately so enable only needs to flip the enabled flag.\n\nRepeating this call replaces any prior pending secret and regenerates codes.\n\nRequires: session cookie + X-CSRF-Token.", + "operationId": "post_totp_setup_api_auth_totp_setup_post", + "parameters": [ + { + "name": "X-CSRF-Token", + "in": "header", + "required": false, + "schema": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "X-Csrf-Token" + } + } + ], + "responses": { + "200": { + "description": "Successful Response", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/TotpSetupResponse" + } + } + } + }, + "422": { + "description": "Validation Error", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/HTTPValidationError" + } + } + } + } + } + } + }, + "/api/auth/totp/enable": { + "post": { + "tags": [ + "api-session" + ], + "summary": "Post Totp Enable", + "description": "Enable TOTP by confirming with the current 6-digit code from the authenticator app.\n\nRequires a prior call to POST /api/auth/totp/setup (so that a pending secret\nexists). On success, totp_enabled becomes True.\n\nReturns 400 if the code is wrong or there is no pending secret.\nRequires: session cookie + X-CSRF-Token.", + "operationId": "post_totp_enable_api_auth_totp_enable_post", + "parameters": [ + { + "name": "X-CSRF-Token", + "in": "header", + "required": false, + "schema": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "X-Csrf-Token" + } + } + ], + "requestBody": { + "required": true, + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/TotpEnableRequest" + } + } + } + }, + "responses": { + "204": { + "description": "Successful Response" + }, + "422": { + "description": "Validation Error", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/HTTPValidationError" + } + } + } + } + } + } + }, + "/api/auth/totp/disable": { + "post": { + "tags": [ + "api-session" + ], + "summary": "Post Totp Disable", + "description": "Disable TOTP. The caller must provide exactly one of:\n- ``password``: the user's current login password, OR\n- ``code``: the current 6-digit TOTP code.\n\nOn success: totp_enabled=False, totp_secret cleared, all recovery codes deleted.\nReturns 400 if neither credential matches or neither is provided.\nRequires: session cookie + X-CSRF-Token.", + "operationId": "post_totp_disable_api_auth_totp_disable_post", + "parameters": [ + { + "name": "X-CSRF-Token", + "in": "header", + "required": false, + "schema": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "X-Csrf-Token" + } + } + ], + "requestBody": { + "required": true, + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/TotpDisableRequest" + } + } + } + }, + "responses": { + "204": { + "description": "Successful Response" + }, + "422": { + "description": "Validation Error", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/HTTPValidationError" + } + } + } + } + } + } + }, + "/api/auth/totp": { + "get": { + "tags": [ + "api-session" + ], + "summary": "Get Totp Status", + "description": "Return the current TOTP status for the authenticated user.\n\nResponse contains only ``{\"enabled\": bool}``.\nSecret and recovery codes are NEVER returned here.\nRequires: session cookie only (no CSRF — read-only).", + "operationId": "get_totp_status_api_auth_totp_get", + "responses": { + "200": { + "description": "Successful Response", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/TotpStatusResponse" + } + } + } + } + } + } + }, "/homeassistant/publish": { "post": { "tags": [ @@ -1549,6 +1727,90 @@ ], "title": "StatusResponse" }, + "TotpDisableRequest": { + "properties": { + "password": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Password" + }, + "code": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Code" + } + }, + "type": "object", + "title": "TotpDisableRequest", + "description": "Disable TOTP by proving identity.\n\nExactly one of ``password`` or ``code`` must be provided." + }, + "TotpEnableRequest": { + "properties": { + "code": { + "type": "string", + "title": "Code" + } + }, + "type": "object", + "required": [ + "code" + ], + "title": "TotpEnableRequest", + "description": "The user confirms setup by providing the 6-digit TOTP code." + }, + "TotpSetupResponse": { + "properties": { + "secret": { + "type": "string", + "title": "Secret" + }, + "otpauth_uri": { + "type": "string", + "title": "Otpauth Uri" + }, + "recovery_codes": { + "items": { + "type": "string" + }, + "type": "array", + "title": "Recovery Codes" + } + }, + "type": "object", + "required": [ + "secret", + "otpauth_uri", + "recovery_codes" + ], + "title": "TotpSetupResponse", + "description": "Returned once after a setup call.\n\n``secret`` and ``recovery_codes`` are **one-time plaintext values**.\nThey are never returned again by any subsequent API call.\nThe frontend must display and instruct the user to save them before confirming." + }, + "TotpStatusResponse": { + "properties": { + "enabled": { + "type": "boolean", + "title": "Enabled" + } + }, + "type": "object", + "required": [ + "enabled" + ], + "title": "TotpStatusResponse", + "description": "Minimal status response — never exposes secret or recovery codes." + }, "ValidationError": { "properties": { "loc": { diff --git a/openapi/openapi.yaml b/openapi/openapi.yaml index 423f543..c26a51b 100644 --- a/openapi/openapi.yaml +++ b/openapi/openapi.yaml @@ -577,6 +577,157 @@ paths: application/json: schema: $ref: '#/components/schemas/HTTPValidationError' + /api/auth/totp/setup: + post: + tags: + - api-session + summary: Post Totp Setup + description: 'Generate a new pending TOTP secret, otpauth URI, and one-time + recovery codes. + + + The secret is stored in the DB but TOTP is NOT yet enabled (totp_enabled stays + + False until the user confirms with POST /api/auth/totp/enable). + + + Recovery codes are returned here as plaintext exactly once; their Argon2 hashes + + are persisted immediately so enable only needs to flip the enabled flag. + + + Repeating this call replaces any prior pending secret and regenerates codes. + + + Requires: session cookie + X-CSRF-Token.' + operationId: post_totp_setup_api_auth_totp_setup_post + parameters: + - name: X-CSRF-Token + in: header + required: false + schema: + anyOf: + - type: string + - type: 'null' + title: X-Csrf-Token + responses: + '200': + description: Successful Response + content: + application/json: + schema: + $ref: '#/components/schemas/TotpSetupResponse' + '422': + description: Validation Error + content: + application/json: + schema: + $ref: '#/components/schemas/HTTPValidationError' + /api/auth/totp/enable: + post: + tags: + - api-session + summary: Post Totp Enable + description: 'Enable TOTP by confirming with the current 6-digit code from the + authenticator app. + + + Requires a prior call to POST /api/auth/totp/setup (so that a pending secret + + exists). On success, totp_enabled becomes True. + + + Returns 400 if the code is wrong or there is no pending secret. + + Requires: session cookie + X-CSRF-Token.' + operationId: post_totp_enable_api_auth_totp_enable_post + parameters: + - name: X-CSRF-Token + in: header + required: false + schema: + anyOf: + - type: string + - type: 'null' + title: X-Csrf-Token + requestBody: + required: true + content: + application/json: + schema: + $ref: '#/components/schemas/TotpEnableRequest' + responses: + '204': + description: Successful Response + '422': + description: Validation Error + content: + application/json: + schema: + $ref: '#/components/schemas/HTTPValidationError' + /api/auth/totp/disable: + post: + tags: + - api-session + summary: Post Totp Disable + description: 'Disable TOTP. The caller must provide exactly one of: + + - ``password``: the user''s current login password, OR + + - ``code``: the current 6-digit TOTP code. + + + On success: totp_enabled=False, totp_secret cleared, all recovery codes deleted. + + Returns 400 if neither credential matches or neither is provided. + + Requires: session cookie + X-CSRF-Token.' + operationId: post_totp_disable_api_auth_totp_disable_post + parameters: + - name: X-CSRF-Token + in: header + required: false + schema: + anyOf: + - type: string + - type: 'null' + title: X-Csrf-Token + requestBody: + required: true + content: + application/json: + schema: + $ref: '#/components/schemas/TotpDisableRequest' + responses: + '204': + description: Successful Response + '422': + description: Validation Error + content: + application/json: + schema: + $ref: '#/components/schemas/HTTPValidationError' + /api/auth/totp: + get: + tags: + - api-session + summary: Get Totp Status + description: 'Return the current TOTP status for the authenticated user. + + + Response contains only ``{"enabled": bool}``. + + Secret and recovery codes are NEVER returned here. + + Requires: session cookie only (no CSRF — read-only).' + operationId: get_totp_status_api_auth_totp_get + responses: + '200': + description: Successful Response + content: + application/json: + schema: + $ref: '#/components/schemas/TotpStatusResponse' /homeassistant/publish: post: tags: @@ -1079,6 +1230,71 @@ components: required: - status title: StatusResponse + TotpDisableRequest: + properties: + password: + anyOf: + - type: string + - type: 'null' + title: Password + code: + anyOf: + - type: string + - type: 'null' + title: Code + type: object + title: TotpDisableRequest + description: 'Disable TOTP by proving identity. + + + Exactly one of ``password`` or ``code`` must be provided.' + TotpEnableRequest: + properties: + code: + type: string + title: Code + type: object + required: + - code + title: TotpEnableRequest + description: The user confirms setup by providing the 6-digit TOTP code. + TotpSetupResponse: + properties: + secret: + type: string + title: Secret + otpauth_uri: + type: string + title: Otpauth Uri + recovery_codes: + items: + type: string + type: array + title: Recovery Codes + type: object + required: + - secret + - otpauth_uri + - recovery_codes + title: TotpSetupResponse + description: 'Returned once after a setup call. + + + ``secret`` and ``recovery_codes`` are **one-time plaintext values**. + + They are never returned again by any subsequent API call. + + The frontend must display and instruct the user to save them before confirming.' + TotpStatusResponse: + properties: + enabled: + type: boolean + title: Enabled + type: object + required: + - enabled + title: TotpStatusResponse + description: Minimal status response — never exposes secret or recovery codes. ValidationError: properties: loc: diff --git a/requirements.in b/requirements.in index f32ec3f..1eb089d 100644 --- a/requirements.in +++ b/requirements.in @@ -4,6 +4,7 @@ argon2-cffi>=25.1,<26.0 fastapi>=0.115,<0.116 httpx>=0.28,<1.0 pydantic-settings>=2.6,<3.0 +pyotp>=2.9,<3.0 python-multipart>=0.0.12,<1.0 pyyaml>=6.0,<7.0 sqlalchemy>=2.0,<3.0 diff --git a/requirements.txt b/requirements.txt index 933a262..389f4e7 100644 --- a/requirements.txt +++ b/requirements.txt @@ -59,6 +59,8 @@ pydantic-core==2.46.2 # via pydantic pydantic-settings==2.13.1 # via -r requirements.in +pyotp==2.10.0 + # via -r requirements.in python-dotenv==1.2.2 # via # pydantic-settings diff --git a/tests/test_api_totp.py b/tests/test_api_totp.py new file mode 100644 index 0000000..40f4e6c --- /dev/null +++ b/tests/test_api_totp.py @@ -0,0 +1,419 @@ +"""Tests for M4-T05: TOTP service + setup/enable/disable/status API.""" +from __future__ import annotations + +import pyotp +import pytest +from fastapi.testclient import TestClient +from sqlalchemy import create_engine, select +from sqlalchemy.orm import Session + +from app.db import reset_db_caches +from app.models.auth import AuthUser, RecoveryCode +from app.services.auth import verify_password + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _login(client: TestClient) -> str: + """Log in and return the CSRF token from the response.""" + resp = client.post( + "/api/auth/login", + json={"username": "admin", "password": "test-password"}, + ) + assert resp.status_code == 200, resp.text + return resp.json()["csrf_token"] + + +def _setup(client: TestClient, csrf: str) -> dict: + resp = client.post( + "/api/auth/totp/setup", + headers={"X-CSRF-Token": csrf}, + ) + return resp + + +def _enable(client: TestClient, csrf: str, code: str): # noqa: ANN201 + return client.post( + "/api/auth/totp/enable", + json={"code": code}, + headers={"X-CSRF-Token": csrf}, + ) + + +def _disable(client: TestClient, csrf: str, *, password: str | None = None, code: str | None = None): + body: dict = {} + if password is not None: + body["password"] = password + if code is not None: + body["code"] = code + return client.post( + "/api/auth/totp/disable", + json=body, + headers={"X-CSRF-Token": csrf}, + ) + + +def _status(client: TestClient) -> dict: + resp = client.get("/api/auth/totp") + assert resp.status_code == 200, resp.text + return resp.json() + + +# --------------------------------------------------------------------------- +# Fixtures — reuse conftest app/client via the session-aware totp_client fixture +# --------------------------------------------------------------------------- + + +@pytest.fixture +def totp_client(auth_database): + """TestClient + a direct SQLAlchemy engine so tests can inspect DB state.""" + from app.main import create_app + + app_url = auth_database["app_url"] + engine = create_engine(app_url, connect_args={"check_same_thread": False}) + fastapi_app = create_app() + with TestClient(fastapi_app) as tc: + yield tc, engine + engine.dispose() + reset_db_caches() + + +# --------------------------------------------------------------------------- +# POST /api/auth/totp/setup +# --------------------------------------------------------------------------- + + +def test_setup_returns_secret_uri_and_recovery_codes(totp_client): + client, engine = totp_client + csrf = _login(client) + + resp = _setup(client, csrf) + + assert resp.status_code == 200, resp.text + body = resp.json() + assert "secret" in body + assert "otpauth_uri" in body + assert "recovery_codes" in body + assert len(body["recovery_codes"]) == 10 + # Each code must match xxxx-xxxx pattern + for rc in body["recovery_codes"]: + parts = rc.split("-") + assert len(parts) == 2 + assert all(len(p) == 4 for p in parts) + # URI must start with otpauth://totp/ + assert body["otpauth_uri"].startswith("otpauth://totp/") + # Secret must be a valid base32 string (pyotp will accept it) + assert len(body["secret"]) >= 16 + + +def test_setup_does_not_enable_totp(totp_client): + """After setup, totp_enabled must still be False.""" + client, engine = totp_client + csrf = _login(client) + + _setup(client, csrf) + + status_body = _status(client) + assert status_body["enabled"] is False + + +def test_setup_persists_secret_and_code_hashes(totp_client): + """After setup, DB has totp_secret set and recovery code hashes stored.""" + client, engine = totp_client + csrf = _login(client) + + resp = _setup(client, csrf) + body = resp.json() + + with Session(engine) as db: + user = db.scalar(select(AuthUser).where(AuthUser.username == "admin")) + assert user is not None + # Secret is stored + assert user.totp_secret == body["secret"] + # Enabled is still False + assert user.totp_enabled is False + # Recovery codes are stored as hashes + codes = db.execute(select(RecoveryCode).where(RecoveryCode.user_id == user.id)).scalars().all() + assert len(codes) == 10 + # Hashes are not the same as plaintext + for i, rc in enumerate(codes): + assert rc.code_hash != body["recovery_codes"][i] + # But they must verify correctly with Argon2 + assert verify_password(body["recovery_codes"][i], rc.code_hash) + + +def test_setup_twice_replaces_pending_state(totp_client): + """Re-running setup clears old pending secret + codes and issues new ones.""" + client, engine = totp_client + csrf = _login(client) + + first = _setup(client, csrf).json() + second = _setup(client, csrf).json() + + # Different secret each time + assert first["secret"] != second["secret"] + # Different codes + assert first["recovery_codes"] != second["recovery_codes"] + + # DB has exactly 10 codes (not 20) + with Session(engine) as db: + user = db.scalar(select(AuthUser).where(AuthUser.username == "admin")) + codes = db.execute( + select(RecoveryCode).where(RecoveryCode.user_id == user.id) + ).scalars().all() + assert len(codes) == 10 + + +# --------------------------------------------------------------------------- +# POST /api/auth/totp/enable +# --------------------------------------------------------------------------- + + +def test_enable_with_wrong_code_fails(totp_client): + client, engine = totp_client + csrf = _login(client) + + _setup(client, csrf) + + resp = _enable(client, csrf, "000000") + assert resp.status_code == 400 + + +def test_enable_with_wrong_code_does_not_enable(totp_client): + client, engine = totp_client + csrf = _login(client) + + _setup(client, csrf) + _enable(client, csrf, "000000") + + assert _status(client)["enabled"] is False + + +def test_enable_with_correct_code_succeeds(totp_client): + """Using pyotp.TOTP(secret).now() to generate the correct code.""" + client, engine = totp_client + csrf = _login(client) + + setup_body = _setup(client, csrf).json() + secret = setup_body["secret"] + + correct_code = pyotp.TOTP(secret).now() + resp = _enable(client, csrf, correct_code) + assert resp.status_code == 204, resp.text + + +def test_enable_with_correct_code_sets_totp_enabled(totp_client): + client, engine = totp_client + csrf = _login(client) + + setup_body = _setup(client, csrf).json() + secret = setup_body["secret"] + + _enable(client, csrf, pyotp.TOTP(secret).now()) + + assert _status(client)["enabled"] is True + + +def test_enable_persists_recovery_code_hashes(totp_client): + """Recovery codes are already hashed in DB after enable (not plaintext).""" + client, engine = totp_client + csrf = _login(client) + + setup_body = _setup(client, csrf).json() + secret = setup_body["secret"] + + _enable(client, csrf, pyotp.TOTP(secret).now()) + + with Session(engine) as db: + user = db.scalar(select(AuthUser).where(AuthUser.username == "admin")) + codes = db.execute(select(RecoveryCode).where(RecoveryCode.user_id == user.id)).scalars().all() + assert len(codes) == 10 + for i, rc in enumerate(codes): + plaintext = setup_body["recovery_codes"][i] + # code_hash must NOT be the plaintext + assert rc.code_hash != plaintext + # But verify_password must return True (Argon2 hash matches) + assert verify_password(plaintext, rc.code_hash) + + +# --------------------------------------------------------------------------- +# POST /api/auth/totp/disable +# --------------------------------------------------------------------------- + + +def _do_full_enable(client: TestClient, csrf: str) -> str: + """Perform setup + enable and return the secret.""" + setup_body = _setup(client, csrf).json() + secret = setup_body["secret"] + _enable(client, csrf, pyotp.TOTP(secret).now()) + return secret + + +def test_disable_with_password_succeeds(totp_client): + client, engine = totp_client + csrf = _login(client) + + _do_full_enable(client, csrf) + assert _status(client)["enabled"] is True + + resp = _disable(client, csrf, password="test-password") + assert resp.status_code == 204, resp.text + + +def test_disable_clears_enabled_flag(totp_client): + client, engine = totp_client + csrf = _login(client) + + _do_full_enable(client, csrf) + _disable(client, csrf, password="test-password") + + assert _status(client)["enabled"] is False + + +def test_disable_clears_secret_and_recovery_codes(totp_client): + client, engine = totp_client + csrf = _login(client) + + _do_full_enable(client, csrf) + _disable(client, csrf, password="test-password") + + with Session(engine) as db: + user = db.scalar(select(AuthUser).where(AuthUser.username == "admin")) + assert user is not None + assert user.totp_secret is None + assert user.totp_enabled is False + codes = db.execute(select(RecoveryCode).where(RecoveryCode.user_id == user.id)).scalars().all() + assert len(codes) == 0 + + +def test_disable_with_totp_code_succeeds(totp_client): + client, engine = totp_client + csrf = _login(client) + + secret = _do_full_enable(client, csrf) + current_code = pyotp.TOTP(secret).now() + + resp = _disable(client, csrf, code=current_code) + assert resp.status_code == 204, resp.text + assert _status(client)["enabled"] is False + + +def test_disable_with_wrong_password_fails(totp_client): + client, engine = totp_client + csrf = _login(client) + + _do_full_enable(client, csrf) + + resp = _disable(client, csrf, password="wrong-password") + assert resp.status_code == 400 + assert _status(client)["enabled"] is True + + +def test_disable_with_no_credential_fails(totp_client): + client, engine = totp_client + csrf = _login(client) + + _do_full_enable(client, csrf) + + resp = _disable(client, csrf) # neither password nor code + assert resp.status_code == 400 + + +# --------------------------------------------------------------------------- +# GET /api/auth/totp — status endpoint +# --------------------------------------------------------------------------- + + +def test_status_does_not_return_secret(totp_client): + client, engine = totp_client + csrf = _login(client) + + setup_body = _setup(client, csrf).json() + secret = setup_body["secret"] + _enable(client, csrf, pyotp.TOTP(secret).now()) + + status_resp = client.get("/api/auth/totp") + assert status_resp.status_code == 200 + body = status_resp.json() + # Must contain ONLY enabled + assert set(body.keys()) == {"enabled"} + assert "secret" not in str(body) + assert "recovery_codes" not in str(body) + assert "otpauth" not in str(body) + + +def test_status_initially_disabled(totp_client): + client, _ = totp_client + _login(client) + + body = _status(client) + assert body["enabled"] is False + + +def test_status_after_enable(totp_client): + client, _ = totp_client + csrf = _login(client) + + setup_body = _setup(client, csrf).json() + _enable(client, csrf, pyotp.TOTP(setup_body["secret"]).now()) + + body = _status(client) + assert body["enabled"] is True + + +# --------------------------------------------------------------------------- +# Security: unauthenticated / missing CSRF +# --------------------------------------------------------------------------- + + +def test_setup_unauthenticated_returns_401(client: TestClient): + resp = client.post("/api/auth/totp/setup", headers={"X-CSRF-Token": "x"}) + assert resp.status_code == 401 + + +def test_setup_missing_csrf_returns_403(totp_client): + client, _ = totp_client + _login(client) + # No X-CSRF-Token header + resp = client.post("/api/auth/totp/setup") + assert resp.status_code == 403 + + +def test_enable_unauthenticated_returns_401(client: TestClient): + resp = client.post( + "/api/auth/totp/enable", + json={"code": "123456"}, + headers={"X-CSRF-Token": "x"}, + ) + assert resp.status_code == 401 + + +def test_enable_missing_csrf_returns_403(totp_client): + client, _ = totp_client + _login(client) + resp = client.post("/api/auth/totp/enable", json={"code": "123456"}) + assert resp.status_code == 403 + + +def test_disable_unauthenticated_returns_401(client: TestClient): + resp = client.post( + "/api/auth/totp/disable", + json={"password": "test-password"}, + headers={"X-CSRF-Token": "x"}, + ) + assert resp.status_code == 401 + + +def test_disable_missing_csrf_returns_403(totp_client): + client, _ = totp_client + _login(client) + resp = client.post("/api/auth/totp/disable", json={"password": "test-password"}) + assert resp.status_code == 403 + + +def test_status_unauthenticated_returns_401(client: TestClient): + resp = client.get("/api/auth/totp") + assert resp.status_code == 401