From 81bd32f613c12ec11a553840a951344b91b0243c Mon Sep 17 00:00:00 2001 From: Tianyu Liu Date: Sun, 21 Jun 2026 22:08:55 +0200 Subject: [PATCH] M4-T06: add TOTP second factor to login (totp_code + recovery codes) --- app/api/routes/api/session.py | 30 ++ app/schemas/session.py | 1 + app/services/totp.py | 13 + docs/design/m4-login-hardening.md | 2 +- openapi/openapi.json | 11 + openapi/openapi.yaml | 5 + tests/test_api_session.py | 443 +++++++++++++++++++++++++++++- tests/test_api_totp.py | 110 ++++++++ 8 files changed, 613 insertions(+), 2 deletions(-) diff --git a/app/api/routes/api/session.py b/app/api/routes/api/session.py index 25c72ec..a587a21 100644 --- a/app/api/routes/api/session.py +++ b/app/api/routes/api/session.py @@ -110,6 +110,36 @@ def post_login( detail="invalid username or password", ) + # --- TOTP second-factor check (only when TOTP is enabled for the user) --- + if user.totp_enabled: + if not body.totp_code: + # Password correct but no TOTP code supplied: signal the front-end to + # prompt for the second factor. Do NOT issue a session. + # Deliberate: this is a normal two-step protocol step from a legitimate + # user — we do NOT register a throttle failure here. The attacker must + # already have the correct password to reach this branch, and counting + # each normal first-step as a failure would risk locking out the real + # admin on every login attempt. + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail={"totp_required": True}, + ) + + # TOTP code supplied — verify against time-based code or a recovery code. + totp_ok = totp_service.verify_totp_code(user, body.totp_code) + if not totp_ok: + totp_ok = totp_service.verify_recovery_code(db, user=user, code=body.totp_code) + + if not totp_ok: + # Second-factor failure is an active attack signal — register failure + # so that repeated wrong TOTP/recovery-code guesses trigger back-off. + if settings.auth_login_throttle_enabled: + login_throttle.register_failure(db, ip=client_ip, username=body.username) + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="invalid username or password", + ) + # --- Success: clear back-off state and issue session --- if settings.auth_login_throttle_enabled: login_throttle.clear(db, ip=client_ip, username=body.username) diff --git a/app/schemas/session.py b/app/schemas/session.py index 6eecaac..71e1180 100644 --- a/app/schemas/session.py +++ b/app/schemas/session.py @@ -16,6 +16,7 @@ class SessionResponse(BaseModel): class LoginRequest(BaseModel): username: str password: str + totp_code: str | None = None class PasswordChangeRequest(BaseModel): diff --git a/app/services/totp.py b/app/services/totp.py index 464421b..cc4627d 100644 --- a/app/services/totp.py +++ b/app/services/totp.py @@ -190,6 +190,19 @@ def disable( return True +def verify_totp_code(user: AuthUser, code: str) -> bool: + """Verify a 6-digit TOTP code for a user. + + Returns True if the code is valid (±1 time window), False otherwise. + The user must have a ``totp_secret`` set; returns False if not. + + This is the public entry-point for T06 login two-factor verification. + """ + if not user.totp_secret: + return False + return _verify_totp_code(user.totp_secret, code) + + def verify_recovery_code(db: Session, *, user: AuthUser, code: str) -> bool: """Verify and consume a one-time recovery code. diff --git a/docs/design/m4-login-hardening.md b/docs/design/m4-login-hardening.md index e831804..7aeb867 100644 --- a/docs/design/m4-login-hardening.md +++ b/docs/design/m4-login-hardening.md @@ -212,7 +212,7 @@ Phase B（TOTP，可选配） - **Reviewer checklist**: 恢复码只存哈希、明文仅一次；secret 不进 OpenAPI 示例/日志；`requirements.txt` 锁定 `pyotp`。 ### M4-T06 — 登录二因子（login 加 `totp_code`） -- **Status**: `todo` · **Depends**: M4-T05, M4-T02 +- **Status**: `done` · **Depends**: M4-T05, M4-T02 - **Files**: `modify app/api/routes/api/session.py`、`app/schemas/session.py`（`LoginRequest.totp_code` 可选）、`app/services/totp.py`（校验 + 消费恢复码）；`modify tests/test_api_session.py`、`tests/test_api_totp.py` - **Steps**: 密码通过后：若 `totp_enabled` 且无 `totp_code` → 401 `{totp_required:true}`（不发 session、退避照算）；有 `totp_code` → 校验 TOTP 或恢复码（命中恢复码置 `used_at`）→ 通过发 session；未启用 → 现状。 - **Out of scope / 不要碰**: 不改 setup/enable（T05）；不动退避公式（T02）。 diff --git a/openapi/openapi.json b/openapi/openapi.json index 95f3c5f..7956d51 100644 --- a/openapi/openapi.json +++ b/openapi/openapi.json @@ -1347,6 +1347,17 @@ "password": { "type": "string", "title": "Password" + }, + "totp_code": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Totp Code" } }, "type": "object", diff --git a/openapi/openapi.yaml b/openapi/openapi.yaml index c26a51b..dadbd96 100644 --- a/openapi/openapi.yaml +++ b/openapi/openapi.yaml @@ -977,6 +977,11 @@ components: password: type: string title: Password + totp_code: + anyOf: + - type: string + - type: 'null' + title: Totp Code type: object required: - username diff --git a/tests/test_api_session.py b/tests/test_api_session.py index 5d932b1..b5b4f0a 100644 --- a/tests/test_api_session.py +++ b/tests/test_api_session.py @@ -1,7 +1,24 @@ -"""Tests for M2-T02: GET /api/session, POST /api/auth/login, /logout, /password.""" +"""Tests for M2-T02: GET /api/session, POST /api/auth/login, /logout, /password. + +Also covers M4-T06: two-factor login regression tests (TOTP not enabled → same +as before; TOTP enabled → two-step protocol). +""" from __future__ import annotations +from pathlib import Path + +import pyotp +import pytest +from alembic import command +from alembic.config import Config from fastapi.testclient import TestClient +from sqlalchemy import create_engine +from sqlalchemy.orm import Session + +from app.db import reset_db_caches +from app.config import get_settings +from app.main import create_app +from app.services.login_throttle import N_FREE # --------------------------------------------------------------------------- @@ -330,3 +347,427 @@ def test_session_response_has_no_secret_fields(client: TestClient) -> None: assert set(body.keys()) == {"user", "csrf_token"} # user must have exactly these keys assert set(body["user"].keys()) == {"username", "force_password_change"} + + +# --------------------------------------------------------------------------- +# M4-T06 — Two-factor login (TOTP second step) +# Fixtures & helpers +# --------------------------------------------------------------------------- + + +def _make_app_alembic_config(database_url: str) -> Config: + cfg = Config("alembic_app.ini") + cfg.set_main_option("sqlalchemy.url", database_url) + return cfg + + +@pytest.fixture +def totp_login_client(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): + """Full client with TOTP-capable DB (throttle disabled for simplicity).""" + db_path = tmp_path / "totp_login_test.db" + db_url = f"sqlite:///{db_path}" + + monkeypatch.setenv("APP_DATABASE_URL", db_url) + monkeypatch.setenv("AUTH_BOOTSTRAP_USERNAME", "admin") + monkeypatch.setenv("AUTH_BOOTSTRAP_PASSWORD", "test-password") + monkeypatch.setenv("AUTH_COOKIE_SECURE_OVERRIDE", "false") + monkeypatch.setenv("AUTH_LOGIN_THROTTLE_ENABLED", "false") + get_settings.cache_clear() + reset_db_caches() + + command.upgrade(_make_app_alembic_config(db_url), "head") + reset_db_caches() + + app = create_app() + with TestClient(app) as tc: + yield tc + + get_settings.cache_clear() + reset_db_caches() + + +@pytest.fixture +def totp_login_throttle_client(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): + """Full client with TOTP + throttle ENABLED (for throttle-interaction tests).""" + db_path = tmp_path / "totp_throttle_test.db" + db_url = f"sqlite:///{db_path}" + + monkeypatch.setenv("APP_DATABASE_URL", db_url) + monkeypatch.setenv("AUTH_BOOTSTRAP_USERNAME", "admin") + monkeypatch.setenv("AUTH_BOOTSTRAP_PASSWORD", "test-password") + monkeypatch.setenv("AUTH_COOKIE_SECURE_OVERRIDE", "false") + monkeypatch.setenv("AUTH_LOGIN_THROTTLE_ENABLED", "true") + get_settings.cache_clear() + reset_db_caches() + + command.upgrade(_make_app_alembic_config(db_url), "head") + reset_db_caches() + + engine = create_engine(db_url, connect_args={"check_same_thread": False}) + app = create_app() + with TestClient(app) as tc: + yield tc, engine + engine.dispose() + get_settings.cache_clear() + reset_db_caches() + + +def _enable_totp(client: TestClient) -> tuple[str, list[str]]: + """Log in, setup TOTP, enable it; return (secret, recovery_codes).""" + login_resp = client.post( + "/api/auth/login", + json={"username": "admin", "password": "test-password"}, + ) + assert login_resp.status_code == 200, login_resp.text + csrf = login_resp.json()["csrf_token"] + + setup_resp = client.post("/api/auth/totp/setup", headers={"X-CSRF-Token": csrf}) + assert setup_resp.status_code == 200, setup_resp.text + setup_body = setup_resp.json() + secret = setup_body["secret"] + recovery_codes = setup_body["recovery_codes"] + + enable_resp = client.post( + "/api/auth/totp/enable", + json={"code": pyotp.TOTP(secret).now()}, + headers={"X-CSRF-Token": csrf}, + ) + assert enable_resp.status_code == 204, enable_resp.text + + # Log out so subsequent login tests start fresh + client.post("/api/auth/logout", headers={"X-CSRF-Token": csrf}) + return secret, recovery_codes + + +# --------------------------------------------------------------------------- +# M4-T06 — Regression: TOTP not enabled → login unchanged +# --------------------------------------------------------------------------- + + +class TestLoginWithoutTotpEnabled: + """TOTP not enabled: login behaves exactly as before T06.""" + + def test_login_without_totp_code_succeeds(self, totp_login_client: TestClient) -> None: + """Standard login (no totp_code field at all) still works when TOTP is off.""" + resp = totp_login_client.post( + "/api/auth/login", + json={"username": "admin", "password": "test-password"}, + ) + assert resp.status_code == 200 + body = resp.json() + assert "user" in body + assert "csrf_token" in body + + def test_login_with_null_totp_code_succeeds(self, totp_login_client: TestClient) -> None: + """Explicitly sending totp_code=null works fine when TOTP is off.""" + resp = totp_login_client.post( + "/api/auth/login", + json={"username": "admin", "password": "test-password", "totp_code": None}, + ) + assert resp.status_code == 200 + + def test_login_sets_session_cookie(self, totp_login_client: TestClient) -> None: + resp = totp_login_client.post( + "/api/auth/login", + json={"username": "admin", "password": "test-password"}, + ) + assert resp.status_code == 200 + assert "home_automation_session=" in resp.headers.get("set-cookie", "").lower() + + def test_wrong_password_still_401(self, totp_login_client: TestClient) -> None: + resp = totp_login_client.post( + "/api/auth/login", + json={"username": "admin", "password": "wrong"}, + ) + assert resp.status_code == 401 + # Must be the standard generic message + assert resp.json()["detail"] == "invalid username or password" + + def test_wrong_password_no_cookie(self, totp_login_client: TestClient) -> None: + resp = totp_login_client.post( + "/api/auth/login", + json={"username": "admin", "password": "wrong"}, + ) + assert resp.status_code == 401 + assert "home_automation_session=" not in resp.headers.get("set-cookie", "").lower() + + +# --------------------------------------------------------------------------- +# M4-T06 — TOTP enabled: two-step protocol +# --------------------------------------------------------------------------- + + +class TestLoginWithTotpEnabled: + """TOTP enabled: missing/wrong/correct code semantics.""" + + def test_missing_totp_code_returns_401_with_totp_required( + self, totp_login_client: TestClient + ) -> None: + """Correct password but no totp_code → 401 with totp_required signal.""" + _enable_totp(totp_login_client) + resp = totp_login_client.post( + "/api/auth/login", + json={"username": "admin", "password": "test-password"}, + ) + assert resp.status_code == 401 + body = resp.json() + # detail must be a dict with totp_required=true + assert isinstance(body["detail"], dict) + assert body["detail"].get("totp_required") is True + + def test_totp_required_response_has_no_session_cookie( + self, totp_login_client: TestClient + ) -> None: + """totp_required 401 must NOT set a session cookie.""" + _enable_totp(totp_login_client) + resp = totp_login_client.post( + "/api/auth/login", + json={"username": "admin", "password": "test-password"}, + ) + assert resp.status_code == 401 + assert "home_automation_session=" not in resp.headers.get("set-cookie", "").lower() + + def test_wrong_totp_code_returns_generic_401( + self, totp_login_client: TestClient + ) -> None: + """Wrong TOTP code → generic 401 (no totp_required signal).""" + _enable_totp(totp_login_client) + resp = totp_login_client.post( + "/api/auth/login", + json={"username": "admin", "password": "test-password", "totp_code": "000000"}, + ) + assert resp.status_code == 401 + detail = resp.json()["detail"] + # Must be a plain string (not dict with totp_required) + assert isinstance(detail, str) + assert detail == "invalid username or password" + + def test_wrong_totp_code_no_cookie(self, totp_login_client: TestClient) -> None: + """Wrong TOTP → no cookie.""" + _enable_totp(totp_login_client) + resp = totp_login_client.post( + "/api/auth/login", + json={"username": "admin", "password": "test-password", "totp_code": "000000"}, + ) + assert resp.status_code == 401 + assert "home_automation_session=" not in resp.headers.get("set-cookie", "").lower() + + def test_correct_totp_code_issues_session(self, totp_login_client: TestClient) -> None: + """Correct TOTP code → 200 + cookie + csrf.""" + secret, _ = _enable_totp(totp_login_client) + resp = totp_login_client.post( + "/api/auth/login", + json={ + "username": "admin", + "password": "test-password", + "totp_code": pyotp.TOTP(secret).now(), + }, + ) + assert resp.status_code == 200 + body = resp.json() + assert "user" in body + assert "csrf_token" in body + assert "home_automation_session=" in resp.headers.get("set-cookie", "").lower() + + def test_totp_required_detail_not_on_wrong_password( + self, totp_login_client: TestClient + ) -> None: + """Wrong password must NOT carry totp_required (cannot reveal TOTP state).""" + _enable_totp(totp_login_client) + resp = totp_login_client.post( + "/api/auth/login", + json={"username": "admin", "password": "wrong-password"}, + ) + assert resp.status_code == 401 + detail = resp.json()["detail"] + assert isinstance(detail, str), "Wrong-password 401 must not expose totp_required" + if isinstance(detail, dict): + assert "totp_required" not in detail + + +# --------------------------------------------------------------------------- +# M4-T06 — Recovery codes can be used for login (one-time) +# --------------------------------------------------------------------------- + + +class TestRecoveryCodeLogin: + """Recovery codes work as second factor and are consumed after use.""" + + def test_recovery_code_login_succeeds(self, totp_login_client: TestClient) -> None: + _, recovery_codes = _enable_totp(totp_login_client) + code = recovery_codes[0] + resp = totp_login_client.post( + "/api/auth/login", + json={"username": "admin", "password": "test-password", "totp_code": code}, + ) + assert resp.status_code == 200, resp.text + + def test_recovery_code_is_one_time_only(self, totp_login_client: TestClient) -> None: + """Using the same recovery code twice: second use must fail (401).""" + _, recovery_codes = _enable_totp(totp_login_client) + code = recovery_codes[0] + + # First use: success + resp1 = totp_login_client.post( + "/api/auth/login", + json={"username": "admin", "password": "test-password", "totp_code": code}, + ) + assert resp1.status_code == 200, resp1.text + + # Log out + csrf = resp1.json()["csrf_token"] + totp_login_client.post("/api/auth/logout", headers={"X-CSRF-Token": csrf}) + + # Second use: must fail + resp2 = totp_login_client.post( + "/api/auth/login", + json={"username": "admin", "password": "test-password", "totp_code": code}, + ) + assert resp2.status_code == 401, ( + "Recovery code must be rejected on second use" + ) + + def test_recovery_code_used_at_set_in_db(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """After consuming a recovery code, its used_at is set in the DB.""" + from app.models.auth import RecoveryCode + from sqlalchemy import select + + db_path = tmp_path / "rc_used_test.db" + db_url = f"sqlite:///{db_path}" + + monkeypatch.setenv("APP_DATABASE_URL", db_url) + monkeypatch.setenv("AUTH_BOOTSTRAP_USERNAME", "admin") + monkeypatch.setenv("AUTH_BOOTSTRAP_PASSWORD", "test-password") + monkeypatch.setenv("AUTH_COOKIE_SECURE_OVERRIDE", "false") + monkeypatch.setenv("AUTH_LOGIN_THROTTLE_ENABLED", "false") + get_settings.cache_clear() + reset_db_caches() + command.upgrade(_make_app_alembic_config(db_url), "head") + reset_db_caches() + + engine = create_engine(db_url, connect_args={"check_same_thread": False}) + app = create_app() + with TestClient(app) as tc: + secret, recovery_codes = _enable_totp(tc) + code = recovery_codes[0] + + tc.post( + "/api/auth/login", + json={"username": "admin", "password": "test-password", "totp_code": code}, + ) + + with Session(engine) as db: + used_codes = db.execute( + select(RecoveryCode).where(RecoveryCode.used_at.isnot(None)) + ).scalars().all() + assert len(used_codes) >= 1, "At least one recovery code should have used_at set" + + engine.dispose() + get_settings.cache_clear() + reset_db_caches() + + def test_other_recovery_codes_still_work_after_one_consumed( + self, totp_login_client: TestClient + ) -> None: + """Consuming one recovery code leaves the others intact.""" + _, recovery_codes = _enable_totp(totp_login_client) + + # Use first code + resp1 = totp_login_client.post( + "/api/auth/login", + json={"username": "admin", "password": "test-password", "totp_code": recovery_codes[0]}, + ) + assert resp1.status_code == 200 + csrf = resp1.json()["csrf_token"] + totp_login_client.post("/api/auth/logout", headers={"X-CSRF-Token": csrf}) + + # Second code must still work + resp2 = totp_login_client.post( + "/api/auth/login", + json={"username": "admin", "password": "test-password", "totp_code": recovery_codes[1]}, + ) + assert resp2.status_code == 200, ( + "Second recovery code should still be valid after first is consumed" + ) + + +# --------------------------------------------------------------------------- +# M4-T06 — Throttle interaction with TOTP second step +# --------------------------------------------------------------------------- + + +class TestThrottleWithTotpEnabled: + """Verify throttle semantics when TOTP is enabled: + - Missing code (totp_required) does NOT count as a throttle failure. + - Wrong TOTP/recovery code DOES count as a throttle failure. + - Throttle check at step 1 still gates step 2. + """ + + def test_missing_code_does_not_register_throttle_failure( + self, totp_login_throttle_client + ) -> None: + """Sending correct password without totp_code must not count as a failure. + + If this test sends N_FREE+1 such requests without accumulating throttle + failures, the next request will still get 401 (totp_required) not 429. + """ + client, engine = totp_login_throttle_client + _enable_totp(client) + + # Send N_FREE + 2 requests with correct password but no totp_code. + for i in range(N_FREE + 2): + resp = client.post( + "/api/auth/login", + json={"username": "admin", "password": "test-password"}, + ) + # Must never become 429 — each is a legitimate first-step + assert resp.status_code == 401, ( + f"Iteration {i}: expected 401 (totp_required), got {resp.status_code}" + ) + body = resp.json() + assert isinstance(body["detail"], dict) + assert body["detail"].get("totp_required") is True + + def test_wrong_totp_code_accumulates_throttle_failures( + self, totp_login_throttle_client + ) -> None: + """Wrong TOTP codes should eventually trigger 429.""" + client, engine = totp_login_throttle_client + _enable_totp(client) + + # Send N_FREE + 1 wrong-code requests — each should register a failure + for _ in range(N_FREE + 1): + client.post( + "/api/auth/login", + json={"username": "admin", "password": "test-password", "totp_code": "000000"}, + ) + + # Next attempt (even with correct password + no code) should hit 429 + resp = client.post( + "/api/auth/login", + json={"username": "admin", "password": "test-password"}, + ) + assert resp.status_code == 429, ( + "After N_FREE+1 wrong TOTP codes the throttle should kick in" + ) + + def test_throttle_blocks_step_one_when_in_window( + self, totp_login_throttle_client + ) -> None: + """Once throttled (by wrong codes), even step-1 attempt is rejected 429.""" + client, engine = totp_login_throttle_client + _enable_totp(client) + + # Accumulate failures via wrong TOTP codes + for _ in range(N_FREE + 1): + client.post( + "/api/auth/login", + json={"username": "admin", "password": "test-password", "totp_code": "000000"}, + ) + + # Throttle check is before password — even wrong password should get 429 + resp = client.post( + "/api/auth/login", + json={"username": "admin", "password": "wrong"}, + ) + assert resp.status_code == 429 diff --git a/tests/test_api_totp.py b/tests/test_api_totp.py index 40f4e6c..f0a3a6c 100644 --- a/tests/test_api_totp.py +++ b/tests/test_api_totp.py @@ -417,3 +417,113 @@ def test_disable_missing_csrf_returns_403(totp_client): def test_status_unauthenticated_returns_401(client: TestClient): resp = client.get("/api/auth/totp") assert resp.status_code == 401 + + +# --------------------------------------------------------------------------- +# M4-T06 — verify_recovery_code service-level tests +# --------------------------------------------------------------------------- + + +def _full_setup_and_enable(client: TestClient) -> tuple[str, list[str]]: + """Log in, run setup, enable TOTP, return (secret, plaintext_recovery_codes).""" + csrf = _login(client) + setup_body = _setup(client, csrf).json() + secret = setup_body["secret"] + recovery_codes = setup_body["recovery_codes"] + _enable(client, csrf, pyotp.TOTP(secret).now()) + return secret, recovery_codes + + +class TestVerifyRecoveryCodeService: + """Unit-level checks for totp_service.verify_recovery_code (used in T06 login).""" + + def test_verify_recovery_code_returns_true_on_valid_code(self, totp_client): + """verify_recovery_code returns True for an unused code.""" + from app.models.auth import AuthUser + from sqlalchemy import select + from app.services import totp as totp_svc + + client, engine = totp_client + _, recovery_codes = _full_setup_and_enable(client) + code = recovery_codes[0] + + with Session(engine) as db: + user = db.scalar(select(AuthUser).where(AuthUser.username == "admin")) + result = totp_svc.verify_recovery_code(db, user=user, code=code) + assert result is True + + def test_verify_recovery_code_returns_false_on_used_code(self, totp_client): + """verify_recovery_code returns False when the code was already consumed.""" + from app.models.auth import AuthUser + from sqlalchemy import select + from app.services import totp as totp_svc + + client, engine = totp_client + _, recovery_codes = _full_setup_and_enable(client) + code = recovery_codes[0] + + with Session(engine) as db: + user = db.scalar(select(AuthUser).where(AuthUser.username == "admin")) + # First call consumes + assert totp_svc.verify_recovery_code(db, user=user, code=code) is True + # Reload user after commit + db.expire_all() + user = db.scalar(select(AuthUser).where(AuthUser.username == "admin")) + # Second call must fail + assert totp_svc.verify_recovery_code(db, user=user, code=code) is False + + def test_verify_recovery_code_returns_false_on_invalid_code(self, totp_client): + """verify_recovery_code returns False for a completely wrong code.""" + from app.models.auth import AuthUser + from sqlalchemy import select + from app.services import totp as totp_svc + + client, engine = totp_client + _full_setup_and_enable(client) + + with Session(engine) as db: + user = db.scalar(select(AuthUser).where(AuthUser.username == "admin")) + result = totp_svc.verify_recovery_code(db, user=user, code="xxxx-xxxx") + assert result is False + + def test_verify_totp_code_correct(self, totp_client): + """verify_totp_code returns True for the current TOTP code.""" + from app.models.auth import AuthUser + from sqlalchemy import select + from app.services import totp as totp_svc + + client, engine = totp_client + secret, _ = _full_setup_and_enable(client) + + with Session(engine) as db: + user = db.scalar(select(AuthUser).where(AuthUser.username == "admin")) + result = totp_svc.verify_totp_code(user, pyotp.TOTP(secret).now()) + assert result is True + + def test_verify_totp_code_wrong(self, totp_client): + """verify_totp_code returns False for a wrong code.""" + from app.models.auth import AuthUser + from sqlalchemy import select + from app.services import totp as totp_svc + + client, engine = totp_client + _full_setup_and_enable(client) + + with Session(engine) as db: + user = db.scalar(select(AuthUser).where(AuthUser.username == "admin")) + result = totp_svc.verify_totp_code(user, "000000") + assert result is False + + def test_verify_totp_code_no_secret(self, totp_client): + """verify_totp_code returns False if user has no secret (TOTP not set up).""" + from app.models.auth import AuthUser + from sqlalchemy import select + from app.services import totp as totp_svc + + client, engine = totp_client + + with Session(engine) as db: + user = db.scalar(select(AuthUser).where(AuthUser.username == "admin")) + assert user.totp_secret is None + result = totp_svc.verify_totp_code(user, "123456") + assert result is False