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>
This commit is contained in:
co-authored by
Claude Fable 5
parent
6258cf5d12
commit
dfe3f0b118
@@ -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
|
||||
|
||||
@@ -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 联动 |
|
||||
|
||||
---
|
||||
|
||||
## 十、健康检查
|
||||
|
||||
+22
-3
@@ -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`.
|
||||
|
||||
|
||||
@@ -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`。
|
||||
|
||||
## 快速启动
|
||||
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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",
|
||||
|
||||
+15
-4
@@ -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}
|
||||
|
||||
@@ -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":
|
||||
|
||||
@@ -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,
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user