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 # Every workspace role is allowed the same baseline read capabilities. Keep this # explicit so new UI read endpoints do not fail closed merely because a role's # mutation scopes differ. STUDIO_BASELINE_READ_SCOPES = frozenset({ "templates:read", "operations:read", "jobs:read", "assets:read", "mcp:read", "system:read", "social:accounts:read", "social:posts:read", "social:schedules:read", "social:analytics:read", "analytics:read", "generation:providers:read", "generation:requests:read", "ai:read", "copilot:read", "projects:read", "members:read", "teams:read", "comments:read", "approvals:read", }) _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.") role_scopes = DEFAULT_ROLE_SCOPES[effective_role] | STUDIO_BASELINE_READ_SCOPES return replace( context, role=effective_role, scopes=frozenset(context.scopes.intersection(role_scopes)), workspace_id=principal.workspace_id, user_id=principal.user_id, membership_id=principal.membership_id, membership_role=raw_role, )