diff --git a/paskia/fastapi/admin.py b/paskia/fastapi/admin.py index 07ecba4..382aad1 100644 --- a/paskia/fastapi/admin.py +++ b/paskia/fastapi/admin.py @@ -38,7 +38,7 @@ async def auth_exception_handler(_request, exc: authz.AuthException): @app.exception_handler(Exception) -async def general_exception_handler(_request, exc: Exception): +async def general_exception_handler(_request, exc: Exception): # pragma: no cover logging.exception("Unhandled exception in admin app") return JSONResponse(status_code=500, content={"detail": "Internal server error"}) @@ -139,7 +139,9 @@ async def admin_update_org( current = await db.instance.get_organization(str(org_uuid)) display_name = payload.get("display_name") or current.display_name - permissions = payload.get("permissions") or current.permissions or [] + permissions = payload.get("permissions") + if permissions is None: + permissions = current.permissions or [] # Sanity check: prevent removing permissions that would break current user's admin access org_admin_perm = f"auth:org:{org_uuid}" @@ -398,7 +400,7 @@ async def admin_update_user_role( # Sanity check: prevent admin from removing their own access if ctx.user.uuid == user_uuid: new_role_obj = next((r for r in roles if r.display_name == new_role), None) - if new_role_obj: + if new_role_obj: # pragma: no branch - always true, role validated above has_admin_access = ( "auth:admin" in new_role_obj.permissions or f"auth:org:{org_uuid}" in new_role_obj.permissions @@ -432,7 +434,7 @@ async def admin_create_user_registration_link( host=request.headers.get("host"), max_age="5m", ) - if ( + if ( # pragma: no cover - defense in depth, authz.verify already checked "auth:admin" not in ctx.role.permissions and f"auth:org:{org_uuid}" not in ctx.role.permissions ): @@ -482,7 +484,7 @@ async def admin_get_user_detail( match=permutil.has_any, host=request.headers.get("host"), ) - if ( + if ( # pragma: no cover - defense in depth, authz.verify already checked "auth:admin" not in ctx.role.permissions and f"auth:org:{org_uuid}" not in ctx.role.permissions ): @@ -496,7 +498,7 @@ async def admin_get_user_detail( for cid in cred_ids: try: c = await db.instance.get_credential_by_id(cid) - except ValueError: + except ValueError: # pragma: no cover - race condition handling continue aaguid_str = str(c.aaguid) aaguids.add(aaguid_str) @@ -631,7 +633,7 @@ async def admin_update_user_display_name( match=permutil.has_any, host=request.headers.get("host"), ) - if ( + if ( # pragma: no cover - defense in depth, authz.verify already checked "auth:admin" not in ctx.role.permissions and f"auth:org:{org_uuid}" not in ctx.role.permissions ): @@ -668,7 +670,7 @@ async def admin_delete_user_credential( host=request.headers.get("host"), max_age="5m", ) - if ( + if ( # pragma: no cover - defense in depth, authz.verify already checked "auth:admin" not in ctx.role.permissions and f"auth:org:{org_uuid}" not in ctx.role.permissions ): @@ -699,7 +701,7 @@ async def admin_delete_user_session( match=permutil.has_any, host=request.headers.get("host"), ) - if ( + if ( # pragma: no cover - defense in depth, authz.verify already checked "auth:admin" not in ctx.role.permissions and f"auth:org:{org_uuid}" not in ctx.role.permissions ): @@ -818,7 +820,7 @@ async def admin_rename_permission( perm = await db.instance.get_permission(old_id) display_name = perm.display_name rename_fn = getattr(db.instance, "rename_permission", None) - if not rename_fn: + if not rename_fn: # pragma: no cover - all current backends support rename raise ValueError("Permission renaming not supported by this backend") await rename_fn(old_id, new_id, display_name) return {"status": "ok"} diff --git a/paskia/fastapi/api.py b/paskia/fastapi/api.py index c395de3..ebc020d 100644 --- a/paskia/fastapi/api.py +++ b/paskia/fastapi/api.py @@ -65,7 +65,9 @@ async def auth_exception_handler(_request: Request, exc: authz.AuthException): @app.exception_handler(Exception) -async def general_exception_handler(_request: Request, exc: Exception): +async def general_exception_handler( + _request: Request, exc: Exception +): # pragma: no cover logging.exception("Unhandled exception in API app") return JSONResponse(status_code=500, content={"detail": "Internal server error"}) diff --git a/pyproject.toml b/pyproject.toml index 6e278f0..fd1cad6 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -33,6 +33,9 @@ version-file = "paskia/_version.py" dev = [ "ruff>=0.1.0", "coverage[toml]>=7.0.0", + "pytest>=8.0.0", + "pytest-asyncio>=0.24.0", + "httpx>=0.27.0", ] [tool.coverage.run] @@ -52,6 +55,14 @@ show_missing = true [tool.coverage.html] directory = "coverage-html" +[tool.pytest.ini_options] +asyncio_mode = "auto" +asyncio_default_fixture_loop_scope = "function" +testpaths = ["tests"] +filterwarnings = [ + "ignore::DeprecationWarning", +] + [tool.ruff] target-version = "py39" line-length = 88 @@ -61,6 +72,15 @@ select = ["E", "F", "I", "N", "W", "UP"] ignore = ["E501"] # Line too long isort.known-first-party = ["paskia"] +[dependency-groups] +dev = [ + "coverage>=7.12.0", + "httpx>=0.28.1", + "pytest>=9.0.1", + "pytest-asyncio>=1.3.0", + "pytest-cov>=7.0.0", +] + [project.scripts] paskia = "paskia.fastapi.__main__:main" diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..bc9758d --- /dev/null +++ b/tests/__init__.py @@ -0,0 +1 @@ +# Paskia API Tests diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..705055b --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,261 @@ +""" +Pytest configuration and fixtures for Paskia API tests. + +FastAPI provides excellent testing support through httpx.ASGITransport, +which allows us to make async requests directly to the ASGI app without +running a server. + +Since we can't emulate WebAuthn passkeys, we create sessions directly +in the database to test authenticated endpoints. +""" + +import asyncio +import os +from collections.abc import AsyncGenerator +from datetime import datetime, timezone +from uuid import UUID + +import httpx +import pytest +import pytest_asyncio +import uuid7 + +from paskia import globals +from paskia.db import Credential, Org, Permission, Role, User +from paskia.db.sql import DB +from paskia.fastapi.session import AUTH_COOKIE_NAME +from paskia.sansio import Passkey +from paskia.util.tokens import create_token, session_key + +# Use in-memory SQLite for tests +os.environ["PASKIA_DB"] = "sqlite+aiosqlite:///:memory:" + + +@pytest.fixture(scope="session") +def event_loop(): + """Create an event loop for the test session.""" + loop = asyncio.get_event_loop_policy().new_event_loop() + yield loop + loop.close() + + +@pytest_asyncio.fixture(scope="function") +async def test_db() -> AsyncGenerator[DB, None]: + """Create an in-memory SQLite database for testing. + + We use :memory: for speed - each test gets a fresh database. + """ + db = DB("sqlite+aiosqlite:///:memory:") + await db.init_db() + globals.db._instance = db + yield db + # Clean up + globals.db._instance = None + + +@pytest_asyncio.fixture(scope="function") +async def passkey_instance() -> Passkey: + """Initialize a passkey instance for testing.""" + pk = Passkey( + rp_id="localhost", + rp_name="Test RP", + origins=["http://localhost:4401"], + ) + globals.passkey._instance = pk + yield pk + globals.passkey._instance = None + + +@pytest_asyncio.fixture(scope="function") +async def test_org(test_db: DB, admin_permission: Permission) -> Org: + """Create a test organization with admin permission.""" + org = Org( + uuid=uuid7.create(), + display_name="Test Organization", + permissions=["auth:admin"], # Org can grant this permission + ) + await test_db.create_organization(org) + return org + + +@pytest_asyncio.fixture(scope="function") +async def admin_permission(test_db: DB) -> Permission: + """Create the auth:admin permission.""" + perm = Permission(id="auth:admin", display_name="Master Admin") + await test_db.create_permission(perm) + return perm + + +@pytest_asyncio.fixture(scope="function") +async def test_role(test_db: DB, test_org: Org, admin_permission: Permission) -> Role: + """Create a test role with admin permission.""" + role = Role( + uuid=uuid7.create(), + org_uuid=test_org.uuid, + display_name="Test Admin Role", + permissions=["auth:admin", f"auth:org:{test_org.uuid}"], + ) + await test_db.create_role(role) + return role + + +@pytest_asyncio.fixture(scope="function") +async def user_role(test_db: DB, test_org: Org) -> Role: + """Create a test role without admin permission (regular user).""" + role = Role( + uuid=uuid7.create(), + org_uuid=test_org.uuid, + display_name="User Role", + permissions=[], + ) + await test_db.create_role(role) + return role + + +@pytest_asyncio.fixture(scope="function") +async def test_user(test_db: DB, test_role: Role) -> User: + """Create a test user with admin role.""" + user = User( + uuid=uuid7.create(), + display_name="Test Admin", + role_uuid=test_role.uuid, + created_at=datetime.now(timezone.utc), + visits=0, + ) + await test_db.create_user(user) + return user + + +@pytest_asyncio.fixture(scope="function") +async def regular_user(test_db: DB, user_role: Role) -> User: + """Create a regular test user without admin permissions.""" + user = User( + uuid=uuid7.create(), + display_name="Regular User", + role_uuid=user_role.uuid, + created_at=datetime.now(timezone.utc), + visits=0, + ) + await test_db.create_user(user) + return user + + +@pytest_asyncio.fixture(scope="function") +async def test_credential(test_db: DB, test_user: User) -> Credential: + """Create a test credential for the admin user.""" + credential = Credential( + uuid=uuid7.create(), + credential_id=os.urandom(32), + user_uuid=test_user.uuid, + aaguid=UUID("00000000-0000-0000-0000-000000000000"), + public_key=os.urandom(64), + sign_count=0, + created_at=datetime.now(timezone.utc), + last_used=None, + last_verified=None, + ) + await test_db.create_credential(credential) + return credential + + +@pytest_asyncio.fixture(scope="function") +async def regular_credential(test_db: DB, regular_user: User) -> Credential: + """Create a test credential for the regular user.""" + credential = Credential( + uuid=uuid7.create(), + credential_id=os.urandom(32), + user_uuid=regular_user.uuid, + aaguid=UUID("00000000-0000-0000-0000-000000000000"), + public_key=os.urandom(64), + sign_count=0, + created_at=datetime.now(timezone.utc), + last_used=None, + last_verified=None, + ) + await test_db.create_credential(credential) + return credential + + +@pytest_asyncio.fixture(scope="function") +async def session_token( + test_db: DB, test_user: User, test_credential: Credential +) -> str: + """Create a session for the admin user and return the token.""" + token = create_token() + await test_db.create_session( + user_uuid=test_user.uuid, + credential_uuid=test_credential.uuid, + key=session_key(token), + host="localhost:4401", + ip="127.0.0.1", + user_agent="pytest", + renewed=datetime.now(timezone.utc), + ) + return token + + +@pytest_asyncio.fixture(scope="function") +async def regular_session_token( + test_db: DB, regular_user: User, regular_credential: Credential +) -> str: + """Create a session for a regular user and return the token.""" + token = create_token() + await test_db.create_session( + user_uuid=regular_user.uuid, + credential_uuid=regular_credential.uuid, + key=session_key(token), + host="localhost:4401", + ip="127.0.0.1", + user_agent="pytest", + renewed=datetime.now(timezone.utc), + ) + return token + + +@pytest_asyncio.fixture(scope="function") +async def reset_token(test_db: DB, test_user: User, test_credential: Credential) -> str: + """Create a reset token for the test user.""" + from paskia.authsession import reset_expires + from paskia.util.passphrase import generate + from paskia.util.tokens import reset_key + + token = generate() + await test_db.create_reset_token( + user_uuid=test_user.uuid, + key=reset_key(token), + expiry=reset_expires(), + token_type="reset", + ) + return token + + +@pytest_asyncio.fixture(scope="function") +async def client( + test_db: DB, passkey_instance: Passkey +) -> AsyncGenerator[httpx.AsyncClient, None]: + """Create an async test client for the FastAPI app. + + Note: We import the app inside the fixture to ensure globals are + initialized first. + """ + # Import app after globals are set + from paskia.fastapi.mainapp import app + + transport = httpx.ASGITransport(app=app) + async with httpx.AsyncClient( + transport=transport, + base_url="http://localhost:4401", + ) as client: + yield client + + +def auth_headers(token: str) -> dict[str, str]: + """Return headers with auth cookie set.""" + return {"Cookie": f"{AUTH_COOKIE_NAME}={token}"} + + +def auth_cookie(token: str) -> httpx.Cookies: + """Return cookies dict with auth cookie.""" + cookies = httpx.Cookies() + cookies.set(AUTH_COOKIE_NAME, token, domain="localhost") + return cookies diff --git a/tests/test_admin.py b/tests/test_admin.py new file mode 100644 index 0000000..c7296d9 --- /dev/null +++ b/tests/test_admin.py @@ -0,0 +1,1565 @@ +""" +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 +""" + +from datetime import datetime, timezone +from uuid import UUID + +import httpx +import pytest +import pytest_asyncio +import uuid7 + +from paskia.db import Credential, Org, Permission, Role, User +from paskia.db.sql import DB +from paskia.util.tokens import create_token, encode_session_key, session_key +from tests.conftest import auth_headers + +# -------------------- Additional Fixtures -------------------- + + +@pytest_asyncio.fixture(scope="function") +async def second_org(test_db: DB) -> Org: + """Create a second organization for deletion tests.""" + org = Org( + uuid=uuid7.create(), + display_name="Second Organization", + permissions=[], + ) + await test_db.create_organization(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( + uuid=uuid7.create(), + org_uuid=second_org.uuid, + display_name="Second Org Admin Role", + permissions=["auth:admin"], + ) + await test_db.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( + uuid=uuid7.create(), + display_name="Second Org User", + role_uuid=second_org_role.uuid, + created_at=datetime.now(timezone.utc), + visits=0, + ) + await test_db.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.""" + import os + + credential = Credential( + uuid=uuid7.create(), + credential_id=os.urandom(32), + user_uuid=second_org_user.uuid, + aaguid=UUID("00000000-0000-0000-0000-000000000000"), + public_key=os.urandom(64), + sign_count=0, + created_at=datetime.now(timezone.utc), + last_used=datetime.now(timezone.utc), + last_verified=datetime.now(timezone.utc), + ) + await test_db.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.""" + token = create_token() + await test_db.create_session( + user_uuid=second_org_user.uuid, + credential_uuid=second_org_credential.uuid, + key=session_key(token), + host="localhost:4401", + ip="127.0.0.1", + user_agent="pytest", + renewed=datetime.now(timezone.utc), + ) + return token + + +@pytest_asyncio.fixture(scope="function") +async def org_admin_role(test_db: DB, test_org: Org) -> Role: + """Create a role with org admin permission only (no global admin).""" + role = Role( + uuid=uuid7.create(), + org_uuid=test_org.uuid, + display_name="Org Admin Role", + permissions=[f"auth:org:{test_org.uuid}"], + ) + await test_db.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( + uuid=uuid7.create(), + display_name="Org Admin User", + role_uuid=org_admin_role.uuid, + created_at=datetime.now(timezone.utc), + visits=5, + last_seen=datetime.now(timezone.utc), + ) + await test_db.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.""" + import os + + credential = Credential( + uuid=uuid7.create(), + credential_id=os.urandom(32), + user_uuid=org_admin_user.uuid, + aaguid=UUID("00000000-0000-0000-0000-000000000000"), + public_key=os.urandom(64), + sign_count=0, + created_at=datetime.now(timezone.utc), + last_used=datetime.now(timezone.utc), + last_verified=None, + ) + await test_db.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.""" + token = create_token() + await test_db.create_session( + user_uuid=org_admin_user.uuid, + credential_uuid=org_admin_credential.uuid, + key=session_key(token), + host="localhost:4401", + ip="127.0.0.1", + user_agent="pytest", + renewed=datetime.now(timezone.utc), + ) + return token + + +@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(id="test:grantable:perm", display_name="Grantable Perm") + await test_db.create_permission(perm) + # Add to org's grantable permissions + await test_db.add_permission_to_organization(str(test_org.uuid), perm.id) + 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/orgs") + 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"] + + +# -------------------- Admin App Root -------------------- + + +class TestAdminAppRoot: + """Tests for the admin app root endpoint""" + + @pytest.mark.asyncio + async def test_admin_app_root_with_auth( + self, client: httpx.AsyncClient, session_token: str + ): + """Admin app root returns HTML when authenticated.""" + response = await client.get( + "/auth/api/admin/", + headers={**auth_headers(session_token), "Host": "localhost:4401"}, + ) + assert response.status_code == 200 + assert "text/html" in response.headers.get("content-type", "") + + +# -------------------- 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/orgs") + 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/orgs", + 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/orgs", + headers={**auth_headers(session_token), "Host": "localhost:4401"}, + ) + assert response.status_code == 200 + data = response.json() + assert isinstance(data, list) + assert len(data) >= 1 + # Check org structure + org = data[0] + assert "uuid" in org + assert "display_name" in org + assert "roles" in org + assert "users" in org + + @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/orgs", + 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 = [o["uuid"] for o in data] + 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.put( + 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.put( + f"/auth/api/admin/orgs/{test_org.uuid}", + json={ + "display_name": "Org Admin Updated Name", + "permissions": [f"auth:org:{test_org.uuid}"], # Keep org admin perm + }, + 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.""" + # First create and add the org admin perm to the org's grantable perms + org_admin_perm_id = f"auth:org:{test_org.uuid}" + perm = Permission(id=org_admin_perm_id, display_name="Org Admin") + try: + await test_db.create_permission(perm) + except Exception: + pass # Permission may already exist + + # Add it to the org's permissions + await test_db.add_permission_to_organization( + str(test_org.uuid), org_admin_perm_id + ) + + # Try to remove all permissions including org admin perm + response = await client.put( + f"/auth/api/admin/orgs/{test_org.uuid}", + json={ + "display_name": "Try Remove Own Perm", + "permissions": [], # Remove org admin perm from org's permissions + }, + headers={**auth_headers(org_admin_session_token), "Host": "localhost:4401"}, + ) + assert response.status_code == 400 + data = response.json() + assert "Cannot remove organization admin permission" in data["detail"] + + @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( + uuid=uuid7.create(), + display_name="Org To Delete", + permissions=[], + ) + await test_db.create_organization(org_to_delete) + + # Create some org-specific permissions to test cleanup + org_perm = Permission( + id=f"test:org:{org_to_delete.uuid}:feature", display_name="Org Feature" + ) + await test_db.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 + await client.post( + "/auth/api/admin/permissions", + json={"id": "test:org:addable", "display_name": "Addable"}, + headers={**auth_headers(session_token), "Host": "localhost:4401"}, + ) + + # Add it to the org + response = await client.post( + f"/auth/api/admin/orgs/{test_org.uuid}/permission?permission_id=test:org:addable", + 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).""" + response = await client.post( + f"/auth/api/admin/orgs/{test_org.uuid}/permission?permission_id=auth:admin", + 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 + await client.post( + "/auth/api/admin/permissions", + json={"id": "test:org:removable", "display_name": "Removable"}, + headers={**auth_headers(session_token), "Host": "localhost:4401"}, + ) + await client.post( + f"/auth/api/admin/orgs/{test_org.uuid}/permission?permission_id=test:org:removable", + headers={**auth_headers(session_token), "Host": "localhost:4401"}, + ) + + # Remove it + response = await client.delete( + f"/auth/api/admin/orgs/{test_org.uuid}/permission?permission_id=test:org:removable", + 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).""" + response = await client.delete( + f"/auth/api/admin/orgs/{test_org.uuid}/permission?permission_id=auth:admin", + 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": [grantable_permission.id], + }, + 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(id="test:not:grantable", display_name="Not Grantable") + await test_db.create_permission(perm) + + response = await client.post( + f"/auth/api/admin/orgs/{test_org.uuid}/roles", + json={ + "display_name": "Bad Role", + "permissions": ["test:not:grantable"], + }, + 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.put( + f"/auth/api/admin/orgs/{test_org.uuid}/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, session_token: str, test_org, second_org_role + ): + """Cannot update role from another org.""" + response = await client.put( + f"/auth/api/admin/orgs/{test_org.uuid}/roles/{second_org_role.uuid}", + json={"display_name": "Try Update Wrong Org"}, + headers={**auth_headers(session_token), "Host": "localhost:4401"}, + ) + assert response.status_code == 404 + data = response.json() + assert "Role not found" in data["detail"] + + @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.put( + f"/auth/api/admin/orgs/{test_org.uuid}/roles/{user_role.uuid}", + json={"permissions": [grantable_permission.id]}, + 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(id="test:not:grantable:update", display_name="Not Grantable") + await test_db.create_permission(perm) + + response = await client.put( + f"/auth/api/admin/orgs/{test_org.uuid}/roles/{user_role.uuid}", + json={"permissions": ["test:not:grantable:update"]}, + 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 cannot remove their own admin permissions.""" + response = await client.put( + f"/auth/api/admin/orgs/{test_org.uuid}/roles/{test_role.uuid}", + json={"permissions": []}, # Remove all permissions + headers={**auth_headers(session_token), "Host": "localhost:4401"}, + ) + assert response.status_code == 400 + data = response.json() + assert "Cannot update your own role" 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/orgs/{test_org.uuid}/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, session_token: str, test_org, second_org_role + ): + """Cannot delete role from another org.""" + response = await client.delete( + f"/auth/api/admin/orgs/{test_org.uuid}/roles/{second_org_role.uuid}", + headers={**auth_headers(session_token), "Host": "localhost:4401"}, + ) + assert response.status_code == 404 + data = response.json() + assert "Role not found" in data["detail"] + + @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/orgs/{test_org.uuid}/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/orgs/{test_org.uuid}/users/{test_user.uuid}", + headers={**auth_headers(session_token), "Host": "localhost:4401"}, + ) + assert response.status_code == 200 + data = response.json() + assert "display_name" in data + 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/orgs/{test_org.uuid}/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, session_token: str, test_org, second_org_user + ): + """Getting user from another org should return 404.""" + response = await client.get( + f"/auth/api/admin/orgs/{test_org.uuid}/users/{second_org_user.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_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/orgs/{test_org.uuid}/users/{org_admin_user.uuid}", + headers={**auth_headers(org_admin_session_token), "Host": "localhost:4401"}, + ) + assert response.status_code == 200 + data = response.json() + assert "display_name" in data + + @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.put( + f"/auth/api/admin/orgs/{test_org.uuid}/users/{test_user.uuid}/display-name", + 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.put( + f"/auth/api/admin/orgs/{test_org.uuid}/users/{fake_uuid}/display-name", + 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, session_token: str, test_org, second_org_user + ): + """Updating user from another org should return 404.""" + response = await client.put( + f"/auth/api/admin/orgs/{test_org.uuid}/users/{second_org_user.uuid}/display-name", + json={"display_name": "New Name"}, + headers={**auth_headers(session_token), "Host": "localhost:4401"}, + ) + assert response.status_code == 404 + + @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.put( + f"/auth/api/admin/orgs/{test_org.uuid}/users/{test_user.uuid}/display-name", + json={"display_name": " "}, + headers={**auth_headers(session_token), "Host": "localhost:4401"}, + ) + assert response.status_code == 400 + data = response.json() + assert "display_name required" 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.put( + f"/auth/api/admin/orgs/{test_org.uuid}/users/{test_user.uuid}/display-name", + json={"display_name": "x" * 100}, + headers={**auth_headers(session_token), "Host": "localhost:4401"}, + ) + assert response.status_code == 400 + data = response.json() + assert "too long" in data["detail"] + + @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.put( + f"/auth/api/admin/orgs/{test_org.uuid}/users/{regular_user.uuid}/role", + json={"role": user_role.display_name}, + 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.put( + f"/auth/api/admin/orgs/{test_org.uuid}/users/{test_user.uuid}/role", + json={}, + headers={**auth_headers(session_token), "Host": "localhost:4401"}, + ) + assert response.status_code == 400 + data = response.json() + assert "role 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.put( + f"/auth/api/admin/orgs/{test_org.uuid}/users/{fake_uuid}/role", + json={"role": "User Role"}, + headers={**auth_headers(session_token), "Host": "localhost:4401"}, + ) + assert response.status_code == 400 + 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, session_token: str, test_org, second_org_user + ): + """Updating role for user in another org should fail.""" + response = await client.put( + f"/auth/api/admin/orgs/{test_org.uuid}/users/{second_org_user.uuid}/role", + json={"role": "User Role"}, + headers={**auth_headers(session_token), "Host": "localhost:4401"}, + ) + assert response.status_code == 400 + data = response.json() + assert "does not belong" in data["detail"] + + @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.put( + f"/auth/api/admin/orgs/{test_org.uuid}/users/{test_user.uuid}/role", + json={"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_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.put( + f"/auth/api/admin/orgs/{test_org.uuid}/users/{org_admin_user.uuid}/role", + json={"role": user_role.display_name}, + 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.put( + f"/auth/api/admin/orgs/{test_org.uuid}/users/{test_user.uuid}/role", + json={"role": test_role.display_name}, + 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/orgs/{test_org.uuid}/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/orgs/{test_org.uuid}/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, session_token: str, test_org, second_org_user + ): + """Creating reset link for user in another org should fail.""" + response = await client.post( + f"/auth/api/admin/orgs/{test_org.uuid}/users/{second_org_user.uuid}/create-link", + headers={**auth_headers(session_token), "Host": "localhost:4401"}, + ) + assert response.status_code == 404 + data = response.json() + assert "not found in organization" in data["detail"] + + @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( + uuid=uuid7.create(), + display_name="User Without Creds", + role_uuid=user_role.uuid, + created_at=datetime.now(timezone.utc), + visits=0, + ) + await test_db.create_user(user_no_cred) + + response = await client.post( + f"/auth/api/admin/orgs/{test_org.uuid}/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 + + +# -------------------- 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/orgs/{test_org.uuid}/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/orgs/{test_org.uuid}/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, + session_token: str, + test_org, + second_org_user, + second_org_credential, + ): + """Deleting credential for user in another org should fail.""" + response = await client.delete( + f"/auth/api/admin/orgs/{test_org.uuid}/users/{second_org_user.uuid}/credentials/{second_org_credential.uuid}", + headers={**auth_headers(session_token), "Host": "localhost:4401"}, + ) + assert response.status_code == 404 + + +# -------------------- 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_token = create_token() + extra_key = session_key(extra_token) + await test_db.create_session( + user_uuid=test_user.uuid, + credential_uuid=test_credential.uuid, + key=extra_key, + host="other.host:4401", + ip="192.168.1.1", + user_agent="other-agent", + renewed=datetime.now(timezone.utc), + ) + + encoded_key = encode_session_key(extra_key) + response = await client.delete( + f"/auth/api/admin/orgs/{test_org.uuid}/users/{test_user.uuid}/sessions/{encoded_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.""" + encoded_key = encode_session_key(session_key(session_token)) + response = await client.delete( + f"/auth/api/admin/orgs/{test_org.uuid}/users/{test_user.uuid}/sessions/{encoded_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/orgs/{test_org.uuid}/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, + session_token: str, + test_org, + second_org_user, + ): + """Deleting session for user in another org should fail.""" + response = await client.delete( + f"/auth/api/admin/orgs/{test_org.uuid}/users/{second_org_user.uuid}/sessions/fake-session", + headers={**auth_headers(session_token), "Host": "localhost:4401"}, + ) + assert response.status_code == 404 + + @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 ID format should fail.""" + response = await client.delete( + f"/auth/api/admin/orgs/{test_org.uuid}/users/{test_user.uuid}/sessions/invalid!!id", + headers={**auth_headers(session_token), "Host": "localhost:4401"}, + ) + assert response.status_code == 400 + data = response.json() + assert "Invalid session identifier" 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_key = session_key(create_token()) + encoded_key = encode_session_key(fake_key) + response = await client.delete( + f"/auth/api/admin/orgs/{test_org.uuid}/users/{test_user.uuid}/sessions/{encoded_key}", + 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/permissions", + headers={**auth_headers(session_token), "Host": "localhost:4401"}, + ) + assert response.status_code == 200 + data = response.json() + assert isinstance(data, list) + # Should include at least auth:admin + perm_ids = [p["id"] for p in data] + assert "auth:admin" in perm_ids + + @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 grantable permissions.""" + response = await client.get( + "/auth/api/admin/permissions", + 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_ids = [p["id"] for p in data] + assert grantable_permission.id in perm_ids + # Should NOT see auth:admin (not grantable by org) + assert "auth:admin" not in perm_ids + + @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={"id": "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={"id": "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(id="test:updateable", display_name="Updateable") + await test_db.create_permission(perm) + + response = await client.put( + "/auth/api/admin/permission?permission_id=test:updateable&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 + ): + """Updating permission with empty name should fail.""" + response = await client.put( + "/auth/api/admin/permission?permission_id=test:perm&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_rename_permission( + self, client: httpx.AsyncClient, session_token: str, test_db: DB + ): + """Admin should be able to rename a permission.""" + # Create permission first + perm = Permission(id="test:renameable2", display_name="Renameable") + await test_db.create_permission(perm) + + response = await client.post( + "/auth/api/admin/permission/rename", + json={"old_id": "test:renameable2", "new_id": "test:renamed2"}, + headers={**auth_headers(session_token), "Host": "localhost:4401"}, + ) + assert response.status_code == 200 + + @pytest.mark.asyncio + async def test_rename_permission_missing_ids( + self, client: httpx.AsyncClient, session_token: str + ): + """Renaming permission without IDs should fail.""" + response = await client.post( + "/auth/api/admin/permission/rename", + 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_rename_permission_auth_admin_fails( + self, client: httpx.AsyncClient, session_token: str + ): + """Cannot rename the auth:admin permission.""" + response = await client.post( + "/auth/api/admin/permission/rename", + json={"old_id": "auth:admin", "new_id": "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_rename_permission_with_display_name( + self, client: httpx.AsyncClient, session_token: str, test_db: DB + ): + """Renaming permission can also update display name.""" + perm = Permission(id="test:rename:withname", display_name="Old Name") + await test_db.create_permission(perm) + + response = await client.post( + "/auth/api/admin/permission/rename", + json={ + "old_id": "test:rename:withname", + "new_id": "test:renamed:withname", + "display_name": "New Display Name", + }, + 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(id="test:deleteable", display_name="Deleteable") + await test_db.create_permission(perm) + + response = await client.delete( + "/auth/api/admin/permission?permission_id=test:deleteable", + 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_fails( + self, client: httpx.AsyncClient, session_token: str + ): + """Cannot delete the auth:admin permission.""" + response = await client.delete( + "/auth/api/admin/permission?permission_id=auth:admin", + headers={**auth_headers(session_token), "Host": "localhost:4401"}, + ) + assert response.status_code == 400 + data = response.json() + assert "Cannot delete the master admin" 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/orgs/{test_org.uuid}/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/orgs/{test_org.uuid}/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.put( + f"/auth/api/admin/orgs/{test_org.uuid}/users/{test_user.uuid}/display-name", + 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/orgs/{test_org.uuid}/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/orgs/{test_org.uuid}/users/{test_user.uuid}/sessions/some-session", + headers={**auth_headers(regular_session_token), "Host": "localhost:4401"}, + ) + assert response.status_code == 403 diff --git a/tests/test_api.py b/tests/test_api.py new file mode 100644 index 0000000..7cf16ae --- /dev/null +++ b/tests/test_api.py @@ -0,0 +1,591 @@ +""" +Tests for the core API endpoints (/auth/api/). + +These tests cover: +- /auth/api/settings - Public settings endpoint +- /auth/api/validate - Session validation +- /auth/api/forward - Forward auth for reverse proxies +- /auth/api/logout - Session logout +- /auth/api/user-info - User information +- /auth/api/set-session - Set session from bearer token +""" + +from datetime import datetime, timezone + +import httpx +import pytest + +from tests.conftest import auth_headers + + +class TestSettingsEndpoint: + """Tests for GET /auth/api/settings""" + + @pytest.mark.asyncio + async def test_get_settings_returns_rp_info(self, client: httpx.AsyncClient): + """Settings endpoint should return RP configuration.""" + response = await client.get("/auth/api/settings") + assert response.status_code == 200 + data = response.json() + assert "rp_id" in data + assert "rp_name" in data + assert "session_cookie" in data + assert data["rp_id"] == "localhost" + assert data["rp_name"] == "Test RP" + assert data["session_cookie"] == "__Host-paskia" + + @pytest.mark.asyncio + async def test_settings_includes_ui_base_path(self, client: httpx.AsyncClient): + """Settings should include UI base path.""" + response = await client.get("/auth/api/settings") + data = response.json() + assert "ui_base_path" in data + + +class TestValidateEndpoint: + """Tests for POST /auth/api/validate""" + + @pytest.mark.asyncio + async def test_validate_without_auth_returns_401(self, client: httpx.AsyncClient): + """Validate without session should return 401.""" + response = await client.post("/auth/api/validate") + assert response.status_code == 401 + + @pytest.mark.asyncio + async def test_validate_with_invalid_token_returns_error( + self, client: httpx.AsyncClient + ): + """Validate with invalid token should return 4xx error.""" + response = await client.post( + "/auth/api/validate", + headers=auth_headers("invalid_token!!"), + ) + # Invalid token format returns 400, expired/missing returns 401 + assert response.status_code in (400, 401) + + @pytest.mark.asyncio + async def test_validate_with_valid_token_returns_200( + self, client: httpx.AsyncClient, session_token: str + ): + """Validate with valid session should return success.""" + response = await client.post( + "/auth/api/validate", + headers={**auth_headers(session_token), "Host": "localhost:4401"}, + ) + assert response.status_code == 200 + data = response.json() + assert data["valid"] is True + assert "user_uuid" in data + + @pytest.mark.asyncio + async def test_validate_with_permission_check( + self, client: httpx.AsyncClient, session_token: str + ): + """Validate should check permissions when provided.""" + # Admin user should pass admin permission check + response = await client.post( + "/auth/api/validate?perm=auth:admin", + headers={**auth_headers(session_token), "Host": "localhost:4401"}, + ) + assert response.status_code == 200 + + @pytest.mark.asyncio + async def test_validate_permission_denied_for_regular_user( + self, client: httpx.AsyncClient, regular_session_token: str + ): + """Regular user should fail admin permission check.""" + response = await client.post( + "/auth/api/validate?perm=auth:admin", + headers={ + **auth_headers(regular_session_token), + "Host": "localhost:4401", + }, + ) + assert response.status_code == 403 + + +class TestForwardEndpoint: + """Tests for GET /auth/api/forward (reverse proxy auth)""" + + @pytest.mark.asyncio + async def test_forward_without_auth_returns_401(self, client: httpx.AsyncClient): + """Forward auth without session should return 401.""" + response = await client.get("/auth/api/forward") + assert response.status_code == 401 + + @pytest.mark.asyncio + async def test_forward_401_json_response(self, client: httpx.AsyncClient): + """Forward auth 401 should include auth iframe info for JSON clients.""" + response = await client.get( + "/auth/api/forward", + headers={"Accept": "application/json"}, + ) + assert response.status_code == 401 + data = response.json() + assert "auth" in data + assert "iframe" in data["auth"] + assert "mode" in data["auth"] + assert data["auth"]["mode"] == "login" + + @pytest.mark.asyncio + async def test_forward_with_valid_session_returns_204( + self, client: httpx.AsyncClient, session_token: str + ): + """Forward auth with valid session should return 204 with headers.""" + response = await client.get( + "/auth/api/forward", + headers={**auth_headers(session_token), "Host": "localhost:4401"}, + ) + assert response.status_code == 204 + # Check Remote-* headers + assert "Remote-User" in response.headers + assert "Remote-Name" in response.headers + assert "Remote-Groups" in response.headers + assert "Remote-Org" in response.headers + + @pytest.mark.asyncio + async def test_forward_with_permission_returns_204( + self, client: httpx.AsyncClient, session_token: str + ): + """Forward auth with valid permission should return 204.""" + response = await client.get( + "/auth/api/forward?perm=auth:admin", + headers={**auth_headers(session_token), "Host": "localhost:4401"}, + ) + assert response.status_code == 204 + + @pytest.mark.asyncio + async def test_forward_permission_denied_returns_403( + self, client: httpx.AsyncClient, regular_session_token: str + ): + """Forward auth with missing permission should return 403.""" + response = await client.get( + "/auth/api/forward?perm=auth:admin", + headers={ + **auth_headers(regular_session_token), + "Host": "localhost:4401", + }, + ) + assert response.status_code == 403 + + @pytest.mark.asyncio + async def test_forward_403_json_includes_forbidden_mode( + self, client: httpx.AsyncClient, regular_session_token: str + ): + """403 response should include forbidden mode for iframe.""" + response = await client.get( + "/auth/api/forward?perm=auth:admin", + headers={ + **auth_headers(regular_session_token), + "Host": "localhost:4401", + "Accept": "application/json", + }, + ) + assert response.status_code == 403 + data = response.json() + assert "auth" in data + assert data["auth"]["mode"] == "forbidden" + + +class TestLogoutEndpoint: + """Tests for POST /auth/api/logout""" + + @pytest.mark.asyncio + async def test_logout_without_session_returns_message( + self, client: httpx.AsyncClient + ): + """Logout without session should return already logged out message.""" + response = await client.post("/auth/api/logout") + assert response.status_code == 200 + data = response.json() + assert "message" in data + assert "Already logged out" in data["message"] + + @pytest.mark.asyncio + async def test_logout_with_valid_session( + self, client: httpx.AsyncClient, session_token: str + ): + """Logout with valid session should succeed and clear session.""" + response = await client.post( + "/auth/api/logout", + headers={**auth_headers(session_token), "Host": "localhost:4401"}, + ) + assert response.status_code == 200 + data = response.json() + assert "Logged out successfully" in data["message"] + + # Verify session is no longer valid + response2 = await client.post( + "/auth/api/validate", + headers={**auth_headers(session_token), "Host": "localhost:4401"}, + ) + assert response2.status_code == 401 + + +class TestUserInfoEndpoint: + """Tests for POST /auth/api/user-info""" + + @pytest.mark.asyncio + async def test_user_info_without_auth_returns_401(self, client: httpx.AsyncClient): + """User info without session should return 401.""" + response = await client.post("/auth/api/user-info") + assert response.status_code == 401 + + @pytest.mark.asyncio + async def test_user_info_with_valid_session( + self, client: httpx.AsyncClient, session_token: str, test_user + ): + """User info with valid session should return user data.""" + response = await client.post( + "/auth/api/user-info", + headers={**auth_headers(session_token), "Host": "localhost:4401"}, + ) + assert response.status_code == 200 + data = response.json() + assert "user" in data + assert data["user"]["user_uuid"] == str(test_user.uuid) + assert data["user"]["user_name"] == test_user.display_name + + @pytest.mark.asyncio + async def test_user_info_includes_credentials( + self, client: httpx.AsyncClient, session_token: str + ): + """User info should include user's credentials.""" + response = await client.post( + "/auth/api/user-info", + headers={**auth_headers(session_token), "Host": "localhost:4401"}, + ) + assert response.status_code == 200 + data = response.json() + assert "credentials" in data + assert len(data["credentials"]) >= 1 + + @pytest.mark.asyncio + async def test_user_info_includes_sessions( + self, client: httpx.AsyncClient, session_token: str + ): + """User info should include user's active sessions.""" + response = await client.post( + "/auth/api/user-info", + headers={**auth_headers(session_token), "Host": "localhost:4401"}, + ) + assert response.status_code == 200 + data = response.json() + assert "sessions" in data + assert len(data["sessions"]) >= 1 + + @pytest.mark.asyncio + async def test_user_info_includes_permissions( + self, client: httpx.AsyncClient, session_token: str + ): + """User info should include user's permissions.""" + response = await client.post( + "/auth/api/user-info", + headers={**auth_headers(session_token), "Host": "localhost:4401"}, + ) + assert response.status_code == 200 + data = response.json() + assert "permissions" in data + + +class TestSetSessionEndpoint: + """Tests for POST /auth/api/set-session""" + + @pytest.mark.asyncio + async def test_set_session_without_bearer_returns_403( + self, client: httpx.AsyncClient + ): + """Set session without bearer token should return 403.""" + response = await client.post("/auth/api/set-session") + assert response.status_code == 403 + + @pytest.mark.asyncio + async def test_set_session_with_valid_bearer_token( + self, client: httpx.AsyncClient, session_token: str + ): + """Set session with valid bearer token should set cookie.""" + response = await client.post( + "/auth/api/set-session", + headers={ + "Authorization": f"Bearer {session_token}", + "Host": "localhost:4401", + }, + ) + assert response.status_code == 200 + data = response.json() + assert "user_uuid" in data + # Check that Set-Cookie header is present + assert "set-cookie" in response.headers + + +class TestErrorHandling: + """Tests for API error handling""" + + @pytest.mark.asyncio + async def test_invalid_endpoint_returns_404(self, client: httpx.AsyncClient): + """Request to non-existent endpoint should return 404.""" + response = await client.get("/auth/api/nonexistent") + assert response.status_code == 404 + + @pytest.mark.asyncio + async def test_error_response_on_bad_token(self, client: httpx.AsyncClient): + """Bad token should return error response.""" + response = await client.post( + "/auth/api/validate", + headers=auth_headers("expired_token!"), + ) + # Malformed token returns 400, expired returns 401 + assert response.status_code in (400, 401) + + +class TestForwardAuthHtmlResponse: + """Tests for forward auth HTML responses""" + + @pytest.mark.asyncio + async def test_forward_401_html_response(self, client: httpx.AsyncClient): + """Forward auth 401 should return HTML page for browser requests.""" + response = await client.get( + "/auth/api/forward", + headers={"Accept": "text/html"}, + ) + assert response.status_code == 401 + assert "text/html" in response.headers.get("content-type", "") + # HTML response should contain the mode data attribute + assert b"data-mode" in response.content or b"mode" in response.content + + @pytest.mark.asyncio + async def test_forward_403_html_response( + self, client: httpx.AsyncClient, regular_session_token: str + ): + """Forward auth 403 should return HTML page for browser requests.""" + response = await client.get( + "/auth/api/forward?perm=auth:admin", + headers={ + **auth_headers(regular_session_token), + "Host": "localhost:4401", + "Accept": "text/html", + }, + ) + assert response.status_code == 403 + assert "text/html" in response.headers.get("content-type", "") + + @pytest.mark.asyncio + async def test_forward_with_expired_session_clears_cookie( + self, client: httpx.AsyncClient + ): + """Forward auth with expired session should trigger clear_session path.""" + # Use a well-formed but non-existent session token + fake_token = "aaaaaaaaaaaaaaaa" # Exactly 16 characters + response = await client.get( + "/auth/api/forward", + headers={ + **auth_headers(fake_token), + "Host": "localhost:4401", + "Accept": "application/json", + }, + ) + assert response.status_code == 401 + # Verify the response contains auth info for re-login + data = response.json() + assert "auth" in data + assert data["auth"]["mode"] == "login" + + +class TestUserInfoWithResetToken: + """Tests for user-info endpoint with reset tokens""" + + @pytest.mark.asyncio + async def test_user_info_with_invalid_reset_token(self, client: httpx.AsyncClient): + """User info with invalid reset token format should return 401.""" + # Invalid format - not a well-formed passphrase (wrong separator) + response = await client.post( + "/auth/api/user-info?reset=invalid-token-format", + ) + # Invalid format raises ValueError which gets converted to 401 HTTPException + assert response.status_code == 401 + data = response.json() + assert "Invalid reset token" in data["detail"] + + @pytest.mark.asyncio + async def test_user_info_with_nonexistent_reset_token( + self, client: httpx.AsyncClient + ): + """User info with well-formed but non-existent reset token should return 401.""" + # We need a well-formed passphrase that doesn't exist in DB + from paskia.util.passphrase import generate + + fake_token = generate() # Generates a well-formed token + response = await client.post( + f"/auth/api/user-info?reset={fake_token}", + ) + # Should return 401 for non-existent token + assert response.status_code == 401 + + @pytest.mark.asyncio + async def test_user_info_with_valid_reset_token( + self, client: httpx.AsyncClient, reset_token: str, test_user + ): + """User info with valid reset token should return minimal user info.""" + response = await client.post( + f"/auth/api/user-info?reset={reset_token}", + ) + assert response.status_code == 200 + data = response.json() + assert "user" in data + + +class TestSetSessionErrors: + """Tests for set-session error cases""" + + @pytest.mark.asyncio + async def test_set_session_with_invalid_bearer_token( + self, client: httpx.AsyncClient + ): + """Set session with invalid (malformed) bearer token should return 400.""" + response = await client.post( + "/auth/api/set-session", + headers={ + "Authorization": "Bearer invalid_token_here", # Wrong length (18 chars) + "Host": "localhost:4401", + }, + ) + # Invalid token format returns 400 + assert response.status_code == 400 + + @pytest.mark.asyncio + async def test_set_session_with_nonexistent_token(self, client: httpx.AsyncClient): + """Set session with valid format but non-existent token should fail.""" + # Use a well-formed 16-char token that doesn't exist in DB + fake_token = "aaaaaaaaaaaaaaaa" # Exactly 16 characters + response = await client.post( + "/auth/api/set-session", + headers={ + "Authorization": f"Bearer {fake_token}", + "Host": "localhost:4401", + }, + ) + # Non-existent session returns 400 (ValueError -> 400) + assert response.status_code == 400 + + +class TestValidateSessionRefresh: + """Tests for session refresh behavior in validate endpoint""" + + @pytest.mark.asyncio + async def test_validate_does_not_refresh_within_interval( + self, client: httpx.AsyncClient, session_token: str + ): + """Validate should not refresh session if within refresh interval.""" + # First call - may or may not refresh depending on session age + response1 = await client.post( + "/auth/api/validate", + headers={**auth_headers(session_token), "Host": "localhost:4401"}, + ) + assert response1.status_code == 200 + + # Second call immediately after - should NOT refresh (within 5 min interval) + response2 = await client.post( + "/auth/api/validate", + headers={**auth_headers(session_token), "Host": "localhost:4401"}, + ) + assert response2.status_code == 200 + data = response2.json() + # Session shouldn't be renewed since we're within the refresh interval + assert data["renewed"] is False + + @pytest.mark.asyncio + async def test_validate_with_expired_session_during_refresh( + self, client: httpx.AsyncClient, test_db + ): + """Validate should handle session expiry during refresh attempt.""" + from paskia.util.tokens import create_token + + # Create a token but don't create a session for it + token = create_token() + response = await client.post( + "/auth/api/validate", + headers={**auth_headers(token), "Host": "localhost:4401"}, + ) + # Should return 401 for non-existent session + assert response.status_code == 401 + + @pytest.mark.asyncio + async def test_validate_session_refresh_fails_concurrent_logout( + self, + client: httpx.AsyncClient, + test_db, + test_user, + test_credential, + ): + """Validate should return 401 if session disappears during refresh.""" + from datetime import timedelta + + from paskia.util.tokens import create_token, session_key + + # Create a session with an old renewed time to trigger refresh + token = create_token() + old_time = datetime.now(timezone.utc) - timedelta(minutes=10) + await test_db.create_session( + user_uuid=test_user.uuid, + credential_uuid=test_credential.uuid, + key=session_key(token), + host="localhost:4401", + ip="127.0.0.1", + user_agent="pytest", + renewed=old_time, + ) + + # Delete the session right before validate tries to refresh + await test_db.delete_session(session_key(token)) + + response = await client.post( + "/auth/api/validate", + headers={**auth_headers(token), "Host": "localhost:4401"}, + ) + # Session was found initially but disappeared during refresh + assert response.status_code == 401 + + +class TestForwardAuthMaxAge: + """Tests for forward auth max_age parameter""" + + @pytest.mark.asyncio + async def test_forward_with_max_age_recent_auth( + self, client: httpx.AsyncClient, session_token: str + ): + """Forward auth with max_age should pass for recent authentication.""" + response = await client.get( + "/auth/api/forward?max_age=1h", + headers={**auth_headers(session_token), "Host": "localhost:4401"}, + ) + # Recently authenticated session should pass + assert response.status_code == 204 + + @pytest.mark.asyncio + async def test_forward_with_invalid_max_age_format( + self, client: httpx.AsyncClient, session_token: str + ): + """Forward auth with invalid max_age format should log warning but succeed.""" + response = await client.get( + "/auth/api/forward?max_age=invalid", + headers={**auth_headers(session_token), "Host": "localhost:4401"}, + ) + # Invalid format is logged but request proceeds + assert response.status_code == 204 + + +class TestValidateWithMaxAge: + """Tests for validate endpoint with max_age parameter""" + + @pytest.mark.asyncio + async def test_validate_with_max_age( + self, client: httpx.AsyncClient, session_token: str + ): + """Validate with max_age should check authentication age.""" + response = await client.post( + "/auth/api/validate?max_age=1h", + headers={**auth_headers(session_token), "Host": "localhost:4401"}, + ) + # This exercises the max_age path - but isn't defined in validate + # Actually validate doesn't have max_age - this tests that unknown params are ignored + assert response.status_code == 200 diff --git a/tests/test_user.py b/tests/test_user.py new file mode 100644 index 0000000..04ab84a --- /dev/null +++ b/tests/test_user.py @@ -0,0 +1,184 @@ +""" +Tests for the user API endpoints (/auth/api/user/). + +These tests cover user self-service operations: +- Display name update +- Logout all sessions +- Session management (delete specific session) +- Credential management (delete credential) +- Device addition link creation +""" + +import httpx +import pytest + +from tests.conftest import auth_headers + + +class TestUserDisplayName: + """Tests for PUT /auth/api/user/display-name""" + + @pytest.mark.asyncio + async def test_update_display_name_requires_auth(self, client: httpx.AsyncClient): + """Update display name without auth should return 401.""" + response = await client.put( + "/auth/api/user/display-name", + json={"display_name": "New Name"}, + ) + assert response.status_code == 401 + + @pytest.mark.asyncio + async def test_update_display_name_success( + self, client: httpx.AsyncClient, session_token: str + ): + """User should be able to update their display name.""" + response = await client.put( + "/auth/api/user/display-name", + json={"display_name": "Updated 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_display_name_empty_fails( + self, client: httpx.AsyncClient, session_token: str + ): + """Empty display name should fail.""" + response = await client.put( + "/auth/api/user/display-name", + json={"display_name": ""}, + headers={**auth_headers(session_token), "Host": "localhost:4401"}, + ) + assert response.status_code == 400 + + @pytest.mark.asyncio + async def test_update_display_name_too_long_fails( + self, client: httpx.AsyncClient, session_token: str + ): + """Display name over 64 chars should fail.""" + long_name = "x" * 100 + response = await client.put( + "/auth/api/user/display-name", + json={"display_name": long_name}, + headers={**auth_headers(session_token), "Host": "localhost:4401"}, + ) + assert response.status_code == 400 + + +class TestUserLogoutAll: + """Tests for POST /auth/api/user/logout-all""" + + @pytest.mark.asyncio + async def test_logout_all_requires_auth(self, client: httpx.AsyncClient): + """Logout all without auth should return already logged out.""" + response = await client.post("/auth/api/user/logout-all") + assert response.status_code == 200 + data = response.json() + assert "Already logged out" in data["message"] + + @pytest.mark.asyncio + async def test_logout_all_success( + self, client: httpx.AsyncClient, session_token: str + ): + """User should be able to logout from all sessions.""" + response = await client.post( + "/auth/api/user/logout-all", + headers={**auth_headers(session_token), "Host": "localhost:4401"}, + ) + assert response.status_code == 200 + data = response.json() + assert "Logged out" in data["message"] + + # Verify session is invalidated + response2 = await client.post( + "/auth/api/validate", + headers={**auth_headers(session_token), "Host": "localhost:4401"}, + ) + assert response2.status_code == 401 + + +class TestUserSessionManagement: + """Tests for DELETE /auth/api/user/session/{session_id}""" + + @pytest.mark.asyncio + async def test_delete_session_requires_auth(self, client: httpx.AsyncClient): + """Delete session without auth should return 401.""" + response = await client.delete("/auth/api/user/session/fake-session-id") + assert response.status_code == 401 + + @pytest.mark.asyncio + async def test_delete_invalid_session_fails( + self, client: httpx.AsyncClient, session_token: str + ): + """Deleting invalid session ID should fail.""" + response = await client.delete( + "/auth/api/user/session/invalid-session-id", + headers={**auth_headers(session_token), "Host": "localhost:4401"}, + ) + assert response.status_code == 400 + + @pytest.mark.asyncio + async def test_delete_nonexistent_session_returns_404( + self, client: httpx.AsyncClient, session_token: str + ): + """Deleting a properly-formatted but nonexistent session returns 404.""" + # Use a valid format but non-existent session key + fake_session = "c2Vzc0FBQUFBQUFBQUFBQUFBQUE" # base64 of "sessAAAAAAAAAAAAAAAA" + response = await client.delete( + f"/auth/api/user/session/{fake_session}", + headers={**auth_headers(session_token), "Host": "localhost:4401"}, + ) + assert response.status_code == 404 + + +class TestUserCredentialManagement: + """Tests for DELETE /auth/api/user/credential/{uuid}""" + + @pytest.mark.asyncio + async def test_delete_credential_requires_auth(self, client: httpx.AsyncClient): + """Delete credential without auth should return 401.""" + response = await client.delete( + "/auth/api/user/credential/00000000-0000-0000-0000-000000000000" + ) + assert response.status_code == 401 + + @pytest.mark.asyncio + async def test_delete_credential_success( + self, client: httpx.AsyncClient, session_token: str, test_credential + ): + """User can delete their credential.""" + response = await client.delete( + f"/auth/api/user/credential/{test_credential.uuid}", + headers={**auth_headers(session_token), "Host": "localhost:4401"}, + ) + # Note: API allows deleting even the only credential + assert response.status_code == 200 + data = response.json() + assert "deleted" in data["message"].lower() + + +class TestUserCreateLink: + """Tests for POST /auth/api/user/create-link""" + + @pytest.mark.asyncio + async def test_create_link_requires_auth(self, client: httpx.AsyncClient): + """Create link without auth should return 401.""" + response = await client.post("/auth/api/user/create-link") + assert response.status_code == 401 + + @pytest.mark.asyncio + async def test_create_link_success( + self, client: httpx.AsyncClient, session_token: str + ): + """User should be able to create a device addition link.""" + response = await client.post( + "/auth/api/user/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 + assert "message" in data