Gan, Jimmy
1a37f34401
- Add max_length validation to LoginRequest (username 128, password 1024) - Replace raw request.json() with Pydantic models in RBAC override/update endpoints - Replace raw request.json() with Pydantic models in passkey register/login/delete - Bind passkey challenges to session-bound challenge_id (prevents cross-session replay) - Gate audit log and security log endpoints behind admin role - Fix fragile opc_db.json.dumps() → import json directly - Add COOKIE_SECURE=False startup warning (suppress with ALLOW_INSECURE_COOKIES) Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
516 lines
20 KiB
Python
516 lines
20 KiB
Python
"""
|
|
Integration tests for authentication flow.
|
|
"""
|
|
|
|
import pyotp
|
|
|
|
|
|
class TestLoginFlow:
|
|
"""Test login endpoint."""
|
|
|
|
def test_login_success_no_2fa(self, test_app, mock_config, temp_auth_file):
|
|
"""Test successful login without 2FA."""
|
|
from auth_service import get_password_hash, save_password_hash
|
|
|
|
# Set up password
|
|
password = "test_password"
|
|
hashed = get_password_hash(password)
|
|
save_password_hash(hashed)
|
|
|
|
response = test_app.post(
|
|
"/api/auth/login", json={"username": "testadmin", "password": password, "totp_code": ""}
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["user"] == "testadmin"
|
|
assert data["role"] == "admin"
|
|
assert data["has_2fa"] is False
|
|
assert "access_token" in data
|
|
assert "refresh_token" in data
|
|
assert data["token_type"] == "bearer"
|
|
|
|
def test_login_success_with_2fa(self, test_app, mock_config, temp_auth_file, sample_totp_secret):
|
|
"""Test successful login with 2FA."""
|
|
from auth_service import get_password_hash, save_password_hash, save_totp_secret
|
|
|
|
password = "test_password"
|
|
hashed = get_password_hash(password)
|
|
save_password_hash(hashed)
|
|
save_totp_secret(sample_totp_secret)
|
|
|
|
totp = pyotp.TOTP(sample_totp_secret)
|
|
valid_code = totp.now()
|
|
|
|
response = test_app.post(
|
|
"/api/auth/login", json={"username": "testadmin", "password": password, "totp_code": valid_code}
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["user"] == "testadmin"
|
|
assert data["has_2fa"] is True
|
|
|
|
def test_login_wrong_password(self, test_app, mock_config, temp_auth_file):
|
|
"""Test login with wrong password."""
|
|
from auth_service import get_password_hash, save_password_hash
|
|
|
|
password = "correct_password"
|
|
hashed = get_password_hash(password)
|
|
save_password_hash(hashed)
|
|
|
|
response = test_app.post(
|
|
"/api/auth/login", json={"username": "testadmin", "password": "wrong_password", "totp_code": ""}
|
|
)
|
|
|
|
assert response.status_code == 401
|
|
assert "Invalid credentials" in response.json()["detail"]
|
|
|
|
def test_login_wrong_username(self, test_app, mock_config, temp_auth_file):
|
|
"""Test login with wrong username."""
|
|
from auth_service import get_password_hash, save_password_hash
|
|
|
|
password = "test_password"
|
|
hashed = get_password_hash(password)
|
|
save_password_hash(hashed)
|
|
|
|
response = test_app.post(
|
|
"/api/auth/login", json={"username": "wronguser", "password": password, "totp_code": ""}
|
|
)
|
|
|
|
assert response.status_code == 401
|
|
|
|
def test_login_missing_2fa_code(self, test_app, mock_config, temp_auth_file, sample_totp_secret):
|
|
"""Test login with 2FA enabled but no code provided."""
|
|
from auth_service import get_password_hash, save_password_hash, save_totp_secret
|
|
|
|
password = "test_password"
|
|
hashed = get_password_hash(password)
|
|
save_password_hash(hashed)
|
|
save_totp_secret(sample_totp_secret)
|
|
|
|
response = test_app.post(
|
|
"/api/auth/login", json={"username": "testadmin", "password": password, "totp_code": ""}
|
|
)
|
|
|
|
assert response.status_code == 403
|
|
assert "2FA code required" in response.json()["detail"]
|
|
|
|
def test_login_invalid_2fa_code(self, test_app, mock_config, temp_auth_file, sample_totp_secret):
|
|
"""Test login with invalid 2FA code."""
|
|
from auth_service import get_password_hash, save_password_hash, save_totp_secret
|
|
|
|
password = "test_password"
|
|
hashed = get_password_hash(password)
|
|
save_password_hash(hashed)
|
|
save_totp_secret(sample_totp_secret)
|
|
|
|
response = test_app.post(
|
|
"/api/auth/login", json={"username": "testadmin", "password": password, "totp_code": "000000"}
|
|
)
|
|
|
|
assert response.status_code == 401
|
|
|
|
def test_login_sets_cookies(self, test_app, mock_config, temp_auth_file):
|
|
"""Test that login sets auth cookies."""
|
|
from auth_service import get_password_hash, save_password_hash
|
|
|
|
password = "test_password"
|
|
hashed = get_password_hash(password)
|
|
save_password_hash(hashed)
|
|
|
|
response = test_app.post(
|
|
"/api/auth/login", json={"username": "testadmin", "password": password, "totp_code": ""}
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
# Check that cookies are set
|
|
assert "nas_access_token" in response.cookies
|
|
assert "nas_refresh_token" in response.cookies
|
|
|
|
|
|
class TestRefreshFlow:
|
|
"""Test token refresh endpoint."""
|
|
|
|
def test_refresh_with_valid_token(self, test_app, mock_config, temp_auth_file, valid_refresh_token):
|
|
"""Test refreshing with valid refresh token."""
|
|
response = test_app.post("/api/auth/refresh", json={"refresh_token": valid_refresh_token})
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "access_token" in data
|
|
assert "refresh_token" in data
|
|
assert data["access_token"] != valid_refresh_token
|
|
|
|
def test_refresh_with_cookie(self, test_app, mock_config, temp_auth_file, valid_refresh_token):
|
|
"""Test refreshing with refresh token in cookie."""
|
|
test_app.cookies.set("nas_refresh_token", valid_refresh_token)
|
|
|
|
response = test_app.post("/api/auth/refresh", json={})
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "access_token" in data
|
|
|
|
def test_refresh_with_invalid_token(self, test_app, mock_config, temp_auth_file):
|
|
"""Test refreshing with invalid token."""
|
|
response = test_app.post("/api/auth/refresh", json={"refresh_token": "invalid.token.here"})
|
|
|
|
assert response.status_code == 401
|
|
|
|
def test_refresh_with_expired_token(self, test_app, mock_config, temp_auth_file, expired_access_token):
|
|
"""Test refreshing with expired token."""
|
|
response = test_app.post("/api/auth/refresh", json={"refresh_token": expired_access_token})
|
|
|
|
assert response.status_code == 401
|
|
|
|
def test_refresh_missing_token(self, test_app, mock_config, temp_auth_file):
|
|
"""Test refresh without providing token."""
|
|
response = test_app.post("/api/auth/refresh", json={})
|
|
|
|
assert response.status_code == 401
|
|
assert "Missing refresh token" in response.json()["detail"]
|
|
|
|
def test_refresh_with_access_token_fails(self, test_app, mock_config, temp_auth_file, valid_access_token):
|
|
"""Test that access token cannot be used for refresh."""
|
|
response = test_app.post("/api/auth/refresh", json={"refresh_token": valid_access_token})
|
|
|
|
assert response.status_code == 401
|
|
assert "Invalid token type" in response.json()["detail"]
|
|
|
|
|
|
class TestLogoutFlow:
|
|
"""Test logout endpoint."""
|
|
|
|
def test_logout_success(self, test_app, mock_config, temp_auth_file, admin_headers):
|
|
"""Test successful logout."""
|
|
response = test_app.post("/api/auth/logout", headers=admin_headers)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["ok"] is True
|
|
|
|
def test_logout_clears_cookies(self, test_app, mock_config, temp_auth_file, admin_headers):
|
|
"""Test that logout clears auth cookies."""
|
|
response = test_app.post("/api/auth/logout", headers=admin_headers)
|
|
|
|
assert response.status_code == 200
|
|
# Cookies should be deleted (max-age=0 or expires in past)
|
|
# TestClient doesn't fully simulate cookie deletion, but we can check response
|
|
|
|
def test_logout_without_auth(self, test_app, mock_config, temp_auth_file):
|
|
"""Test logout without authentication - should succeed as logout doesn't require auth."""
|
|
response = test_app.post("/api/auth/logout")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["ok"] is True
|
|
|
|
|
|
class TestUserInfo:
|
|
"""Test user info endpoint."""
|
|
|
|
def test_get_user_info(self, test_app, mock_config, temp_auth_file, temp_rbac_file, admin_headers):
|
|
"""Test getting current user info."""
|
|
response = test_app.get("/api/auth/me", headers=admin_headers)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["username"] == "testuser"
|
|
assert data["role"] == "admin"
|
|
assert "pages" in data
|
|
assert "sidebar_links" in data
|
|
|
|
def test_get_user_info_without_auth(self, test_app, mock_config, temp_auth_file):
|
|
"""Test getting user info without authentication."""
|
|
response = test_app.get("/api/auth/me")
|
|
|
|
assert response.status_code == 401
|
|
|
|
|
|
class TestProxyAuth:
|
|
"""Test proxy authentication endpoint."""
|
|
|
|
def test_proxy_auth_from_trusted_source(self, test_app, mock_config, temp_auth_file, temp_rbac_file):
|
|
"""Test proxy auth from trusted proxy."""
|
|
response = test_app.get(
|
|
"/api/auth/proxy", headers={"Remote-User": "proxyuser", "Remote-Groups": "dashboard_admin"}
|
|
)
|
|
|
|
# Note: This test may need adjustment based on actual proxy auth implementation
|
|
# The test client's default host may not be in trusted proxies
|
|
assert response.status_code in [200, 403]
|
|
|
|
def test_proxy_auth_from_untrusted_source(self, test_app, mock_config, temp_auth_file):
|
|
"""Test proxy auth from untrusted source."""
|
|
# Mock request from untrusted IP
|
|
response = test_app.get(
|
|
"/api/auth/proxy", headers={"Remote-User": "proxyuser", "Remote-Groups": "dashboard_admin"}
|
|
)
|
|
|
|
# Should reject untrusted proxy
|
|
assert response.status_code == 403
|
|
|
|
|
|
class TestPasswordChange:
|
|
"""Test password change endpoint."""
|
|
|
|
def test_change_password_success(self, test_app, mock_config, temp_auth_file, admin_headers):
|
|
"""Test successful password change."""
|
|
from auth_service import get_password_hash, save_password_hash
|
|
|
|
old_password = "old_password"
|
|
new_password = "new_password_123"
|
|
|
|
# Set initial password
|
|
hashed = get_password_hash(old_password)
|
|
save_password_hash(hashed)
|
|
|
|
response = test_app.post(
|
|
"/api/auth/change-password",
|
|
headers=admin_headers,
|
|
json={"current_password": old_password, "new_password": new_password},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
|
|
def test_change_password_wrong_old_password(self, test_app, mock_config, temp_auth_file, admin_headers):
|
|
"""Test password change with wrong old password."""
|
|
from auth_service import get_password_hash, save_password_hash
|
|
|
|
old_password = "old_password"
|
|
hashed = get_password_hash(old_password)
|
|
save_password_hash(hashed)
|
|
|
|
response = test_app.post(
|
|
"/api/auth/change-password",
|
|
headers=admin_headers,
|
|
json={"current_password": "wrong_password", "new_password": "new_password_123"},
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
|
|
def test_change_password_without_auth(self, test_app, mock_config, temp_auth_file):
|
|
"""Test password change without authentication."""
|
|
response = test_app.post("/api/auth/change-password", json={"current_password": "old", "new_password": "new"})
|
|
|
|
assert response.status_code == 401
|
|
|
|
def test_change_password_too_short(self, test_app, mock_config, temp_auth_file, admin_headers):
|
|
"""Test password change with password too short."""
|
|
from auth_service import get_password_hash, save_password_hash
|
|
|
|
old_password = "old_password"
|
|
hashed = get_password_hash(old_password)
|
|
save_password_hash(hashed)
|
|
|
|
response = test_app.post(
|
|
"/api/auth/change-password",
|
|
headers=admin_headers,
|
|
json={"current_password": old_password, "new_password": "short"},
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
assert "at least 8 characters" in response.json()["detail"]
|
|
|
|
|
|
class TestRBACConfig:
|
|
"""Test RBAC configuration endpoints."""
|
|
|
|
def test_get_rbac_config_as_admin(self, test_app, mock_config, temp_auth_file, temp_rbac_file, admin_headers):
|
|
"""Test getting RBAC config as admin."""
|
|
response = test_app.get("/api/auth/rbac/config", headers=admin_headers)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "role_defaults" in data or "user_overrides" in data
|
|
|
|
def test_get_rbac_config_as_member(self, test_app, mock_config, temp_auth_file, temp_rbac_file):
|
|
"""Test that non-admin cannot get RBAC config."""
|
|
from datetime import timedelta
|
|
|
|
from auth_service import create_access_token
|
|
|
|
member_token = create_access_token(
|
|
data={"sub": "memberuser", "role": "member"}, expires_delta=timedelta(minutes=30)
|
|
)
|
|
headers = {"Authorization": f"Bearer {member_token}"}
|
|
|
|
response = test_app.get("/api/auth/rbac/config", headers=headers)
|
|
|
|
assert response.status_code == 403
|
|
|
|
def test_get_rbac_config_without_auth(self, test_app, mock_config, temp_auth_file):
|
|
"""Test getting RBAC config without authentication."""
|
|
response = test_app.get("/api/auth/rbac/config")
|
|
|
|
assert response.status_code == 401
|
|
|
|
def test_set_user_override(self, test_app, mock_config, temp_auth_file, temp_rbac_file, admin_headers):
|
|
"""Test setting user page override."""
|
|
response = test_app.put(
|
|
"/api/auth/rbac/overrides/testuser", headers=admin_headers, json={"pages": ["overview", "docker"]}
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["ok"] is True
|
|
|
|
def test_set_user_override_missing_pages(
|
|
self, test_app, mock_config, temp_auth_file, temp_rbac_file, admin_headers
|
|
):
|
|
"""Test setting user override without pages field."""
|
|
response = test_app.put("/api/auth/rbac/overrides/testuser", headers=admin_headers, json={})
|
|
|
|
assert response.status_code == 422
|
|
|
|
def test_set_user_override_as_member(self, test_app, mock_config, temp_auth_file, temp_rbac_file):
|
|
"""Test that non-admin cannot set user override."""
|
|
from datetime import timedelta
|
|
|
|
from auth_service import create_access_token
|
|
|
|
member_token = create_access_token(
|
|
data={"sub": "memberuser", "role": "member"}, expires_delta=timedelta(minutes=30)
|
|
)
|
|
headers = {"Authorization": f"Bearer {member_token}"}
|
|
|
|
response = test_app.put("/api/auth/rbac/overrides/testuser", headers=headers, json={"pages": ["overview"]})
|
|
|
|
assert response.status_code == 403
|
|
|
|
def test_delete_user_override(self, test_app, mock_config, temp_auth_file, temp_rbac_file, admin_headers):
|
|
"""Test deleting user override."""
|
|
# First set an override
|
|
test_app.put("/api/auth/rbac/overrides/testuser", headers=admin_headers, json={"pages": ["overview"]})
|
|
|
|
# Then delete it
|
|
response = test_app.delete("/api/auth/rbac/overrides/testuser", headers=admin_headers)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["ok"] is True
|
|
|
|
def test_delete_user_override_as_member(self, test_app, mock_config, temp_auth_file, temp_rbac_file):
|
|
"""Test that non-admin cannot delete user override."""
|
|
from datetime import timedelta
|
|
|
|
from auth_service import create_access_token
|
|
|
|
member_token = create_access_token(
|
|
data={"sub": "memberuser", "role": "member"}, expires_delta=timedelta(minutes=30)
|
|
)
|
|
headers = {"Authorization": f"Bearer {member_token}"}
|
|
|
|
response = test_app.delete("/api/auth/rbac/overrides/testuser", headers=headers)
|
|
|
|
assert response.status_code == 403
|
|
|
|
def test_update_role_config(self, test_app, mock_config, temp_auth_file, temp_rbac_file, admin_headers):
|
|
"""Test updating role configuration."""
|
|
response = test_app.put(
|
|
"/api/auth/rbac/roles/member",
|
|
headers=admin_headers,
|
|
json={"pages": ["overview", "docker"], "sidebar_links": ["navidrome", "jellyfin"]},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["ok"] is True
|
|
|
|
def test_update_role_config_with_wildcard(
|
|
self, test_app, mock_config, temp_auth_file, temp_rbac_file, admin_headers
|
|
):
|
|
"""Test updating role with wildcard pages."""
|
|
response = test_app.put(
|
|
"/api/auth/rbac/roles/admin", headers=admin_headers, json={"pages": "*", "sidebar_links": "*"}
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
|
|
def test_update_role_config_missing_pages(
|
|
self, test_app, mock_config, temp_auth_file, temp_rbac_file, admin_headers
|
|
):
|
|
"""Test updating role without pages field."""
|
|
response = test_app.put(
|
|
"/api/auth/rbac/roles/member", headers=admin_headers, json={"sidebar_links": ["navidrome"]}
|
|
)
|
|
|
|
assert response.status_code == 422
|
|
|
|
def test_update_role_config_invalid_pages_type(
|
|
self, test_app, mock_config, temp_auth_file, temp_rbac_file, admin_headers
|
|
):
|
|
"""Test updating role with invalid pages type."""
|
|
response = test_app.put("/api/auth/rbac/roles/member", headers=admin_headers, json={"pages": "invalid"})
|
|
|
|
assert response.status_code == 400
|
|
assert "pages must be" in response.json()["detail"]
|
|
|
|
def test_update_role_config_as_member(self, test_app, mock_config, temp_auth_file, temp_rbac_file):
|
|
"""Test that non-admin cannot update role config."""
|
|
from datetime import timedelta
|
|
|
|
from auth_service import create_access_token
|
|
|
|
member_token = create_access_token(
|
|
data={"sub": "memberuser", "role": "member"}, expires_delta=timedelta(minutes=30)
|
|
)
|
|
headers = {"Authorization": f"Bearer {member_token}"}
|
|
|
|
response = test_app.put("/api/auth/rbac/roles/member", headers=headers, json={"pages": ["overview"]})
|
|
|
|
assert response.status_code == 403
|
|
|
|
|
|
class TestPreferences:
|
|
"""Test user preferences endpoints."""
|
|
|
|
def test_get_preferences(self, test_app, mock_config, temp_auth_file, temp_rbac_file, admin_headers):
|
|
"""Test getting user preferences."""
|
|
response = test_app.get("/api/auth/preferences", headers=admin_headers)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "sidebar_order" in data
|
|
|
|
def test_get_preferences_without_auth(self, test_app, mock_config, temp_auth_file):
|
|
"""Test getting preferences without authentication."""
|
|
response = test_app.get("/api/auth/preferences")
|
|
|
|
assert response.status_code == 401
|
|
|
|
def test_save_preferences(self, test_app, mock_config, temp_auth_file, temp_rbac_file, admin_headers):
|
|
"""Test saving user preferences."""
|
|
response = test_app.put(
|
|
"/api/auth/preferences",
|
|
headers=admin_headers,
|
|
json={"sidebar_order": {"main": ["overview", "docker", "files"], "media": ["navidrome", "jellyfin"]}},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["ok"] is True
|
|
|
|
def test_save_preferences_missing_sidebar_order(
|
|
self, test_app, mock_config, temp_auth_file, temp_rbac_file, admin_headers
|
|
):
|
|
"""Test saving preferences without sidebar_order."""
|
|
response = test_app.put("/api/auth/preferences", headers=admin_headers, json={})
|
|
|
|
assert response.status_code == 400
|
|
assert "Missing sidebar_order" in response.json()["detail"]
|
|
|
|
def test_save_preferences_invalid_type(self, test_app, mock_config, temp_auth_file, temp_rbac_file, admin_headers):
|
|
"""Test saving preferences with invalid type."""
|
|
response = test_app.put("/api/auth/preferences", headers=admin_headers, json={"sidebar_order": "invalid"})
|
|
|
|
assert response.status_code == 400
|
|
assert "Missing sidebar_order dict" in response.json()["detail"]
|
|
|
|
def test_save_preferences_without_auth(self, test_app, mock_config, temp_auth_file):
|
|
"""Test saving preferences without authentication."""
|
|
response = test_app.put("/api/auth/preferences", json={"sidebar_order": {}})
|
|
|
|
assert response.status_code == 401
|