LeoVasanko
52f3b26630
'**.example.com' covers the apex and subdomains at any depth;
'*.example.com' covers exactly one subdomain level (neither apex nor
deeper) — analogous to permission scope wildcards, and sidestepping the
DNS/TLS/nginx ambiguity around '*.'. This also allows excluding the apex
where wanted. The seeded/default entry becomes '**.{rp-id}' (init,
add-domain, legacy empty-origins conversion, branch-era '*' sanitize
rewrite).
2157 lines
78 KiB
Python
2157 lines
78 KiB
Python
"""
|
|
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"
|