Add verified Studio identity authorization (#7)

- Add verified Studio identity authorization (8ce9df18095dfcf741801d791643e4562c0d67fb)
- Resolve Studio subjects through active memberships (64d5cff23a23dd9810ec3bac00b9dcbd6fb2d567)
- Add Studio identity HMAC verifier (00c3273205d40650fd696fe6a5342fa54f6ddce9)
- Test Studio identity signature validation (5bceb55f5a456c6e086cfe83995f398dace91118)
- Enforce role-limited Studio authorization (c8a03912905d61e4b068453fd932bf3d3bc5fa9d)
- Use membership-constrained Studio context (3d2941fbc1c0c3b2b394e7477cbd015b0c934092)
- Test Studio identity and privilege boundaries (85cfd93573829c0a4d9ed909cc769d509dd59e4b)
- Test Studio tenant membership resolution (b97b02f73a33c92e46427a2244d5206811c871c6)
- Organize Studio identity regression test imports (a760c151a2641ff55f12be447ffc77d915a3e823)
- Match Studio HMAC canonicalization (a4fde0f17460628ababf1a5dc3d2f092d690acaa)
- Align identity tests with Studio signature format (e446305c6a659b2909b63ec65959ae8cffde354f)

app/security/middleware.py

@@ -17,6 +17,7 @@ from app.security.errors import ForbiddenError, RateLimitError, UnauthorizedErro
 from app.security.policy import ScopePolicy
 from app.security.rate_limit import APIKeyRateLimiter, RateLimitLease
 from app.security.service import APIKeyService
+from app.security.studio_identity import constrain_context_to_membership, verify_studio_identity
 
 logger = get_logger(__name__)
 
@@ -56,6 +57,32 @@ class APIKeyAuthenticationMiddleware(BaseHTTPMiddleware):
         try:
             api_key = self._bearer_token(request.headers.get("authorization"))
             context = await self.api_keys.authenticate(api_key)
+            studio_header_names = (
+                "x-mr-studio-subject",
+                "x-mr-studio-email",
+                "x-mr-studio-name",
+                "x-mr-studio-timestamp",
+                "x-mr-studio-signature",
+            )
+            if any(request.headers.get(name) is not None for name in studio_header_names):
+                secret = (
+                    self.settings.studio_identity_secret.get_secret_value()
+                    if self.settings.studio_identity_secret
+                    else ""
+                )
+                try:
+                    identity = verify_studio_identity(request.headers, secret)
+                except ValueError as exc:
+                    logger.warning("rejected Studio identity envelope: %s", str(exc))
+                    raise UnauthorizedError from exc
+                principal = await self.api_keys.tenants.resolve_studio_subject(identity.subject)
+                try:
+                    # The shared key authenticates transport only; persisted
+                    # membership capabilities constrain its effective scopes.
+                    context = constrain_context_to_membership(context, principal)
+                except ValueError as exc:
+                    logger.warning("rejected Studio membership role: %s", str(exc))
+                    raise ForbiddenError from exc
             request.state.auth = context
             context_token = auth_context.set(context)
             required_scope = await self.policy.required_scope(request)

app/security/studio_identity.py

@@ -0,0 +1,90 @@
+from __future__ import annotations
+
+import hmac
+import re
+import time
+from collections.abc import Mapping
+from dataclasses import dataclass, replace
+
+from app.security.context import AuthContext
+from app.security.scopes import DEFAULT_ROLE_SCOPES
+from app.security.tenancy import TenantPrincipal
+
+_HEX_SHA256 = re.compile(r"^[a-f0-9]{64}$")
+
+
+@dataclass(frozen=True, slots=True)
+class StudioIdentity:
+    subject: str
+    email: str
+    name: str
+    timestamp: int
+
+
+def verify_studio_identity(
+    headers: Mapping[str, str],
+    secret: str,
+    *,
+    now: float | None = None,
+    max_age_seconds: int = 300,
+) -> StudioIdentity:
+    """Verify the Studio proxy's HMAC envelope; never trust unsigned identity headers."""
+    names = {
+        "subject": "x-mr-studio-subject",
+        "email": "x-mr-studio-email",
+        "name": "x-mr-studio-name",
+        "timestamp": "x-mr-studio-timestamp",
+        "signature": "x-mr-studio-signature",
+    }
+    values = {key: (headers.get(header) or "").strip() for key, header in names.items()}
+    if not secret or len(secret.encode("utf-8")) < 32:
+        raise ValueError("Studio identity verification is not configured.")
+    if any(not value for value in values.values()):
+        raise ValueError("Studio identity envelope is incomplete.")
+    subject, email, name, timestamp_text, signature = (
+        values["subject"], values["email"], values["name"], values["timestamp"], values["signature"]
+    )
+    if len(subject) > 255 or any(ord(ch) < 32 for ch in subject):
+        raise ValueError("Studio subject is malformed.")
+    if len(email) > 320 or len(name) > 160:
+        raise ValueError("Studio profile fields exceed their limits.")
+    if not timestamp_text.isascii() or not timestamp_text.isdecimal() or len(timestamp_text) > 12:
+        raise ValueError("Studio timestamp is malformed.")
+    if not _HEX_SHA256.fullmatch(signature):
+        raise ValueError("Studio signature is malformed.")
+    timestamp = int(timestamp_text)
+    current = time.time() if now is None else now
+    if abs(current - timestamp) > max_age_seconds:
+        raise ValueError("Studio identity envelope has expired.")
+    canonical = "\n".join((timestamp_text, subject, email, name)).encode("utf-8")
+    expected = hmac.digest(secret.encode("utf-8"), canonical, "sha256").hex()
+    if not hmac.compare_digest(expected, signature):
+        raise ValueError("Studio identity signature is invalid.")
+    return StudioIdentity(subject=subject, email=email, name=name, timestamp=timestamp)
+
+
+def constrain_context_to_membership(
+    context: AuthContext, principal: TenantPrincipal
+) -> AuthContext:
+    """Intersect API-key scopes with persisted workspace-role capabilities."""
+    raw_role = principal.membership_role.strip().lower()
+    role_map = {
+        "owner": "developer",
+        "admin": "developer",
+        "developer": "developer",
+        "operator": "operator",
+        "viewer": "viewer",
+        "member": "viewer",
+    }
+    effective_role = role_map.get(raw_role)
+    if effective_role is None:
+        raise ValueError("Unsupported workspace membership role.")
+    return replace(
+        context,
+        role=effective_role,
+        scopes=frozenset(context.scopes.intersection(DEFAULT_ROLE_SCOPES[effective_role])),
+        workspace_id=principal.workspace_id,
+        user_id=principal.user_id,
+        membership_id=principal.membership_id,
+        membership_role=raw_role,
+    )

app/security/tenancy.py

@@ -100,6 +100,38 @@ class TenantService:
                 membership_role=membership.role,
             )
 
