Add verified Studio identity authorization

app/projects/api.py

@@ -66,8 +66,8 @@ async def update_notification_preference(request: Request, payload: dict[str, An
 
 @router.post("/workspace/invitations", response_model=InvitationResponse)
 async def invite_member(request: Request, payload: InvitationCreate) -> InvitationResponse:
-    workspace_id, _, _, _ = _identity(request)
-    return await request.app.state.container.collaboration.invite_member(workspace_id, payload.email, payload.role)
+    workspace_id, actor_user_id, _, _ = _identity(request)
+    return await request.app.state.container.collaboration.invite_member(workspace_id, actor_user_id, payload.email, payload.role)
 
 @router.patch("/workspace/teams/{team_id}", response_model=TeamResponse)
 async def update_team(request: Request, team_id: str, payload: TeamBase) -> TeamResponse:
@@ -82,7 +82,8 @@ async def archive_team(request: Request, team_id: str) -> Response:
 
 @router.get("/workspace/workflows/{workflow_id}/requests", response_model=list[ApprovalRequest])
 async def list_approval_requests(request: Request, workflow_id: str) -> list[ApprovalRequest]:
-    return await request.app.state.container.approval.list_requests(workflow_id)
+    workspace_id, _, _, _ = _identity(request)
+    return await request.app.state.container.approval.list_requests(workspace_id, workflow_id)
 
 
 @router.post("/workspace/requests/{request_id}/approve", response_model=ApprovalRequest)
@@ -90,16 +91,16 @@ async def approve_request(
     request: Request,
     request_id: str
 ) -> ApprovalRequest:
-    _, user_id, _, _ = _identity(request)
-    return await request.app.state.container.approval.approve_request(request_id, user_id)
+    workspace_id, user_id, _, _ = _identity(request)
+    return await request.app.state.container.approval.approve_request(workspace_id, request_id, user_id)
 
 @router.post("/workspace/requests/{request_id}/reject", response_model=ApprovalRequest)
 async def reject_request(
     request: Request,
     request_id: str
 ) -> ApprovalRequest:
-    _, user_id, _, _ = _identity(request)
-    return await request.app.state.container.approval.reject_request(request_id, user_id)
+    workspace_id, user_id, _, _ = _identity(request)
+    return await request.app.state.container.approval.reject_request(workspace_id, request_id, user_id)
 
 @router.post("/workspace/requests/{request_id}/comments", response_model=ReviewComment)
 async def add_review_comment(
@@ -327,7 +328,8 @@ async def list_project_collaborators(
     request: Request,
     project_id: UUID
 ) -> list[MemberResponse]:
-    return await request.app.state.container.collaboration.list_project_collaborators(str(project_id))
+    workspace_id, _, _, _ = _identity(request)
+    return await request.app.state.container.collaboration.list_project_collaborators(workspace_id, str(project_id))
 
 @router.post("/{project_id}/collaborators", response_model=MemberResponse)
 async def add_project_collaborator(
@@ -347,7 +349,8 @@ async def remove_project_collaborator(
     project_id: UUID,
     user_id: str
 ) -> Response:
-    await request.app.state.container.collaboration.remove_project_collaborator(str(project_id), user_id)
+    workspace_id, _, _, _ = _identity(request)
+    await request.app.state.container.collaboration.remove_project_collaborator(workspace_id, str(project_id), user_id)
     return Response(status_code=status.HTTP_204_NO_CONTENT)
 
 

app/projects/repositories/approval_repository.py

@@ -30,9 +30,14 @@ class ApprovalRepository:
             await session.refresh(request)
             return request
 
-    async def update_request_status(self, request_id: str, status: str) -> ApprovalRequest:
+    async def update_request_status(self, workspace_id: str, request_id: str, status: str) -> ApprovalRequest:
         async with self.database.session() as session:
-            request = await session.get(ApprovalRequest, request_id)
+            request = await session.scalar(
+                select(ApprovalRequest).where(
+                    ApprovalRequest.id == request_id,
+                    ApprovalRequest.workspace_id == workspace_id,
+                )
+            )
             if not request:
                 raise Exception("Request not found")
             request.status = status
@@ -40,15 +45,28 @@ class ApprovalRepository:
             await session.refresh(request)
             return request
 
-    async def get_request(self, request_id: str) -> ApprovalRequest:
+    async def get_request(self, workspace_id: str, request_id: str) -> ApprovalRequest:
         async with self.database.session() as session:
-            request = await session.get(ApprovalRequest, request_id)
+            request = await session.scalar(
+                select(ApprovalRequest).where(
+                    ApprovalRequest.id == request_id,
+                    ApprovalRequest.workspace_id == workspace_id,
+                )
+            )
             if not request:
                 raise Exception("Request not found")
             return request
 
     async def add_review_comment(self, request_id: str, user_id: str, workspace_id: str, content: str) -> ReviewComment:
         async with self.database.session() as session:
+            request = await session.scalar(
+                select(ApprovalRequest).where(
+                    ApprovalRequest.id == request_id,
+                    ApprovalRequest.workspace_id == workspace_id,
+                )
+            )
+            if request is None:
+                raise ValueError("Approval request was not found in this workspace.")
             comment = ReviewComment(
                 request_id=request_id,
                 user_id=user_id,
@@ -60,10 +78,13 @@ class ApprovalRepository:
             await session.refresh(comment)
             return comment
 
-    async def list_requests(self, workflow_id: str) -> list[ApprovalRequest]:
+    async def list_requests(self, workspace_id: str, workflow_id: str) -> list[ApprovalRequest]:
         async with self.database.session() as session:
             result = await session.scalars(
-                select(ApprovalRequest).where(ApprovalRequest.workflow_id == workflow_id)
+                select(ApprovalRequest).where(
+                    ApprovalRequest.workspace_id == workspace_id,
+                    ApprovalRequest.workflow_id == workflow_id,
+                )
             )
             return list(result.all())
 

app/projects/repositories/collaboration_repository.py

@@ -5,6 +5,7 @@ from app.security.database import SecurityDatabase
 from sqlalchemy import select, func
 from app.projects.models.collaboration import Team, Invitation, ProjectCollaborator, CollaborationActivity
 from app.security.models import WorkspaceMembership
+from app.projects.models.project import Project
 
 class CollaborationRepository:
     def __init__(self, database: SecurityDatabase) -> None:
@@ -148,15 +149,35 @@ class CollaborationRepository:
             )
             return list(result.all())
 
-    async def list_project_collaborators(self, project_id: str) -> list[ProjectCollaborator]:
+    async def list_project_collaborators(self, workspace_id: str, project_id: str) -> list[ProjectCollaborator]:
         async with self.database.session() as session:
             result = await session.scalars(
-                select(ProjectCollaborator).where(ProjectCollaborator.project_id == project_id)
+                select(ProjectCollaborator)
+                .join(Project, Project.id == ProjectCollaborator.project_id)
+                .where(
+                    ProjectCollaborator.workspace_id == workspace_id,
+                    ProjectCollaborator.project_id == project_id,
+                    Project.workspace_id == workspace_id,
+                )
             )
             return list(result.all())
 
     async def add_project_collaborator(self, workspace_id: str, project_id: str, user_id: str, role: str) -> ProjectCollaborator:
         async with self.database.session() as session:
+            project = await session.scalar(
+                select(Project).where(Project.id == project_id, Project.workspace_id == workspace_id)
+            )
+            membership = await session.scalar(
+                select(WorkspaceMembership).where(
+                    WorkspaceMembership.workspace_id == workspace_id,
+                    WorkspaceMembership.user_id == user_id,
+                    WorkspaceMembership.status == "active",
+                )
+            )
+            if project is None or membership is None:
+                raise ValueError("Project and collaborator must belong to the authenticated workspace.")
+            if role not in {"viewer", "editor", "admin", "member"}:
+                raise ValueError("Unsupported project collaborator role.")
             collaborator = ProjectCollaborator(
                 workspace_id=workspace_id,
                 project_id=project_id,
@@ -168,12 +189,16 @@ class CollaborationRepository:
             await session.refresh(collaborator)
             return collaborator
 
-    async def remove_project_collaborator(self, project_id: str, user_id: str) -> None:
+    async def remove_project_collaborator(self, workspace_id: str, project_id: str, user_id: str) -> None:
         async with self.database.session() as session:
             collaborator = await session.scalar(
-                select(ProjectCollaborator).where(
+                select(ProjectCollaborator)
+                .join(Project, Project.id == ProjectCollaborator.project_id)
+                .where(
+                    ProjectCollaborator.workspace_id == workspace_id,
                     ProjectCollaborator.project_id == project_id,
-                    ProjectCollaborator.user_id == user_id
+                    ProjectCollaborator.user_id == user_id,
+                    Project.workspace_id == workspace_id,
                 )
             )
             if collaborator:

app/projects/services/approval_service.py

@@ -15,8 +15,8 @@ class ApprovalService:
         state = await self.editor_repository.get_state(workspace_id, project_id)
         return await self.repository.create_request(workspace_id, workflow_id, project_id, user_id, state.revision)
 
-    async def approve_request(self, request_id: str, actor_user_id: str) -> ApprovalRequest:
-        request = await self.repository.get_request(request_id)
+    async def approve_request(self, workspace_id: str, request_id: str, actor_user_id: str) -> ApprovalRequest:
+        request = await self.repository.get_request(workspace_id, request_id)
         if request.created_by == actor_user_id:
              raise Exception("Separation of duties: Cannot approve your own submission.")
         
@@ -25,13 +25,13 @@ class ApprovalService:
         if state.revision != request.editor_revision:
             raise Exception("Content has changed since request; approval is stale.")
 
-        return await self.repository.update_request_status(request_id, "approved")
+        return await self.repository.update_request_status(workspace_id, request_id, "approved")
 
-    async def reject_request(self, request_id: str, actor_user_id: str) -> ApprovalRequest:
-        return await self.repository.update_request_status(request_id, "rejected")
+    async def reject_request(self, workspace_id: str, request_id: str, actor_user_id: str) -> ApprovalRequest:
+        return await self.repository.update_request_status(workspace_id, request_id, "rejected")
 
     async def add_comment(self, request_id: str, user_id: str, workspace_id: str, content: str) -> ReviewComment:
         return await self.repository.add_review_comment(request_id, user_id, workspace_id, content)
 
-    async def list_requests(self, workflow_id: str) -> list[ApprovalRequest]:
-        return await self.repository.list_requests(workflow_id)
+    async def list_requests(self, workspace_id: str, workflow_id: str) -> list[ApprovalRequest]:
+        return await self.repository.list_requests(workspace_id, workflow_id)

app/projects/services/collaboration_service.py

@@ -33,7 +33,10 @@ class CollaborationService:
     async def archive_team(self, workspace_id: str, team_id: str) -> None:
         await self.repository.archive_team(workspace_id, team_id)
 
-    async def invite_member(self, workspace_id: str, email: str, role: str) -> InvitationResponse:
+    async def invite_member(self, workspace_id: str, actor_user_id: str, email: str, role: str) -> InvitationResponse:
+        actor_membership = await self.repository.get_membership(workspace_id, actor_user_id)
+        if actor_membership is None or actor_membership.status != "active" or actor_membership.role not in {"owner", "admin"}:
+            raise CollaborationUnauthorizedError("Only workspace owners or admins can invite members.")
         if role not in {"admin", "member"}:
             raise ValueError("Invitation role must be admin or member.")
         token = secrets.token_urlsafe(32)
@@ -101,20 +104,32 @@ class CollaborationService:
         if actor_user_id == target_user_id and new_role == 'admin':
             raise CollaborationUnauthorizedError("Cannot elevate your own privileges.")
 
-        # Perform update
+        if new_role not in {"admin", "member"}:
+            raise ValueError("Member role must be admin or member.")
+        target_membership = await self.repository.get_membership(workspace_id, target_user_id)
+        if target_membership is None:
+            raise TeamMemberNotFoundError("Member not found.")
+        if target_membership.role == "admin" and new_role != "admin":
+            admin_count = await self.repository.count_workspace_admins(workspace_id)
+            if admin_count <= 1:
+                raise CollaborationUnauthorizedError("Cannot demote the last administrator.")
+        # Perform update only after validating the actor, target, and allowed role.
         await self.repository.update_member_role(workspace_id, target_user_id, new_role)
 
     async def get_membership(self, workspace_id: str, user_id: str) -> WorkspaceMembership | None:
         return await self.repository.get_membership(workspace_id, user_id)
 
-    async def list_project_collaborators(self, project_id: str) -> list[MemberResponse]:
-        collaborators = await self.repository.list_project_collaborators(project_id)
+    async def list_project_collaborators(self, workspace_id: str, project_id: str) -> list[MemberResponse]:
+        collaborators = await self.repository.list_project_collaborators(workspace_id, project_id)
         return [MemberResponse(id=c.id, workspace_id=c.workspace_id, user_id=c.user_id, role=c.role, created_at=c.created_at.isoformat()) for c in collaborators]
 
     async def add_project_collaborator(self, workspace_id: str, project_id: str, user_id: str, role: str) -> MemberResponse:
         collaborator = await self.repository.add_project_collaborator(workspace_id, project_id, user_id, role)
         return MemberResponse(id=collaborator.id, workspace_id=collaborator.workspace_id, user_id=collaborator.user_id, role=collaborator.role, created_at=collaborator.created_at.isoformat())
 
+    async def remove_project_collaborator(self, workspace_id: str, project_id: str, user_id: str) -> None:
+        await self.repository.remove_project_collaborator(workspace_id, project_id, user_id)
+
     async def record_activity(self, workspace_id: str, user_id: str, action: str, entity_id: str, entity_type: str, metadata: dict[str, Any]) -> None:
         await self.repository.record_activity(workspace_id, user_id, action, entity_id, entity_type, metadata)
 

app/security/cli.py

@@ -3,12 +3,16 @@ from __future__ import annotations
 import argparse
 import asyncio
 import json
+from uuid import uuid4
+
+from sqlalchemy import select
 
 from app.core.config import get_settings
 from app.security.database import SecurityDatabase
 from app.security.schemas import APIKeyCreate, APIKeyView
 from app.security.service import APIKeyService
 from app.security.tenancy import TenantService
+from app.security.models import User, Workspace, WorkspaceMembership
 
 
 async def _create(arguments: argparse.Namespace) -> None:
@@ -33,6 +37,58 @@ async def _create(arguments: argparse.Namespace) -> None:
         await database.close()
 
 
+
+async def _provision_studio_user(arguments: argparse.Namespace) -> None:
+    """Explicit operator-driven onboarding; never grant access from a browser request."""
+    settings = get_settings()
+    database = SecurityDatabase(settings.database_url, auto_migrate=settings.security_auto_migrate)
+    await database.initialize()
+    try:
+        async with database.session() as session:
+            existing = await session.scalar(select(User).where(User.subject == arguments.subject))
+            if existing is not None:
+                raise SystemExit("That identity subject is already provisioned; refusing to reassign it.")
+            workspace = await session.scalar(
+                select(Workspace).where(Workspace.slug == arguments.workspace_slug)
+            )
+            if workspace is not None:
+                raise SystemExit("Workspace slug already exists; refusing to attach an initial owner implicitly.")
+            user_id = str(uuid4())
+            workspace_id = str(uuid4())
+            user = User(
+                id=user_id,
+                subject=arguments.subject,
+                display_name=arguments.display_name,
+                metadata_json={"provisioned_by": "security-cli"},
+            )
+            workspace = Workspace(
+                id=workspace_id,
+                slug=arguments.workspace_slug,
+                name=arguments.workspace_name,
+                metadata_json={"provisioned_by": "security-cli"},
+            )
+            membership = WorkspaceMembership(
+                id=str(uuid4()),
+                workspace_id=workspace_id,
+                user_id=user_id,
+                role=arguments.role,
+                status="active",
+            )
+            session.add_all([user, workspace, membership])
+            await session.commit()
+            print(json.dumps({
+                "user_id": user_id,
+                "subject": arguments.subject,
+                "workspace_id": workspace_id,
+                "workspace_slug": arguments.workspace_slug,
+                "membership_id": membership.id,
+                "role": arguments.role,
+                "status": "active",
+            }, indent=2))
+    finally:
+        await database.close()
+
+
 def _bootstrap(arguments: argparse.Namespace) -> None:
     material = APIKeyService.generate_material(arguments.environment)
     print("Store the API key in your password manager. It will not be shown again.\n")
@@ -49,6 +105,15 @@ def main() -> None:
         "generate-bootstrap", help="Generate a key and hash-only bootstrap settings"
     )
     bootstrap.add_argument("--environment", choices=("live", "test"), default="live")
+    provision = commands.add_parser(
+        "provision-studio-user",
+        help="Explicitly provision a verified Supabase subject and initial workspace membership",
+    )
+    provision.add_argument("--subject", required=True, help="Exact Supabase auth user UUID/subject")
+    provision.add_argument("--display-name", required=True)
+    provision.add_argument("--workspace-slug", required=True)
+    provision.add_argument("--workspace-name", required=True)
+    provision.add_argument("--role", choices=("owner", "admin", "member"), default="owner")
     create = commands.add_parser("create", help="Create a key directly in the configured database")
     create.add_argument("--name", required=True)
     create.add_argument("--environment", choices=("live", "test"), default="live")
@@ -57,6 +122,8 @@ def main() -> None:
     arguments = parser.parse_args()
     if arguments.command == "generate-bootstrap":
         _bootstrap(arguments)
+    elif arguments.command == "provision-studio-user":
+        asyncio.run(_provision_studio_user(arguments))
     else:
         asyncio.run(_create(arguments))
 

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:

docs/STUDIO_IDENTITY_SETUP.md

@@ -0,0 +1,55 @@
+# MediaRouter Studio identity and tenant isolation
+
+MediaRouter Studio authenticates users with Supabase on the Next.js server and forwards a short-lived, signed identity envelope to MediaRouter. The shared MediaRouter API token is a server-to-server transport credential; it must never be exposed to browser code.
+
+## Required configuration
+
+Set these values in both environments:
+
+- **Studio / Vercel:** `STUDIO_IDENTITY_SECRET` and `MEDIAROUTER_API_TOKEN`. Keep them server-only; do not use a `NEXT_PUBLIC_` prefix.
+- **MediaRouter / Hugging Face Space:** `STUDIO_IDENTITY_SECRET` with the exact same secret value, plus the existing API authentication and production database configuration.
+
+Use a cryptographically random secret of at least 32 bytes. Generate one with a password manager or a secure local random generator. Do not paste it into source control, logs, or chat. Restart/redeploy both services after setting it.
+
+The Studio proxy signs the UTF-8 bytes of four newline-separated values in this order: timestamp (Unix seconds), Supabase user ID, normalized email, display name. It uses HMAC-SHA256 and sends the lowercase hexadecimal signature in `x-mr-studio-signature`. The backend validates the signature in constant time and rejects envelopes more than 300 seconds from server time.
+
+## Tenant onboarding is explicit
+
+A valid Supabase session proves identity, but it does **not** automatically grant a MediaRouter workspace. Before a user can use tenant-backed Studio APIs, the backend security database must contain:
+
+1. An active `users` row whose `subject` exactly matches the Supabase user ID.
+2. Exactly one active `workspace_memberships` row for that user, linked to an active `workspaces` row.
+3. A supported membership role: `owner`, `admin`, `developer`, `operator`, `viewer`, or `member`.
+
+Users with no active membership or more than one active workspace membership are denied. This is deliberate: the current Studio identity envelope has no workspace selector, and the backend must not guess a tenant. Do not solve onboarding by trusting a workspace ID from a browser request or by assigning every newly signed-in user an owner role.
+
+For a controlled first-user bootstrap, an operator with access to the backend's configured security database can run the explicit CLI command from the backend project environment:
+
+```sh
+python -m app.security.cli provision-studio-user \
+  --subject "<exact Supabase auth user ID>" \
+  --display-name "<display name>" \
+  --workspace-slug "<unique-lowercase-workspace-slug>" \
+  --workspace-name "<workspace name>" \
+  --role owner
+```
+
+This command refuses to reassign an already-provisioned subject or reuse an existing workspace slug. It requires direct operator/database access and must not be exposed as an unauthenticated HTTP endpoint. For subsequent members, use the authorized invitation workflow; invitation acceptance still needs to be wired to the identity provider before invitations alone grant access.
+
+## Authorization behavior
+
+The backend treats the shared API key as transport authentication only when a signed Studio identity is present. It resolves the Supabase subject against persisted backend membership, replaces the key's tenant identity with that persisted principal, and intersects API-key scopes with the membership role's default scopes. Workspace owners/admins are mapped to the backend `developer` capability set, not the global API `admin` role. Viewer/member access is restricted to the viewer capability set.
+
+If any Studio identity header is present, the complete envelope and signature are required. The Studio proxy also fails closed when it cannot validate a Supabase session; it must not forward the shared token as a fallback for an unidentified browser request.
+
+## Production verification checklist
+
+- [ ] Set the same random `STUDIO_IDENTITY_SECRET` in Vercel and the HF Space.
+- [ ] Confirm `MEDIAROUTER_API_TOKEN` is only a server-side Vercel environment variable.
+- [ ] Provision one explicit backend user and workspace membership for each Studio user, or implement an authorized invitation flow.
+- [ ] Use a non-superuser tenant database role and verify forced RLS policies on production tables; a passing application-level check alone is not proof of database isolation.
+- [ ] Test with two distinct Studio accounts: create/read/update/delete a project and access an asset/job in account A, then confirm account B cannot read, mutate, cancel, or download A's resources.
+- [ ] Test expired and tampered signatures, disabled users/memberships/workspaces, and users with zero or multiple active memberships.
+- [ ] Run the backend unit/integration tests and inspect production logs for denied cross-tenant access before merging/deploying.
+
+This document describes the intended configuration and checks; it is not evidence that the live deployment has been configured or that the checklist has passed.

tests/test_approval_workspace_scope.py

@@ -0,0 +1,38 @@
+from __future__ import annotations
+
+from unittest.mock import AsyncMock
+
+import pytest
+
+from app.projects.repositories.approval_repository import ApprovalRepository
+from app.projects.services.approval_service import ApprovalService
+
+
+@pytest.mark.asyncio
+async def test_approval_repository_queries_are_workspace_scoped() -> None:
+    repository = ApprovalRepository(None)
+    # Repository calls require a real database; this regression test protects
+    # the service contract so routes cannot omit the authenticated workspace.
+    repository.list_requests = AsyncMock(return_value=[])
+    service = ApprovalService(repository, None)
+    result = await service.list_requests("workspace-a", "workflow-a")
+    assert result == []
+    repository.list_requests.assert_awaited_once_with("workspace-a", "workflow-a")
+
+
+@pytest.mark.asyncio
+async def test_approval_actions_pass_workspace_to_repository() -> None:
+    repository = ApprovalRepository(None)
+    request = type("Request", (), {
+        "created_by": "submitter",
+        "project_id": "project-a",
+        "editor_revision": 1,
+        "workspace_id": "workspace-a",
+    })()
+    repository.get_request = AsyncMock(return_value=request)
+    repository.update_request_status = AsyncMock(return_value=request)
+    editor = type("Editor", (), {"get_state": AsyncMock(return_value=type("State", (), {"revision": 1})())})()
+    service = ApprovalService(repository, editor)
+    await service.approve_request("workspace-a", "request-a", "reviewer")
+    repository.get_request.assert_awaited_once_with("workspace-a", "request-a")
+    repository.update_request_status.assert_awaited_once_with("workspace-a", "request-a", "approved")

tests/test_project_collaboration_authorization.py

@@ -1,23 +1,53 @@
 from __future__ import annotations
 
-import unittest
+from unittest.mock import AsyncMock
 
-from app.projects.repositories.collaboration_repository import CollaborationRepository
-from app.projects.services.collaboration_service import CollaborationService
-
-
-class CollaborationAuthorizationTests(unittest.TestCase):
-    def test_project_collaborator_list_requires_project_workspace(self) -> None:
-        repository = CollaborationRepository(None)
-        service = CollaborationService(repository)
+import pytest
 
-        with self.assertRaises(Exception):
-            service.list_project_collaborators("project-from-another-workspace")
-
-    def test_project_collaborator_removal_requires_project_workspace(self) -> None:
-        repository = CollaborationRepository(None)
-        service = CollaborationService(repository)
+from app.projects.services.collaboration_service import CollaborationService
 
-        with self.assertRaises(Exception):
-            service.remove_project_collaborator("project-from-another-workspace", "user-id")
 
+@pytest.mark.asyncio
+async def test_project_collaborator_listing_passes_authenticated_workspace() -> None:
+    repository = type("Repository", (), {})()
+    repository.list_project_collaborators = AsyncMock(return_value=[])
+    service = CollaborationService(repository)
+    assert await service.list_project_collaborators("workspace-a", "project-a") == []
+    repository.list_project_collaborators.assert_awaited_once_with("workspace-a", "project-a")
+
+
+@pytest.mark.asyncio
+async def test_project_collaborator_removal_passes_authenticated_workspace() -> None:
+    repository = type("Repository", (), {})()
+    repository.remove_project_collaborator = AsyncMock()
+    service = CollaborationService(repository)
+    await service.remove_project_collaborator("workspace-a", "project-a", "user-b")
+    repository.remove_project_collaborator.assert_awaited_once_with(
+        "workspace-a", "project-a", "user-b"
+    )
+
+
+@pytest.mark.asyncio
+async def test_invitation_requires_workspace_admin_or_owner() -> None:
+    repository = type("Repository", (), {})()
+    repository.get_membership = AsyncMock(return_value=type(
+        "Membership", (), {"role": "member", "status": "active"}
+    )())
+    service = CollaborationService(repository)
+    with pytest.raises(Exception, match="Only workspace owners or admins"):
+        await service.invite_member("workspace-a", "user-a", "new@example.com", "member")
+    repository.invite_member_with_hash = AsyncMock()
+    repository.invite_member_with_hash.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_member_role_changes_reject_privileged_unknown_roles() -> None:
+    repository = type("Repository", (), {})()
+    repository.get_membership = AsyncMock(return_value=type(
+        "Membership", (), {"role": "admin", "status": "active", "user_id": "admin-a"}
+    )())
+    repository.update_member_role = AsyncMock()
+    service = CollaborationService(repository)
+    with pytest.raises(ValueError, match="Member role must be admin or member"):
+        await service.update_member_role("workspace-a", "admin-a", "user-b", "owner")
+    repository.update_member_role.assert_not_awaited()

tests/test_studio_identity.py

@@ -0,0 +1,228 @@
+from __future__ import annotations
+
+import hashlib
+import hmac
+from uuid import uuid4
+
+import pytest
+from sqlalchemy import select
+
+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 scoped.membership_role == "member"
+    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
+) -> tuple[str, list[str]]:
+    user_id = str(uuid4())
+    workspace_ids: list[str] = []
+    async with database.session() as session:
+        rows = [User(id=user_id, subject=subject, display_name="Test user")]
+        for index in range(workspace_count):
+            workspace_id = str(uuid4())
+            workspace_ids.append(workspace_id)
+            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, workspace_ids
+
+
+async def _database(tmp_path, filename: str) -> SecurityDatabase:
+    database = SecurityDatabase(
+        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, 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 == workspace_ids[0]
+        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 = await _database(tmp_path, "studio-ambiguous.db")
+    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 = 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(
+                    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()