diff --git a/backend/src/backend/auth.py b/backend/src/backend/auth.py index a49d789..c7bc069 100644 --- a/backend/src/backend/auth.py +++ b/backend/src/backend/auth.py @@ -22,6 +22,7 @@ from sqlalchemy.ext.asyncio import AsyncSession from backend.dependencies import database_session from common.auth.jwt import JwtError, issue_jwt, verify_jwt_token +from common.auth.membership import resolve_is_system_admin from common.auth.passwords import verify_password from common.db.models import Roles, Users, WorkspaceMembers, Workspaces from common.ids import new_ulid @@ -72,26 +73,6 @@ def _user_payload( } -async def _resolve_is_system_admin( - session: AsyncSession, - user: Users, -) -> bool: - """Return True iff the user holds a platform-scoped admin role. - - The check is: ``Users.status == 'active'`` AND - ``Users.platform_role_id`` points to a ``Roles`` row whose - ``role_code == 'admin'``. Any other shape (no platform_role_id, - disabled user, wrong role code) returns False — the frontend reads - this to decide whether to show the system-admin entry point. - """ - if user.status != "active" or user.platform_role_id is None: - return False - platform_role = await session.scalar( - select(Roles).where(Roles.role_id == user.platform_role_id) - ) - return platform_role is not None and platform_role.role_code == "admin" - - def _workspace_payload( workspace: Workspaces, role: Roles, @@ -187,7 +168,7 @@ async def login( token = issue_jwt(user.user_id, ttl_seconds=COOKIE_TTL_SECONDS) _set_session_cookie(request, response, token) - is_system_admin = await _resolve_is_system_admin(session, user) + is_system_admin = await resolve_is_system_admin(session, user) return { "request_id": new_ulid(), @@ -266,7 +247,7 @@ async def me( if user_role_code is None and rows: user_role_code = rows[0][1].role_code - is_system_admin = await _resolve_is_system_admin(session, user) + is_system_admin = await resolve_is_system_admin(session, user) return { "request_id": new_ulid(), diff --git a/backend/src/backend/dependencies.py b/backend/src/backend/dependencies.py index ec2806e..ea79f91 100644 --- a/backend/src/backend/dependencies.py +++ b/backend/src/backend/dependencies.py @@ -31,10 +31,15 @@ from dataclasses import dataclass from typing import AsyncIterator from fastapi import Depends, HTTPException, Query, Request, status +from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from common.auth.jwt import JwtError, verify_jwt_token -from common.auth.membership import MembershipError, load_active_membership +from common.auth.membership import ( + MembershipError, + load_active_membership, + resolve_is_system_admin, +) from common.db import session_scope from common.db.models import Roles, Users, Workspaces from common.ids import new_ulid @@ -49,10 +54,18 @@ class RequestContext: user: Users workspace: Workspaces role: Roles + # True when the requester holds the platform-scoped admin role (via + # Users.platform_role_id). Same flag exposed on ``/api/v1/auth/me`` + # so the frontend can render the platform-admin entry point. System + # admins have full control over every workspace — including disabled + # ones — so ``request_context`` lets them through and this flag is + # the single signal handlers use to gate platform-only operations. + is_system_admin: bool = False @property def is_admin(self) -> bool: - return self.role.role_code == "admin" + """Workspace admin OR system admin (the latter is strictly stronger).""" + return self.is_system_admin or self.role.role_code == "admin" async def database_session(request: Request) -> AsyncIterator[AsyncSession]: @@ -97,7 +110,7 @@ async def request_context( ), session: AsyncSession = Depends(database_session), ) -> RequestContext: - """Verify JWT and load the user's active membership for ``workspace_id``. + """Verify JWT and load the user's membership for ``workspace_id``. The ``request_id`` comes from the ``X-Request-ID`` header if present, else from a freshly minted ULID. Handlers receive the @@ -105,21 +118,59 @@ async def request_context( header-based implementation, so the 56 ``Depends(request_context)`` call sites in this repo stay working without changes — they just now pass ``?workspace_id=`` instead of the old headers. + + System admins (Users.platform_role_id → admin role) bypass the + active-membership requirement: they can address disabled workspaces + because they own the platform. Non-admin users still need an + active ``WorkspaceMembers`` row in an active ``Workspaces`` row. """ user = await current_user(request, session) - try: - _user, workspace, role = await load_active_membership( - session, user.user_id, workspace_id, - ) - except MembershipError as exc: - raise HTTPException( - status.HTTP_403_FORBIDDEN, - "active workspace membership is required", - ) from exc + is_system_admin = await resolve_is_system_admin(session, user) + + if is_system_admin: + # System admin: any workspace (active or disabled) is fine. + # Still 404 if the workspace_id is genuinely unknown — the + # ``?workspace_id=`` query param is part of the URL contract. + workspace = await session.get(Workspaces, workspace_id) + if workspace is None: + raise HTTPException( + status.HTTP_404_NOT_FOUND, + "workspace not found", + ) + # Synthesize a role object so the rest of ``RequestContext`` + # (and downstream ``is_admin`` checks) keep working without + # branching on whether a real WorkspaceMembers row exists. + role = await _load_admin_role(session) + if role is None: + # The seed migration creates this row, so this is a + # hard config error if it's missing. + raise HTTPException( + status.HTTP_500_INTERNAL_SERVER_ERROR, + "admin role not configured", + ) + else: + try: + _user, workspace, role = await load_active_membership( + session, user.user_id, workspace_id, + ) + except MembershipError as exc: + raise HTTPException( + status.HTTP_403_FORBIDDEN, + "active workspace membership is required", + ) from exc + request_id = request.headers.get("X-Request-ID") or new_ulid() return RequestContext( request_id=request_id, user=user, workspace=workspace, role=role, + is_system_admin=is_system_admin, + ) + + +async def _load_admin_role(session: AsyncSession) -> Roles | None: + """Return the singleton ``role_code='admin'`` row (None if absent).""" + return await session.scalar( + select(Roles).where(Roles.role_code == "admin") ) diff --git a/backend/src/backend/main.py b/backend/src/backend/main.py index 4b77137..78a8a8d 100644 --- a/backend/src/backend/main.py +++ b/backend/src/backend/main.py @@ -10,7 +10,13 @@ from fastapi.routing import APIRoute from common.config import settings from common.db import create_database_engine, create_session_factory from common.service_app import create_service_app -from common.storage import AsyncStorageBackend, PURPOSE_BUCKETS, build_storage_config, create_storage +from common.storage import ( + AsyncStorageBackend, + PURPOSE_BUCKETS, + actual_bucket_name, + build_storage_config, + create_storage, +) from backend.admin import router as admin_router from backend.auth import router as auth_router from backend.platform import router as platform_router @@ -31,10 +37,13 @@ async def lifespan(app: Any) -> AsyncIterator[None]: # Storage API is part of the backend process. Platform routers call # the helpers in ``backend.services.storage`` directly (in-process), - # so no HTTP client is needed. ``build_storage_config`` picks s3 vs - # local based on ``settings.storage_backend``. + # so no HTTP client is needed. The dict is keyed by the actual + # bucket name (e.g. "versions"), matching ``UploadSessions.bucket_name`` + # and ``StorageObjects.bucket_name`` so call sites can do + # ``object_stores[upload.bucket_name].put(...)`` directly. app.state.object_stores: dict[str, AsyncStorageBackend] = { - name: create_storage(build_storage_config(name)) for name in PURPOSE_BUCKETS + actual_bucket_name(purpose): create_storage(build_storage_config(purpose)) + for purpose in PURPOSE_BUCKETS } app.state.default_bucket = settings.s3_workspace_bucket runtime_http_client = httpx.AsyncClient( diff --git a/backend/src/backend/storage_api.py b/backend/src/backend/storage_api.py index b3f2612..59c1ab0 100644 --- a/backend/src/backend/storage_api.py +++ b/backend/src/backend/storage_api.py @@ -23,6 +23,7 @@ from common.service_app import create_service_app from common.storage import ( AsyncStorageBackend, PURPOSE_BUCKETS, + actual_bucket_name, build_storage_config, create_storage, ) @@ -62,20 +63,26 @@ def safe_file_name(value: str) -> str: return name -# Map an upload's usage_type to the S3 bucket that should hold the -# resulting object. ``usage_type`` is the only signal available at the -# storage edge (the request comes from either the public API or the -# internal schedule worker), so we make the routing decision in one place -# here and let every other layer — server-object create, multipart upload, -# direct put — inherit the mapping. +# Map an upload's usage_type to its purpose (which then resolves to the +# actual bucket / directory via ``actual_bucket_name``). Keeping the +# purpose as the intermediate value means s3 mode and local mode share +# the same routing logic — only the final ``actual_bucket_name`` differs. +USAGE_TYPE_TO_PURPOSE: dict[str, str] = { + "working_copy": "workspace", + "public_script": "workspace", + "data_resource": "workspace", + "snapshot": "workspace", + "version_artifact": "version", + "run_log": "run_log", + "run_result": "run_log", +} + +# Pre-resolved bucket map (for read-only callers like services/storage.py). +# Re-resolved at module load; re-resolve via resolve_bucket() if the +# workspace.artifact_bucket override matters. BUCKET_FOR_USAGE: dict[str, str] = { - "working_copy": settings.s3_workspace_bucket, - "public_script": settings.s3_workspace_bucket, - "data_resource": settings.s3_workspace_bucket, - "snapshot": settings.s3_workspace_bucket, - "version_artifact": settings.s3_version_bucket, - "run_log": settings.s3_run_log_bucket, - "run_result": settings.s3_run_log_bucket, + usage_type: actual_bucket_name(purpose) + for usage_type, purpose in USAGE_TYPE_TO_PURPOSE.items() } @@ -92,7 +99,8 @@ def resolve_bucket( """ if workspace.artifact_bucket: return workspace.artifact_bucket - return BUCKET_FOR_USAGE.get(usage_type, settings.s3_workspace_bucket) + purpose = USAGE_TYPE_TO_PURPOSE.get(usage_type, "workspace") + return actual_bucket_name(purpose) def storage_payload(item: StorageObjects) -> dict[str, Any]: @@ -124,9 +132,12 @@ async def lifespan(app: Any) -> AsyncIterator[None]: app.state.session_factory = create_session_factory(engine) # Buckets are pre-provisioned by the deployment; the storage layer no # longer auto-creates them. ``build_storage_config`` picks s3 vs local - # based on ``settings.storage_backend``. + # based on ``settings.storage_backend``. The dict is keyed by the + # actual bucket name so ``object_stores[upload.bucket_name]`` + # works without a reverse mapping. app.state.object_stores: dict[str, AsyncStorageBackend] = { - name: create_storage(build_storage_config(name)) for name in PURPOSE_BUCKETS + actual_bucket_name(purpose): create_storage(build_storage_config(purpose)) + for purpose in PURPOSE_BUCKETS } app.state.default_bucket = settings.s3_workspace_bucket try: diff --git a/common/src/common/auth/membership.py b/common/src/common/auth/membership.py index b6e528d..fed91ad 100644 --- a/common/src/common/auth/membership.py +++ b/common/src/common/auth/membership.py @@ -22,6 +22,34 @@ class MembershipError(Exception): """Raised when the user is not an active member of the workspace.""" +async def resolve_is_system_admin( + session: AsyncSession, + user: Users, +) -> bool: + """Return True iff the user holds a platform-scoped admin role. + + The check is: ``Users.status == 'active'`` AND + ``Users.platform_role_id`` points to a ``Roles`` row whose + ``role_code == 'admin'``. Any other shape (no platform_role_id, + disabled user, wrong role code) returns False — the frontend reads + this to decide whether to show the system-admin entry point. + + System admins own the platform: they can address disabled workspaces, + bypass per-workspace membership checks, etc. Anything that wants to + gate "platform-only" behavior (deleting a workspace, soft-deleting + a user globally, …) should consult this flag — it is the single + source of truth. + """ + from sqlalchemy import select + + if user.status != "active" or user.platform_role_id is None: + return False + platform_role = await session.scalar( + select(Roles).where(Roles.role_id == user.platform_role_id) + ) + return platform_role is not None and platform_role.role_code == "admin" + + async def load_active_membership( session: AsyncSession, user_id: str, @@ -66,4 +94,5 @@ async def load_active_membership( __all__ = [ "MembershipError", "load_active_membership", + "resolve_is_system_admin", ] diff --git a/common/src/common/storage/__init__.py b/common/src/common/storage/__init__.py index 16e078b..bc82ba0 100644 --- a/common/src/common/storage/__init__.py +++ b/common/src/common/storage/__init__.py @@ -22,6 +22,7 @@ from .base import AsyncStorageBackend, ObjectMeta, StorageBackend from .factory import ( PURPOSE_BUCKETS, RCLONE_REMOTE_NAME, + actual_bucket_name, build_storage_config, create_storage, rclone_remote_spec, @@ -32,6 +33,7 @@ from .registry import register_backend, registered_backends __all__ = [ "create_storage", "build_storage_config", + "actual_bucket_name", "workspaces_root", "rclone_remote_spec", "RCLONE_REMOTE_NAME", diff --git a/common/src/common/storage/factory.py b/common/src/common/storage/factory.py index a0d0e78..46ca7cb 100644 --- a/common/src/common/storage/factory.py +++ b/common/src/common/storage/factory.py @@ -62,6 +62,22 @@ def create_storage(config: Dict[str, Any]) -> AnyStorageBackend: PURPOSE_BUCKETS: tuple[str, ...] = ("workspace", "version", "run_log", "trash") +def actual_bucket_name(purpose: str) -> str: + """把 purpose 名称解析成实际桶路径 / 名（runtime 数据会存在这个字符串里）。 + + - s3 模式:`settings.s3__bucket`(e.g. ``"versions"``) + - local 模式:`${local_storage_base_dir}/`(e.g. ``"/data/version"``) + + 这是 ``app.state.object_stores`` 的 dict key——``UploadSessions.bucket_name`` + 和 ``StorageObjects.bucket_name`` 都存这个值,所以 dict 必须用这个串 + 做 key 才能在 ``object_stores[upload.bucket_name]`` 那里直接命中。 + """ + from common.config import settings # 延迟 import 避免循环 + if settings.storage_backend == "local": + return str(Path(settings.local_storage_base_dir) / purpose) + return getattr(settings, f"s3_{purpose}_bucket") + + def build_storage_config(bucket_name: str) -> Dict[str, Any]: """根据 ``settings.storage_backend`` 构造 ``create_storage()`` 的入参。