+    async def resolve_studio_subject(self, subject: str) -> TenantPrincipal:
+        """Resolve a verified identity only when it has exactly one active workspace."""
+        async with self.database.session() as session:
+            user = await session.scalar(
+                select(User).where(User.subject == subject, User.status == "active")
+            )
+            if user is None:
+                raise ForbiddenError
+            rows = list(
+                (
+                    await session.execute(
+                        select(WorkspaceMembership, Workspace)
+                        .join(Workspace, Workspace.id == WorkspaceMembership.workspace_id)
+                        .where(
+                            WorkspaceMembership.user_id == user.id,
+                            WorkspaceMembership.status == "active",
+                            Workspace.status == "active",
+                        )
+                    )
+                ).all()
+            )
+            # Never guess a workspace when membership is absent or ambiguous.
+            if len(rows) != 1:
+                raise ForbiddenError
+            membership, workspace = rows[0]
+            return TenantPrincipal(
+                workspace_id=workspace.id,
+                user_id=user.id,
+                membership_id=membership.id,
+                membership_role=membership.role,
+            )
+
     async def bind_api_key(
         self, *, api_key_id: str, workspace_id: str, user_id: str
     ) -> TenantPrincipal:

tests/test_studio_identity.py

@@ -0,0 +1,167 @@
+from __future__ import annotations
+
+import hashlib
+import hmac
+from uuid import uuid4
+
+import pytest
+
+from app.security.context import AuthContext
+from app.security.database import SecurityDatabase
+from app.security.errors import ForbiddenError
+from app.security.models import User, Workspace, WorkspaceMembership
+from app.security.studio_identity import constrain_context_to_membership, verify_studio_identity
+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]:
+    signature = hmac.new(
+        SECRET.encode(),
+        "\n".join((str(timestamp), subject, email, name)).encode(),
+        hashlib.sha256,
+    ).hexdigest()
+    return {
+        "x-mr-studio-subject": subject,
+        "x-mr-studio-email": email,
+        "x-mr-studio-name": name,
+        "x-mr-studio-timestamp": str(timestamp),
+        "x-mr-studio-signature": signature,
+    }
+
+
+def test_valid_studio_identity_signature_is_accepted() -> None:
+    identity = verify_studio_identity(signed_headers(), SECRET, now=1_800_000_000)
+    assert identity.subject == "supabase-user-1"
+    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"),
+])
+def test_tampered_identity_fields_are_rejected(changed_header: str, changed_value: str) -> None:
+    headers = signed_headers()
+    headers[changed_header] = changed_value
+    with pytest.raises(ValueError, match="signature"):
+        verify_studio_identity(headers, SECRET, now=1_800_000_000)
+
+
+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)
+
+
+def test_incomplete_envelope_and_short_secret_are_rejected() -> None:
+    with pytest.raises(ValueError, match="incomplete"):
+        verify_studio_identity({"x-mr-studio-subject": "user"}, SECRET, now=1_800_000_000)
+    with pytest.raises(ValueError, match="not configured"):
+        verify_studio_identity(signed_headers(), "too-short", now=1_800_000_000)
+
+
+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,
+    )
+    principal = TenantPrincipal(
+        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 "admin" not in scoped.scopes
+    assert "projects:delete" not in scoped.scopes
+    assert "projects:read" in scoped.scopes
+
+
+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,
+    )
+    principal = TenantPrincipal(
+        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:
+    user_id = str(uuid4())
+    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())
+            rows.append(
+                Workspace(
+                    id=workspace_id,
+                    slug=f"test-{index}-{uuid4().hex[:8]}",
+                    name=f"Workspace {index}",
+                )
+            )
+            rows.append(
+                WorkspaceMembership(
+                    id=str(uuid4()),
+                    workspace_id=workspace_id,
+                    user_id=user_id,
+                    role="member",
+                    status="active",
+                )
+            )
+        session.add_all(rows)
+        await session.commit()
+    return user_id
+
+
+async def test_studio_subject_resolves_one_active_membership(tmp_path) -> None:
+    database = SecurityDatabase(
+        f"sqlite+aiosqlite:///{tmp_path / 'studio-tenant.db'}", auto_migrate=True
+    )
+    await database.initialize()
+    try:
+        user_id = 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.membership_id
+        assert principal.membership_role == "member"
+    finally:
+        await database.close()
+
+
+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()
+    try:
+        await _seed_subject(database, "supabase-subject-many", 2)
+        with pytest.raises(ForbiddenError):
+            await TenantService(database).resolve_studio_subject("supabase-subject-many")
+    finally:
+        await database.close()
+
+
+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()
+    try:
+        with pytest.raises(ForbiddenError):
+            await TenantService(database).resolve_studio_subject("not-provisioned")
+    finally:
+        await database.close()