Spaces:
Running
Running
| from __future__ import annotations | |
| import asyncio | |
| from pathlib import Path | |
| from fastapi.testclient import TestClient | |
| from app.core.config import Settings | |
| from app.security.models import WorkspaceMembership | |
| from app.security.schemas import APIKeyCreate | |
| from main import create_app | |
| def _settings(tmp_path: Path) -> Settings: | |
| return Settings( | |
| _env_file=None, | |
| auth_enabled=True, | |
| studio_identity_secret="test-only-studio-identity-secret-material", | |
| database_url=f"sqlite+aiosqlite:///{tmp_path / 'security.db'}", | |
| social_database_url=f"sqlite+aiosqlite:///{tmp_path / 'social.db'}", | |
| social_auto_migrate=True, | |
| social_worker_enabled=False, | |
| social_oauth_encryption_key="test-only-encryption-material", | |
| temp_dir=tmp_path / "temp", | |
| output_dir=tmp_path / "outputs", | |
| cleanup_interval_seconds=3600, | |
| whisper_model="tiny", | |
| max_workers=1, | |
| ) | |
| def test_api_key_admin_routes_are_workspace_scoped_and_role_gated(tmp_path: Path) -> None: | |
| settings = _settings(tmp_path) | |
| application = create_app(settings) | |
| container = application.state.container | |
| asyncio.run(container.security_database.initialize()) | |
| owner_record, owner_secret = asyncio.run( | |
| container.api_keys.create( | |
| APIKeyCreate(name="Workspace owner", environment="test", role="admin", scopes=["admin"]), | |
| created_by="tests", | |
| ) | |
| ) | |
| owner_context = asyncio.run(container.api_keys.authenticate(owner_secret)) | |
| same_workspace_record, same_workspace_secret = asyncio.run( | |
| container.api_keys.create( | |
| APIKeyCreate(name="Same workspace key", environment="test", role=None, scopes=[]), | |
| created_by=owner_record.id, | |
| workspace_id=owner_context.workspace_id, | |
| user_id=owner_context.user_id, | |
| ) | |
| ) | |
| other_record, _ = asyncio.run( | |
| container.api_keys.create( | |
| APIKeyCreate(name="Other workspace key", environment="test", role="admin", scopes=["admin"]), | |
| created_by="tests", | |
| ) | |
| ) | |
| headers = {"Authorization": f"Bearer {owner_secret}"} | |
| with TestClient(application) as client: | |
| listed = client.get("/v1/api-keys", headers=headers) | |
| assert listed.status_code == 200 | |
| assert listed.json()["total"] == 2 | |
| ids = {item["id"] for item in listed.json()["items"]} | |
| assert ids == {owner_record.id, same_workspace_record.id} | |
| assert client.get(f"/v1/api-keys/{other_record.id}", headers=headers).status_code == 404 | |
| # A regular key in the owner's workspace must not inherit key-admin | |
| # privileges merely because its persisted membership role is owner. | |
| regular_key_response = client.get( | |
| "/v1/api-keys", | |
| headers={"Authorization": f"Bearer {same_workspace_secret}"}, | |
| ) | |
| assert regular_key_response.status_code == 403 | |
| async def downgrade_membership() -> None: | |
| async with container.security_database.session() as session: | |
| membership = await session.get(WorkspaceMembership, owner_context.membership_id) | |
| assert membership is not None | |
| membership.role = "member" | |
| await session.commit() | |
| asyncio.run(downgrade_membership()) | |
| denied = client.get("/v1/api-keys", headers=headers) | |
| assert denied.status_code == 403 | |
| asyncio.run(container.security_database.close()) | |