mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-08-11 00:30:12 -04:00
c236fdc650
The /system/appliance endpoint is fetched by the SPA's i18n bootstrap on mount to seed locale, hostname, timezone, and the chrony NTP-gate state BEFORE any login state exists. The route handler itself has no auth dependency and the test_route_auth_coverage allowlist correctly marks it public, but the global auth_middleware in main.py — which short-circuits every /api/ path not in PUBLIC_API_ROUTES — was never told about it. Result: every browser session on an auth-enabled install logged a 401 on the appliance endpoint before login. Added /api/v1/system/appliance to PUBLIC_API_ROUTES with a comment pointing at the dual-list pattern so this doesn't drift again, and a regression test in TestAuthMiddlewarePublicRoutes that posts /auth/setup to turn auth on, then asserts the endpoint returns 200 with the documented shape (hostname / timezone / locale / time_synced fields all present).
1012 lines
37 KiB
Python
1012 lines
37 KiB
Python
"""Integration tests for Authentication API endpoints.
|
|
|
|
Tests the full request/response cycle for /api/v1/auth/ and /api/v1/users/ endpoints.
|
|
"""
|
|
|
|
import pytest
|
|
from httpx import AsyncClient
|
|
|
|
|
|
class TestAuthStatusAPI:
|
|
"""Integration tests for /api/v1/auth/status endpoint."""
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_get_auth_status_disabled(self, async_client: AsyncClient):
|
|
"""Verify auth status returns disabled when not configured."""
|
|
response = await async_client.get("/api/v1/auth/status")
|
|
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
assert "auth_enabled" in result
|
|
assert result["auth_enabled"] is False
|
|
assert result["requires_setup"] is True
|
|
|
|
|
|
class TestAuthSetupAPI:
|
|
"""Integration tests for /api/v1/auth/setup endpoint."""
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_setup_auth_disabled(self, async_client: AsyncClient):
|
|
"""Verify auth can be set up with auth disabled (no password required)."""
|
|
response = await async_client.post(
|
|
"/api/v1/auth/setup",
|
|
json={"auth_enabled": False},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
assert result["auth_enabled"] is False
|
|
assert result["admin_created"] is False
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_setup_auth_enabled_requires_credentials(self, async_client: AsyncClient):
|
|
"""Verify enabling auth requires admin username and password."""
|
|
response = await async_client.post(
|
|
"/api/v1/auth/setup",
|
|
json={"auth_enabled": True},
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
assert "Admin username and password are required" in response.json()["detail"]
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_setup_auth_enabled_with_credentials(self, async_client: AsyncClient):
|
|
"""Verify auth can be enabled with admin credentials."""
|
|
response = await async_client.post(
|
|
"/api/v1/auth/setup",
|
|
json={
|
|
"auth_enabled": True,
|
|
"admin_username": "testadmin",
|
|
"admin_password": "TestPass1!",
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
assert result["auth_enabled"] is True
|
|
assert result["admin_created"] is True
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_setup_weak_password_rejected_when_creating_new_admin(self, async_client: AsyncClient):
|
|
"""Complexity is enforced only when a new admin is being created."""
|
|
response = await async_client.post(
|
|
"/api/v1/auth/setup",
|
|
json={
|
|
"auth_enabled": True,
|
|
"admin_username": "weakpw_admin",
|
|
"admin_password": "NoSpecial1",
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
assert "special character" in response.json()["detail"].lower()
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_setup_reenable_with_existing_admin_ignores_password(self, async_client: AsyncClient, db_session):
|
|
"""Re-enabling auth when an admin already exists must not reject the placeholder
|
|
password the frontend still sends. Regression for the LDAP re-enable flow that
|
|
previously 422'd because the Pydantic schema enforced complexity unconditionally.
|
|
"""
|
|
from backend.app.core.auth import get_password_hash
|
|
from backend.app.models.user import User
|
|
|
|
existing = User(
|
|
username="existing_admin",
|
|
# pragma: allowlist secret — test fixture only, not a real credential
|
|
password_hash=get_password_hash("DoesNotMatter1!"), # noqa: S106
|
|
role="admin",
|
|
is_active=True,
|
|
)
|
|
db_session.add(existing)
|
|
await db_session.commit()
|
|
|
|
response = await async_client.post(
|
|
"/api/v1/auth/setup",
|
|
json={
|
|
"auth_enabled": True,
|
|
"admin_username": "irrelevant",
|
|
"admin_password": "NoSpecial1",
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
assert result["auth_enabled"] is True
|
|
assert result["admin_created"] is False
|
|
|
|
|
|
class TestAuthLoginAPI:
|
|
"""Integration tests for /api/v1/auth/login endpoint."""
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_login_auth_disabled(self, async_client: AsyncClient):
|
|
"""Verify login fails when auth is not enabled."""
|
|
response = await async_client.post(
|
|
"/api/v1/auth/login",
|
|
json={"username": "admin", "password": "password"},
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
assert "Authentication is not enabled" in response.json()["detail"]
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_login_success(self, async_client: AsyncClient):
|
|
"""Verify login succeeds with valid credentials after setup."""
|
|
# First enable auth
|
|
await async_client.post(
|
|
"/api/v1/auth/setup",
|
|
json={
|
|
"auth_enabled": True,
|
|
"admin_username": "logintest",
|
|
"admin_password": "LoginPass1!",
|
|
},
|
|
)
|
|
|
|
# Now login
|
|
response = await async_client.post(
|
|
"/api/v1/auth/login",
|
|
json={"username": "logintest", "password": "LoginPass1!"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
assert "access_token" in result
|
|
assert result["token_type"] == "bearer"
|
|
assert result["user"]["username"] == "logintest"
|
|
assert result["user"]["role"] == "admin"
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_login_invalid_credentials(self, async_client: AsyncClient):
|
|
"""Verify login fails with invalid credentials."""
|
|
# First enable auth
|
|
await async_client.post(
|
|
"/api/v1/auth/setup",
|
|
json={
|
|
"auth_enabled": True,
|
|
"admin_username": "invalidtest",
|
|
"admin_password": "CorrectPass1!",
|
|
},
|
|
)
|
|
|
|
# Try login with wrong password
|
|
response = await async_client.post(
|
|
"/api/v1/auth/login",
|
|
json={"username": "invalidtest", "password": "wrongpassword"},
|
|
)
|
|
|
|
assert response.status_code == 401
|
|
assert "Incorrect username or password" in response.json()["detail"]
|
|
|
|
|
|
class TestAuthMeAPI:
|
|
"""Integration tests for /api/v1/auth/me endpoint."""
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_me_without_token(self, async_client: AsyncClient):
|
|
"""Verify /me fails without authentication token."""
|
|
response = await async_client.get("/api/v1/auth/me")
|
|
|
|
assert response.status_code == 401
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_me_with_valid_token(self, async_client: AsyncClient):
|
|
"""Verify /me returns user info with valid token."""
|
|
# Setup and login
|
|
await async_client.post(
|
|
"/api/v1/auth/setup",
|
|
json={
|
|
"auth_enabled": True,
|
|
"admin_username": "metest",
|
|
"admin_password": "MePass1!",
|
|
},
|
|
)
|
|
|
|
login_response = await async_client.post(
|
|
"/api/v1/auth/login",
|
|
json={"username": "metest", "password": "MePass1!"},
|
|
)
|
|
token = login_response.json()["access_token"]
|
|
|
|
# Get current user
|
|
response = await async_client.get(
|
|
"/api/v1/auth/me",
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
assert result["username"] == "metest"
|
|
assert result["role"] == "admin"
|
|
assert result["is_active"] is True
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_me_with_api_key_bearer(self, async_client: AsyncClient, db_session):
|
|
"""Verify /me returns synthetic admin user when using API key via Bearer token."""
|
|
from backend.app.core.auth import generate_api_key
|
|
from backend.app.models.api_key import APIKey
|
|
|
|
# Create an API key directly in the database
|
|
full_key, key_hash, key_prefix = generate_api_key()
|
|
api_key = APIKey(name="test-kiosk", key_hash=key_hash, key_prefix=key_prefix, enabled=True)
|
|
db_session.add(api_key)
|
|
await db_session.commit()
|
|
|
|
# Call /me with the API key as Bearer token
|
|
response = await async_client.get(
|
|
"/api/v1/auth/me",
|
|
headers={"Authorization": f"Bearer {full_key}"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
assert result["id"] == 0
|
|
assert result["username"].startswith("api-key:")
|
|
assert result["role"] == "admin"
|
|
assert result["is_admin"] is True
|
|
assert result["is_active"] is True
|
|
assert len(result["permissions"]) > 0
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_me_with_api_key_header(self, async_client: AsyncClient, db_session):
|
|
"""Verify /me returns synthetic admin user when using X-API-Key header."""
|
|
from backend.app.core.auth import generate_api_key
|
|
from backend.app.models.api_key import APIKey
|
|
|
|
full_key, key_hash, key_prefix = generate_api_key()
|
|
api_key = APIKey(name="test-kiosk-header", key_hash=key_hash, key_prefix=key_prefix, enabled=True)
|
|
db_session.add(api_key)
|
|
await db_session.commit()
|
|
|
|
response = await async_client.get(
|
|
"/api/v1/auth/me",
|
|
headers={"X-API-Key": full_key},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
assert result["id"] == 0
|
|
assert result["username"].startswith("api-key:")
|
|
assert result["is_admin"] is True
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_me_with_invalid_api_key(self, async_client: AsyncClient):
|
|
"""Verify /me rejects invalid API key."""
|
|
response = await async_client.get(
|
|
"/api/v1/auth/me",
|
|
headers={"Authorization": "Bearer bb_invalid_key_value"},
|
|
)
|
|
|
|
assert response.status_code == 401
|
|
|
|
|
|
class TestUsersAPI:
|
|
"""Integration tests for /api/v1/users/ endpoints."""
|
|
|
|
@pytest.fixture
|
|
async def auth_token(self, async_client: AsyncClient):
|
|
"""Setup auth and return admin token."""
|
|
await async_client.post(
|
|
"/api/v1/auth/setup",
|
|
json={
|
|
"auth_enabled": True,
|
|
"admin_username": "usersadmin",
|
|
"admin_password": "AdminPass1!",
|
|
},
|
|
)
|
|
|
|
login_response = await async_client.post(
|
|
"/api/v1/auth/login",
|
|
json={"username": "usersadmin", "password": "AdminPass1!"},
|
|
)
|
|
return login_response.json()["access_token"]
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_list_users_requires_auth(self, async_client: AsyncClient):
|
|
"""Verify listing users requires authentication when auth is enabled."""
|
|
# First enable auth
|
|
await async_client.post(
|
|
"/api/v1/auth/setup",
|
|
json={
|
|
"auth_enabled": True,
|
|
"admin_username": "authreqadmin",
|
|
"admin_password": "AdminPass1!",
|
|
},
|
|
)
|
|
|
|
# Now try to list users without a token
|
|
response = await async_client.get("/api/v1/users/")
|
|
|
|
assert response.status_code == 401
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_list_users_as_admin(self, async_client: AsyncClient, auth_token: str):
|
|
"""Verify admin can list users."""
|
|
response = await async_client.get(
|
|
"/api/v1/users/",
|
|
headers={"Authorization": f"Bearer {auth_token}"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
assert isinstance(result, list)
|
|
assert len(result) >= 1 # At least the admin user
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_create_user(self, async_client: AsyncClient, auth_token: str):
|
|
"""Verify admin can create a new user."""
|
|
response = await async_client.post(
|
|
"/api/v1/users/",
|
|
headers={"Authorization": f"Bearer {auth_token}"},
|
|
json={
|
|
"username": "newuser",
|
|
"password": "Newuserpass1!",
|
|
"role": "user",
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 201
|
|
result = response.json()
|
|
assert result["username"] == "newuser"
|
|
assert result["role"] == "user"
|
|
assert result["is_active"] is True
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_create_user_duplicate_username(self, async_client: AsyncClient, auth_token: str):
|
|
"""Verify creating user with duplicate username fails."""
|
|
# Create first user
|
|
await async_client.post(
|
|
"/api/v1/users/",
|
|
headers={"Authorization": f"Bearer {auth_token}"},
|
|
json={
|
|
"username": "duplicateuser",
|
|
"password": "Password123!",
|
|
"role": "user",
|
|
},
|
|
)
|
|
|
|
# Try to create duplicate
|
|
response = await async_client.post(
|
|
"/api/v1/users/",
|
|
headers={"Authorization": f"Bearer {auth_token}"},
|
|
json={
|
|
"username": "duplicateuser",
|
|
"password": "Password456!",
|
|
"role": "user",
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
assert "Username already exists" in response.json()["detail"]
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_update_user(self, async_client: AsyncClient, auth_token: str):
|
|
"""Verify admin can update a user."""
|
|
# Create user
|
|
create_response = await async_client.post(
|
|
"/api/v1/users/",
|
|
headers={"Authorization": f"Bearer {auth_token}"},
|
|
json={
|
|
"username": "updateuser",
|
|
"password": "Password123!",
|
|
"role": "user",
|
|
},
|
|
)
|
|
user_id = create_response.json()["id"]
|
|
|
|
# Update user
|
|
response = await async_client.patch(
|
|
f"/api/v1/users/{user_id}",
|
|
headers={"Authorization": f"Bearer {auth_token}"},
|
|
json={"role": "admin"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()["role"] == "admin"
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_delete_user(self, async_client: AsyncClient, auth_token: str, db_session):
|
|
"""Verify admin can delete a user and that all auth-table side effects cascade.
|
|
|
|
The auth-cleanup side effects matter on SQLite (FK enforcement off by default):
|
|
without explicit DELETEs in the endpoint, deleting a user leaves orphan rows
|
|
in user_oidc_links / user_totp / user_otp_codes / api_keys — which would
|
|
block SSO re-login and leak MFA secrets (#1285).
|
|
"""
|
|
from sqlalchemy import select
|
|
|
|
from backend.app.models.api_key import APIKey
|
|
from backend.app.models.long_lived_token import LongLivedToken
|
|
from backend.app.models.oidc_provider import UserOIDCLink
|
|
from backend.app.models.user import User
|
|
from backend.app.models.user_otp_code import UserOTPCode
|
|
from backend.app.models.user_totp import UserTOTP
|
|
|
|
# Create user
|
|
create_response = await async_client.post(
|
|
"/api/v1/users/",
|
|
headers={"Authorization": f"Bearer {auth_token}"},
|
|
json={
|
|
"username": "deleteuser",
|
|
"password": "Password123!",
|
|
"role": "user",
|
|
},
|
|
)
|
|
user_id = create_response.json()["id"]
|
|
|
|
# Delete user
|
|
response = await async_client.delete(
|
|
f"/api/v1/users/{user_id}",
|
|
headers={"Authorization": f"Bearer {auth_token}"},
|
|
)
|
|
|
|
assert response.status_code == 204
|
|
|
|
# All auth-related rows for this user must be gone — see #1285.
|
|
await db_session.commit()
|
|
user_row = await db_session.execute(select(User).where(User.id == user_id))
|
|
assert user_row.scalar_one_or_none() is None, "User row not deleted"
|
|
|
|
for model in (UserOIDCLink, UserTOTP, UserOTPCode, APIKey, LongLivedToken):
|
|
rows = await db_session.execute(select(model).where(model.user_id == user_id))
|
|
assert rows.scalars().all() == [], f"Orphan {model.__name__} rows left after user delete"
|
|
|
|
|
|
class TestAuthDisableAPI:
|
|
"""Integration tests for /api/v1/auth/disable endpoint."""
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_disable_auth(self, async_client: AsyncClient):
|
|
"""Verify admin can disable authentication."""
|
|
# Setup auth
|
|
await async_client.post(
|
|
"/api/v1/auth/setup",
|
|
json={
|
|
"auth_enabled": True,
|
|
"admin_username": "disableadmin",
|
|
"admin_password": "AdminPass1!",
|
|
},
|
|
)
|
|
|
|
# Login to get token
|
|
login_response = await async_client.post(
|
|
"/api/v1/auth/login",
|
|
json={"username": "disableadmin", "password": "AdminPass1!"},
|
|
)
|
|
token = login_response.json()["access_token"]
|
|
|
|
# Disable auth
|
|
response = await async_client.post(
|
|
"/api/v1/auth/disable",
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()["auth_enabled"] is False
|
|
|
|
# Verify auth is now disabled
|
|
status_response = await async_client.get("/api/v1/auth/status")
|
|
assert status_response.json()["auth_enabled"] is False
|
|
|
|
|
|
class TestGroupsAPI:
|
|
"""Integration tests for /api/v1/groups/ endpoints."""
|
|
|
|
@pytest.fixture
|
|
async def auth_token(self, async_client: AsyncClient):
|
|
"""Setup auth and return admin token."""
|
|
await async_client.post(
|
|
"/api/v1/auth/setup",
|
|
json={
|
|
"auth_enabled": True,
|
|
"admin_username": "groupsadmin",
|
|
"admin_password": "AdminPass1!",
|
|
},
|
|
)
|
|
|
|
login_response = await async_client.post(
|
|
"/api/v1/auth/login",
|
|
json={"username": "groupsadmin", "password": "AdminPass1!"},
|
|
)
|
|
return login_response.json()["access_token"]
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_list_groups(self, async_client: AsyncClient, auth_token: str):
|
|
"""Verify listing groups returns default groups."""
|
|
response = await async_client.get(
|
|
"/api/v1/groups/",
|
|
headers={"Authorization": f"Bearer {auth_token}"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
groups = response.json()
|
|
assert isinstance(groups, list)
|
|
# Should have default groups: Administrators, Operators, Viewers
|
|
group_names = [g["name"] for g in groups]
|
|
assert "Administrators" in group_names
|
|
assert "Operators" in group_names
|
|
assert "Viewers" in group_names
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_get_permissions(self, async_client: AsyncClient, auth_token: str):
|
|
"""Verify getting available permissions."""
|
|
response = await async_client.get(
|
|
"/api/v1/groups/permissions",
|
|
headers={"Authorization": f"Bearer {auth_token}"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
permissions = response.json()
|
|
assert isinstance(permissions, dict)
|
|
# Should have permission categories
|
|
assert "Printers" in permissions or len(permissions) > 0
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_create_group(self, async_client: AsyncClient, auth_token: str):
|
|
"""Verify creating a new group."""
|
|
response = await async_client.post(
|
|
"/api/v1/groups/",
|
|
headers={"Authorization": f"Bearer {auth_token}"},
|
|
json={
|
|
"name": "Custom Group",
|
|
"description": "A custom test group",
|
|
"permissions": ["printers:read", "archives:read"],
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 201
|
|
group = response.json()
|
|
assert group["name"] == "Custom Group"
|
|
assert group["description"] == "A custom test group"
|
|
assert "printers:read" in group["permissions"]
|
|
assert group["is_system"] is False
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_update_group(self, async_client: AsyncClient, auth_token: str):
|
|
"""Verify updating a group."""
|
|
# Create a group first
|
|
create_response = await async_client.post(
|
|
"/api/v1/groups/",
|
|
headers={"Authorization": f"Bearer {auth_token}"},
|
|
json={
|
|
"name": "Update Test Group",
|
|
"permissions": ["printers:read"],
|
|
},
|
|
)
|
|
group_id = create_response.json()["id"]
|
|
|
|
# Update the group
|
|
response = await async_client.patch(
|
|
f"/api/v1/groups/{group_id}",
|
|
headers={"Authorization": f"Bearer {auth_token}"},
|
|
json={
|
|
"description": "Updated description",
|
|
"permissions": ["printers:read", "printers:control"],
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
group = response.json()
|
|
assert group["description"] == "Updated description"
|
|
assert "printers:control" in group["permissions"]
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_cannot_delete_system_group(self, async_client: AsyncClient, auth_token: str):
|
|
"""Verify system groups cannot be deleted."""
|
|
# Get the Administrators group
|
|
list_response = await async_client.get(
|
|
"/api/v1/groups/",
|
|
headers={"Authorization": f"Bearer {auth_token}"},
|
|
)
|
|
admin_group = next(g for g in list_response.json() if g["name"] == "Administrators")
|
|
|
|
# Try to delete it
|
|
response = await async_client.delete(
|
|
f"/api/v1/groups/{admin_group['id']}",
|
|
headers={"Authorization": f"Bearer {auth_token}"},
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
assert "system group" in response.json()["detail"].lower()
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_delete_custom_group(self, async_client: AsyncClient, auth_token: str):
|
|
"""Verify custom groups can be deleted."""
|
|
# Create a group
|
|
create_response = await async_client.post(
|
|
"/api/v1/groups/",
|
|
headers={"Authorization": f"Bearer {auth_token}"},
|
|
json={"name": "Delete Test Group"},
|
|
)
|
|
group_id = create_response.json()["id"]
|
|
|
|
# Delete it
|
|
response = await async_client.delete(
|
|
f"/api/v1/groups/{group_id}",
|
|
headers={"Authorization": f"Bearer {auth_token}"},
|
|
)
|
|
|
|
assert response.status_code == 204
|
|
|
|
|
|
class TestUserGroupsAPI:
|
|
"""Integration tests for user-group assignments."""
|
|
|
|
@pytest.fixture
|
|
async def auth_token(self, async_client: AsyncClient):
|
|
"""Setup auth and return admin token."""
|
|
await async_client.post(
|
|
"/api/v1/auth/setup",
|
|
json={
|
|
"auth_enabled": True,
|
|
"admin_username": "usergroupadmin",
|
|
"admin_password": "AdminPass1!",
|
|
},
|
|
)
|
|
|
|
login_response = await async_client.post(
|
|
"/api/v1/auth/login",
|
|
json={"username": "usergroupadmin", "password": "AdminPass1!"},
|
|
)
|
|
return login_response.json()["access_token"]
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_create_user_with_groups(self, async_client: AsyncClient, auth_token: str):
|
|
"""Verify creating a user with group assignments."""
|
|
# Get Operators group ID
|
|
groups_response = await async_client.get(
|
|
"/api/v1/groups/",
|
|
headers={"Authorization": f"Bearer {auth_token}"},
|
|
)
|
|
operators_group = next(g for g in groups_response.json() if g["name"] == "Operators")
|
|
|
|
# Create user with group
|
|
response = await async_client.post(
|
|
"/api/v1/users/",
|
|
headers={"Authorization": f"Bearer {auth_token}"},
|
|
json={
|
|
"username": "groupuser",
|
|
"password": "Password123!",
|
|
"group_ids": [operators_group["id"]],
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 201
|
|
user = response.json()
|
|
assert any(g["name"] == "Operators" for g in user["groups"])
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_add_user_to_group(self, async_client: AsyncClient, auth_token: str):
|
|
"""Verify adding a user to a group."""
|
|
# Create a user
|
|
user_response = await async_client.post(
|
|
"/api/v1/users/",
|
|
headers={"Authorization": f"Bearer {auth_token}"},
|
|
json={"username": "addtogroup", "password": "Password123!"},
|
|
)
|
|
user_id = user_response.json()["id"]
|
|
|
|
# Get Viewers group
|
|
groups_response = await async_client.get(
|
|
"/api/v1/groups/",
|
|
headers={"Authorization": f"Bearer {auth_token}"},
|
|
)
|
|
viewers_group = next(g for g in groups_response.json() if g["name"] == "Viewers")
|
|
|
|
# Add user to group
|
|
response = await async_client.post(
|
|
f"/api/v1/groups/{viewers_group['id']}/users/{user_id}",
|
|
headers={"Authorization": f"Bearer {auth_token}"},
|
|
)
|
|
|
|
assert response.status_code == 204
|
|
|
|
# Verify user is in group
|
|
user_check = await async_client.get(
|
|
f"/api/v1/users/{user_id}",
|
|
headers={"Authorization": f"Bearer {auth_token}"},
|
|
)
|
|
assert any(g["name"] == "Viewers" for g in user_check.json()["groups"])
|
|
|
|
|
|
class TestChangePasswordAPI:
|
|
"""Integration tests for /api/v1/users/me/change-password endpoint."""
|
|
|
|
@pytest.fixture
|
|
async def user_token(self, async_client: AsyncClient):
|
|
"""Setup auth and return regular user token."""
|
|
# Enable auth with admin
|
|
await async_client.post(
|
|
"/api/v1/auth/setup",
|
|
json={
|
|
"auth_enabled": True,
|
|
"admin_username": "pwchangeadmin",
|
|
"admin_password": "AdminPass1!",
|
|
},
|
|
)
|
|
|
|
admin_login = await async_client.post(
|
|
"/api/v1/auth/login",
|
|
json={"username": "pwchangeadmin", "password": "AdminPass1!"},
|
|
)
|
|
admin_token = admin_login.json()["access_token"]
|
|
|
|
# Create a regular user
|
|
await async_client.post(
|
|
"/api/v1/users/",
|
|
headers={"Authorization": f"Bearer {admin_token}"},
|
|
json={"username": "pwchangeuser", "password": "Oldpassword123!"},
|
|
)
|
|
|
|
# Login as regular user
|
|
user_login = await async_client.post(
|
|
"/api/v1/auth/login",
|
|
json={"username": "pwchangeuser", "password": "Oldpassword123!"},
|
|
)
|
|
return user_login.json()["access_token"]
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_change_password_success(self, async_client: AsyncClient, user_token: str):
|
|
"""Verify user can change their own password."""
|
|
response = await async_client.post(
|
|
"/api/v1/users/me/change-password",
|
|
headers={"Authorization": f"Bearer {user_token}"},
|
|
json={
|
|
"current_password": "Oldpassword123!",
|
|
"new_password": "Newpassword456!",
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert "success" in response.json()["message"].lower()
|
|
|
|
# Verify can login with new password
|
|
login_response = await async_client.post(
|
|
"/api/v1/auth/login",
|
|
json={"username": "pwchangeuser", "password": "Newpassword456!"},
|
|
)
|
|
assert login_response.status_code == 200
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_change_password_wrong_current(self, async_client: AsyncClient, user_token: str):
|
|
"""Verify changing password fails with wrong current password."""
|
|
response = await async_client.post(
|
|
"/api/v1/users/me/change-password",
|
|
headers={"Authorization": f"Bearer {user_token}"},
|
|
json={
|
|
"current_password": "wrongpassword",
|
|
"new_password": "Newpassword456!",
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
assert "incorrect" in response.json()["detail"].lower()
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_change_password_requires_auth(self, async_client: AsyncClient):
|
|
"""Verify changing password requires authentication."""
|
|
response = await async_client.post(
|
|
"/api/v1/users/me/change-password",
|
|
json={
|
|
"current_password": "oldpassword",
|
|
"new_password": "Strongpass456!",
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 401
|
|
|
|
|
|
class TestAuthMiddlewarePublicRoutes:
|
|
"""Tests for auth middleware public route configuration.
|
|
|
|
These routes must be accessible without authentication, even when auth is enabled,
|
|
because browser elements like <img src> and <video src> don't send Authorization headers.
|
|
"""
|
|
|
|
@pytest.fixture
|
|
async def enabled_auth(self, async_client: AsyncClient):
|
|
"""Enable auth for testing middleware behavior."""
|
|
await async_client.post(
|
|
"/api/v1/auth/setup",
|
|
json={
|
|
"auth_enabled": True,
|
|
"admin_username": "middlewareadmin",
|
|
"admin_password": "AdminPass1!",
|
|
},
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_auth_status_is_public(self, async_client: AsyncClient, enabled_auth):
|
|
"""Verify /api/v1/auth/status is accessible without auth."""
|
|
response = await async_client.get("/api/v1/auth/status")
|
|
assert response.status_code == 200
|
|
assert "auth_enabled" in response.json()
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_system_appliance_is_public(self, async_client: AsyncClient, enabled_auth):
|
|
"""Verify /api/v1/system/appliance is reachable without a JWT.
|
|
|
|
The SPA's i18n bootstrap fetches this BEFORE login to seed locale,
|
|
hostname, timezone, and NTP-gate state. The route handler has no
|
|
auth dependency, but the global auth_middleware blocks every
|
|
/api/ path not in PUBLIC_API_ROUTES — so without an explicit
|
|
allowlist entry the user sees a 401 in the browser console on
|
|
every page load.
|
|
"""
|
|
response = await async_client.get("/api/v1/system/appliance")
|
|
assert response.status_code == 200, response.text
|
|
body = response.json()
|
|
# Shape contract (no-auth surface):
|
|
for key in ("hostname", "timezone", "locale", "time_synced"):
|
|
assert key in body
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_auth_login_is_public(self, async_client: AsyncClient, enabled_auth):
|
|
"""Verify /api/v1/auth/login is accessible without auth."""
|
|
response = await async_client.post(
|
|
"/api/v1/auth/login",
|
|
json={"username": "middlewareadmin", "password": "AdminPass1!"},
|
|
)
|
|
# Should not return 401 (unauthorized) - it should either succeed or return
|
|
# a different error (like 400 for wrong credentials)
|
|
assert response.status_code != 401 or "token" in response.json()
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_auth_setup_is_public(self, async_client: AsyncClient):
|
|
"""Verify /api/v1/auth/setup is accessible without auth (needed for setup/recovery)."""
|
|
# Don't enable auth first - test that setup endpoint itself is accessible
|
|
response = await async_client.post(
|
|
"/api/v1/auth/setup",
|
|
json={"auth_enabled": False},
|
|
)
|
|
# Should not be 401
|
|
assert response.status_code != 401
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_updates_version_is_public(self, async_client: AsyncClient, enabled_auth):
|
|
"""Verify /api/v1/updates/version is accessible without auth."""
|
|
response = await async_client.get("/api/v1/updates/version")
|
|
# Should not be 401
|
|
assert response.status_code != 401
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_protected_route_requires_auth(self, async_client: AsyncClient, enabled_auth):
|
|
"""Verify non-public routes return 401 without token."""
|
|
response = await async_client.get("/api/v1/printers/")
|
|
assert response.status_code == 401
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_protected_route_works_with_token(self, async_client: AsyncClient, enabled_auth):
|
|
"""Verify non-public routes work with valid token."""
|
|
# Login to get token
|
|
login_response = await async_client.post(
|
|
"/api/v1/auth/login",
|
|
json={"username": "middlewareadmin", "password": "AdminPass1!"},
|
|
)
|
|
token = login_response.json()["access_token"]
|
|
|
|
# Access protected route
|
|
response = await async_client.get(
|
|
"/api/v1/printers/",
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_advanced_auth_status_is_public(self, async_client: AsyncClient, enabled_auth):
|
|
"""Verify /api/v1/auth/advanced-auth/status is accessible without auth."""
|
|
response = await async_client.get("/api/v1/auth/advanced-auth/status")
|
|
# Should not be 401 (must be accessible for login page)
|
|
assert response.status_code != 401
|
|
# Should return valid response (200 with auth status)
|
|
if response.status_code == 200:
|
|
result = response.json()
|
|
assert "advanced_auth_enabled" in result
|
|
assert "smtp_configured" in result
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_forgot_password_is_public(self, async_client: AsyncClient, enabled_auth):
|
|
"""Verify /api/v1/auth/forgot-password is accessible without auth."""
|
|
response = await async_client.post(
|
|
"/api/v1/auth/forgot-password",
|
|
json={"email": "test@example.com"},
|
|
)
|
|
# Should not be 401 (must be accessible for password reset from login page)
|
|
assert response.status_code != 401
|
|
# 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
|