Spaces:
Running
Running
| from __future__ import annotations | |
| import json | |
| from fastapi import Request | |
| PUBLIC_GET_PATHS = frozenset( | |
| {"/", "/health", "/version", "/docs", "/docs/oauth2-redirect", "/openapi.json", "/redoc"} | |
| ) | |
| class ScopePolicy: | |
| """Maps existing and future API route families to stable authorization scopes.""" | |
| def is_public(request: Request) -> bool: | |
| path = request.url.path | |
| callback_segments = tuple(segment for segment in path.split("/") if segment) | |
| return request.method == "GET" and ( | |
| path in PUBLIC_GET_PATHS | |
| # OAuth callbacks deliberately rely on a single-use, short-lived | |
| # state record instead of an API key that a provider cannot send. | |
| # Keep this exemption exact: no nested route may accidentally | |
| # inherit callback's public status. | |
| or callback_segments[:3] == ("v1", "social", "accounts") | |
| and len(callback_segments) == 5 | |
| and callback_segments[-1] == "callback" | |
| ) | |
| async def required_scope(self, request: Request) -> str | None: | |
| path = request.url.path | |
| method = request.method | |
| if path == "/v1/auth/context": | |
| return None | |
| if path.startswith("/v1/social"): | |
| return self._social_scope(path, method) | |
| if path.startswith("/v1/analytics"): | |
| if path == "/v1/analytics/sync" and method == "POST": | |
| return "analytics:sync" | |
| if path.endswith("/cancel") and method == "POST": | |
| return "analytics:sync" | |
| return "analytics:read" | |
| if path.startswith("/v1/generation"): | |
| return self._generation_scope(path, method) | |
| if path.startswith("/v1/ai"): | |
| return self._ai_scope(path, method) | |
| if path.startswith("/v1/copilot"): | |
| return self._copilot_scope(path, method) | |
| if path.startswith("/v1/projects"): | |
| return self._project_scope(path, method) | |
| if path.startswith("/mcp"): | |
| return await self._mcp_scope(request) | |
| if path.startswith("/v1/api-keys") or path.startswith("/v1/audit-logs"): | |
| return "admin" | |
| if path.startswith("/v1/templates/catalog"): | |
| if method == "GET": | |
| return "templates:read" | |
| if path.endswith(("/apply", "/instantiate")): | |
| return "templates:apply" | |
| if method == "POST": | |
| return "templates:create" | |
| if method == "PATCH": | |
| return "templates:update" | |
| if method == "DELETE": | |
| return "templates:delete" | |
| if ( | |
| path == "/v1/templates" | |
| or path == "/v1/templates/categories" | |
| or (path.startswith("/v1/templates/") and path != "/v1/templates/run") | |
| ): | |
| return "templates:read" if method == "GET" else "admin" | |
| if path == "/v1/templates/run": | |
| return "templates:run" | |
| if path.startswith("/v1/jobs"): | |
| if method == "GET": | |
| return "jobs:read" | |
| if method == "DELETE" or path.endswith("/cancel"): | |
| return "jobs:cancel" | |
| return "jobs:create" | |
| if path.startswith("/v1/assets"): | |
| if method == "GET": | |
| return "assets:read" | |
| if method == "DELETE": | |
| return "assets:delete" | |
| return "assets:write" | |
| if path.startswith("/v1/media/") and method == "GET": | |
| return "operations:read" | |
| if ( | |
| path.startswith(("/v1/video", "/v1/audio", "/v1/image", "/v1/whisper", "/v1/ytdlp")) | |
| or path == "/v1/probe" | |
| ): | |
| return "operations:execute" | |
| if method == "GET": | |
| return "system:read" | |
| return "admin" | |
| def _social_scope(path: str, method: str) -> str: | |
| if "/assets" in path: | |
| return "assets:read" if method == "GET" else "assets:write" | |
| if "/analytics" in path: | |
| return "social:analytics:read" | |
| if "/jobs" in path: | |
| return "social:posts:read" | |
| if path.endswith("/calendar"): | |
| return "social:schedules:read" | |
| if path.endswith("/publishing-context"): | |
| return "social:posts:read" | |
| if path.endswith("/queue"): | |
| return "social:posts:read" | |
| if path.endswith("/bulk"): | |
| return "social:schedules:write" | |
| if "/drafts" in path: | |
| return "social:posts:read" if method == "GET" else "social:posts:write" | |
| if "/accounts" in path: | |
| return "social:accounts:read" if method == "GET" else "social:accounts:write" | |
| if "/posts" in path: | |
| if method == "GET": | |
| return "social:posts:read" | |
| if path.endswith("/publish"): | |
| return "social:posts:publish" | |
| if path.endswith("/schedule"): | |
| return "social:schedules:write" | |
| if path.endswith("/reschedule"): | |
| return "social:schedules:write" | |
| if path.endswith("/cancel"): | |
| return "social:posts:write" | |
| return "social:posts:write" | |
| return "social:accounts:read" | |
| def _generation_scope(path: str, method: str) -> str: | |
| if "/providers" in path: | |
| return "generation:providers:read" | |
| if path.endswith("/cancel"): | |
| return "generation:jobs:cancel" | |
| if "/jobs/" in path or method == "GET": | |
| return "generation:requests:read" | |
| return "generation:requests:create" | |
| def _ai_scope(path: str, method: str) -> str: | |
| if path.endswith("/capabilities") or method == "GET": | |
| return "ai:read" | |
| return "ai:generate" | |
| def _copilot_scope(path: str, method: str) -> str: | |
| if method == "GET" or path.endswith("/capabilities"): | |
| return "copilot:read" | |
| return "copilot:execute" | |
| def _project_scope(path: str, method: str) -> str: | |
| if "/editor" in path: | |
| return "projects:read" if method == "GET" else "projects:update" | |
| if "/renders" in path: | |
| return "projects:read" if method == "GET" else "projects:update" | |
| if "/assets" in path or "/jobs" in path: | |
| return "projects:read" if method == "GET" else "projects:update" | |
| if "/workspace/teams" in path: | |
| return { | |
| "GET": "teams:read", | |
| "POST": "teams:create", | |
| "PATCH": "teams:update", | |
| "DELETE": "teams:delete", | |
| }.get(method, "admin") | |
| if "/workspace/invitations" in path: | |
| return "members:invite" | |
| if "/workspace/members" in path: | |
| return { | |
| "DELETE": "members:remove", | |
| "PATCH": "members:update", | |
| "GET": "members:read" | |
| }.get(method, "admin") | |
| if "/workspace/workflows" in path or "/workspace/requests" in path: | |
| if method == "POST": | |
| if "approve" in path or "reject" in path: | |
| return "approvals:review" | |
| if "comments" in path: | |
| return "comments:create" | |
| return "approvals:create" | |
| return "approvals:read" | |
| if "/collaborators" in path: | |
| return "projects:collaborate" | |
| return { | |
| "GET": "projects:read", | |
| "POST": "projects:create", | |
| "PATCH": "projects:update", | |
| "DELETE": "projects:delete", | |
| }.get(method, "admin") | |
| async def _mcp_scope(request: Request) -> str: | |
| if request.method != "POST": | |
| return "mcp:read" | |
| try: | |
| payload = json.loads(await request.body()) | |
| except (json.JSONDecodeError, UnicodeDecodeError): | |
| return "mcp:read" | |
| messages = payload if isinstance(payload, list) else [payload] | |
| methods = { | |
| str(message.get("method", "")) for message in messages if isinstance(message, dict) | |
| } | |
| return "mcp:execute" if "tools/call" in methods else "mcp:read" | |
| def is_job(required_scope: str | None) -> bool: | |
| return required_scope in { | |
| "templates:run", | |
| "operations:execute", | |
| "jobs:create", | |
| "mcp:execute", | |
| "social:posts:publish", | |
| "generation:requests:create", | |
| "analytics:sync", | |
| "ai:generate", | |
| "copilot:execute", | |
| } | |
| def is_upload(request: Request, required_scope: str | None) -> bool: | |
| return request.method in {"POST", "PUT", "PATCH"} and required_scope in { | |
| "templates:run", | |
| "operations:execute", | |
| "jobs:create", | |
| "assets:write", | |
| "mcp:execute", | |
| } | |