Spaces:
Runtime error
Runtime error
| from __future__ import annotations | |
| import contextvars | |
| from dataclasses import dataclass | |
| from datetime import datetime | |
| from app.security.scopes import membership_role_scopes | |
| class AuthContext: | |
| api_key_id: str | |
| key_name: str | |
| key_prefix: str | |
| environment: str | |
| role: str | None | |
| scopes: frozenset[str] | |
| requests_per_minute: int | |
| concurrent_jobs: int | |
| uploads_per_hour: int | |
| processing_bytes_per_day: int | |
| # API keys may be configured without an expiry; callers that omit the | |
| # field therefore receive the same explicit non-expiring semantics. | |
| expires_at: datetime | None = None | |
| # Resolved exclusively by the server from APIKeyPrincipal. These are not | |
| # accepted from REST, MCP, SDK, or n8n request payloads. | |
| workspace_id: str | None = None | |
| user_id: str | None = None | |
| membership_id: str | None = None | |
| membership_role: str | None = None | |
| membership_scopes: frozenset[str] | None = None | |
| def allows(self, required_scope: str) -> bool: | |
| key_allows = required_scope in self.scopes or "admin" in self.scopes | |
| if not key_allows: | |
| return False | |
| # API-key scopes are a credential ceiling; workspace membership is an | |
| # independent tenant ceiling. A powerful key must never elevate a | |
| # viewer/member beyond the role granted by the workspace owner. | |
| if self.membership_role is None: | |
| return True | |
| membership_scopes = self.membership_scopes | |
| if membership_scopes is None: | |
| membership_scopes = membership_role_scopes(self.membership_role) | |
| return required_scope in membership_scopes or "admin" in membership_scopes | |
| auth_context: contextvars.ContextVar[AuthContext | None] = contextvars.ContextVar( | |
| "auth_context", default=None | |
| ) | |
| # HTTP authentication already applies request and concurrency limits before the | |
| # MCP ASGI application runs. Standalone stdio MCP calls use this marker to apply | |
| # the same limiter exactly once inside the transport-neutral registry. | |
| http_auth_applied: contextvars.ContextVar[bool] = contextvars.ContextVar( | |
| "http_auth_applied", default=False | |
| ) | |