Spaces:
Runtime error
Runtime error
| 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}$") | |
| 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, | |
| ) | |