""" Tests for the admin API endpoints (/auth/api/admin/). These tests cover: - Organization management (CRUD) - Role management (CRUD) - User management within orgs - Permission management - Exception handlers - Session management - Credential management """ import os import secrets from datetime import UTC, datetime from urllib.parse import urlsplit from uuid import UUID import httpx import pytest import pytest_asyncio import uuid7 from paskia import db, domains from paskia.db import ( Credential, Org, Permission, Role, User, add_permission_to_org, create_credential, create_org, create_permission, create_role, create_user, ) from paskia.db.operations import DB from paskia.util.crypto import hash_secret from tests.conftest import auth_headers, create_test_image_bytes, create_test_session # -------------------- Additional Fixtures -------------------- @pytest_asyncio.fixture(scope="function") async def second_org(test_db: DB) -> Org: """Create a second organization for deletion tests.""" org = Org.create( display_name="Second Organization", ) create_org(org) return org @pytest_asyncio.fixture(scope="function") async def second_org_role( test_db: DB, second_org: Org, admin_permission: Permission ) -> Role: """Create a role in the second org with admin permission.""" role = Role.create( org=second_org.uuid, display_name="Second Org Admin Role", permissions={admin_permission.uuid}, ) create_role(role) return role @pytest_asyncio.fixture(scope="function") async def second_org_user(test_db: DB, second_org_role: Role) -> User: """Create a user in the second org.""" user = User.create( display_name="Second Org User", role=second_org_role.uuid, ) create_user(user) return user @pytest_asyncio.fixture(scope="function") async def second_org_credential(test_db: DB, second_org_user: User) -> Credential: """Create a credential for the second org user.""" credential = Credential.create( credential_id=os.urandom(32), user=second_org_user.uuid, aaguid=UUID("00000000-0000-0000-0000-000000000000"), public_key=os.urandom(64), sign_count=0, rp_id="localhost", ) create_credential(credential) return credential @pytest_asyncio.fixture(scope="function") async def second_org_session_token( test_db: DB, second_org_user: User, second_org_credential: Credential ) -> str: """Create a session for the second org admin user.""" _db_key, secret = create_test_session( user_uuid=second_org_user.uuid, credential_uuid=second_org_credential.uuid, ) return secret @pytest_asyncio.fixture(scope="function") async def org_admin_role( test_db: DB, test_org: Org, org_admin_permission: Permission ) -> Role: """Create a role with org admin permission only (no global admin).""" role = Role.create( org=test_org.uuid, display_name="Org Admin Role", permissions={org_admin_permission.uuid}, ) create_role(role) return role @pytest_asyncio.fixture(scope="function") async def org_admin_user(test_db: DB, org_admin_role: Role) -> User: """Create a user with org admin permission only.""" user = User.create( display_name="Org Admin User", role=org_admin_role.uuid, ) user.visits = 5 user.last_seen = datetime.now(UTC) create_user(user) return user @pytest_asyncio.fixture(scope="function") async def org_admin_credential(test_db: DB, org_admin_user: User) -> Credential: """Create a credential for the org admin user.""" credential = Credential.create( credential_id=os.urandom(32), user=org_admin_user.uuid, aaguid=UUID("00000000-0000-0000-0000-000000000000"), public_key=os.urandom(64), sign_count=0, rp_id="localhost", ) create_credential(credential) return credential @pytest_asyncio.fixture(scope="function") async def org_admin_session_token( test_db: DB, org_admin_user: User, org_admin_credential: Credential ) -> str: """Create a session for the org admin user.""" _db_key, secret = create_test_session( user_uuid=org_admin_user.uuid, credential_uuid=org_admin_credential.uuid, ) return secret @pytest_asyncio.fixture(scope="function") async def grantable_permission(test_db: DB, test_org: Org) -> Permission: """Create a permission and add it to org's grantable permissions.""" perm = Permission.create(scope="test:grantable:perm", display_name="Grantable Perm") create_permission(perm) # Add to org's grantable permissions add_permission_to_org(test_org.uuid, perm.uuid) return perm # -------------------- Exception Handler Tests -------------------- class TestExceptionHandlers: """Tests for admin app exception handlers""" @pytest.mark.asyncio async def test_auth_exception_handler(self, client: httpx.AsyncClient): """AuthException should return proper JSON with auth info.""" # Accessing admin without auth triggers AuthException response = await client.get("/auth/api/admin/info") assert response.status_code == 401 data = response.json() assert "detail" in data assert "auth" in data assert data["auth"]["mode"] == "login" assert "iframe" in data["auth"] # -------------------- Organization Tests -------------------- class TestAdminOrganizations: """Tests for admin organization endpoints""" @pytest.mark.asyncio async def test_list_orgs_requires_auth(self, client: httpx.AsyncClient): """List orgs without auth should return 401.""" response = await client.get("/auth/api/admin/info") assert response.status_code == 401 @pytest.mark.asyncio async def test_list_orgs_requires_admin_permission( self, client: httpx.AsyncClient, regular_session_token: str ): """List orgs without admin permission should return 403.""" response = await client.get( "/auth/api/admin/info", headers={ **auth_headers(regular_session_token), "Host": "localhost:4401", }, ) assert response.status_code == 403 @pytest.mark.asyncio async def test_list_orgs_with_admin( self, client: httpx.AsyncClient, session_token: str, test_org ): """Admin user should be able to list organizations.""" response = await client.get( "/auth/api/admin/info", headers={**auth_headers(session_token), "Host": "localhost:4401"}, ) assert response.status_code == 200 data = response.json() assert isinstance(data, dict) assert "orgs" in data orgs_data = data["orgs"] assert isinstance(orgs_data, dict) assert len(orgs_data) >= 1 # Check org structure org_data = list(orgs_data.values())[0] assert "org" in org_data org = org_data["org"] assert "uuid" in org assert "display_name" in org assert "roles" in org_data assert "users" in org_data @pytest.mark.asyncio async def test_list_orgs_includes_user_avatar_urls( self, client: httpx.AsyncClient, session_token: str, test_org, test_user, tmp_path, monkeypatch, ): """Admin org payload should include canonical avatar URLs for listed users.""" upload = await client.put( f"/auth/api/user/{test_user.uuid}/profile.webp", files={"file": ("avatar.webp", create_test_image_bytes(), "image/webp")}, headers={**auth_headers(session_token), "Host": "localhost:4401"}, ) assert upload.status_code == 200 response = await client.get( "/auth/api/admin/info", headers={**auth_headers(session_token), "Host": "localhost:4401"}, ) assert response.status_code == 200 data = response.json() listed_user = data["orgs"][str(test_org.uuid)]["users"][str(test_user.uuid)] parts = urlsplit(listed_user["avatar_url"]) assert parts.path.endswith(f"/auth/api/user/{test_user.uuid}/profile.webp") assert parts.query == "" @pytest.mark.asyncio async def test_list_orgs_with_org_admin( self, client: httpx.AsyncClient, org_admin_session_token: str, test_org, second_org, ): """Org admin should only see their own organization.""" response = await client.get( "/auth/api/admin/info", headers={**auth_headers(org_admin_session_token), "Host": "localhost:4401"}, ) assert response.status_code == 200 data = response.json() # Should only see their own org, not the second org org_uuids = [org_data["org"]["uuid"] for org_data in data["orgs"].values()] assert str(test_org.uuid) in org_uuids @pytest.mark.asyncio async def test_create_org_requires_admin( self, client: httpx.AsyncClient, regular_session_token: str ): """Creating org without admin permission should fail.""" response = await client.post( "/auth/api/admin/orgs/", json={"display_name": "New Org"}, headers={ **auth_headers(regular_session_token), "Host": "localhost:4401", }, ) assert response.status_code == 403 @pytest.mark.asyncio async def test_create_org_success( self, client: httpx.AsyncClient, session_token: str ): """Admin should be able to create a new organization.""" response = await client.post( "/auth/api/admin/orgs/", json={"display_name": "New Test Org", "permissions": []}, headers={**auth_headers(session_token), "Host": "localhost:4401"}, ) assert response.status_code == 200 data = response.json() assert "uuid" in data @pytest.mark.asyncio async def test_create_org_with_defaults( self, client: httpx.AsyncClient, session_token: str ): """Admin should be able to create org with default values.""" response = await client.post( "/auth/api/admin/orgs/", json={}, # No display_name or permissions headers={**auth_headers(session_token), "Host": "localhost:4401"}, ) assert response.status_code == 200 data = response.json() assert "uuid" in data @pytest.mark.asyncio async def test_update_org( self, client: httpx.AsyncClient, session_token: str, test_org ): """Admin should be able to update an organization.""" response = await client.patch( f"/auth/api/admin/orgs/{test_org.uuid}", json={"display_name": "Updated Org Name"}, headers={**auth_headers(session_token), "Host": "localhost:4401"}, ) assert response.status_code == 200 data = response.json() assert data["status"] == "ok" @pytest.mark.asyncio async def test_update_org_with_org_admin( self, client: httpx.AsyncClient, org_admin_session_token: str, test_org, ): """Org admin should be able to update their organization.""" response = await client.patch( f"/auth/api/admin/orgs/{test_org.uuid}", json={ "display_name": "Org Admin Updated Name", }, headers={**auth_headers(org_admin_session_token), "Host": "localhost:4401"}, ) assert response.status_code == 200 data = response.json() assert data["status"] == "ok" @pytest.mark.asyncio async def test_update_org_org_admin_cannot_remove_own_perm( self, client: httpx.AsyncClient, org_admin_session_token: str, test_org, test_db: DB, ): """Org admin cannot remove their org admin permission from org's permissions.""" # The auth:org:admin perm is already created and added by org_admin_permission fixture org_admin_perm = next( p for p in db.data().permissions.values() if p.scope == "auth:org:admin" ) # Try to remove org admin perm (this is validated server-side in the remove endpoint) response = await client.delete( f"/auth/api/admin/orgs/{test_org.uuid}/permission?permission_uuid={org_admin_perm.uuid}", headers={**auth_headers(org_admin_session_token), "Host": "localhost:4401"}, ) # This should fail because only global admin can remove perms from org assert response.status_code == 403 @pytest.mark.asyncio async def test_delete_org_own_org_fails( self, client: httpx.AsyncClient, session_token: str, test_org ): """Cannot delete the organization you belong to.""" response = await client.delete( f"/auth/api/admin/orgs/{test_org.uuid}", headers={**auth_headers(session_token), "Host": "localhost:4401"}, ) assert response.status_code == 400 data = response.json() assert "Cannot delete" in data["detail"] @pytest.mark.asyncio async def test_delete_org_success( self, client: httpx.AsyncClient, session_token: str, test_db: DB, ): """Admin should be able to delete another organization.""" # Create org to delete org_to_delete = Org.create( display_name="Org To Delete", ) create_org(org_to_delete) # Create some org-specific permissions to test cleanup org_perm = Permission.create( scope=f"test:org:{org_to_delete.uuid}:feature", display_name="Org Feature", ) create_permission(org_perm) response = await client.delete( f"/auth/api/admin/orgs/{org_to_delete.uuid}", headers={**auth_headers(session_token), "Host": "localhost:4401"}, ) assert response.status_code == 200 data = response.json() assert data["status"] == "ok" # -------------------- Organization Permission Tests -------------------- class TestAdminOrgPermissions: """Tests for managing permissions on organizations""" @pytest.mark.asyncio async def test_add_permission_to_org( self, client: httpx.AsyncClient, session_token: str, test_org ): """Admin should be able to add a permission to an org.""" # First create a permission perm = Permission.create(scope="test:org:addable", display_name="Addable") create_permission(perm) # Add it to the org response = await client.post( f"/auth/api/admin/orgs/{test_org.uuid}/permission?permission_uuid={perm.uuid}", headers={**auth_headers(session_token), "Host": "localhost:4401"}, ) assert response.status_code == 200 data = response.json() assert data["status"] == "ok" @pytest.mark.asyncio async def test_add_permission_to_org_requires_admin( self, client: httpx.AsyncClient, org_admin_session_token: str, test_org, ): """Org admin cannot add permissions to org (requires global admin).""" admin_perm = next( p for p in db.data().permissions.values() if p.scope == "auth:admin" ) response = await client.post( f"/auth/api/admin/orgs/{test_org.uuid}/permission?permission_uuid={admin_perm.uuid}", headers={**auth_headers(org_admin_session_token), "Host": "localhost:4401"}, ) assert response.status_code == 403 @pytest.mark.asyncio async def test_remove_permission_from_org( self, client: httpx.AsyncClient, session_token: str, test_org ): """Admin should be able to remove a permission from an org.""" # First create and add a permission perm = Permission.create(scope="test:org:removable", display_name="Removable") create_permission(perm) await client.post( f"/auth/api/admin/orgs/{test_org.uuid}/permission?permission_uuid={perm.uuid}", headers={**auth_headers(session_token), "Host": "localhost:4401"}, ) # Remove it response = await client.delete( f"/auth/api/admin/orgs/{test_org.uuid}/permission?permission_uuid={perm.uuid}", headers={**auth_headers(session_token), "Host": "localhost:4401"}, ) assert response.status_code == 200 data = response.json() assert data["status"] == "ok" @pytest.mark.asyncio async def test_remove_permission_from_org_requires_admin( self, client: httpx.AsyncClient, org_admin_session_token: str, test_org, ): """Org admin cannot remove permissions from org (requires global admin).""" admin_perm = next( p for p in db.data().permissions.values() if p.scope == "auth:admin" ) response = await client.delete( f"/auth/api/admin/orgs/{test_org.uuid}/permission?permission_uuid={admin_perm.uuid}", headers={**auth_headers(org_admin_session_token), "Host": "localhost:4401"}, ) assert response.status_code == 403 # -------------------- Role Tests -------------------- class TestAdminRoles: """Tests for admin role endpoints""" @pytest.mark.asyncio async def test_create_role_requires_admin( self, client: httpx.AsyncClient, regular_session_token: str, test_org ): """Creating role without admin permission should fail.""" response = await client.post( f"/auth/api/admin/orgs/{test_org.uuid}/roles", json={"display_name": "New Role"}, headers={ **auth_headers(regular_session_token), "Host": "localhost:4401", }, ) assert response.status_code == 403 @pytest.mark.asyncio async def test_create_role_success( self, client: httpx.AsyncClient, session_token: str, test_org ): """Admin should be able to create a new role.""" response = await client.post( f"/auth/api/admin/orgs/{test_org.uuid}/roles", json={"display_name": "Test Role", "permissions": []}, headers={**auth_headers(session_token), "Host": "localhost:4401"}, ) assert response.status_code == 200 data = response.json() assert "uuid" in data @pytest.mark.asyncio async def test_create_role_with_defaults( self, client: httpx.AsyncClient, session_token: str, test_org ): """Admin should be able to create role with default name.""" response = await client.post( f"/auth/api/admin/orgs/{test_org.uuid}/roles", json={}, headers={**auth_headers(session_token), "Host": "localhost:4401"}, ) assert response.status_code == 200 data = response.json() assert "uuid" in data @pytest.mark.asyncio async def test_create_role_with_grantable_permission( self, client: httpx.AsyncClient, session_token: str, test_org, grantable_permission, ): """Admin should be able to create role with grantable permissions.""" response = await client.post( f"/auth/api/admin/orgs/{test_org.uuid}/roles", json={ "display_name": "Role With Perms", "permissions": [str(grantable_permission.uuid)], }, headers={**auth_headers(session_token), "Host": "localhost:4401"}, ) assert response.status_code == 200 data = response.json() assert "uuid" in data @pytest.mark.asyncio async def test_create_role_with_non_grantable_permission( self, client: httpx.AsyncClient, session_token: str, test_org, test_db: DB, ): """Creating role with non-grantable permission should fail.""" # Create permission but don't add to org perm = Permission.create( scope="test:not:grantable", display_name="Not Grantable", ) create_permission(perm) response = await client.post( f"/auth/api/admin/orgs/{test_org.uuid}/roles", json={ "display_name": "Bad Role", "permissions": [str(perm.uuid)], }, headers={**auth_headers(session_token), "Host": "localhost:4401"}, ) assert response.status_code == 400 data = response.json() assert "not grantable" in data["detail"] @pytest.mark.asyncio async def test_update_role( self, client: httpx.AsyncClient, session_token: str, test_org, test_role ): """Admin should be able to update a role.""" response = await client.patch( f"/auth/api/admin/roles/{test_role.uuid}", json={"display_name": "Updated Role Name"}, headers={**auth_headers(session_token), "Host": "localhost:4401"}, ) assert response.status_code == 200 data = response.json() assert data["status"] == "ok" @pytest.mark.asyncio async def test_update_role_wrong_org( self, client: httpx.AsyncClient, org_admin_session_token: str, test_org, second_org_role, ): """Org admin cannot update role from another org.""" response = await client.patch( f"/auth/api/admin/roles/{second_org_role.uuid}", json={"display_name": "Try Update Wrong Org"}, headers={**auth_headers(org_admin_session_token), "Host": "localhost:4401"}, ) assert response.status_code == 403 @pytest.mark.asyncio async def test_update_role_add_grantable_permission( self, client: httpx.AsyncClient, session_token: str, test_org, user_role, grantable_permission, ): """Admin should be able to add grantable permissions to role.""" response = await client.post( f"/auth/api/admin/roles/{user_role.uuid}/permissions/{grantable_permission.uuid}", headers={**auth_headers(session_token), "Host": "localhost:4401"}, ) assert response.status_code == 200 @pytest.mark.asyncio async def test_update_role_add_non_grantable_permission( self, client: httpx.AsyncClient, session_token: str, test_org, user_role, test_db: DB, ): """Adding non-grantable permission to role should fail.""" perm = Permission.create( scope="test:not:grantable:update", display_name="Not Grantable", ) create_permission(perm) response = await client.post( f"/auth/api/admin/roles/{user_role.uuid}/permissions/{perm.uuid}", headers={**auth_headers(session_token), "Host": "localhost:4401"}, ) assert response.status_code == 400 data = response.json() assert "not grantable" in data["detail"] @pytest.mark.asyncio async def test_update_own_role_cannot_remove_admin( self, client: httpx.AsyncClient, session_token: str, test_org, test_role, admin_permission, org_admin_permission, ): """Admin cannot remove their own admin permissions.""" # test_role has both auth:admin and auth:org:admin # Remove auth:admin first (should succeed since org:admin remains) response = await client.delete( f"/auth/api/admin/roles/{test_role.uuid}/permissions/{admin_permission.uuid}", headers={**auth_headers(session_token), "Host": "localhost:4401"}, ) assert response.status_code == 200 # Now try to remove auth:org:admin (should fail - would leave no admin access) response = await client.delete( f"/auth/api/admin/roles/{test_role.uuid}/permissions/{org_admin_permission.uuid}", headers={**auth_headers(session_token), "Host": "localhost:4401"}, ) assert response.status_code == 400 data = response.json() assert "Cannot remove your own admin permissions" in data["detail"] @pytest.mark.asyncio async def test_delete_role( self, client: httpx.AsyncClient, session_token: str, test_org, user_role ): """Admin should be able to delete a role.""" response = await client.delete( f"/auth/api/admin/roles/{user_role.uuid}", headers={**auth_headers(session_token), "Host": "localhost:4401"}, ) assert response.status_code == 200 data = response.json() assert data["status"] == "ok" @pytest.mark.asyncio async def test_delete_role_wrong_org( self, client: httpx.AsyncClient, org_admin_session_token: str, test_org, second_org_role, ): """Org admin cannot delete role from another org.""" response = await client.delete( f"/auth/api/admin/roles/{second_org_role.uuid}", headers={**auth_headers(org_admin_session_token), "Host": "localhost:4401"}, ) assert response.status_code == 403 @pytest.mark.asyncio async def test_delete_own_role_fails( self, client: httpx.AsyncClient, session_token: str, test_org, test_role ): """Admin cannot delete their own role.""" response = await client.delete( f"/auth/api/admin/roles/{test_role.uuid}", headers={**auth_headers(session_token), "Host": "localhost:4401"}, ) assert response.status_code == 400 data = response.json() assert "Cannot delete your own role" in data["detail"] # -------------------- User Tests -------------------- class TestAdminUsersInOrg: """Tests for admin user management within organizations""" @pytest.mark.asyncio async def test_create_user_success( self, client: httpx.AsyncClient, session_token: str, test_org, user_role ): """Admin should be able to create a new user.""" response = await client.post( f"/auth/api/admin/orgs/{test_org.uuid}/users", json={"display_name": "New User", "role": user_role.display_name}, headers={**auth_headers(session_token), "Host": "localhost:4401"}, ) assert response.status_code == 200 data = response.json() assert "uuid" in data @pytest.mark.asyncio async def test_create_user_missing_fields( self, client: httpx.AsyncClient, session_token: str, test_org ): """Creating user without required fields should fail.""" response = await client.post( f"/auth/api/admin/orgs/{test_org.uuid}/users", json={}, headers={**auth_headers(session_token), "Host": "localhost:4401"}, ) assert response.status_code == 400 data = response.json() assert "required" in data["detail"] @pytest.mark.asyncio async def test_create_user_invalid_role( self, client: httpx.AsyncClient, session_token: str, test_org ): """Creating user with non-existent role should fail.""" response = await client.post( f"/auth/api/admin/orgs/{test_org.uuid}/users", json={"display_name": "New User", "role": "NonExistent Role"}, headers={**auth_headers(session_token), "Host": "localhost:4401"}, ) assert response.status_code == 400 data = response.json() assert "Role not found" in data["detail"] @pytest.mark.asyncio async def test_get_user_in_org( self, client: httpx.AsyncClient, session_token: str, test_org, test_user ): """Admin should be able to get user details within an org.""" response = await client.get( f"/auth/api/admin/users/{test_user.uuid}", headers={**auth_headers(session_token), "Host": "localhost:4401"}, ) assert response.status_code == 200 data = response.json() assert "user" in data assert "display_name" in data["user"] assert "credentials" in data assert "sessions" in data assert "aaguid_info" in data assert "org" in data assert "role" in data @pytest.mark.asyncio async def test_get_user_not_found( self, client: httpx.AsyncClient, session_token: str, test_org ): """Getting non-existent user should return 404.""" fake_uuid = uuid7.create() response = await client.get( f"/auth/api/admin/users/{fake_uuid}", headers={**auth_headers(session_token), "Host": "localhost:4401"}, ) assert response.status_code == 404 data = response.json() assert "User not found" in data["detail"] @pytest.mark.asyncio async def test_get_user_wrong_org( self, client: httpx.AsyncClient, org_admin_session_token: str, test_org, second_org_user, ): """Org admin cannot get user from another org.""" response = await client.get( f"/auth/api/admin/users/{second_org_user.uuid}", headers={**auth_headers(org_admin_session_token), "Host": "localhost:4401"}, ) assert response.status_code == 403 @pytest.mark.asyncio async def test_get_user_with_org_admin( self, client: httpx.AsyncClient, org_admin_session_token: str, test_org, org_admin_user, ): """Org admin should be able to get user details.""" response = await client.get( f"/auth/api/admin/users/{org_admin_user.uuid}", headers={**auth_headers(org_admin_session_token), "Host": "localhost:4401"}, ) assert response.status_code == 200 data = response.json() assert "user" in data assert "display_name" in data["user"] @pytest.mark.asyncio async def test_update_user_display_name_in_org( self, client: httpx.AsyncClient, session_token: str, test_org, test_user ): """Admin should be able to update user display name.""" response = await client.patch( f"/auth/api/admin/users/{test_user.uuid}/info", json={"display_name": "Updated Admin Name"}, headers={**auth_headers(session_token), "Host": "localhost:4401"}, ) assert response.status_code == 200 @pytest.mark.asyncio async def test_update_user_display_name_not_found( self, client: httpx.AsyncClient, session_token: str, test_org ): """Updating non-existent user should return 404.""" fake_uuid = uuid7.create() response = await client.patch( f"/auth/api/admin/users/{fake_uuid}/info", json={"display_name": "New Name"}, headers={**auth_headers(session_token), "Host": "localhost:4401"}, ) assert response.status_code == 404 data = response.json() assert "User not found" in data["detail"] @pytest.mark.asyncio async def test_update_user_display_name_wrong_org( self, client: httpx.AsyncClient, org_admin_session_token: str, test_org, second_org_user, ): """Org admin cannot update user from another org.""" response = await client.patch( f"/auth/api/admin/users/{second_org_user.uuid}/info", json={"display_name": "New Name"}, headers={**auth_headers(org_admin_session_token), "Host": "localhost:4401"}, ) assert response.status_code == 403 @pytest.mark.asyncio async def test_update_user_display_name_empty( self, client: httpx.AsyncClient, session_token: str, test_org, test_user ): """Updating user with empty display name should fail.""" response = await client.patch( f"/auth/api/admin/users/{test_user.uuid}/info", json={"display_name": " "}, headers={**auth_headers(session_token), "Host": "localhost:4401"}, ) assert response.status_code == 400 data = response.json() assert "display_name cannot be empty" in data["detail"] @pytest.mark.asyncio async def test_update_user_display_name_too_long( self, client: httpx.AsyncClient, session_token: str, test_org, test_user ): """Updating user with too long display name should fail.""" response = await client.patch( f"/auth/api/admin/users/{test_user.uuid}/info", json={"display_name": "x" * 65}, headers={**auth_headers(session_token), "Host": "localhost:4401"}, ) assert response.status_code == 400 data = response.json() assert "display_name too long" in data["detail"] @pytest.mark.asyncio async def test_admin_can_upload_user_avatar( self, client: httpx.AsyncClient, session_token: str, test_user: User, tmp_path, monkeypatch, ): """Admin should be able to upload avatar for a managed user.""" response = await client.put( f"/auth/api/user/{test_user.uuid}/profile.webp", files={"file": ("avatar.webp", create_test_image_bytes(), "image/webp")}, headers={**auth_headers(session_token), "Host": "localhost:4401"}, ) assert response.status_code == 200 detail = await client.get( f"/auth/api/admin/users/{test_user.uuid}", headers={**auth_headers(session_token), "Host": "localhost:4401"}, ) assert detail.status_code == 200 avatar_url = detail.json()["user"]["avatar_url"] parts = urlsplit(avatar_url) assert parts.path.endswith(f"/auth/api/user/{test_user.uuid}/profile.webp") @pytest.mark.asyncio async def test_update_user_role_in_org( self, client: httpx.AsyncClient, session_token: str, test_org, regular_user, user_role, ): """Admin should be able to change user's role within org.""" # Use regular_user who is in the same org but not the session owner response = await client.patch( f"/auth/api/admin/users/{regular_user.uuid}/role", json={"role_uuid": str(user_role.uuid)}, headers={**auth_headers(session_token), "Host": "localhost:4401"}, ) assert response.status_code == 200 @pytest.mark.asyncio async def test_update_user_role_missing_role( self, client: httpx.AsyncClient, session_token: str, test_org, test_user ): """Updating user role without specifying role should fail.""" response = await client.patch( f"/auth/api/admin/users/{test_user.uuid}/role", json={}, headers={**auth_headers(session_token), "Host": "localhost:4401"}, ) assert response.status_code == 400 data = response.json() assert "role_uuid is required" in data["detail"] @pytest.mark.asyncio async def test_update_user_role_user_not_found( self, client: httpx.AsyncClient, session_token: str, test_org ): """Updating role for non-existent user should fail.""" fake_uuid = uuid7.create() response = await client.patch( f"/auth/api/admin/users/{fake_uuid}/role", json={"role_uuid": str(uuid7.create())}, headers={**auth_headers(session_token), "Host": "localhost:4401"}, ) assert response.status_code == 404 data = response.json() assert "User not found" in data["detail"] @pytest.mark.asyncio async def test_update_user_role_wrong_org( self, client: httpx.AsyncClient, org_admin_session_token: str, test_org, second_org_user, ): """Org admin cannot update role for user in another org.""" response = await client.patch( f"/auth/api/admin/users/{second_org_user.uuid}/role", json={"role_uuid": str(uuid7.create())}, headers={**auth_headers(org_admin_session_token), "Host": "localhost:4401"}, ) assert response.status_code == 403 @pytest.mark.asyncio async def test_update_user_role_invalid_role( self, client: httpx.AsyncClient, session_token: str, test_org, test_user ): """Updating user to non-existent role should fail.""" response = await client.patch( f"/auth/api/admin/users/{test_user.uuid}/role", json={"role_uuid": str(uuid7.create())}, headers={**auth_headers(session_token), "Host": "localhost:4401"}, ) assert response.status_code == 400 data = response.json() assert "Role not found" in data["detail"] @pytest.mark.asyncio async def test_update_own_role_to_non_admin_fails( self, client: httpx.AsyncClient, org_admin_session_token: str, test_org, org_admin_user, user_role, ): """Admin cannot change their own role to non-admin role.""" response = await client.patch( f"/auth/api/admin/users/{org_admin_user.uuid}/role", json={"role_uuid": str(user_role.uuid)}, headers={**auth_headers(org_admin_session_token), "Host": "localhost:4401"}, ) assert response.status_code == 400 data = response.json() assert "without admin permissions" in data["detail"] @pytest.mark.asyncio async def test_update_own_role_to_admin_role_succeeds( self, client: httpx.AsyncClient, session_token: str, test_org, test_user, test_role, ): """Admin can change their own role to another admin role.""" # test_user is already on test_role which has auth:admin # Changing to the same role should succeed (no permission loss) response = await client.patch( f"/auth/api/admin/users/{test_user.uuid}/role", json={"role_uuid": str(test_role.uuid)}, headers={**auth_headers(session_token), "Host": "localhost:4401"}, ) assert response.status_code == 200 @pytest.mark.asyncio async def test_create_user_reset_link( self, client: httpx.AsyncClient, session_token: str, test_org, test_user ): """Admin should be able to create reset links for users.""" response = await client.post( f"/auth/api/admin/users/{test_user.uuid}/create-link", headers={**auth_headers(session_token), "Host": "localhost:4401"}, ) assert response.status_code == 200 data = response.json() assert "url" in data assert "expires" in data @pytest.mark.asyncio async def test_create_user_reset_link_not_found( self, client: httpx.AsyncClient, session_token: str, test_org ): """Creating reset link for non-existent user should fail.""" fake_uuid = uuid7.create() response = await client.post( f"/auth/api/admin/users/{fake_uuid}/create-link", headers={**auth_headers(session_token), "Host": "localhost:4401"}, ) assert response.status_code == 404 data = response.json() assert "User not found" in data["detail"] @pytest.mark.asyncio async def test_create_user_reset_link_wrong_org( self, client: httpx.AsyncClient, org_admin_session_token: str, test_org, second_org_user, ): """Org admin cannot create reset link for user in another org.""" response = await client.post( f"/auth/api/admin/users/{second_org_user.uuid}/create-link", headers={**auth_headers(org_admin_session_token), "Host": "localhost:4401"}, ) assert response.status_code == 403 @pytest.mark.asyncio async def test_create_user_registration_link_without_credentials( self, client: httpx.AsyncClient, session_token: str, test_org, user_role, test_db: DB, ): """Creating link for user without credentials should return registration link.""" # Create user without credentials user_no_cred = User.create( display_name="User Without Creds", role=user_role.uuid, ) create_user(user_no_cred) response = await client.post( f"/auth/api/admin/users/{user_no_cred.uuid}/create-link", headers={**auth_headers(session_token), "Host": "localhost:4401"}, ) assert response.status_code == 200 data = response.json() assert "url" in data # -------------------- User Deletion Tests -------------------- class TestAdminUserDeletion: """Tests for admin user deletion""" @pytest.mark.asyncio async def test_delete_user_success( self, client: httpx.AsyncClient, session_token: str, test_org, user_role, test_db: DB, ): """Admin should be able to delete a user.""" # Create a user to delete user_to_delete = User.create( display_name="User To Delete", role=user_role.uuid, ) create_user(user_to_delete) response = await client.delete( f"/auth/api/admin/users/{user_to_delete.uuid}", headers={**auth_headers(session_token), "Host": "localhost:4401"}, ) assert response.status_code == 200 data = response.json() assert data["status"] == "ok" # Verify user is actually deleted assert user_to_delete.uuid not in db.data().users @pytest.mark.asyncio async def test_delete_user_not_found( self, client: httpx.AsyncClient, session_token: str ): """Deleting non-existent user should return 404.""" fake_uuid = uuid7.create() response = await client.delete( f"/auth/api/admin/users/{fake_uuid}", headers={**auth_headers(session_token), "Host": "localhost:4401"}, ) assert response.status_code == 404 data = response.json() assert "User not found" in data["detail"] @pytest.mark.asyncio async def test_delete_own_user_fails( self, client: httpx.AsyncClient, session_token: str, test_user ): """Admin cannot delete their own account.""" response = await client.delete( f"/auth/api/admin/users/{test_user.uuid}", headers={**auth_headers(session_token), "Host": "localhost:4401"}, ) assert response.status_code == 400 data = response.json() assert "Cannot delete your own account" in data["detail"] @pytest.mark.asyncio async def test_delete_user_wrong_org( self, client: httpx.AsyncClient, org_admin_session_token: str, second_org_user, ): """Org admin cannot delete user from another org.""" response = await client.delete( f"/auth/api/admin/users/{second_org_user.uuid}", headers={**auth_headers(org_admin_session_token), "Host": "localhost:4401"}, ) assert response.status_code == 403 @pytest.mark.asyncio async def test_delete_user_org_admin_success( self, client: httpx.AsyncClient, org_admin_session_token: str, test_org, user_role, test_db: DB, ): """Org admin should be able to delete users in their org.""" # Create a user in the same org to delete user_to_delete = User.create( display_name="Org User To Delete", role=user_role.uuid, ) create_user(user_to_delete) response = await client.delete( f"/auth/api/admin/users/{user_to_delete.uuid}", headers={**auth_headers(org_admin_session_token), "Host": "localhost:4401"}, ) assert response.status_code == 200 data = response.json() assert data["status"] == "ok" @pytest.mark.asyncio async def test_delete_user_regular_user_forbidden( self, client: httpx.AsyncClient, regular_session_token: str, test_user, ): """Regular user trying to delete user should get 403.""" response = await client.delete( f"/auth/api/admin/users/{test_user.uuid}", headers={**auth_headers(regular_session_token), "Host": "localhost:4401"}, ) assert response.status_code == 403 # -------------------- Credential Tests -------------------- class TestAdminCredentials: """Tests for admin credential management""" @pytest.mark.asyncio async def test_delete_user_credential( self, client: httpx.AsyncClient, session_token: str, test_org, test_user, test_credential, ): """Admin should be able to delete a user's credential.""" response = await client.delete( f"/auth/api/admin/users/{test_user.uuid}/credentials/{test_credential.uuid}", headers={**auth_headers(session_token), "Host": "localhost:4401"}, ) assert response.status_code == 200 data = response.json() assert data["status"] == "ok" @pytest.mark.asyncio async def test_delete_credential_user_not_found( self, client: httpx.AsyncClient, session_token: str, test_org ): """Deleting credential for non-existent user should fail.""" fake_user_uuid = uuid7.create() fake_cred_uuid = uuid7.create() response = await client.delete( f"/auth/api/admin/users/{fake_user_uuid}/credentials/{fake_cred_uuid}", headers={**auth_headers(session_token), "Host": "localhost:4401"}, ) assert response.status_code == 404 data = response.json() assert "User not found" in data["detail"] @pytest.mark.asyncio async def test_delete_credential_wrong_org( self, client: httpx.AsyncClient, org_admin_session_token: str, test_org, second_org_user, second_org_credential, ): """Org admin cannot delete credential for user in another org.""" response = await client.delete( f"/auth/api/admin/users/{second_org_user.uuid}/credentials/{second_org_credential.uuid}", headers={**auth_headers(org_admin_session_token), "Host": "localhost:4401"}, ) assert response.status_code == 403 # -------------------- Session Tests -------------------- class TestAdminSessions: """Tests for admin session management""" @pytest.mark.asyncio async def test_delete_user_session( self, client: httpx.AsyncClient, session_token: str, test_org, test_user, test_credential, test_db: DB, ): """Admin should be able to delete a user's session.""" # Create an additional session to delete extra_db_key, _extra_secret = create_test_session( user_uuid=test_user.uuid, credential_uuid=test_credential.uuid, host="other.host:4401", ip="192.168.1.1", user_agent="other-agent", ) response = await client.delete( f"/auth/api/admin/users/{test_user.uuid}/sessions/{extra_db_key}", headers={**auth_headers(session_token), "Host": "localhost:4401"}, ) assert response.status_code == 200 data = response.json() assert data["status"] == "ok" assert data["current_session_terminated"] is False @pytest.mark.asyncio async def test_delete_own_session( self, client: httpx.AsyncClient, session_token: str, test_org, test_user, ): """Admin can delete their own current session.""" session_db_key = hash_secret("cookie", session_token) response = await client.delete( f"/auth/api/admin/users/{test_user.uuid}/sessions/{session_db_key}", headers={**auth_headers(session_token), "Host": "localhost:4401"}, ) assert response.status_code == 200 data = response.json() assert data["current_session_terminated"] is True @pytest.mark.asyncio async def test_delete_session_user_not_found( self, client: httpx.AsyncClient, session_token: str, test_org ): """Deleting session for non-existent user should fail.""" fake_uuid = uuid7.create() response = await client.delete( f"/auth/api/admin/users/{fake_uuid}/sessions/fake-session-id", headers={**auth_headers(session_token), "Host": "localhost:4401"}, ) assert response.status_code == 404 data = response.json() assert "User not found" in data["detail"] @pytest.mark.asyncio async def test_delete_session_wrong_org( self, client: httpx.AsyncClient, org_admin_session_token: str, test_org, second_org_user, ): """Org admin cannot delete session for user in another org.""" response = await client.delete( f"/auth/api/admin/users/{second_org_user.uuid}/sessions/fake-session", headers={**auth_headers(org_admin_session_token), "Host": "localhost:4401"}, ) assert response.status_code == 403 @pytest.mark.asyncio async def test_delete_session_invalid_id( self, client: httpx.AsyncClient, session_token: str, test_org, test_user ): """Deleting session with invalid/non-existent ID should fail.""" response = await client.delete( f"/auth/api/admin/users/{test_user.uuid}/sessions/invalid!!id", headers={**auth_headers(session_token), "Host": "localhost:4401"}, ) assert response.status_code == 404 data = response.json() assert "Session not found" in data["detail"] @pytest.mark.asyncio async def test_delete_session_not_found( self, client: httpx.AsyncClient, session_token: str, test_org, test_user ): """Deleting non-existent session should fail.""" # Use a valid format but non-existent key fake_token = secrets.token_urlsafe(12) response = await client.delete( f"/auth/api/admin/users/{test_user.uuid}/sessions/{fake_token}", headers={**auth_headers(session_token), "Host": "localhost:4401"}, ) assert response.status_code == 404 data = response.json() assert "Session not found" in data["detail"] # -------------------- Permission Tests -------------------- class TestAdminPermissions: """Tests for admin permission management""" @pytest.mark.asyncio async def test_list_permissions( self, client: httpx.AsyncClient, session_token: str ): """Admin should be able to list all permissions.""" response = await client.get( "/auth/api/admin/info", headers={**auth_headers(session_token), "Host": "localhost:4401"}, ) assert response.status_code == 200 data = response.json() assert isinstance(data, dict) assert "permissions" in data permissions_data = data["permissions"] assert isinstance(permissions_data, dict) # Should include at least auth:admin perm_scopes = [p["scope"] for p in permissions_data.values()] assert "auth:admin" in perm_scopes @pytest.mark.asyncio async def test_list_permissions_org_admin( self, client: httpx.AsyncClient, org_admin_session_token: str, test_org, grantable_permission, ): """Org admin should only see permissions their org can grant.""" response = await client.get( "/auth/api/admin/info", headers={**auth_headers(org_admin_session_token), "Host": "localhost:4401"}, ) assert response.status_code == 200 data = response.json() # Should only see permissions the org can grant perm_scopes = [p["scope"] for p in data["permissions"].values()] assert grantable_permission.scope in perm_scopes # test_org CAN grant auth:admin (it's in org.permissions), so org admin sees it assert "auth:admin" in perm_scopes # Should also see auto-created org admin permission assert "auth:org:admin" in perm_scopes @pytest.mark.asyncio async def test_create_permission( self, client: httpx.AsyncClient, session_token: str ): """Admin should be able to create new permissions.""" response = await client.post( "/auth/api/admin/permissions/", json={"scope": "test:create:permission", "display_name": "Test Permission"}, headers={**auth_headers(session_token), "Host": "localhost:4401"}, ) assert response.status_code == 200 data = response.json() assert data["status"] == "ok" @pytest.mark.asyncio async def test_create_permission_missing_fields( self, client: httpx.AsyncClient, session_token: str ): """Creating permission without required fields should fail.""" response = await client.post( "/auth/api/admin/permissions/", json={}, headers={**auth_headers(session_token), "Host": "localhost:4401"}, ) assert response.status_code == 400 data = response.json() assert "required" in data["detail"] @pytest.mark.asyncio async def test_create_permission_requires_admin( self, client: httpx.AsyncClient, regular_session_token: str ): """Creating permission without admin should fail.""" response = await client.post( "/auth/api/admin/permissions/", json={"scope": "test:forbidden", "display_name": "Forbidden"}, headers={ **auth_headers(regular_session_token), "Host": "localhost:4401", }, ) assert response.status_code == 403 @pytest.mark.asyncio async def test_update_permission( self, client: httpx.AsyncClient, session_token: str, test_db: DB ): """Admin should be able to update a permission.""" # Create permission first perm = Permission.create(scope="test:updateable", display_name="Updateable") create_permission(perm) response = await client.patch( f"/auth/api/admin/permissions/{perm.uuid}?display_name=Updated%20Name", headers={**auth_headers(session_token), "Host": "localhost:4401"}, ) assert response.status_code == 200 data = response.json() assert data["status"] == "ok" @pytest.mark.asyncio async def test_update_permission_empty_name( self, client: httpx.AsyncClient, session_token: str, test_db: DB ): """Updating permission with empty name should fail.""" # Create permission first perm = Permission.create(scope="test:perm", display_name="Test Perm") create_permission(perm) response = await client.patch( f"/auth/api/admin/permissions/{perm.uuid}?display_name=", headers={**auth_headers(session_token), "Host": "localhost:4401"}, ) assert response.status_code == 400 data = response.json() assert "display_name is required" in data["detail"] @pytest.mark.asyncio async def test_update_permission_scope( self, client: httpx.AsyncClient, session_token: str, test_db: DB ): """Admin should be able to update a permission's scope via PATCH.""" # Create permission first perm = Permission.create(scope="test:renameable2", display_name="Renameable") create_permission(perm) response = await client.patch( f"/auth/api/admin/permissions/{perm.uuid}?scope=test:renamed2", headers={**auth_headers(session_token), "Host": "localhost:4401"}, ) assert response.status_code == 200 @pytest.mark.asyncio async def test_update_permission_auth_admin_scope_fails( self, client: httpx.AsyncClient, session_token: str ): """Cannot change the auth:admin permission scope.""" # Get the auth:admin permission perms = list(db.data().permissions.values()) admin_perm = next(p for p in perms if p.scope == "auth:admin") response = await client.patch( f"/auth/api/admin/permissions/{admin_perm.uuid}?scope=auth:superadmin", headers={**auth_headers(session_token), "Host": "localhost:4401"}, ) assert response.status_code == 400 data = response.json() assert "Cannot rename the master admin" in data["detail"] @pytest.mark.asyncio async def test_update_permission_scope_and_display_name( self, client: httpx.AsyncClient, session_token: str, test_db: DB ): """Updating permission can change scope and display name together.""" perm = Permission.create(scope="test:rename:withname", display_name="Old Name") create_permission(perm) response = await client.patch( f"/auth/api/admin/permissions/{perm.uuid}?scope=test:renamed:withname&display_name=New%20Display%20Name", headers={**auth_headers(session_token), "Host": "localhost:4401"}, ) assert response.status_code == 200 @pytest.mark.asyncio async def test_delete_permission( self, client: httpx.AsyncClient, session_token: str, test_db: DB ): """Admin should be able to delete a permission.""" # Create permission first perm = Permission.create(scope="test:deleteable", display_name="Deleteable") create_permission(perm) response = await client.delete( f"/auth/api/admin/permissions/{perm.uuid}", headers={**auth_headers(session_token), "Host": "localhost:4401"}, ) assert response.status_code == 200 data = response.json() assert data["status"] == "ok" @pytest.mark.asyncio async def test_delete_permission_auth_admin_last_one_fails( self, client: httpx.AsyncClient, session_token: str ): """Cannot delete the only auth:admin permission (would lock out admin).""" # Get the auth:admin permission perms = list(db.data().permissions.values()) admin_perm = next(p for p in perms if p.scope == "auth:admin") response = await client.delete( f"/auth/api/admin/permissions/{admin_perm.uuid}", headers={**auth_headers(session_token), "Host": "localhost:4401"}, ) assert response.status_code == 400 data = response.json() assert "lock you out of admin access" in data["detail"] @pytest.mark.asyncio async def test_delete_permission_auth_admin_with_another_succeeds( self, client: httpx.AsyncClient, session_token: str, test_db: DB ): """Can delete an auth:admin permission if another accessible one exists.""" # Create a second auth:admin permission (no domain restriction) perm2 = Permission.create(scope="auth:admin", display_name="Secondary Admin") create_permission(perm2) # Get the original auth:admin permission (the one created in setup) perms = list(db.data().permissions.values()) admin_perms = [p for p in perms if p.scope == "auth:admin"] # Delete the first one (not the one we just created) original_admin_perm = next(p for p in admin_perms if p.uuid != perm2.uuid) # Now we can delete the original one response = await client.delete( f"/auth/api/admin/permissions/{original_admin_perm.uuid}", headers={**auth_headers(session_token), "Host": "localhost:4401"}, ) assert response.status_code == 200 data = response.json() assert data["status"] == "ok" @pytest.mark.asyncio async def test_delete_permission_auth_admin_domain_mismatch_fails( self, client: httpx.AsyncClient, session_token: str, test_db: DB ): """Cannot delete auth:admin if remaining one has mismatched domain.""" # Create a second auth:admin permission with a different domain perm2 = Permission.create( scope="auth:admin", display_name="Other Domain Admin", domain="other.example.com", ) create_permission(perm2) # Cannot delete the original one because the remaining one is not accessible # Get the original auth:admin permission perms = list(db.data().permissions.values()) admin_perms = [p for p in perms if p.scope == "auth:admin" and p.domain is None] original_admin_perm = admin_perms[0] # The one without domain response = await client.delete( f"/auth/api/admin/permissions/{original_admin_perm.uuid}", headers={**auth_headers(session_token), "Host": "localhost:4401"}, ) assert response.status_code == 400 data = response.json() assert "lock you out of admin access" in data["detail"] @pytest.mark.asyncio async def test_remove_auth_admin_from_own_org_fails( self, client: httpx.AsyncClient, session_token: str, test_org ): """Cannot remove auth:admin permission from your own organization.""" admin_perm = next( p for p in db.data().permissions.values() if p.scope == "auth:admin" ) response = await client.delete( f"/auth/api/admin/orgs/{test_org.uuid}/permission?permission_uuid={admin_perm.uuid}", headers={**auth_headers(session_token), "Host": "localhost:4401"}, ) assert response.status_code == 400 data = response.json() assert "lock you out of admin access" in data["detail"] # -------------------- Edge Cases for AuthException in Org-Admin Checks -------------------- class TestOrgAdminAuthExceptions: """Tests for org admin AuthException branches that require specific permission checks.""" @pytest.mark.asyncio async def test_create_reset_link_regular_user_forbidden( self, client: httpx.AsyncClient, regular_session_token: str, test_org, test_user, ): """Regular user (not org admin) trying to create reset link should get 403.""" response = await client.post( f"/auth/api/admin/users/{test_user.uuid}/create-link", headers={**auth_headers(regular_session_token), "Host": "localhost:4401"}, ) assert response.status_code == 403 @pytest.mark.asyncio async def test_get_user_detail_regular_user_forbidden( self, client: httpx.AsyncClient, regular_session_token: str, test_org, test_user, ): """Regular user trying to get user details should get 403.""" response = await client.get( f"/auth/api/admin/users/{test_user.uuid}", headers={**auth_headers(regular_session_token), "Host": "localhost:4401"}, ) assert response.status_code == 403 @pytest.mark.asyncio async def test_update_display_name_regular_user_forbidden( self, client: httpx.AsyncClient, regular_session_token: str, test_org, test_user, ): """Regular user trying to update display name should get 403.""" response = await client.patch( f"/auth/api/admin/users/{test_user.uuid}/info", json={"display_name": "New Name"}, headers={**auth_headers(regular_session_token), "Host": "localhost:4401"}, ) assert response.status_code == 403 @pytest.mark.asyncio async def test_delete_credential_regular_user_forbidden( self, client: httpx.AsyncClient, regular_session_token: str, test_org, test_user, test_credential, ): """Regular user trying to delete credential should get 403.""" response = await client.delete( f"/auth/api/admin/users/{test_user.uuid}/credentials/{test_credential.uuid}", headers={**auth_headers(regular_session_token), "Host": "localhost:4401"}, ) assert response.status_code == 403 @pytest.mark.asyncio async def test_delete_session_regular_user_forbidden( self, client: httpx.AsyncClient, regular_session_token: str, test_org, test_user, ): """Regular user trying to delete session should get 403.""" response = await client.delete( f"/auth/api/admin/users/{test_user.uuid}/sessions/some-session", headers={**auth_headers(regular_session_token), "Host": "localhost:4401"}, ) assert response.status_code == 403 class TestDomains: """Tests for the domain management API (/auth/api/admin/domains/).""" async def _set_auth_host(self, client, session_token, test_user, test_credential): """Configure an auth host on the localhost domain, as the admin UI would.""" r = await client.patch( "/auth/api/admin/domains/localhost", json={ "rp_name": "", "origins": { "auth.localhost": {"auth_host": True}, "localhost": True, }, }, headers={**auth_headers(session_token), "Host": "localhost:4401"}, ) assert r.status_code == 200, r.text domain_cfg = db.data().config.domains["localhost"] assert domains.auth_host_url(domain_cfg) == "https://auth.localhost" domain = domains.registry().get("localhost") assert domain.own_auth_host == "auth.localhost" assert domain.auth_site_url == "https://auth.localhost/" # Session for requests coming from the auth host (sessions are host-bound) _, token = create_test_session( test_user.uuid, test_credential.uuid, host="auth.localhost" ) return {**auth_headers(token), "Host": "auth.localhost"} @pytest.mark.asyncio async def test_list_domains(self, client: httpx.AsyncClient, session_token: str): r = await client.get( "/auth/api/admin/domains/", headers={**auth_headers(session_token), "Host": "localhost:4401"}, ) assert r.status_code == 200, r.text data = r.json() assert len(data) == 1 domain = data[0] assert domain["rp_id"] == "localhost" assert domain["origins"] == {"**.localhost": True} assert "related" not in domain assert domain["auth_host"] is None assert domain["site_url"] == "http://localhost:4401" @pytest.mark.asyncio async def test_domains_require_master_admin( self, client: httpx.AsyncClient, regular_session_token: str ): r = await client.get( "/auth/api/admin/domains/", headers={**auth_headers(regular_session_token), "Host": "localhost:4401"}, ) assert r.status_code in (401, 403) @pytest.mark.asyncio async def test_remove_auth_host_updates_runtime( self, client: httpx.AsyncClient, session_token: str, test_user, test_credential, ): """Removing the auth host mark must clear it from runtime config and URLs.""" headers = await self._set_auth_host( client, session_token, test_user, test_credential ) # The dialog still lists the old auth host among origins, so it is sent back r = await client.patch( "/auth/api/admin/domains/localhost", json={ "rp_name": "", "origins": {"auth.localhost": True, "localhost": True}, }, headers=headers, ) assert r.status_code == 200, r.text domain_cfg = db.data().config.domains["localhost"] assert domains.auth_host_url(domain_cfg) is None domain = domains.registry().get("localhost") assert domain.own_auth_host is None assert domain.ui_base_path == "/auth/" # Site URL derivation is stateless: with the auth host mark removed, # the exact rp-id origin becomes the site URL. assert domain.auth_site_url == "https://localhost/auth/" # GET and settings reflect the cleared state r = await client.get( "/auth/api/admin/domains/", headers={**auth_headers(session_token), "Host": "localhost:4401"}, ) assert r.json()[0]["origins"] == {"auth.localhost": True, "localhost": True} r = await client.get("/auth/api/settings") assert r.json()["auth_host"] is None assert r.json()["own_auth_host"] is None assert r.json()["ui_base_path"] == "/auth/" # Middleware no longer redirects to the removed auth host r = await client.get( "/auth/admin", headers={**auth_headers(session_token), "Host": "localhost:4401"}, follow_redirects=False, ) assert "auth.localhost" not in r.headers.get("location", "") @pytest.mark.asyncio async def test_remove_auth_host_without_origins_falls_back( self, client: httpx.AsyncClient, session_token: str, ): """Emptying a domain's origins table must not keep the removed auth host in derived URLs. Only possible on a domain other than the one in use — the lockout guard refuses it there.""" headers = {**auth_headers(session_token), "Host": "localhost:4401"} r = await client.post( "/auth/api/admin/domains/", json={ "rp_id": "example.com", "origins": { "auth.example.com": {"auth_host": True}, "app.example.com": True, }, }, headers=headers, ) assert r.status_code == 200, r.text domain = domains.registry().get("example.com") assert domain.own_auth_host == "auth.example.com" assert "auth.example.com" in domain.site_url r = await client.patch( "/auth/api/admin/domains/example.com", json={"rp_name": "", "origins": {}}, headers=headers, ) assert r.status_code == 200, r.text domain = domains.registry().get("example.com") assert domain.own_auth_host is None assert domain.ui_base_path == "/auth/" assert "auth.example.com" not in domain.site_url assert "auth.example.com" not in domain.auth_site_url @pytest.mark.asyncio async def test_create_and_delete_domain( self, client: httpx.AsyncClient, session_token: str ): headers = {**auth_headers(session_token), "Host": "localhost:4401"} r = await client.post( "/auth/api/admin/domains/", json={ "rp_id": "example.com", "rp_name": "Example", "origins": {"app.example.com": True, "unrelated-site.com": True}, }, headers=headers, ) assert r.status_code == 200, r.text r = await client.get("/auth/api/admin/domains/", headers=headers) domains_list = {domain["rp_id"]: domain for domain in r.json()} assert set(domains_list) == {"localhost", "example.com"} created = domains_list["example.com"] assert created["rp_name"] == "Example" # In-domain and related origins live in one table; classification # is derived from the rp-id assert created["origins"] == { "app.example.com": True, "unrelated-site.com": True, } r = await client.delete("/auth/api/admin/domains/example.com", headers=headers) assert r.status_code == 200, r.text assert "example.com" not in db.data().config.domains assert domains.registry().get("example.com") is None @pytest.mark.asyncio async def test_create_domain_validation( self, client: httpx.AsyncClient, session_token: str ): headers = {**auth_headers(session_token), "Host": "localhost:4401"} # rp_id is required r = await client.post("/auth/api/admin/domains/", json={}, headers=headers) assert r.status_code == 400 # Duplicate rp-id r = await client.post( "/auth/api/admin/domains/", json={"rp_id": "localhost"}, headers=headers ) assert r.status_code == 400 # Invalid rp-id r = await client.post( "/auth/api/admin/domains/", json={"rp_id": "not a domain!"}, headers=headers ) assert r.status_code == 400 # An auth host must be within the rp-id domain r = await client.post( "/auth/api/admin/domains/", json={ "rp_id": "example.com", "origins": {"auth.other.com": {"auth_host": True}}, }, headers=headers, ) assert r.status_code == 400 # Related origin host may not collide across domains r = await client.post( "/auth/api/admin/domains/", json={"rp_id": "example.com", "origins": {"shared-app.com": True}}, headers=headers, ) assert r.status_code == 200 r = await client.post( "/auth/api/admin/domains/", json={"rp_id": "other.com", "origins": {"shared-app.com": True}}, headers=headers, ) assert r.status_code == 400 # Cross-domain entries are related origins — accepted in the same table r = await client.post( "/auth/api/admin/domains/", json={"rp_id": "another.com", "origins": {"elsewhere.com": True}}, headers=headers, ) assert r.status_code == 200 # Plain '*' is rejected — wildcards must be explicit ('**.another.com') r = await client.post( "/auth/api/admin/domains/", json={"rp_id": "star.com", "origins": {"*": True}}, headers=headers, ) assert r.status_code == 400 @pytest.mark.asyncio async def test_delete_domain_guards( self, client: httpx.AsyncClient, session_token: str, test_credential ): headers = {**auth_headers(session_token), "Host": "localhost:4401"} # Cannot delete the last domain r = await client.delete("/auth/api/admin/domains/localhost", headers=headers) assert r.status_code == 400 # Unknown domain r = await client.delete("/auth/api/admin/domains/nope.com", headers=headers) assert r.status_code == 400 # A domain with credentials still registered under it cannot be deleted r = await client.post( "/auth/api/admin/domains/", json={"rp_id": "example.com"}, headers=headers ) assert r.status_code == 200 cred = Credential.create( credential_id=secrets.token_bytes(32), user=test_credential.user_uuid, aaguid=UUID("00000000-0000-0000-0000-000000000000"), public_key=secrets.token_bytes(64), sign_count=0, rp_id="example.com", ) create_credential(cred) r = await client.delete("/auth/api/admin/domains/example.com", headers=headers) assert r.status_code == 400 @pytest.mark.asyncio async def test_update_domain_refuses_self_lockout( self, client: httpx.AsyncClient, session_token: str ): """An allow-list excluding the admin's current host is refused.""" headers = {**auth_headers(session_token), "Host": "localhost:4401"} # Allow-list without the current host and no auth host → lockout r = await client.patch( "/auth/api/admin/domains/localhost", json={"rp_name": "", "origins": {"auth.localhost": True}}, headers=headers, ) assert r.status_code == 400 assert "lock you out" in r.text # Emptying the origins table entirely is likewise a lockout r = await client.patch( "/auth/api/admin/domains/localhost", json={"rp_name": "", "origins": {}}, headers=headers, ) assert r.status_code == 400 assert "lock you out" in r.text # Allow-list including the current host is fine r = await client.patch( "/auth/api/admin/domains/localhost", json={"rp_name": "", "origins": {"localhost:4401": True}}, headers=headers, ) assert r.status_code == 200, r.text # An allow-list without the current host is also fine when an auth # host is set: ceremonies move there (and it is always allowed). # Done last: with an auth host set, the API here routes differently. r = await client.patch( "/auth/api/admin/domains/localhost", json={ "rp_name": "", "origins": {"auth.localhost": {"auth_host": True}}, }, headers=headers, ) assert r.status_code == 200, r.text @pytest.mark.asyncio async def test_delete_current_domain_refused( self, client: httpx.AsyncClient, session_token: str ): headers = {**auth_headers(session_token), "Host": "localhost:4401"} r = await client.post( "/auth/api/admin/domains/", json={"rp_id": "example.com"}, headers=headers ) assert r.status_code == 200 # Deleting the domain in use is refused even if it has no credentials r = await client.delete("/auth/api/admin/domains/localhost", headers=headers) assert r.status_code == 400 assert "currently using" in r.text # Deleting another domain while authenticated here is fine r = await client.delete("/auth/api/admin/domains/example.com", headers=headers) assert r.status_code == 200, r.text @pytest.mark.asyncio async def test_no_cross_domain_auth_host_fallback( self, client: httpx.AsyncClient, session_token: str, test_user, test_credential, ): """A domain without its own auth host reports none — there is no cross-domain fallback to another domain's auth host.""" headers = await self._set_auth_host( client, session_token, test_user, test_credential ) r = await client.post( "/auth/api/admin/domains/", json={"rp_id": "example.com"}, headers=headers ) assert r.status_code == 200 # Settings on the example.com host report no auth host r = await client.get("/auth/api/settings", headers={"Host": "example.com"}) assert r.status_code == 200 assert r.json()["rp_id"] == "example.com" assert r.json()["auth_host"] is None assert r.json()["own_auth_host"] is None # The localhost domain still reports its own auth host r = await client.get("/auth/api/settings", headers={"Host": "auth.localhost"}) assert r.status_code == 200 assert r.json()["auth_host"] == "auth.localhost" assert r.json()["own_auth_host"] == "auth.localhost"