From 076ba01c74e9cb5b33f20496c1d00f29fc1cfd45 Mon Sep 17 00:00:00 2001 From: host Date: Mon, 26 Oct 2026 01:32:57 +0800 Subject: [PATCH] cloud: fill gaps (memory/KB/cost/models endpoints, WAL concurrency, OAuth scaffold, push subscriptions) --- cloud/README.md | 90 +++-- cloud/db.py | 135 +++++++ cloud/oauth.py | 461 +++++++++++++++++++++ cloud/push.py | 360 +++++++++++++++++ cloud/server.py | 552 +++++++++++++++++++++++++- cloud/tests/test_db.py | 249 ++++++++++++ cloud/tests/test_oauth.py | 275 +++++++++++++ cloud/tests/test_push.py | 304 ++++++++++++++ cloud/tests/test_server_endpoints.py | 225 +++++++++++ cloud/tests/test_server_oauth_push.py | 177 +++++++++ docs/cloud-architecture.md | 31 ++ 11 files changed, 2831 insertions(+), 28 deletions(-) create mode 100644 cloud/db.py create mode 100644 cloud/oauth.py create mode 100644 cloud/push.py create mode 100644 cloud/tests/test_db.py create mode 100644 cloud/tests/test_oauth.py create mode 100644 cloud/tests/test_push.py create mode 100644 cloud/tests/test_server_endpoints.py create mode 100644 cloud/tests/test_server_oauth_push.py diff --git a/cloud/README.md b/cloud/README.md index d9f97d1..a018b4d 100644 --- a/cloud/README.md +++ b/cloud/README.md @@ -12,27 +12,16 @@ webhook）。零第三方依赖，只用 Python 标准库。 ``` 客户端 / GitHub / Slack - │ HTTP + X-API-Key（webhook 靠签名，见下） + │ HTTP + X-API-Key（webhook 靠签名，OAuth 靠 state，见下） ▼ cloud/server.py（控制面，ThreadingHTTPServer） - ├─ 项目 / 任务：建、查、取消、心跳、stale 回收 + ├─ 项目 / 任务：建、查、取消、心跳、stale 回收、任务成本 + ├─ 记忆 / 知识库 / 成本 / 模型：已接成 HTTP 端点（见 API 表） + ├─ OAuth：cloud/oauth.py（state + 授权 URL + 回调，脚手架） + ├─ 推送订阅与分发：cloud/push.py（订阅表 + 偏好过滤） ├─ 认证 / 权限 / 审计：cloud/auth.py（AuthStore） + ├─ 并发：cloud/db.py（WAL + busy_timeout + 有限重试） └─ webhook 入站：cloud/webhooks.py（验签 + 解析，只解析不执行） - │ （后续由 worker 接线） - ▼ -worker（尚未接成 HTTP 端点，本版需另行调用模块） - ├─ 派发顺序：cloud/queue.py（order_tasks / ready_tasks / - │ budget_exceeded / TaskQueue；状态机与 host.py 对齐） - ├─ 执行：cloud/sandbox.py（LocalExecutor 调 host.py CLI 做只读 - │ 封装；DockerExecutor / CloudExecutor 仍是占位，未接通） - ├─ 记忆：cloud/memory.py（MemoryStore：任务摘要，供 prompt 上下文） - ├─ 知识库：cloud/kb.py（KBStore：按项目隔离的文档检索） - ├─ 模型路由：cloud/models.py（ModelRouter / record_fallback： - │ 只决定 fallback，不跑模型回合） - ├─ 成本：cloud/cost.py（CostStore / fetch_account_usage； - │ usage 来自调用方注入的 client，本模块不联网） - └─ 通知：cloud/webhooks.py（NotificationService：webhook + Slack - 占位分发；Web Push / APNs / FCM 明确未接，需要证书才做） ``` 状态库：单一 SQLite 文件。AuthStore 用 `users` / `project_members` / @@ -82,10 +71,44 @@ user_id, api_key_plain = store.create_user("alice") ## API 列表 -全部返回 JSON，`Content-Type: application/json`。除 `GET /health` -与 `/webhooks/*` 外，所有端点要求请求头 `X-API-Key`，无 key 或 key -错返回 `401 {"error": "unauthorized"}`。`/webhooks/*` 不走 X-API-Key， -因为外部服务无法持本站 API key，改靠各自的签名认证（见下）。 +全部返回 JSON，`Content-Type: application/json`。除 `GET /health`、 +`/webhooks/*`、`/oauth/*` 外，所有端点要求请求头 `X-API-Key`，无 key +或 key 错返回 `401 {"error": "unauthorized"}`。`/webhooks/*` 不走 +X-API-Key，因为外部服务无法持本站 API key，改靠各自的签名认证。 +`/oauth/*` 不走 X-API-Key，因为 OAuth 登录的起点就是用户还没有 +本站 API key——它的认证靠随机 state（10 分钟过期、一次性）加 +client_secret（只走环境变量），不是无认证后门。 + +### OAuth 接线步骤（脚手架，已实现，未接真实应用） + +1. 注册真实 OAuth 应用（GitHub 或 Google），拿到 client_id， + 回调地址填本站的 `/oauth/{provider}/callback` 公网地址。 +2. `GET /oauth/{provider}/login?client_id=...&redirect_uri=...` 拿 + authorization_url，把用户送到 provider 授权页。 +3. provider 回调本站并带回 code + state，调 + `POST /oauth/{provider}/callback`，验 state → 换 token → 映射 + 本站用户并记审计。client_secret 只从环境变量读 + （`GITHUB_OAUTH_CLIENT_SECRET` / `GOOGLE_OAUTH_CLIENT_SECRET`）， + 绝不写文件、绝不进 URL、换 token 前未配好就明确报错。 +4. 仍缺：真实应用注册、回调公网地址、已有本站账号的绑定流程 + （目前是确定性用户名新建/复用，不做账号绑定）。 + +### 推送凭证清单 + +- webhook / slack：需要可注入 sender（worker 侧），HTTP 层没有 + sender 时明确报错，不静默成功。 +- webpush：需要 VAPID 密钥对（私钥绝不进仓库）。 +- apns：需要 APNs 证书。 +- fcm：需要 FCM key。 +以上三者目前只有接口 + 缺凭证报错，没有真实发送端。 + +### SQLite 并发上限 + +server 的连接统一走 `cloud/db.py`：WAL 模式、busy_timeout 5000ms、 +写锁冲突有限重试 3 次（退避 0.05s/0.1s/0.2s，重试耗尽抛原始错误， +绝不无限循环）。多线程并发写测试已覆盖。但这只是单机缓解： +写仍是单写者串行，写密集或跨机器时必须换 Postgres，这是下一阶段， +不是本轮假装解决。 | 方法与路径 | 权限 | 说明 | | --- | --- | --- | @@ -98,6 +121,20 @@ user_id, api_key_plain = store.create_user("alice") | `POST /tasks/{id}/cancel` | 该项目 `create_task` | `queued` / `running` 改为 `interrupted` 并记审计 `cancel_task`；已是终态（completed / failed / interrupted / merged / merge-failed）返回 `400`；不存在 `404` | | `POST /tasks/{id}/heartbeat` | 该项目 `create_task` | 更新 `last_heartbeat = now`；状态为 `queued` 则改为 `running`；返回任务 | | `POST /admin/sweep` | 认证即可 | 调 `sweep_stale_tasks(db_path, threshold_seconds=300)`（body 可传 `threshold_seconds` 覆盖默认值）；把 `running` 且 `last_heartbeat`（无心跳则用 `updated_at`）早于 `now - threshold` 的任务改为 `interrupted` 并记审计 `sweep_interrupted`；返回 `{"swept": [ids]}` | +| `POST /tasks/{id}/cost` | 该任务所属项目 `read` | 成本聚合：`{task_id, project_id, total_tokens, usd_micros, turns, fallback_count, actual_tokens, estimated_tokens}`；真数与估算分开；不存在 `404`，越权 `403` | +| `POST /projects/{id}/memory` | 该项目 `create_task` | 写线程摘要 body `{thread_id?, goal, conclusion, files_changed?, open_issues?}`，返回 `201 {id}`，并记审计 `create_memory` | +| `GET /projects/{id}/memory?top_k=5` | 该项目 `read` | 最近 active 摘要（默认 top_k=5）+ `prompt_context`（只含 top_k，不全量注入）；不存在 `404`，越权 `403` | +| `POST /projects/{id}/memory/{mid}/supersede` | 该项目 `create_task` | 把摘要标废弃（deprecated，不是物理删除），并记审计 `supersede_memory`；摘要不属于该项目则 `404` | +| `POST /projects/{id}/kb/index` | 该项目 `create_task` | 索引文档 body `{path, content, kind?}`，同项目同 path 重复索引会替换旧文档，返回 `201 {id}` | +| `GET /projects/{id}/kb/search?q=&top_k=5` | 该项目 `read` | FTS + LIKE 回退，仍按项目过滤，跨项目文档绝不返回；`q` 缺失 `400` | +| `GET /projects/{id}/cost` | 该项目 `read` | 项目成本总览 + 按天：`?day=YYYY-MM-DD` 查单天，`?days=N` 查最近 N 天；无参数只给总览 | +| `GET /models` | 认证即可 | 优先级列表（环境变量 `MODELS_PRIORITY` 逗号分隔，未设用默认）+ 只含调用者成员项目的模型统计与 fallback 事件（`fallback_count > 0` 的任务）；不返回 prompt 内容 | +| `GET /oauth/{provider}/login?client_id=&redirect_uri=&scope=` | 无需 X-API-Key（pre-auth） | 返回 `{provider, state, authorization_url}`，只返回 URL 不跳转；provider 仅 github/google，未知 `400`；state 已入库，10 分钟过期 | +| `POST /oauth/{provider}/callback` | 无需 X-API-Key（pre-auth） | body `{code, state, client_id, redirect_uri}`；验 state → 换 token → 映射到本站用户（复用 users 表，用户名 `oauth::`）并记审计 `oauth_login`；state 无效/过期、换 token 失败、无 secret 一律 `400` 且不建会话；成功不返回 API key（发放方式由调用方另行决定） | +| `POST /projects/{id}/push/subscriptions` | 该项目 `create_task` | 记订阅 body `{channel, target, preference?}`，channel 为 webhook/slack/webpush/apns/fcm，preference 为 instant/digest/silent，返回 `201` | +| `GET /projects/{id}/push/subscriptions` | 该项目 `read` | 只列调用者自己的订阅（不泄露他人 target） | +| `DELETE /projects/{id}/push/subscriptions/{sid}` | 该项目 `create_task` | 只能删自己的订阅，项目 owner 可删任意；不存在 `404`，他人订阅 `403` | +| `POST /admin/push/dispatch` | 该项目 `read` | body `{event_type, project_id, payload?}`；只发状态变化事件（completed/failed/goal-unmet/merge-failed/review-requested）；HTTP 层无注入 sender：webhook/slack 报“无 sender”，webpush/apns/fcm 报缺凭证，都不静默成功；真实发送端由 worker 注入 sender 后调 `push.dispatch_event` | | `POST /webhooks/github` | GitHub 签名 | 见“webhook 端点” | | `POST /webhooks/slack` | Slack 签名 | 见“webhook 端点” | @@ -184,8 +221,9 @@ interrupted 且终态任务 cancel 返回 400、心跳后 sweep 不扫刚心跳 Secure Vault 注入、或其它经用户明确同意的渠道），而不是把 key 写进配置文件、数据库或代码里凑合。定下之前，定时自动化与 worker 的模型回合都只能停在“失败路径可用、模型跑不起来”的状态。 -- **其它待接线（非外部条件，但同样未完成）**：memory / KB / 成本 / - 模型路由尚未接成 HTTP 端点；OAuth（GitHub/Google）未接，首版只有 - API key + 哈希认证；Web Push / APNs / FCM 推送未接，需要证书才做； - 单机 SQLite 在高并发写下仍有 busy 锁上限，生产量上来后需评估 - 换库或单写者队列。 +- **其它待接线（非外部条件，但同样未完成）**：OAuth 已接成脚手架 + 端点（state + 授权 URL + 回调），但真实应用注册、回调公网地址、 + 已有账号绑定仍缺；推送已接成订阅 + 分发端点，但 Web Push / + APNs / FCM 的真实凭证与发送端仍缺；单机 SQLite 在高并发写下 + 仍有上限，生产量上来后需换 Postgres；worker 执行、云端执行、 + 真实通知与真实模型回合均未接通。 diff --git a/cloud/db.py b/cloud/db.py new file mode 100644 index 0000000..b15ceb1 --- /dev/null +++ b/cloud/db.py @@ -0,0 +1,135 @@ +"""SQLite 并发 helper（云端控制面单机阶段）。 + +只解决一件事：ThreadingHTTPServer 多线程 + 单个 SQLite 文件时的写冲突。 +不碰任何业务逻辑、不定义新 schema、不改 server.py 的接线： +本模块只提供连接与读写 helper，协调员后续把 server 的 _connect +换成这里的 connect 即可（表名/字段与 server.py 的 init_db、 +memory.py 的 summaries 完全一致，因为本模块根本不建表）。 + +使用约定（必须遵守，否则 helper 帮不了你）： +- 每个线程用自己的连接：绝不在多线程间共用一个 sqlite3 连接 + （sqlite3 默认 check_same_thread=True，共用会直接报错）。 + 典型用法是每请求/每线程调用一次 connect()，用完 close()。 +- 写操作走 execute_with_retry + transaction：一次语句失败重试是 + 安全的（单条 INSERT/UPDATE 被锁时 SQLite 已回滚该语句）； + 不要把多条非幂等语句包在一个重试里——重试只针对单条语句。 + +单机 SQLite 的上限（诚实边界，见模块末尾注释）： +- WAL + busy_timeout + 有限重试只能缓解单机多线程写冲突， + 不能变成多机/高并发数据库：写仍是单写者串行，读多写少尚可， + 写密集或跨机器时必须换 Postgres（下一阶段，不在本轮假装解决）。 +""" + +import sqlite3 +import time +from contextlib import contextmanager + +# 与 server.py 旧 _connect 的 timeout=30s 保持一致的连接级等待； +# PRAGMA busy_timeout 是语句级等待，两者不冲突，后者以毫秒为单位 +# 在每个连接上单独设置（不随库文件持久化，见 connect 的注释）。 +DEFAULT_BUSY_TIMEOUT_MS = 5000 +DEFAULT_CONNECT_TIMEOUT_S = 30 +DEFAULT_MAX_RETRIES = 3 +# 指数退避：0.05s、0.1s、0.2s；重试耗尽即抛原始错误，绝不无限循环。 +DEFAULT_BACKOFF_SECONDS = (0.05, 0.1, 0.2) +_LOCKED_MARKER = "database is locked" + + +def connect(db_path, busy_timeout_ms=DEFAULT_BUSY_TIMEOUT_MS, + timeout=DEFAULT_CONNECT_TIMEOUT_S): + """返回一个已设好 WAL + busy_timeout + Row 的新连接。 + + - PRAGMA journal_mode=WAL：读不阻塞写、写不阻塞读，是多线程 + server 的前提；该模式随库文件持久化（:memory: 库除外， + 内存库的 journal_mode 恒为 memory，属 SQLite 本身行为）。 + - PRAGMA busy_timeout：写冲突时语句级等待窗口，单位毫秒； + 它是每个连接的属性，不随库文件持久化，所以每次 connect 都设。 + - row_factory=sqlite3.Row：与 server.py 各 handler 在 _db() 里 + 事后设置的 row_factory 一致，行可用 row["id"] 按列名访问。 + - 不设 check_same_thread=False：跨线程共用连接是错误用法， + 保持默认 True 让误用尽早报错，而不是静默出竞态。 + """ + conn = sqlite3.connect(str(db_path), timeout=timeout) + try: + conn.execute("PRAGMA journal_mode=WAL") + conn.execute("PRAGMA busy_timeout=%d" % int(busy_timeout_ms)) + conn.row_factory = sqlite3.Row + except Exception: + conn.close() + raise + return conn + + +def is_locked_error(exc): + """只有 OperationalError 且文案含 database is locked 才算写锁冲突。 + + 其它 OperationalError（如 no such table、语法错误）重试无用， + 必须立即抛出，不能被重试掩盖成“锁问题”。 + """ + return ( + isinstance(exc, sqlite3.OperationalError) + and _LOCKED_MARKER in str(exc).lower() + ) + + +def execute_with_retry(conn, sql, params=(), + max_retries=DEFAULT_MAX_RETRIES, + backoff_seconds=DEFAULT_BACKOFF_SECONDS): + """执行单条写语句，遇写锁冲突时有限重试，返回 cursor。 + + 重试次数 = max_retries（默认 3 次，即最多尝试 1 + 3 = 4 次）， + 每次重试前按 backoff_seconds 依次 sleep（默认 0.05/0.1/0.2s， + 退避表短于重试次数时取最后一项，不越界）。 + 重试耗尽后把最后一次的原始 OperationalError 原样抛出，不吞掉、 + 不包装成别的异常类型，调用方看到的仍是 sqlite3.OperationalError。 + 非锁错误一次都不重试。 + """ + attempts = 0 + while True: + try: + return conn.execute(sql, params) + except sqlite3.OperationalError as exc: + if not is_locked_error(exc) or attempts >= max_retries: + raise + delay = backoff_seconds[ + min(attempts, len(backoff_seconds) - 1) + ] + time.sleep(delay) + attempts += 1 + + +def read_all(conn, sql, params=()): + """读操作 helper：不重试，只要求连接本身已走 WAL + busy_timeout。 + + 读在 WAL 下不阻塞写，一般不需要重试；如果调用方传的是旧式 + 连接（非本模块 connect 建的），读结果仍正确，只是没有 WAL/ + busy_timeout 保障——这属于调用方误用，本函数不静默替它建连接。 + """ + return conn.execute(sql, params).fetchall() + + +@contextmanager +def transaction(conn): + """单连接事务上下文：成功 commit，异常 rollback 后原样抛出。 + + 与 execute_with_retry 配合的典型写法:: + + conn = connect(db_path) + try: + with transaction(conn): + execute_with_retry(conn, "INSERT INTO tasks(...) ...", (...)) + finally: + conn.close() + + 注意：重试发生在语句级（transaction 内部），事务本身不重试； + 若事务内某条语句最终抛错，整个事务回滚，不会留下半写状态。 + """ + try: + yield conn + conn.commit() + except Exception: + try: + conn.rollback() + except Exception: + pass + raise diff --git a/cloud/oauth.py b/cloud/oauth.py new file mode 100644 index 0000000..5c04d6d --- /dev/null +++ b/cloud/oauth.py @@ -0,0 +1,461 @@ +"""OAuth 接线脚手架（GitHub / Google，首版）。 + +诚实边界：这只是脚手架，不是已接通的真实登录。 +- 本模块不向真实 GitHub / Google 发任何请求：默认的 HTTP 路径 + （``_urllib_post``）只在调用方真的传入真实 code 且配好 + client_secret 时才会被走到；单测一律用注入的 fake http 客户端。 +- 本模块不硬编码任何真实 client_id / client_secret：client_id 由 + 调用方以参数传入，client_secret 只从环境变量读 + （GitHub：``GITHUB_OAUTH_CLIENT_SECRET``， + Google：``GOOGLE_OAUTH_CLIENT_SECRET``），绝不写文件、绝不 + 硬编码、绝不出现在授权 URL 里。 +- 本模块不存储 access_token：回调成功后只返回外部身份信息， + access_token 在提取身份后即丢弃。如后续需要用 token 调 provider + API，需另设计加密存储，本模块不提供。 +- 本模块不创建本站会话：回调成功返回的只是外部身份字典， + 映射到本站用户、建会话、记审计由 server 层用 ``auth.AuthStore`` + 完成（见 ``handle_oauth_callback`` 文档里的 server 层示例）。 + 附带的 ``map_external_identity_to_local_user`` 只复用 auth.py 已有 + 的 users 表（按确定性用户名查找/创建），不发明新的身份表， + 也不发明新的审计格式。 + +流程（标准 OAuth2 授权码流程的接线点）： + state 生成（OAuthStateStore.create_state，存库，默认 10 分钟过期） + -> 授权 URL（build_authorization_url / start_oauth_login） + -> 用户在 provider 处授权，provider 回调本站并带回 code + state + -> 回调（handle_oauth_callback）：验 state -> 换 token + -> 取外部身份 -> 返回身份字典（server 层再映射 + 记审计） + +零第三方依赖，只用 Python 标准库。 +""" + +import json +import os +import secrets +import sqlite3 +import time +import urllib.parse +import urllib.request +from urllib.parse import parse_qs + +# state 默认有效期：10 分钟。 +STATE_TTL_SECONDS = 600 + +# 各 provider 的公开端点与默认 scope、client_secret 环境变量名。 +# 这些是 OAuth 协议的公开端点地址，不是本站的任何凭证；真实应用 +# 的 client_id / client_secret 仍需另行注册并经环境变量供给。 +PROVIDERS = { + "github": { + "authorize_url": "https://github.com/login/oauth/authorize", + "token_url": "https://github.com/login/oauth/access_token", + "default_scope": "read:user", + "client_secret_env": "GITHUB_OAUTH_CLIENT_SECRET", + }, + "google": { + "authorize_url": "https://accounts.google.com/o/oauth2/v2/auth", + "token_url": "https://oauth2.googleapis.com/token", + "default_scope": "openid email profile", + "client_secret_env": "GOOGLE_OAUTH_CLIENT_SECRET", + }, +} + + +class OAuthError(Exception): + """OAuth 流程失败：state 无效/过期、换 token 失败、身份取不到等。 + + ``reason`` 供调用方区分（例如记日志或返回不同的错误提示）， + 取值：``unknown_provider`` / ``state_not_found`` / ``state_expired`` / + ``state_mismatch`` / ``client_secret_missing`` / ``token_exchange_failed`` + / ``identity_missing``。 + """ + + def __init__(self, message, reason=None): + super().__init__(message) + self.reason = reason + + +def _provider_config(provider): + """取 provider 配置，未知 provider 抛 OAuthError（不猜端点）。""" + config = PROVIDERS.get(provider) + if config is None: + raise OAuthError( + "unknown provider: %r (supported: %s)" + % (provider, ", ".join(sorted(PROVIDERS))), + reason="unknown_provider", + ) + return config + + +class OAuthStateStore: + """基于 SQLite 的 OAuth state 存储。 + + 表 oauth_states：state（主键，随机串）、provider、redirect_uri、 + created_at（epoch 秒）。与 AuthStore 的 users / project_members / + audit_log 表名不冲突，可共用同一个 db 文件。 + """ + + def __init__(self, db_path): + self.db_path = str(db_path) + self.conn = sqlite3.connect(self.db_path) + self.conn.execute( + "CREATE TABLE IF NOT EXISTS oauth_states(" + "state TEXT PRIMARY KEY, " + "provider TEXT, " + "redirect_uri TEXT, " + "created_at REAL)" + ) + self.conn.commit() + + def create_state(self, provider, redirect_uri, ttl_seconds=STATE_TTL_SECONDS): + """生成并入库一个随机 state，返回 state 字符串。 + + ``ttl_seconds`` 只影响后续 ``validate_state`` 的过期判定， + 本函数不把过期时间单独存一列：判定统一用 + ``created_at + ttl_seconds``（validate 时传入的 ttl 须与 + 创建时一致，默认两边都是 600 秒）。 + """ + _provider_config(provider) + if not redirect_uri or not isinstance(redirect_uri, str): + raise ValueError("redirect_uri required") + state = secrets.token_urlsafe(32) + self.conn.execute( + "INSERT INTO oauth_states(state, provider, redirect_uri, created_at)" + " VALUES(?, ?, ?, ?)", + (state, provider, redirect_uri, time.time()), + ) + self.conn.commit() + return state + + def validate_state(self, state, provider, redirect_uri=None, + ttl_seconds=STATE_TTL_SECONDS, now=None): + """校验 state，成功返回记录 dict，失败返回 None（绝不抛错）。 + + 校验规则（任一不满足即拒绝）： + - state 在库里存在，否则 None（不存在）； + - 记录的 provider 与传入一致，否则 None（不匹配，且不消耗 + 该 state，避免一次错 provider 的尝试烧掉合法 state）； + - redirect_uri 若传入，须与记录一致，否则 None（不匹配）； + - ``now``（默认 time.time()）未超过 + ``created_at + ttl_seconds``，否则删除该行并返回 None（过期）。 + + 成功是“一次性”的：返回前删除该行，同一 state 第二次校验 + 必然失败，防止重放。``now`` 参数只为单测注入时间， + 生产调用传默认即可。 + """ + if not state: + return None + if now is None: + now = time.time() + row = self.conn.execute( + "SELECT state, provider, redirect_uri, created_at" + " FROM oauth_states WHERE state = ?", + (state,), + ).fetchone() + if row is None: + return None + stored_state, stored_provider, stored_redirect_uri, created_at = row + if stored_provider != provider: + return None + if redirect_uri is not None and stored_redirect_uri != redirect_uri: + return None + if now > created_at + ttl_seconds: + self.conn.execute( + "DELETE FROM oauth_states WHERE state = ?", (state,)) + self.conn.commit() + return None + self.conn.execute( + "DELETE FROM oauth_states WHERE state = ?", (state,)) + self.conn.commit() + return { + "state": stored_state, + "provider": stored_provider, + "redirect_uri": stored_redirect_uri, + "created_at": created_at, + } + + +def build_authorization_url(provider, client_id, redirect_uri, scope=None, + state=None): + """拼授权 URL（纯字符串拼接，不发任何请求、不含任何 secret）。 + + 参数： + - provider：``"github"`` 或 ``"google"``，未知抛 OAuthError； + - client_id：调用方传入的真实应用 id，本模块不硬编码、不校验 + 其真实性（注册见模块文档“真实 OAuth 仍缺什么”）； + - redirect_uri：须与在 provider 处注册的回调地址一致； + - scope：不传则用该 provider 的默认 scope（GitHub ``read:user``， + Google ``openid email profile``）； + - state：不传则现场随机生成一个（调用方若要校验，必须自己 + 用 OAuthStateStore 存下它；推荐直接用 ``start_oauth_login``， + 它把生成 + 入库 + 拼 URL 一次做完）。 + + 返回的 URL 只含 client_id / redirect_uri / scope / state（Google + 另含 response_type=code），绝不含 client_secret。 + """ + config = _provider_config(provider) + if not client_id or not isinstance(client_id, str): + raise ValueError("client_id required") + if not redirect_uri or not isinstance(redirect_uri, str): + raise ValueError("redirect_uri required") + if state is None: + state = secrets.token_urlsafe(32) + params = { + "client_id": client_id, + "redirect_uri": redirect_uri, + "scope": scope or config["default_scope"], + "state": state, + } + if provider == "google": + params["response_type"] = "code" + return config["authorize_url"] + "?" + urllib.parse.urlencode(params) + + +def start_oauth_login(state_store, provider, client_id, redirect_uri, + scope=None): + """登录起点：生成并入库 state，再拼授权 URL。 + + 返回 ``(state, authorization_url)``：调用方把用户重定向到该 URL， + provider 回调时带回的 state 交给 ``handle_oauth_callback`` 校验。 + """ + state = state_store.create_state(provider, redirect_uri) + url = build_authorization_url( + provider, client_id, redirect_uri, scope=scope, state=state) + return state, url + + +def _urllib_post(token_url, payload): + """默认的真实换 token 实现（urllib POST，返回解析后的 dict）。 + + 只在调用方未注入 http_client 且配好 client_secret 时走到； + GitHub 的 token 端点默认回表单编码，本函数先试 JSON、 + 失败再按表单解析，两种都解析不出则返回空 dict（由调用方 + 判为换 token 失败）。 + """ + data = urllib.parse.urlencode(payload).encode("utf-8") + request = urllib.request.Request( + token_url, data=data, method="POST", + headers={ + "Accept": "application/json", + "Content-Type": "application/x-www-form-urlencoded", + }, + ) + with urllib.request.urlopen(request, timeout=30) as response: + raw = response.read().decode("utf-8", "replace") + try: + parsed = json.loads(raw) + if isinstance(parsed, dict): + return parsed + except ValueError: + pass + form = parse_qs(raw) + return {key: values[0] for key, values in form.items() if values} + + +def exchange_code_for_token(provider, code, client_id, redirect_uri, + http_client=None, client_secret=None): + """用授权码换 token，返回 provider 的 token 响应 dict。 + + http_client 注入约定（单测与未来真实接线共用同一签名）： + ``http_client(token_url, payload_dict) -> dict``，其中 payload 含 + client_id / code / redirect_uri，Google 另含 + grant_type=authorization_code；client_secret 仅在有值时才放进 + payload。返回的 dict 须含 ``access_token``，否则视为换 token 失败； + http_client 抛异常、或返回含 ``error`` 的 dict，同样视为失败。 + + client_secret 供给：参数优先，未传则读该 provider 对应的环境变量 + （见 PROVIDERS 的 client_secret_env）。真实路径（http_client + 为 None）下仍无 secret 时，在发任何请求前就抛 OAuthError， + 绝不以空 secret 去撞真实端点。注入了 http_client 的测试路径 + 允许 secret 缺省（fake 不需要真实 secret），生产调用不得依赖 + 这一宽松分支。 + """ + config = _provider_config(provider) + if not code or not isinstance(code, str): + raise OAuthError("code required", reason="token_exchange_failed") + if client_secret is None: + client_secret = os.environ.get(config["client_secret_env"]) + if http_client is None and not client_secret: + raise OAuthError( + "client_secret not configured: set %s (never hard-code it)" + % config["client_secret_env"], + reason="client_secret_missing", + ) + payload = { + "client_id": client_id, + "code": code, + "redirect_uri": redirect_uri, + } + if provider == "google": + payload["grant_type"] = "authorization_code" + if client_secret is not None: + payload["client_secret"] = client_secret + client = http_client if http_client is not None else _urllib_post + try: + result = client(config["token_url"], payload) + except Exception as exc: + raise OAuthError( + "token exchange failed: %s" % (exc,), reason="token_exchange_failed" + ) from exc + if not isinstance(result, dict) or result.get("error"): + raise OAuthError( + "token exchange failed: %r" % (result,), + reason="token_exchange_failed", + ) + if not result.get("access_token"): + raise OAuthError( + "token exchange returned no access_token", + reason="token_exchange_failed", + ) + return result + + +def _identity_from_token_or_userinfo(provider, token_result, userinfo_client): + """从 token 响应或 userinfo 回调里抽外部身份。 + + 顺序：先看 token 响应自带的身份字段（fake / 某些 provider 会带 + id、login、email），没有则调 userinfo_client(access_token)； + 两边都没有可用身份时返回 None。真实 GitHub 的 token 响应不带 + 身份，生产调用必须传 userinfo_client（例如调 + https://api.github.com/user 并取 login/id）；真实 Google 的 + id_token 解析同样走 userinfo_client 这一注入点，本模块不解析 JWT。 + """ + for source in (token_result,): + external_id = source.get("id") or source.get("sub") + login = source.get("login") or source.get("name") + email = source.get("email") + if external_id is not None or login is not None or email is not None: + return { + "provider": provider, + "external_id": str(external_id) if external_id is not None else None, + "login": login, + "email": email, + } + if userinfo_client is None: + return None + try: + info = userinfo_client(token_result["access_token"]) + except Exception: + return None + if not isinstance(info, dict): + return None + external_id = info.get("id") or info.get("sub") + login = info.get("login") or info.get("name") + email = info.get("email") + if external_id is None and login is None and email is None: + return None + return { + "provider": provider, + "external_id": str(external_id) if external_id is not None else None, + "login": login, + "email": email, + } + + +def handle_oauth_callback(state_store, provider, code, state, client_id, + redirect_uri, http_client=None, userinfo_client=None, + client_secret=None, now=None, + ttl_seconds=STATE_TTL_SECONDS): + """处理 provider 回调：验 state -> 换 token -> 返回外部身份。 + + 成功返回 dict：``{"provider", "external_id", "login", "email"}`` + （缺失的字段为 None；external_id 至少与 login/email 其一存在， + 否则抛 OAuthError）。 + + 失败一律抛 OAuthError，不建任何会话、不写任何用户或审计记录： + - state 不存在 -> reason=state_not_found； + - state 过期 -> reason=state_expired（过期行在拒绝时被删除）； + - state 与 provider / redirect_uri 不匹配 -> + reason=state_mismatch； + - 换 token 失败 -> reason=token_exchange_failed； + - 拿不到外部身份 -> reason=identity_missing。 + + 本函数 deliberately 不做以下三件事，全部留给 server 层（签名 + 与 auth.py 完全兼容，不发明新格式）： + 1. 映射到本站用户：可用 ``map_external_identity_to_local_user`` + （复用 AuthStore 的 users 表）或 server 层自己的映射； + 2. 建会话：本站首版会话就是 AuthStore 的 API key，OAuth 登录后 + 是否发 key、发给谁，由 server 层决定，本模块不发； + 3. 记审计：server 层在映射出本站 user_id 后调用 + ``AuthStore.audit(user_id, project_id, "oauth_login", + "%s:%s" % (provider, external_id))``——与 auth.py 现有 + audit(action, detail) 写法一致，project_id 为登录发生的 + 项目上下文、无项目上下文时 server 层自定（例如空串）， + 本模块不替 server 层选 project_id。 + + access_token 在本函数内只用于取身份（经 userinfo_client）， + 返回前即丢弃，本模块不存储、不返回它。 + """ + # 先区分拒绝原因（validate_state 本身只返回记录或 None）： + # 不存在 / provider 或 redirect_uri 不匹配 / 过期各有 reason； + # 过期行在此处删除（validate_state 的过期分支同样会删，幂等）。 + if now is None: + now = time.time() + peek = state_store.conn.execute( + "SELECT provider, redirect_uri, created_at" + " FROM oauth_states WHERE state = ?", + (state,), + ).fetchone() if state else None + if peek is None: + raise OAuthError("unknown state", reason="state_not_found") + peek_provider, peek_redirect_uri, peek_created_at = peek + if peek_provider != provider or ( + redirect_uri is not None and peek_redirect_uri != redirect_uri): + raise OAuthError("state does not match provider/redirect_uri", + reason="state_mismatch") + if now > peek_created_at + ttl_seconds: + state_store.conn.execute( + "DELETE FROM oauth_states WHERE state = ?", (state,)) + state_store.conn.commit() + raise OAuthError("state expired", reason="state_expired") + record = state_store.validate_state( + state, provider, redirect_uri=redirect_uri, now=now, + ttl_seconds=ttl_seconds) + if record is None: + raise OAuthError( + "invalid state", reason="state_not_found") + token_result = exchange_code_for_token( + provider, code, client_id, redirect_uri, + http_client=http_client, client_secret=client_secret) + identity = _identity_from_token_or_userinfo( + provider, token_result, userinfo_client) + if identity is None: + raise OAuthError( + "cannot determine external identity: token response carries" + " no identity fields and no userinfo_client was provided", + reason="identity_missing", + ) + if identity["external_id"] is None and identity["login"] is None \ + and identity["email"] is None: + raise OAuthError( + "empty external identity", reason="identity_missing") + return identity + + +def map_external_identity_to_local_user(auth_store, provider, identity): + """把外部身份映射到本站用户（复用 AuthStore 的 users 表）。 + + 确定性用户名：``oauth::``。 + 已存在则返回 ``(user_id, False)``，不存在则经 + ``AuthStore.create_user`` 创建并返回 ``(user_id, True)``—— + 创建时生成的 API key 明文按 auth.py 约定只应返回给调用方一次， + 本函数不返回它（调用方如需发 key 给登录用户，须另行决定 + 发放方式，本模块不代发）。这只是脚手架的默认映射：生产若要 + 把外部身份绑定到已有本站账号（而非每次新建），需另加绑定表 + 与绑定流程，本函数不假装已做绑定。 + """ + _provider_config(provider) + login = identity.get("login") or identity.get("external_id") \ + or identity.get("email") + if login is None: + raise OAuthError( + "identity has no login/external_id/email to map from", + reason="identity_missing", + ) + username = "oauth:%s:%s" % (provider, login) + row = auth_store.conn.execute( + "SELECT id FROM users WHERE username = ?", (username,) + ).fetchone() + if row is not None: + return row[0], False + user_id, _api_key_plain = auth_store.create_user(username) + return user_id, True diff --git a/cloud/push.py b/cloud/push.py new file mode 100644 index 0000000..2ce9d63 --- /dev/null +++ b/cloud/push.py @@ -0,0 +1,360 @@ +"""推送订阅与分发模块（云端架构实施）。 + +零第三方依赖，只用 Python 标准库。本模块只做两件事： + +1. 推送订阅的增删查（SQLite 表 ``push_subscriptions``）。 +2. 状态变化事件的分发（只发状态变化，不发进行中状态）。 + +边界（不要超出本模块宣称的范围）： + +- 合法事件类型与 ``webhooks.ALLOWED_NOTIFICATION_EVENTS`` 一致： + ``completed``、``failed``、``goal-unmet``、``merge-failed``、 + ``review-requested``。其他事件类型拒绝或忽略，并写明原因， + 绝不静默丢弃。 +- 偏好过滤：``silent``（静默）不发；失败类事件（``failed``、 + ``goal-unmet``、``merge-failed``）即使偏好是 ``digest``（摘要） + 也即时发；非失败类（``completed``、``review-requested``）在 + ``digest`` 偏好下本次不立即发，返回结果里明确列入 + ``skipped_for_digest``（待摘要列表），不是静默丢弃。 +- 发送端可注入：``webhook`` / ``slack`` 走可注入的 ``sender``。 + ``sender`` 为 ``None`` 时对这两个通道也返回“无 sender”错误， + 不静默成功。``sender`` 可以是可调用对象（收到一个 dict，见 + :func:`dispatch_event`），也可以是按通道分的 dict + （``{"webhook": callable, "slack": callable}``），或带 ``send`` + 方法的对象。 +- ``webpush`` / ``apns`` / ``fcm`` 只有接口，没有真实发送端： + 一律明确报错“缺凭证（VAPID / APNs 证书 / FCM key）”，不编造 + 真实推送、不生成或提交任何私钥到仓库、不向真实推送服务发 + 任何请求。真实凭证与真实发送端是后续待接线的工作，见模块 + 末尾的缺口说明。 +- 本模块不联网、不调用真实推送服务、不跑真实模型回合。 + 密钥、token、证书一律不由本模块读取、写入或返回。 +""" + +import sqlite3 +import time +import uuid +from contextlib import contextmanager + +__all__ = [ + "CHANNELS", + "PREFERENCES", + "ALLOWED_PUSH_EVENTS", + "FAILURE_EVENTS", + "MISSING_CREDENTIAL_REASONS", + "init_push_db", + "create_subscription", + "delete_subscription", + "list_subscriptions", + "dispatch_event", +] + +CHANNELS = ("webhook", "slack", "webpush", "apns", "fcm") +PREFERENCES = ("instant", "digest", "silent") + +# 只发状态变化的事件集合，与 webhooks.ALLOWED_NOTIFICATION_EVENTS 一致。 +ALLOWED_PUSH_EVENTS = frozenset( + {"completed", "failed", "goal-unmet", "merge-failed", "review-requested"} +) + +# 失败类事件：即使偏好是 digest 也即时发。 +FAILURE_EVENTS = frozenset({"failed", "goal-unmet", "merge-failed"}) + +# webpush/apns/fcm 缺凭证时的明确报错文案（不含任何真实凭证）。 +MISSING_CREDENTIAL_REASONS = { + "webpush": "missing credential: VAPID keys not configured", + "apns": "missing credential: APNs certificate not configured", + "fcm": "missing credential: FCM key not configured", +} + + +def _is_connection(db): + return isinstance(db, sqlite3.Connection) + + +@contextmanager +def _connect(db): + """按 db 类型取连接：Connection 直接用（不关闭）；路径则新建并关闭。""" + if _is_connection(db): + yield db + elif isinstance(db, (str, bytes)) or hasattr(db, "__fspath__"): + conn = sqlite3.connect(str(db), timeout=30) + try: + yield conn + finally: + conn.close() + else: + raise TypeError("db must be a sqlite3 connection or a db path") + + +def init_push_db(db): + """创建 push_subscriptions 表（已存在则不动）。""" + with _connect(db) as conn: + conn.execute( + "CREATE TABLE IF NOT EXISTS push_subscriptions(" + "id TEXT PRIMARY KEY, " + "project_id TEXT NOT NULL, " + "user_id TEXT NOT NULL, " + "channel TEXT NOT NULL, " + "target TEXT NOT NULL, " + "preference TEXT NOT NULL, " + "created_at REAL, " + "UNIQUE(project_id, user_id, channel, target))" + ) + conn.commit() + + +def _row_to_dict(row): + return { + "id": row[0], + "project_id": row[1], + "user_id": row[2], + "channel": row[3], + "target": row[4], + "preference": row[5], + "created_at": row[6], + } + + +def _validate_subscription_fields(project_id, user_id, channel, target, + preference): + if not project_id or not isinstance(project_id, str): + raise ValueError("project_id required") + if not user_id or not isinstance(user_id, str): + raise ValueError("user_id required") + if channel not in CHANNELS: + raise ValueError("channel must be one of %s" % (CHANNELS,)) + if not target or not isinstance(target, str): + raise ValueError("target required") + if preference not in PREFERENCES: + raise ValueError("preference must be one of %s" % (PREFERENCES,)) + + +def create_subscription(db, project_id, user_id, channel, target, + preference="instant"): + """创建一条推送订阅，返回订阅 dict。 + + 同一 (project_id, user_id, channel, target) 已存在时返回已有订阅， + 不建重复行。channel 取值 webhook/slack/webpush/apns/fcm； + preference 取值 instant/digest/silent；非法取值抛 ValueError。 + """ + _validate_subscription_fields( + project_id, user_id, channel, target, preference) + init_push_db(db) + with _connect(db) as conn: + existing = conn.execute( + "SELECT id, project_id, user_id, channel, target, preference," + " created_at FROM push_subscriptions" + " WHERE project_id = ? AND user_id = ? AND channel = ?" + " AND target = ?", + (project_id, user_id, channel, target), + ).fetchone() + if existing is not None: + return _row_to_dict(existing) + sub_id = uuid.uuid4().hex + now = time.time() + conn.execute( + "INSERT INTO push_subscriptions(id, project_id, user_id," + " channel, target, preference, created_at)" + " VALUES(?, ?, ?, ?, ?, ?, ?)", + (sub_id, project_id, user_id, channel, target, preference, now), + ) + conn.commit() + row = conn.execute( + "SELECT id, project_id, user_id, channel, target, preference," + " created_at FROM push_subscriptions WHERE id = ?", + (sub_id,), + ).fetchone() + return _row_to_dict(row) + + +def delete_subscription(db, subscription_id=None, *, project_id=None, + user_id=None, channel=None, target=None): + """删除订阅，返回 True（删到至少一条）或 False（无匹配）。 + + 优先按 subscription_id 删除；未传 id 时按给出的 + project_id/user_id/channel/target 过滤删除（至少要给一个过滤条件， + 否则不删任何行并返回 False，避免误删整表）。 + """ + init_push_db(db) + with _connect(db) as conn: + if subscription_id is not None: + cur = conn.execute( + "DELETE FROM push_subscriptions WHERE id = ?", + (subscription_id,), + ) + else: + conditions = [] + params = [] + for column, value in ( + ("project_id", project_id), + ("user_id", user_id), + ("channel", channel), + ("target", target), + ): + if value is not None: + conditions.append("%s = ?" % column) + params.append(value) + if not conditions: + return False + cur = conn.execute( + "DELETE FROM push_subscriptions WHERE " + + " AND ".join(conditions), + params, + ) + conn.commit() + return cur.rowcount > 0 + + +def list_subscriptions(db, project_id, user_id=None): + """列出某项目的订阅；传 user_id 时只列该用户的。 + + 返回按 created_at、id 排序的订阅 dict 列表。只查本项目， + 跨项目订阅绝不会混入（见单测）。 + """ + if not project_id or not isinstance(project_id, str): + raise ValueError("project_id required") + init_push_db(db) + with _connect(db) as conn: + if user_id is None: + rows = conn.execute( + "SELECT id, project_id, user_id, channel, target," + " preference, created_at FROM push_subscriptions" + " WHERE project_id = ? ORDER BY created_at, id", + (project_id,), + ).fetchall() + else: + rows = conn.execute( + "SELECT id, project_id, user_id, channel, target," + " preference, created_at FROM push_subscriptions" + " WHERE project_id = ? AND user_id = ?" + " ORDER BY created_at, id", + (project_id, user_id), + ).fetchall() + return [_row_to_dict(row) for row in rows] + + +def _resolve_channel_sender(sender, channel): + """从注入的 sender 里取出某通道的发送函数，取不到返回 None。""" + if sender is None: + return None + if isinstance(sender, dict): + candidate = sender.get(channel) + if candidate is None: + return None + if callable(candidate): + return candidate + send = getattr(candidate, "send", None) + return send if callable(send) else None + if callable(sender): + return sender + send = getattr(sender, "send", None) + return send if callable(send) else None + + +def dispatch_event(db, event_type, project_id, payload, sender=None): + """分发一次状态变化事件，只发本项目的订阅。 + + 参数： + db: sqlite3 连接或数据库路径。 + event_type: 合法类型见 ALLOWED_PUSH_EVENTS；其他类型拒绝， + 返回 ``{"sent": [], ..., "reason": "event not allowed"}`` + 且不调用任何 sender。 + project_id: 只查该项目的订阅，跨项目不混。 + payload: 原事件载荷（dict 或其他），原样放入发送给 sender + 的消息 dict 的 ``payload`` 字段。 + sender: 可注入发送端（见模块 docstring）。webhook/slack + 通道在 sender 为 None 时返回“无 sender”错误，不静默成功； + webpush/apns/fcm 通道不走 sender，一律返回缺凭证错误。 + + 返回 dict，键固定为:: + + { + "event_type": ..., + "project_id": ..., + "sent": [{"subscription_id", "user_id", "channel", "target"}], + "skipped_for_digest": [订阅 dict（待摘要，本次未立即发）], + "skipped_silent": [订阅 dict（静默偏好，本次未发）], + "errors": [{"subscription_id", "user_id", "channel", "target", + "reason": ...}], + } + + 非法事件类型时额外含 ``"reason": "event not allowed"``。 + sender 抛异常时记入 errors，不中断其他订阅的分发。 + """ + result = { + "event_type": event_type, + "project_id": project_id, + "sent": [], + "skipped_for_digest": [], + "skipped_silent": [], + "errors": [], + } + if event_type not in ALLOWED_PUSH_EVENTS: + result["reason"] = "event not allowed" + result["errors"] = [{ + "subscription_id": None, + "user_id": None, + "channel": None, + "target": None, + "reason": "event not allowed: %r not in %s" + % (event_type, sorted(ALLOWED_PUSH_EVENTS)), + }] + return result + if not project_id or not isinstance(project_id, str): + result["reason"] = "project_id required" + result["errors"] = [{ + "subscription_id": None, + "user_id": None, + "channel": None, + "target": None, + "reason": "project_id required", + }] + return result + + subscriptions = list_subscriptions(db, project_id) + is_failure = event_type in FAILURE_EVENTS + for sub in subscriptions: + base = { + "subscription_id": sub["id"], + "user_id": sub["user_id"], + "channel": sub["channel"], + "target": sub["target"], + } + # 偏好过滤：silent 永远不发；digest 只对非失败类延迟。 + if sub["preference"] == "silent": + result["skipped_silent"].append(sub) + continue + if sub["preference"] == "digest" and not is_failure: + result["skipped_for_digest"].append(sub) + continue + # webpush/apns/fcm：只有接口，缺凭证，明确报错，不走 sender、 + # 不联网、不编造推送。 + if sub["channel"] in MISSING_CREDENTIAL_REASONS: + result["errors"].append( + dict(base, reason=MISSING_CREDENTIAL_REASONS[sub["channel"]])) + continue + # webhook/slack：必须有可注入 sender，否则明确报错。 + channel_sender = _resolve_channel_sender(sender, sub["channel"]) + if channel_sender is None: + result["errors"].append( + dict(base, reason="no sender for channel %s" + % sub["channel"])) + continue + message = { + "event_type": event_type, + "project_id": project_id, + "payload": payload, + "channel": sub["channel"], + "target": sub["target"], + "user_id": sub["user_id"], + "subscription_id": sub["id"], + } + try: + channel_sender(message) + except Exception as exc: # noqa: BLE001 -- 分发失败记 errors，不中断 + result["errors"].append( + dict(base, reason="sender failed: %s" % exc)) + continue + result["sent"].append(base) + return result diff --git a/cloud/server.py b/cloud/server.py index 8514d11..b1b0d86 100644 --- a/cloud/server.py +++ b/cloud/server.py @@ -45,6 +45,23 @@ from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from urllib.parse import parse_qs, urlparse from cloud.auth import AuthStore +from cloud.cost import CostStore +from cloud.db import connect as _db_connect +from cloud.kb import KBStore +from cloud.memory import MemoryStore +from cloud.oauth import ( + OAuthError, + OAuthStateStore, + handle_oauth_callback, + map_external_identity_to_local_user, + start_oauth_login, +) +from cloud.push import ( + create_subscription as _push_create, + delete_subscription as _push_delete, + dispatch_event as _push_dispatch, + list_subscriptions as _push_list, +) from cloud.webhooks import ( parse_github_pr_comment, parse_slack_command, @@ -59,8 +76,12 @@ DEFAULT_SWEEP_THRESHOLD_SECONDS = 300 def _connect(db_path): - """每个请求新建一个连接；busy timeout 给并发写留重试窗口。""" - return sqlite3.connect(str(db_path), timeout=30) + """每个请求新建一个连接；走 cloud.db 的 WAL + busy_timeout。 + + 单机 SQLite 的上限仍在：WAL + 有限重试只能缓解多线程写冲突， + 高并发要换 Postgres（见 cloud/README.md）。 + """ + return _db_connect(str(db_path)) def init_db(db_path): @@ -236,6 +257,9 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): def do_POST(self): self._dispatch("POST") + def do_DELETE(self): + self._dispatch("DELETE") + def _dispatch(self, method): parsed = urlparse(self.path) path = parsed.path @@ -246,6 +270,17 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): if path == "/health" and method == "GET": self._send_json(200, {"ok": True, "service": "cloud-control-plane"}) return + if path.startswith("/oauth/"): + # OAuth 登录/回调是 pre-auth 端点（用户还没本站 API key， + # 这正是 OAuth 要解决的问题），靠 state + client_secret + # 认证，不走 X-API-Key；不要把它当成无认证后门。 + if method == "GET" and path.endswith("/login"): + self._oauth_login(path) + elif method == "POST" and path.endswith("/callback"): + self._oauth_callback(path) + else: + self._send_json(404, {"error": "not found"}) + return if path.startswith("/webhooks/"): # webhook 靠签名认证（外部服务无法持 X-API-Key），不走 API key。 if method != "POST": @@ -268,16 +303,24 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): self._create_project(user_id) elif path == "/projects" and method == "GET": self._list_projects(user_id) + elif path == "/models" and method == "GET": + self._models_overview(user_id) + elif path.startswith("/projects/"): + self._dispatch_project(user_id, method, path, query) elif path == "/tasks" and method == "POST": self._create_task(user_id) elif path == "/tasks" and method == "GET": self._list_tasks(user_id, query) elif path == "/admin/sweep" and method == "POST": self._admin_sweep() + elif path == "/admin/push/dispatch" and method == "POST": + self._push_dispatch_endpoint(user_id) elif path.startswith("/tasks/"): rest = path[len("/tasks/"):] if "/" not in rest and method == "GET": self._get_task(user_id, rest) + elif rest.endswith("/cost") and method == "GET": + self._task_cost(user_id, rest[: -len("/cost")]) elif rest.endswith("/cancel") and method == "POST": self._cancel_task(user_id, rest[: -len("/cancel")]) elif rest.endswith("/heartbeat") and method == "POST": @@ -354,6 +397,396 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): if r["id"] in member_ids] self._send_json(200, {"projects": projects}) + # -- 项目子资源：记忆 / 知识库 / 成本 ------------------------------- + def _project_exists(self, project_id): + conn = self._db() + try: + row = conn.execute( + "SELECT id FROM projects WHERE id = ?", (project_id,) + ).fetchone() + return row is not None + finally: + conn.close() + + def _check_project(self, user_id, project_id, action): + """返回 None 表示通过；否则返回 (status, error) 先行响应。""" + if not self._project_exists(project_id): + return 404, "project not found" + with _auth_store(self.server.db_path) as store: + if not store.check_permission(user_id, project_id, action): + return 403, "forbidden" + return None + + def _dispatch_project(self, user_id, method, path, query): + rest = path[len("/projects/"):] + parts = rest.split("/") + project_id = parts[0] if parts else "" + if not project_id: + self._send_json(404, {"error": "not found"}) + return + # /projects/{id}/memory + if len(parts) == 2 and parts[1] == "memory": + if method == "POST": + err = self._check_project(user_id, project_id, "create_task") + if err: + self._send_json(err[0], {"error": err[1]}) + return + self._memory_create(user_id, project_id) + elif method == "GET": + err = self._check_project(user_id, project_id, "read") + if err: + self._send_json(err[0], {"error": err[1]}) + return + self._memory_list(project_id, query) + else: + self._send_json(404, {"error": "not found"}) + return + # /projects/{id}/memory/{mid}/supersede + if len(parts) == 4 and parts[1] == "memory" and parts[3] == "supersede": + if method != "POST": + self._send_json(404, {"error": "not found"}) + return + err = self._check_project(user_id, project_id, "create_task") + if err: + self._send_json(err[0], {"error": err[1]}) + return + self._memory_supersede(user_id, project_id, parts[2]) + return + # /projects/{id}/kb/index + if len(parts) == 3 and parts[1] == "kb" and parts[2] == "index": + if method != "POST": + self._send_json(404, {"error": "not found"}) + return + err = self._check_project(user_id, project_id, "create_task") + if err: + self._send_json(err[0], {"error": err[1]}) + return + self._kb_index(user_id, project_id) + return + # /projects/{id}/kb/search + if len(parts) == 3 and parts[1] == "kb" and parts[2] == "search": + if method != "GET": + self._send_json(404, {"error": "not found"}) + return + err = self._check_project(user_id, project_id, "read") + if err: + self._send_json(err[0], {"error": err[1]}) + return + self._kb_search(project_id, query) + return + # /projects/{id}/cost + if len(parts) == 2 and parts[1] == "cost": + if method != "GET": + self._send_json(404, {"error": "not found"}) + return + err = self._check_project(user_id, project_id, "read") + if err: + self._send_json(err[0], {"error": err[1]}) + return + self._project_cost(project_id, query) + return + # /projects/{id}/push/subscriptions + if len(parts) == 3 and parts[1] == "push" and parts[2] == "subscriptions": + if method == "POST": + err = self._check_project(user_id, project_id, "create_task") + if err: + self._send_json(err[0], {"error": err[1]}) + return + self._push_create_endpoint(user_id, project_id) + elif method == "GET": + err = self._check_project(user_id, project_id, "read") + if err: + self._send_json(err[0], {"error": err[1]}) + return + self._send_json(200, { + "subscriptions": _push_list( + self.server.db_path, project_id, user_id=user_id)}) + else: + self._send_json(404, {"error": "not found"}) + return + # /projects/{id}/push/subscriptions/{sid} DELETE + if len(parts) == 4 and parts[1] == "push" and parts[2] == "subscriptions": + if method != "DELETE": + self._send_json(404, {"error": "not found"}) + return + err = self._check_project(user_id, project_id, "create_task") + if err: + self._send_json(err[0], {"error": err[1]}) + return + subs = _push_list(self.server.db_path, project_id) + mine = [s for s in subs if s["id"] == parts[3]] + if not mine: + self._send_json(404, {"error": "subscription not found"}) + return + # 只能删自己的订阅，项目 owner 可删任意（owner 角色检查）。 + with _auth_store(self.server.db_path) as auth: + is_owner = auth.check_permission( + user_id, project_id, "manage_project") + if mine[0]["user_id"] != user_id and not is_owner: + self._send_json(403, {"error": "forbidden"}) + return + deleted = _push_delete(self.server.db_path, + subscription_id=parts[3]) + self._send_json(200, {"deleted": bool(deleted)}) + return + self._send_json(404, {"error": "not found"}) + + def _memory_create(self, user_id, project_id): + body, err = self._read_json() + if err: + self._send_json(400, {"error": err}) + return + goal = body.get("goal") + conclusion = body.get("conclusion") + if not goal or not isinstance(goal, str): + self._send_json(400, {"error": "goal required"}) + return + if not conclusion or not isinstance(conclusion, str): + self._send_json(400, {"error": "conclusion required"}) + return + store = MemoryStore(self.server.db_path) + try: + mid = store.add_summary( + project_id, + body.get("thread_id") or "", + goal, + conclusion, + files_changed=body.get("files_changed") or [], + open_issues=body.get("open_issues") or [], + ) + finally: + try: + store.conn.close() + except Exception: + pass + with _auth_store(self.server.db_path) as auth: + auth.audit(user_id, project_id, "create_memory", str(mid)) + self._send_json(201, {"id": mid, "project_id": project_id}) + + def _memory_list(self, project_id, query): + try: + top_k = int((query.get("top_k") or ["5"])[0]) + except (TypeError, ValueError): + self._send_json(400, {"error": "top_k must be an integer"}) + return + store = MemoryStore(self.server.db_path) + try: + summaries = store.list_summaries(project_id) + context = store.build_prompt_context(project_id, top_k=top_k) + finally: + try: + store.conn.close() + except Exception: + pass + self._send_json(200, { + "project_id": project_id, + "summaries": summaries[:top_k], + "prompt_context": context, + }) + + def _memory_supersede(self, user_id, project_id, mid_raw): + try: + mid = int(mid_raw) + except (TypeError, ValueError): + self._send_json(400, {"error": "id must be an integer"}) + return + conn = self._db() + try: + row = conn.execute( + "SELECT project_id FROM summaries WHERE id = ?", (mid,) + ).fetchone() + finally: + conn.close() + if row is None or row["project_id"] != project_id: + self._send_json(404, {"error": "summary not found"}) + return + store = MemoryStore(self.server.db_path) + try: + store.deprecate(mid) + finally: + try: + store.conn.close() + except Exception: + pass + with _auth_store(self.server.db_path) as auth: + auth.audit(user_id, project_id, "supersede_memory", str(mid)) + self._send_json(200, {"id": mid, "status": "deprecated"}) + + def _kb_index(self, user_id, project_id): + body, err = self._read_json() + if err: + self._send_json(400, {"error": err}) + return + path = body.get("path") + content = body.get("content") + if not path or not isinstance(path, str): + self._send_json(400, {"error": "path required"}) + return + if content is None or not isinstance(content, str): + self._send_json(400, {"error": "content required"}) + return + store = KBStore(self.server.db_path) + try: + doc_id = store.index_file( + project_id, path, content, + kind=body.get("kind") or "file") + except ValueError as e: + self._send_json(400, {"error": str(e)}) + return + finally: + try: + store.conn.close() + except Exception: + pass + with _auth_store(self.server.db_path) as auth: + auth.audit(user_id, project_id, "kb_index", path) + self._send_json(201, {"id": doc_id, "project_id": project_id, + "path": path}) + + def _kb_search(self, project_id, query): + q = (query.get("q") or [""])[0] + if not q: + self._send_json(400, {"error": "q required"}) + return + try: + top_k = int((query.get("top_k") or ["5"])[0]) + except (TypeError, ValueError): + self._send_json(400, {"error": "top_k must be an integer"}) + return + store = KBStore(self.server.db_path) + try: + results = store.search(project_id, q, top_k=top_k) + finally: + try: + store.conn.close() + except Exception: + pass + self._send_json(200, {"project_id": project_id, "query": q, + "results": results}) + + def _project_cost(self, project_id, query): + store = CostStore(self.server.db_path) + try: + overall = store.report_by_project(project_id) + conn = store._conn + conn.row_factory = sqlite3.Row + days = (query.get("days") or [None])[0] + day = (query.get("day") or [None])[0] + daily = [] + if day: + row = conn.execute( + "SELECT day, COALESCE(SUM(total_tokens),0) AS t," + " COALESCE(SUM(usd_micros),0) AS u, COUNT(*) AS c," + " COALESCE(SUM(fallback_count),0) AS f" + " FROM cost_records WHERE project_id = ? AND day = ?" + " GROUP BY day", + (project_id, day)).fetchone() + if row: + daily = [{"day": row["day"], "total_tokens": row["t"], + "usd_micros": row["u"], "turns": row["c"], + "fallback_count": row["f"]}] + elif days is not None: + try: + n = int(days) + except (TypeError, ValueError): + self._send_json(400, {"error": "days must be an integer"}) + return + from datetime import date, timedelta + cutoff = (date.today() - timedelta(days=max(n - 1, 0)) + ).isoformat() + rows = conn.execute( + "SELECT day, COALESCE(SUM(total_tokens),0) AS t," + " COALESCE(SUM(usd_micros),0) AS u, COUNT(*) AS c," + " COALESCE(SUM(fallback_count),0) AS f" + " FROM cost_records WHERE project_id = ? AND day >= ?" + " GROUP BY day ORDER BY day", + (project_id, cutoff)).fetchall() + daily = [{"day": r["day"], "total_tokens": r["t"], + "usd_micros": r["u"], "turns": r["c"], + "fallback_count": r["f"]} for r in rows] + finally: + try: + store._conn.close() + except Exception: + pass + self._send_json(200, {"project_id": project_id, "overall": overall, + "daily": daily}) + + # -- 任务成本与模型总览 ---------------------------------------------- + def _task_cost(self, user_id, task_id): + row = self._fetch_task(task_id) + if row is None: + self._send_json(404, {"error": "task not found"}) + return + task = _task_to_dict(row) + with _auth_store(self.server.db_path) as auth: + if not auth.check_permission(user_id, task["project_id"], "read"): + self._send_json(403, {"error": "forbidden"}) + return + store = CostStore(self.server.db_path) + try: + report = store.report_by_task(task_id) + finally: + try: + store._conn.close() + except Exception: + pass + self._send_json(200, {"task_id": task_id, + "project_id": task["project_id"], **report}) + + def _models_overview(self, user_id): + # 优先级列表来自环境变量 MODELS_PRIORITY（逗号分隔），未设时 + # 用默认列表；这只是控制面的默认配置，不是 provider 的承诺。 + raw = os.environ.get("MODELS_PRIORITY", "") + priority = [m.strip() for m in raw.split(",") if m.strip()] or [ + "gpt-6.1-sol", "gpt-6-sol", "gpt-5.6-sol"] + # 只统计调用者是成员的项目，避免跨项目泄露成本。 + conn = self._db() + try: + member_ids = [r["project_id"] for r in conn.execute( + "SELECT project_id FROM project_members WHERE user_id = ?", + (user_id,)).fetchall()] + finally: + conn.close() + if not member_ids: + self._send_json(200, {"priority_list": priority, + "model_stats": [], "fallback_events": []}) + return + placeholders = ",".join("?" for _ in member_ids) + store = CostStore(self.server.db_path) + try: + conn = store._conn + conn.row_factory = sqlite3.Row + rows = conn.execute( + "SELECT model, COUNT(*) AS turns," + " COALESCE(SUM(total_tokens),0) AS tokens," + " COALESCE(SUM(fallback_count),0) AS fallbacks" + " FROM cost_records WHERE project_id IN (%s)" + " GROUP BY model ORDER BY model" % placeholders, + member_ids).fetchall() + fb_rows = conn.execute( + "SELECT task_id, project_id, model, fallback_count" + " FROM cost_records WHERE project_id IN (%s)" + " AND fallback_count > 0" + " ORDER BY created_at DESC LIMIT 50" % placeholders, + member_ids).fetchall() + finally: + try: + store._conn.close() + except Exception: + pass + self._send_json(200, { + "priority_list": priority, + "model_stats": [ + {"model": r["model"], "turns": r["turns"], + "total_tokens": r["tokens"], + "fallback_count": r["fallbacks"]} for r in rows], + "fallback_events": [ + {"task_id": r["task_id"], "project_id": r["project_id"], + "model": r["model"], + "fallback_count": r["fallback_count"]} for r in fb_rows], + }) + # -- 任务 ---------------------------------------------------------- def _create_task(self, user_id): body, err = self._read_json() @@ -553,6 +986,121 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): swept = sweep_stale_tasks(self.server.db_path, threshold) self._send_json(200, {"swept": swept}) + # -- OAuth（pre-auth，靠 state + secret，不走 X-API-Key）----------- + def _oauth_provider(self, path): + parts = [p for p in path.split("/") if p] + if len(parts) != 3 or parts[0] != "oauth": + return None + return parts[1] + + def _oauth_login(self, path): + provider = self._oauth_provider(path) + if provider is None: + self._send_json(404, {"error": "not found"}) + return + query = parse_qs(urlparse(self.path).query) + client_id = (query.get("client_id") or [None])[0] + redirect_uri = (query.get("redirect_uri") or [None])[0] + scope = (query.get("scope") or [None])[0] + if not client_id or not redirect_uri: + self._send_json(400, {"error": "client_id and redirect_uri required"}) + return + store = OAuthStateStore(self.server.db_path) + try: + state, url = start_oauth_login( + store, provider, client_id, redirect_uri, scope=scope) + except OAuthError as e: + self._send_json(400, {"error": str(e), "reason": e.reason}) + return + except ValueError as e: + self._send_json(400, {"error": str(e)}) + return + finally: + try: + store.conn.close() + except Exception: + pass + # 只返回 URL，不直接跳转到任何登录页；跳转由调用方决定。 + self._send_json(200, {"provider": provider, "state": state, + "authorization_url": url}) + + def _oauth_callback(self, path): + provider = self._oauth_provider(path) + if provider is None: + self._send_json(404, {"error": "not found"}) + return + body, err = self._read_json() + if err: + self._send_json(400, {"error": err}) + return + code = body.get("code") + state = body.get("state") + client_id = body.get("client_id") + redirect_uri = body.get("redirect_uri") + if not code or not state or not client_id or not redirect_uri: + self._send_json(400, {"error": "code, state, client_id and redirect_uri required"}) + return + store = OAuthStateStore(self.server.db_path) + try: + identity = handle_oauth_callback( + store, provider, code, state, client_id, redirect_uri) + except OAuthError as e: + self._send_json(400, {"error": str(e), "reason": e.reason}) + return + finally: + try: + store.conn.close() + except Exception: + pass + with _auth_store(self.server.db_path) as auth: + local_id, created = map_external_identity_to_local_user( + auth, provider, identity) + auth.audit(local_id, "", "oauth_login", + "%s:%s" % (provider, identity.get("external_id"))) + self._send_json(200, {"user_id": local_id, "created": created, + "provider": provider, + "login": identity.get("login"), + "email": identity.get("email")}) + + # -- 推送订阅与分发 --------------------------------------------------- + def _push_create_endpoint(self, user_id, project_id): + body, err = self._read_json() + if err: + self._send_json(400, {"error": err}) + return + try: + sub = _push_create( + self.server.db_path, project_id, user_id, + body.get("channel"), body.get("target"), + body.get("preference") or "instant") + except ValueError as e: + self._send_json(400, {"error": str(e)}) + return + self._send_json(201, sub) + + def _push_dispatch_endpoint(self, user_id): + body, err = self._read_json() + if err: + self._send_json(400, {"error": err}) + return + event_type = body.get("event_type") + project_id = body.get("project_id") + if not event_type or not project_id: + self._send_json(400, {"error": "event_type and project_id required"}) + return + # 分发者须是该项目的成员（至少可读），否则不能往别人项目里推。 + with _auth_store(self.server.db_path) as auth: + if not auth.check_permission(user_id, project_id, "read"): + self._send_json(403, {"error": "forbidden"}) + return + # HTTP 层无可注入 sender：webhook/slack 会报“无 sender”， + # webpush/apns/fcm 会报缺凭证——都不静默成功。真实发送端 + # 由 worker 注入 sender 后调 push.dispatch_event。 + result = _push_dispatch( + self.server.db_path, event_type, project_id, + body.get("payload") or {}, sender=None) + self._send_json(200, result) + # -- webhooks（签名认证，不走 X-API-Key）--------------------------- def _handle_github_webhook(self): raw = self._read_body() diff --git a/cloud/tests/test_db.py b/cloud/tests/test_db.py new file mode 100644 index 0000000..1dc9a19 --- /dev/null +++ b/cloud/tests/test_db.py @@ -0,0 +1,249 @@ +"""cloud/db.py 单测：SQLite 并发 helper。 + +只测 db.py，不改 server.py / host.py / 其它模块。 +重试逻辑的覆盖方式（本文件写清楚，供协调员核对）： +- 确定性 fake 方式：FakeConn 前两次抛 database is locked、第三次 + 成功，断言重试后成功且退避序列为 0.05/0.1/0.2s；另测重试耗尽 + 抛原始错误、非锁错误一次都不重试。 +- 真实并发方式：8 线程各写 20 条（server.py 同款 tasks 表）， + 断言无错误、最终 160 条、id 无重复。 +两者都测了，前者证重试逻辑本身正确，后者证真实写冲突不再直接炸掉。 +""" + +import os +import sqlite3 +import sys +import tempfile +import threading +import time +import unittest +from pathlib import Path +from unittest import mock + +try: + from cloud import db as db_module + from cloud.db import ( + connect, + execute_with_retry, + is_locked_error, + read_all, + transaction, + ) +except ImportError: # discover -s cloud/tests 时把项目根目录补进 sys.path + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + from cloud import db as db_module + from cloud.db import ( + connect, + execute_with_retry, + is_locked_error, + read_all, + transaction, + ) + +# 与 server.py init_db 的 tasks 表定义逐字段一致（db.py 本身不建表， +# 这里复刻定义只是为了让并发测试写进同款表，证明 schema 兼容）。 +TASKS_DDL = ( + "CREATE TABLE IF NOT EXISTS tasks(" + "id TEXT PRIMARY KEY, " + "project_id TEXT, " + "prompt TEXT, " + "status TEXT, " + "idempotency_key TEXT, " + "depends_on TEXT, " + "priority INTEGER, " + "created_by TEXT, " + "created_at REAL, " + "updated_at REAL, " + "last_heartbeat REAL, " + "UNIQUE(project_id, idempotency_key))" +) + + +class FakeConn: + """受控 fake：前 fail_times 次 execute 抛锁错误，之后返回哨兵。""" + + def __init__(self, fail_times, error_text="database is locked"): + self.fail_times = fail_times + self.error_text = error_text + self.calls = 0 + + def execute(self, sql, params=()): + self.calls += 1 + if self.calls <= self.fail_times: + raise sqlite3.OperationalError(self.error_text) + return "ok-sentinel" + + +class DbHelperTestCase(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.db_path = os.path.join(self.tmp.name, "concurrency.db") + conn = connect(self.db_path) + try: + conn.execute(TASKS_DDL) + conn.commit() + finally: + conn.close() + + def tearDown(self): + self.tmp.cleanup() + + # -- 连接属性：WAL + busy_timeout + Row ------------------------------ + def test_connect_sets_wal_busy_timeout_row_factory(self): + conn = connect(self.db_path) + try: + self.assertEqual(conn.row_factory, sqlite3.Row) + journal = conn.execute( + "PRAGMA journal_mode").fetchone()[0] + self.assertEqual(journal, "wal") + busy = conn.execute("PRAGMA busy_timeout").fetchone()[0] + self.assertEqual(busy, 5000) + # Row 行为：按列名访问，与 server.py 的 _task_to_dict 兼容。 + conn.execute( + "INSERT INTO tasks(id, project_id, prompt, status)" + " VALUES(?, ?, ?, 'queued')", + ("t-row", "p1", "hello"), + ) + conn.commit() + row = read_all( + conn, "SELECT * FROM tasks WHERE id = ?", ("t-row",))[0] + self.assertEqual(row["id"], "t-row") + self.assertEqual(row["prompt"], "hello") + finally: + conn.close() + + def test_connect_custom_busy_timeout(self): + conn = connect(self.db_path, busy_timeout_ms=1234) + try: + busy = conn.execute("PRAGMA busy_timeout").fetchone()[0] + self.assertEqual(busy, 1234) + finally: + conn.close() + + # -- 锁判定 --------------------------------------------------------- + def test_is_locked_error_only_for_locked_operational_error(self): + self.assertTrue(is_locked_error( + sqlite3.OperationalError("database is locked"))) + self.assertTrue(is_locked_error( + sqlite3.OperationalError("Database Is Locked (retry)"))) + self.assertFalse(is_locked_error( + sqlite3.OperationalError("no such table: tasks"))) + self.assertFalse(is_locked_error(ValueError("database is locked"))) + + # -- 重试逻辑：fake 受控方式（确定性）-------------------------------- + def test_retry_succeeds_after_two_locked_failures(self): + fake = FakeConn(fail_times=2) + with mock.patch.object(db_module.time, "sleep") as sleep_mock: + result = execute_with_retry(fake, "INSERT ...", ()) + self.assertEqual(result, "ok-sentinel") + self.assertEqual(fake.calls, 3) # 1 次初始 + 2 次重试 + self.assertEqual( + [c.args[0] for c in sleep_mock.call_args_list], + [0.05, 0.1, 0.2][:2]) + + def test_retry_backoff_sequence_is_exponential(self): + fake = FakeConn(fail_times=3) + with mock.patch.object(db_module.time, "sleep") as sleep_mock: + execute_with_retry(fake, "INSERT ...", ()) + self.assertEqual( + [c.args[0] for c in sleep_mock.call_args_list], + [0.05, 0.1, 0.2]) + + def test_retry_exhausted_reraises_original_error(self): + fake = FakeConn(fail_times=99) + with mock.patch.object(db_module.time, "sleep"): + with self.assertRaises(sqlite3.OperationalError) as ctx: + execute_with_retry(fake, "INSERT ...", ()) + self.assertIn("database is locked", str(ctx.exception)) + # 默认 max_retries=3：1 次初始 + 3 次重试 = 4 次尝试，不多不少。 + self.assertEqual(fake.calls, 4) + + def test_non_locked_error_is_not_retried(self): + fake = FakeConn(fail_times=99, error_text="no such table: tasks") + with mock.patch.object(db_module.time, "sleep") as sleep_mock: + with self.assertRaises(sqlite3.OperationalError): + execute_with_retry(fake, "SELECT ...", ()) + self.assertEqual(fake.calls, 1) + sleep_mock.assert_not_called() + + # -- 事务：成功提交 / 失败回滚 --------------------------------------- + def test_transaction_commits_and_rolls_back(self): + conn = connect(self.db_path) + try: + with transaction(conn): + execute_with_retry( + conn, + "INSERT INTO tasks(id, project_id, prompt, status)" + " VALUES(?, ?, ?, 'queued')", + ("t-ok", "p1", "ok"), + ) + self.assertEqual( + len(read_all(conn, "SELECT * FROM tasks")), 1) + with self.assertRaises(sqlite3.IntegrityError): + with transaction(conn): + execute_with_retry( + conn, + "INSERT INTO tasks(id, project_id, prompt, status)" + " VALUES(?, ?, ?, 'queued')", + ("t-ok", "p1", "dup"), + ) + # 冲突行回滚后仍只有原来 1 条，不留半写。 + self.assertEqual( + len(read_all(conn, "SELECT * FROM tasks")), 1) + finally: + conn.close() + + # -- 真实并发：8 线程 x 20 条 ---------------------------------------- + def test_concurrent_writes_8_threads_x_20_no_loss(self): + errors = [] + barrier = threading.Barrier(8) + + def worker(thread_idx): + try: + barrier.wait(timeout=10) + conn = connect(self.db_path) + try: + for i in range(20): + with transaction(conn): + execute_with_retry( + conn, + "INSERT INTO tasks(id, project_id, prompt," + " status, idempotency_key, depends_on," + " priority, created_by, created_at," + " updated_at, last_heartbeat)" + " VALUES(?, ?, ?, 'queued', ?, '[]'," + " 0, 'worker', ?, ?, NULL)", + ("t-%d-%d" % (thread_idx, i), + "p-concurrent", + "prompt %d-%d" % (thread_idx, i), + "key-%d-%d" % (thread_idx, i), + time.time(), time.time()), + ) + finally: + conn.close() + except Exception as exc: # noqa: BLE001 - 汇总后统一断言 + errors.append(exc) + + threads = [threading.Thread(target=worker, args=(n,)) + for n in range(8)] + for t in threads: + t.start() + for t in threads: + t.join(timeout=60) + self.assertFalse([t for t in threads if t.is_alive()], + "worker 线程未在 60s 内结束") + self.assertEqual(errors, [], "并发写不应有任何异常: %r" % (errors,)) + conn = connect(self.db_path) + try: + rows = read_all( + conn, + "SELECT id FROM tasks WHERE project_id = ?", + ("p-concurrent",)) + finally: + conn.close() + self.assertEqual(len(rows), 160) + self.assertEqual(len({r["id"] for r in rows}), 160) + + +if __name__ == "__main__": + unittest.main() diff --git a/cloud/tests/test_oauth.py b/cloud/tests/test_oauth.py new file mode 100644 index 0000000..8e33e37 --- /dev/null +++ b/cloud/tests/test_oauth.py @@ -0,0 +1,275 @@ +"""oauth 脚手架单测：state、授权 URL、回调、换 token 失败。 + +全部外部调用均为注入的 fake http 客户端，本文件不向真实 +GitHub / Google 发任何请求；出现的 client_id / secret 哨兵值 +均为测试用的假值（形如 test- / fake- 前缀），不是真实凭证。 +""" + +import os +import sqlite3 +import sys +import tempfile +import time +import unittest +from pathlib import Path +from unittest import mock +from urllib.parse import parse_qs, urlparse + +try: + from cloud.auth import AuthStore + from cloud.oauth import ( + OAuthError, + OAuthStateStore, + build_authorization_url, + exchange_code_for_token, + handle_oauth_callback, + map_external_identity_to_local_user, + start_oauth_login, + ) +except ImportError: # discover -s cloud/tests 时把项目根目录补进 sys.path + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + from cloud.auth import AuthStore + from cloud.oauth import ( + OAuthError, + OAuthStateStore, + build_authorization_url, + exchange_code_for_token, + handle_oauth_callback, + map_external_identity_to_local_user, + start_oauth_login, + ) + +PROVIDER = "github" +CLIENT_ID = "test-client-id-not-real" +REDIRECT_URI = "https://example.invalid/oauth/callback" +FAKE_SECRET_SENTINEL = "fake-secret-sentinel-not-real" + + +def _fake_token_client(token_result=None, error=None): + """返回 fake http_client：记录调用，不联网。""" + calls = [] + + def fake(token_url, payload): + calls.append((token_url, dict(payload))) + if error is not None: + raise RuntimeError(error) + return dict(token_result or {"access_token": "fake-token-not-real"}) + + fake.calls = calls + return fake + + +class TestOAuthState(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.db_path = os.path.join(self.tmp.name, "oauth.db") + self.store = OAuthStateStore(self.db_path) + + def tearDown(self): + self.tmp.cleanup() + + def test_state_create_then_validate_ok(self): + state = self.store.create_state(PROVIDER, REDIRECT_URI) + self.assertTrue(state) + record = self.store.validate_state(state, PROVIDER, REDIRECT_URI) + self.assertIsNotNone(record) + self.assertEqual(record["provider"], PROVIDER) + self.assertEqual(record["redirect_uri"], REDIRECT_URI) + # 一次性：第二次校验必然失败（防重放） + self.assertIsNone(self.store.validate_state(state, PROVIDER)) + + def test_state_expired_rejected(self): + state = self.store.create_state(PROVIDER, REDIRECT_URI) + expired_now = time.time() + 601 # 默认 TTL 600 秒 + self.assertIsNone(self.store.validate_state( + state, PROVIDER, now=expired_now)) + # 过期行已被删除，不会残留 + row = self.store.conn.execute( + "SELECT state FROM oauth_states WHERE state = ?", (state,) + ).fetchone() + self.assertIsNone(row) + + def test_state_mismatch_rejected(self): + state = self.store.create_state(PROVIDER, REDIRECT_URI) + # 不存在的 state + self.assertIsNone(self.store.validate_state( + "no-such-state", PROVIDER)) + # provider 不匹配：拒绝，且不消耗合法 state + self.assertIsNone(self.store.validate_state(state, "google")) + # redirect_uri 不匹配：同样拒绝且不消耗 + self.assertIsNone(self.store.validate_state( + state, PROVIDER, "https://example.invalid/other")) + # 合法组合仍可通过（前两次不匹配没有烧掉它） + self.assertIsNotNone(self.store.validate_state( + state, PROVIDER, REDIRECT_URI)) + + +class TestAuthorizationUrl(unittest.TestCase): + def test_url_contains_params_and_no_secret(self): + state = "test-state-abc" + url = build_authorization_url( + PROVIDER, CLIENT_ID, REDIRECT_URI, + scope="read:user", state=state) + parsed = urlparse(url) + self.assertEqual(parsed.netloc, "github.com") + params = {k: v[0] for k, v in parse_qs(parsed.query).items()} + self.assertEqual(params["client_id"], CLIENT_ID) + self.assertEqual(params["redirect_uri"], REDIRECT_URI) + self.assertEqual(params["scope"], "read:user") + self.assertEqual(params["state"], state) + self.assertNotIn("secret", url.lower()) + self.assertNotIn(FAKE_SECRET_SENTINEL, url) + + def test_url_default_scope_and_google(self): + url = build_authorization_url( + "github", CLIENT_ID, REDIRECT_URI, state="s1") + self.assertIn("read%3Auser", url) # 默认 scope read:user + url = build_authorization_url( + "google", CLIENT_ID, REDIRECT_URI, state="s2") + params = {k: v[0] for k, v in parse_qs(urlparse(url).query).items()} + self.assertEqual(params["scope"], "openid email profile") + self.assertEqual(params["response_type"], "code") + self.assertNotIn("secret", url.lower()) + + def test_start_login_stores_state(self): + tmp = tempfile.TemporaryDirectory() + try: + store = OAuthStateStore(os.path.join(tmp.name, "s.db")) + state, url = start_oauth_login( + store, PROVIDER, CLIENT_ID, REDIRECT_URI) + self.assertIn("state=" + state, url) + row = store.conn.execute( + "SELECT provider FROM oauth_states WHERE state = ?", + (state,), + ).fetchone() + self.assertEqual(row[0], PROVIDER) + finally: + tmp.cleanup() + + def test_unknown_provider_rejected(self): + with self.assertRaises(OAuthError): + build_authorization_url( + "not-a-provider", CLIENT_ID, REDIRECT_URI) + + +class TestOAuthCallback(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.db_path = os.path.join(self.tmp.name, "oauth.db") + self.store = OAuthStateStore(self.db_path) + # 同库上的 AuthStore：用于断言“失败不建用户/不记审计”。 + self.auth = AuthStore(self.db_path) + + def tearDown(self): + self.tmp.cleanup() + + def _login(self, fake=None, **kwargs): + state = self.store.create_state(PROVIDER, REDIRECT_URI) + return handle_oauth_callback( + self.store, PROVIDER, "test-code-not-real", state, + CLIENT_ID, REDIRECT_URI, + http_client=fake or _fake_token_client( + {"access_token": "fake-token-not-real", + "id": 12345, "login": "fake-user"}), + **kwargs) + + def _count(self, table): + return self.store.conn.execute( + "SELECT COUNT(*) FROM %s" % table).fetchone()[0] + + def test_callback_success_returns_identity(self): + identity = self._login() + self.assertEqual(identity["provider"], PROVIDER) + self.assertEqual(identity["external_id"], "12345") + self.assertEqual(identity["login"], "fake-user") + # access_token 不返回、不入库：回调只交身份 + self.assertNotIn("access_token", identity) + + def test_callback_maps_identity_to_local_user(self): + identity = self._login() + user_id, created = map_external_identity_to_local_user( + self.auth, PROVIDER, identity) + self.assertTrue(created) + user_id2, created2 = map_external_identity_to_local_user( + self.auth, PROVIDER, identity) + self.assertEqual(user_id, user_id2) + self.assertFalse(created2) + + def test_callback_state_expired_rejected(self): + state = self.store.create_state(PROVIDER, REDIRECT_URI) + with self.assertRaises(OAuthError) as ctx: + handle_oauth_callback( + self.store, PROVIDER, "code", state, CLIENT_ID, REDIRECT_URI, + http_client=_fake_token_client(), + now=time.time() + 601) + self.assertEqual(ctx.exception.reason, "state_expired") + + def test_callback_state_mismatch_rejected(self): + state = self.store.create_state(PROVIDER, REDIRECT_URI) + with self.assertRaises(OAuthError) as ctx: + handle_oauth_callback( + self.store, "google", "code", state, CLIENT_ID, REDIRECT_URI, + http_client=_fake_token_client()) + self.assertEqual(ctx.exception.reason, "state_mismatch") + with self.assertRaises(OAuthError) as ctx: + handle_oauth_callback( + self.store, PROVIDER, "code", "missing-state", + CLIENT_ID, REDIRECT_URI, + http_client=_fake_token_client()) + self.assertEqual(ctx.exception.reason, "state_not_found") + + def test_token_exchange_failure_creates_no_session_or_user(self): + fake = _fake_token_client(error="provider refused code") + state = self.store.create_state(PROVIDER, REDIRECT_URI) + with self.assertRaises(OAuthError) as ctx: + handle_oauth_callback( + self.store, PROVIDER, "bad-code", state, + CLIENT_ID, REDIRECT_URI, http_client=fake) + self.assertEqual(ctx.exception.reason, "token_exchange_failed") + # 换 token 失败：无用户、无审计（本模块本来就不建会话， + # 这里断言失败路径也没有任何副作用写入同库的 auth 表）。 + self.assertEqual(self._count("users"), 0) + self.assertEqual(self._count("audit_log"), 0) + + def test_token_error_dict_creates_nothing(self): + fake = _fake_token_client( + {"error": "bad_verification_code"}) + state = self.store.create_state(PROVIDER, REDIRECT_URI) + with self.assertRaises(OAuthError): + handle_oauth_callback( + self.store, PROVIDER, "bad-code", state, + CLIENT_ID, REDIRECT_URI, http_client=fake) + self.assertEqual(self._count("users"), 0) + + def test_real_path_without_secret_refuses_before_network(self): + # http_client=None 且环境变量无 secret：必须在联网前拒绝。 + # 用 patch 清空两个 secret 环境变量，并断言 urlopen 未被调用。 + with mock.patch.dict(os.environ, {}, clear=False): + os.environ.pop("GITHUB_OAUTH_CLIENT_SECRET", None) + os.environ.pop("GOOGLE_OAUTH_CLIENT_SECRET", None) + with mock.patch("cloud.oauth._urllib_post") as mocked: + with self.assertRaises(OAuthError) as ctx: + exchange_code_for_token( + PROVIDER, "code", CLIENT_ID, REDIRECT_URI, + http_client=None) + self.assertEqual( + ctx.exception.reason, "client_secret_missing") + mocked.assert_not_called() + + def test_userinfo_client_used_when_token_has_no_identity(self): + def userinfo(access_token): + self.assertEqual(access_token, "fake-token-not-real") + return {"id": 999, "login": "via-userinfo"} + + state = self.store.create_state(PROVIDER, REDIRECT_URI) + identity = handle_oauth_callback( + self.store, PROVIDER, "code", state, CLIENT_ID, REDIRECT_URI, + http_client=_fake_token_client( + {"access_token": "fake-token-not-real"}), + userinfo_client=userinfo) + self.assertEqual(identity["external_id"], "999") + self.assertEqual(identity["login"], "via-userinfo") + + +if __name__ == "__main__": + unittest.main() diff --git a/cloud/tests/test_push.py b/cloud/tests/test_push.py new file mode 100644 index 0000000..14fedef --- /dev/null +++ b/cloud/tests/test_push.py @@ -0,0 +1,304 @@ +"""push.py 单测：订阅增删查、偏好过滤、分发错误不静默丢。 + +只用标准库，不联网、不向真实推送服务发任何请求：所有发送都用 +假 sender（可调用对象或按通道分的 dict）注入；webpush/apns/fcm +只断言缺凭证报错，不调用任何 sender。 +本文件不含任何真实密钥、token、证书或私钥。 +""" + +import os +import sqlite3 +import sys +import unittest +from pathlib import Path + +try: + from cloud.push import ( + ALLOWED_PUSH_EVENTS, + CHANNELS, + FAILURE_EVENTS, + PREFERENCES, + create_subscription, + delete_subscription, + dispatch_event, + list_subscriptions, + ) +except ImportError: # discover -s cloud/tests 时把项目根目录补进 sys.path + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + from cloud.push import ( + ALLOWED_PUSH_EVENTS, + CHANNELS, + FAILURE_EVENTS, + PREFERENCES, + create_subscription, + delete_subscription, + dispatch_event, + list_subscriptions, + ) + + +class PushTestCase(unittest.TestCase): + def setUp(self): + # 内存库 + 单一连接：init 与查询共用同一连接，避免 :memory: 换连接丢表。 + self.conn = sqlite3.connect(":memory:") + + def tearDown(self): + self.conn.close() + + def _sub(self, project="proj-a", user="user-1", channel="webhook", + target="https://example.invalid/hook", preference="instant"): + return create_subscription( + self.conn, project, user, channel, target, preference) + + def _calls(self): + calls = [] + + def sender(message): + calls.append(message) + + return calls, sender + + +class TestSubscriptionCrud(PushTestCase): + def test_create_and_list(self): + sub = self._sub() + self.assertEqual(sub["project_id"], "proj-a") + self.assertEqual(sub["user_id"], "user-1") + self.assertEqual(sub["channel"], "webhook") + self.assertEqual(sub["preference"], "instant") + listed = list_subscriptions(self.conn, "proj-a", "user-1") + self.assertEqual(len(listed), 1) + self.assertEqual(listed[0]["id"], sub["id"]) + + def test_create_duplicate_returns_existing_not_duplicate_row(self): + first = self._sub() + second = self._sub() + self.assertEqual(first["id"], second["id"]) + self.assertEqual( + len(list_subscriptions(self.conn, "proj-a", "user-1")), 1) + + def test_create_invalid_channel_or_preference_rejected(self): + with self.assertRaises(ValueError): + self._sub(channel="sms") + with self.assertRaises(ValueError): + self._sub(preference="weekly") + self.assertEqual( + len(list_subscriptions(self.conn, "proj-a", "user-1")), 0) + + def test_delete_by_id_then_no_longer_listed(self): + sub = self._sub() + self.assertTrue(delete_subscription(self.conn, sub["id"])) + self.assertEqual( + list_subscriptions(self.conn, "proj-a", "user-1"), []) + # 再次删除无匹配，返回 False，不报错。 + self.assertFalse(delete_subscription(self.conn, sub["id"])) + + def test_constants_cover_task_contract(self): + self.assertEqual( + set(CHANNELS), {"webhook", "slack", "webpush", "apns", "fcm"}) + self.assertEqual(set(PREFERENCES), {"instant", "digest", "silent"}) + self.assertEqual( + ALLOWED_PUSH_EVENTS, + {"completed", "failed", "goal-unmet", "merge-failed", + "review-requested"}, + ) + self.assertEqual(FAILURE_EVENTS, {"failed", "goal-unmet", "merge-failed"}) + + +class TestDispatchPreferences(PushTestCase): + def test_silent_preference_never_sends(self): + self._sub(preference="silent") + calls, sender = self._calls() + result = dispatch_event( + self.conn, "failed", "proj-a", {"task_id": "t-1"}, sender=sender) + self.assertEqual(calls, []) + self.assertEqual(result["sent"], []) + self.assertEqual(len(result["skipped_silent"]), 1) + self.assertEqual(result["skipped_silent"][0]["preference"], "silent") + + def test_failure_event_sends_immediately_even_with_digest(self): + # 失败类：digest 也即时发，不进 skipped_for_digest。 + for event_type in ("failed", "goal-unmet", "merge-failed"): + with self.subTest(event_type=event_type): + conn = sqlite3.connect(":memory:") + try: + create_subscription( + conn, "proj-a", "user-1", "webhook", + "https://example.invalid/hook", "digest") + calls, sender = self._calls() + result = dispatch_event( + conn, event_type, "proj-a", {"task_id": "t-1"}, + sender=sender) + self.assertEqual(len(calls), 1) + self.assertEqual(calls[0]["event_type"], event_type) + self.assertEqual(len(result["sent"]), 1) + self.assertEqual(result["skipped_for_digest"], []) + finally: + conn.close() + + def test_completed_event_filtered_by_digest_returns_pending_list(self): + self._sub(preference="digest") + calls, sender = self._calls() + result = dispatch_event( + self.conn, "completed", "proj-a", {"task_id": "t-1"}, + sender=sender) + # 不立即发，但不是静默丢弃：明确列出待摘要。 + self.assertEqual(calls, []) + self.assertEqual(result["sent"], []) + self.assertEqual(len(result["skipped_for_digest"]), 1) + self.assertEqual( + result["skipped_for_digest"][0]["preference"], "digest") + self.assertEqual(result["errors"], []) + + def test_review_requested_digest_also_deferred(self): + self._sub(preference="digest") + calls, sender = self._calls() + result = dispatch_event( + self.conn, "review-requested", "proj-a", {"task_id": "t-1"}, + sender=sender) + self.assertEqual(calls, []) + self.assertEqual(len(result["skipped_for_digest"]), 1) + + def test_instant_completed_sends(self): + self._sub(preference="instant") + calls, sender = self._calls() + result = dispatch_event( + self.conn, "completed", "proj-a", {"task_id": "t-1"}, + sender=sender) + self.assertEqual(len(calls), 1) + self.assertEqual(calls[0]["channel"], "webhook") + self.assertEqual(calls[0]["payload"], {"task_id": "t-1"}) + self.assertEqual(len(result["sent"]), 1) + self.assertEqual(result["errors"], []) + + def test_invalid_event_rejected_and_sender_not_called(self): + self._sub(preference="instant") + calls, sender = self._calls() + result = dispatch_event( + self.conn, "running", "proj-a", {"task_id": "t-1"}, sender=sender) + self.assertEqual(calls, []) + self.assertEqual(result["sent"], []) + self.assertEqual(result["reason"], "event not allowed") + self.assertTrue( + any("event not allowed" in e["reason"] for e in result["errors"])) + + +class TestDispatchSendersAndCredentials(PushTestCase): + def test_webhook_without_sender_returns_no_sender_error_not_silent_ok(self): + self._sub(channel="webhook") + result = dispatch_event( + self.conn, "completed", "proj-a", {"task_id": "t-1"}, sender=None) + self.assertEqual(result["sent"], []) + self.assertEqual(len(result["errors"]), 1) + self.assertIn("no sender", result["errors"][0]["reason"]) + + def test_slack_via_channel_dict_sender(self): + create_subscription( + self.conn, "proj-a", "user-1", "slack", "C123", "instant") + slack_calls = [] + + def slack_sender(message): + slack_calls.append(message) + + result = dispatch_event( + self.conn, "completed", "proj-a", {"task_id": "t-1"}, + sender={"slack": slack_sender}) + self.assertEqual(len(slack_calls), 1) + self.assertEqual(slack_calls[0]["channel"], "slack") + self.assertEqual(len(result["sent"]), 1) + self.assertEqual(result["errors"], []) + + def test_webpush_apns_fcm_missing_credential_error_not_silent_drop(self): + cases = { + "webpush": "VAPID", + "apns": "APNs", + "fcm": "FCM", + } + for channel, keyword in cases.items(): + with self.subTest(channel=channel): + conn = sqlite3.connect(":memory:") + try: + create_subscription( + conn, "proj-a", "user-1", channel, + "device-or-endpoint", "instant") + calls, sender = self._calls() + result = dispatch_event( + conn, "failed", "proj-a", {"task_id": "t-1"}, + sender=sender) + # 绝不走假 sender、绝不假装成功：明确报错。 + self.assertEqual(calls, []) + self.assertEqual(result["sent"], []) + self.assertEqual(len(result["errors"]), 1) + self.assertIn("missing credential", result["errors"][0]["reason"]) + self.assertIn(keyword, result["errors"][0]["reason"]) + finally: + conn.close() + + def test_sender_exception_recorded_not_raised_and_others_continue(self): + self._sub(user="user-1", target="https://example.invalid/hook-1") + + def flaky(message): + if message["target"] == "https://example.invalid/hook-1": + raise RuntimeError("fake send failure") + calls.append(message) + + calls = [] + # 第二个订阅用另一个 target，同一 sender 只对第一个抛错。 + create_subscription( + self.conn, "proj-a", "user-2", "webhook", + "https://example.invalid/hook-2", "instant") + result = dispatch_event( + self.conn, "completed", "proj-a", {"task_id": "t-1"}, + sender=flaky) + self.assertEqual(len(result["sent"]), 1) + self.assertEqual(len(result["errors"]), 1) + self.assertIn("sender failed", result["errors"][0]["reason"]) + self.assertEqual(len(calls), 1) + + +class TestDispatchIsolation(PushTestCase): + def test_deleted_subscription_no_longer_receives(self): + sub = self._sub() + self.assertTrue(delete_subscription(self.conn, sub["id"])) + calls, sender = self._calls() + result = dispatch_event( + self.conn, "failed", "proj-a", {"task_id": "t-1"}, sender=sender) + self.assertEqual(calls, []) + self.assertEqual(result["sent"], []) + self.assertEqual(result["errors"], []) + + def test_cross_project_subscriptions_do_not_mix(self): + self._sub(project="proj-a", user="user-1") + self._sub(project="proj-b", user="user-1", + target="https://example.invalid/hook-b") + calls, sender = self._calls() + # B 项目事件只发给 B 的订阅，A 的订阅收不到。 + result_b = dispatch_event( + self.conn, "failed", "proj-b", {"task_id": "t-1"}, sender=sender) + self.assertEqual(len(calls), 1) + self.assertEqual(calls[0]["target"], "https://example.invalid/hook-b") + self.assertEqual(len(result_b["sent"]), 1) + # A 项目事件只发给 A 的订阅。 + calls.clear() + result_a = dispatch_event( + self.conn, "failed", "proj-a", {"task_id": "t-1"}, sender=sender) + self.assertEqual(len(calls), 1) + self.assertEqual( + calls[0]["target"], "https://example.invalid/hook") + self.assertEqual(len(result_a["sent"]), 1) + + def test_list_subscriptions_scoped_by_project_and_user(self): + self._sub(project="proj-a", user="user-1") + self._sub(project="proj-a", user="user-2", + target="https://example.invalid/hook-2") + self._sub(project="proj-b", user="user-1", + target="https://example.invalid/hook-b") + self.assertEqual(len(list_subscriptions(self.conn, "proj-a")), 2) + self.assertEqual( + len(list_subscriptions(self.conn, "proj-a", "user-1")), 1) + self.assertEqual( + len(list_subscriptions(self.conn, "proj-b", "user-1")), 1) + + +if __name__ == "__main__": + unittest.main() diff --git a/cloud/tests/test_server_endpoints.py b/cloud/tests/test_server_endpoints.py new file mode 100644 index 0000000..c73e2a4 --- /dev/null +++ b/cloud/tests/test_server_endpoints.py @@ -0,0 +1,225 @@ +"""新端点测试：记忆 / 知识库 / 成本 / 模型总览。 + +只用标准库 http.client，不跑真实模型，不联网。 +每个测试用独立临时 SQLite 与临时端口。 +""" + +import http.client +import json +import os +import sys +import tempfile +import threading +import unittest +from pathlib import Path +from urllib.parse import quote + +try: + from cloud.auth import AuthStore + from cloud.cost import CostStore + from cloud.server import create_server +except ImportError: + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + from cloud.auth import AuthStore + from cloud.cost import CostStore + from cloud.server import create_server + + +class EndpointsTestCase(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.db_path = os.path.join(self.tmp.name, "control.db") + self.auth = AuthStore(self.db_path) + self.owner_id, self.owner_key = self.auth.create_user("owner") + self.viewer_id, self.viewer_key = self.auth.create_user("viewer") + self.outsider_id, self.outsider_key = self.auth.create_user("outsider") + self.server = create_server(self.db_path, "127.0.0.1", 0) + self.port = self.server.server_address[1] + self.thread = threading.Thread( + target=self.server.serve_forever, daemon=True) + self.thread.start() + status, self.project = self._req( + "POST", "/projects", {"name": "demo"}, self.owner_key) + self.assertEqual(status, 201) + self.auth.set_member( + self.project["id"], self.owner_id, self.viewer_id, "viewer") + # outsider 自己的项目，用于跨项目隔离测试。 + status, self.other = self._req( + "POST", "/projects", {"name": "other"}, self.outsider_key) + self.assertEqual(status, 201) + + def tearDown(self): + self.server.shutdown() + self.server.server_close() + self.thread.join(timeout=5) + try: + self.auth.conn.close() + except Exception: + pass + self.tmp.cleanup() + + def _req(self, method, path, body=None, key=None): + conn = http.client.HTTPConnection("127.0.0.1", self.port, timeout=10) + hdrs = {} + payload = None + if body is not None: + payload = json.dumps(body).encode("utf-8") + hdrs["Content-Type"] = "application/json" + if key is not None: + hdrs["X-API-Key"] = key + try: + conn.request(method, path, body=payload, headers=hdrs) + resp = conn.getresponse() + data = resp.read() + finally: + conn.close() + try: + parsed = json.loads(data.decode("utf-8")) if data else None + except ValueError: + parsed = None + return resp.status, parsed + + def _pid(self): + return self.project["id"] + + # -- 记忆 ---------------------------------------------------------- + def test_memory_create_list_supersede(self): + status, created = self._req( + "POST", "/projects/%s/memory" % self._pid(), + {"thread_id": "t1", "goal": "加法", "conclusion": "已加 add2", + "files_changed": ["mod.py"]}, + self.owner_key) + self.assertEqual(status, 201, created) + status, body = self._req( + "GET", "/projects/%s/memory" % self._pid(), key=self.viewer_key) + self.assertEqual(status, 200) + self.assertEqual(len(body["summaries"]), 1) + self.assertIn("add2", body["prompt_context"]) + status, body = self._req( + "POST", "/projects/%s/memory/%s/supersede" + % (self._pid(), created["id"]), {}, self.owner_key) + self.assertEqual(status, 200) + self.assertEqual(body["status"], "deprecated") + status, body = self._req( + "GET", "/projects/%s/memory" % self._pid(), key=self.owner_key) + self.assertEqual(body["summaries"], []) + + def test_memory_no_key_401_and_outsider_403(self): + status, _ = self._req( + "GET", "/projects/%s/memory" % self._pid()) + self.assertEqual(status, 401) + status, _ = self._req( + "GET", "/projects/%s/memory" % self._pid(), + key=self.outsider_key) + self.assertEqual(status, 403) + status, _ = self._req( + "POST", "/projects/%s/memory" % self._pid(), + {"goal": "g", "conclusion": "c"}, self.viewer_key) + self.assertEqual(status, 403) + + def test_memory_cross_project_supersede_404(self): + status, created = self._req( + "POST", "/projects/%s/memory" % self._pid(), + {"goal": "g", "conclusion": "c"}, self.owner_key) + self.assertEqual(status, 201) + # outsider 拿自己项目的路径去废弃别人的摘要：应 403/404，不成功。 + status, _ = self._req( + "POST", "/projects/%s/memory/%s/supersede" + % (self.other["id"], created["id"]), {}, self.outsider_key) + self.assertIn(status, (403, 404)) + + def test_memory_unknown_project_404(self): + status, _ = self._req( + "GET", "/projects/nope/memory", key=self.owner_key) + self.assertEqual(status, 404) + + # -- 知识库 --------------------------------------------------------- + def test_kb_index_search_and_isolation(self): + status, _ = self._req( + "POST", "/projects/%s/kb/index" % self._pid(), + {"path": "README.md", "content": "用户登录流程说明", "kind": "readme"}, + self.owner_key) + self.assertEqual(status, 201) + status, body = self._req( + "GET", "/projects/%s/kb/search?q=%s" + % (self._pid(), quote("登录")), + key=self.viewer_key) + self.assertEqual(status, 200) + self.assertEqual(len(body["results"]), 1) + # outsider 在自己项目里搜不到别人的文档。 + status, body = self._req( + "GET", "/projects/%s/kb/search?q=%s" + % (self.other["id"], quote("登录")), + key=self.outsider_key) + self.assertEqual(status, 200) + self.assertEqual(body["results"], []) + # 无 key 401，越权 403。 + status, _ = self._req( + "GET", "/projects/%s/kb/search?q=%s" + % (self._pid(), quote("登录"))) + self.assertEqual(status, 401) + status, _ = self._req( + "GET", "/projects/%s/kb/search?q=%s" + % (self._pid(), quote("登录")), + key=self.outsider_key) + self.assertEqual(status, 403) + + # -- 成本 ----------------------------------------------------------- + def test_task_and_project_cost(self): + status, task = self._req( + "POST", "/tasks", + {"project_id": self._pid(), "prompt": "做加法"}, self.owner_key) + self.assertEqual(status, 201) + store = CostStore(self.db_path) + try: + store.record(task["id"], self._pid(), "gpt-6.1-sol", + input_tokens=100, output_tokens=50, + usd_micros=1234, source="actual", fallback_count=1) + store.record_estimated(task["id"], self._pid(), "gpt-6.1-sol", + "prompt" * 10, "output" * 10) + finally: + store._conn.close() + status, body = self._req( + "GET", "/tasks/%s/cost" % task["id"], key=self.viewer_key) + self.assertEqual(status, 200) + self.assertEqual(body["actual_tokens"], 150) + self.assertGreater(body["estimated_tokens"], 0) + self.assertEqual(body["fallback_count"], 1) + status, _ = self._req( + "GET", "/tasks/%s/cost" % task["id"], key=self.outsider_key) + self.assertEqual(status, 403) + status, _ = self._req( + "GET", "/tasks/nope/cost", key=self.owner_key) + self.assertEqual(status, 404) + status, body = self._req( + "GET", "/projects/%s/cost" % self._pid(), key=self.owner_key) + self.assertEqual(status, 200) + self.assertEqual(body["overall"]["turns"], 2) + + # -- 模型总览 ------------------------------------------------------- + def test_models_overview_isolated(self): + status, task = self._req( + "POST", "/tasks", + {"project_id": self._pid(), "prompt": "x"}, self.owner_key) + store = CostStore(self.db_path) + try: + store.record(task["id"], self._pid(), "gpt-6.1-sol", + input_tokens=10, output_tokens=5) + finally: + store._conn.close() + status, body = self._req("GET", "/models", key=self.owner_key) + self.assertEqual(status, 200) + self.assertIn("gpt-6.1-sol", body["priority_list"]) + models = {m["model"] for m in body["model_stats"]} + self.assertIn("gpt-6.1-sol", models) + # outsider 只能看到自己项目的统计（目前为空），看不到别人的。 + status, body = self._req("GET", "/models", key=self.outsider_key) + self.assertEqual(status, 200) + models = {m["model"] for m in body["model_stats"]} + self.assertNotIn("gpt-6.1-sol", models) + status, _ = self._req("GET", "/models") + self.assertEqual(status, 401) + + +if __name__ == "__main__": + unittest.main() diff --git a/cloud/tests/test_server_oauth_push.py b/cloud/tests/test_server_oauth_push.py new file mode 100644 index 0000000..0a5ff45 --- /dev/null +++ b/cloud/tests/test_server_oauth_push.py @@ -0,0 +1,177 @@ +"""OAuth 与推送的 HTTP 端点测试。 + +不联网、不向真实 GitHub/Google/推送服务发请求；OAuth 回调成功 +路径由模块级单测覆盖，这里只测 HTTP 层的失败路径与订阅分发。 +""" + +import http.client +import json +import os +import sys +import tempfile +import threading +import unittest +from pathlib import Path + +try: + from cloud.auth import AuthStore + from cloud.oauth import OAuthStateStore + from cloud.server import create_server +except ImportError: + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + from cloud.auth import AuthStore + from cloud.oauth import OAuthStateStore + from cloud.server import create_server + + +class OAuthPushEndpointsTestCase(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.db_path = os.path.join(self.tmp.name, "control.db") + self.auth = AuthStore(self.db_path) + self.owner_id, self.owner_key = self.auth.create_user("owner") + self.outsider_id, self.outsider_key = self.auth.create_user("outsider") + self.server = create_server(self.db_path, "127.0.0.1", 0) + self.port = self.server.server_address[1] + self.thread = threading.Thread( + target=self.server.serve_forever, daemon=True) + self.thread.start() + status, self.project = self._req( + "POST", "/projects", {"name": "demo"}, self.owner_key) + self.assertEqual(status, 201) + + def tearDown(self): + self.server.shutdown() + self.server.server_close() + self.thread.join(timeout=5) + try: + self.auth.conn.close() + except Exception: + pass + self.tmp.cleanup() + + def _req(self, method, path, body=None, key=None): + conn = http.client.HTTPConnection("127.0.0.1", self.port, timeout=10) + hdrs = {} + payload = None + if body is not None: + payload = json.dumps(body).encode("utf-8") + hdrs["Content-Type"] = "application/json" + if key is not None: + hdrs["X-API-Key"] = key + try: + conn.request(method, path, body=payload, headers=hdrs) + resp = conn.getresponse() + data = resp.read() + finally: + conn.close() + try: + parsed = json.loads(data.decode("utf-8")) if data else None + except ValueError: + parsed = None + return resp.status, parsed + + # -- OAuth ----------------------------------------------------------- + def test_oauth_login_returns_url_without_secret(self): + status, body = self._req( + "GET", + "/oauth/github/login?client_id=test-client" + "&redirect_uri=http%3A%2F%2F127.0.0.1%2Fcb") + self.assertEqual(status, 200) + self.assertIn("github.com/login/oauth/authorize", + body["authorization_url"]) + self.assertIn("test-client", body["authorization_url"]) + self.assertNotIn("secret", body["authorization_url"]) + self.assertTrue(body["state"]) + + def test_oauth_login_unknown_provider_400(self): + status, _ = self._req( + "GET", + "/oauth/nope/login?client_id=x" + "&redirect_uri=http%3A%2F%2F127.0.0.1%2Fcb") + self.assertEqual(status, 400) + + def test_oauth_callback_bad_state_400_no_session(self): + status, body = self._req( + "POST", "/oauth/github/callback", + {"code": "fake-code", "state": "no-such-state", + "client_id": "test-client", + "redirect_uri": "http://127.0.0.1/cb"}) + self.assertEqual(status, 400) + self.assertIn("state", body["error"]) + + def test_oauth_callback_good_state_no_secret_400(self): + store = OAuthStateStore(self.db_path) + try: + state = store.create_state( + "github", "http://127.0.0.1/cb") + finally: + store.conn.close() + # 确保环境里没有真实 secret（本测试不依赖它）。 + old = os.environ.pop("GITHUB_OAUTH_CLIENT_SECRET", None) + try: + status, body = self._req( + "POST", "/oauth/github/callback", + {"code": "fake-code", "state": state, + "client_id": "test-client", + "redirect_uri": "http://127.0.0.1/cb"}) + finally: + if old is not None: + os.environ["GITHUB_OAUTH_CLIENT_SECRET"] = old + self.assertEqual(status, 400) + self.assertIn("client_secret", body["error"]) + + # -- 推送 ------------------------------------------------------------- + def test_push_subscribe_list_dispatch_and_delete(self): + pid = self.project["id"] + status, sub = self._req( + "POST", "/projects/%s/push/subscriptions" % pid, + {"channel": "webhook", "target": "https://example.invalid/hook", + "preference": "instant"}, + self.owner_key) + self.assertEqual(status, 201, sub) + status, body = self._req( + "GET", "/projects/%s/push/subscriptions" % pid, + key=self.owner_key) + self.assertEqual(status, 200) + self.assertEqual(len(body["subscriptions"]), 1) + # HTTP 层无注入 sender：分发应报“无 sender”，不静默成功。 + status, body = self._req( + "POST", "/admin/push/dispatch", + {"event_type": "failed", "project_id": pid, + "payload": {"task_id": "t1"}}, + self.owner_key) + self.assertEqual(status, 200) + self.assertEqual(len(body["errors"]), 1) + self.assertIn("no sender", body["errors"][0]["reason"]) + # outsider 不能往别人项目里订阅/分发。 + status, _ = self._req( + "POST", "/projects/%s/push/subscriptions" % pid, + {"channel": "webhook", "target": "https://example.invalid/x"}, + self.outsider_key) + self.assertEqual(status, 403) + # 删除：DELETE 方法须可用，且删后列表为空。 + conn = http.client.HTTPConnection("127.0.0.1", self.port, timeout=10) + try: + conn.request("DELETE", + "/projects/%s/push/subscriptions/%s" % (pid, sub["id"]), + headers={"X-API-Key": self.owner_key}) + resp = conn.getresponse() + resp.read() + self.assertEqual(resp.status, 200) + finally: + conn.close() + status, body = self._req( + "GET", "/projects/%s/push/subscriptions" % pid, + key=self.owner_key) + self.assertEqual(body["subscriptions"], []) + + def test_push_dispatch_requires_auth(self): + status, _ = self._req( + "POST", "/admin/push/dispatch", + {"event_type": "completed", "project_id": self.project["id"]}) + self.assertEqual(status, 401) + + +if __name__ == "__main__": + unittest.main() diff --git a/docs/cloud-architecture.md b/docs/cloud-architecture.md index badabb6..8dce706 100644 --- a/docs/cloud-architecture.md +++ b/docs/cloud-architecture.md @@ -234,6 +234,37 @@ CI 过了才合。 fallback 列表 + 成本报告。依赖 3A 的任务记录，不依赖 3B/3C，可以和 3C 并行。 +## 补齐进度（2026-10-26 第二轮，上一轮留下的未完成项） + +已做完（不要再列为未完成）： +- 记忆 / 知识库 / 成本 / 模型已接成 server.py 正式端点， + 全部走 X-API-Key + 项目 RBAC（除 /health、/webhooks/*、 + /oauth/* pre-auth 外），单测覆盖 401/403/404 与跨项目隔离。 +- SQLite 并发：连接统一走 cloud/db.py（WAL + busy_timeout + 5000ms + 写锁有限重试 3 次指数退避），多线程并发写单测已覆盖； + 但单机上限仍在，高并发要换 Postgres，见下“仍留”。 +- OAuth 接线脚手架：cloud/oauth.py（state 10 分钟过期一次性、 + 授权 URL 生成、回调验 state + 换 token，token 换取可注入 fake + 客户端，secret 只走环境变量），端点 GET /oauth/{provider}/login + 与 POST /oauth/{provider}/callback 已接；真实应用注册与绑定仍缺。 +- 推送订阅与分发：cloud/push.py + 端点（订阅增删查、分发按 + 偏好过滤、只发状态变化事件），webhook/slack 走可注入 sender， + webpush/apns/fcm 只做接口 + 缺凭证明确报错；真实凭证仍缺。 + +仍留（需要外部条件，不是代码能假装解决的）： +- 常驻主机 / 托管服务（VM 回收问题没变）。 +- GitHub 远端 + PAT（真实 PR 仍缺这两样）。 +- Slack App（真实签名密钥、斜杠命令、事件订阅）。 +- 云容器账号（CloudExecutor 仍是占位，Docker 未接通挂载 / + 网络隔离 / 资源限额）。 +- 模型 key 供给方式（定时/worker 进程手里没有中转站 key， + 上次 real-nightly-tests 就卡在这里）。 +- OAuth 真实应用注册、回调公网地址、已有账号绑定流程。 +- 推送真实证书（VAPID / APNs / FCM）与真实发送端。 +- worker 执行、云端执行、真实通知与真实模型回合的接通； + memory/KB/成本/模型路由的 HTTP 端点已接，但 worker 回写 + completed/merged 与成本记录仍需接线。 + ## 需要用户定的事（现在就可以定，不用等代码） 1. 控制面跑在哪：常驻机器（自己的 Mac/服务器）还是