774 lines
28 KiB
Python
774 lines
28 KiB
Python
"""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
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _api_login(client: TestClient, *, username: str = "admin", password: str = "test-password"):
|
|
"""Log in via POST /api/auth/login and return the response."""
|
|
return client.post(
|
|
"/api/auth/login",
|
|
json={"username": username, "password": password},
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# GET /api/session — unauthenticated
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_get_session_unauthenticated_returns_401(client: TestClient) -> None:
|
|
response = client.get("/api/session")
|
|
assert response.status_code == 401
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# GET /api/session — authenticated (via Jinja login)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_get_session_authenticated_returns_user_and_csrf(client: TestClient) -> None:
|
|
_api_login(client)
|
|
|
|
response = client.get("/api/session")
|
|
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
assert "user" in body
|
|
assert "csrf_token" in body
|
|
assert body["user"]["username"] == "admin"
|
|
assert isinstance(body["user"]["force_password_change"], bool)
|
|
assert isinstance(body["csrf_token"], str)
|
|
assert body["csrf_token"] # non-empty
|
|
|
|
|
|
def test_get_session_does_not_leak_password(client: TestClient) -> None:
|
|
_api_login(client)
|
|
response = client.get("/api/session")
|
|
body_str = str(response.json())
|
|
assert "test-password" not in body_str
|
|
assert "password_hash" not in body_str
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# POST /api/auth/login
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_post_login_valid_credentials_returns_200_with_session(client: TestClient) -> None:
|
|
response = _api_login(client)
|
|
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
assert "user" in body
|
|
assert "csrf_token" in body
|
|
assert body["user"]["username"] == "admin"
|
|
assert isinstance(body["csrf_token"], str)
|
|
assert body["csrf_token"]
|
|
|
|
|
|
def test_post_login_sets_httponly_session_cookie(client: TestClient) -> None:
|
|
response = _api_login(client)
|
|
|
|
assert response.status_code == 200
|
|
set_cookie = response.headers.get("set-cookie", "").lower()
|
|
assert "home_automation_session=" in set_cookie
|
|
assert "httponly" in set_cookie
|
|
assert "samesite=lax" in set_cookie
|
|
assert "path=/" in set_cookie
|
|
|
|
|
|
def test_post_login_cookie_secure_flag_follows_settings(client: TestClient) -> None:
|
|
"""In test mode AUTH_COOKIE_SECURE_OVERRIDE=false so secure should be absent."""
|
|
response = _api_login(client)
|
|
|
|
assert response.status_code == 200
|
|
set_cookie = response.headers.get("set-cookie", "").lower()
|
|
# secure is absent because AUTH_COOKIE_SECURE_OVERRIDE=false in conftest
|
|
assert "secure" not in set_cookie
|
|
|
|
|
|
def test_post_login_invalid_credentials_returns_401(client: TestClient) -> None:
|
|
response = _api_login(client, password="wrong-password")
|
|
|
|
assert response.status_code == 401
|
|
# No session cookie should be set
|
|
assert "set-cookie" not in response.headers or (
|
|
"home_automation_session=" not in response.headers.get("set-cookie", "").lower()
|
|
)
|
|
|
|
|
|
def test_post_login_unknown_user_returns_401(client: TestClient) -> None:
|
|
response = _api_login(client, username="nobody", password="irrelevant")
|
|
assert response.status_code == 401
|
|
|
|
|
|
def test_post_login_does_not_require_csrf_header(client: TestClient) -> None:
|
|
"""Login is unauthenticated; no X-CSRF-Token should be required."""
|
|
response = client.post(
|
|
"/api/auth/login",
|
|
json={"username": "admin", "password": "test-password"},
|
|
# Deliberately omit X-CSRF-Token
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
|
|
def test_post_login_allows_subsequent_authenticated_request(client: TestClient) -> None:
|
|
login_resp = _api_login(client)
|
|
assert login_resp.status_code == 200
|
|
|
|
# GET /api/session should now succeed (cookie was set on the client)
|
|
session_resp = client.get("/api/session")
|
|
assert session_resp.status_code == 200
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# POST /api/auth/logout
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_post_logout_unauthenticated_returns_401(client: TestClient) -> None:
|
|
response = client.post("/api/auth/logout", headers={"X-CSRF-Token": "token"})
|
|
assert response.status_code == 401
|
|
|
|
|
|
def test_post_logout_authenticated_missing_csrf_returns_403(client: TestClient) -> None:
|
|
_api_login(client)
|
|
response = client.post("/api/auth/logout")
|
|
assert response.status_code == 403
|
|
|
|
|
|
def test_post_logout_authenticated_empty_csrf_returns_403(client: TestClient) -> None:
|
|
_api_login(client)
|
|
response = client.post("/api/auth/logout", headers={"X-CSRF-Token": ""})
|
|
assert response.status_code == 403
|
|
|
|
|
|
def test_post_logout_authenticated_with_csrf_returns_204(client: TestClient) -> None:
|
|
_api_login(client)
|
|
response = client.post("/api/auth/logout", headers={"X-CSRF-Token": "any-non-empty-value"})
|
|
assert response.status_code == 204
|
|
|
|
|
|
def test_post_logout_invalidates_session(client: TestClient) -> None:
|
|
_api_login(client)
|
|
|
|
# Verify session is active
|
|
assert client.get("/api/session").status_code == 200
|
|
|
|
# Logout
|
|
client.post("/api/auth/logout", headers={"X-CSRF-Token": "token"})
|
|
|
|
# Session should now be gone
|
|
assert client.get("/api/session").status_code == 401
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# POST /api/auth/password
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_post_password_unauthenticated_returns_401(client: TestClient) -> None:
|
|
response = client.post(
|
|
"/api/auth/password",
|
|
json={
|
|
"current_password": "test-password",
|
|
"new_password": "new-password-123",
|
|
"confirm_password": "new-password-123",
|
|
},
|
|
headers={"X-CSRF-Token": "token"},
|
|
)
|
|
assert response.status_code == 401
|
|
|
|
|
|
def test_post_password_authenticated_missing_csrf_returns_403(client: TestClient) -> None:
|
|
_api_login(client)
|
|
response = client.post(
|
|
"/api/auth/password",
|
|
json={
|
|
"current_password": "test-password",
|
|
"new_password": "new-password-123",
|
|
"confirm_password": "new-password-123",
|
|
},
|
|
)
|
|
assert response.status_code == 403
|
|
|
|
|
|
def test_post_password_success_returns_204(client: TestClient) -> None:
|
|
_api_login(client)
|
|
response = client.post(
|
|
"/api/auth/password",
|
|
json={
|
|
"current_password": "test-password",
|
|
"new_password": "new-password-123",
|
|
"confirm_password": "new-password-123",
|
|
},
|
|
headers={"X-CSRF-Token": "token"},
|
|
)
|
|
assert response.status_code == 204
|
|
|
|
|
|
def test_post_password_wrong_current_password_returns_400(client: TestClient) -> None:
|
|
_api_login(client)
|
|
response = client.post(
|
|
"/api/auth/password",
|
|
json={
|
|
"current_password": "wrong-current",
|
|
"new_password": "new-password-123",
|
|
"confirm_password": "new-password-123",
|
|
},
|
|
headers={"X-CSRF-Token": "token"},
|
|
)
|
|
assert response.status_code == 400
|
|
# Error message must be generic — no leaking which check failed
|
|
detail = response.json().get("detail", "")
|
|
assert "current password is invalid" not in detail
|
|
assert detail == "password change failed"
|
|
|
|
|
|
def test_post_password_mismatched_new_passwords_returns_400(client: TestClient) -> None:
|
|
_api_login(client)
|
|
response = client.post(
|
|
"/api/auth/password",
|
|
json={
|
|
"current_password": "test-password",
|
|
"new_password": "new-password-123",
|
|
"confirm_password": "different-password-123",
|
|
},
|
|
headers={"X-CSRF-Token": "token"},
|
|
)
|
|
assert response.status_code == 400
|
|
assert response.json()["detail"] == "password change failed"
|
|
|
|
|
|
def test_post_password_too_short_returns_400(client: TestClient) -> None:
|
|
_api_login(client)
|
|
response = client.post(
|
|
"/api/auth/password",
|
|
json={
|
|
"current_password": "test-password",
|
|
"new_password": "short",
|
|
"confirm_password": "short",
|
|
},
|
|
headers={"X-CSRF-Token": "token"},
|
|
)
|
|
assert response.status_code == 400
|
|
assert response.json()["detail"] == "password change failed"
|
|
|
|
|
|
def test_post_password_same_as_current_returns_400(client: TestClient) -> None:
|
|
_api_login(client)
|
|
response = client.post(
|
|
"/api/auth/password",
|
|
json={
|
|
"current_password": "test-password",
|
|
"new_password": "test-password",
|
|
"confirm_password": "test-password",
|
|
},
|
|
headers={"X-CSRF-Token": "token"},
|
|
)
|
|
assert response.status_code == 400
|
|
assert response.json()["detail"] == "password change failed"
|
|
|
|
|
|
def test_post_password_success_sets_force_password_change_false(client: TestClient) -> None:
|
|
"""After successful password change, force_password_change should be False."""
|
|
_api_login(client)
|
|
|
|
# The bootstrap user always has force_password_change=True; change it
|
|
resp = client.post(
|
|
"/api/auth/password",
|
|
json={
|
|
"current_password": "test-password",
|
|
"new_password": "new-password-123",
|
|
"confirm_password": "new-password-123",
|
|
},
|
|
headers={"X-CSRF-Token": "token"},
|
|
)
|
|
assert resp.status_code == 204
|
|
|
|
# Session still active; force_password_change should now be False
|
|
session_resp = client.get("/api/session")
|
|
assert session_resp.status_code == 200
|
|
assert session_resp.json()["user"]["force_password_change"] is False
|
|
|
|
|
|
def test_post_password_does_not_revoke_session(client: TestClient) -> None:
|
|
"""After password change, the session remains valid (not revoked)."""
|
|
_api_login(client)
|
|
|
|
client.post(
|
|
"/api/auth/password",
|
|
json={
|
|
"current_password": "test-password",
|
|
"new_password": "new-password-123",
|
|
"confirm_password": "new-password-123",
|
|
},
|
|
headers={"X-CSRF-Token": "token"},
|
|
)
|
|
|
|
# Session must still be active
|
|
assert client.get("/api/session").status_code == 200
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Response schema correctness — no secrets in session response
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_session_response_has_no_secret_fields(client: TestClient) -> None:
|
|
login_resp = _api_login(client)
|
|
assert login_resp.status_code == 200
|
|
body = login_resp.json()
|
|
|
|
# Must have exactly these top-level keys
|
|
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
|