test: cover inactive Studio tenant identities

tests/test_studio_identity.py

@@ -17,8 +17,13 @@ from app.security.tenancy import TenantPrincipal, TenantService
 SECRET = "a" * 64
 
 
-def signed_headers(*, timestamp: int = 1_800_000_000, subject: str = "supabase-user-1",
-                   email: str = "user@example.com", name: str = "Studio User") -> dict[str, str]:
+def signed_headers(
+    *,
+    timestamp: int = 1_800_000_000,
+    subject: str = "supabase-user-1",
+    email: str = "user@example.com",
+    name: str = "Studio User",
+) -> dict[str, str]:
     signature = hmac.new(
         SECRET.encode(),
         "\n".join((str(timestamp), subject, email, name)).encode(),
@@ -39,11 +44,14 @@ def test_valid_studio_identity_signature_is_accepted() -> None:
     assert identity.email == "user@example.com"
 
 
-@pytest.mark.parametrize("changed_header,changed_value", [
-    ("x-mr-studio-subject", "another-user"),
-    ("x-mr-studio-email", "attacker@example.com"),
-    ("x-mr-studio-name", "Administrator"),
-])
+@pytest.mark.parametrize(
+    ("changed_header", "changed_value"),
+    [
+        ("x-mr-studio-subject", "another-user"),
+        ("x-mr-studio-email", "attacker@example.com"),
+        ("x-mr-studio-name", "Administrator"),
+    ],
+)
 def test_tampered_identity_fields_are_rejected(changed_header: str, changed_value: str) -> None:
     headers = signed_headers()
     headers[changed_header] = changed_value
@@ -53,7 +61,9 @@ def test_tampered_identity_fields_are_rejected(changed_header: str, changed_valu
 
 def test_expired_identity_is_rejected() -> None:
     with pytest.raises(ValueError, match="expired"):
-        verify_studio_identity(signed_headers(timestamp=1_800_000_000), SECRET, now=1_800_000_301)
+        verify_studio_identity(
+            signed_headers(timestamp=1_800_000_000), SECRET, now=1_800_000_301
+        )
 
 
 def test_incomplete_envelope_and_short_secret_are_rejected() -> None:
@@ -65,18 +75,28 @@ def test_incomplete_envelope_and_short_secret_are_rejected() -> None:
 
 def test_shared_admin_key_cannot_escalate_viewer_membership() -> None:
     context = AuthContext(
-        api_key_id="key", key_name="Studio proxy", key_prefix="mp_live_12345678",
-        environment="live", role="admin", scopes=frozenset({"admin", "projects:delete", "projects:read"}),
-        requests_per_minute=100, concurrent_jobs=10, uploads_per_hour=20,
-        processing_bytes_per_day=1000, expires_at=None,
+        api_key_id="key",
+        key_name="Studio proxy",
+        key_prefix="mp_live_12345678",
+        environment="live",
+        role="admin",
+        scopes=frozenset({"admin", "projects:delete", "projects:read"}),
+        requests_per_minute=100,
+        concurrent_jobs=10,
+        uploads_per_hour=20,
+        processing_bytes_per_day=1000,
+        expires_at=None,
     )
     principal = TenantPrincipal(
-        workspace_id="workspace-a", user_id="user-a", membership_id="membership-a",
+        workspace_id="workspace-a",
+        user_id="user-a",
+        membership_id="membership-a",
         membership_role="member",
     )
     scoped = constrain_context_to_membership(context, principal)
     assert scoped.workspace_id == "workspace-a"
     assert scoped.user_id == "user-a"
+    assert scoped.membership_role == "member"
     assert "admin" not in scoped.scopes
     assert "projects:delete" not in scoped.scopes
     assert "projects:read" in scoped.scopes
@@ -84,27 +104,39 @@ def test_shared_admin_key_cannot_escalate_viewer_membership() -> None:
 
 def test_unknown_membership_role_fails_closed() -> None:
     context = AuthContext(
-        api_key_id="key", key_name="Studio proxy", key_prefix="mp_live_12345678",
-        environment="live", role="admin", scopes=frozenset({"admin"}),
-        requests_per_minute=100, concurrent_jobs=10, uploads_per_hour=20,
-        processing_bytes_per_day=1000, expires_at=None,
+        api_key_id="key",
+        key_name="Studio proxy",
+        key_prefix="mp_live_12345678",
+        environment="live",
+        role="admin",
+        scopes=frozenset({"admin"}),
+        requests_per_minute=100,
+        concurrent_jobs=10,
+        uploads_per_hour=20,
+        processing_bytes_per_day=1000,
+        expires_at=None,
     )
     principal = TenantPrincipal(
-        workspace_id="workspace-a", user_id="user-a", membership_id="membership-a",
+        workspace_id="workspace-a",
+        user_id="user-a",
+        membership_id="membership-a",
         membership_role="superuser",
     )
     with pytest.raises(ValueError, match="Unsupported"):
         constrain_context_to_membership(context, principal)
 
 
-
-async def _seed_subject(database: SecurityDatabase, subject: str, workspace_count: int) -> str:
+async def _seed_subject(
+    database: SecurityDatabase, subject: str, workspace_count: int
+) -> tuple[str, list[str]]:
     user_id = str(uuid4())
+    workspace_ids: list[str] = []
     async with database.session() as session:
         user = User(id=user_id, subject=subject, display_name="Test user")
         rows = [user]
         for index in range(workspace_count):
             workspace_id = str(uuid4())
+            workspace_ids.append(workspace_id)
             rows.append(
                 Workspace(
                     id=workspace_id,
@@ -123,19 +155,24 @@ async def _seed_subject(database: SecurityDatabase, subject: str, workspace_coun
             )
         session.add_all(rows)
         await session.commit()
-    return user_id
+    return user_id, workspace_ids
 
 
-async def test_studio_subject_resolves_one_active_membership(tmp_path) -> None:
+async def _database(tmp_path, filename: str) -> SecurityDatabase:
     database = SecurityDatabase(
-        f"sqlite+aiosqlite:///{tmp_path / 'studio-tenant.db'}", auto_migrate=True
+        f"sqlite+aiosqlite:///{tmp_path / filename}", auto_migrate=True
     )
     await database.initialize()
+    return database
+
+
+async def test_studio_subject_resolves_one_active_membership(tmp_path) -> None:
+    database = await _database(tmp_path, "studio-tenant.db")
     try:
-        user_id = await _seed_subject(database, "supabase-subject-one", 1)
+        user_id, workspace_ids = await _seed_subject(database, "supabase-subject-one", 1)
         principal = await TenantService(database).resolve_studio_subject("supabase-subject-one")
         assert principal.user_id == user_id
-        assert principal.workspace_id
+        assert principal.workspace_id == workspace_ids[0]
         assert principal.membership_id
         assert principal.membership_role == "member"
     finally:
@@ -143,10 +180,7 @@ async def test_studio_subject_resolves_one_active_membership(tmp_path) -> None:
 
 
 async def test_studio_subject_fails_closed_for_ambiguous_memberships(tmp_path) -> None:
-    database = SecurityDatabase(
-        f"sqlite+aiosqlite:///{tmp_path / 'studio-ambiguous.db'}", auto_migrate=True
-    )
-    await database.initialize()
+    database = await _database(tmp_path, "studio-ambiguous.db")
     try:
         await _seed_subject(database, "supabase-subject-many", 2)
         with pytest.raises(ForbiddenError):
@@ -156,12 +190,39 @@ async def test_studio_subject_fails_closed_for_ambiguous_memberships(tmp_path) -
 
 
 async def test_unknown_studio_subject_is_not_auto_provisioned(tmp_path) -> None:
-    database = SecurityDatabase(
-        f"sqlite+aiosqlite:///{tmp_path / 'studio-unknown.db'}", auto_migrate=True
-    )
-    await database.initialize()
+    database = await _database(tmp_path, "studio-unknown.db")
     try:
         with pytest.raises(ForbiddenError):
             await TenantService(database).resolve_studio_subject("not-provisioned")
     finally:
         await database.close()
+
+
+@pytest.mark.parametrize("inactive_entity", ["user", "workspace", "membership"])
+async def test_inactive_identity_or_membership_is_denied(tmp_path, inactive_entity: str) -> None:
+    database = await _database(tmp_path, f"inactive-{inactive_entity}.db")
+    try:
+        user_id, workspace_ids = await _seed_subject(
+            database, f"supabase-inactive-{inactive_entity}", 1
+        )
+        async with database.session() as session:
+            if inactive_entity == "user":
+                entity = await session.get(User, user_id)
+            elif inactive_entity == "workspace":
+                entity = await session.get(Workspace, workspace_ids[0])
+            else:
+                entity = await session.scalar(
+                    __import__("sqlalchemy").select(WorkspaceMembership).where(
+                        WorkspaceMembership.user_id == user_id,
+                        WorkspaceMembership.workspace_id == workspace_ids[0],
+                    )
+                )
+            assert entity is not None
+            entity.status = "disabled"
+            await session.commit()
+        with pytest.raises(ForbiddenError):
+            await TenantService(database).resolve_studio_subject(
+                f"supabase-inactive-{inactive_entity}"
+            )
+    finally:
+        await database.close()