Spaces:
Running
Running
File size: 3,475 Bytes
75d4046 f73fb2e 75d4046 18400bf 75d4046 f73fb2e 75d4046 18400bf 75d4046 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 | 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())
|