diff --git a/.env.example b/.env.example index 0952238..2d6ab09 100644 --- a/.env.example +++ b/.env.example @@ -17,6 +17,12 @@ DATABASE_URL=mysql+asyncmy://root:change-me@127.0.0.1:3306/model_platform?charse DEMO_AUTH_ENABLED=true JWT_SECRET=change-this-development-secret +# ============================================================================ +# CRITICAL: must set BEFORE first run. The initial admin user is seeded by +# the deployment bootstrap. Never keep the development default in production. +# ============================================================================ +INITIAL_ADMIN_PASSWORD=admin12345 + # Object storage (S3-compatible, RustFS). # RUSTFS_ENDPOINT is the single upstream URL consumed by all 4 services: # - nginx (via scripts/nginx-entrypoint.sh, which parses host + port) @@ -36,4 +42,5 @@ RUSTFS_SECRET_KEY=change-me RUSTFS_WORKSPACE_BUCKET=workspaces RUSTFS_VERSION_BUCKET=versions RUSTFS_RUN_LOG_BUCKET=run-logs - +RUSTFS_TRASH_BUCKET=trash +RUSTFS_TRASH_RETENTION_DAYS=30 diff --git a/backend/Dockerfile b/backend/Dockerfile index be8a6ba..f810b03 100644 --- a/backend/Dockerfile +++ b/backend/Dockerfile @@ -8,10 +8,10 @@ WORKDIR /app COPY --from=ghcr.io/astral-sh/uv:latest /uv /bin/uv RUN ( \ - sed -i 's/deb.debian.org/mirrors.cloud.aliyuncs.com/g' /etc/apt/sources.list.d/debian.sources 2>/dev/null || \ - sed -i 's/deb.debian.org/mirrors.cloud.aliyuncs.com/g' /etc/apt/sources.list 2>/dev/null || \ - sed -i 's/archive.ubuntu.com/mirrors.cloud.aliyuncs.com/g' /etc/apt/sources.list 2>/dev/null || \ - sed -i 's/security.ubuntu.com/mirrors.cloud.aliyuncs.com/g' /etc/apt/sources.list 2>/dev/null \ + sed -i 's/deb.debian.org/mirrors.aliyun.com/g' /etc/apt/sources.list.d/debian.sources 2>/dev/null || \ + sed -i 's/deb.debian.org/mirrors.aliyun.com/g' /etc/apt/sources.list 2>/dev/null || \ + sed -i 's/archive.ubuntu.com/mirrors.aliyun.com/g' /etc/apt/sources.list 2>/dev/null || \ + sed -i 's/security.ubuntu.com/mirrors.aliyun.com/g' /etc/apt/sources.list 2>/dev/null \ ) || ( \ sed -i 's/deb.debian.org/mirrors.aliyun.com/g' /etc/apt/sources.list.d/debian.sources 2>/dev/null || \ sed -i 's/deb.debian.org/mirrors.aliyun.com/g' /etc/apt/sources.list 2>/dev/null || \ diff --git a/backend/pyproject.toml b/backend/pyproject.toml index e15b90d..893bbda 100644 --- a/backend/pyproject.toml +++ b/backend/pyproject.toml @@ -11,6 +11,8 @@ dependencies = [ "alembic==1.18.5", "cryptography==49.0.0", "gunicorn>=26.0.0", + "passlib==1.7.4", + "bcrypt>=4.0,<4.1", ] [tool.uv.sources] diff --git a/backend/src/backend/auth.py b/backend/src/backend/auth.py index 9fb1928..f302db3 100644 --- a/backend/src/backend/auth.py +++ b/backend/src/backend/auth.py @@ -35,18 +35,18 @@ router = APIRouter(tags=["auth"]) # (future extension). COOKIE_NAME = "access_token" COOKIE_TTL_SECONDS = 24 * 60 * 60 -COOKIE_SECURE = True COOKIE_SAMESITE = "lax" -def _set_session_cookie(response: Response, token: str) -> None: +def _set_session_cookie(request: Request, response: Response, token: str) -> None: + forwarded_scheme = request.headers.get("x-forwarded-proto", request.url.scheme) response.set_cookie( key=COOKIE_NAME, value=token, max_age=COOKIE_TTL_SECONDS, path="/", httponly=True, - secure=COOKIE_SECURE, + secure=forwarded_scheme == "https", samesite=COOKIE_SAMESITE, ) @@ -159,7 +159,7 @@ async def login( user_role_code = rows[0][1].role_code token = issue_jwt(user.user_id, ttl_seconds=COOKIE_TTL_SECONDS) - _set_session_cookie(response, token) + _set_session_cookie(request, response, token) return { "request_id": new_ulid(), diff --git a/backend/src/backend/dependencies.py b/backend/src/backend/dependencies.py index 615a915..ec2806e 100644 --- a/backend/src/backend/dependencies.py +++ b/backend/src/backend/dependencies.py @@ -1,22 +1,48 @@ +"""FastAPI dependencies for the public API. + +Two distinct concerns live here: + +* :func:`database_session` — yields a transactional ``AsyncSession`` + bound to the request's app-state factory. The session commits on + success and rolls back on exception via ``common.db.session_scope``. + +* :func:`request_context` — verifies the ``access_token`` cookie + (HS256-signed JWT), then loads the user's active membership for the + ``workspace_id`` query parameter. Returns a :class:`RequestContext` + dataclass that downstream handlers read for identity, role, and the + request id. + +The 56 ``Depends(request_context)`` call sites elsewhere in the +backend expect this exact dataclass shape; the workspace and role are +always present, so handlers do not need to handle the "no workspace +selected" case. + +Routes that need a workspace context must declare the query parameter +explicitly, even when the path also contains a resource id (e.g. +``/schedules/{schedule_id}``). This keeps authorization decisions +local to the request and prevents the "implicit workspace" footgun +where a path-resource lookup quietly overrides what the user asked +for. +""" + from __future__ import annotations from dataclasses import dataclass from typing import AsyncIterator -from fastapi import Depends, Header, HTTPException, Request, status -from sqlalchemy import select +from fastapi import Depends, HTTPException, Query, Request, status 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.db import session_scope -from common.db.models import ( - Roles, - Users, - WorkspaceMembers, - Workspaces, -) +from common.db.models import Roles, Users, Workspaces from common.ids import new_ulid +ACCESS_TOKEN_COOKIE = "access_token" + + @dataclass(frozen=True) class RequestContext: request_id: str @@ -34,42 +60,66 @@ async def database_session(request: Request) -> AsyncIterator[AsyncSession]: yield session +async def current_user( + request: Request, + session: AsyncSession = Depends(database_session), +) -> Users: + """Resolve the authenticated user from the ``access_token`` cookie. + + Returns the active ``Users`` row or raises 401. Does NOT load a + workspace — use :func:`request_context` for handlers that need + a workspace-scoped context, or use ``Depends(current_user)`` for + workspace-agnostic endpoints (e.g. ``/api/v1/auth/me``). + """ + token = request.cookies.get(ACCESS_TOKEN_COOKIE) + if not token: + raise HTTPException(status.HTTP_401_UNAUTHORIZED, "not authenticated") + try: + payload = verify_jwt_token(token) + except JwtError as exc: + raise HTTPException(status.HTTP_401_UNAUTHORIZED, str(exc)) from exc + user_id = payload.get("sub") + if not user_id: + raise HTTPException(status.HTTP_401_UNAUTHORIZED, "invalid token") + user = await session.get(Users, user_id) + if user is None or user.status != "active": + raise HTTPException(status.HTTP_401_UNAUTHORIZED, "user not found") + return user + + async def request_context( request: Request, - x_user_id: str = Header(alias="X-User-ID"), - x_workspace_id: str = Header(alias="X-Workspace-ID"), - x_request_id: str | None = Header(default=None, alias="X-Request-ID"), + workspace_id: str = Query( + ..., + min_length=26, + max_length=26, + description="Workspace context for this request (CHAR(26) ULID).", + ), + session: AsyncSession = Depends(database_session), ) -> RequestContext: - async with request.app.state.session_factory() as session: - statement = ( - select(Users, Workspaces, Roles) - .join( - WorkspaceMembers, - WorkspaceMembers.user_id == Users.user_id, - ) - .join( - Workspaces, - Workspaces.workspace_id == WorkspaceMembers.workspace_id, - ) - .join(Roles, Roles.role_id == WorkspaceMembers.role_id) - .where( - Users.user_id == x_user_id, - Users.status == "active", - WorkspaceMembers.workspace_id == x_workspace_id, - WorkspaceMembers.member_status == "active", - Workspaces.status == "active", - ) - ) - row = (await session.execute(statement)).one_or_none() - if row is None: - raise HTTPException( - status.HTTP_403_FORBIDDEN, - "active workspace membership is required", - ) - user, workspace, role = row - return RequestContext( - request_id=x_request_id or new_ulid(), - user=user, - workspace=workspace, - role=role, + """Verify JWT and load the user's active 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 + same :class:`RequestContext` shape they had under the old + 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. + """ + 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 + request_id = request.headers.get("X-Request-ID") or new_ulid() + return RequestContext( + request_id=request_id, + user=user, + workspace=workspace, + role=role, + ) diff --git a/backend/src/backend/jupyter.py b/backend/src/backend/jupyter.py index 43dbf94..bcbbc6c 100644 --- a/backend/src/backend/jupyter.py +++ b/backend/src/backend/jupyter.py @@ -1,129 +1,24 @@ from __future__ import annotations -import base64 -import hashlib -import hmac -import json import re -import time from typing import Optional from fastapi import APIRouter, Depends, HTTPException, Request, Response from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials from sqlalchemy import select - -from backend.dependencies import ( - RequestContext, - database_session, - request_context, -) -from backend.runtime_client import RuntimeClientError -from common.config import settings -from common.db.models import Scripts, Users, WorkspaceMembers, Workspaces from sqlalchemy.ext.asyncio import AsyncSession +from backend.dependencies import database_session +from backend.runtime_client import RuntimeClientError +from common.auth.jwt import JwtError, verify_jwt_token +from common.auth.membership import MembershipError, load_active_membership +from common.db.models import Scripts + router = APIRouter(tags=["jupyter"]) security = HTTPBearer(auto_error=False) -JWT_SECRET = settings.jwt_secret -JWT_ALGORITHM = "HS256" - - -def _b64decode(value: str) -> bytes: - padding = "=" * (-len(value) % 4) - return base64.urlsafe_b64decode(value + padding) - - -def _b64encode(value: bytes) -> str: - return base64.urlsafe_b64encode(value).rstrip(b"=").decode() - - -def create_jwt_token(user_id: str, *, expires_seconds: int = 3600) -> str: - header = _b64encode( - json.dumps( - {"alg": JWT_ALGORITHM, "typ": "JWT"}, - separators=(",", ":"), - ).encode() - ) - expires_at = int(time.time()) + expires_seconds - payload = _b64encode( - json.dumps( - {"sub": user_id, "exp": expires_at}, - separators=(",", ":"), - ).encode() - ) - signing_input = f"{header}.{payload}".encode() - signature = _b64encode( - hmac.new( - JWT_SECRET.encode(), - signing_input, - hashlib.sha256, - ).digest() - ) - return f"{header}.{payload}.{signature}" - - -def verify_jwt_token(token: str) -> dict: - """Parse a signed JWT from the access_token cookie. - - Returns a payload dict containing at least ``sub`` (user_id) and - ``exp``; raises 401 on missing, malformed, or expired tokens. The - current implementation is intentionally minimal — replace with a - real verification once a public-key/HSM source is wired in. - """ - if not token: - raise HTTPException( - status_code=401, - detail="Missing Authentication Token", - ) - - try: - header_b64, payload_b64, signature_b64 = token.split(".", 2) - except ValueError as exc: - raise HTTPException( - status_code=401, - detail="Invalid Authentication Token", - ) from exc - - signing_input = f"{header_b64}.{payload_b64}".encode() - expected = hmac.new( - JWT_SECRET.encode(), - signing_input, - hashlib.sha256, - ).digest() - try: - signature = _b64decode(signature_b64) - except Exception as exc: # pragma: no cover - malformed b64 - raise HTTPException( - status_code=401, - detail="Invalid Authentication Token", - ) from exc - if not hmac.compare_digest(expected, signature): - raise HTTPException( - status_code=401, - detail="Invalid Authentication Token", - ) - - try: - payload = json.loads(_b64decode(payload_b64)) - except Exception as exc: - raise HTTPException( - status_code=401, - detail="Invalid Authentication Token", - ) from exc - - exp = payload.get("exp") - if not isinstance(exp, (int, float)) or exp < time.time(): - raise HTTPException( - status_code=401, - detail="Token expired", - ) - - return payload - - def extract_notebook_path( uri: str, workspace_id: str, @@ -168,75 +63,23 @@ async def check_notebook_is_locked( return bool(is_locked) -async def load_active_membership( +async def load_active_membership_or_403( session: AsyncSession, user_id: str, workspace_id: str, -) -> tuple[Users, Workspaces]: - """Resolve the user's active membership in the workspace.""" - statement = ( - select(Users, Workspaces) - .join( - WorkspaceMembers, - WorkspaceMembers.user_id == Users.user_id, - ) - .join( - Workspaces, - Workspaces.workspace_id == WorkspaceMembers.workspace_id, - ) - .where( - Users.user_id == user_id, - Users.status == "active", - WorkspaceMembers.workspace_id == workspace_id, - WorkspaceMembers.member_status == "active", - Workspaces.status == "active", - ) - ) - row = (await session.execute(statement)).one_or_none() - if row is None: +): + """Resolve the user's active membership, raising 403 if missing. + + Thin wrapper around :func:`common.auth.membership.load_active_membership` + that maps the library's ``MembershipError`` to a FastAPI 403. + """ + try: + return await load_active_membership(session, user_id, workspace_id) + except MembershipError as exc: raise HTTPException( status_code=403, detail="active workspace membership is required", - ) - return row - - -@router.post("/api/v1/auth/demo-session") -async def create_demo_session( - request: Request, - response: Response, - context: RequestContext = Depends(request_context), -) -> dict: - """Issue the short-lived HttpOnly cookie used by the self-hosted UI. - - The normal request-context check still requires an active user/workspace - membership. This endpoint is disabled by default and must be explicitly - enabled by the deployment configuration. - """ - if not settings.demo_auth_enabled: - raise HTTPException(status_code=404, detail="Not Found") - expires_seconds = 3600 - response.set_cookie( - "access_token", - create_jwt_token( - context.user.user_id, - expires_seconds=expires_seconds, - ), - max_age=expires_seconds, - httponly=True, - secure=request.url.scheme == "https", - samesite="lax", - path="/", - ) - return { - "request_id": context.request_id, - "data": { - "user_id": context.user.user_id, - "workspace_id": context.workspace.workspace_id, - "expires_at": int(time.time()) + expires_seconds, - }, - "meta": {}, - } + ) from exc @router.get("/api/v1/auth/jupyter") @@ -266,7 +109,13 @@ async def verify_jupyter_access( cookie_token = request.cookies.get("access_token") bearer_token = auth.credentials if auth else None token = cookie_token or bearer_token - payload = verify_jwt_token(token) + try: + payload = verify_jwt_token(token) + except JwtError as exc: + raise HTTPException( + status_code=401, + detail=str(exc), + ) from exc user_id = payload.get("sub") if not user_id: raise HTTPException( @@ -274,7 +123,7 @@ async def verify_jupyter_access( detail="Invalid Authentication Token", ) - await load_active_membership(session, user_id, workspace_id) + await load_active_membership_or_403(session, user_id, workspace_id) notebook_path = extract_notebook_path(original_uri, workspace_id) if notebook_path and await check_notebook_is_locked( diff --git a/backend/src/backend/main.py b/backend/src/backend/main.py index 61fcc40..c390cd4 100644 --- a/backend/src/backend/main.py +++ b/backend/src/backend/main.py @@ -12,6 +12,7 @@ from common.db import create_database_engine, create_session_factory from common.service_app import create_service_app from common.storage import RustFSObjectStore from backend.admin import router as admin_router +from backend.auth import router as auth_router from backend.jupyter import router as jupyter_router from backend.resources import router as resources_router from backend.runtime_client import RuntimeClient @@ -34,12 +35,13 @@ async def lifespan(app: Any) -> AsyncIterator[None]: access_key=settings.rustfs_access_key, secret_key=settings.rustfs_secret_key, ) - # Ensure all three purpose-named buckets exist; the storage edge picks + # Ensure all four purpose-named buckets exist; the storage edge picks # the right one per upload (see resolve_bucket in storage_api.py). for bucket in ( settings.rustfs_workspace_bucket, settings.rustfs_version_bucket, settings.rustfs_run_log_bucket, + settings.rustfs_trash_bucket, ): await asyncio.to_thread( app.state.object_store.ensure_bucket, @@ -69,6 +71,7 @@ app = create_service_app( settings.service_name, lifespan=lifespan, ) +app.include_router(auth_router) app.include_router(jupyter_router) app.include_router(resources_router) app.include_router(schedule_runs_router) diff --git a/backend/src/backend/schedule_runs.py b/backend/src/backend/schedule_runs.py index 233d953..8c4bd6a 100644 --- a/backend/src/backend/schedule_runs.py +++ b/backend/src/backend/schedule_runs.py @@ -1,6 +1,5 @@ from __future__ import annotations -import hashlib from datetime import UTC, datetime from typing import Any, Literal @@ -13,19 +12,22 @@ from common.db.models import ( ScheduleNodeRuns, ScheduleRuns, ) -from common.eventing import add_outbox_event, utcnow -from common.ids import new_ulid from backend.dependencies import ( RequestContext, database_session, request_context, ) from common.schemas import StrictModel -from backend.schedules import ( - graph_rows, - schedule_row, - validate_dag, +from common.scheduler import ( + DagTooLarge, + InvalidDag, + InvalidNodeArguments, + ScheduleNotFound, + TriggerError, + create_scheduled_run, + normalize_idempotency_key, ) +from common.ids import new_ulid router = APIRouter(tags=["schedule-runs"]) @@ -51,46 +53,23 @@ def _iso(value: datetime | None) -> str | None: return value.astimezone(UTC).isoformat() -def _normalized_idempotency_key( - workspace_id: str, - schedule_id: str, - value: str, -) -> str: - normalized = value.strip() - if len(normalized) < 8: - raise HTTPException( - status.HTTP_422_UNPROCESSABLE_ENTITY, - "Idempotency-Key must contain at least 8 characters", +def _http_error_from_trigger(exc: TriggerError) -> HTTPException: + if isinstance(exc, ScheduleNotFound): + return HTTPException(status.HTTP_404_NOT_FOUND, str(exc)) + if isinstance(exc, InvalidDag): + return HTTPException( + status.HTTP_409_CONFLICT, + detail={ + "code": "SCHEDULE_DAG_INVALID", + "message": str(exc), + "errors": exc.errors, + }, ) - digest = hashlib.sha256( - f"{workspace_id}:{schedule_id}:{normalized}".encode("utf-8") - ).hexdigest() - return f"run:v1:{digest}" - - -def _arguments(value: dict[str, Any] | None) -> list[str]: - payload = value or {} - raw = payload.get("_args") - result = [str(item) for item in raw] if isinstance(raw, list) else [] - for key, item in payload.items(): - if key == "_args": - continue - option = f"--{key.replace('_', '-')}" - if item is True: - result.append(option) - elif item is False or item is None: - continue - elif isinstance(item, list): - for list_item in item: - result.extend((option, str(list_item))) - elif isinstance(item, (str, int, float)): - result.extend((option, str(item))) - else: - raise HTTPException( - status.HTTP_422_UNPROCESSABLE_ENTITY, - f"node argument {key!r} must be a scalar or list", - ) - return result + if isinstance(exc, DagTooLarge): + return HTTPException(status.HTTP_409_CONFLICT, str(exc)) + if isinstance(exc, InvalidNodeArguments): + return HTTPException(status.HTTP_422_UNPROCESSABLE_ENTITY, exc.message) + return HTTPException(status.HTTP_409_CONFLICT, str(exc)) def run_summary(item: ScheduleRuns) -> dict[str, Any]: @@ -183,125 +162,40 @@ async def run_schedule_now( session: AsyncSession = Depends(database_session), ) -> dict[str, Any]: reason = payload.reason if payload is not None else "manual_run" - key = _normalized_idempotency_key( - context.workspace.workspace_id, - schedule_id, - idempotency_key, - ) - existing = await session.scalar( - select(ScheduleRuns).where(ScheduleRuns.idempotency_key == key) - ) - if existing is not None: - if ( - existing.workspace_id != context.workspace.workspace_id - or existing.schedule_id != schedule_id - ): - raise HTTPException( - status.HTTP_409_CONFLICT, - "Idempotency-Key belongs to another schedule run", - ) - return { - "request_id": context.request_id, - "data": await run_detail(existing, session), - "meta": {"reused": True}, - } - - schedule = await schedule_row( - schedule_id, - context, - session, - for_update=True, - ) - node_rows, edges = await graph_rows(schedule_id, session) - nodes = [row[0] for row in node_rows] - validation = validate_dag(nodes, edges) - if not validation["valid"] or not nodes: - raise HTTPException( - status.HTTP_409_CONFLICT, - detail={ - "code": "SCHEDULE_DAG_INVALID", - "message": "schedule must contain a valid non-empty DAG", - "errors": validation["errors"], - }, + trigger_type: Literal["manual", "cron"] = "cron" if reason == "cron" else "manual" + try: + key = normalize_idempotency_key( + context.workspace.workspace_id, + schedule_id, + idempotency_key, ) - if len(nodes) > 100 or len(edges) > 500: + except TriggerError as exc: raise HTTPException( - status.HTTP_409_CONFLICT, - "schedule exceeds the v1 execution size limit", - ) + status.HTTP_422_UNPROCESSABLE_ENTITY, str(exc), + ) from exc - snapshot = { - "schedule_name": schedule.schedule_name, - "workflow_version": schedule.workflow_version, - "max_concurrency": schedule.max_concurrency, - "failure_policy": schedule.failure_policy, - "nodes": [ - { - "node_id": node.node_id, - "node_key": node.node_key, - "versions_id": version.versions_id, - "script_type": script.script_type, - "artifact_object_id": version.artifact_object_id, - "artifact_path": version.artifact_path, - "timeout_seconds": node.timeout_seconds, - "retry_count": node.retry_count, - "retry_interval_sec": node.retry_interval_sec, - "arguments": _arguments(node.arguments_json), - } - for node, version, script in node_rows - ], - "edges": [ - { - "source_node_id": edge.source_node_id, - "target_node_id": edge.target_node_id, - } - for edge in edges - ], - } - now = utcnow() - run = ScheduleRuns( - run_id=new_ulid(), - schedule_id=schedule.schedule_id, - workspace_id=schedule.workspace_id, - workflow_version=schedule.workflow_version, - trigger_type="cron" if reason == "cron" else "manual", - idempotency_key=key, - run_status="queued", - state_version=0, - schedule_snapshot=snapshot, - queued_at=now, - triggered_by=context.user.user_id, - ) - session.add(run) - schedule.last_run_at = now - await add_outbox_event( - session, - event_type="schedule.run.requested", - producer="platform-api", - trace_id=context.request_id, - aggregate_type="schedule_run", - aggregate_id=run.run_id, - idempotency_key=key, - payload={ - "workspace_id": run.workspace_id, - "schedule_id": run.schedule_id, - "run_id": run.run_id, - "workflow_version": run.workflow_version, - "trigger_type": run.trigger_type, - "triggered_by": run.triggered_by, - "schedule_snapshot": snapshot, - }, - ) - await session.flush() - # Commit before yielding so the Outbox row is visible to the executor's - # next MySQL poll — the executor's _database_event_loop picks it up. - # We intentionally do NOT HTTP-push; see backend/schedule_client.py. + try: + run, is_new = await create_scheduled_run( + session, + schedule_id=schedule_id, + workspace_id=context.workspace.workspace_id, + triggered_by_user_id=context.user.user_id, + trigger_type=trigger_type, + idempotency_key=key, + trace_id=context.request_id, + ) + except TriggerError as exc: + raise _http_error_from_trigger(exc) from exc + + # Commit before responding so the Outbox row is visible to the + # executor's next MySQL poll — the executor's _database_event_loop + # picks it up. We intentionally do NOT HTTP-push. await session.commit() await session.refresh(run) return { "request_id": context.request_id, "data": await run_detail(run, session), - "meta": {"reused": False}, + "meta": {"reused": not is_new}, } diff --git a/backend/src/backend/storage_api.py b/backend/src/backend/storage_api.py index 4f8ef35..248aba7 100644 --- a/backend/src/backend/storage_api.py +++ b/backend/src/backend/storage_api.py @@ -124,9 +124,20 @@ async def lifespan(app: Any) -> AsyncIterator[None]: secret_key=settings.rustfs_secret_key, ) app.state.default_bucket = settings.rustfs_workspace_bucket - await asyncio.to_thread( - app.state.object_store.ensure_bucket, - app.state.default_bucket) + # Ensure every purpose-named bucket exists up front, including the + # trash bucket. The trash bucket is shared across all workspaces + # and usage_types; the source key is preserved as a prefix so a + # restore is a same-key move back to the source bucket. + for bucket in ( + settings.rustfs_workspace_bucket, + settings.rustfs_version_bucket, + settings.rustfs_run_log_bucket, + settings.rustfs_trash_bucket, + ): + await asyncio.to_thread( + app.state.object_store.ensure_bucket, + bucket, + ) try: yield finally: @@ -227,13 +238,19 @@ async def create_upload_record( storage_object = await session.get( StorageObjects, upload.storage_object_id) - return { - "upload_id": upload.upload_id, - "status": upload.upload_status, - "storage_object": ( - storage_payload(storage_object) if storage_object else None - ), - } + if storage_object is None or storage_object.object_status != "available": + # The previously-completed object was deleted (or never + # materialized). Treat the idempotency hit as a tombstone + # and fall through to a fresh upload: clear the pointer so + # complete_upload_record re-validates the bucket + key. + upload.storage_object_id = None + upload.upload_status = "created" + else: + return { + "upload_id": upload.upload_id, + "status": upload.upload_status, + "storage_object": storage_payload(storage_object), + } if upload.upload_status not in {"created", "uploading"}: raise HTTPException( status.HTTP_409_CONFLICT, @@ -295,11 +312,14 @@ async def complete_upload_record( raise HTTPException(status.HTTP_404_NOT_FOUND, "upload not found") if upload.upload_status == "completed" and upload.storage_object_id: item = await session.get(StorageObjects, upload.storage_object_id) - if item is None: - raise HTTPException( - status.HTTP_409_CONFLICT, - "completed upload has no storage object") - return item + if item is None or item.object_status != "available": + # The linked storage object was deleted. Reset the upload so + # the caller can re-upload the same bytes and create a + # fresh, available object. + upload.storage_object_id = None + upload.upload_status = "created" + else: + return item if upload.upload_status not in {"created", "uploading"}: raise HTTPException( status.HTTP_409_CONFLICT, @@ -552,6 +572,20 @@ async def delete_object( storage_object_id: str, request: Request, session: AsyncSession = Depends(database_session)) -> dict[str, Any]: + """Soft-delete a storage object. + + The bytes are copied to ``rustfs_trash_bucket`` under the same key + (preserved as ``{source_bucket}/{object_key}`` so a restore is a + same-name move), the source key is then deleted from its origin + bucket, and the row's ``object_status`` flips to ``"deleted"`` with + ``deleted_at`` stamped for the reaper. + + Immutable artifacts (version snapshots, run logs) are not + trashed — the policy is enforced by ``is_immutable`` and an + explicit 409. The reaper will physically delete trashed objects + older than ``rustfs_trash_retention_days`` (out of scope for this + endpoint; the field is the contract). + """ item = await session.scalar( select(StorageObjects) .where(StorageObjects.storage_object_id == storage_object_id) @@ -563,22 +597,145 @@ async def delete_object( raise HTTPException( status.HTTP_409_CONFLICT, "immutable object cannot be deleted") - if item.object_status != "deleted": - if item.storage_backend == "rustfs" and item.bucket_name and item.object_key: + if item.object_status == "deleted": + return { + "data": { + "storage_object_id": storage_object_id, + "object_status": item.object_status, + "trash_key": item.trash_key, + } + } + if item.storage_backend == "rustfs" and item.bucket_name and item.object_key: + trash_key = f"{item.bucket_name}/{item.object_key}" + try: await asyncio.to_thread( - request.app.state.object_store.delete, - bucket_name=item.bucket_name, - object_key=item.object_key) - item.object_status = "deleted" - item.deleted_at = utcnow() + request.app.state.object_store.move_to_trash, + source_bucket=item.bucket_name, + source_key=item.object_key, + trash_bucket=settings.rustfs_trash_bucket, + trash_key=trash_key, + ) + except Exception as exc: + # If the move fails, leave the source intact and surface the + # error. We do NOT mark the row as deleted in that case — + # otherwise we'd have a row pointing to non-existent bytes. + raise HTTPException( + status.HTTP_502_BAD_GATEWAY, + f"failed to move object to trash: {exc}", + ) from exc + item.trash_key = trash_key + item.object_status = "deleted" + item.deleted_at = utcnow() return { "data": { "storage_object_id": storage_object_id, "object_status": item.object_status, + "trash_key": item.trash_key, + "trash_bucket": settings.rustfs_trash_bucket, } } +@app.post( + "/internal/v1/objects/{storage_object_id}/restore") +async def restore_object( + storage_object_id: str, + request: Request, + session: AsyncSession = Depends(database_session)) -> dict[str, Any]: + """Restore a soft-deleted object from the trash bucket. + + Copies the bytes back to the source bucket + key and flips the + row back to ``object_status='available'``. If the source bucket + is missing the object (e.g. the trash copy was the only one), the + restore still works because we copy *from* trash rather than + renaming in place. The trash copy is left in place — the reaper + will collect it on the next sweep; this is intentional so a + failed restore does not destroy the only copy. + """ + item = await session.scalar( + select(StorageObjects) + .where(StorageObjects.storage_object_id == storage_object_id) + .with_for_update() + ) + if item is None: + raise HTTPException(status.HTTP_404_NOT_FOUND, "object not found") + if item.object_status != "deleted": + raise HTTPException( + status.HTTP_409_CONFLICT, + "object is not in trash") + if not item.trash_key or not item.bucket_name or not item.object_key: + raise HTTPException( + status.HTTP_409_CONFLICT, + "object has no trash pointer; cannot restore") + try: + await asyncio.to_thread( + request.app.state.object_store.copy, + source_bucket=settings.rustfs_trash_bucket, + source_key=item.trash_key, + dest_bucket=item.bucket_name, + dest_key=item.object_key, + ) + except Exception as exc: + raise HTTPException( + status.HTTP_502_BAD_GATEWAY, + f"failed to restore from trash: {exc}", + ) from exc + item.object_status = "available" + item.deleted_at = None + # Keep trash_key so the reaper can clean up the duplicate on its + # next pass; we don't try to delete it here because a partial + # failure would leave the user with no data. + return { + "data": storage_payload(item), + } + + +@app.post( + "/internal/v1/admin/trash/purge") +async def purge_trash_object( + payload: dict[str, Any], + request: Request, + session: AsyncSession = Depends(database_session)) -> dict[str, Any]: + """Physically delete a trashed object. + + Admin / reaper endpoint — given a ``storage_object_id``, deletes + the bytes from the trash bucket and hard-deletes the DB row. + The route is split from ``delete_object`` because the regular + delete path is the one users hit, and reaper runs need a way to + finalize the lifecycle without re-entering the soft-delete branch. + """ + storage_object_id = (payload or {}).get("storage_object_id", "").strip() + if not storage_object_id: + raise HTTPException( + status.HTTP_400_BAD_REQUEST, + "storage_object_id is required") + item = await session.scalar( + select(StorageObjects) + .where(StorageObjects.storage_object_id == storage_object_id) + .with_for_update() + ) + if item is None: + return {"data": {"storage_object_id": storage_object_id, "purged": False}} + if item.object_status != "deleted": + raise HTTPException( + status.HTTP_409_CONFLICT, + "object is not in trash; refuse to hard-delete live data") + if item.trash_key: + try: + await asyncio.to_thread( + request.app.state.object_store.delete, + bucket_name=settings.rustfs_trash_bucket, + object_key=item.trash_key, + ) + except Exception as exc: + raise HTTPException( + status.HTTP_502_BAD_GATEWAY, + f"failed to delete from trash: {exc}", + ) from exc + await session.delete(item) + return {"data": {"storage_object_id": storage_object_id, "purged": True}} + + @app.get("/internal/health/storage") async def internal_health() -> dict[str, str]: return {"status": "ready", "service": "storage-api"} diff --git a/common/pyproject.toml b/common/pyproject.toml index e07adbe..3de5494 100644 --- a/common/pyproject.toml +++ b/common/pyproject.toml @@ -10,6 +10,8 @@ dependencies = [ "boto3>=1.34,<2", "fastapi==0.116.1", "pydantic-settings>=2.14.2", + "passlib==1.7.4", + "bcrypt>=4.0,<4.1", ] [build-system] diff --git a/common/src/common/config.py b/common/src/common/config.py index 8dbb5cf..290c2e7 100644 --- a/common/src/common/config.py +++ b/common/src/common/config.py @@ -83,6 +83,22 @@ class Settings(BaseSettings): default="run-logs", description="Bucket for schedule run logs.", ) + rustfs_trash_bucket: str = Field( + default="trash", + description=( + "Bucket for soft-deleted objects. The source bucket key is " + "preserved as a prefix so a restore is a same-key move. " + "Trash is reaped on a schedule out of band." + ), + ) + rustfs_trash_retention_days: int = Field( + default=30, + description=( + "How long a trashed object is retained before reaping. " + "Tracked in the database (StorageObjects.deleted_at) so the " + "reaper can run as a single SQL sweep." + ), + ) # ── local FS roots ──────────────────────────────────────────── workspace_root: str = Field( diff --git a/common/src/common/db/models/storage.py b/common/src/common/db/models/storage.py index 761b403..521b7aa 100644 --- a/common/src/common/db/models/storage.py +++ b/common/src/common/db/models/storage.py @@ -103,6 +103,15 @@ class StorageObjects(Base): TINYINT(1), nullable=False, server_default=text("0") ) deleted_at: Mapped[Optional[datetime.datetime]] = mapped_column(DATETIME(fsp=3)) + trash_key: Mapped[Optional[str]] = mapped_column( + String(1100), + comment=( + "Path inside the trash bucket where the soft-deleted bytes " + "live. Format: '{source_bucket}/{object_key}' so a restore " + "is a same-key copy back to the source bucket. NULL while " + "the row is still available." + ), + ) class DataResources(Base): diff --git a/common/src/common/scheduler/__init__.py b/common/src/common/scheduler/__init__.py index a12a034..fbb997d 100644 --- a/common/src/common/scheduler/__init__.py +++ b/common/src/common/scheduler/__init__.py @@ -16,6 +16,18 @@ from __future__ import annotations from typing import TYPE_CHECKING +from common.scheduler.trigger import ( + SYSTEM_CRON_USER_ID, + DagTooLarge, + InvalidDag, + InvalidNodeArguments, + ScheduleNotFound, + TriggerError, + create_scheduled_run, + normalize_idempotency_key, + parse_node_arguments, +) + if TYPE_CHECKING: # pragma: no cover - typing only from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore @@ -53,7 +65,16 @@ def build_sqlalchemy_jobstore( __all__ = [ + "DagTooLarge", + "InvalidDag", + "InvalidNodeArguments", "JOBSTORE_TABLE", + "SYSTEM_CRON_USER_ID", + "ScheduleNotFound", + "TriggerError", "build_sqlalchemy_jobstore", + "create_scheduled_run", + "normalize_idempotency_key", + "parse_node_arguments", "to_sync_database_url", ] diff --git a/common/src/common/storage/rustfs.py b/common/src/common/storage/rustfs.py index 4d4314d..30c535e 100644 --- a/common/src/common/storage/rustfs.py +++ b/common/src/common/storage/rustfs.py @@ -156,3 +156,50 @@ class RustFSObjectStore: def delete(self, *, bucket_name: str, object_key: str) -> None: self.internal.delete_object(Bucket=bucket_name, Key=object_key) + + def copy( + self, + *, + source_bucket: str, + source_key: str, + dest_bucket: str, + dest_key: str, + ) -> None: + """Server-side copy ``source_bucket/source_key`` → ``dest_bucket/dest_key``. + + ``CopySource`` is a single header string of the form + ``/{bucket}/{key}`` — must NOT be URL-encoded or quoted. + """ + self.internal.copy_object( + Bucket=dest_bucket, + Key=dest_key, + CopySource={"Bucket": source_bucket, "Key": source_key}, + ) + + def move_to_trash( + self, + *, + source_bucket: str, + source_key: str, + trash_bucket: str, + trash_key: str, + ) -> None: + """Copy an object into the trash bucket and delete the source. + + The copy is a server-side operation in RustFS (no data flows + through the client). The source delete is best-effort: if it + fails after the copy succeeds the trash holds the only copy of + the bytes, which is exactly the point — the caller can retry. + """ + self.copy( + source_bucket=source_bucket, + source_key=source_key, + dest_bucket=trash_bucket, + dest_key=trash_key, + ) + try: + self.delete(bucket_name=source_bucket, object_key=source_key) + except ClientError: + # Source was already gone, or transient delete failure — + # the trash copy is what matters; caller logs and moves on. + pass diff --git a/default.conf b/default.conf index 50d8bd4..1f357ce 100644 --- a/default.conf +++ b/default.conf @@ -43,6 +43,18 @@ server { proxy_set_header X-Forwarded-Proto $scheme; proxy_set_header X-Real-IP $remote_addr; proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; + # Explicitly forward the session cookie set by + # POST /api/v1/auth/login. nginx forwards it by default, but + # spelling it out keeps the auth contract visible. + proxy_set_header Cookie $http_cookie; + # Defense in depth: blank out the legacy identity headers so a + # malicious client cannot bypass the cookie-based auth flow + # by stuffing X-User-ID / X-Workspace-ID into the request. + # The backend's RequestContext no longer reads them (it + # derives identity from the access_token cookie), so this is + # belt-and-suspenders against a future regression. + proxy_set_header X-User-ID ""; + proxy_set_header X-Workspace-ID ""; } # ========================================================================= @@ -104,6 +116,15 @@ server { proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; + # Defense in depth: do not let the browser-supplied identity + # headers leak past the auth subrequest. The auth subrequest + # only forwards the Cookie + Authorization it cares about; + # the actual Jupyter upstream is fully trusted (the address + # comes from the backend's runtime registry), so a leaked + # X-User-ID here would not matter for the proxy target but + # could pollute audit logs. + proxy_set_header X-User-ID ""; + proxy_set_header X-Workspace-ID ""; } # 2. 内部 Auth 子请求 location @@ -123,6 +144,13 @@ server { proxy_set_header Cookie $http_cookie; proxy_set_header Authorization $http_authorization; + # Defense in depth: the auth subrequest reads the session + # cookie / Authorization header, not the legacy identity + # headers. Blank them out so a poisoned client header cannot + # be confused for an authenticated identity if the backend + # code is ever refactored to read them again. + proxy_set_header X-User-ID ""; + proxy_set_header X-Workspace-ID ""; } # 拒绝其余非法路径 diff --git a/docker-compose.yml b/docker-compose.yml index 99bb6e4..117cec6 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -15,6 +15,7 @@ services: - head environment: DATABASE_URL: ${DATABASE_URL:?DATABASE_URL is required} + INITIAL_ADMIN_PASSWORD: ${INITIAL_ADMIN_PASSWORD:-admin12345} web: build: @@ -55,6 +56,7 @@ services: SERVICE_NAME: model-platform-backend JWT_SECRET: ${JWT_SECRET:?JWT_SECRET is required} DEMO_AUTH_ENABLED: ${DEMO_AUTH_ENABLED:-false} + INITIAL_ADMIN_PASSWORD: ${INITIAL_ADMIN_PASSWORD:-admin12345} RUNTIME_API_URL: http://runtime:8000 RUSTFS_ENDPOINT: ${RUSTFS_ENDPOINT:?RUSTFS_ENDPOINT is required} RUSTFS_ACCESS_KEY: ${RUSTFS_ACCESS_KEY:?RUSTFS_ACCESS_KEY is required} @@ -62,6 +64,8 @@ services: RUSTFS_WORKSPACE_BUCKET: ${RUSTFS_WORKSPACE_BUCKET:-workspaces} RUSTFS_VERSION_BUCKET: ${RUSTFS_VERSION_BUCKET:-versions} RUSTFS_RUN_LOG_BUCKET: ${RUSTFS_RUN_LOG_BUCKET:-run-logs} + RUSTFS_TRASH_BUCKET: ${RUSTFS_TRASH_BUCKET:-trash} + RUSTFS_TRASH_RETENTION_DAYS: ${RUSTFS_TRASH_RETENTION_DAYS:-30} READINESS_TARGETS: ${MYSQL_HOST:?MYSQL_HOST is required}:${MYSQL_PORT:-3306},${RUSTFS_HOST:?RUSTFS_HOST is required}:${RUSTFS_PORT:-9000},runtime:8000 depends_on: migrate: diff --git a/frontend/app/app.css b/frontend/app/app.css index ded572f..c5fff16 100644 --- a/frontend/app/app.css +++ b/frontend/app/app.css @@ -25,3 +25,86 @@ * { box-sizing: border-box; } html, body { margin: 0; min-width: 1024px; min-height: 100%; } button, input, select, textarea { font: inherit; } + +.auth-loading, +.login-page { + min-height: 100vh; + display: flex; + align-items: center; + justify-content: center; +} + +.auth-loading { + gap: 12px; + color: #61758a; +} + +.auth-loading > span { + width: 20px; + height: 20px; + border: 2px solid #bfd5e9; + border-top-color: #1677ff; + border-radius: 50%; + animation: auth-spin 0.8s linear infinite; +} + +@keyframes auth-spin { to { transform: rotate(360deg); } } + +.login-page { + padding: 48px; + background: + radial-gradient(circle at 20% 15%, rgba(22, 119, 255, 0.12), transparent 34%), + linear-gradient(145deg, #eef5fb, #f8fafc 55%, #edf3f8); +} + +.login-card { + width: 420px; + padding: 42px; + border: 1px solid #dbe6ef; + border-radius: 18px; + background: rgba(255, 255, 255, 0.96); + box-shadow: 0 22px 60px rgba(37, 63, 88, 0.14); +} + +.login-brand { + width: 46px; + height: 46px; + display: grid; + place-items: center; + border-radius: 13px; + color: white; + background: linear-gradient(145deg, #1177e8, #25a1f2); +} + +.login-kicker { + margin: 26px 0 8px; + color: #1677ff; + font-size: 11px; + font-weight: 750; + letter-spacing: 0.14em; +} + +.login-card h1 { margin: 0; font-size: 27px; } +.login-description { margin: 10px 0 28px; color: #728398; } +.login-card form { display: grid; gap: 18px; } +.login-card label { display: grid; gap: 8px; color: #465b70; font-size: 13px; } +.login-card input { + width: 100%; + padding: 12px 14px; + border: 1px solid #cad8e5; + border-radius: 9px; + outline: none; + color: #18334f; + background: white; +} +.login-card input:focus { border-color: #1677ff; box-shadow: 0 0 0 3px rgba(22, 119, 255, 0.12); } +.login-card button { + min-height: 44px; + border: 0; + border-radius: 9px; + color: white; + background: #1677ff; + cursor: pointer; +} +.login-card button:disabled { opacity: 0.65; cursor: wait; } +.login-error { margin: -4px 0 0; color: #d4380d; font-size: 13px; } diff --git a/frontend/app/context/AuthContext.tsx b/frontend/app/context/AuthContext.tsx new file mode 100644 index 0000000..643348a --- /dev/null +++ b/frontend/app/context/AuthContext.tsx @@ -0,0 +1,256 @@ +import { + createContext, + type ReactNode, + useCallback, + useContext, + useEffect, + useMemo, + useState, +} from "react"; +import { useLocation, useNavigate } from "react-router"; + +import * as rawApi from "../services/api"; +import type { WorkspaceBoundApi } from "../services/api"; + +export type AuthUser = { + user_id: string; + username: string; + display_name: string; + email: string | null; + status: string; + role_code: string | null; +}; + +export type AuthWorkspace = { + workspace_id: string; + workspace_code: string; + workspace_name: string; + role_code: string; + role_name: string; +}; + +type AuthSession = { + user: AuthUser; + workspaces: AuthWorkspace[]; + default_workspace_id: string | null; +}; + +type ApiEnvelope = { + data: T; +}; + +type AuthContextValue = { + user: AuthUser | null; + workspaces: AuthWorkspace[]; + currentWorkspace: AuthWorkspace | null; + loading: boolean; + login: (username: string, password: string) => Promise; + logout: () => Promise; + setCurrentWorkspace: (workspaceId: string) => void; +}; + +const AuthContext = createContext(null); +const workspaceStorageKey = "model-platform-current-workspace"; + +async function authRequest(path: string, init: RequestInit = {}): Promise { + const response = await fetch(path, { + ...init, + credentials: "same-origin", + headers: { + ...(init.body ? { "Content-Type": "application/json" } : {}), + ...init.headers, + }, + }); + const payload = await response.json().catch(() => ({})) as + | ApiEnvelope + | { detail?: string }; + if (!response.ok) { + throw new Error( + "detail" in payload && typeof payload.detail === "string" + ? payload.detail + : `请求失败（HTTP ${response.status}）`, + ); + } + return (payload as ApiEnvelope).data; +} + +function selectWorkspace( + session: AuthSession, + preferredId?: string | null, +): AuthWorkspace | null { + return session.workspaces.find((item) => item.workspace_id === preferredId) + ?? session.workspaces.find( + (item) => item.workspace_id === session.default_workspace_id, + ) + ?? session.workspaces[0] + ?? null; +} + +export function AuthProvider({ children }: { children: ReactNode }) { + const location = useLocation(); + const navigate = useNavigate(); + const [user, setUser] = useState(null); + const [workspaces, setWorkspaces] = useState([]); + const [currentWorkspace, setWorkspace] = useState(null); + const [loading, setLoading] = useState(true); + + const applySession = useCallback((session: AuthSession) => { + const storedId = typeof window === "undefined" + ? null + : window.localStorage.getItem(workspaceStorageKey); + const workspace = selectWorkspace(session, storedId); + setUser(session.user); + setWorkspaces(session.workspaces); + setWorkspace(workspace); + if (workspace && typeof window !== "undefined") { + window.localStorage.setItem(workspaceStorageKey, workspace.workspace_id); + } + }, []); + + useEffect(() => { + let cancelled = false; + void authRequest("/api/v1/auth/me") + .then((session) => { + if (!cancelled) applySession(session); + }) + .catch(() => { + if (!cancelled) { + setUser(null); + setWorkspaces([]); + setWorkspace(null); + } + }) + .finally(() => { + if (!cancelled) setLoading(false); + }); + return () => { + cancelled = true; + }; + }, [applySession]); + + useEffect(() => { + if (loading) return; + if (!user && location.pathname !== "/login") { + navigate("/login", { replace: true }); + } else if (user && location.pathname === "/login") { + navigate("/workbench", { replace: true }); + } + }, [loading, location.pathname, navigate, user]); + + const login = useCallback(async (username: string, password: string) => { + const session = await authRequest("/api/v1/auth/login", { + method: "POST", + body: JSON.stringify({ username, password }), + }); + applySession(session); + navigate("/workbench", { replace: true }); + }, [applySession, navigate]); + + const logout = useCallback(async () => { + try { + await authRequest<{ logged_out: boolean }>("/api/v1/auth/logout", { + method: "POST", + }); + } finally { + setUser(null); + setWorkspaces([]); + setWorkspace(null); + navigate("/login", { replace: true }); + } + }, [navigate]); + + const setCurrentWorkspace = useCallback((workspaceId: string) => { + const workspace = workspaces.find((item) => item.workspace_id === workspaceId); + if (!workspace) return; + setWorkspace(workspace); + window.localStorage.setItem(workspaceStorageKey, workspace.workspace_id); + }, [workspaces]); + + const value = useMemo(() => ({ + user, + workspaces, + currentWorkspace, + loading, + login, + logout, + setCurrentWorkspace, + }), [currentWorkspace, loading, login, logout, setCurrentWorkspace, user, workspaces]); + + if (loading) { + return ( +
+ + 正在验证登录状态… +
+ ); + } + + return {children}; +} + +export function useAuth(): AuthContextValue { + const context = useContext(AuthContext); + if (!context) { + throw new Error("useAuth 必须在 AuthProvider 内使用"); + } + return context; +} + +export function useApi(): WorkspaceBoundApi { + const { currentWorkspace } = useAuth(); + const workspaceId = currentWorkspace?.workspace_id ?? ""; + + return useMemo(() => ({ + listScripts: () => rawApi.listScripts(workspaceId), + createScript: (input) => rawApi.createScript(workspaceId, input), + uploadScript: (file, parentPath, visibility) => + rawApi.uploadScript(workspaceId, file, parentPath, visibility), + updateScript: (scriptId, input) => + rawApi.updateScript(workspaceId, scriptId, input), + deleteScript: (scriptId) => rawApi.deleteScript(workspaceId, scriptId), + listWorkspaceDirectories: () => rawApi.listWorkspaceDirectories(workspaceId), + createWorkspaceDirectory: (directoryName, parentPath) => + rawApi.createWorkspaceDirectory(workspaceId, directoryName, parentPath), + deleteWorkspaceDirectory: (path) => + rawApi.deleteWorkspaceDirectory(workspaceId, path), + acquireFileLock: (script) => rawApi.acquireFileLock(workspaceId, script), + heartbeatFileLock: (session) => rawApi.heartbeatFileLock(workspaceId, session), + releaseFileLock: (session) => rawApi.releaseFileLock(workspaceId, session), + releaseFileLockOnUnload: (session) => + rawApi.releaseFileLockOnUnload(workspaceId, session), + createJupyterAccessTicket: (session) => + rawApi.createJupyterAccessTicket(workspaceId, session), + listScriptVersions: (scriptId) => rawApi.listScriptVersions(workspaceId, scriptId), + publishScriptVersion: (input) => rawApi.publishScriptVersion(workspaceId, input), + listSchedules: () => rawApi.listSchedules(workspaceId), + getSchedule: (scheduleId) => rawApi.getSchedule(workspaceId, scheduleId), + createSchedule: (input) => rawApi.createSchedule(workspaceId, input), + updateSchedule: (scheduleId, input) => + rawApi.updateSchedule(workspaceId, scheduleId, input), + deleteSchedule: (scheduleId, workflowVersion) => + rawApi.deleteSchedule(workspaceId, scheduleId, workflowVersion), + listScheduleArtifacts: () => rawApi.listScheduleArtifacts(workspaceId), + hideScheduleArtifact: (versionsId) => + rawApi.hideScheduleArtifact(workspaceId, versionsId), + listEmployees: () => rawApi.listEmployees(workspaceId), + createEmployee: (input) => rawApi.createEmployee(workspaceId, input), + updateEmployee: (userId, input) => + rawApi.updateEmployee(workspaceId, userId, input), + deleteEmployee: (userId) => rawApi.deleteEmployee(workspaceId, userId), + createScheduleNode: (scheduleId, input) => + rawApi.createScheduleNode(workspaceId, scheduleId, input), + updateScheduleNode: (scheduleId, nodeId, input) => + rawApi.updateScheduleNode(workspaceId, scheduleId, nodeId, input), + deleteScheduleNode: (scheduleId, nodeId, workflowVersion) => + rawApi.deleteScheduleNode(workspaceId, scheduleId, nodeId, workflowVersion), + createScheduleEdge: (scheduleId, input) => + rawApi.createScheduleEdge(workspaceId, scheduleId, input), + deleteScheduleEdge: (scheduleId, edgeId, workflowVersion) => + rawApi.deleteScheduleEdge(workspaceId, scheduleId, edgeId, workflowVersion), + validateSchedule: (scheduleId) => rawApi.validateSchedule(workspaceId, scheduleId), + previewCron: (input) => rawApi.previewCron(workspaceId, input), + runScheduleNow: (scheduleId) => rawApi.runScheduleNow(workspaceId, scheduleId), + listScheduleRuns: (input) => rawApi.listScheduleRuns(workspaceId, input), + getScheduleRun: (runId) => rawApi.getScheduleRun(workspaceId, runId), + }), [workspaceId]); +} diff --git a/frontend/app/features/admin/AdminPages.tsx b/frontend/app/features/admin/AdminPages.tsx index 93f3bf2..cc75cd6 100644 --- a/frontend/app/features/admin/AdminPages.tsx +++ b/frontend/app/features/admin/AdminPages.tsx @@ -2,13 +2,9 @@ import { type FormEvent, useEffect, useState } from "react"; import { ApiRequestError, - createEmployee, - deleteEmployee, - demoContext, - listEmployees, - updateEmployee, type Employee, } from "../../services/api"; +import { useApi, useAuth } from "../../context/AuthContext"; import Icon from "../../components/Icon"; import "../../styles/admin.css"; import "../../styles/dashboard.css"; @@ -28,13 +24,17 @@ export function DashboardPage({ online: boolean; onNavigate: (page: "scripts" | "schedules" | "system") => void; }) { + const { user, currentWorkspace } = useAuth(); return (
MODEL DEVELOPMENT PLATFORM -

下午好，{demoContext.userName}

-

当前位于 {demoContext.workspaceName}，可以继续构建脚本或配置调度。

+

下午好，{user?.display_name ?? "用户"}

+

+ 当前位于 {currentWorkspace?.workspace_name ?? "（未选择 Workspace）"} + ，可以继续构建脚本或配置调度。 +

{online ? "服务正常" : "服务连接中"}
@@ -112,18 +112,20 @@ export function SystemAdminPage({ onNotify: (notice: Notice) => void; onConnectionChange: (online: boolean) => void; }) { + const api = useApi(); + const { user, currentWorkspace } = useAuth(); const [employees, setEmployees] = useState([]); const [loading, setLoading] = useState(true); const [saving, setSaving] = useState(false); const [editing, setEditing] = useState(null); const [dialogOpen, setDialogOpen] = useState(false); const [form, setForm] = useState(EMPTY_FORM); - const canManage = demoContext.roleCode === "admin"; + const canManage = user?.role_code === "admin"; const load = async (): Promise => { setLoading(true); try { - setEmployees(await listEmployees()); + setEmployees(await api.listEmployees()); onConnectionChange(true); } catch (error) { onConnectionChange(false); @@ -164,7 +166,7 @@ export function SystemAdminPage({ setSaving(true); try { if (editing) { - const updated = await updateEmployee(editing.user_id, { + const updated = await api.updateEmployee(editing.user_id, { display_name: form.display_name.trim(), email: form.email.trim() || null, role_code: form.role_code, @@ -174,7 +176,7 @@ export function SystemAdminPage({ (item) => item.user_id === updated.user_id ? updated : item, )); } else { - const created = await createEmployee({ + const created = await api.createEmployee({ username: form.username.trim(), display_name: form.display_name.trim(), email: form.email.trim() || null, @@ -197,7 +199,7 @@ export function SystemAdminPage({ const remove = async (employee: Employee): Promise => { if (!window.confirm(`确定从当前 Workspace 删除员工“${employee.display_name}”吗？`)) return; try { - await deleteEmployee(employee.user_id); + await api.deleteEmployee(employee.user_id); setEmployees((current) => current.filter((item) => item.user_id !== employee.user_id)); onNotify({ tone: "success", message: "员工已删除" }); } catch (error) { @@ -211,7 +213,11 @@ export function SystemAdminPage({ return (
-
系统管理

员工管理

{demoContext.workspaceName} · {employees.length} 名员工

+
+ 系统管理 +

员工管理

+

{currentWorkspace?.workspace_name ?? "（未选择 Workspace）"} · {employees.length} 名员工

+
diff --git a/frontend/app/features/platform/ModelPlatformApp.tsx b/frontend/app/features/platform/ModelPlatformApp.tsx index 3f0a50d..a5bda9d 100644 --- a/frontend/app/features/platform/ModelPlatformApp.tsx +++ b/frontend/app/features/platform/ModelPlatformApp.tsx @@ -10,24 +10,6 @@ import { import { useLocation, useNavigate } from "react-router"; import { - acquireFileLock, - createWorkspaceDirectory, - createScript, - createJupyterAccessTicket, - deleteScript, - deleteWorkspaceDirectory, - demoContext, - demoUsers, - demoWorkspaces, - heartbeatFileLock, - listScripts, - listScriptVersions, - listWorkspaceDirectories, - publishScriptVersion, - releaseFileLock, - releaseFileLockOnUnload, - setDemoContext, - uploadScript, type ActiveEditSession, type ScriptItem, type ScriptType, @@ -35,6 +17,7 @@ import { type Visibility, type WorkspaceDirectory, } from "../../services/api"; +import { useApi, useAuth } from "~/context/AuthContext"; import Icon from "../../components/Icon"; import SchedulePage from "../schedules/SchedulePage"; import { DashboardPage, SystemAdminPage } from "../admin/AdminPages"; @@ -123,9 +106,27 @@ function mergeDirectories( ).values()]; } export default function ModelPlatformApp() { + const { currentWorkspace } = useAuth(); + if (!currentWorkspace) { + return ( +
+
+ 加载中… +
+
+ ); + } + return ; +} + +function AuthenticatedModelPlatformApp() { const location = useLocation(); const navigate = useNavigate(); const activePage = pageFromPath(location.pathname); + const auth = useAuth(); + const { user, workspaces, setCurrentWorkspace, logout } = auth; + const currentWorkspace = auth.currentWorkspace!; + const api = useApi(); const [scripts, setScripts] = useState([]); const [directories, setDirectories] = useState([]); const [selectedId, setSelectedId] = useState(null); @@ -144,7 +145,6 @@ export default function ModelPlatformApp() { }>({ open: false, parentPath: "", name: "", busy: false }); const [contextMenu, setContextMenu] = useState(null); const [workspaceMenuOpen, setWorkspaceMenuOpen] = useState(false); - const [userMenuOpen, setUserMenuOpen] = useState(false); const [uploadParentPath, setUploadParentPath] = useState(""); const [uploading, setUploading] = useState(false); const uploadInputRef = useRef(null); @@ -176,8 +176,8 @@ export default function ModelPlatformApp() { setRefreshing(silent); try { const [items, folderItems] = await Promise.all([ - listScripts(), - listWorkspaceDirectories(), + api.listScripts(), + api.listWorkspaceDirectories(), ]); setScripts(items); setDirectories(folderItems); @@ -248,7 +248,7 @@ export default function ModelPlatformApp() { } let ignore = false; setVersionsLoading(true); - void listScriptVersions(selectedId) + void api.listScriptVersions(selectedId) .then((items) => { if (!ignore) setVersions(items); }) @@ -282,7 +282,7 @@ export default function ModelPlatformApp() { return; } heartbeatRunning = true; - void heartbeatFileLock(current) + void api.heartbeatFileLock(current) .then((updated) => { setEditSession((active) => active && active.edit_session_id === updated.edit_session_id @@ -319,7 +319,7 @@ export default function ModelPlatformApp() { if (!current || current.edit_session_id !== editSession.edit_session_id) { return; } - void createJupyterAccessTicket(current) + void api.createJupyterAccessTicket(current) .then((ticket) => { setEditSession((active) => active && active.edit_session_id === ticket.edit_session_id @@ -342,7 +342,7 @@ export default function ModelPlatformApp() { if (!editSession) return; const handleUnload = () => { const current = editSessionRef.current; - if (current) releaseFileLockOnUnload(current); + if (current) api.releaseFileLockOnUnload(current); }; window.addEventListener("beforeunload", handleUnload); return () => window.removeEventListener("beforeunload", handleUnload); @@ -356,24 +356,17 @@ export default function ModelPlatformApp() { ); }, [keyword, scripts]); - const memberScriptGroups = [...demoUsers] - .sort((left, right) => ( - Number(right.userId === demoContext.userId) - - Number(left.userId === demoContext.userId) - )) - .map((user) => { - const memberScripts = filteredScripts.filter( - (item) => item.owner_user_id === user.userId, + const memberScriptGroups = (() => { + const currentUserScripts = filteredScripts.filter( + (item) => item.owner_user_id === user?.user_id, ); - const inferred = inferredDirectories(memberScripts); - return { - user, - scripts: memberScripts, - directories: user.userId === demoContext.userId - ? mergeDirectories(directories, inferred) - : inferred, - }; - }); + const inferred = inferredDirectories(currentUserScripts); + return [{ + user: user, + scripts: currentUserScripts, + directories: mergeDirectories(directories, inferred), + }]; + })(); const selected = scripts.find((item) => item.script_id === selectedId) ?? null; const selectScript = (scriptId: string | null) => { @@ -413,7 +406,7 @@ export default function ModelPlatformApp() { let newlyAcquired = false; try { if (active && active.script_id !== script.script_id) { - await releaseFileLock(active); + await api.releaseFileLock(active); setEmbeddedJupyterUrl(null); setEditSession(null); editSessionRef.current = null; @@ -422,20 +415,20 @@ export default function ModelPlatformApp() { if (!requestIsCurrent()) return; if (!active) { - active = await acquireFileLock(script); + active = await api.acquireFileLock(script); newlyAcquired = true; } if (!requestIsCurrent()) { if (active) { - await releaseFileLock(active); + await api.releaseFileLock(active); clearSessionIfActive(active); } return; } - const ticket = await createJupyterAccessTicket(active); + const ticket = await api.createJupyterAccessTicket(active); if (!requestIsCurrent()) { - await releaseFileLock(active); + await api.releaseFileLock(active); clearSessionIfActive(active); return; } @@ -456,7 +449,7 @@ export default function ModelPlatformApp() { } catch (error) { if (newlyAcquired && active) { try { - await releaseFileLock(active); + await api.releaseFileLock(active); } catch { // The database lease is the final safety net if compensation cannot reach Runtime. } @@ -513,7 +506,7 @@ export default function ModelPlatformApp() { } setEditBusy(true); try { - await releaseFileLock(active); + await api.releaseFileLock(active); setEmbeddedJupyterUrl(null); setEditSession(null); editSessionRef.current = null; @@ -572,7 +565,7 @@ export default function ModelPlatformApp() { if (!publishTarget) return; setPublishing(true); try { - const version = await publishScriptVersion({ + const version = await api.publishScriptVersion({ script: publishTarget, releaseNote, visibility: publishVisibility, @@ -602,7 +595,7 @@ export default function ModelPlatformApp() { if (!form.name.trim()) return; setCreating(true); try { - const created = await createScript(form); + const created = await api.createScript(form); setScripts((items) => [created, ...items]); selectScript(created.script_id); setCreateOpen(false); @@ -658,7 +651,7 @@ export default function ModelPlatformApp() { let lastCreated: ScriptItem | null = null; try { for (const file of files) { - lastCreated = await uploadScript(file, uploadParentPath); + lastCreated = await api.uploadScript(file, uploadParentPath); } await load(true); if (lastCreated) selectScript(lastCreated.script_id); @@ -684,7 +677,7 @@ export default function ModelPlatformApp() { if (!folderDialog.name.trim()) return; setFolderDialog((current) => ({ ...current, busy: true })); try { - await createWorkspaceDirectory( + await api.createWorkspaceDirectory( folderDialog.name.trim(), folderDialog.parentPath, ); @@ -718,7 +711,7 @@ export default function ModelPlatformApp() { if (editSessionRef.current?.script_id === script.script_id) return; } try { - await deleteScript(script.script_id); + await api.deleteScript(script.script_id); if (selectedIdRef.current === script.script_id) selectScript(null); await load(true); setToast({ @@ -752,7 +745,7 @@ export default function ModelPlatformApp() { if (editSessionRef.current?.script_id === activeScript.script_id) return; } try { - const result = await deleteWorkspaceDirectory(path); + const result = await api.deleteWorkspaceDirectory(path); const selectedScript = scripts.find( (item) => item.script_id === selectedIdRef.current, ); @@ -851,36 +844,29 @@ export default function ModelPlatformApp() { {apiOnline ? "服务已连接" : "服务未连接"}
- {workspaceMenuOpen && (
- {demoWorkspaces.map((workspace) => ( - ))}
)}
- + - {userMenuOpen && ( -
- {demoUsers.map((user) => ( - - ))} -
- )}
@@ -949,18 +935,18 @@ export default function ModelPlatformApp() { <> {memberScriptGroups.map((group) => ( ))} {filteredScripts.length === 0 && ( @@ -1045,13 +1031,13 @@ export default function ModelPlatformApp() {
) : activePage === "schedules" ? ( ) : activePage === "system" ? ( @@ -1458,4 +1444,4 @@ export default function ModelPlatformApp() { )} ); -} \ No newline at end of file +} diff --git a/frontend/app/features/platform/ScriptWorkspace.tsx b/frontend/app/features/platform/ScriptWorkspace.tsx new file mode 100644 index 0000000..4786539 --- /dev/null +++ b/frontend/app/features/platform/ScriptWorkspace.tsx @@ -0,0 +1,327 @@ +import Icon from "../../components/Icon"; +import type { + ActiveEditSession, + ScriptItem, + StableVersion, +} from "../../services/api"; +import { scriptIcon } from "./WorkspaceTree"; + +type ToastState = { + tone: "success" | "error" | "info"; + message: string; +}; + +type ScriptWorkspaceProps = { + script: ScriptItem; + editSession: ActiveEditSession | null; + jupyterUrl: string | null; + editBusy: boolean; + openError: string | null; + latestVersion: StableVersion | null; + versionsLoading: boolean; + onOpenEditor: () => void; + onEndEditing: () => void; + onClose: () => void; + onPublish: () => void; + onInfo: (toast: ToastState) => void; +}; + +function formatTime(value: string) { + return new Intl.DateTimeFormat("zh-CN", { + month: "2-digit", + day: "2-digit", + hour: "2-digit", + minute: "2-digit", + hour12: false, + }).format(new Date(value)); +} + +function formatBytes(value: number) { + if (value < 1024) return `${value} B`; + return `${(value / 1024).toFixed(1)} KB`; +} + +function shortHash(value: string) { + return value ? `${value.slice(0, 8)}…${value.slice(-6)}` : "—"; +} + +function confineJupyterFrame(frame: HTMLIFrameElement): void { + try { + const document = frame.contentDocument; + if (!document?.documentElement) return; + const keepInside = (): void => { + document.querySelectorAll("button,[role='button']").forEach( + (element) => { + const label = `${element.getAttribute("aria-label") ?? ""} ${ + element.getAttribute("title") ?? "" + } ${element.textContent ?? ""}`.trim(); + if (/^open in\b/i.test(label) || /\bopen in\.\.\./i.test(label)) { + element.style.setProperty("display", "none", "important"); + } + }, + ); + document.querySelectorAll("a[target]").forEach((link) => { + if (["_blank", "_top", "_parent"].includes(link.target)) { + link.target = "_self"; + } + }); + }; + keepInside(); + new MutationObserver(keepInside).observe(document.documentElement, { + childList: true, + subtree: true, + }); + document.addEventListener("click", (event) => { + const target = event.target as HTMLElement | null; + const link = target?.closest?.("a") as HTMLAnchorElement | null; + if (link && ["_blank", "_top", "_parent"].includes(link.target)) { + link.target = "_self"; + } + }, true); + } catch { + // The iframe remains sandboxed even if its document is not yet accessible. + } +} + +export function ScriptWorkspace({ + script, + editSession, + jupyterUrl, + editBusy, + openError, + latestVersion, + versionsLoading, + onOpenEditor, + onEndEditing, + onClose, + onPublish, + onInfo, +}: ScriptWorkspaceProps) { + const isNotebook = script.script_type === "notebook"; + const isEditing = editSession?.session_status === "active"; + return ( + <> +
+
+ + + + {script.script_name} + +
+ +
+ +
+
+ + + + 工作副本 + + {script.script_name} +
+
+ {isEditing ? ( + + ) : ( + + )} + + + + {isEditing + ? isNotebook + ? "Demo 无锁模式 · Kernel 已连接" + : "Demo 无锁模式 · 编辑中" + : latestVersion + ? `最新 ${latestVersion.version_label}` + : "工作副本已就绪"} + +
+
+ +
+ {isEditing && jupyterUrl ? ( +
+
+ + + Workspace Jupyter Server + + + {isNotebook ? "Notebook Session · Python 3 Kernel" : "Jupyter 文本编辑器"} + + + Runtime {editSession.runtime_id.slice(-8)} + +
+