public_git/bambuddy
1
0
Fork 0
mirror of https://github.com/maziggy/bambuddy.git synced 2026-10-18 11:56:01 +02:00
Code Issues Packages Projects Releases Wiki Activity

feat: Two-Factor Authentication (TOTP, Email OTP) and OIDC/SSO – full implementation with admin UI (#933)

Browse Source 
feat: Two-Factor Authentication (TOTP, Email OTP) and OIDC/SSO – full implementation with admin UI (#933)
This commit is contained in:
Sn0rrii
2026-04-13 13:24:28 +02:00
committed by GitHub
parent 8af0966e68
commit ba1c97c808
44 changed files with 10473 additions and 290 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/tests/conftest.py
+4
View File
@@ -70,6 +70,7 @@ async def test_engine():
ams_label,
api_key,
archive,
auth_ephemeral,
color_catalog,
external_link,
filament,
@@ -78,6 +79,7 @@ async def test_engine():
maintenance,
notification,
notification_template,
oidc_provider,
print_queue,
printer,
project,
@@ -94,6 +96,8 @@ async def test_engine():
spoolbuddy_device,
user,
user_email_pref,
user_otp_code,
user_totp,
virtual_printer,
)
backend/tests/integration/test_advanced_auth_api.py
+111 -27
View File
@@ -22,7 +22,7 @@ SMTP_DATA = {
}
async def _setup_admin(async_client: AsyncClient, username: str = "admin", password: str = "adminpass123"):
async def _setup_admin(async_client: AsyncClient, username: str = "admin", password: str = "AdminPass1!"):
"""Enable auth and create admin user, return admin token."""
await async_client.post(
"/api/v1/auth/setup",
@@ -47,7 +47,7 @@ async def _setup_smtp_and_advanced_auth(async_client: AsyncClient, token: str):
async def _create_regular_user(
async_client: AsyncClient, token: str, username: str = "regular", password: str = "regularpass123"
async_client: AsyncClient, token: str, username: str = "regular", password: str = "Regularpass1!"
):
"""Create a regular (non-admin) user and return their token."""
headers = {"Authorization": f"Bearer {token}"}
@@ -68,7 +68,7 @@ class TestSMTPConfigAPI:
@pytest.fixture
async def admin_token(self, async_client: AsyncClient):
return await _setup_admin(async_client, "smtpadmin", "adminpass123")
return await _setup_admin(async_client, "smtpadmin", "AdminPass1!")
@pytest.mark.asyncio
@pytest.mark.integration
@@ -100,7 +100,7 @@ class TestSMTPConfigAPI:
@pytest.mark.integration
async def test_smtp_settings_requires_admin(self, async_client: AsyncClient, admin_token: str):
"""Non-admin user gets 403 on SMTP endpoints."""
user_token = await _create_regular_user(async_client, admin_token, "smtpregular", "pass123456")
user_token = await _create_regular_user(async_client, admin_token, "smtpregular", "Pass12345!")
headers = {"Authorization": f"Bearer {user_token}"}
response = await async_client.post("/api/v1/auth/smtp", headers=headers, json=SMTP_DATA)
@@ -143,7 +143,7 @@ class TestAdvancedAuthToggleAPI:
@pytest.fixture
async def admin_token(self, async_client: AsyncClient):
return await _setup_admin(async_client, "toggleadmin", "adminpass123")
return await _setup_admin(async_client, "toggleadmin", "AdminPass1!")
@pytest.mark.asyncio
@pytest.mark.integration
@@ -193,7 +193,7 @@ class TestAdvancedAuthToggleAPI:
@pytest.mark.integration
async def test_enable_requires_admin(self, async_client: AsyncClient, admin_token: str):
"""Non-admin user gets 403 on enable/disable."""
user_token = await _create_regular_user(async_client, admin_token, "toggleregular", "pass123456")
user_token = await _create_regular_user(async_client, admin_token, "toggleregular", "Pass12345!")
headers = {"Authorization": f"Bearer {user_token}"}
response = await async_client.post("/api/v1/auth/advanced-auth/enable", headers=headers)
@@ -208,7 +208,7 @@ class TestEmailLoginAPI:
@pytest.fixture
async def admin_token(self, async_client: AsyncClient):
return await _setup_admin(async_client, "emailadmin", "adminpass123")
return await _setup_admin(async_client, "emailadmin", "AdminPass1!")
@pytest.mark.asyncio
@pytest.mark.integration
@@ -233,13 +233,13 @@ class TestEmailLoginAPI:
await async_client.patch(
f"/api/v1/users/{user_id}",
headers=headers,
json={"password": "knownpassword123"},
json={"password": "Knownpassword1!"},
)
# Login with email
response = await async_client.post(
"/api/v1/auth/login",
json={"username": "emailuser@test.com", "password": "knownpassword123"},
json={"username": "emailuser@test.com", "password": "Knownpassword1!"},
)
assert response.status_code == 200
assert "access_token" in response.json()
@@ -262,12 +262,12 @@ class TestEmailLoginAPI:
await async_client.patch(
f"/api/v1/users/{user_id}",
headers=headers,
json={"password": "casepassword123"},
json={"password": "Casepassword1!"},
)
response = await async_client.post(
"/api/v1/auth/login",
json={"username": "CASEUSER@TEST.COM", "password": "casepassword123"},
json={"username": "CASEUSER@TEST.COM", "password": "Casepassword1!"},
)
assert response.status_code == 200
assert "access_token" in response.json()
@@ -282,13 +282,13 @@ class TestEmailLoginAPI:
await async_client.post(
"/api/v1/users/",
headers=headers,
json={"username": "noemail", "password": "noEmailPass1", "email": "noemail@test.com", "role": "user"},
json={"username": "noemail", "password": "NoEmailPass1!", "email": "noemail@test.com", "role": "user"},
)
# Try to login with email — should fail since advanced auth is off
response = await async_client.post(
"/api/v1/auth/login",
json={"username": "noemail@test.com", "password": "noEmailPass1"},
json={"username": "noemail@test.com", "password": "NoEmailPass1!"},
)
assert response.status_code == 401
@@ -310,13 +310,13 @@ class TestEmailLoginAPI:
await async_client.patch(
f"/api/v1/users/{user_id}",
headers=headers,
json={"password": "usernamepass123"},
json={"password": "Usernamepass1!"},
)
# Login with username (not email)
response = await async_client.post(
"/api/v1/auth/login",
json={"username": "usernameuser", "password": "usernamepass123"},
json={"username": "usernameuser", "password": "Usernamepass1!"},
)
assert response.status_code == 200
assert "access_token" in response.json()
@@ -327,7 +327,7 @@ class TestForgotPasswordAPI:
@pytest.fixture
async def admin_token(self, async_client: AsyncClient):
return await _setup_admin(async_client, "forgotadmin", "adminpass123")
return await _setup_admin(async_client, "forgotadmin", "AdminPass1!")
@pytest.mark.asyncio
@pytest.mark.integration
@@ -388,7 +388,13 @@ class TestForgotPasswordAPI:
@pytest.mark.asyncio
@pytest.mark.integration
async def test_forgot_password_changes_password(self, async_client: AsyncClient, admin_token: str):
"""After forgot-password, old password stops working."""
"""After forgot-password + confirm, old password stops working and new one works.
H-6: The flow is now token-based: /forgot-password issues a reset link and
/forgot-password/confirm consumes the token and sets the new password.
"""
from unittest.mock import AsyncMock
headers = {"Authorization": f"Bearer {admin_token}"}
with patch("backend.app.api.routes.users.send_email"):
@@ -403,37 +409,66 @@ class TestForgotPasswordAPI:
await async_client.patch(
f"/api/v1/users/{user_id}",
headers=headers,
json={"password": "originalpass123"},
json={"password": "Originalpass1!"},
)
# Verify login works with original password
login_resp = await async_client.post(
"/api/v1/auth/login",
json={"username": "resetme", "password": "originalpass123"},
json={"username": "resetme", "password": "Originalpass1!"},
)
assert login_resp.status_code == 200
# Trigger forgot password
with patch("backend.app.api.routes.auth.send_email"):
await async_client.post(
# Trigger forgot-password and capture the reset URL (contains the token)
captured: dict[str, str] = {}
async def _capture_link_email(db, username, reset_url):
captured["reset_url"] = reset_url
return ("subject", "body", "<body/>")
with (
patch(
"backend.app.api.routes.auth.create_password_reset_link_email_from_template",
side_effect=_capture_link_email,
),
patch("backend.app.api.routes.auth.send_email"),
):
resp = await async_client.post(
"/api/v1/auth/forgot-password",
json={"email": "resetme@test.com"},
)
assert resp.status_code == 200
assert "reset_url" in captured, "Reset URL not captured — email function was not called"
# Extract the token from the captured URL and confirm the reset
reset_token = captured["reset_url"].split("reset_token=")[1]
confirm_resp = await async_client.post(
"/api/v1/auth/forgot-password/confirm",
json={"token": reset_token, "new_password": "Newpass456!"},
)
assert confirm_resp.status_code == 200
# Old password should no longer work
login_resp = await async_client.post(
"/api/v1/auth/login",
json={"username": "resetme", "password": "originalpass123"},
json={"username": "resetme", "password": "Originalpass1!"},
)
assert login_resp.status_code == 401
# New password must work
login_resp = await async_client.post(
"/api/v1/auth/login",
json={"username": "resetme", "password": "Newpass456!"},
)
assert login_resp.status_code == 200
class TestAdminResetPasswordAPI:
"""Integration tests for admin password reset endpoint."""
@pytest.fixture
async def admin_token(self, async_client: AsyncClient):
return await _setup_admin(async_client, "resetadmin", "adminpass123")
return await _setup_admin(async_client, "resetadmin", "AdminPass1!")
@pytest.mark.asyncio
@pytest.mark.integration
@@ -467,7 +502,7 @@ class TestAdminResetPasswordAPI:
async def test_reset_password_requires_admin(self, async_client: AsyncClient, admin_token: str):
"""Non-admin user gets 403 on reset-password."""
# Create regular user before enabling advanced auth (no email required)
user_token = await _create_regular_user(async_client, admin_token, "resetregular", "pass123456")
user_token = await _create_regular_user(async_client, admin_token, "resetregular", "Pass12345!")
with patch("backend.app.api.routes.users.send_email"):
await _setup_smtp_and_advanced_auth(async_client, admin_token)
@@ -522,7 +557,7 @@ class TestAdminResetPasswordAPI:
create_resp = await async_client.post(
"/api/v1/users/",
headers=headers,
json={"username": "noemailuser", "password": "noemail123456", "role": "user"},
json={"username": "noemailuser", "password": "Noemail12345!", "role": "user"},
)
user_id = create_resp.json()["id"]
@@ -543,7 +578,7 @@ class TestUserCreationAdvancedAuth:
@pytest.fixture
async def admin_token(self, async_client: AsyncClient):
return await _setup_admin(async_client, "createadmin", "adminpass123")
return await _setup_admin(async_client, "createadmin", "AdminPass1!")
@pytest.mark.asyncio
@pytest.mark.integration
@@ -628,3 +663,52 @@ class TestUserCreationAdvancedAuth:
result = response.json()
assert "email" in result
assert result["email"] == "emailresp@test.com"
# ===========================================================================
# M-1: OIDC/LDAP users must not be able to use the password reset flow
# ===========================================================================
class TestAuthSourcePasswordResetBlocking:
"""Forgot-password must silently skip OIDC and LDAP users (M-1)."""
@pytest.fixture
async def admin_token(self, async_client: AsyncClient):
return await _setup_admin(async_client, "authsrcadmin", "AdminPass1!")
@pytest.mark.asyncio
@pytest.mark.integration
async def test_forgot_password_silently_skips_oidc_user(
self, async_client: AsyncClient, admin_token: str, db_session
):
"""forgot-password for an OIDC user returns 200 but does NOT send email."""
from backend.app.core.auth import get_password_hash
from backend.app.models.user import User
headers = {"Authorization": f"Bearer {admin_token}"}
await async_client.post("/api/v1/auth/smtp", headers=headers, json=SMTP_DATA)
await async_client.post("/api/v1/auth/advanced-auth/enable", headers=headers)
# Directly insert an OIDC-sourced user into the DB
oidc_user = User(
username="oidcpwreset",
email="oidcpwreset@test.com",
auth_source="oidc",
password_hash=get_password_hash("irrelevant"),
role="user",
is_active=True,
)
db_session.add(oidc_user)
await db_session.commit()
with patch("backend.app.api.routes.auth.send_email") as mock_send:
response = await async_client.post(
"/api/v1/auth/forgot-password",
json={"email": "oidcpwreset@test.com"},
)
# Anti-enumeration: still returns 200
assert response.status_code == 200
# But no email is sent for OIDC users
mock_send.assert_not_called()
backend/tests/integration/test_auth_api.py
+88 -34
View File
@@ -61,7 +61,7 @@ class TestAuthSetupAPI:
json={
"auth_enabled": True,
"admin_username": "testadmin",
"admin_password": "testpassword123",
"admin_password": "TestPass1!",
},
)
@@ -96,14 +96,14 @@ class TestAuthLoginAPI:
json={
"auth_enabled": True,
"admin_username": "logintest",
"admin_password": "loginpassword123",
"admin_password": "LoginPass1!",
},
)
# Now login
response = await async_client.post(
"/api/v1/auth/login",
json={"username": "logintest", "password": "loginpassword123"},
json={"username": "logintest", "password": "LoginPass1!"},
)
assert response.status_code == 200
@@ -123,7 +123,7 @@ class TestAuthLoginAPI:
json={
"auth_enabled": True,
"admin_username": "invalidtest",
"admin_password": "correctpassword",
"admin_password": "CorrectPass1!",
},
)
@@ -158,13 +158,13 @@ class TestAuthMeAPI:
json={
"auth_enabled": True,
"admin_username": "metest",
"admin_password": "mepassword123",
"admin_password": "MePass1!",
},
)
login_response = await async_client.post(
"/api/v1/auth/login",
json={"username": "metest", "password": "mepassword123"},
json={"username": "metest", "password": "MePass1!"},
)
token = login_response.json()["access_token"]
@@ -254,13 +254,13 @@ class TestUsersAPI:
json={
"auth_enabled": True,
"admin_username": "usersadmin",
"admin_password": "adminpassword123",
"admin_password": "AdminPass1!",
},
)
login_response = await async_client.post(
"/api/v1/auth/login",
json={"username": "usersadmin", "password": "adminpassword123"},
json={"username": "usersadmin", "password": "AdminPass1!"},
)
return login_response.json()["access_token"]
@@ -274,7 +274,7 @@ class TestUsersAPI:
json={
"auth_enabled": True,
"admin_username": "authreqadmin",
"admin_password": "adminpassword123",
"admin_password": "AdminPass1!",
},
)
@@ -306,7 +306,7 @@ class TestUsersAPI:
headers={"Authorization": f"Bearer {auth_token}"},
json={
"username": "newuser",
"password": "newuserpassword",
"password": "Newuserpass1!",
"role": "user",
},
)
@@ -327,7 +327,7 @@ class TestUsersAPI:
headers={"Authorization": f"Bearer {auth_token}"},
json={
"username": "duplicateuser",
"password": "password123",
"password": "Password123!",
"role": "user",
},
)
@@ -338,7 +338,7 @@ class TestUsersAPI:
headers={"Authorization": f"Bearer {auth_token}"},
json={
"username": "duplicateuser",
"password": "password456",
"password": "Password456!",
"role": "user",
},
)
@@ -356,7 +356,7 @@ class TestUsersAPI:
headers={"Authorization": f"Bearer {auth_token}"},
json={
"username": "updateuser",
"password": "password123",
"password": "Password123!",
"role": "user",
},
)
@@ -382,7 +382,7 @@ class TestUsersAPI:
headers={"Authorization": f"Bearer {auth_token}"},
json={
"username": "deleteuser",
"password": "password123",
"password": "Password123!",
"role": "user",
},
)
@@ -410,14 +410,14 @@ class TestAuthDisableAPI:
json={
"auth_enabled": True,
"admin_username": "disableadmin",
"admin_password": "adminpassword123",
"admin_password": "AdminPass1!",
},
)
# Login to get token
login_response = await async_client.post(
"/api/v1/auth/login",
json={"username": "disableadmin", "password": "adminpassword123"},
json={"username": "disableadmin", "password": "AdminPass1!"},
)
token = login_response.json()["access_token"]
@@ -446,13 +446,13 @@ class TestGroupsAPI:
json={
"auth_enabled": True,
"admin_username": "groupsadmin",
"admin_password": "adminpassword123",
"admin_password": "AdminPass1!",
},
)
login_response = await async_client.post(
"/api/v1/auth/login",
json={"username": "groupsadmin", "password": "adminpassword123"},
json={"username": "groupsadmin", "password": "AdminPass1!"},
)
return login_response.json()["access_token"]
@@ -592,13 +592,13 @@ class TestUserGroupsAPI:
json={
"auth_enabled": True,
"admin_username": "usergroupadmin",
"admin_password": "adminpassword123",
"admin_password": "AdminPass1!",
},
)
login_response = await async_client.post(
"/api/v1/auth/login",
json={"username": "usergroupadmin", "password": "adminpassword123"},
json={"username": "usergroupadmin", "password": "AdminPass1!"},
)
return login_response.json()["access_token"]
@@ -619,7 +619,7 @@ class TestUserGroupsAPI:
headers={"Authorization": f"Bearer {auth_token}"},
json={
"username": "groupuser",
"password": "password123",
"password": "Password123!",
"group_ids": [operators_group["id"]],
},
)
@@ -636,7 +636,7 @@ class TestUserGroupsAPI:
user_response = await async_client.post(
"/api/v1/users/",
headers={"Authorization": f"Bearer {auth_token}"},
json={"username": "addtogroup", "password": "password123"},
json={"username": "addtogroup", "password": "Password123!"},
)
user_id = user_response.json()["id"]
@@ -675,13 +675,13 @@ class TestChangePasswordAPI:
json={
"auth_enabled": True,
"admin_username": "pwchangeadmin",
"admin_password": "adminpassword123",
"admin_password": "AdminPass1!",
},
)
admin_login = await async_client.post(
"/api/v1/auth/login",
json={"username": "pwchangeadmin", "password": "adminpassword123"},
json={"username": "pwchangeadmin", "password": "AdminPass1!"},
)
admin_token = admin_login.json()["access_token"]
@@ -689,13 +689,13 @@ class TestChangePasswordAPI:
await async_client.post(
"/api/v1/users/",
headers={"Authorization": f"Bearer {admin_token}"},
json={"username": "pwchangeuser", "password": "oldpassword123"},
json={"username": "pwchangeuser", "password": "Oldpassword123!"},
)
# Login as regular user
user_login = await async_client.post(
"/api/v1/auth/login",
json={"username": "pwchangeuser", "password": "oldpassword123"},
json={"username": "pwchangeuser", "password": "Oldpassword123!"},
)
return user_login.json()["access_token"]
@@ -707,8 +707,8 @@ class TestChangePasswordAPI:
"/api/v1/users/me/change-password",
headers={"Authorization": f"Bearer {user_token}"},
json={
"current_password": "oldpassword123",
"new_password": "newpassword456",
"current_password": "Oldpassword123!",
"new_password": "Newpassword456!",
},
)
@@ -718,7 +718,7 @@ class TestChangePasswordAPI:
# Verify can login with new password
login_response = await async_client.post(
"/api/v1/auth/login",
json={"username": "pwchangeuser", "password": "newpassword456"},
json={"username": "pwchangeuser", "password": "Newpassword456!"},
)
assert login_response.status_code == 200
@@ -731,7 +731,7 @@ class TestChangePasswordAPI:
headers={"Authorization": f"Bearer {user_token}"},
json={
"current_password": "wrongpassword",
"new_password": "newpassword456",
"new_password": "Newpassword456!",
},
)
@@ -746,7 +746,7 @@ class TestChangePasswordAPI:
"/api/v1/users/me/change-password",
json={
"current_password": "oldpassword",
"new_password": "newpassword",
"new_password": "Strongpass456!",
},
)
@@ -768,7 +768,7 @@ class TestAuthMiddlewarePublicRoutes:
json={
"auth_enabled": True,
"admin_username": "middlewareadmin",
"admin_password": "adminpassword123",
"admin_password": "AdminPass1!",
},
)
@@ -786,7 +786,7 @@ class TestAuthMiddlewarePublicRoutes:
"""Verify /api/v1/auth/login is accessible without auth."""
response = await async_client.post(
"/api/v1/auth/login",
json={"username": "middlewareadmin", "password": "adminpassword123"},
json={"username": "middlewareadmin", "password": "AdminPass1!"},
)
# Should not return 401 (unauthorized) - it should either succeed or return
# a different error (like 400 for wrong credentials)
@@ -826,7 +826,7 @@ class TestAuthMiddlewarePublicRoutes:
# Login to get token
login_response = await async_client.post(
"/api/v1/auth/login",
json={"username": "middlewareadmin", "password": "adminpassword123"},
json={"username": "middlewareadmin", "password": "AdminPass1!"},
)
token = login_response.json()["access_token"]
@@ -863,3 +863,57 @@ class TestAuthMiddlewarePublicRoutes:
# Will likely be 400 (advanced auth not enabled) but that's okay -
# the important thing is it's not blocked by auth middleware
assert response.status_code in [200, 400]
# ===========================================================================
# H-1: Input length validation
# ===========================================================================
class TestInputLengthValidation:
"""LoginRequest and SetupRequest must reject oversized inputs (H-1)."""
@pytest.mark.asyncio
@pytest.mark.integration
async def test_login_password_too_long_rejected(self, async_client: AsyncClient):
"""Password exceeding 256 characters must be rejected with 422."""
response = await async_client.post(
"/api/v1/auth/login",
json={"username": "admin", "password": "x" * 257},
)
assert response.status_code == 422
@pytest.mark.asyncio
@pytest.mark.integration
async def test_login_username_too_long_rejected(self, async_client: AsyncClient):
"""Username exceeding 150 characters must be rejected with 422."""
response = await async_client.post(
"/api/v1/auth/login",
json={"username": "u" * 151, "password": "password"},
)
assert response.status_code == 422
@pytest.mark.asyncio
@pytest.mark.integration
async def test_setup_password_too_long_rejected(self, async_client: AsyncClient):
"""SetupRequest admin_password exceeding 256 characters must be rejected with 422."""
response = await async_client.post(
"/api/v1/auth/setup",
json={
"auth_enabled": True,
"admin_username": "admin",
"admin_password": "x" * 257,
},
)
assert response.status_code == 422
@pytest.mark.asyncio
@pytest.mark.integration
async def test_login_password_at_limit_accepted(self, async_client: AsyncClient):
"""Password of exactly 256 characters must pass schema validation (may fail auth)."""
response = await async_client.post(
"/api/v1/auth/login",
json={"username": "admin", "password": "x" * 256},
)
# Schema accepts it; auth may reject with 401 (auth disabled) or 400
assert response.status_code != 422
backend/tests/integration/test_client_ip.py
+130
View File
@@ -0,0 +1,130 @@
"""Unit tests for _get_client_ip (M-R9-A / M-R10-A).
Covers:
- Direct connection without TRUSTED_PROXY_IPS → returns client.host
- Trusted proxy with XFF → walks right-to-left, returns first non-proxy IP
- Spoofed XFF from an untrusted client → client.host is returned
- Multiple trusted proxies in chain → returns leftmost non-proxy entry
- All XFF entries are trusted proxies → falls back to leftmost
- Empty XFF header with trusted proxy → returns direct_ip
- No client (client=None) → returns unique per-request token
"""
from __future__ import annotations
from unittest.mock import MagicMock, patch
def _make_request(client_host: str | None, xff: str = "") -> MagicMock:
"""Create a minimal mock Request with given client.host and X-Forwarded-For."""
req = MagicMock()
if client_host is None:
req.client = None
else:
req.client = MagicMock()
req.client.host = client_host
req.headers = MagicMock()
req.headers.get = lambda key, default="": xff if key == "X-Forwarded-For" else default
return req
def _call(request, trusted: frozenset[str]) -> str:
from backend.app.api.routes.auth import _get_client_ip
with patch("backend.app.api.routes.auth._TRUSTED_PROXY_IPS", trusted):
return _get_client_ip(request)
# ---------------------------------------------------------------------------
# No proxy configured (TRUSTED_PROXY_IPS empty)
# ---------------------------------------------------------------------------
def test_no_proxy_returns_client_host():
req = _make_request("1.2.3.4")
assert _call(req, frozenset()) == "1.2.3.4"
def test_no_proxy_xff_ignored():
"""XFF must be ignored when TRUSTED_PROXY_IPS is not set."""
req = _make_request("1.2.3.4", xff="9.9.9.9")
assert _call(req, frozenset()) == "1.2.3.4"
# ---------------------------------------------------------------------------
# Trusted proxy present; direct peer is the proxy
# ---------------------------------------------------------------------------
def test_trusted_proxy_returns_rightmost_non_proxy():
"""Single proxy: XFF = client_ip; direct_ip = proxy_ip → return client."""
proxy = "10.0.0.1"
client = "203.0.113.5"
req = _make_request(proxy, xff=client)
assert _call(req, frozenset({proxy})) == client
def test_trusted_proxy_chain_skips_proxy_ips():
"""Multi-hop: client → proxy1 → proxy2 (direct) → app.
XFF = 'client, proxy1'; direct = proxy2. Should return client."""
proxy1 = "10.0.0.1"
proxy2 = "10.0.0.2"
client = "198.51.100.7"
req = _make_request(proxy2, xff=f"{client}, {proxy1}")
assert _call(req, frozenset({proxy1, proxy2})) == client
def test_all_xff_entries_are_proxies_falls_back_to_leftmost():
"""When every XFF entry is a trusted proxy, return the leftmost (original) entry."""
proxy1 = "10.0.0.1"
proxy2 = "10.0.0.2"
req = _make_request(proxy2, xff=f"{proxy1}, {proxy2}")
assert _call(req, frozenset({proxy1, proxy2})) == proxy1
def test_empty_xff_with_trusted_proxy_returns_direct_ip():
"""Trusted proxy but no XFF header → fall through to direct_ip."""
proxy = "10.0.0.1"
req = _make_request(proxy, xff="")
assert _call(req, frozenset({proxy})) == proxy
# ---------------------------------------------------------------------------
# Spoofed XFF from an untrusted client
# ---------------------------------------------------------------------------
def test_spoofed_xff_from_untrusted_client_ignored():
"""Client not in TRUSTED_PROXY_IPS → XFF is ignored; client.host returned."""
untrusted_client = "203.0.113.99"
req = _make_request(untrusted_client, xff="1.1.1.1")
assert _call(req, frozenset({"10.0.0.1"})) == untrusted_client
# ---------------------------------------------------------------------------
# No client (transport layer provides no address)
# ---------------------------------------------------------------------------
def test_no_client_returns_unique_token():
"""When request.client is None, each call returns a unique rate-limit sentinel."""
req1 = _make_request(None)
req2 = _make_request(None)
ip1 = _call(req1, frozenset())
ip2 = _call(req2, frozenset())
assert ip1.startswith("__no_ip_")
assert ip2.startswith("__no_ip_")
assert ip1 != ip2, "Each missing-client request must get a distinct sentinel"
# ---------------------------------------------------------------------------
# Whitespace in XFF values
# ---------------------------------------------------------------------------
def test_xff_with_extra_whitespace_trimmed():
"""IPs in XFF with leading/trailing spaces are handled correctly."""
proxy = "10.0.0.1"
client = "192.0.2.33"
req = _make_request(proxy, xff=f" {client} , {proxy} ")
assert _call(req, frozenset({proxy})) == client
backend/tests/integration/test_mfa_api.py
+3016
View File
File diff suppressed because it is too large Load Diff
backend/tests/integration/test_ownership_permissions.py
+9 -9
View File
@@ -22,14 +22,14 @@ class TestOwnershipPermissionsSetup:
json={
"auth_enabled": True,
"admin_username": "ownershipadmin",
"admin_password": "adminpassword123",
"admin_password": "AdminPass1!",
},
)
# Login as admin
admin_login = await async_client.post(
"/api/v1/auth/login",
json={"username": "ownershipadmin", "password": "adminpassword123"},
json={"username": "ownershipadmin", "password": "AdminPass1!"},
)
admin_token = admin_login.json()["access_token"]
admin_user = admin_login.json()["user"]
@@ -49,7 +49,7 @@ class TestOwnershipPermissionsSetup:
headers={"Authorization": f"Bearer {admin_token}"},
json={
"username": "operator1",
"password": "operatorpass123",
"password": "Operatorpass1!",
"group_ids": [operators_group["id"]],
},
)
@@ -58,7 +58,7 @@ class TestOwnershipPermissionsSetup:
# Login as operator
operator_login = await async_client.post(
"/api/v1/auth/login",
json={"username": "operator1", "password": "operatorpass123"},
json={"username": "operator1", "password": "Operatorpass1!"},
)
operator_token = operator_login.json()["access_token"]
@@ -68,7 +68,7 @@ class TestOwnershipPermissionsSetup:
headers={"Authorization": f"Bearer {admin_token}"},
json={
"username": "operator2",
"password": "operatorpass123",
"password": "Operatorpass1!",
"group_ids": [operators_group["id"]],
},
)
@@ -76,7 +76,7 @@ class TestOwnershipPermissionsSetup:
operator2_login = await async_client.post(
"/api/v1/auth/login",
json={"username": "operator2", "password": "operatorpass123"},
json={"username": "operator2", "password": "Operatorpass1!"},
)
operator2_token = operator2_login.json()["access_token"]
@@ -86,14 +86,14 @@ class TestOwnershipPermissionsSetup:
headers={"Authorization": f"Bearer {admin_token}"},
json={
"username": "viewer1",
"password": "viewerpass123",
"password": "Viewerpass1!",
"group_ids": [viewers_group["id"]],
},
)
viewer_login = await async_client.post(
"/api/v1/auth/login",
json={"username": "viewer1", "password": "viewerpass123"},
json={"username": "viewer1", "password": "Viewerpass1!"},
)
viewer_token = viewer_login.json()["access_token"]
@@ -721,7 +721,7 @@ class TestUserItemsCountAndDeletion(TestOwnershipPermissionsSetup):
headers={"Authorization": f"Bearer {auth_setup['admin_token']}"},
json={
"username": "deletewithitems",
"password": "password123",
"password": "Password123!",
},
)
user_id = create_response.json()["id"]
backend/tests/integration/test_security.py
+796
View File
@@ -0,0 +1,796 @@
"""Security tests for the 8 coverage gaps identified in the maintainer review.
Gap 1: encryption.py has zero tests
Gap 2: JWT revocation (revoke_jti, is_jti_revoked, _is_token_fresh) untested
Gap 3: OIDC exchange token replay untested
Gap 4: OIDC email_verified claim handling untested
Gap 5: Email OTP max-attempts invalidation untested
Gap 6: OIDC callback error redirects (SSRF protection) undertested
Gap 7: Login rate limiting untested
Gap 8: challenge_id cookie binding untested
"""
from __future__ import annotations
import base64
import secrets
import time
from datetime import datetime, timedelta, timezone
from unittest.mock import AsyncMock, MagicMock, patch
import jwt as pyjwt
import pytest
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import rsa
from httpx import AsyncClient
from sqlalchemy.ext.asyncio import AsyncSession
from backend.app.models.auth_ephemeral import AuthEphemeralToken
from backend.app.models.user import User
AUTH_SETUP_URL = "/api/v1/auth/setup"
LOGIN_URL = "/api/v1/auth/login"
LOGOUT_URL = "/api/v1/auth/logout"
ME_URL = "/api/v1/auth/me"
def _auth_header(token: str) -> dict[str, str]:
return {"Authorization": f"Bearer {token}"}
def _norm_pw(password: str) -> str:
"""Ensure password meets complexity requirements (I4: SetupRequest now validates)."""
if not any(c.isupper() for c in password):
password = password[0].upper() + password[1:]
if not any(c not in "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789" for c in password):
password = password + "!"
return password
async def _setup_and_login(client: AsyncClient, username: str, password: str) -> str:
password = _norm_pw(password)
await client.post(
AUTH_SETUP_URL,
json={"auth_enabled": True, "admin_username": username, "admin_password": password},
)
resp = await client.post(LOGIN_URL, json={"username": username, "password": password})
assert resp.status_code == 200
return resp.json()["access_token"]
def _make_test_rsa_key():
def _b64url(n: int, length: int) -> str:
return base64.urlsafe_b64encode(n.to_bytes(length, "big")).rstrip(b"=").decode()
private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
private_pem = private_key.private_bytes(
serialization.Encoding.PEM,
serialization.PrivateFormat.TraditionalOpenSSL,
serialization.NoEncryption(),
)
pub_numbers = private_key.public_key().public_numbers()
jwks = {
"keys": [
{
"kty": "RSA",
"use": "sig",
"alg": "RS256",
"kid": "test-kid-1",
"n": _b64url(pub_numbers.n, 256),
"e": _b64url(pub_numbers.e, 3),
}
]
}
return private_pem, jwks
# ===========================================================================
# Gap 1: encryption.py unit tests
# ===========================================================================
class TestEncryption:
"""encrypt/decrypt round-trips, plaintext passthrough, RuntimeError on missing key."""
def test_encrypt_decrypt_roundtrip_with_key(self):
from cryptography.fernet import Fernet
test_key = Fernet.generate_key().decode()
import backend.app.core.encryption as enc_mod
original = enc_mod._fernet_instance
original_warn = enc_mod._warn_shown
try:
enc_mod._fernet_instance = None
enc_mod._warn_shown = False
with patch.dict("os.environ", {"MFA_ENCRYPTION_KEY": test_key}):
ciphertext = enc_mod.mfa_encrypt("my-totp-secret")
assert ciphertext.startswith("fernet:")
assert enc_mod.mfa_decrypt(ciphertext) == "my-totp-secret"
finally:
enc_mod._fernet_instance = original
enc_mod._warn_shown = original_warn
def test_plaintext_passthrough_without_key(self):
import backend.app.core.encryption as enc_mod
original = enc_mod._fernet_instance
original_warn = enc_mod._warn_shown
try:
enc_mod._fernet_instance = None
enc_mod._warn_shown = False
with patch.dict("os.environ", {}, clear=True):
env = {k: v for k, v in __import__("os").environ.items() if k != "MFA_ENCRYPTION_KEY"}
with patch.dict("os.environ", env, clear=True):
result = enc_mod.mfa_encrypt("plaintext-secret")
assert result == "plaintext-secret"
assert enc_mod.mfa_decrypt("plaintext-secret") == "plaintext-secret"
finally:
enc_mod._fernet_instance = original
enc_mod._warn_shown = original_warn
def test_decrypt_raises_runtime_error_without_key_for_encrypted_value(self):
import backend.app.core.encryption as enc_mod
original = enc_mod._fernet_instance
original_warn = enc_mod._warn_shown
try:
enc_mod._fernet_instance = None
enc_mod._warn_shown = False
# A value with the fernet: prefix but no key configured
env = {k: v for k, v in __import__("os").environ.items() if k != "MFA_ENCRYPTION_KEY"}
with (
patch.dict("os.environ", env, clear=True),
pytest.raises(RuntimeError, match="MFA_ENCRYPTION_KEY must be set"),
):
enc_mod.mfa_decrypt("fernet:gAAAAA-fake-ciphertext")
finally:
enc_mod._fernet_instance = original
enc_mod._warn_shown = original_warn
# ===========================================================================
# Gap 2: JWT revocation — revoke_jti, is_jti_revoked, _is_token_fresh, /me
# ===========================================================================
class TestJWTRevocation:
"""JWT revocation and token freshness checks."""
@pytest.mark.asyncio
@pytest.mark.integration
async def test_revoke_jti_and_is_jti_revoked(self, async_client: AsyncClient, db_session: AsyncSession):
"""revoke_jti stores the JTI; is_jti_revoked returns True afterwards."""
from backend.app.core.auth import is_jti_revoked, revoke_jti
test_jti = secrets.token_urlsafe(16)
expires = datetime.now(timezone.utc) + timedelta(hours=1)
assert not await is_jti_revoked(test_jti)
await revoke_jti(test_jti, expires, username="testuser")
assert await is_jti_revoked(test_jti)
@pytest.mark.asyncio
@pytest.mark.integration
async def test_revoke_jti_idempotent(self, async_client: AsyncClient):
"""Double-revocation of the same JTI should not raise."""
from backend.app.core.auth import is_jti_revoked, revoke_jti
jti = secrets.token_urlsafe(16)
expires = datetime.now(timezone.utc) + timedelta(hours=1)
await revoke_jti(jti, expires)
await revoke_jti(jti, expires) # must not raise
assert await is_jti_revoked(jti)
def test_is_token_fresh_rejects_none_iat(self):
"""_is_token_fresh returns False when iat is None (I1 hard cutoff)."""
from backend.app.core.auth import _is_token_fresh
user = MagicMock()
user.password_changed_at = None
assert _is_token_fresh(None, user) is False
def test_is_token_fresh_rejects_token_before_password_change(self):
"""_is_token_fresh returns False when iat predates password_changed_at."""
from backend.app.core.auth import _is_token_fresh
now = datetime.now(timezone.utc)
user = MagicMock()
user.password_changed_at = now
old_iat = (now - timedelta(hours=1)).timestamp()
assert _is_token_fresh(old_iat, user) is False
def test_is_token_fresh_accepts_token_after_password_change(self):
"""_is_token_fresh returns True when iat is after password_changed_at."""
from backend.app.core.auth import _is_token_fresh
now = datetime.now(timezone.utc)
user = MagicMock()
user.password_changed_at = now - timedelta(hours=1)
recent_iat = now.timestamp()
assert _is_token_fresh(recent_iat, user) is True
def test_is_token_fresh_returns_true_when_no_password_change(self):
"""_is_token_fresh returns True when password_changed_at is None (I2 migration not yet run)."""
from backend.app.core.auth import _is_token_fresh
user = MagicMock()
user.password_changed_at = None
assert _is_token_fresh(time.time(), user) is True
@pytest.mark.asyncio
@pytest.mark.integration
async def test_me_endpoint_rejects_token_after_logout(self, async_client: AsyncClient):
"""After logout, the bearer token must be rejected by /me (B1 + revocation)."""
token = await _setup_and_login(async_client, "sec_logout_me", "sec_logout_me1")
# Token works before logout
me_resp = await async_client.get(ME_URL, headers=_auth_header(token))
assert me_resp.status_code == 200
# Logout
logout_resp = await async_client.post(LOGOUT_URL, headers=_auth_header(token))
assert logout_resp.status_code == 200
# Token must now be rejected
me_after = await async_client.get(ME_URL, headers=_auth_header(token))
assert me_after.status_code == 401
# ===========================================================================
# Gap 3: OIDC exchange token replay
# ===========================================================================
class TestOIDCExchangeReplay:
"""A single-use OIDC exchange token cannot be redeemed twice."""
@pytest.mark.asyncio
@pytest.mark.integration
async def test_exchange_token_is_single_use(self, async_client: AsyncClient, db_session: AsyncSession):
"""The second call to /oidc/exchange with the same token returns 401."""
exchange_token = secrets.token_urlsafe(32)
db_session.add(
AuthEphemeralToken(
token=exchange_token,
token_type="oidc_exchange",
username="oidc_replay_user",
expires_at=datetime.now(timezone.utc) + timedelta(minutes=5),
)
)
await db_session.commit()
# Seed the user so the exchange can resolve it
from backend.app.core.auth import get_password_hash
from backend.app.core.database import async_session, seed_default_groups
async with async_session() as db:
result = await db.execute(__import__("sqlalchemy").select(User).where(User.username == "oidc_replay_user"))
if result.scalar_one_or_none() is None:
db.add(
User(
username="oidc_replay_user",
password_hash=get_password_hash("pw"),
is_active=True,
)
)
await db.commit()
first = await async_client.post("/api/v1/auth/oidc/exchange", json={"oidc_token": exchange_token})
assert first.status_code == 200
second = await async_client.post("/api/v1/auth/oidc/exchange", json={"oidc_token": exchange_token})
assert second.status_code == 401
# ===========================================================================
# Gap 4: OIDC email_verified claim handling
# ===========================================================================
class TestOIDCEmailVerified:
"""email_verified: False/absent must not link OIDC identity to an existing email."""
@pytest.mark.asyncio
@pytest.mark.integration
async def test_unverified_email_does_not_link_to_existing_user(
self, async_client: AsyncClient, db_session: AsyncSession
):
"""If email_verified is False, the OIDC callback must not auto-link by email."""
private_pem, jwks_data = _make_test_rsa_key()
issuer = "https://idp.evtest.example.com"
client_id = "ev-client"
nonce = secrets.token_urlsafe(16)
now = int(time.time())
id_token = pyjwt.encode(
{
"sub": "ev-sub-new",
"iss": issuer,
"aud": client_id,
"nonce": nonce,
"email": "existing@example.com",
"email_verified": False, # <-- must be ignored
"iat": now,
"exp": now + 300,
},
private_pem,
algorithm="RS256",
headers={"kid": "test-kid-1"},
)
admin_token = await _setup_and_login(async_client, "ev_admin", "ev_admin1")
# Create existing user with the same email (use strong password for validator)
create_user_resp = await async_client.post(
"/api/v1/users",
json={"username": "existing_email_user", "password": "Str0ng!Pass", "email": "existing@example.com"},
headers=_auth_header(admin_token),
)
assert create_user_resp.status_code in (200, 201), create_user_resp.json()
# Create OIDC provider
create_resp = await async_client.post(
"/api/v1/auth/oidc/providers",
json={
"name": "EV-IdP",
"issuer_url": issuer,
"client_id": client_id,
"client_secret": "secret",
"scopes": "openid email",
"is_enabled": True,
"auto_create_users": True,
},
headers=_auth_header(admin_token),
)
assert create_resp.status_code == 201
provider_id = create_resp.json()["id"]
state = secrets.token_urlsafe(32)
code_verifier = secrets.token_urlsafe(48)
db_session.add(
AuthEphemeralToken(
token=state,
token_type="oidc_state",
provider_id=provider_id,
nonce=nonce,
code_verifier=code_verifier,
expires_at=datetime.now(timezone.utc) + timedelta(minutes=5),
)
)
await db_session.commit()
discovery_doc = {
"issuer": issuer,
"authorization_endpoint": f"{issuer}/auth",
"token_endpoint": f"{issuer}/token",
"jwks_uri": f"{issuer}/.well-known/jwks.json",
}
class _MockResp:
def __init__(self, data):
self._data = data
self.status_code = 200
self.is_success = True
self.text = str(data)
def json(self):
return self._data
def raise_for_status(self):
pass
class _MockHttpxClientEV:
def __init__(self, *args, **kwargs):
pass
async def __aenter__(self):
return self
async def __aexit__(self, *_):
pass
async def get(self, url, **kwargs):
if "jwks" in url:
return _MockResp(jwks_data)
return _MockResp(discovery_doc)
async def post(self, url, **kwargs):
return _MockResp({"access_token": "mock", "token_type": "Bearer", "id_token": id_token})
with patch("backend.app.api.routes.mfa.httpx.AsyncClient", _MockHttpxClientEV):
await async_client.get(
f"/api/v1/auth/oidc/callback?code=test-code&state={state}",
follow_redirects=False,
)
# Callback must NOT link to the existing_email_user — a new user is created
# instead (because the email claim was ignored due to email_verified=False).
# Either a new user is provisioned (redirect with oidc_token) or the callback
# fails. In either case, the existing user must not have an OIDC link.
from sqlalchemy import select as sa_select
from backend.app.models.oidc_provider import UserOIDCLink
link_result = await db_session.execute(
sa_select(UserOIDCLink)
.join(User, UserOIDCLink.user_id == User.id)
.where(User.email == "existing@example.com")
)
link = link_result.scalar_one_or_none()
assert link is None, "Existing user must not be auto-linked when email_verified is False"
# ===========================================================================
# Gap 5: Email OTP max-attempts invalidation
# ===========================================================================
class TestEmailOTPMaxAttempts:
"""After MAX_ATTEMPTS wrong codes, the OTP is permanently invalidated."""
@pytest.mark.asyncio
@pytest.mark.integration
async def test_email_otp_invalidated_after_max_attempts(self, async_client: AsyncClient, db_session: AsyncSession):
from passlib.context import CryptContext
from sqlalchemy import select as sa_select
from backend.app.models.user_otp_code import UserOTPCode
_pwd_ctx = CryptContext(schemes=["pbkdf2_sha256"], deprecated="auto")
admin_token = await _setup_and_login(async_client, "otp_max_admin", "otp_max_admin1")
# Enable email OTP for admin user
result = await db_session.execute(sa_select(User).where(User.username == "otp_max_admin"))
user = result.scalar_one()
user.email = "otpmax@example.com"
await db_session.commit()
setup_code = "123456"
from backend.app.models.auth_ephemeral import AuthEphemeralToken as AET
setup_token = secrets.token_urlsafe(32)
db_session.add(
AET(
token=setup_token,
token_type="email_otp_setup",
username="otp_max_admin",
nonce=_pwd_ctx.hash(setup_code),
expires_at=datetime.now(timezone.utc) + timedelta(minutes=10),
)
)
await db_session.commit()
await async_client.post(
"/api/v1/auth/2fa/email/enable/confirm",
json={"setup_token": setup_token, "code": setup_code},
headers=_auth_header(admin_token),
)
# Login to get pre_auth_token
login_resp = await async_client.post(
LOGIN_URL, json={"username": "otp_max_admin", "password": "Otp_max_admin1"}
)
pre_auth_token = login_resp.json()["pre_auth_token"]
# Insert an OTP record directly (bypassing SMTP)
real_code = "654321"
otp = UserOTPCode(
user_id=user.id,
code_hash=_pwd_ctx.hash(real_code),
attempts=0,
used=False,
expires_at=datetime.now(timezone.utc) + timedelta(minutes=10),
)
db_session.add(otp)
await db_session.commit()
# Submit MAX_ATTEMPTS wrong codes
from backend.app.api.routes.mfa import MAX_2FA_ATTEMPTS
for _ in range(MAX_2FA_ATTEMPTS):
r = await async_client.post(
"/api/v1/auth/2fa/verify",
json={"pre_auth_token": pre_auth_token, "code": "000000", "method": "email"},
)
# Each attempt must fail with 401
assert r.status_code == 401
# After max attempts, the correct code is also rejected (either OTP
# invalidated → 401, or rate limit hit → 429). Either means locked out.
final = await async_client.post(
"/api/v1/auth/2fa/verify",
json={"pre_auth_token": pre_auth_token, "code": real_code, "method": "email"},
)
assert final.status_code in (401, 429), f"Expected lockout, got {final.status_code}: {final.json()}"
# ===========================================================================
# Gap 6: OIDC callback SSRF protection — invalid authorization_endpoint scheme
# ===========================================================================
class TestOIDCSSRFProtection:
"""authorization_endpoint with non-http(s) scheme must be rejected."""
@pytest.mark.asyncio
@pytest.mark.integration
async def test_invalid_authorization_endpoint_scheme_rejected(
self, async_client: AsyncClient, db_session: AsyncSession
):
issuer = "https://idp.ssrf.example.com"
client_id = "ssrf-client"
admin_token = await _setup_and_login(async_client, "ssrf_admin", "ssrf_admin1")
create_resp = await async_client.post(
"/api/v1/auth/oidc/providers",
json={
"name": "SSRF-IdP",
"issuer_url": issuer,
"client_id": client_id,
"client_secret": "secret",
"scopes": "openid",
"is_enabled": True,
"auto_create_users": False,
},
headers=_auth_header(admin_token),
)
assert create_resp.status_code == 201
provider_id = create_resp.json()["id"]
# Discovery doc returns a javascript: authorization_endpoint
malicious_discovery = {
"issuer": issuer,
"authorization_endpoint": "javascript:alert(1)", # <-- malicious
"token_endpoint": f"{issuer}/token",
"jwks_uri": f"{issuer}/.well-known/jwks.json",
}
class _MockResp:
def __init__(self, data):
self._data = data
self.status_code = 200
self.is_success = True
self.text = str(data)
def json(self):
return self._data
def raise_for_status(self):
pass
class _MockHttpxClientSSRF:
def __init__(self, *args, **kwargs):
pass
async def __aenter__(self):
return self
async def __aexit__(self, *_):
pass
async def get(self, url, **kwargs):
return _MockResp(malicious_discovery)
async def post(self, url, **kwargs):
return _MockResp({})
with patch("backend.app.api.routes.mfa.httpx.AsyncClient", _MockHttpxClientSSRF):
# oidc_authorize uses a path parameter, not query param
authorize_resp = await async_client.get(
f"/api/v1/auth/oidc/authorize/{provider_id}",
follow_redirects=False,
)
# Must be rejected with 502 — B2 guard rejects invalid authorization_endpoint scheme
assert authorize_resp.status_code == 502, authorize_resp.json()
detail = authorize_resp.json().get("detail", "").lower()
assert "authorization_endpoint" in detail or "invalid" in detail
# ===========================================================================
# Gap 7: Login rate limiting
# ===========================================================================
class TestLoginRateLimiting:
"""10+ failed logins for the same username must return 429."""
@pytest.mark.asyncio
@pytest.mark.integration
async def test_excessive_failed_logins_return_429(self, async_client: AsyncClient):
from backend.app.api.routes.mfa import MAX_LOGIN_ATTEMPTS
# Setup auth but do NOT log in
await async_client.post(
AUTH_SETUP_URL,
json={"auth_enabled": True, "admin_username": "ratelimit_user", "admin_password": "Ratelimit_pw1"},
)
status_codes = []
for _ in range(MAX_LOGIN_ATTEMPTS + 2):
resp = await async_client.post(
LOGIN_URL,
json={"username": "ratelimit_user", "password": "wrong_password"},
)
status_codes.append(resp.status_code)
# The last attempts must be 429 (Too Many Requests)
assert status_codes[-1] == 429, f"Expected 429 after {MAX_LOGIN_ATTEMPTS} failures, got: {status_codes}"
# ===========================================================================
# Gap 8: challenge_id cookie binding
# ===========================================================================
class TestChallengeIdCookieBinding:
"""A pre-auth token stolen from session A cannot be used from session B."""
@pytest.mark.asyncio
@pytest.mark.integration
async def test_pre_auth_token_rejected_without_matching_cookie(
self, async_client: AsyncClient, db_session: AsyncSession
):
import pyotp
from passlib.context import CryptContext
_pwd_ctx = CryptContext(schemes=["pbkdf2_sha256"], deprecated="auto")
# Set up user with TOTP
await _setup_and_login(async_client, "cookie_bind_user", "cookie_bind_pw1")
secret = pyotp.random_base32()
totp_obj = pyotp.TOTP(secret)
from sqlalchemy import select as sa_select
from backend.app.models.user_totp import UserTOTP
result = await db_session.execute(sa_select(User).where(User.username == "cookie_bind_user"))
user = result.scalar_one()
db_session.add(UserTOTP(user_id=user.id, secret=secret, is_enabled=True))
await db_session.commit()
# Login from "session A" — gets a pre_auth_token and a 2fa_challenge cookie
login_resp = await async_client.post(
LOGIN_URL, json={"username": "cookie_bind_user", "password": "Cookie_bind_pw1"}
)
assert login_resp.status_code == 200
assert login_resp.json()["requires_2fa"] is True
pre_auth_token = login_resp.json()["pre_auth_token"]
# The async_client jar now holds the 2fa_challenge cookie for session A
# Simulate session B by creating a new client WITHOUT the cookie
from httpx import ASGITransport, AsyncClient as FreshClient
from backend.app.main import app
async with FreshClient(transport=ASGITransport(app=app), base_url="http://test") as session_b:
# Attempt to use session A's pre_auth_token from session B (no cookie)
verify_resp = await session_b.post(
"/api/v1/auth/2fa/verify",
json={
"pre_auth_token": pre_auth_token,
"code": totp_obj.now(),
"method": "totp",
},
)
# Must be rejected — pre_auth_token is bound to session A's cookie
assert verify_resp.status_code == 401, (
f"Expected 401 for token replay from cookieless session, got {verify_resp.status_code}: "
f"{verify_resp.json()}"
)
# ===========================================================================
# C2: Security-header middleware
# ===========================================================================
class TestSecurityHeaders:
"""Every HTTP response must include standard security headers (C2)."""
@pytest.mark.asyncio
@pytest.mark.integration
async def test_security_headers_present(self, async_client: AsyncClient):
"""GET /api/v1/auth/me (unauthenticated → 401) still carries security headers."""
resp = await async_client.get(ME_URL)
assert resp.status_code == 401 # sanity — no auth token
assert resp.headers.get("x-content-type-options") == "nosniff"
assert resp.headers.get("x-frame-options") == "SAMEORIGIN"
assert resp.headers.get("referrer-policy") == "strict-origin-when-cross-origin"
csp = resp.headers.get("content-security-policy", "")
assert "default-src 'self'" in csp
assert "script-src 'self'" in csp
assert "frame-ancestors 'none'" in csp
assert "object-src 'none'" in csp
@pytest.mark.asyncio
@pytest.mark.integration
async def test_hsts_absent_for_http(self, async_client: AsyncClient):
"""HSTS must NOT be set over plain HTTP (test transport uses http)."""
resp = await async_client.get(ME_URL)
assert "strict-transport-security" not in resp.headers
# ===========================================================================
# I3: Rate-limit bucket interaction — IP spray vs. username spray
# ===========================================================================
class TestRateLimitBuckets:
"""IP-spray and username-spray must each trip the correct independent bucket."""
@pytest.mark.asyncio
@pytest.mark.integration
async def test_ip_spray_trips_ip_bucket(self, async_client: AsyncClient):
"""20 failed logins from one IP across 20 different usernames trips the IP bucket.
Each per-username bucket only has 1 failure (well below MAX_LOGIN_ATTEMPTS=10),
so the username bucket is never the reason for the 429.
"""
from unittest.mock import patch as _patch
unique_ip = "10.99.1.1"
# Ensure auth is enabled
await async_client.post(
AUTH_SETUP_URL,
json={"auth_enabled": True, "admin_username": "spray_ip_admin", "admin_password": "SprayIp_admin1"},
)
status_codes: list[int] = []
with _patch("backend.app.api.routes.auth._get_client_ip", return_value=unique_ip):
for i in range(22):
resp = await async_client.post(
LOGIN_URL,
json={"username": f"spray_ip_victim_{i}", "password": "wrong"},
)
status_codes.append(resp.status_code)
# The first 20 attempts fail with 401; the 21st+ must be 429 (IP bucket full)
assert status_codes[-1] == 429, f"Expected 429 after 20 IP-spray failures, got: {status_codes}"
# No single username saw more than one attempt → username buckets not tripped
non_429 = [c for c in status_codes[:-2] if c == 429]
assert not non_429, f"Username bucket triggered early: {status_codes}"
@pytest.mark.asyncio
@pytest.mark.integration
async def test_username_spray_trips_username_bucket(self, async_client: AsyncClient):
"""One username targeted from 10+ different IPs trips the username bucket.
Each per-IP bucket only sees 1 failure, so no IP bucket is tripped.
The username bucket (max 10) is what fires the 429.
"""
from unittest.mock import patch as _patch
from backend.app.api.routes.mfa import MAX_LOGIN_ATTEMPTS
# Ensure auth is enabled
await async_client.post(
AUTH_SETUP_URL,
json={
"auth_enabled": True,
"admin_username": "spray_uname_admin",
"admin_password": "SprayUname_admin1",
},
)
target_username = "spray_uname_victim"
status_codes: list[int] = []
for i in range(MAX_LOGIN_ATTEMPTS + 2):
rotating_ip = f"10.99.2.{i + 1}"
with _patch("backend.app.api.routes.auth._get_client_ip", return_value=rotating_ip):
resp = await async_client.post(
LOGIN_URL,
json={"username": target_username, "password": "wrong"},
)
status_codes.append(resp.status_code)
# After MAX_LOGIN_ATTEMPTS failures for same username the bucket fires
assert status_codes[-1] == 429, (
f"Expected 429 after {MAX_LOGIN_ATTEMPTS} username-spray failures, got: {status_codes}"
)
backend/tests/unit/test_mfa_helpers.py
+49
View File
@@ -0,0 +1,49 @@
"""Unit tests for 2FA helper functions in mfa.py."""
import base64
import string
import pytest
from passlib.context import CryptContext
from backend.app.api.routes.mfa import _generate_backup_codes, _generate_totp_qr_b64
class TestBackupCodeGeneration:
"""Tests for backup code helpers."""
def test_generates_ten_codes(self):
plain, hashed = _generate_backup_codes()
assert len(plain) == 10
assert len(hashed) == 10
def test_codes_are_eight_chars(self):
plain, _ = _generate_backup_codes()
for code in plain:
assert len(code) == 8
def test_codes_are_alphanumeric(self):
allowed = set(string.ascii_uppercase + string.digits)
plain, _ = _generate_backup_codes()
for code in plain:
assert all(c in allowed for c in code)
def test_hashes_verify_against_plain(self):
ctx = CryptContext(schemes=["pbkdf2_sha256"], deprecated="auto")
plain, hashed = _generate_backup_codes()
for p, h in zip(plain, hashed, strict=True):
assert ctx.verify(p, h)
def test_codes_are_unique(self):
plain, _ = _generate_backup_codes()
assert len(set(plain)) == 10
class TestTOTPQRCode:
"""Tests for QR code generation helper."""
def test_generates_base64_png(self):
uri = "otpauth://totp/Bambuddy:testuser?secret=BASE32SECRET&issuer=Bambuddy"
result = _generate_totp_qr_b64(uri)
decoded = base64.b64decode(result)
assert decoded[:4] == b"\x89PNG"