fix(security): P0-1 — port exposure + service-token auth on /internal/* + jupyter RPC

The fix lands in three concentric layers, all backed by a single
INTERNAL_SERVICE_TOKEN shared secret so we have one mechanism
instead of three:

1. docker-compose: drop the backend.ports: 8891:8000 and
   runtime.ports: 8892:8000 mappings. Nginx is the only host
   ingress again (architecture §2.2).
2. /internal/v1/*: the storage control plane had six endpoints, five
   of which were dead code (frontend already migrated to
   /api/v1/data-resources/* with JWT; schedule only ever called
   POST /internal/v1/objects). Delete the dead routes, mount the
   one survivor with Depends(require_internal_service) that
   compares the X-Internal-Service-Token header against
   settings.internal_service_token with secrets.compare_digest.
3. POST /api/v1/jupyter on the runtime container: previously open
   inside the Docker network. Same token mechanism — backend's
   runtime_http_client now carries the header, runtime's
   handle_jupyter_action requires the same header. /api/v1/health
   stays open for the Nginx and compose healthchecks.

The schedule worker was already configured to call
POST /internal/v1/objects; build_storage_http_client now
sets the token header so its existing call site keeps working
without changes.

Files touched:
  backend/src/backend/storage_api.py   # 5 dead routes deleted + token guard
  backend/src/backend/main.py          # runtime_http_client header
  runtime/src/runtime/main.py          # require_internal_service Depends
  common/src/common/config.py          # internal_service_token setting
  schedule/src/schedule/service.py     # httpx client header
  docker-compose.yml                   # ports dropped, INTERNAL_SERVICE_TOKEN env
  .env.example                         # INTERNAL_SERVICE_TOKEN placeholder
  API.md / README.md / DEVELOP.md      # §9 trimmed to 1 endpoint

Verified:
  compileall -> 0 errors
  pytest backend/tests -> 37 passed
  in-process ASGI smoke:
    POST /internal/v1/objects no/wrong/correct token -> 401/401/200
    POST /api/v1/jupyter   no/wrong/correct token -> 401/401/200
    5 deleted internal routes -> 404
  docker compose config (with env) -> OK

P0-1 still has one open sub-item (rclone RC --rc-no-auth) that
the user has explicitly deferred; not touched here.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

.env.example

@@ -73,3 +73,13 @@ S3_TRASH_RETENTION_DAYS=30
 # after writing new workspace files. Default points at the runtime service
 # over the compose network.
 RCLONE_RC_URL=http://runtime:5572
+
+# ============================================================================
+# Service-to-service auth (P0-1 fix).
+# Backend's /internal/v1/* storage control plane requires this shared secret.
+# The schedule worker reads the same value and sends it as the
+# ``X-Internal-Service-Token`` header. Value MUST match between backend and
+# schedule. Generate a random 64-char string for any non-dev deployment:
+#   python -c "import secrets; print(secrets.token_urlsafe(48))"
+# ============================================================================
+INTERNAL_SERVICE_TOKEN=change-me-internal-service-token

API.md

@@ -27,7 +27,7 @@
 6. [管理后台 (`/api/v1/admin/...`)](#六管理后台)
 7. [系统管理 (`/api/v1/platform/...`)](#七系统管理-apiv1platform)
 8. [Jupyter 路由 (Nginx `auth_request`)](#八jupyter-路由)
-9. [对象存储控制面 (`/internal/v1/...`,同进程 RPC)](#九对象存储控制面)
+9. [对象存储控制面 (`/internal/v1/objects`,服务间 RPC + Token 鉴权)](#九对象存储控制面)
 10. [健康检查](#十健康检查)
 
 ---
@@ -1006,53 +1006,29 @@ Nginx 把这些状态原样透传给浏览器,前端可在 `onerror` 里判断
 
 ## 九、对象存储控制面
 
-> 路径前缀 `/internal/v1/...`,**前端不要直接调用**。这是 backend 内部
-> 异步消息处理(Schedule worker)用的 RPC 端点,经 in-process ASGI 直接
-> 转发(`storage_app` 路由被 `app.include_router` 进同一个 backend 进程),
-> 外部无法访问。
+> **范围**:Schedule worker 调用 backend 写 `run_log` / `run_result` 用的
+> 单端点 RPC。**前端不要直接调用,也不要把这个路径用于任何用户输入**。
+> 历史上有 6 个 `/internal/v1/*` 端点,经过 P0-1 修复后只剩这 1 个;
+> 删除的端点全部已迁移到 JWT 保护的 `/api/v1/data-resources/*` 与
+> `/api/v1/scripts/*` 路由(见 §五、§三)。
 
 底层抽象:`common.storage.AsyncStorageBackend`(`put/get/delete/exists/stat/
 list/get_url/copy`)。按 `settings.storage_backend` 选实现:`"s3"` 走
 S3-兼容服务,`"local"` 走 `LOCAL_STORAGE_BASE_DIR` 子目录。
 
-### 9.1 `POST /internal/v1/uploads`
+### 鉴权(P0-1 后)
 
-创建上传会话。`Idempotency-Key` 必填,同 key + 同元数据 → 复用;同 key + 不同元数据 → 409。
+所有 `/internal/v1/*` 路由都会校验请求头里的 service token:
 
-```json
-{
-  "workspace_id": "...",
-  "user_id": "...",
-  "usage_type": "working_copy",
-  "file_name": "train.py",
-  "content_type": "text/x-python",
-  "expected_size_bytes": 1024,
-  "expected_hash": "<optional sha256 hex>",
-  "idempotency_key": "...",
-  "visibility": "private",
-  "is_immutable": false
-}
+```
+X-Internal-Service-Token: <settings.internal_service_token>
 ```
 
-返回 `{upload_id, status, upload_path, expires_at}`。`upload_path` 是
-第 9.2 步要 PUT 的端点(本进程内 `/internal/v1/uploads/{upload_id}`)。
+- backend 与 schedule 必须把同一个值注入到 `INTERNAL_SERVICE_TOKEN` 环境变量。
+- 缺失或不相符 → `401`。
+- backend 配置为空 → `503`(`internal service token not configured`)。
 
-### 9.2 `PUT /internal/v1/uploads/{upload_id}`
-
-完成上传(server-proxied PUT)。**请求体即原始字节**,`Content-Type:
-application/octet-stream`。后端 `await request.body()` 读字节 → 校验
-size + sha256 → 调 `await backend.put(key, bytes, content_type=...,
-metadata={"sha256": ...})` → 写 `StorageObjects` 行 → 标 session 为
-completed。最大 100 MiB。
-
-> 历史:旧版本这一步是 `POST /uploads/{id}/complete`,靠
-> presigned-PUT + head() 验证。已被 server-proxied PUT 取代。
-
-### 9.3 `POST /internal/v1/uploads/{upload_id}/abort`
-
-主动放弃。删除可能已经写了一半的对象字节,释放 `UploadSessions` 行。
-
-### 9.4 `POST /internal/v1/objects`
+### 9.1 `POST /internal/v1/objects`
 
 **单步创建**(不走两步上传,字节 base64 进 JSON 体)。适用 < 100 KiB
 对象(避免 multipart/大请求体的前端复杂度)。内部直接调
@@ -1068,24 +1044,13 @@ metadata={"sha256": ...})`。
   "content_type": "text/plain",
   "content_base64": "PHN0ZXAtY29udGVudD4=",
   "visibility": "private",
-  "is_immutable": false,
+  "is_immutable": true,
   "idempotency_key": "...",
   "relative_path": null
 }
 ```
 
-### 9.5 `POST /internal/v1/objects/{storage_object_id}/download-url`
-
-生成 presigned GET URL(s3 模式:`AsyncStorageBackend.get_url()`;
-local 模式:目前抛 `NotImplementedError`,需要 native FS serving 配合
-nginx 静态 location)。
-
-### 9.6 `DELETE /internal/v1/objects/{storage_object_id}`
-
-软删。`is_immutable == 1` 的对象拒绝删除。`move_to_trash` 走跨后端
-`copy + delete`(同一进程内的两个 backend 实例)。
-
-### 9.7 usage_type → 桶路由(自动)
+### 9.2 usage_type → 桶路由(自动)
 
 | usage_type | 实际桶(env var) | 默认桶名 |
 |---|---|---|
@@ -1099,6 +1064,16 @@ nginx 静态 location)。
 桶在 `STORAGE_BACKEND=s3` 时是 4 个独立 S3 bucket,在
 `STORAGE_BACKEND=local` 时是 `LOCAL_STORAGE_BASE_DIR` 下的 4 个子目录。
 
+### 9.3 已删除的端点(P0-1 收纳)
+
+| 旧端点 | 替代路由 | 说明 |
+|---|---|---|
+| `POST /internal/v1/uploads` | `POST /api/v1/data-resources/uploads`(JWT) | 前端走公开路径 |
+| `PUT /internal/v1/uploads/{id}` | `PUT /api/v1/data-resources/uploads/{id}` | 同上 |
+| `POST /internal/v1/uploads/{id}/abort` | (客户端取消即可) | 无后端状态 |
+| `POST /internal/v1/objects/{id}/download-url` | 各自的公开路由生成 presigned URL | download URL 由公开路由返回 |
+| `DELETE /internal/v1/objects/{id}` | 公开路由的删除操作 | 与 data-resource 联动 |
+
 ---
 
 ## 十、健康检查

DEVELOP.md

@@ -18,7 +18,7 @@ common/                Pure-Python shared library
   schemas.py           StrictModel base
   utils.py             get_free_port, start_process
 
-backend/               Public FastAPI service + internal /internal/v1/* sub-app
+backend/               Public FastAPI service + tiny /internal/v1/objects RPC
   main.py              lifespan + route registration
   jupyter.py           /api/v1/auth/jupyter — the ONLY auth entry
   scripts.py           CRUD for scripts/notebooks (object storage via AsyncStorageBackend)
@@ -27,7 +27,7 @@ backend/               Public FastAPI service + internal /internal/v1/* sub-app
   schedule_schemas.py  Pydantic request/response models
   admin.py             Admin endpoints
   resources.py         Misc data resources
-  storage_api.py       /internal/v1/* (sub-app merged into main)
+  storage_api.py       /internal/v1/objects — single token-guarded endpoint (P0-1)
   storage_client.py    Stub (HTTP client removed post-migration; rewrite pending)
   schedule_client.py   Placeholder module (was the HTTP-push executor client)
   runtime_client.py    Self-contained httpx wrapper for the runtime
@@ -41,7 +41,7 @@ schedule/              Schedule Executor (DAG worker)
   worker.py            NodeExecutor (notebook / python execution)
   service.py           SchedulerService facade (composes the three)
   main.py              Lifespan + FastAPI app
-  storage_client.py    Stub (SchedulerStorageClient rewrite pending — use AsyncStorageBackend directly)
+  storage_client.py    SchedulerStorageClient — talks to backend /internal/v1/objects
   execution.py         execute_artifact (notebook + python paths)
   notebook_runner.py   Subprocess entry point (nbclient)
 
@@ -164,6 +164,25 @@ grep -rnE 'os\.(environ\[?["\x27][A-Z_]+|getenv\(["\x27][A-Z_]+)' --include="*.p
   4. Calls `RuntimeClient.get_workspace` / `start_workspace`.
   5. Returns `x-upstream-addr` + `x-jupyter-internal-token` response
      headers. **Browser never holds the runtime token.**
+
+### Service-to-service auth (P0-1 fix)
+
+- Schedule → Backend single endpoint ``POST /internal/v1/objects`` is
+  guarded by ``require_internal_service`` in ``backend.storage_api``.
+- The token header is ``X-Internal-Service-Token`` (case-insensitive
+  on the wire because FastAPI ``Header`` lowercase-matches the name
+  ``x-internal-service-token``); the secret value comes from
+  ``settings.internal_service_token`` / env ``INTERNAL_SERVICE_TOKEN``.
+- Comparison uses ``secrets.compare_digest`` — never equality.
+- Backend and schedule must be configured with the same value; a
+  mismatch fails fast at the first notebook run (``401``) which is
+  intentional. ``.env.example`` ships a placeholder
+  ``change-me-internal-service-token`` and the docker-compose
+  ``${INTERNAL_SERVICE_TOKEN:?...}`` reference forces production
+  deployments to set a real value.
+- Removing the legacy backend / runtime host-port mappings
+  (``8891:8000`` / ``8892:8000``) is part of the same fix — no service
+  is reachable from the host except Nginx anymore.
 - Nginx captures the headers via `auth_request_set` and proxies to the
   upstream sub-process with `Authorization: token $jupyter_token`.
 

README.md

@@ -36,7 +36,8 @@
               ▼                                     ▼
    ┌──────────────────┐               ┌──────────────────────┐
    │ FastAPI Backend  │               │ Runtime (Jupyter)     │
-   │ + /internal/v1   │   control     │ - rclone FUSE mount  │
+   │ + /internal/v1   │   control     │ - rclone FUSE mount  │ (P0-1)
+   │   /objects       │   token-Auth  │                       │
    │ (storage)        ├──────────────►│ - subprocess pool    │
    │ - DAG CRUD       │               │   (per workspace)    │
    │ - script CRUD    │               └──────────┬───────────┘
@@ -93,12 +94,12 @@ default.conf                 Nginx 模板（挂载，启动时渲染）
 | 服务 | 镜像 | 暴露 | 用途 |
 |---|---|---|---|
 | `web` | `nginx:alpine` | 宿主机 `:8888` → `:80` | SPA、`/api/` 反向代理、`/jupyter/{ws}/` auth_request 代理、`/storage/` S3 直通（仅 s3 模式） |
-| `backend` | `Dockerfile` | 仅内网 | DAG CRUD、script CRUD、schedule 触发、`/api/v1/auth/jupyter`、`/internal/v1/*` 存储控制面 |
+| `backend` | `Dockerfile` | 仅内网 | DAG CRUD、script CRUD、schedule 触发、`/api/v1/auth/jupyter`、`/internal/v1/objects` 服务间 RPC(共享 `INTERNAL_SERVICE_TOKEN` 鉴权,P0-1)|
 | `runtime` | `Dockerfile` | 仅内网 | 每个 workspace 一个 Jupyter 子进程池、rclone FUSE 挂载 `workspace` 桶（s3 模式） |
 | `schedule` | `Dockerfile` | 仅内网 | cron tick + DAG 执行（轮询 MySQL Outbox） |
 
-架构**故意只暴露一个宿主机端口**（网关）；其他服务都在 Docker 内网。
-这一点在 `docker-compose.yml` 里强制执行 — backend / runtime / schedule 都没有 `ports:`。
+架构**故意只暴露一个宿主机端口**(网关);其他服务都在 Docker 内网。
+这一点在 `docker-compose.yml` 里强制执行 — backend / runtime / schedule 都没有 `ports:`。在 P0-1 之前,backend 与 runtime 曾短暂地把 `8891` / `8892` 映射到宿主机;此映射已被删除,改用 `INTERNAL_SERVICE_TOKEN` 头对 `/internal/v1/*` 做服务间鉴权,见 `API.md §9`。
 
 ## 快速启动
 

backend/src/backend/main.py

@@ -52,9 +52,15 @@ async def lifespan(app: Any) -> AsyncIterator[None]:
         for purpose in PURPOSE_BUCKETS
     }
     app.state.default_bucket = settings.s3_workspace_bucket
+    # P0-1 fix: runtime's /api/v1/jupyter is token-guarded. The token is
+    # the same ``INTERNAL_SERVICE_TOKEN`` value used by /internal/v1/* —
+    # reusing one mechanism instead of inventing a second one.
     runtime_http_client = httpx.AsyncClient(
         base_url=settings.runtime_api_url,
         timeout=httpx.Timeout(30.0),
+        headers={
+            "X-Internal-Service-Token": settings.internal_service_token,
+        },
     )
     app.state.runtime_client = RuntimeClient(runtime_http_client)
     # Short timeout — refresh is best-effort and runs in a BackgroundTask.

backend/src/backend/storage_api.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import hashlib
+import secrets
 from collections.abc import AsyncIterator
 from datetime import UTC, datetime, timedelta
 from pathlib import PurePosixPath
@@ -19,22 +20,49 @@ from common.ids import new_ulid
 from common.storage import USAGE_TYPE_TO_PURPOSE, actual_bucket_name, build_storage_uri
 from common.storage.schemas import (
     CreateUploadRequest,
-    DownloadUrlRequest,
     ServerObjectRequest,
 )
-from fastapi import APIRouter, Depends, HTTPException, Request, status
+from fastapi import APIRouter, Depends, Header, HTTPException, Request, status
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from backend.services.storage import (
-    create_download_url_payload,
     create_server_object_payload,
     create_upload_record,
-    soft_delete_object,
     upload_bytes_to_session,
 )
 
 
+# Header name for the service-to-service token. Schedule worker is the
+# only legitimate caller — every other consumer goes through the
+# JWT-protected ``/api/v1/data-resources/*`` routes. The header name
+# mirrors the ``X-Internal-*`` convention used elsewhere in the stack.
+INTERNAL_SERVICE_TOKEN_HEADER = "x-internal-service-token"
+
+
+def require_internal_service(
+    x_internal_service_token: str | None = Header(default=None),
+) -> None:
+    """Enforce a shared secret for /internal/v1/* routes.
+
+    Compares the supplied header against ``settings.internal_service_token``
+    with a constant-time check. The token is configured identically on the
+    backend and the schedule container via ``INTERNAL_SERVICE_TOKEN``; the
+    default in ``Settings`` is a development-only placeholder that callers
+    must override in any non-dev deployment.
+    """
+    expected = settings.internal_service_token
+    if not expected:
+        raise HTTPException(
+            status.HTTP_503_SERVICE_UNAVAILABLE,
+            "internal service token not configured",
+        )
+    if not x_internal_service_token or not secrets.compare_digest(
+        x_internal_service_token, expected
+    ):
+        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "internal service token required")
+
+
 def utcnow() -> datetime:
     return datetime.now(UTC).replace(tzinfo=None)
 
@@ -358,82 +386,25 @@ async def upload_bytes_to_session(
     return item
 
 
-@router.post("/v1/uploads")
-async def create_upload(
-    payload: CreateUploadRequest,
-    request: Request,
-    session: AsyncSession = Depends(database_session),
-) -> dict[str, Any]:
-    return {
-        "data": await create_upload_record(payload, session, request),
-    }
-
-
-@router.put("/v1/uploads/{upload_id}")
-async def upload_bytes(
-    upload_id: str, request: Request, session: AsyncSession = Depends(database_session)
-) -> dict[str, Any]:
-    """Server-proxied upload: PUT raw bytes in the request body. Replaces the
-    old ``POST /uploads/{id}/complete`` flow that paired presigned-PUT with
-    a head()-validate step.
-    """
-    item = await upload_bytes_to_session(upload_id, session, request)
-    return {"data": storage_payload(item)}
-
-
-@router.post("/v1/uploads/{upload_id}/abort")
-async def abort_upload(
-    upload_id: str, request: Request, session: AsyncSession = Depends(database_session)
-) -> dict[str, Any]:
-    upload = await session.scalar(
-        select(UploadSessions)
-        .where(UploadSessions.upload_id == upload_id)
-        .with_for_update()
-    )
-    if upload is None:
-        raise HTTPException(status.HTTP_404_NOT_FOUND, "upload not found")
-    if upload.upload_status == "completed":
-        raise HTTPException(
-            status.HTTP_409_CONFLICT, "completed upload cannot be aborted"
-        )
-    if upload.upload_status != "aborted":
-        await request.app.state.object_stores[upload.bucket_name].delete(
-            upload.object_key
-        )
-        upload.upload_status = "aborted"
-    return {"data": {"upload_id": upload_id, "status": "aborted"}}
-
-
-@router.post("/v1/objects")
+@router.post(
+    "/v1/objects",
+    dependencies=[Depends(require_internal_service)],
+)
 async def create_server_object(
     payload: ServerObjectRequest,
     request: Request,
     session: AsyncSession = Depends(database_session),
 ) -> dict[str, Any]:
+    """Server-side single-call object upload used by the schedule worker.
+
+    Token-guarded: the only legitimate caller is the schedule service
+    that writes ``run_log`` / ``run_result`` artifacts after a notebook
+    finishes. Frontend users upload through the JWT-protected
+    ``/api/v1/data-resources/*`` routes instead.
+    """
     return await create_server_object_payload(payload, request, session)
 
 
-@router.post("/v1/objects/{storage_object_id}/download-url")
-async def create_download_url(
-    storage_object_id: str,
-    payload: DownloadUrlRequest,
-    request: Request,
-    session: AsyncSession = Depends(database_session),
-) -> dict[str, Any]:
-    item = await session.get(StorageObjects, storage_object_id)
-    return await create_download_url_payload(item, payload, request)
-
-
-@router.delete("/v1/objects/{storage_object_id}")
-async def delete_object(
-    storage_object_id: str,
-    request: Request,
-    session: AsyncSession = Depends(database_session),
-) -> dict[str, Any]:
-    """Soft-delete a storage object. See ``backend.services.storage.soft_delete_object``."""
-    return await soft_delete_object(storage_object_id, request, session)
-
-
 @router.post("/v1/objects/{storage_object_id}/restore")
 async def restore_object(
     storage_object_id: str,

common/src/common/config.py

@@ -152,6 +152,22 @@ class Settings(BaseSettings):
         description="Schedule → Backend HTTP base URL (cron post-back).",
     )
 
+    # ── service-to-service auth for /internal/v1/* (P0-1 fix) ─────
+    # Shared secret between backend and the schedule worker. The schedule
+    # posts the value in the ``X-Internal-Service-Token`` header when it
+    # uploads ``run_log`` / ``run_result`` artifacts. Backend's storage
+    # API rejects requests whose header does not match this value.
+    # Override via ``INTERNAL_SERVICE_TOKEN``; the placeholder default is
+    # safe for local dev with the matching schedule config but must be
+    # replaced in any non-dev deployment.
+    internal_service_token: str = Field(
+        default="dev-only-internal-token-not-for-production",
+        description=(
+            "Shared secret for service-to-service auth on /internal/v1/*. "
+            "Set identically on backend and schedule via INTERNAL_SERVICE_TOKEN."
+        ),
+    )
+
     # ── runtime public base URL ──────────────────────────────────
     public_base_url: str = Field(
         default="http://runtime",

docker-compose.yml

@@ -72,6 +72,9 @@ services:
         max-file: "10"
     restart: unless-stopped
     # No host port: architecture §2.2 — only Nginx is externally reachable.
+    # The previous ``8891:8000`` mapping (P0-1) was removed: the
+    # ``/internal/v1/*`` storage control plane is now guarded by a
+    # shared ``INTERNAL_SERVICE_TOKEN`` instead of network isolation.
     # No local-FS volume: backend stores everything in S3 (S3_*).
     environment:
       DATABASE_URL: ${DATABASE_URL:?DATABASE_URL is required}
@@ -81,6 +84,8 @@ services:
       DEMO_AUTH_ENABLED: ${DEMO_AUTH_ENABLED:-false}
       INITIAL_ADMIN_PASSWORD: ${INITIAL_ADMIN_PASSWORD:-admin12345}
       RUNTIME_API_URL: http://runtime:8000
+      # P0-1 fix: shared secret required by /internal/v1/* routes.
+      INTERNAL_SERVICE_TOKEN: ${INTERNAL_SERVICE_TOKEN:?INTERNAL_SERVICE_TOKEN is required}
       STORAGE_BACKEND: ${STORAGE_BACKEND:-s3}
       LOCAL_STORAGE_BASE_DIR: ${LOCAL_STORAGE_BASE_DIR:-/data}
       # S3_* only matter when STORAGE_BACKEND=s3. Defaults are kept so local
@@ -101,8 +106,6 @@ services:
         condition: service_completed_successfully
       runtime:
         condition: service_healthy
-    ports:
-      - 8891:8000
     volumes:
       - ${PWD}:/app
       - ./data:/data
@@ -125,8 +128,10 @@ services:
         max-size: "200m"
         max-file: "10"
     restart: unless-stopped
-    ports:
-      - 8892:8000
+    # No host port: architecture §2.2 — only Nginx is externally reachable.
+    # The previous ``8892:8000`` mapping (P0-1) was removed: the runtime
+    # container is reachable only from the Docker internal network and
+    # Nginx-authenticated Jupyter paths.
     cap_add:
       - SYS_ADMIN
     devices:
@@ -137,6 +142,10 @@ services:
     environment:
       DATABASE_URL: ${DATABASE_URL:?DATABASE_URL is required}
       SERVICE_NAME: runtime-manager
+      # P0-1 fix: runtime's /api/v1/jupyter is token-guarded. The same
+      # ``INTERNAL_SERVICE_TOKEN`` value backend uses for /internal/v1/*
+      # auth — see ``require_internal_service`` in runtime/main.py.
+      INTERNAL_SERVICE_TOKEN: ${INTERNAL_SERVICE_TOKEN:?INTERNAL_SERVICE_TOKEN is required}
       STORAGE_BACKEND: ${STORAGE_BACKEND:-s3}
       LOCAL_STORAGE_BASE_DIR: ${LOCAL_STORAGE_BASE_DIR:-/data}
       # WORKSPACES_ROOT defaults to /data/workspace (settings.workspaces_root);
@@ -190,6 +199,8 @@ services:
       SERVICE_NAME: schedule-executor
       SCHEDULE_EVENT_NAMESPACE: ${SCHEDULE_EVENT_NAMESPACE:-model-platform-local}
       BACKEND_API_URL: http://backend:8000
+      # P0-1 fix: must match the backend's INTERNAL_SERVICE_TOKEN exactly.
+      INTERNAL_SERVICE_TOKEN: ${INTERNAL_SERVICE_TOKEN:?INTERNAL_SERVICE_TOKEN is required}
       STORAGE_BACKEND: ${STORAGE_BACKEND:-s3}
       LOCAL_STORAGE_BASE_DIR: ${LOCAL_STORAGE_BASE_DIR:-/data}
       S3_ENDPOINT: ${S3_ENDPOINT:-http://s3:9000}

runtime/src/runtime/main.py

@@ -8,12 +8,21 @@ surface is two endpoints; everything else is lifespan orchestration.
 from __future__ import annotations
 
 import asyncio
+import secrets
+from collections.abc import AsyncIterator
 from contextlib import asynccontextmanager
 from typing import Literal
 
-from fastapi import FastAPI, HTTPException
+from fastapi import Depends, FastAPI, Header, HTTPException, status
 from loguru import logger
 from pydantic import BaseModel, ConfigDict
+
+# Import Settings directly: the runtime process does not own a DB
+# session, so we can't reuse ``request_context``. We DO reuse the same
+# shared secret idea (``/internal/v1/*`` uses the exact same value),
+# because inventing a second token scheme for one extra hop would be
+# pure complexity.
+from common.config import settings
 from runtime.mount import start_rclone_mount, stop_rclone_mount
 from runtime.process import (
     JUPYTER_PROCESSES,
@@ -27,6 +36,37 @@ from runtime.process import (
     stop_workspace,
 )
 
+INTERNAL_SERVICE_TOKEN_HEADER = "x-internal-service-token"
+
+
+def require_internal_service(
+    x_internal_service_token: str | None = Header(default=None),
+) -> None:
+    """Enforce a shared secret on /api/v1/jupyter.
+
+    The only legitimate caller is the backend (already authenticated
+    via cookie / Bearer JWT), which forwards the request after running
+    ``request_context``. The runtime process does not own a DB session
+    so it cannot validate the JWT itself — the token header is the
+    cheaper defense-in-depth equivalent.
+
+    Empty backend config -> 503 (we'd rather fail loud than silently
+    allow all callers when deployment hasn't been initialised).
+    """
+    expected = settings.internal_service_token
+    if not expected:
+        raise HTTPException(
+            status.HTTP_503_SERVICE_UNAVAILABLE,
+            "internal service token not configured",
+        )
+    if not x_internal_service_token or not secrets.compare_digest(
+        x_internal_service_token, expected
+    ):
+        raise HTTPException(
+            status.HTTP_401_UNAUTHORIZED,
+            "internal service token required",
+        )
+
 
 class JupyterActionRequest(BaseModel):
     action: Literal["start", "stop", "list", "get"]
@@ -91,7 +131,7 @@ def healthz() -> dict:
     return {"status": "ok"}
 
 
-@app.post("/api/v1/jupyter")
+@app.post("/api/v1/jupyter", dependencies=[Depends(require_internal_service)])
 async def handle_jupyter_action(req: JupyterActionRequest) -> dict:
     match req.action:
         case "start":

schedule/src/schedule/service.py

@@ -225,12 +225,20 @@ def build_storage_http_client() -> httpx.AsyncClient:
     The schedule service no longer needs to call any user-facing
     endpoint (cron trigger writes directly to the DB now), but the
     storage endpoints at ``/internal/v1/...`` still live on the
-    backend process and are reached via this client. Auth is not
-    required — the client is bound to the shared Docker network.
+    backend process and are reached via this client.
+
+    Auth (P0-1 fix): every backend ``/internal/v1/*`` route checks the
+    ``X-Internal-Service-Token`` header against
+    ``settings.internal_service_token``. The schedule reads the same
+    env var so a value mismatch causes a 401 on the first upload — that
+    is intentional, not a bug.
     """
     return httpx.AsyncClient(
         base_url=settings.backend_api_url,
         timeout=httpx.Timeout(60.0),
+        headers={
+            "X-Internal-Service-Token": settings.internal_service_token,
+        },
     )