diff --git a/.env.docker.example b/.env.docker.example index 7098ad3..de51856 100644 --- a/.env.docker.example +++ b/.env.docker.example @@ -3,6 +3,8 @@ BIND_ADDRESS=127.0.0.1 API_PORT=3000 DASHBOARD_PORT=3001 MOCK_HOOK_PORT=3101 +# Add the external dashboard hostname for shared deployments; no wildcards. +DASHBOARD_ALLOWED_HOSTS=127.0.0.1,localhost,[::1] # Replace these local-only defaults before any shared or production deployment. # Generate independent values with: openssl rand -hex 32 @@ -11,6 +13,17 @@ DB_USER=waitqueue DB_PASSWORD='replace-with-a-strong-password' DB_ROOT_PASSWORD='replace-with-a-different-strong-password' +# Optional application boundary. Empty values preserve the local compatibility mode. +# Generate a shared API token with: openssl rand -hex 32 +WAITQUEUE_API_TOKEN= +# Comma-separated exact origins, for example: https://worker.example.com +HOOK_URL_ALLOWLIST= +# Development only. Keep false in shared and production environments. +HOOK_URL_ALLOW_PRIVATE=false +REQUEST_BODY_LIMIT_BYTES=32768 +RATE_LIMIT_MAX_REQUESTS=0 +RATE_LIMIT_WINDOW_MS=60000 + # Scheduler settings HOOK_TIMEOUT_MS=10000 CHECK_TASK_DIFF_CRON=0 * * * * * diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index b2ebd66..b5dbde7 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -221,6 +221,7 @@ jobs: DASHBOARD_PORT: 3001 DB_DATABASE: waitqueue_ci DB_USER: waitqueue_ci + HOOK_URL_ALLOWLIST: https://worker.example.com,http://127.0.0.1:3101 steps: - name: Check out repository uses: actions/checkout@v7 @@ -233,11 +234,14 @@ jobs: set -Eeuo pipefail db_password="$(openssl rand -hex 24)" db_root_password="$(openssl rand -hex 24)" + api_token="$(openssl rand -hex 32)" echo "::add-mask::$db_password" echo "::add-mask::$db_root_password" + echo "::add-mask::$api_token" { echo "DB_PASSWORD=$db_password" echo "DB_ROOT_PASSWORD=$db_root_password" + echo "WAITQUEUE_API_TOKEN=$api_token" } >> "$GITHUB_ENV" - name: Validate Compose configuration @@ -269,10 +273,96 @@ jobs: '.code == 0 and .data.status == "ready" and .data.dependencies.mysql == "ok" and .data.dependencies.redis == "ok"' \ <<< "$ready_json" > /dev/null + unauthenticated_status="$(curl --silent --show-error --output /dev/null --write-out '%{http_code}' \ + "http://127.0.0.1:${API_PORT}/waitqueue/admin/overview")" + test "$unauthenticated_status" = "401" + + wrong_token_status="$(curl --silent --show-error --output /dev/null --write-out '%{http_code}' \ + --header 'Authorization: Bearer definitely-wrong' \ + "http://127.0.0.1:${API_PORT}/waitqueue/admin/overview")" + test "$wrong_token_status" = "401" + + authenticated_json="$(curl --fail --silent --show-error \ + --header "Authorization: Bearer ${WAITQUEUE_API_TOKEN}" \ + "http://127.0.0.1:${API_PORT}/waitqueue/admin/overview")" + jq --exit-status '.code == 0 and .data.summary.queueCount == 0' \ + <<< "$authenticated_json" > /dev/null + + private_hook_status="$(curl --silent --show-error --output /dev/null --write-out '%{http_code}' \ + --request POST \ + --header "Authorization: Bearer ${WAITQUEUE_API_TOKEN}" \ + --header 'Content-Type: application/json' \ + --data '{"namespace":"blocked","hookUrl":"http://127.0.0.1:3101/callback"}' \ + "http://127.0.0.1:${API_PORT}/waitqueue/queue/newQueue")" + test "$private_hook_status" = "400" + + oversized_status="$(python3 -c \ + 'import json; print(json.dumps({"namespace":"large","hookUrl":"https://worker.example.com/" + "x" * 40000}))' \ + | curl --silent --show-error --output /dev/null --write-out '%{http_code}' \ + --request POST \ + --header "Authorization: Bearer ${WAITQUEUE_API_TOKEN}" \ + --header 'Content-Type: application/json' \ + --data-binary @- \ + "http://127.0.0.1:${API_PORT}/waitqueue/queue/newQueue")" + test "$oversized_status" = "413" + dashboard_html="$(curl --fail --silent --show-error --retry 10 --retry-all-errors --retry-delay 2 \ "http://127.0.0.1:${DASHBOARD_PORT}/")" grep --fixed-strings --quiet 'WaitQueue Control Room' <<< "$dashboard_html" + proxy_content_type_status="$(curl --silent --show-error --output /dev/null --write-out '%{http_code}' \ + --request POST \ + --header 'Content-Type: text/plain' \ + --data 'not-json' \ + "http://127.0.0.1:${DASHBOARD_PORT}/waitqueue/queue/newQueue")" + test "$proxy_content_type_status" = "415" + + proxy_jsonp_status="$(curl --silent --show-error --output /dev/null --write-out '%{http_code}' \ + --request POST \ + --header 'Content-Type: application/jsonp' \ + --data '{}' \ + "http://127.0.0.1:${DASHBOARD_PORT}/waitqueue/queue/newQueue")" + test "$proxy_jsonp_status" = "415" + + untrusted_host_status="$(curl --silent --show-error --output /dev/null --write-out '%{http_code}' \ + --header "Host: attacker.example:${DASHBOARD_PORT}" \ + "http://127.0.0.1:${DASHBOARD_PORT}/waitqueue/admin/overview")" + test "$untrusted_host_status" = "403" + + proxy_oversized_status="$(python3 -c \ + 'import json; print(json.dumps({"namespace":"large","hookUrl":"https://worker.example.com/" + "x" * 40000}))' \ + | curl --silent --show-error --output /dev/null --write-out '%{http_code}' \ + --request POST \ + --header 'Content-Type: application/json' \ + --data-binary @- \ + "http://127.0.0.1:${DASHBOARD_PORT}/waitqueue/queue/newQueue")" + test "$proxy_oversized_status" = "413" + + proxy_queue_json="$(curl --fail --silent --show-error \ + --request POST \ + --header 'Authorization: Bearer browser-controlled-value' \ + --header 'Cookie: session=browser-controlled-value' \ + --header 'Content-Type: application/json' \ + --data '{"namespace":"ci","hookUrl":"https://worker.example.com/callback","currMaxCount":2,"crontab":{"run":"0 0 0 1 1 *","check":"0 0 0 1 1 *","expire":"0 0 0 1 1 *"}}' \ + "http://127.0.0.1:${DASHBOARD_PORT}/waitqueue/queue/newQueue")" + jq --exit-status '.code == 0' <<< "$proxy_queue_json" > /dev/null + + proxy_task_json="$(curl --fail --silent --show-error \ + --request POST \ + --header 'Content-Type: application/json' \ + --data '{"namespace":"ci","hookUrl":"https://worker.example.com/callback","taskId":"ci-proxy-task"}' \ + "http://127.0.0.1:${DASHBOARD_PORT}/waitqueue/scheduler/addTask")" + jq --exit-status '.code == 0' <<< "$proxy_task_json" > /dev/null + + proxied_json="$(curl --fail --silent --show-error \ + "http://127.0.0.1:${DASHBOARD_PORT}/waitqueue/admin/overview")" + jq --exit-status \ + '.code == 0 and .data.summary.queueCount == 1 and .data.summary.waiting == 1' \ + <<< "$proxied_json" > /dev/null + + docker compose exec -T dashboard sh -eu -c \ + '! grep --recursive --fixed-strings --quiet "$WAITQUEUE_API_TOKEN" /app/.next/static' + - name: Verify migration history and idempotency shell: bash run: | diff --git a/Dockerfile b/Dockerfile index 7450907..155624b 100644 --- a/Dockerfile +++ b/Dockerfile @@ -74,8 +74,6 @@ RUN --mount=type=cache,id=waitqueue-dashboard-pnpm,target=/pnpm/store \ FROM dashboard-deps AS dashboard-build -ARG WAITQUEUE_API_URL=http://api:3000 -ENV WAITQUEUE_API_URL=$WAITQUEUE_API_URL ENV NODE_ENV=production COPY admin-dashboard/next.config.js admin-dashboard/tsconfig.json ./ @@ -119,4 +117,3 @@ EXPOSE 3101 HEALTHCHECK --interval=15s --timeout=5s --start-period=5s --retries=3 \ CMD node -e "fetch('http://127.0.0.1:3101/health').then((response) => process.exit(response.ok ? 0 : 1)).catch(() => process.exit(1))" CMD ["node", "mock-hook.mjs"] - diff --git a/README.md b/README.md index d8e9aa7..2adf8b2 100644 --- a/README.md +++ b/README.md @@ -6,7 +6,7 @@ ![WaitQueue Control Room](docs/control-room.jpg) -> 当前定位是内部服务与二次开发基础设施。API 尚未内置鉴权，`hookUrl` 也会被服务端主动访问；生产部署必须放在可信网络或认证网关之后。 +> 当前定位是内部服务与二次开发基础设施。项目内置可选 Bearer token、精确回调 origin 允许列表、请求大小限制和轻量限流；为兼容本地开发，token 与允许列表默认为空，共享或生产环境必须显式开启并放在带 TLS 和用户认证的网关之后。 ## 它解决什么问题 @@ -72,13 +72,16 @@ Compose 会按 `MySQL → 数据库迁移 → API → 控制台` 的顺序启动 - API 存活检查：[http://127.0.0.1:3000/waitqueue/health](http://127.0.0.1:3000/waitqueue/health) - API 就绪检查：[http://127.0.0.1:3000/waitqueue/ready](http://127.0.0.1:3000/waitqueue/ready) -默认只监听 `127.0.0.1`，数据库凭据也只为隔离的本地体验准备。共享或生产环境应先复制配置并替换两个密码；生产暴露应通过带认证与 TLS 的网关完成，而不是把本项目端口直接绑定到公网： +默认只监听 `127.0.0.1`，数据库凭据也只为隔离的本地体验准备。空 token 和空回调允许列表是兼容模式，不是生产安全默认值。共享或生产环境应先复制配置，替换两个数据库密码，用 `openssl rand -hex 32` 生成 API token，并按实际回调服务填写精确 origin： ```bash cp .env.docker.example .env +# 编辑 .env：填写独立密码、API token、回调 origin 和外部控制台主机名 docker compose up --build --detach --wait ``` +`HOOK_URL_ALLOWLIST` 是逗号分隔的精确 origin，例如 `https://worker.example.com,https://jobs.example.net:8443`；不支持通配符、路径、query 或 fragment。严格模式还会拒绝本机/私网字面量，并在真正连接时校验且固定 DNS 解析结果。生产暴露应通过带用户认证与 TLS 的网关完成，而不是把本项目端口直接绑定到公网。 + 查看状态与日志： ```bash @@ -98,7 +101,7 @@ docker compose down docker compose --profile demo up --build --detach --wait ``` -此时注册队列时使用容器网络地址 `http://mock-hook:3101/callback`。宿主机仍可通过 `http://127.0.0.1:3101/health` 检查示例回调。 +此时注册队列时使用容器网络地址 `http://mock-hook:3101/callback`。如果已开启回调允许列表，还需在根目录 `.env` 中同时加入 `HOOK_URL_ALLOWLIST=http://mock-hook:3101` 和 `HOOK_URL_ALLOW_PRIVATE=true`。后者是仅供隔离演示环境的显式逃生开关，共享/生产必须保持 `false`。宿主机仍可通过 `http://127.0.0.1:3101/health` 检查示例回调。 MySQL 与 Redis 数据保存在命名卷中。只有确认要清空全部队列配置和运行态时，才执行 `docker compose down --volumes`；该操作不可从 Compose 自动恢复。 @@ -151,7 +154,11 @@ corepack pnpm --dir wait-queue build corepack pnpm --dir wait-queue migrate ``` -默认配置可直接连接本机 `waitqueue` 数据库和 Redis；非默认账号、端口或密码请在迁移前修改 `wait-queue/.env`。迁移器会按版本执行 `V*.sql`、校验历史文件并跳过已经应用的版本；不要手工重放 SQL 文件。`U*.sql` 是破坏性回滚脚本，不属于正常启动流程。 +默认配置可直接连接本机 `waitqueue` 数据库和 Redis；非默认账号、端口或密码请在迁移前修改 `wait-queue/.env`。如需开启安全边界，还要在 `wait-queue/.env` 填写 `WAITQUEUE_API_TOKEN` 和 `HOOK_URL_ALLOWLIST`，并在后续的 `admin-dashboard/.env.local` 填写同一个 token。token 只由两个服务端读取，不应放入 `NEXT_PUBLIC_*` 变量。 + +下文的手动演示回调位于本机；若同时演示严格允许列表，请设置 `HOOK_URL_ALLOWLIST=http://127.0.0.1:3101` 和 `HOOK_URL_ALLOW_PRIVATE=true`。这个逃生开关只为隔离的本地演示准备，共享/生产必须保持 `false`。 + +迁移器会按版本执行 `V*.sql`、校验历史文件并跳过已经应用的版本；不要手工重放 SQL 文件。`U*.sql` 是破坏性回滚脚本，不属于正常启动流程。 从已有数据库升级前，先做可恢复性已验证的备份，并至少完成以下预检： @@ -191,9 +198,15 @@ corepack pnpm --dir admin-dashboard dev 也可以直接在控制台完成这两步。下面的 curl 便于验证 API： +```bash +# 本地兼容模式保持为空；开启鉴权时改为与后端一致的值。 +export WAITQUEUE_API_TOKEN='' +``` + ```bash curl -X POST http://127.0.0.1:3000/waitqueue/queue/newQueue \ -H 'Content-Type: application/json' \ + -H "Authorization: Bearer ${WAITQUEUE_API_TOKEN}" \ -d '{ "namespace": "demo", "hookUrl": "http://127.0.0.1:3101/callback", @@ -207,6 +220,7 @@ curl -X POST http://127.0.0.1:3000/waitqueue/queue/newQueue \ curl -X POST http://127.0.0.1:3000/waitqueue/scheduler/addTask \ -H 'Content-Type: application/json' \ + -H "Authorization: Bearer ${WAITQUEUE_API_TOKEN}" \ -d '{ "namespace": "demo", "hookUrl": "http://127.0.0.1:3101/callback", @@ -285,8 +299,14 @@ docker compose config --quiet | `DB_USER` / `DB_PASSWORD` | `root` / 空 | MySQL 凭据 | | `REDIS_HOST` / `REDIS_PORT` | `127.0.0.1` / `6379` | Redis 地址 | | `REDIS_PASSWORD` | 空 | Redis 密码 | +| `WAITQUEUE_API_TOKEN` | 空 | 非空时要求管理、队列和调度 API 携带同值 Bearer token | +| `HOOK_URL_ALLOWLIST` | 空 | 逗号分隔的精确 HTTP(S) origin；非空时拒绝其他回调地址 | +| `HOOK_URL_ALLOW_PRIVATE` | `false` | 仅本地演示使用；`true` 时允许已显式列入的本机/私网回调 | +| `REQUEST_BODY_LIMIT_BYTES` | `32768` | JSON 请求体上限，单位字节 | +| `RATE_LIMIT_MAX_REQUESTS` | `0` | 单进程、单客户端窗口内的最大请求数；`0` 关闭 | +| `RATE_LIMIT_WINDOW_MS` | `60000` | 限流固定窗口，单位毫秒 | -端口和超时必须为正整数。cron 使用“秒 分 时 日 月 周”六段格式。 +端口、超时和安全数值必须符合表中约束，无效值会让进程在启动时失败。cron 使用“秒 分 时 日 月 周”六段格式。 ### 控制台 @@ -295,12 +315,14 @@ docker compose config --quiet | 变量 | 默认值 | 说明 | | --- | --- | --- | | `WAITQUEUE_API_URL` | `http://127.0.0.1:3000` | Next.js 服务端代理的后端地址 | +| `WAITQUEUE_API_TOKEN` | 空 | 与后端相同的共享 token，由服务端代理注入 | +| `DASHBOARD_ALLOWED_HOSTS` | `127.0.0.1,localhost,[::1]` | 逗号分隔的精确控制台主机名，用于阻断 DNS rebinding | -生产环境应在 `build` 和 `start` 阶段提供一致的值。这个变量只在 Next.js 服务端使用，不会打进浏览器代码。 +三个变量都在 `start` 运行时由 Next.js 服务端读取，不会打进浏览器代码或镜像构建层。`DASHBOARD_ALLOWED_HOSTS` 不接受 scheme、路径或通配符，比较时忽略端口；共享域名部署必须显式加入外部 hostname，反向代理应保留或改写为该允许值。 ## HTTP API -所有路径都以 `/waitqueue` 开头，请求和响应使用 JSON。成功响应统一为： +所有路径都以 `/waitqueue` 开头，请求和响应使用 JSON。`WAITQUEUE_API_TOKEN` 非空时，`/admin/*`、`/queue/*` 和 `/scheduler/*` 必须带 `Authorization: Bearer `；`GET /health`、`GET /ready` 与 `OPTIONS` 保持无鉴权，便于探针与预检。成功响应统一为： ```json { @@ -310,7 +332,14 @@ docker compose config --quiet } ``` -参数错误返回 HTTP 400，资源不存在返回 404，不支持的方法返回 405，未处理异常返回 500。调用方应同时判断 HTTP 状态码和响应体 `code`。 +参数错误返回 HTTP 400，未认证返回 401，资源不存在返回 404，请求体过大返回 413，限流返回 429 并带 `Retry-After`，不支持的方法返回 405，未处理异常返回 500。调用方应同时判断 HTTP 状态码和响应体 `code`。 + +### 安全边界 + +- API token 是服务间共享凭据，不是用户登录或细粒度授权。控制台的服务端代理只转发三个明确的 API，丢弃浏览器传入的 Authorization、Cookie 和转发头，再注入服务端 token。任何能访问控制台的人仍可借此操作队列，因此共享部署仍需认证网关。 +- 回调允许列表按 WHATWG URL 归一化后精确比较 origin，每次真正发送前会再校验。严格模式拒绝 loopback、link-local、私网与本地主机名；域名的所有 DNS 结果也会在连接前校验，实际 socket 固定使用已校验地址。回调不跟随 3xx。`HOOK_URL_ALLOW_PRIVATE=true` 仅用于显式列入的隔离本地演示服务。 +- 内存限流按 API 进程和直连 IP 生效，不信任 `X-Forwarded-For`。多副本或公网环境要由网关补充全局限流；对出站网络要求更强隔离时，应配置出站代理或网络策略。 +- 所有写请求（成功或失败）与鉴权/限流拒绝都会生成结构化审计日志；请求体、token、Cookie、完整回调 URL 和 taskId 不会进入审计字段。 ### 健康检查 @@ -477,14 +506,15 @@ docker compose config --quiet - 浅色/深色主题与移动端布局； - 离线、过期、加载和空数据状态。 -浏览器只请求当前控制台域名；Next.js 根据 `WAITQUEUE_API_URL` 代理 API，因此后端无需 CORS。页面不伪造历史趋势、成功率或平均耗时，因为当前存储模型没有这些数据。 +浏览器只请求当前控制台域名；Next.js 运行时服务端代理按白名单转发 API，并在配置时注入 `WAITQUEUE_API_TOKEN`，因此后端无需 CORS，token 也不进入浏览器 bundle。页面不伪造历史趋势、成功率或平均耗时，因为当前存储模型没有这些数据。 更多前端说明见 [admin-dashboard/README.md](admin-dashboard/README.md)。 ## 已知边界 -- 所有 API 当前都没有鉴权；只应暴露在可信网络中，并由网关补充认证、授权与限流。 -- `hookUrl` 会被服务端主动请求。开放注册能力前必须增加主机/网段白名单，防范 SSRF。 +- API token 和回调允许列表为空时保持兼容模式；该模式只适合隔离的本地开发。共享环境必须显式配置两者。 +- 内置 token 是单一共享凭据，控制台代理也不是用户登录系统；多用户环境仍需由网关补充认证、授权、TLS 和全局限流。 +- `hookUrl` 会被服务端主动请求。严格模式已校验并固定 DNS 结果，但出站代理/网络策略仍是生产环境必要的纵深防御。不要在共享或生产环境开启 `HOOK_URL_ALLOW_PRIVATE`。 - cron 在应用进程内运行，没有 leader election；当前推荐单实例，多实例会重复触发 `check` / `expire`。 - Redis 是任务运行态的唯一存储，应按恢复目标配置持久化、高可用和备份。 - 当前是单节点 ioredis 客户端；使用 Redis Cluster 前应改为 Cluster 客户端，并给同一队列的 key 添加一致 hash tag。 diff --git a/admin-dashboard/.env.example b/admin-dashboard/.env.example index 07f9f50..ecde2ba 100644 --- a/admin-dashboard/.env.example +++ b/admin-dashboard/.env.example @@ -1,2 +1,6 @@ # Backend origin used by the Next.js same-origin proxy. WAITQUEUE_API_URL=http://127.0.0.1:3000 +# Optional backend Bearer token. This is read only by the Next.js server. +WAITQUEUE_API_TOKEN= +# Comma-separated exact dashboard hostnames. Ports in Host headers are ignored. +DASHBOARD_ALLOWED_HOSTS=127.0.0.1,localhost,[::1] diff --git a/admin-dashboard/README.md b/admin-dashboard/README.md index c27ca4b..e7da705 100644 --- a/admin-dashboard/README.md +++ b/admin-dashboard/README.md @@ -48,8 +48,12 @@ corepack pnpm --dir admin-dashboard dev ```dotenv WAITQUEUE_API_URL=http://127.0.0.1:3000 +WAITQUEUE_API_TOKEN= +DASHBOARD_ALLOWED_HOSTS=127.0.0.1,localhost,[::1] ``` +后端开启 `WAITQUEUE_API_TOKEN` 时，这里必须填写同一值。该值是服务端共享凭据，不要改名为 `NEXT_PUBLIC_*`。`DASHBOARD_ALLOWED_HOSTS` 是无通配符的精确 hostname 列表，比较时忽略端口；使用共享域名或反向代理时必须加入外部 hostname，并让代理保留/改写为该 Host。 + ## 生产构建 ```bash @@ -58,31 +62,32 @@ corepack pnpm --dir admin-dashboard build corepack pnpm --dir admin-dashboard start ``` -`dev` 和 `start` 都固定监听 3001，避免与后端默认的 3000 冲突。构建和启动时应提供同一个 `WAITQUEUE_API_URL`。 +`dev` 和 `start` 都固定监听 3001，避免与后端默认的 3000 冲突。`WAITQUEUE_API_URL`、`WAITQUEUE_API_TOKEN` 和 `DASHBOARD_ALLOWED_HOSTS` 在服务启动后按请求读取，无需作为 Docker build argument；同一份构建产物可在不同环境复用。 ### Docker Compose -仓库根目录的 Compose 会把控制台构建为 Next.js standalone 镜像，并在构建阶段将 API 代理固定到容器网络中的 `http://api:3000`： +仓库根目录的 Compose 会把控制台构建为 Next.js standalone 镜像，并在运行时将 API 代理指向容器网络中的 `http://api:3000`： ```bash docker compose up --build --detach --wait ``` -启动完成后访问 [http://127.0.0.1:3001](http://127.0.0.1:3001)。这里必须在构建阶段提供 `WAITQUEUE_API_URL`，因为 Next.js 会把 rewrite 写进构建产物；只在容器启动时修改该变量不足以改变已生成的代理规则。控制台镜像以非 root 用户和只读文件系统运行。 +启动完成后访问 [http://127.0.0.1:3001](http://127.0.0.1:3001)。Compose 将同一个 `WAITQUEUE_API_TOKEN` 仅注入 API 与控制台运行时，不会写入镜像构建层。控制台镜像以非 root 用户和只读文件系统运行。 ## 数据链路 ```text Browser └─ /waitqueue/*（同源） - └─ Next.js rewrite - └─ WAITQUEUE_API_URL - ├─ GET /waitqueue/admin/overview - ├─ POST /waitqueue/queue/newQueue - └─ POST /waitqueue/scheduler/addTask + └─ Next.js internal rewrite + └─ /api/waitqueue/*（服务端白名单代理） + └─ WAITQUEUE_API_URL + 服务端 Bearer token + ├─ GET /waitqueue/admin/overview + ├─ POST /waitqueue/queue/newQueue + └─ POST /waitqueue/scheduler/addTask ``` -`WAITQUEUE_API_URL` 只由 Next.js 服务端读取，不会进入浏览器 bundle。后端无需开启 CORS。 +代理先用 `DASHBOARD_ALLOWED_HOSTS` 精确校验请求 Host，再只接受上图三组 method/path；POST 只接受 JSON 且限制为 32 KiB。它不转发浏览器传入的 Authorization、Cookie、Host 或转发头，禁止上游重定向，只复制必要的响应头。服务端变量不会进入浏览器 bundle；后端无需开启 CORS。 ## 技术与目录 @@ -92,6 +97,7 @@ Browser admin-dashboard/ ├── src/pages/_app.tsx # 全局样式与页面入口 ├── src/pages/index.tsx # 数据读取、交互与控制室页面 +├── src/pages/api/waitqueue/ # 运行时白名单代理与 token 注入 ├── src/style/global.css # 设计 token、主题与基础样式 ├── src/style/dashboard.module.css # 工作台布局、状态组件和响应式样式 ├── next.config.js # API 同源代理 @@ -102,11 +108,12 @@ HTTP 使用浏览器原生 `fetch`，页面状态使用 React hooks；没有 Red ## 安全边界 -控制台没有独立登录页，后端 API 也尚未内置鉴权。部署时必须： +控制台的服务端代理可以隐藏后端共享 token，但它不是用户身份认证。任何能访问控制台的人都能借代理读取或修改队列。部署时必须： - 将控制台和 API 放到可信网络或认证网关之后； -- 对写接口增加认证、授权和审计； -- 限制可注册的 `hookUrl` 主机与网段，防止 SSRF； +- 后端与控制台配置同一个高强度 `WAITQUEUE_API_TOKEN`； +- 将控制台外部 hostname 精确加入 `DASHBOARD_ALLOWED_HOSTS`； +- 后端配置精确 `HOOK_URL_ALLOWLIST`，并按需配置出站网络策略； - 不把管理端直接暴露到公网。 完整启动流程、API 和回调协议见仓库根目录 [README.md](../README.md)。 diff --git a/admin-dashboard/next.config.js b/admin-dashboard/next.config.js index 2704dbe..1d7eef9 100644 --- a/admin-dashboard/next.config.js +++ b/admin-dashboard/next.config.js @@ -1,6 +1,4 @@ /** @type {import('next').NextConfig} */ -const apiOrigin = (process.env.WAITQUEUE_API_URL || 'http://127.0.0.1:3000').replace(/\/$/, ''); - module.exports = { output: 'standalone', reactStrictMode: true, @@ -11,7 +9,7 @@ module.exports = { return [ { source: '/waitqueue/:path*', - destination: `${apiOrigin}/waitqueue/:path*`, + destination: '/api/waitqueue/:path*', }, ]; }, diff --git a/admin-dashboard/src/pages/api/waitqueue/[...path].ts b/admin-dashboard/src/pages/api/waitqueue/[...path].ts new file mode 100644 index 0000000..1f9e3f7 --- /dev/null +++ b/admin-dashboard/src/pages/api/waitqueue/[...path].ts @@ -0,0 +1,138 @@ +import type { NextApiRequest, NextApiResponse } from 'next'; + +const DEFAULT_API_ORIGIN = 'http://127.0.0.1:3000'; +const DEFAULT_ALLOWED_HOSTS = '127.0.0.1,localhost,[::1]'; +const PROXY_TIMEOUT_MS = 10_000; +const JSON_CONTENT_TYPE = 'application/json'; + +const ALLOWED_ROUTES = new Map([ + ['admin/overview', 'GET'], + ['queue/newQueue', 'POST'], + ['scheduler/addTask', 'POST'], +]); + +type ErrorEnvelope = { + code: 1; + msg: string; + data: never[]; +}; + +export const config = { + api: { + bodyParser: { + sizeLimit: '32kb', + }, + }, +}; + +function errorEnvelope(msg: string): ErrorEnvelope { + return { code: 1, msg, data: [] }; +} + +function normalizedHostname(authority: string): string { + if (!authority || /[\\/\s]/.test(authority)) throw new Error('invalid host authority'); + const parsed = new URL(`http://${authority}`); + if (parsed.username || parsed.password || parsed.pathname !== '/' || parsed.search || parsed.hash) { + throw new Error('invalid host authority'); + } + return parsed.hostname.toLowerCase().replace(/^\[|\]$/g, '').replace(/\.+$/, ''); +} + +function requestHostAllowed(request: NextApiRequest): boolean { + try { + const configured = process.env.DASHBOARD_ALLOWED_HOSTS || DEFAULT_ALLOWED_HOSTS; + const allowedHosts = configured.split(',').map((value) => normalizedHostname(value.trim())); + return allowedHosts.includes(normalizedHostname(request.headers.host || '')); + } catch { + return false; + } +} + +function upstreamOrigin(): string { + const raw = process.env.WAITQUEUE_API_URL || DEFAULT_API_ORIGIN; + const parsed = new URL(raw); + if ( + (parsed.protocol !== 'http:' && parsed.protocol !== 'https:') || + parsed.username || + parsed.password || + parsed.pathname !== '/' || + parsed.search || + parsed.hash + ) { + throw new Error('WAITQUEUE_API_URL must be an HTTP(S) origin without credentials or a path'); + } + return parsed.origin; +} + +function routePath(request: NextApiRequest): string | undefined { + const segments = request.query.path; + if (!Array.isArray(segments) || segments.some((segment) => !segment)) return undefined; + return segments.join('/'); +} + +function copyResponseHeader(response: Response, target: NextApiResponse, name: string): void { + const value = response.headers.get(name); + if (value) target.setHeader(name, value); +} + +export default async function handler(request: NextApiRequest, response: NextApiResponse): Promise { + if (!requestHostAllowed(request)) { + response.status(403).json(errorEnvelope('host not allowed')); + return; + } + + const path = routePath(request); + const allowedMethod = path ? ALLOWED_ROUTES.get(path) : undefined; + if (!path || !allowedMethod) { + response.status(404).json(errorEnvelope('route not found')); + return; + } + + if (request.method !== allowedMethod) { + response.setHeader('Allow', allowedMethod); + response.status(405).json(errorEnvelope('method not allowed')); + return; + } + + if (allowedMethod === 'POST') { + const contentType = request.headers['content-type'] || ''; + const mediaType = contentType.split(';', 1)[0].trim().toLowerCase(); + if (mediaType !== JSON_CONTENT_TYPE) { + response.status(415).json(errorEnvelope('content type must be application/json')); + return; + } + } + + const controller = new AbortController(); + const timeout = setTimeout(() => controller.abort(), PROXY_TIMEOUT_MS); + try { + const headers: Record = { accept: JSON_CONTENT_TYPE }; + if (allowedMethod === 'POST') headers['content-type'] = JSON_CONTENT_TYPE; + + const token = process.env.WAITQUEUE_API_TOKEN?.trim(); + if (token) headers.authorization = `Bearer ${token}`; + + const upstream = await fetch(`${upstreamOrigin()}/waitqueue/${path}`, { + method: allowedMethod, + headers, + body: allowedMethod === 'POST' ? JSON.stringify(request.body) : undefined, + redirect: 'manual', + signal: controller.signal, + }); + if (upstream.status >= 300 && upstream.status < 400) { + throw new Error('unexpected upstream redirect'); + } + + for (const header of ['content-type', 'cache-control', 'retry-after']) { + copyResponseHeader(upstream, response, header); + } + const body = await upstream.text(); + response.status(upstream.status); + if (body) response.send(body); + else response.end(); + } catch { + response.status(502).json(errorEnvelope('upstream service unavailable')); + } finally { + clearTimeout(timeout); + } +} diff --git a/compose.yaml b/compose.yaml index 12e8f97..a07b6fe 100644 --- a/compose.yaml +++ b/compose.yaml @@ -73,6 +73,12 @@ services: environment: <<: *api-environment APP_PORT: 3000 + WAITQUEUE_API_TOKEN: ${WAITQUEUE_API_TOKEN:-} + HOOK_URL_ALLOWLIST: ${HOOK_URL_ALLOWLIST:-} + HOOK_URL_ALLOW_PRIVATE: ${HOOK_URL_ALLOW_PRIVATE:-false} + REQUEST_BODY_LIMIT_BYTES: ${REQUEST_BODY_LIMIT_BYTES:-32768} + RATE_LIMIT_MAX_REQUESTS: ${RATE_LIMIT_MAX_REQUESTS:-0} + RATE_LIMIT_WINDOW_MS: ${RATE_LIMIT_WINDOW_MS:-60000} ports: - "${BIND_ADDRESS:-127.0.0.1}:${API_PORT:-3000}:3000" depends_on: @@ -105,11 +111,11 @@ services: context: . dockerfile: Dockerfile target: dashboard - args: - WAITQUEUE_API_URL: http://api:3000 restart: unless-stopped environment: WAITQUEUE_API_URL: http://api:3000 + WAITQUEUE_API_TOKEN: ${WAITQUEUE_API_TOKEN:-} + DASHBOARD_ALLOWED_HOSTS: ${DASHBOARD_ALLOWED_HOSTS:-127.0.0.1,localhost,[::1]} PORT: 3001 HOSTNAME: 0.0.0.0 ports: diff --git a/wait-queue/.env.example b/wait-queue/.env.example index 27679d3..1a435ba 100644 --- a/wait-queue/.env.example +++ b/wait-queue/.env.example @@ -3,6 +3,16 @@ HOOK_TIMEOUT_MS=10000 CHECK_TASK_DIFF_CRON=0 * * * * * CRON_TIMEZONE=Asia/Shanghai +# Leave empty only for trusted local development. Set a strong value before exposing the API. +WAITQUEUE_API_TOKEN= +# Comma-separated exact public HTTP(S) origins, for example: https://worker.example.com +HOOK_URL_ALLOWLIST= +# Development-only escape hatch for an explicitly allowlisted local callback. +HOOK_URL_ALLOW_PRIVATE=false +REQUEST_BODY_LIMIT_BYTES=32768 +RATE_LIMIT_MAX_REQUESTS=0 +RATE_LIMIT_WINDOW_MS=60000 + DB_HOST=127.0.0.1 DB_PORT=3306 DB_DATABASE=waitqueue diff --git a/wait-queue/src/app.ts b/wait-queue/src/app.ts index 6ba5288..ccb5bbf 100644 --- a/wait-queue/src/app.ts +++ b/wait-queue/src/app.ts @@ -3,29 +3,73 @@ import Router from '@koa/router' import koaPino from 'koa-pino-logger' import bodyParser from 'koa-bodyparser' import { CronJob } from 'cron' -import { queueRoutes } from './routes/queue' -import { schedulerRoutes } from './routes/scheduler' +import { createQueueRoutes } from './routes/queue' +import { createSchedulerRoutes } from './routes/scheduler' import { adminRoutes } from './routes/admin' import { errorHandler } from './middleware/error_handler' +import { auditMiddleware, createBearerAuth, createRateLimit } from './middleware/security' import response from './utils/response' import { HttpError } from './utils/http_error' import { env } from './conf/env' import { daoMysql } from './conf/db' import { redisCli } from './conf/redis' import { Timer } from './lib/timer' -import { createBackgroundContext, logger } from './common/logger' +import { + createBackgroundContext, + LOGGER_REDACT_PATHS, + logger, + safeErrorSerializer, + safeRequestSerializer, +} from './common/logger' import { createReadinessCheck, ReadinessCheck, ReadinessResult } from './service/readiness' +import { + createSecurityConfigurationWarner, + createSecurityConfig, + SecurityConfigInput, +} from './security/config' +import { FixedWindowRateLimiter } from './security/rate_limit' +import { HookUrlPolicy } from './security/hook_url_policy' + +const warnSecurityConfiguration = createSecurityConfigurationWarner(logger) export interface CreateAppOptions { readinessCheck?: ReadinessCheck + security?: SecurityConfigInput + rateLimitClock?: () => number + requestLogStream?: NodeJS.WritableStream } export function createApp(options: CreateAppOptions = {}): Koa { const app = new Koa() const readinessCheck = options.readinessCheck ?? createReadinessCheck() - app.use(koaPino()) + const security = createSecurityConfig(options.security ?? env.security) + const hookUrlPolicy = new HookUrlPolicy(security.hookUrlAllowlist, { + allowPrivate: security.allowPrivateHookUrls, + }) + const rateLimiter = new FixedWindowRateLimiter( + security.rateLimitMaxRequests, + security.rateLimitWindowMs, + options.rateLimitClock + ) + app.use( + koaPino( + { + redact: { paths: [...LOGGER_REDACT_PATHS], censor: '[REDACTED]' }, + serializers: { err: safeErrorSerializer, req: safeRequestSerializer }, + }, + options.requestLogStream as any + ) + ) + app.use(auditMiddleware) app.use(errorHandler) - app.use(bodyParser()) + app.use(createRateLimit(rateLimiter)) + app.use(createBearerAuth(security.apiToken)) + app.use( + bodyParser({ + enableTypes: ['json'], + jsonLimit: `${security.requestBodyLimitBytes}b`, + }) + ) app.use(async (ctx, next) => { await next() if (ctx.status === 404 && !ctx.body) { @@ -34,7 +78,7 @@ export function createApp(options: CreateAppOptions = {}): Koa { } }) - const router = new Router({ prefix: '/waitqueue' }) + const router = new Router({ prefix: '/waitqueue', sensitive: true }) router.get('/health', (ctx) => response.success(ctx, { status: 'ok' })) router.get('/ready', async (ctx) => { ctx.set('Cache-Control', 'no-store') @@ -59,8 +103,8 @@ export function createApp(options: CreateAppOptions = {}): Koa { response.success(ctx, { status: 'ready', dependencies: readiness.dependencies }) }) router.use('/admin', adminRoutes.routes()) - router.use('/scheduler', schedulerRoutes.routes()) - router.use('/queue', queueRoutes.routes()) + router.use('/scheduler', createSchedulerRoutes(hookUrlPolicy).routes()) + router.use('/queue', createQueueRoutes(hookUrlPolicy).routes()) app.use(router.routes()) app.use( @@ -74,6 +118,7 @@ export function createApp(options: CreateAppOptions = {}): Koa { } export async function start() { + warnSecurityConfiguration(env.security) const database = daoMysql.getInstance() const redis = redisCli.getInstance() let timer: Timer | undefined diff --git a/wait-queue/src/common/logger.ts b/wait-queue/src/common/logger.ts index ee5de5c..966db57 100644 --- a/wait-queue/src/common/logger.ts +++ b/wait-queue/src/common/logger.ts @@ -1,7 +1,90 @@ import { Context } from 'koa' import createLogger from 'pino' -export const logger = createLogger({ name: 'waitqueue' }) +export const LOGGER_REDACT_PATHS = Object.freeze([ + 'req.headers.authorization', + 'req.headers.cookie', + 'request.headers.authorization', + 'request.headers.cookie', + 'headers.authorization', + 'headers.cookie', + 'authorization', + 'cookie', + 'apiToken', + 'token', + 'password', + 'hookUrl', + 'taskId', + 'taskIds', + 'context.apiToken', + 'context.token', + 'context.password', + 'context.url', + 'context.hookUrl', + 'context.taskId', + 'context.taskIds', + '*.headers.authorization', + '*.headers.cookie', + '*.apiToken', + '*.token', + '*.password', + 'database.password', + 'redis.password', + 'err.config.password', + 'err.options.password', + '*.hookUrl', + '*.taskId', + '*.taskIds', +]) + +function safeErrorIdentifier(value: unknown, fallback: string): string { + return typeof value === 'string' && /^[A-Za-z0-9_.-]{1,64}$/.test(value) ? value : fallback +} + +const SAFE_REQUEST_PATHS = new Set([ + '/waitqueue/health', + '/waitqueue/ready', + '/waitqueue/admin/overview', + '/waitqueue/queue/newQueue', + '/waitqueue/scheduler/addTask', +]) + +export function safeLogPath(value: unknown): string { + if (typeof value !== 'string') return '/[invalid]' + let pathname: string + try { + pathname = new URL(value, 'http://request.invalid').pathname + } catch { + return '/[invalid]' + } + if (SAFE_REQUEST_PATHS.has(pathname)) return pathname + if (pathname === '/waitqueue' || pathname.startsWith('/waitqueue/')) return '/waitqueue/[unmatched]' + return '/[unmatched]' +} + +export function safeRequestSerializer(request: any): { id?: string | number; method: string; path: string } { + const serialized: { id?: string | number; method: string; path: string } = { + method: safeErrorIdentifier(request?.method, 'UNKNOWN'), + path: safeLogPath(request?.url ?? request?.path), + } + if (typeof request?.id === 'string' || typeof request?.id === 'number') serialized.id = request.id + return serialized +} + +export function safeErrorSerializer(error: any): { type: string; code?: string } { + const serialized: { type: string; code?: string } = { + type: safeErrorIdentifier(error?.type ?? error?.name ?? error?.constructor?.name, 'Error'), + } + const code = safeErrorIdentifier(error?.code, '') + if (code) serialized.code = code + return serialized +} + +export const logger = createLogger({ + name: 'waitqueue', + redact: { paths: [...LOGGER_REDACT_PATHS], censor: '[REDACTED]' }, + serializers: { err: safeErrorSerializer, req: safeRequestSerializer }, +}) export function createBackgroundContext(): Context { return { diff --git a/wait-queue/src/conf/env.ts b/wait-queue/src/conf/env.ts index d27eb73..cef686a 100644 --- a/wait-queue/src/conf/env.ts +++ b/wait-queue/src/conf/env.ts @@ -1,4 +1,5 @@ import 'dotenv/config' +import { readSecurityConfig } from '../security/config' function readPositiveInteger(name: string, fallback: number): number { const raw = process.env[name] @@ -28,4 +29,5 @@ export const env = Object.freeze({ port: readPositiveInteger('REDIS_PORT', 6379), password: process.env.REDIS_PASSWORD || undefined, }), + security: readSecurityConfig(), }) diff --git a/wait-queue/src/lib/task_manager.ts b/wait-queue/src/lib/task_manager.ts index 8652e02..6474075 100644 --- a/wait-queue/src/lib/task_manager.ts +++ b/wait-queue/src/lib/task_manager.ts @@ -4,6 +4,8 @@ import { Redis } from 'ioredis' import { Service } from './service' import { TaskRun } from './task_run' import { redisCli } from '../conf/redis' +import { env } from '../conf/env' +import { HookUrlPolicy } from '../security/hook_url_policy' const CLAIM_TASKS_SCRIPT = ` local maxRunning = tonumber(ARGV[1]) @@ -69,7 +71,8 @@ interface TaskClaim { } export class TaskManager extends Service { - private url: string + private queueId: number + private namespace: string private runningKey: string // 正在执行的任务 private waitingKey: string // 等待执行的任务 private taskRunningCount: number // 并发执行的任务数 @@ -82,23 +85,27 @@ export class TaskManager extends Service { url: string, runningKey: string, waitingKey: string, - taskRunningCount: number + taskRunningCount: number, + hookUrlPolicy: HookUrlPolicy = new HookUrlPolicy(env.security.hookUrlAllowlist, { + allowPrivate: env.security.allowPrivateHookUrls, + }) ) { super(ctx) - this.url = url + this.queueId = queueId + this.namespace = namespace this.runningKey = runningKey this.waitingKey = waitingKey this.taskRunningCount = taskRunningCount - this.taskRunInstance = new TaskRun(this.ctx, this.url, queueId, namespace) + this.taskRunInstance = new TaskRun(this.ctx, url, queueId, namespace, hookUrlPolicy) this.redis = redisCli.getInstance() } private async dispatchTask({ taskId, claimToken }: TaskClaim): Promise { try { await this.taskRunInstance.run(taskId) - this.selfLog('task trigger success', taskId) + this.selfLog('task trigger succeeded') } catch (error: any) { - this.baseLogError(`task trigger failed: ${taskId}`, error) + this.baseLogError('task trigger failed', error) try { const requeued = await this.redis.eval( REQUEUE_TASK_SCRIPT, @@ -109,11 +116,10 @@ export class TaskManager extends Service { claimToken ) this.selfLog( - requeued === 1 ? 'task trigger failed; task returned to waiting queue' : 'task trigger failed; stale claim ignored', - taskId + requeued === 1 ? 'task trigger failed; task returned to waiting queue' : 'task trigger failed; stale claim ignored' ) } catch (redisError) { - this.baseLogError(`failed to return task to waiting queue: ${taskId}`, redisError) + this.baseLogError('failed to return task to waiting queue', redisError) } } } @@ -163,7 +169,7 @@ export class TaskManager extends Service { this.selfLog(`runTask: claimed task count: ${claims.length}`) await Promise.all( claims.map((claim) => { - this.selfLog('runTask: prepare exec task', claim.taskId) + this.selfLog('runTask: prepare exec task') return this.dispatchTask(claim) }) ) @@ -181,7 +187,7 @@ export class TaskManager extends Service { this.selfLog('CheckStatus: check task status start') const taskMap = await this.redis.hgetall(this.runningKey) const taskIds = Object.keys(taskMap) - this.selfLog('CheckStatus: 正在执行中的任务 id ', taskIds.join(',')) + this.selfLog(`CheckStatus: running task count: ${taskIds.length}`) const completeIds = await this.taskRunInstance.checkTaskStatus( taskIds.filter((item) => { return item !== '' @@ -189,7 +195,7 @@ export class TaskManager extends Service { ) const releasedIds = await this.releaseTasks(taskMap, completeIds) if (releasedIds.length) { - this.selfLog('CheckStatus: 从 runningkey 中移除的已完成任务 id ', releasedIds.join(',')) + this.selfLog(`CheckStatus: released completed task count: ${releasedIds.length}`) } } @@ -203,16 +209,16 @@ export class TaskManager extends Service { const taskMap = await this.redis.hgetall(this.runningKey) const expiredIds = await this.taskRunInstance.expireTasks() - this.selfLog('ExpireTask: 任务实际过期任务 id 列表 ', expiredIds.join(',')) + this.selfLog(`ExpireTask: expired task count: ${expiredIds.length}`) // 获取实际已过期但是仍然在缓存执行队列中的任务 id const releasedIds = await this.releaseTasks(taskMap, expiredIds) if (releasedIds.length) { - this.selfLog('ExpireTask: 从 runningkey 中移除的过期任务 id ', releasedIds.join(',')) + this.selfLog(`ExpireTask: released expired task count: ${releasedIds.length}`) } } - selfLog(message: string, taskId?: string): void { - this.baseLogInfo(`${this.url}${'|taskId: ' + taskId}|${message}`) + selfLog(message: string): void { + this.baseLogInfo(message, { queueId: this.queueId, namespace: this.namespace }) } } diff --git a/wait-queue/src/lib/task_run.ts b/wait-queue/src/lib/task_run.ts index 399726c..c2575f4 100644 --- a/wait-queue/src/lib/task_run.ts +++ b/wait-queue/src/lib/task_run.ts @@ -1,6 +1,125 @@ import { Context } from 'koa' +import { lookup as dnsLookup } from 'node:dns/promises' +import * as http from 'node:http' +import * as https from 'node:https' +import { isIP } from 'node:net' import { Service } from './service' import { env } from '../conf/env' +import { HookUrlPolicy } from '../security/hook_url_policy' + +const MAX_CALLBACK_RESPONSE_BYTES = 1_048_576 + +export interface CallbackResponse { + status: number + body: string +} + +export interface CallbackRequestOptions { + body: string + signal: AbortSignal + hookUrlPolicy: HookUrlPolicy +} + +export type CallbackTransport = (url: URL, options: CallbackRequestOptions) => Promise +export type CallbackAddressResolver = ( + hostname: string +) => Promise + +function hostnameWithoutBrackets(hostname: string): string { + return hostname.replace(/^\[|\]$/g, '') +} + +const defaultAddressResolver: CallbackAddressResolver = async (hostname) => + dnsLookup(hostname, { all: true, verbatim: true }) + +function abortError(): Error { + const error = new Error('callback request aborted') + error.name = 'AbortError' + return error +} + +function withAbortSignal(operation: Promise, signal: AbortSignal): Promise { + if (signal.aborted) return Promise.reject(abortError()) + return new Promise((resolve, reject) => { + const onAbort = () => reject(abortError()) + signal.addEventListener('abort', onAbort, { once: true }) + operation.then( + (value) => { + signal.removeEventListener('abort', onAbort) + resolve(value) + }, + (error) => { + signal.removeEventListener('abort', onAbort) + reject(error) + } + ) + }) +} + +export async function resolvePinnedCallbackAddresses( + url: URL, + hookUrlPolicy: HookUrlPolicy, + resolveAddresses: CallbackAddressResolver = defaultAddressResolver +): Promise { + if (!hookUrlPolicy.enforcesPublicAddresses) return undefined + const hostname = hostnameWithoutBrackets(url.hostname) + if (isIP(hostname)) return undefined + const addresses = await resolveAddresses(hostname) + if (addresses.length === 0) throw new Error('callback hostname did not resolve') + addresses.forEach(({ address }) => hookUrlPolicy.assertAllowedAddress(address)) + return addresses +} + +export function createCallbackTransport( + resolveAddresses: CallbackAddressResolver = defaultAddressResolver +): CallbackTransport { + return async (url, options) => { + const pinnedAddresses = await withAbortSignal( + resolvePinnedCallbackAddresses(url, options.hookUrlPolicy, resolveAddresses), + options.signal + ) + const requestOptions: http.RequestOptions = { + method: 'POST', + headers: { + accept: 'application/json', + 'content-type': 'application/json', + 'content-length': Buffer.byteLength(options.body), + }, + agent: false, + signal: options.signal, + } + if (pinnedAddresses) { + requestOptions.lookup = ((_hostname: string, lookupOptions: { all?: boolean }, callback: Function) => { + if (lookupOptions.all) callback(null, pinnedAddresses) + else callback(null, pinnedAddresses[0].address, pinnedAddresses[0].family) + }) as any + } + + return new Promise((resolve, reject) => { + const request = (url.protocol === 'https:' ? https : http).request(url, requestOptions, (response) => { + const chunks: Buffer[] = [] + let receivedBytes = 0 + response.on('data', (chunk: Buffer | string) => { + const buffer = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk) + receivedBytes += buffer.length + if (receivedBytes > MAX_CALLBACK_RESPONSE_BYTES) { + response.destroy(new Error('callback response exceeded the size limit')) + return + } + chunks.push(buffer) + }) + response.on('error', reject) + response.on('end', () => { + resolve({ status: response.statusCode ?? 0, body: Buffer.concat(chunks).toString('utf8') }) + }) + }) + request.on('error', reject) + request.end(options.body) + }) + } +} + +export const postCallback = createCallbackTransport() export enum TASK_TYPE_CODE { run = 'run', @@ -12,12 +131,25 @@ export class TaskRun extends Service { private url: string private queueId: number private namespace: string + private hookUrlPolicy: HookUrlPolicy + private callbackTransport: CallbackTransport - constructor(ctx: Context, url: string, queueId: number, namespace: string) { + constructor( + ctx: Context, + url: string, + queueId: number, + namespace: string, + hookUrlPolicy: HookUrlPolicy = new HookUrlPolicy(env.security.hookUrlAllowlist, { + allowPrivate: env.security.allowPrivateHookUrls, + }), + callbackTransport: CallbackTransport = postCallback + ) { super(ctx) this.url = url this.queueId = queueId this.namespace = namespace + this.hookUrlPolicy = hookUrlPolicy + this.callbackTransport = callbackTransport } async run(taskId: string): Promise { @@ -57,19 +189,24 @@ export class TaskRun extends Service { const body = taskIds ? { type, queueId: this.queueId, namespace: this.namespace, taskIds } : { type, queueId: this.queueId, namespace: this.namespace } - this.baseLogInfo(`callback ${type}`, { url: this.url, taskIds }) + const callbackUrl = this.hookUrlPolicy.assertAllowed(this.url) + this.baseLogInfo('callback requested', { + callbackType: type, + queueId: this.queueId, + namespace: this.namespace, + taskCount: taskIds?.length ?? 0, + }) const abortController = new AbortController() const timeout = setTimeout(() => abortController.abort(), env.hookTimeoutMs) try { - const res = await fetch(this.url, { - method: 'POST', - headers: { 'content-type': 'application/json' }, + const res = await this.callbackTransport(callbackUrl, { body: JSON.stringify(body), signal: abortController.signal, + hookUrlPolicy: this.hookUrlPolicy, }) if (res.status !== 200) throw new Error(`${type} callback returned HTTP ${res.status}`) - const responseBody = await res.text() + const responseBody = res.body if (!responseBody) return {} try { return JSON.parse(responseBody) diff --git a/wait-queue/src/lib/timer.ts b/wait-queue/src/lib/timer.ts index c5bd08a..8959f98 100644 --- a/wait-queue/src/lib/timer.ts +++ b/wait-queue/src/lib/timer.ts @@ -6,6 +6,7 @@ import { Service } from './service' import { TaskManager } from './task_manager' import { getRunningKey, getWaitingKey } from '../common/cache' import { env } from '../conf/env' +import { HookUrlPolicy } from '../security/hook_url_policy' interface TaskJob { cron: string @@ -34,12 +35,19 @@ export class Timer extends Service { private checkTaskJobMap: Map private expireTaskJobMap: Map private queueDao: ModelCtor - constructor(ctx: Context) { + private hookUrlPolicy: HookUrlPolicy + constructor( + ctx: Context, + hookUrlPolicy: HookUrlPolicy = new HookUrlPolicy(env.security.hookUrlAllowlist, { + allowPrivate: env.security.allowPrivateHookUrls, + }) + ) { super(ctx) this.runTaskJobMap = runTaskJob this.checkTaskJobMap = checkTaskJob this.expireTaskJobMap = expireTaskJob this.queueDao = QueueDao + this.hookUrlPolicy = hookUrlPolicy } initializeQueueList(queueIds: number[] = []): Promise { @@ -85,7 +93,7 @@ export class Timer extends Service { } queueUniqKey(queueInfo: QueueAttributes) { - return JSON.stringify([queueInfo.namespace, queueInfo.url]) + return `queue:${queueInfo.id}` } syncQueueJob(queueInfo: QueueAttributes) { @@ -98,9 +106,16 @@ export class Timer extends Service { queueInfo.url, getRunningKey(queueInfo.namespace, queueInfo.id), getWaitingKey(queueInfo.namespace, queueInfo.id), - queueInfo.count + queueInfo.count, + this.hookUrlPolicy ) - const queueSignature = JSON.stringify([queueInfo.id, queueInfo.namespace, queueInfo.url, queueInfo.count]) + const queueSignature = JSON.stringify([ + queueInfo.id, + queueInfo.namespace, + queueInfo.url, + queueInfo.count, + this.hookUrlPolicy.configurationKey, + ]) const replacements: JobReplacement[] = [] ;[ { job: this.runTaskJobMap, cronTab: queueInfo.runCrontab, execFunc: taskInstance.runTask.bind(taskInstance) }, diff --git a/wait-queue/src/middleware/security.ts b/wait-queue/src/middleware/security.ts new file mode 100644 index 0000000..0deecca --- /dev/null +++ b/wait-queue/src/middleware/security.ts @@ -0,0 +1,105 @@ +import { createHash, timingSafeEqual } from 'crypto' +import { Context, Next } from 'koa' +import { HttpError } from '../utils/http_error' +import { FixedWindowRateLimiter } from '../security/rate_limit' +import { safeLogPath } from '../common/logger' + +export type AuditRejection = 'authentication' | 'rate_limit' + +const PUBLIC_CONTROL_PATHS = new Set(['/waitqueue/health', '/waitqueue/ready']) + +function isControlPath(path: string): boolean { + return path === '/waitqueue' || path.startsWith('/waitqueue/') +} + +function isSecurityExempt(ctx: Context): boolean { + if (ctx.method === 'OPTIONS') return true + const normalizedPath = ctx.path.toLowerCase() + return !isControlPath(normalizedPath) || PUBLIC_CONTROL_PATHS.has(normalizedPath) +} + +function bearerToken(header: string): string | undefined { + const match = /^Bearer ([^\s]+)$/i.exec(header) + return match?.[1] +} + +function tokensMatch(expectedDigest: Buffer, actual: string | undefined): boolean { + const actualDigest = createHash('sha256').update(actual ?? '').digest() + return timingSafeEqual(expectedDigest, actualDigest) +} + +export function createBearerAuth(apiToken: string | undefined) { + const expectedDigest = apiToken ? createHash('sha256').update(apiToken).digest() : undefined + return async (ctx: Context, next: Next): Promise => { + if (!expectedDigest || isSecurityExempt(ctx)) { + await next() + return + } + + if (!tokensMatch(expectedDigest, bearerToken(ctx.get('authorization')))) { + ctx.state.auditRejection = 'authentication' satisfies AuditRejection + ctx.set('WWW-Authenticate', 'Bearer') + throw new HttpError(401, 'authentication required') + } + await next() + } +} + +function clientKey(ctx: Context): string { + return ctx.ip || ctx.req.socket.remoteAddress || 'unknown' +} + +export function createRateLimit(limiter: FixedWindowRateLimiter) { + return async (ctx: Context, next: Next): Promise => { + if (isSecurityExempt(ctx)) { + await next() + return + } + + const decision = limiter.consume(clientKey(ctx)) + if (!decision.allowed) { + ctx.state.auditRejection = 'rate_limit' satisfies AuditRejection + ctx.set('Retry-After', String(decision.retryAfterSeconds)) + throw new HttpError(429, 'too many requests') + } + await next() + } +} + +const WRITE_METHODS = new Set(['POST', 'PUT', 'PATCH', 'DELETE']) + +function successfulWriteAction(ctx: Context): string { + if (ctx.method === 'POST' && ctx.path === '/waitqueue/queue/newQueue') return 'queue.configure' + if (ctx.method === 'POST' && ctx.path === '/waitqueue/scheduler/addTask') return 'task.enqueue' + return 'api.write' +} + +export async function auditMiddleware(ctx: Context, next: Next): Promise { + const startedAt = Date.now() + try { + await next() + } finally { + const rejection = ctx.state.auditRejection as AuditRejection | undefined + const writeRequest = WRITE_METHODS.has(ctx.method) + const successfulWrite = writeRequest && ctx.status >= 200 && ctx.status < 400 + if (!rejection && !writeRequest) return + + const audit = { + event: 'api_audit', + action: + rejection === 'authentication' + ? 'auth.denied' + : rejection === 'rate_limit' + ? 'rate_limit.denied' + : successfulWriteAction(ctx), + outcome: rejection ? 'denied' : successfulWrite ? 'succeeded' : 'failed', + method: ctx.method, + path: safeLogPath(ctx.path), + statusCode: ctx.status, + durationMs: Math.max(0, Date.now() - startedAt), + } + if (rejection) ctx.log.warn({ audit }, 'api request rejected') + else if (successfulWrite) ctx.log.info({ audit }, 'api write succeeded') + else ctx.log.warn({ audit }, 'api write failed') + } +} diff --git a/wait-queue/src/routes/admin.ts b/wait-queue/src/routes/admin.ts index 9179233..ff27da9 100644 --- a/wait-queue/src/routes/admin.ts +++ b/wait-queue/src/routes/admin.ts @@ -2,7 +2,7 @@ import Router from '@koa/router' import { AdminService } from '../service/admin' import response from '../utils/response' -const adminRoutes = new Router() +const adminRoutes = new Router({ sensitive: true }) adminRoutes.get('/overview', async (ctx) => { ctx.set('Cache-Control', 'no-store') diff --git a/wait-queue/src/routes/queue.ts b/wait-queue/src/routes/queue.ts index 0b40e9d..047862e 100644 --- a/wait-queue/src/routes/queue.ts +++ b/wait-queue/src/routes/queue.ts @@ -2,13 +2,17 @@ import Router from '@koa/router' import { QueueService } from '../service/queue' import response from '../utils/response' import { validateNewQueueInput } from '../utils/validation' +import { HookUrlPolicy, permissiveHookUrlPolicy } from '../security/hook_url_policy' -const queueRoutes = new Router() +export function createQueueRoutes(hookUrlPolicy: HookUrlPolicy = permissiveHookUrlPolicy): Router { + const queueRoutes = new Router({ sensitive: true }) -queueRoutes.post('/newQueue', async (ctx) => { - const input = validateNewQueueInput(ctx.request.body) - const result = await new QueueService(ctx).newQueue(input) - response.success(ctx, result) -}) + queueRoutes.post('/newQueue', async (ctx) => { + const input = validateNewQueueInput(ctx.request.body, hookUrlPolicy) + const result = await new QueueService(ctx, hookUrlPolicy).newQueue(input) + response.success(ctx, result) + }) + return queueRoutes +} -export { queueRoutes } +export const queueRoutes = createQueueRoutes() diff --git a/wait-queue/src/routes/scheduler.ts b/wait-queue/src/routes/scheduler.ts index 7311a4c..7e12ac2 100644 --- a/wait-queue/src/routes/scheduler.ts +++ b/wait-queue/src/routes/scheduler.ts @@ -2,13 +2,17 @@ import Router from '@koa/router' import response from '../utils/response' import { SchedulerService } from '../service/scheduler' import { validateAddTaskInput } from '../utils/validation' +import { HookUrlPolicy, permissiveHookUrlPolicy } from '../security/hook_url_policy' -const schedulerRoutes = new Router() +export function createSchedulerRoutes(hookUrlPolicy: HookUrlPolicy = permissiveHookUrlPolicy): Router { + const schedulerRoutes = new Router({ sensitive: true }) -schedulerRoutes.post('/addTask', async (ctx) => { - const input = validateAddTaskInput(ctx.request.body) - const result = await new SchedulerService(ctx).addTask(input) - response.success(ctx, result) -}) + schedulerRoutes.post('/addTask', async (ctx) => { + const input = validateAddTaskInput(ctx.request.body, hookUrlPolicy) + const result = await new SchedulerService(ctx).addTask(input) + response.success(ctx, result) + }) + return schedulerRoutes +} -export { schedulerRoutes } +export const schedulerRoutes = createSchedulerRoutes() diff --git a/wait-queue/src/security/config.ts b/wait-queue/src/security/config.ts new file mode 100644 index 0000000..46a376d --- /dev/null +++ b/wait-queue/src/security/config.ts @@ -0,0 +1,164 @@ +import { normalizeAllowedOrigins } from './hook_url_policy' + +export interface SecurityConfig { + apiToken?: string + hookUrlAllowlist: readonly string[] + allowPrivateHookUrls: boolean + requestBodyLimitBytes: number + rateLimitMaxRequests: number + rateLimitWindowMs: number +} + +export type SecurityConfigInput = Partial + +export const DEFAULT_SECURITY_CONFIG: SecurityConfig = Object.freeze({ + apiToken: undefined, + hookUrlAllowlist: Object.freeze([]), + allowPrivateHookUrls: false, + requestBodyLimitBytes: 32_768, + rateLimitMaxRequests: 0, + rateLimitWindowMs: 60_000, +}) + +function assertInteger(name: string, value: number, minimum: number): number { + if (!Number.isInteger(value) || value < minimum) { + throw new Error(`${name} must be an integer greater than or equal to ${minimum}`) + } + return value +} + +function assertBoolean(name: string, value: boolean): boolean { + if (typeof value !== 'boolean') throw new Error(`${name} must be a boolean`) + return value +} + +function normalizeApiToken(value: string | undefined): string | undefined { + if (value === undefined || value === '') return undefined + if (value.trim() === '') throw new Error('WAITQUEUE_API_TOKEN must not contain only whitespace') + if (/\s/.test(value)) throw new Error('WAITQUEUE_API_TOKEN must not contain whitespace') + return value +} + +export function createSecurityConfig(input: SecurityConfigInput = {}): SecurityConfig { + const config = { + apiToken: normalizeApiToken(input.apiToken ?? DEFAULT_SECURITY_CONFIG.apiToken), + hookUrlAllowlist: normalizeAllowedOrigins(input.hookUrlAllowlist ?? DEFAULT_SECURITY_CONFIG.hookUrlAllowlist), + allowPrivateHookUrls: assertBoolean( + 'HOOK_URL_ALLOW_PRIVATE', + input.allowPrivateHookUrls ?? DEFAULT_SECURITY_CONFIG.allowPrivateHookUrls + ), + requestBodyLimitBytes: assertInteger( + 'REQUEST_BODY_LIMIT_BYTES', + input.requestBodyLimitBytes ?? DEFAULT_SECURITY_CONFIG.requestBodyLimitBytes, + 1 + ), + rateLimitMaxRequests: assertInteger( + 'RATE_LIMIT_MAX_REQUESTS', + input.rateLimitMaxRequests ?? DEFAULT_SECURITY_CONFIG.rateLimitMaxRequests, + 0 + ), + rateLimitWindowMs: assertInteger( + 'RATE_LIMIT_WINDOW_MS', + input.rateLimitWindowMs ?? DEFAULT_SECURITY_CONFIG.rateLimitWindowMs, + 1 + ), + } + return Object.freeze({ ...config, hookUrlAllowlist: Object.freeze([...config.hookUrlAllowlist]) }) +} + +function readInteger( + environment: NodeJS.ProcessEnv, + name: string, + fallback: number, + minimum: number +): number { + const raw = environment[name] + if (raw === undefined || raw === '') return fallback + const value = Number(raw) + return assertInteger(name, value, minimum) +} + +function readAllowedOrigins(environment: NodeJS.ProcessEnv): readonly string[] { + const raw = environment.HOOK_URL_ALLOWLIST + if (raw === undefined || raw.trim() === '') return [] + const values = raw.split(',').map((value) => value.trim()) + if (values.some((value) => value === '')) { + throw new Error('HOOK_URL_ALLOWLIST must be a comma-separated list of non-empty origins') + } + try { + return normalizeAllowedOrigins(values) + } catch (error) { + throw new Error(`HOOK_URL_ALLOWLIST is invalid: ${(error as Error).message}`) + } +} + +function readBoolean(environment: NodeJS.ProcessEnv, name: string, fallback: boolean): boolean { + const raw = environment[name] + if (raw === undefined || raw === '') return fallback + if (raw === 'true') return true + if (raw === 'false') return false + throw new Error(`${name} must be either true or false`) +} + +export function readSecurityConfig(environment: NodeJS.ProcessEnv = process.env): SecurityConfig { + return createSecurityConfig({ + apiToken: environment.WAITQUEUE_API_TOKEN, + hookUrlAllowlist: readAllowedOrigins(environment), + allowPrivateHookUrls: readBoolean( + environment, + 'HOOK_URL_ALLOW_PRIVATE', + DEFAULT_SECURITY_CONFIG.allowPrivateHookUrls + ), + requestBodyLimitBytes: readInteger( + environment, + 'REQUEST_BODY_LIMIT_BYTES', + DEFAULT_SECURITY_CONFIG.requestBodyLimitBytes, + 1 + ), + rateLimitMaxRequests: readInteger( + environment, + 'RATE_LIMIT_MAX_REQUESTS', + DEFAULT_SECURITY_CONFIG.rateLimitMaxRequests, + 0 + ), + rateLimitWindowMs: readInteger( + environment, + 'RATE_LIMIT_WINDOW_MS', + DEFAULT_SECURITY_CONFIG.rateLimitWindowMs, + 1 + ), + }) +} + +interface WarningLogger { + warn(bindings: Record, message: string): unknown +} + +export function createSecurityConfigurationWarner(log: WarningLogger): (config: SecurityConfig) => void { + let warnedAboutAuthentication = false + let warnedAboutHookPolicy = false + let warnedAboutPrivateHooks = false + return (config: SecurityConfig) => { + if (!config.apiToken && !warnedAboutAuthentication) { + warnedAboutAuthentication = true + log.warn( + { configuration: 'WAITQUEUE_API_TOKEN' }, + 'control API authentication is disabled; configure a token before exposing the service' + ) + } + if (config.hookUrlAllowlist.length === 0 && !warnedAboutHookPolicy) { + warnedAboutHookPolicy = true + log.warn( + { configuration: 'HOOK_URL_ALLOWLIST' }, + 'callback origin allowlist is empty; configure exact origins before exposing the service' + ) + } + if (config.allowPrivateHookUrls && !warnedAboutPrivateHooks) { + warnedAboutPrivateHooks = true + log.warn( + { configuration: 'HOOK_URL_ALLOW_PRIVATE' }, + 'private and local callback targets are enabled; use this override only for isolated development' + ) + } + } +} diff --git a/wait-queue/src/security/hook_url_policy.ts b/wait-queue/src/security/hook_url_policy.ts new file mode 100644 index 0000000..8b19c46 --- /dev/null +++ b/wait-queue/src/security/hook_url_policy.ts @@ -0,0 +1,190 @@ +export class HookUrlPolicyError extends Error { + constructor(message: string) { + super(message) + this.name = 'HookUrlPolicyError' + } +} + +export interface HookUrlPolicyOptions { + allowPrivate?: boolean +} + +function normalizedHostname(hostname: string): string { + return hostname.toLowerCase().replace(/^\[|\]$/g, '').replace(/\.+$/, '') +} + +function ipv4Octets(address: string): number[] | undefined { + const parts = address.split('.') + if (parts.length !== 4 || parts.some((part) => !/^\d{1,3}$/.test(part))) return undefined + const octets = parts.map(Number) + return octets.every((octet) => octet >= 0 && octet <= 255) ? octets : undefined +} + +function isNonPublicIpv4(address: string): boolean { + const octets = ipv4Octets(address) + if (!octets) return false + const [a, b, c] = octets + return ( + a === 0 || + a === 10 || + a === 127 || + (a === 100 && b >= 64 && b <= 127) || + (a === 169 && b === 254) || + (a === 172 && b >= 16 && b <= 31) || + (a === 192 && b === 168) || + (a === 192 && b === 0 && (c === 0 || c === 2)) || + (a === 198 && (b === 18 || b === 19)) || + (a === 198 && b === 51 && c === 100) || + (a === 203 && b === 0 && c === 113) || + a >= 224 + ) +} + +function ipv6Bytes(address: string): number[] | undefined { + let normalized = address.toLowerCase() + if (normalized.includes('.')) { + const lastColon = normalized.lastIndexOf(':') + const octets = ipv4Octets(normalized.slice(lastColon + 1)) + if (lastColon < 0 || !octets) return undefined + normalized = `${normalized.slice(0, lastColon)}:${((octets[0] << 8) | octets[1]).toString(16)}:${( + (octets[2] << 8) | + octets[3] + ).toString(16)}` + } + + const halves = normalized.split('::') + if (halves.length > 2) return undefined + const left = halves[0] ? halves[0].split(':') : [] + const right = halves.length === 2 && halves[1] ? halves[1].split(':') : [] + if (halves.length === 1 && left.length !== 8) return undefined + const missing = 8 - left.length - right.length + if (missing < (halves.length === 2 ? 1 : 0)) return undefined + const groups = [...left, ...Array(missing).fill('0'), ...right] + if (groups.length !== 8 || groups.some((group) => !/^[0-9a-f]{1,4}$/.test(group))) return undefined + return groups.flatMap((group) => { + const value = Number.parseInt(group, 16) + return [value >> 8, value & 0xff] + }) +} + +function isNonPublicIpv6(address: string): boolean { + const bytes = ipv6Bytes(address) + if (!bytes) return false + const unspecified = bytes.every((byte) => byte === 0) + const loopback = bytes.slice(0, 15).every((byte) => byte === 0) && bytes[15] === 1 + const ipv4Mapped = bytes.slice(0, 10).every((byte) => byte === 0) && bytes[10] === 0xff && bytes[11] === 0xff + if (ipv4Mapped) return isNonPublicIpv4(bytes.slice(12).join('.')) + const ipv4Compatible = bytes.slice(0, 12).every((byte) => byte === 0) + if (ipv4Compatible) return isNonPublicIpv4(bytes.slice(12).join('.')) + const wellKnownNat64 = + bytes[0] === 0x00 && + bytes[1] === 0x64 && + bytes[2] === 0xff && + bytes[3] === 0x9b && + bytes.slice(4, 12).every((byte) => byte === 0) + if (wellKnownNat64) return isNonPublicIpv4(bytes.slice(12).join('.')) + const sixToFour = bytes[0] === 0x20 && bytes[1] === 0x02 + if (sixToFour) return isNonPublicIpv4(bytes.slice(2, 6).join('.')) + return ( + unspecified || + loopback || + (bytes[0] & 0xfe) === 0xfc || + (bytes[0] === 0xfe && (bytes[1] & 0xc0) === 0x80) || + (bytes[0] === 0xfe && (bytes[1] & 0xc0) === 0xc0) || + bytes[0] === 0xff || + (bytes[0] === 0x20 && bytes[1] === 0x01 && bytes[2] === 0x00 && bytes[3] === 0x00) || + (bytes[0] === 0x20 && bytes[1] === 0x01 && bytes[2] === 0x0d && bytes[3] === 0xb8) + ) +} + +export function isPrivateOrLocalHostname(hostname: string): boolean { + const value = normalizedHostname(hostname) + if (!value) return true + if (value.includes(':')) return isNonPublicIpv6(value) + if (ipv4Octets(value)) return isNonPublicIpv4(value) + return ( + !value.includes('.') || + value === 'localhost' || + value.endsWith('.localhost') || + value === 'localdomain' || + value.endsWith('.localdomain') || + value.endsWith('.local') || + value === 'internal' || + value.endsWith('.internal') || + value === 'home.arpa' || + value.endsWith('.home.arpa') + ) +} + +function parseHttpUrl(value: string, fieldName: string): URL { + let parsed: URL + try { + parsed = new URL(value) + } catch { + throw new HookUrlPolicyError(`${fieldName} must be a valid HTTP(S) URL`) + } + + if (parsed.protocol !== 'http:' && parsed.protocol !== 'https:') { + throw new HookUrlPolicyError(`${fieldName} must be a valid HTTP(S) URL`) + } + if (parsed.username || parsed.password) { + throw new HookUrlPolicyError(`${fieldName} must not contain credentials`) + } + return parsed +} + +export function normalizeAllowedOrigins(values: readonly string[]): readonly string[] { + const origins = new Set() + for (const rawValue of values) { + const value = rawValue.trim() + if (!value) throw new HookUrlPolicyError('allowed origin must not be empty') + const parsed = parseHttpUrl(value, 'allowed origin') + if ( + (parsed.pathname && parsed.pathname !== '/') || + parsed.search || + parsed.hash || + value.includes('?') || + value.includes('#') + ) { + throw new HookUrlPolicyError('allowed origin must not contain a path, query, or fragment') + } + origins.add(parsed.origin) + } + return Object.freeze([...origins]) +} + +export class HookUrlPolicy { + private readonly allowedOrigins: ReadonlySet + private readonly allowPrivate: boolean + readonly configurationKey: string + + constructor(origins: readonly string[] = [], options: HookUrlPolicyOptions = {}) { + const normalizedOrigins = normalizeAllowedOrigins(origins) + this.allowedOrigins = new Set(normalizedOrigins) + this.allowPrivate = options.allowPrivate ?? false + this.configurationKey = JSON.stringify([normalizedOrigins, this.allowPrivate]) + } + + get enforcesPublicAddresses(): boolean { + return this.allowedOrigins.size > 0 && !this.allowPrivate + } + + assertAllowedAddress(address: string): void { + if (this.enforcesPublicAddresses && isPrivateOrLocalHostname(address)) { + throw new HookUrlPolicyError('hookUrl resolved to a private or local address') + } + } + + assertAllowed(value: string): URL { + const parsed = parseHttpUrl(value, 'hookUrl') + if (this.allowedOrigins.size > 0 && !this.allowedOrigins.has(parsed.origin)) { + throw new HookUrlPolicyError('hookUrl origin is not allowed') + } + if (this.enforcesPublicAddresses && isPrivateOrLocalHostname(parsed.hostname)) { + throw new HookUrlPolicyError('hookUrl must not target a private or local address') + } + return parsed + } +} + +export const permissiveHookUrlPolicy = new HookUrlPolicy() diff --git a/wait-queue/src/security/rate_limit.ts b/wait-queue/src/security/rate_limit.ts new file mode 100644 index 0000000..bcecbf8 --- /dev/null +++ b/wait-queue/src/security/rate_limit.ts @@ -0,0 +1,68 @@ +export interface RateLimitDecision { + allowed: boolean + retryAfterSeconds: number +} + +interface WindowEntry { + count: number + resetAt: number +} + +export const MAX_RATE_LIMIT_CLIENTS = 10_000 + +export class FixedWindowRateLimiter { + private readonly entries = new Map() + + constructor( + private readonly maxRequests: number, + private readonly windowMs: number, + private readonly now: () => number = Date.now, + private readonly maxClients: number = MAX_RATE_LIMIT_CLIENTS + ) { + if (!Number.isInteger(maxRequests) || maxRequests < 0) throw new Error('maxRequests must be a non-negative integer') + if (!Number.isInteger(windowMs) || windowMs <= 0) throw new Error('windowMs must be a positive integer') + if (!Number.isInteger(maxClients) || maxClients <= 0 || maxClients > MAX_RATE_LIMIT_CLIENTS) { + throw new Error(`maxClients must be an integer between 1 and ${MAX_RATE_LIMIT_CLIENTS}`) + } + } + + get size(): number { + return this.entries.size + } + + consume(key: string): RateLimitDecision { + if (this.maxRequests === 0) return { allowed: true, retryAfterSeconds: 0 } + + const now = this.now() + let entry = this.entries.get(key) + if (entry && now >= entry.resetAt) { + this.entries.delete(key) + entry = undefined + } + + if (!entry) { + this.ensureCapacity(now) + entry = { count: 0, resetAt: now + this.windowMs } + this.entries.set(key, entry) + } + + entry.count += 1 + const allowed = entry.count <= this.maxRequests + return { + allowed, + retryAfterSeconds: allowed ? 0 : Math.max(1, Math.ceil((entry.resetAt - now) / 1000)), + } + } + + private ensureCapacity(now: number): void { + if (this.entries.size < this.maxClients) return + for (const [key, entry] of this.entries) { + if (now >= entry.resetAt) this.entries.delete(key) + } + while (this.entries.size >= this.maxClients) { + const oldestKey = this.entries.keys().next().value as string | undefined + if (oldestKey === undefined) break + this.entries.delete(oldestKey) + } + } +} diff --git a/wait-queue/src/service/queue.ts b/wait-queue/src/service/queue.ts index fc394ca..441ee4e 100644 --- a/wait-queue/src/service/queue.ts +++ b/wait-queue/src/service/queue.ts @@ -5,12 +5,21 @@ import { ModelCtor } from 'sequelize' import { Timer } from '../lib/timer' import { NewQueueRequest, OperationResult } from '../types/api' import { createBackgroundContext } from '../common/logger' +import { HookUrlPolicy } from '../security/hook_url_policy' +import { env } from '../conf/env' export class QueueService extends Service { private queueDao: ModelCtor - constructor(ctx: Context) { + private hookUrlPolicy: HookUrlPolicy + constructor( + ctx: Context, + hookUrlPolicy: HookUrlPolicy = new HookUrlPolicy(env.security.hookUrlAllowlist, { + allowPrivate: env.security.allowPrivateHookUrls, + }) + ) { super(ctx) this.queueDao = QueueDao + this.hookUrlPolicy = hookUrlPolicy } async newQueue(params: NewQueueRequest): Promise { @@ -36,7 +45,7 @@ export class QueueService extends Service { }) } - await new Timer(createBackgroundContext()).initializeQueueList([queue.id]) + await new Timer(createBackgroundContext(), this.hookUrlPolicy).initializeQueueList([queue.id]) return { isOk: true } } diff --git a/wait-queue/src/service/scheduler.ts b/wait-queue/src/service/scheduler.ts index 16abb28..cb054be 100644 --- a/wait-queue/src/service/scheduler.ts +++ b/wait-queue/src/service/scheduler.ts @@ -30,7 +30,7 @@ export class SchedulerService extends Service { throw new HttpError(404, 'queue not found; register it before adding tasks') } - this.baseLogInfo(`TaskManager-${namespace}|url:${hookUrl}|taskId:${taskId}|addTask: push task to queue`) + this.baseLogInfo('task added to waiting queue', { queueId: queueRes.id, namespace }) await this.redis.lpush(getWaitingKey(queueRes.namespace, queueRes.id), taskId) return { isOk: true, diff --git a/wait-queue/src/utils/validation.ts b/wait-queue/src/utils/validation.ts index 7429003..dba4dda 100644 --- a/wait-queue/src/utils/validation.ts +++ b/wait-queue/src/utils/validation.ts @@ -1,6 +1,11 @@ import { CronTime } from 'cron' import { AddTaskRequest, NewQueueRequest, QueueCrontab } from '../types/api' import { HttpError } from './http_error' +import { + HookUrlPolicy, + HookUrlPolicyError, + permissiveHookUrlPolicy, +} from '../security/hook_url_policy' export const DEFAULT_QUEUE_CONCURRENCY = 5 export const DEFAULT_QUEUE_CRONTAB: QueueCrontab = Object.freeze({ @@ -30,13 +35,13 @@ function requiredString(source: JsonObject, field: string, maxLength: number): s return normalized } -function hookUrl(source: JsonObject): string { +function hookUrl(source: JsonObject, policy: HookUrlPolicy): string { const value = requiredString(source, 'hookUrl', 255) try { - const parsed = new URL(value) - if (parsed.protocol !== 'http:' && parsed.protocol !== 'https:') throw new Error('unsupported protocol') - } catch { - throw new HttpError(400, 'hookUrl must be a valid HTTP(S) URL') + policy.assertAllowed(value) + } catch (error) { + if (error instanceof HookUrlPolicyError) throw new HttpError(400, error.message) + throw error } return value } @@ -57,7 +62,10 @@ function cronExpression(value: unknown, field: keyof QueueCrontab): string { return normalized } -export function validateNewQueueInput(value: unknown): NewQueueRequest { +export function validateNewQueueInput( + value: unknown, + hookUrlPolicy: HookUrlPolicy = permissiveHookUrlPolicy +): NewQueueRequest { const body = asObject(value) const rawConcurrency = body.currMaxCount ?? DEFAULT_QUEUE_CONCURRENCY if (!Number.isInteger(rawConcurrency) || (rawConcurrency as number) < 1 || (rawConcurrency as number) > 1000) { @@ -66,7 +74,7 @@ export function validateNewQueueInput(value: unknown): NewQueueRequest { const rawCrontab = body.crontab === undefined ? {} : asObject(body.crontab) return { - hookUrl: hookUrl(body), + hookUrl: hookUrl(body, hookUrlPolicy), namespace: requiredString(body, 'namespace', 64), currMaxCount: rawConcurrency as number, crontab: { @@ -77,10 +85,13 @@ export function validateNewQueueInput(value: unknown): NewQueueRequest { } } -export function validateAddTaskInput(value: unknown): AddTaskRequest { +export function validateAddTaskInput( + value: unknown, + hookUrlPolicy: HookUrlPolicy = permissiveHookUrlPolicy +): AddTaskRequest { const body = asObject(value) return { - hookUrl: hookUrl(body), + hookUrl: hookUrl(body, hookUrlPolicy), namespace: requiredString(body, 'namespace', 64), taskId: requiredString(body, 'taskId', 256), } diff --git a/wait-queue/test/security.test.js b/wait-queue/test/security.test.js new file mode 100644 index 0000000..d0a6f3d --- /dev/null +++ b/wait-queue/test/security.test.js @@ -0,0 +1,514 @@ +const test = require('node:test') +const assert = require('node:assert/strict') +const { once } = require('node:events') +const { Writable } = require('node:stream') + +const { createApp } = require('../dist/app.js') +const { QueueDao } = require('../dist/dao/queue_dao.js') +const { redisCli } = require('../dist/conf/redis.js') +const { Timer } = require('../dist/lib/timer.js') +const { + createSecurityConfigurationWarner, + createSecurityConfig, + readSecurityConfig, +} = require('../dist/security/config.js') +const { HookUrlPolicy } = require('../dist/security/hook_url_policy.js') +const { FixedWindowRateLimiter } = require('../dist/security/rate_limit.js') +const { validateAddTaskInput, validateNewQueueInput } = require('../dist/utils/validation.js') +const { HttpError } = require('../dist/utils/http_error.js') + +function logSink(lines = undefined) { + return new Writable({ + write(chunk, _encoding, callback) { + if (lines) lines.push(String(chunk)) + callback() + }, + }) +} + +async function startTestApp(t, options = {}) { + const server = createApp({ requestLogStream: logSink(), ...options }).listen(0, '127.0.0.1') + await once(server, 'listening') + t.after(() => new Promise((resolve, reject) => server.close((error) => (error ? reject(error) : resolve())))) + const address = server.address() + assert.ok(address && typeof address === 'object') + return `http://127.0.0.1:${address.port}` +} + +test('security environment configuration validates numeric values, tokens, and exact origins', () => { + assert.deepEqual(readSecurityConfig({}), { + apiToken: undefined, + hookUrlAllowlist: [], + allowPrivateHookUrls: false, + requestBodyLimitBytes: 32768, + rateLimitMaxRequests: 0, + rateLimitWindowMs: 60000, + }) + + const parsed = readSecurityConfig({ + WAITQUEUE_API_TOKEN: 'secret', + HOOK_URL_ALLOWLIST: 'https://worker.example.com, https://jobs.example.net:8443/', + HOOK_URL_ALLOW_PRIVATE: 'true', + REQUEST_BODY_LIMIT_BYTES: '1024', + RATE_LIMIT_MAX_REQUESTS: '20', + RATE_LIMIT_WINDOW_MS: '5000', + }) + assert.equal(parsed.apiToken, 'secret') + assert.deepEqual(parsed.hookUrlAllowlist, [ + 'https://worker.example.com', + 'https://jobs.example.net:8443', + ]) + assert.equal(parsed.allowPrivateHookUrls, true) + assert.equal(parsed.requestBodyLimitBytes, 1024) + assert.equal(parsed.rateLimitMaxRequests, 20) + assert.equal(parsed.rateLimitWindowMs, 5000) + + for (const environment of [ + { REQUEST_BODY_LIMIT_BYTES: '0' }, + { RATE_LIMIT_MAX_REQUESTS: '-1' }, + { RATE_LIMIT_WINDOW_MS: '1.5' }, + { WAITQUEUE_API_TOKEN: ' ' }, + { WAITQUEUE_API_TOKEN: 'abc def' }, + { HOOK_URL_ALLOW_PRIVATE: 'yes' }, + { HOOK_URL_ALLOWLIST: 'https://worker.example.com/path' }, + { HOOK_URL_ALLOWLIST: 'https://worker.example.com?' }, + { HOOK_URL_ALLOWLIST: 'redis://worker.example.com' }, + { HOOK_URL_ALLOWLIST: 'https://user:pass@worker.example.com' }, + ]) { + assert.throws(() => readSecurityConfig(environment)) + } + assert.throws(() => createSecurityConfig({ requestBodyLimitBytes: Number.NaN })) +}) + +test('insecure compatibility defaults emit each startup warning only once', () => { + const warnings = [] + const warn = createSecurityConfigurationWarner({ + warn(bindings, message) { + warnings.push({ bindings, message }) + }, + }) + const insecure = createSecurityConfig() + warn(insecure) + warn(insecure) + assert.equal(warnings.length, 2) + assert.deepEqual( + warnings.map(({ bindings }) => bindings.configuration).sort(), + ['HOOK_URL_ALLOWLIST', 'WAITQUEUE_API_TOKEN'] + ) + assert.ok(warnings.every(({ message }) => !message.includes('undefined'))) +}) + +test('Bearer authentication protects control APIs while health, readiness, and OPTIONS stay open', async (t) => { + const originalFindAll = QueueDao.findAll + QueueDao.findAll = async () => [] + t.after(() => { + QueueDao.findAll = originalFindAll + }) + + const baseUrl = await startTestApp(t, { + security: { apiToken: 'test-secret' }, + readinessCheck: async () => ({ + ready: true, + dependencies: { mysql: 'ok', redis: 'ok' }, + }), + }) + + assert.equal((await fetch(`${baseUrl}/waitqueue/health`)).status, 200) + assert.equal((await fetch(`${baseUrl}/waitqueue/ready`)).status, 200) + assert.equal((await fetch(`${baseUrl}/waitqueue/admin/overview`, { method: 'OPTIONS' })).status, 200) + + const missing = await fetch(`${baseUrl}/waitqueue/admin/overview`) + assert.equal(missing.status, 401) + assert.equal(missing.headers.get('www-authenticate'), 'Bearer') + assert.equal((await missing.json()).msg, 'authentication required') + + assert.equal( + ( + await fetch(`${baseUrl}/waitqueue/admin/overview`, { + headers: { authorization: 'Bearer wrong-secret' }, + }) + ).status, + 401 + ) + assert.equal( + ( + await fetch(`${baseUrl}/waitqueue/admin/overview`, { + headers: { authorization: 'Bearer test-secret' }, + }) + ).status, + 200 + ) + assert.equal((await fetch(`${baseUrl}/waitqueue/future-control-route`)).status, 401) + assert.equal( + ( + await fetch(`${baseUrl}/waitqueue/future-control-route`, { + headers: { authorization: 'Bearer test-secret' }, + }) + ).status, + 404, + 'new control-plane routes must be authenticated by default' + ) + + for (const path of [ + '/WAITQUEUE/ADMIN/overview', + '/waitqueue/ADMIN/overview', + '/waitqueue/QUEUE/newQueue', + '/waitqueue/SCHEDULER/addTask', + ]) { + const method = path.endsWith('overview') ? 'GET' : 'POST' + assert.equal( + (await fetch(`${baseUrl}${path}`, { method })).status, + 401, + `case variants of protected paths must not bypass authentication: ${path}` + ) + assert.equal( + ( + await fetch(`${baseUrl}${path}`, { + method, + headers: { authorization: 'Bearer test-secret' }, + }) + ).status, + 404, + `routing must reject non-canonical path casing: ${path}` + ) + } +}) + +test('authentication and rate limiting run before the JSON-only bounded body parser', async (t) => { + const baseUrl = await startTestApp(t, { + security: { apiToken: 'test-secret', requestBodyLimitBytes: 64 }, + }) + const oversizedBody = JSON.stringify({ hookUrl: `https://worker.example.com/${'x'.repeat(200)}` }) + + const unauthorized = await fetch(`${baseUrl}/waitqueue/queue/newQueue`, { + method: 'POST', + headers: { 'content-type': 'application/json', authorization: 'Bearer wrong-secret' }, + body: oversizedBody, + }) + assert.equal(unauthorized.status, 401, 'auth rejection must happen before parsing an oversized body') + + const oversized = await fetch(`${baseUrl}/waitqueue/queue/newQueue`, { + method: 'POST', + headers: { 'content-type': 'application/json', authorization: 'Bearer test-secret' }, + body: oversizedBody, + }) + assert.equal(oversized.status, 413) + + const form = await fetch(`${baseUrl}/waitqueue/queue/newQueue`, { + method: 'POST', + headers: { 'content-type': 'application/x-www-form-urlencoded', authorization: 'Bearer test-secret' }, + body: 'hookUrl=https%3A%2F%2Fworker.example.com', + }) + assert.equal(form.status, 400) + assert.equal((await form.json()).msg, 'hookUrl is required') +}) + +test('fixed-window limiter resets on the injected clock, returns Retry-After, and stays bounded', async (t) => { + let now = 10_000 + const baseUrl = await startTestApp(t, { + security: { rateLimitMaxRequests: 2, rateLimitWindowMs: 5000 }, + rateLimitClock: () => now, + }) + + assert.equal((await fetch(`${baseUrl}/waitqueue/admin/missing`)).status, 404) + assert.equal((await fetch(`${baseUrl}/waitqueue/admin/missing`)).status, 404) + const limited = await fetch(`${baseUrl}/waitqueue/admin/missing`) + assert.equal(limited.status, 429) + assert.equal(limited.headers.get('retry-after'), '5') + assert.equal((await fetch(`${baseUrl}/waitqueue/health`)).status, 200, 'health is not rate limited') + + now += 5000 + assert.equal((await fetch(`${baseUrl}/waitqueue/admin/missing`)).status, 404) + + const bounded = new FixedWindowRateLimiter(1, 1000, () => now, 2) + bounded.consume('client-1') + bounded.consume('client-2') + bounded.consume('client-3') + assert.equal(bounded.size, 2) +}) + +test('hook URL policy rejects credentials and non-allowlisted origins at request validation', () => { + const policy = new HookUrlPolicy(['https://worker.example.com']) + const base = { namespace: 'billing', hookUrl: 'https://worker.example.com/tasks' } + assert.equal(validateNewQueueInput(base, policy).hookUrl, base.hookUrl) + assert.equal(validateAddTaskInput({ ...base, taskId: 'task-1' }, policy).hookUrl, base.hookUrl) + + for (const hookUrl of [ + 'https://other.example.com/tasks', + 'https://user:pass@worker.example.com/tasks', + 'file:///tmp/callback', + ]) { + assert.throws( + () => validateNewQueueInput({ ...base, hookUrl }, policy), + (error) => error instanceof HttpError && error.status === 400 + ) + } + + for (const hookUrl of [ + 'http://127.0.0.1:3101/callback', + 'http://127.255.255.254/callback', + 'http://2130706433/callback', + 'http://0x7f000001/callback', + 'http://localhost:3101/callback', + 'http://10.0.0.8/callback', + 'http://192.168.1.8/callback', + 'http://169.254.169.254/latest/meta-data', + 'http://[::1]:3101/callback', + 'http://[::ffff:127.0.0.1]/callback', + 'http://[::127.0.0.1]/callback', + 'http://[64:ff9b::127.0.0.1]/callback', + 'http://[2002:7f00:1::]/callback', + 'http://[fc00::1]/callback', + 'http://[fe80::1]/callback', + 'http://mock-hook:3101/callback', + ]) { + const strictPolicy = new HookUrlPolicy([new URL(hookUrl).origin]) + assert.throws( + () => validateNewQueueInput({ ...base, hookUrl }, strictPolicy), + (error) => + error instanceof HttpError && + error.status === 400 && + error.message === 'hookUrl must not target a private or local address' + ) + } + + const localHookUrl = 'http://mock-hook:3101/callback' + assert.equal( + validateNewQueueInput({ ...base, hookUrl: localHookUrl }).hookUrl, + localHookUrl, + 'empty allowlist keeps the legacy local-development behavior' + ) + const explicitDevelopmentPolicy = new HookUrlPolicy(['http://mock-hook:3101'], { + allowPrivate: true, + }) + assert.equal( + validateNewQueueInput({ ...base, hookUrl: localHookUrl }, explicitDevelopmentPolicy).hookUrl, + localHookUrl + ) +}) + +test('audit logs use final statuses and redact credentials and callback identifiers', async (t) => { + const originalFindOne = QueueDao.findOne + const originalFindOrCreate = QueueDao.findOrCreate + const redis = redisCli.getInstance() + const originalLpush = redis.lpush + const originalInitializeQueueList = Timer.prototype.initializeQueueList + let timerHookUrlPolicy + QueueDao.findOne = async () => ({ id: 7, namespace: 'billing' }) + QueueDao.findOrCreate = async () => [{ id: 7, async update() {} }, true] + redis.lpush = async () => 1 + Timer.prototype.initializeQueueList = async function () { + timerHookUrlPolicy = this.hookUrlPolicy + } + t.after(() => { + QueueDao.findOne = originalFindOne + QueueDao.findOrCreate = originalFindOrCreate + redis.lpush = originalLpush + Timer.prototype.initializeQueueList = originalInitializeQueueList + }) + + const lines = [] + let now = 0 + const server = createApp({ + requestLogStream: logSink(lines), + security: { + apiToken: 'audit-secret-token', + hookUrlAllowlist: ['https://private-worker.example.com'], + rateLimitMaxRequests: 2, + rateLimitWindowMs: 1000, + }, + rateLimitClock: () => now, + }).listen(0, '127.0.0.1') + await once(server, 'listening') + t.after(() => new Promise((resolve, reject) => server.close((error) => (error ? reject(error) : resolve())))) + const address = server.address() + assert.ok(address && typeof address === 'object') + const url = `http://127.0.0.1:${address.port}/waitqueue/scheduler/addTask` + const queueUrl = `http://127.0.0.1:${address.port}/waitqueue/queue/newQueue` + const body = JSON.stringify({ + hookUrl: 'https://private-worker.example.com/callback', + namespace: 'billing', + taskId: 'private-task-id', + }) + + assert.equal( + ( + await fetch(url, { + method: 'POST', + headers: { 'content-type': 'application/json', authorization: 'Bearer audit-secret-token' }, + body, + }) + ).status, + 200 + ) + assert.equal( + ( + await fetch(queueUrl, { + method: 'POST', + headers: { + 'content-type': 'application/json', + authorization: 'Bearer audit-secret-token', + cookie: 'session=private-cookie-value', + }, + body: JSON.stringify({ + hookUrl: 'https://private-worker.example.com/callback', + namespace: 'billing', + }), + }) + ).status, + 200 + ) + assert.equal( + ( + await fetch(url, { + method: 'POST', + headers: { 'content-type': 'application/json', authorization: 'Bearer audit-secret-token' }, + body, + }) + ).status, + 429 + ) + now = 1000 + assert.equal( + ( + await fetch(url, { + method: 'POST', + headers: { 'content-type': 'application/json', authorization: 'Bearer wrong-secret-token' }, + body, + }) + ).status, + 401 + ) + + const records = lines + .join('') + .trim() + .split('\n') + .filter(Boolean) + .map(JSON.parse) + const audits = records.filter((record) => record.audit?.event === 'api_audit') + assert.ok( + audits.some( + (record) => + record.audit.action === 'task.enqueue' && + record.audit.outcome === 'succeeded' && + record.audit.statusCode === 200 && + Number.isInteger(record.audit.durationMs) + ) + ) + assert.equal( + timerHookUrlPolicy?.configurationKey, + new HookUrlPolicy(['https://private-worker.example.com']).configurationKey, + 'route validation and scheduled callbacks must share the same hook URL policy' + ) + assert.ok( + audits.some( + (record) => record.audit.action === 'queue.configure' && record.audit.statusCode === 200 + ) + ) + assert.ok( + audits.some((record) => record.audit.action === 'rate_limit.denied' && record.audit.statusCode === 429) + ) + assert.ok(audits.some((record) => record.audit.action === 'auth.denied' && record.audit.statusCode === 401)) + + const serialized = JSON.stringify(records) + assert.doesNotMatch(serialized, /audit-secret-token|wrong-secret-token|private-cookie-value/) + assert.doesNotMatch(serialized, /private-worker\.example\.com|private-task-id/) + assert.ok(records.every((record) => record.req?.headers === undefined)) +}) + +test('error serialization drops messages, SQL parameters, command arguments, and causes', async (t) => { + const originalFindOrCreate = QueueDao.findOrCreate + const sensitiveError = new Error( + 'database failure for https://secret-worker.example/callback and private-task-id' + ) + sensitiveError.code = 'ER_SYNTHETIC' + sensitiveError.sql = 'INSERT INTO queue VALUES ("https://secret-worker.example/callback")' + sensitiveError.parameters = ['private-task-id'] + sensitiveError.command = { args: ['private-api-token'] } + sensitiveError.cause = new Error('private-cause-value') + QueueDao.findOrCreate = async () => { + throw sensitiveError + } + t.after(() => { + QueueDao.findOrCreate = originalFindOrCreate + }) + + const lines = [] + const server = createApp({ + requestLogStream: logSink(lines), + security: { apiToken: 'private-api-token' }, + }).listen(0, '127.0.0.1') + await once(server, 'listening') + t.after(() => new Promise((resolve, reject) => server.close((error) => (error ? reject(error) : resolve())))) + const address = server.address() + assert.ok(address && typeof address === 'object') + const response = await fetch( + `http://127.0.0.1:${address.port}/waitqueue/queue/newQueue?token=private-query-secret&hookUrl=private-query-url`, + { + method: 'POST', + headers: { + 'content-type': 'application/json', + authorization: 'Bearer private-api-token', + }, + body: JSON.stringify({ + hookUrl: 'https://secret-worker.example/callback', + namespace: 'billing', + }), + } + ) + assert.equal(response.status, 500) + assert.equal((await response.json()).msg, 'internal server error') + const unknownPathResponse = await fetch( + `http://127.0.0.1:${address.port}/waitqueue/private-path-secret?token=private-unknown-query`, + { + method: 'POST', + headers: { + 'content-type': 'application/json', + authorization: 'Bearer private-api-token', + }, + body: '{}', + } + ) + assert.equal(unknownPathResponse.status, 404) + + const serialized = lines.join('') + assert.doesNotMatch( + serialized, + /private-api-token|private-query-secret|private-query-url|private-path-secret|private-unknown-query|secret-worker\.example|private-task-id|private-cause-value|INSERT INTO/ + ) + const records = serialized + .trim() + .split('\n') + .filter(Boolean) + .map(JSON.parse) + const errorRecord = records.find((record) => record.err?.code === 'ER_SYNTHETIC') + assert.deepEqual(errorRecord?.err, { type: 'Error', code: 'ER_SYNTHETIC' }) + assert.ok( + records.some( + (record) => + record.req?.method === 'POST' && + record.req.path === '/waitqueue/queue/newQueue' && + record.req.url === undefined + ), + 'request logs must contain only a canonical path without URL queries' + ) + assert.ok( + records.some( + (record) => + record.audit?.action === 'queue.configure' && + record.audit.outcome === 'failed' && + record.audit.statusCode === 500 + ), + 'failed write attempts must remain auditable without serializing their payloads' + ) + assert.ok( + records.some( + (record) => + record.audit?.action === 'api.write' && + record.audit.path === '/waitqueue/[unmatched]' && + record.audit.statusCode === 404 + ), + 'unknown write paths must be normalized before audit logging' + ) +}) diff --git a/wait-queue/test/task_run.test.js b/wait-queue/test/task_run.test.js index 88a1a40..2285cf3 100644 --- a/wait-queue/test/task_run.test.js +++ b/wait-queue/test/task_run.test.js @@ -1,7 +1,14 @@ const test = require('node:test') const assert = require('node:assert/strict') +const http = require('node:http') +const { once } = require('node:events') -const { TaskRun } = require('../dist/lib/task_run.js') +const { + TaskRun, + createCallbackTransport, + resolvePinnedCallbackAddresses, +} = require('../dist/lib/task_run.js') +const { HookUrlPolicy } = require('../dist/security/hook_url_policy.js') function createContext() { return { @@ -11,36 +18,32 @@ function createContext() { } test('run, check, and expire callbacks include queueId and namespace', async (t) => { - const originalFetch = globalThis.fetch const calls = [] - globalThis.fetch = async (url, init) => { - const body = JSON.parse(init.body) - calls.push({ url, init, body }) + const callbackTransport = async (url, options) => { + const body = JSON.parse(options.body) + calls.push({ url, options, body }) const responseBody = body.type === 'run' ? '' : JSON.stringify({ data: { taskIds: body.type === 'check' ? ['task-2'] : ['task-3'] } }) - return { - status: 200, - async text() { - return responseBody - }, - } + return { status: 200, body: responseBody } } - t.after(() => { - globalThis.fetch = originalFetch - }) - const taskRun = new TaskRun(createContext(), 'https://worker.example.com/tasks', 7, 'billing') + const taskRun = new TaskRun( + createContext(), + 'https://worker.example.com/tasks', + 7, + 'billing', + undefined, + callbackTransport + ) await taskRun.run('task-1') assert.deepEqual(await taskRun.checkTaskStatus(['task-1', 'task-2']), ['task-2']) assert.deepEqual(await taskRun.expireTasks(), ['task-3']) assert.equal(calls.length, 3) - assert.ok(calls.every(({ url }) => url === 'https://worker.example.com/tasks')) - assert.ok(calls.every(({ init }) => init.method === 'POST')) - assert.ok(calls.every(({ init }) => init.headers['content-type'] === 'application/json')) - assert.ok(calls.every(({ init }) => init.signal instanceof AbortSignal)) + assert.ok(calls.every(({ url }) => url.href === 'https://worker.example.com/tasks')) + assert.ok(calls.every(({ options }) => options.signal instanceof AbortSignal)) assert.deepEqual( calls.map(({ body }) => body), [ @@ -50,3 +53,161 @@ test('run, check, and expire callbacks include queueId and namespace', async (t) ] ) }) + +test('TaskRun revalidates callback policy before sending and never follows redirects', async (t) => { + const calls = [] + const callbackTransport = async (url, options) => { + calls.push({ url, options }) + return { status: 302, body: '' } + } + + const denied = new TaskRun( + createContext(), + 'https://blocked.example.com/tasks', + 7, + 'billing', + new HookUrlPolicy(['https://worker.example.com']), + callbackTransport + ) + await assert.rejects(denied.run('private-task-id'), /origin is not allowed/) + assert.equal(calls.length, 0, 'policy rejection must happen immediately before fetch') + + const redirecting = new TaskRun( + createContext(), + 'https://worker.example.com/tasks', + 7, + 'billing', + new HookUrlPolicy(['https://worker.example.com']), + callbackTransport + ) + await assert.rejects(redirecting.run('private-task-id'), /returned HTTP 302/) + assert.equal(calls.length, 1) +}) + +test('strict callback resolution rejects private DNS answers and pins public answers', async () => { + const url = new URL('https://worker.example.com/tasks') + const policy = new HookUrlPolicy(['https://worker.example.com']) + await assert.rejects( + resolvePinnedCallbackAddresses(url, policy, async () => [{ address: '127.0.0.1', family: 4 }]), + /resolved to a private or local address/ + ) + await assert.rejects( + resolvePinnedCallbackAddresses(url, policy, async () => [{ address: '169.254.169.254', family: 4 }]), + /resolved to a private or local address/ + ) + assert.deepEqual( + await resolvePinnedCallbackAddresses(url, policy, async () => [ + { address: '1.1.1.1', family: 4 }, + { address: '2001:4860:4860::8888', family: 6 }, + ]), + [ + { address: '1.1.1.1', family: 4 }, + { address: '2001:4860:4860::8888', family: 6 }, + ] + ) +}) + +test('pinned callback transport connects with the validated address on Node lookup APIs', async (t) => { + let receivedBody + const server = http.createServer((request, response) => { + const chunks = [] + request.on('data', (chunk) => chunks.push(chunk)) + request.on('end', () => { + receivedBody = JSON.parse(Buffer.concat(chunks).toString('utf8')) + response.statusCode = 200 + response.end() + }) + }) + server.listen(0, '127.0.0.1') + await once(server, 'listening') + t.after(() => new Promise((resolve) => server.close(resolve))) + const address = server.address() + assert.ok(address && typeof address === 'object') + const origin = `http://callback.example.test:${address.port}` + const policy = new HookUrlPolicy([origin]) + policy.assertAllowedAddress = () => {} + const transport = createCallbackTransport(async (hostname) => { + assert.equal(hostname, 'callback.example.test') + return [{ address: '127.0.0.1', family: 4 }] + }) + const taskRun = new TaskRun(createContext(), `${origin}/callback`, 7, 'billing', policy, transport) + + await taskRun.run('task-1') + assert.deepEqual(receivedBody, { + type: 'run', + queueId: 7, + namespace: 'billing', + taskIds: ['task-1'], + }) +}) + +test('callback timeout also bounds DNS resolution', async () => { + const policy = new HookUrlPolicy(['https://worker.example.com']) + const transport = createCallbackTransport(() => new Promise(() => {})) + const controller = new AbortController() + const request = transport(new URL('https://worker.example.com/callback'), { + body: '{}', + signal: controller.signal, + hookUrlPolicy: policy, + }) + controller.abort() + await assert.rejects(request, (error) => error?.name === 'AbortError') +}) + +test('default callback transport returns a redirect without requesting its location', async (t) => { + let redirectedRequests = 0 + const destination = http.createServer((_request, response) => { + redirectedRequests += 1 + response.end('unexpected') + }) + destination.listen(0, '127.0.0.1') + await once(destination, 'listening') + t.after(() => new Promise((resolve) => destination.close(resolve))) + const destinationAddress = destination.address() + assert.ok(destinationAddress && typeof destinationAddress === 'object') + + const redirect = http.createServer((_request, response) => { + response.writeHead(302, { + location: `http://127.0.0.1:${destinationAddress.port}/private-destination`, + }) + response.end() + }) + redirect.listen(0, '127.0.0.1') + await once(redirect, 'listening') + t.after(() => new Promise((resolve) => redirect.close(resolve))) + const redirectAddress = redirect.address() + assert.ok(redirectAddress && typeof redirectAddress === 'object') + const origin = `http://127.0.0.1:${redirectAddress.port}` + const taskRun = new TaskRun( + createContext(), + `${origin}/callback`, + 7, + 'billing', + new HookUrlPolicy([origin], { allowPrivate: true }) + ) + + await assert.rejects(taskRun.run('private-task-id'), /returned HTTP 302/) + assert.equal(redirectedRequests, 0) +}) + +test('default callback transport rejects oversized callback responses', async (t) => { + const server = http.createServer((_request, response) => { + response.statusCode = 200 + response.end(Buffer.alloc(1_048_577, 'x')) + }) + server.listen(0, '127.0.0.1') + await once(server, 'listening') + t.after(() => new Promise((resolve) => server.close(resolve))) + const address = server.address() + assert.ok(address && typeof address === 'object') + const origin = `http://127.0.0.1:${address.port}` + const taskRun = new TaskRun( + createContext(), + `${origin}/callback`, + 7, + 'billing', + new HookUrlPolicy([origin], { allowPrivate: true }) + ) + + await assert.rejects(taskRun.run('task-1'), /response exceeded the size limit/) +}) diff --git a/wait-queue/test/timer.test.js b/wait-queue/test/timer.test.js index 98b282d..fb84639 100644 --- a/wait-queue/test/timer.test.js +++ b/wait-queue/test/timer.test.js @@ -11,6 +11,7 @@ function replaceModule(modulePath, exports) { } const cronInstances = [] +const taskManagerPolicies = [] class FakeCronJob { constructor(cron, onTick) { if (cron === 'invalid-cron') throw new Error('invalid cron') @@ -37,6 +38,9 @@ const cronPath = require.resolve('cron') replaceModule(queueDaoPath, { QueueDao: { async findAll() { return queueRows } } }) replaceModule(taskManagerPath, { TaskManager: class { + constructor(...args) { + taskManagerPolicies.push(args.at(-1)) + } async runTask() {} async checkTaskStatus() {} async expireTask() {} @@ -45,6 +49,7 @@ replaceModule(taskManagerPath, { replaceModule(cronPath, { CronJob: FakeCronJob }) const { Timer } = require('../dist/lib/timer.js') +const { HookUrlPolicy } = require('../dist/security/hook_url_policy.js') function createContext() { return { @@ -67,10 +72,12 @@ function queue(count = 2, overrides = {}) { } test('timer reuses unchanged jobs, replaces changed configuration, and stops deleted queues', async () => { - const timer = new Timer(createContext()) + const hookUrlPolicy = new HookUrlPolicy(['https://worker.example.com']) + const timer = new Timer(createContext(), hookUrlPolicy) queueRows = [queue(2)] await timer.initializeQueueList() + assert.equal(taskManagerPolicies[0], hookUrlPolicy, 'timer must pass its callback policy to each task manager') assert.equal(cronInstances.length, 3) assert.ok(cronInstances.every((job) => job.started && !job.stopped))