mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-08-11 00:30:12 -04:00
7c117dc6bc
Archives, the queue and statistics report ownership as a numeric created_by_id, and statistics accept it as a filter, but nothing let an API key discover whose id was whose -- the only user listing returns emails, roles, group membership and full permission sets, so it is administrative and rejects keys. Add GET /users/slim returning id + username only, gated on a new users:read_slim permission mapped to can_read_status. That grants no data a key could not already reach: for API-keyed requests the permission deps return None as current_user, so the stats:filter_by_user guard short-circuits and ?created_by_id=N is already honoured for every N. What was missing was the ability to address the filter, not permission to use it. The full listing stays unmapped = admin-only. Also fix /auth/me, which answered an API key with a synthetic administrator: id 0, role admin, is_admin true and every permission in the enum. A key cannot reach an administrative route at all, so clients building their UI from that response rendered actions that 403 on use. It now reports the key owner's identity, is_admin false, and the permissions the key's scopes actually admit. Ownerless legacy keys keep id 0 but no longer claim admin. --- Source user names from the slim listing where only names are needed (#1894) Stats filter-by-user, the Archives print log filter, the File Manager username autocomplete, the camera-token owner column and the Finance member picker all render nothing but a username, but all of them read the full user listing, which is gated on the admin-level users:read. An operator granted stats:filter_by_user but not users:read got an empty filter with no indication why. Point them at /users/slim under a separate react-query key, since the full listing shares the 'users' key and the two shapes would clobber each other in the cache.
1136 lines
43 KiB
Python
1136 lines
43 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_ownerless_api_key_bearer(self, async_client: AsyncClient, db_session):
|
|
"""A legacy key has no identity to report, but no longer claims admin (#1894)."""
|
|
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 False
|
|
assert result["is_active"] is True
|
|
# can_read_status defaults True, so the scope-derived set is non-empty
|
|
# -- but it is a set, not "every permission there is".
|
|
assert len(result["permissions"]) > 0
|
|
assert "users:create" not in result["permissions"]
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_me_with_ownerless_api_key_header(self, async_client: AsyncClient, db_session):
|
|
"""Same as above via the X-API-Key header rather than Bearer."""
|
|
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 False
|
|
|
|
@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
|
|
|
|
async def _owned_key(self, async_client: AsyncClient, db_session, **scopes):
|
|
"""Set up auth and return (owner, full_key) for a key with ``scopes``.
|
|
|
|
The owner is given an email and a group explicitly rather than relying
|
|
on what /auth/setup happens to seed, so the assertions about what /me
|
|
withholds cannot pass vacuously.
|
|
"""
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import selectinload
|
|
|
|
from backend.app.core.auth import generate_api_key
|
|
from backend.app.models.api_key import APIKey
|
|
from backend.app.models.group import Group
|
|
from backend.app.models.user import User
|
|
|
|
await async_client.post(
|
|
"/api/v1/auth/setup",
|
|
json={
|
|
"auth_enabled": True,
|
|
"admin_username": "keyowner",
|
|
"admin_password": "KeyPass1!",
|
|
},
|
|
)
|
|
owner = (
|
|
await db_session.execute(select(User).where(User.username == "keyowner").options(selectinload(User.groups)))
|
|
).scalar_one()
|
|
owner.email = "keyowner@example.invalid"
|
|
group = Group(name="key-owner-group", description="t", permissions=["printers:read"], is_system=False)
|
|
db_session.add(group)
|
|
await db_session.flush()
|
|
owner.groups.append(group)
|
|
|
|
full_key, key_hash, key_prefix = generate_api_key()
|
|
db_session.add(
|
|
APIKey(name="owned", key_hash=key_hash, key_prefix=key_prefix, enabled=True, user_id=owner.id, **scopes)
|
|
)
|
|
await db_session.commit()
|
|
return owner, full_key
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_me_reports_the_key_owner_not_a_synthetic_admin(self, async_client: AsyncClient, db_session):
|
|
"""The id is the point of #1894 -- it is what created_by_id filters on."""
|
|
owner, full_key = await self._owned_key(async_client, db_session)
|
|
|
|
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"] == owner.id
|
|
assert result["username"] == "keyowner"
|
|
# The owner is an admin; the key still is not, because no key reaches
|
|
# an administrative route regardless of who owns it.
|
|
assert result["is_admin"] is False
|
|
assert result["role"] != "admin"
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_me_withholds_owner_email_and_groups(self, async_client: AsyncClient, db_session):
|
|
"""Identity, not the owner's profile -- anyone holding the key sees this."""
|
|
owner, full_key = await self._owned_key(async_client, db_session)
|
|
assert owner.email is not None and owner.groups # the helper made both non-empty
|
|
|
|
result = (await async_client.get("/api/v1/auth/me", headers={"X-API-Key": full_key})).json()
|
|
|
|
assert result["email"] is None
|
|
assert result["groups"] == []
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_me_permissions_track_the_key_scopes_not_the_owner(self, async_client: AsyncClient, db_session):
|
|
"""A key owned by an admin still reports only what its flags allow."""
|
|
_, full_key = await self._owned_key(
|
|
async_client,
|
|
db_session,
|
|
can_read_status=True,
|
|
can_control_printer=False,
|
|
can_queue=False,
|
|
)
|
|
|
|
perms = (await async_client.get("/api/v1/auth/me", headers={"X-API-Key": full_key})).json()["permissions"]
|
|
|
|
assert "printers:read" in perms # can_read_status
|
|
assert "printers:control" not in perms # can_control_printer is off
|
|
assert "queue:create" not in perms # can_queue is off
|
|
assert "users:create" not in perms # administrative: unmapped for keys
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_me_permissions_are_exactly_what_the_gate_admits(self, async_client: AsyncClient, db_session):
|
|
"""/me must not drift from _check_apikey_permissions.
|
|
|
|
The whole defect in #1894 was a /me response that described a different
|
|
credential than the one the gate enforces, so pin them to each other
|
|
rather than to a hand-written list that can rot. The owner is threaded
|
|
through both sides for the same reason -- the gate narrows to the
|
|
owner's permissions, so a check that skipped the owner would stop
|
|
catching drift the moment the owner is not an administrator.
|
|
"""
|
|
from fastapi import HTTPException
|
|
from sqlalchemy import select
|
|
|
|
from backend.app.core.auth import _check_apikey_permissions, resolve_apikey_owner
|
|
from backend.app.core.permissions import ALL_PERMISSIONS
|
|
from backend.app.models.api_key import APIKey
|
|
|
|
_, full_key = await self._owned_key(async_client, db_session, can_read_status=True, can_control_printer=False)
|
|
|
|
response = await async_client.get("/api/v1/auth/me", headers={"X-API-Key": full_key})
|
|
reported = set(response.json()["permissions"])
|
|
|
|
key = (await db_session.execute(select(APIKey).where(APIKey.name == "owned"))).scalar_one()
|
|
owner = await resolve_apikey_owner(db_session, key)
|
|
for perm in ALL_PERMISSIONS:
|
|
try:
|
|
_check_apikey_permissions(key, [perm], owner=owner)
|
|
except HTTPException:
|
|
assert perm not in reported, f"/me reports '{perm}' but the gate denies it"
|
|
else:
|
|
assert perm in reported, f"the gate admits '{perm}' but /me omits it"
|
|
|
|
|
|
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
|