diff --git a/docs/05-automation-ingestion-review.md b/docs/05-automation-ingestion-review.md new file mode 100644 index 00000000..e29e2c8d --- /dev/null +++ b/docs/05-automation-ingestion-review.md @@ -0,0 +1,265 @@ +# 自动化内容摄取机制 — 全面 Review 与优化方案 + +> 范围:所有"自动往库里新增/刷新内容"的路径 —— GitHub 项目抓取、托管 A2A 代理导入、 +> 用户提交、技能抓取,以及配套的健康探测、场景分类、AgentCard 校验、cron 鉴权。 +> 每条问题都经过代码级复现验证,并附**正反方辩论**与**裁决**。 +> 生成日期:2026-07-11。 + +--- + +## 0. 机制全景 + +内容进入库有三条自动/半自动管道,核心逻辑收敛在 `src/lib/` 由 cron 端点与 CLI 脚本共享: + +| 管道 | 触发入口 | 共享逻辑 | 内容 | 上线策略 | +|------|----------|----------|------|----------| +| GitHub 项目抓取 | `api/cron/scrape-agents` · `scripts/scrape-github-agents.ts` | `lib/scrape-agents.ts` | Agent `kind=PROJECT` | **直接 APPROVED** | +| 托管 A2A 导入 | `api/cron/import-hosted` · `scripts/import-hosted-agents.ts` | `lib/import-hosted.ts` | Agent `kind=HOSTED` | 默认 PENDING(待审) | +| 用户提交 | `api/agents/submit` | — | Agent `HOSTED` | 普通用户 PENDING / admin key APPROVED | +| 技能抓取 | `scripts/scrape-github-skills.ts` 等(手动) | — | Skill `source=GITHUB_SCRAPE` | **直接 APPROVED** | + +配套:`api/cron/health`(探活 HOSTED)、`lib/scenarios.ts`(确定性关键词场景分类)、 +`lib/agentcard.ts`(AgentCard 拉取 + zod 校验 + 基础 SSRF 防护)、`lib/cron-auth.ts` +(`CRON_SECRET` 鉴权,未配置即 fail-closed)。 + +**结构性优点(保留)**:CLI/cron 共用同一 lib、全链路幂等 upsert、AgentCard 有 +大小/超时/schema 三重校验、GitHub 抓取带限流退避、star-farm 启发式过滤、第三方 +hosted agent 默认进审核队列、cron 鉴权 fail-closed。架构底子是健康的,以下是需要 +收口的具体缺陷。 + +--- + +## 1. 🔴 严重 · import-hosted 可劫持已上架条目(slug 碰撞覆盖) + +### 结论:CONFIRMED + +`lib/import-hosted.ts` 的 upsert 按 `slugify(card.name)` 定位。**update 分支不校验 +现有条目的 `kind` / `publisherId`,不把 `status` 重置为 PENDING**,却会覆盖 +`name / description / endpointUrl / cardUrl / securitySchemes`。"默认 PENDING 待审" +只对 **create** 生效,对 **update** 完全失效。 + +### 证据与复现 + +- `lib/import-hosted.ts:103-133`:update 分支写入 endpointUrl/cardUrl 等,无任何身份/状态守卫。 +- 默认注册表 `DEFAULT_REGISTRY = prassanna-ravishankar/a2a-registry`(第三方仓库,任何人可 PR)。 +- slugify 一致性验证:`lib/utils.ts` 的 `slugify`(用户提交用)与 `import-hosted.ts` 内联 + `slugify`(≤80 字符时)对同一 name 产出**完全相同**的 slug。 +- `api/registry` 对 `status=APPROVED` 的条目直接输出 `endpoint`(`registry/route.ts:69`)。 + +**复现路径**:某用户提交 HOSTED "Acme Assistant"(slug `acme-assistant`,admin 审核通过 → +APPROVED)。攻击者向社区注册表 PR 一张 name="Acme Assistant" 的 card → 下次 `import-hosted` +cron:slug 命中该已上架条目 → 走 update → `endpointUrl`/`cardUrl` 被静默替换为攻击者服务, +**且保持 APPROVED 直接对外**。同理可撞 `kind=PROJECT` 条目(card 命名 "vercel next.js" → +`vercel-next-js`),把"仅发现"的开源项目条目翻成一个可调用的恶意 endpoint。 + +### 正方(应修) + +- 这是**内容完整性 + 供应链**双重风险:外部一个 PR 就能改写我们已背书条目的执行端点。 +- registry 是产品的核心承诺("one API to discover all agents"),被投毒直接摧毁信任。 +- 修复面很小(集中在一个 update 分支加守卫),收益/风险比极高。 + +### 反方(可不修 / 风险) + +- 需攻击者知道目标 slug 且社区注册表 PR 被合并 —— 有一定门槛,非零日。 +- 加 publisher/origin 守卫可能挡掉**合法的域名迁移**(agent 换了托管域名),导致刷新失败。 +- 目前生产上 import-hosted 是否在定时跑、注册表是否可写,取决于外部调度(见 #7), + 未跑时风险是潜伏的而非活跃的。 + +### 裁决:修,采用"所有权 + 同源"双守卫 + +在 upsert 前 `findUnique(slug)`,按以下规则: +1. `existing.publisherId !== pub.id` → **跳过**(绝不触碰用户/他人拥有的行)。 +2. `existing.kind !== "HOSTED"` → **跳过**(hosted card 不得覆盖 PROJECT)。 +3. 同源(`origin(existing.cardUrl) === origin(new.cardUrl)`)→ 正常刷新,保留 status。 +4. 异源(域名变更)→ 允许更新字段但 **status 降级为 PENDING** 并 log,让审核台重新裁决 —— + 既不阻断合法迁移(反方顾虑),又不让换域后的端点继承旧背书。 +- 反方的"迁移被挡"由规则 4 化解;门槛低不构成不修理由。 + +--- + +## 2. 🔴 严重 · scrape-agents 撤销人工治理(强制翻回 APPROVED) + +### 结论:CONFIRMED(比初判更重 —— 删除也会被撤销) + +`lib/scrape-agents.ts:161` 的 update 分支**无条件写 `status: "APPROVED"`**;create 分支 +同样以 APPROVED 建行。于是 admin 的两种治理动作都不持久: + +- **REJECT**:admin 把垃圾/低质 repo 标 REJECTED(`admin/agents/[id]` PATCH,schema 允许该值)→ + 下次 cron 走 update → 被翻回 APPROVED。 +- **DELETE**:admin 删除该行 → slug 不存在 → 下次 cron 走 create → 以 APPROVED **重新建出来**。 + +### 证据与复现 + +- `lib/scrape-agents.ts:161` update `{ status: "APPROVED", ... }`;`:164` create `status: "APPROVED"`。 +- `schemas.ts:85` `adminAgentUpdateSchema.status` 含 `REJECTED / DISABLED`;`admin/agents/[id]/route.ts` PATCH 落库。 +- 与 memory 记录直接相关:prod `/api/registry` 顶部混入 star-farm/flagged 条目 —— 人工清理会被自动化持续覆盖。 +- 触发条件:该 repo 仍在 GitHub 搜索结果内且通过 `minStars` + `looksStarFarmed` 过滤(即"看起来正常但被人工判低质"的那类,恰恰最容易复发)。 + +### 正方(应修) + +- 自动化**不应有权撤销人类的显式裁决**,这是治理系统的底线。 +- 直接解释了 memory 里长期存在的"registry 顶部混垃圾"现象,是根因而非表象。 + +### 反方(可不修 / 风险) + +- 若某 repo 被误拒,"自动复活"反而是一种自愈。 +- 保留 REJECTED 墓碑意味着爬虫要多查一次状态(2000 行 × findUnique),有成本。 +- 只改 update、不动 create 的话,删除仍会复活 —— 要彻底需要墓碑机制,复杂度上升。 + +### 裁决:修,墓碑集合 + update 不碰 status + +1. 循环前一次性查出 `status ∈ {REJECTED, DISABLED}` 的 slug 集合(单查询,规避 2000 次往返 —— 化解反方成本顾虑)。 +2. 命中墓碑的 repo **跳过**(尊重 REJECT 与 DELETE-后-被拒 的人工意志)。 +3. update 分支**移除 `status`**,只刷 `stars / description / scenarios`;新发现仍以 create=APPROVED 上线。 +- "误拒自愈"由 admin 手动改回 APPROVED 即可,不该由爬虫代劳。 + +--- + +## 3. 🟠 中 · cron 抓取无耗时可观测 + 超时留下半写(初判"必然超时"已更正) + +### 结论:CONFIRMED(降级 —— 非"默认必然超时",而是"不可观测 + 非事务半写") + +初版说"300s 几乎必然超时"**高估了**。实测口径:默认 `pages=1`、20 条 query、 +`reqDelayMs=900` → 约 18s 搜索 + 最多 2000 次串行 upsert(prod ~15ms/次 ≈ 30s), +合计约 ~50s,**默认配置下并不必然超时**。真正的缺陷是另外两点: + +1. `export const maxDuration = 300` 在 Cloud Run `output:"standalone"` 下**大概率是空操作** —— + 实际上限是 Cloud Run 服务的 request timeout(默认 300s),`maxDuration` 是 Vercel 语义。 + 即代码里那行给不了它声称的保护。 +2. `pages=2~3` 或触发 GitHub 限流(每次等待上限 120s)时,总耗时可逼近/超过服务超时; + 一旦被切断,由于**逐条 upsert 非事务**,会留下部分写入,且响应无耗时/进度字段,**无法判断某次 run 是否只完成了一半**。 + +### 证据与复现 + +- `api/cron/scrape-agents/route.ts:31` `reqDelayMs: 900`;`:11` `maxDuration = 300`。 +- `next.config.ts` `output: "standalone"`(Cloud Run),无 route-timeout 处理 → maxDuration 非有效上限。 +- `lib/scrape-agents.ts` 逐条 `await db.agent.upsert`,无事务、无批处理、响应无 `durationMs`。 + +### 正方(应修) + +- 静默半失败最坏:每天定时任务悄悄只跑一半,没人发现。 +- 加耗时/计数遥测成本极低,收益是可观测性。 + +### 反方(可不修 / 风险) + +- 幂等 upsert 意味着"半写"下次会补齐,数据不会损坏,只是当次不全。 +- 真正的时限治理应在 Cloud Run 服务配置层(调 timeout),代码层改动有限。 + +### 裁决:低成本收口(不做大重构) + +- 三个 cron 响应统一加 `durationMs`(和已有 `ranAt` 配套),让调度日志能看出耗时趋势与截断。 +- 保留 `maxDuration`(无害),但在注释里点明"Cloud Run 下有效上限来自服务 timeout",避免误导。 +- 事务化/批处理留作后续(反方合理:幂等已兜底数据安全,不紧急)。 + +--- + +## 4. 🟠 高 · AgentCard SSRF:重定向 + DNS 两个绕过口 + +### 结论:CONFIRMED + +`lib/agentcard.ts` 的 `assertSafeUrl` 只对**初始 URL 字符串**做静态私网黑名单,存在两个现实绕过: + +1. **重定向**:`fetchOne` 用 `redirect: "follow"`(`agentcard.ts:134`),302 目标**不再过校验** → + 攻击者用一个公网 URL 302 到 `169.254.169.254`(云元数据)即可。 +2. **DNS**:只比对 hostname 字面量,不解析 IP → `http://x.attacker.com` 解析到内网地址即绕过(DNS rebinding)。 + +该链路对**公开的用户提交端点**(`api/agents/submit` → `fetchAgentCard`)同样生效,不止 cron。 +代码 §11 注释已自认是 best-effort TODO,但端点是公开可达的,风险是现实的。 + +### 证据与复现 + +- `agentcard.ts:62-86` `assertSafeUrl` 仅字符串匹配;`:131-135` `redirect: "follow"`。 +- `api/agents/submit/route.ts:55` 未认证前置的 `fetchAgentCard(input.cardUrl)`(有速率限制,但任意登录用户可打)。 + +### 正方(应修) + +- 云元数据端点泄露 = 潜在凭证泄露,SSRF 属高危类。 +- 修法成熟(手动逐跳重定向 + 解析后 IP 校验),不影响正常 card 抓取。 + +### 反方(可不修 / 风险) + +- 彻底防护需 socket 级(connect 时校验 IP)才能杜绝 check-then-connect 的 TOCTOU,纯应用层是"大幅缓解"非"根治"。 +- 严格 DNS 校验可能误伤某些走内网 DNS 但合法的自托管 agent(生产环境少见)。 +- 生产 egress 若已有网络层限制(VPC/防火墙),应用层是纵深防御而非唯一防线。 + +### 裁决:修,实现"手动逐跳 + 每跳 DNS/IP 私网校验" + +- `redirect: "manual"`,自己跟随(上限 3 跳),每跳 `Location` 都过 `assertSafeUrl` + DNS 解析校验。 +- 新增 `assertPublicHost(hostname)`:解析所有 A/AAAA,任一落在私网/环回/链路本地即拒。 +- 明确记录**残留风险**:connect 级 TOCTOU 未根治(反方点),留作 socket-level 后续;当前修复关闭了两个可被单请求利用的现实口子,是纵深防御的应用层一环。 + +--- + +## 5. 🟡 轻 · star-farm 阈值与实测星农 band 不一致(初判"正则重编译"已撤回) + +### 结论:大部分 RETRACTED,仅保留一个可调参数 + +- **撤回**:初版称 `classifyScenarios` "每次调用重编译 20×N 正则" —— **错误**。`scenarios.ts:294` + `MATCHERS` 是**模块级 const,只编译一次**;每次调用只做 `.test()`。无性能缺陷,收回该条。 +- **保留(轻)**:`looksStarFarmed`(`scrape-agents.ts:52-56`)对 `stars < 150` 直接放行不判 fork 比; + 而 memory 记录实测星农 band 在 ~340★ 附近。150 的地板可能仍放进一部分星农 repo。 + +### 正反方 + +- 正方:调高地板或对 150–400★ 区间也看 fork 比,能少放一点垃圾进顶部。 +- 反方:阈值本质是经验值,收紧有**误伤真实小项目**的风险;且 #2 修好后,人工 REJECT 会持久生效, + 自动阈值的重要性下降 —— 用"人工兜底"替代"参数精调"更稳。 + +### 裁决:本轮不改参数,依赖 #2 的人工治理持久化 + +仅在代码注释里补一行,提示 150 地板与观测到的 ~340 band 的关系,留给后续按数据调。不动逻辑,避免误伤。 + +--- + +## 6. 🟡 低 · health cron 无时限 + 静默吞错 + +### 结论:CONFIRMED(低) + +- `api/cron/health/route.ts` **无 `maxDuration`**;HOSTED 增多后 `PROBE_CONCURRENCY=8` × 8s 超时可能撞默认时限。 +- `lib/health.ts:53` `.catch(() => {})` 把写库失败完全吞掉 → 排障时看不到失败。 + +### 正反方 + +- 正方:探测量会随 hosted agent 增长,现在加时限 + 错误计数几乎零成本。 +- 反方:当前 hosted 数量少,离撞限很远;吞错只影响可观测性,不影响正确性。 + +### 裁决:顺手修(零风险) + +加 `maxDuration=120`;`.catch` 改为累加 `writeErrors` 计数并计入 summary,不再静默。 + +--- + +## 7. 🟠 中(流程) · 调度器配置不在代码库 + +### 结论:CONFIRMED + +三个 cron 全靠外部 Cloud Scheduler 触发(HANDOFF 提到 `tako-cron-secret`),但仓库内 +**无任何调度配置**(无 `vercel.json`;`cloudbuild.yaml` 按 memory 是 stale 模板)。 +"哪些 job 在跑、频率、参数(尤其 `?pages=`/`?minStars=`/`?max=`)"完全是带外知识, +无法 review、无法版本化、迁移/重装极易漏配或配错。 + +### 正反方 + +- 正方:调度是自动化的"启动开关",不进版本库等于生产行为不可复现。 +- 反方:Cloud Scheduler 配置含项目 ID 等环境细节,直接入库需注意不泄密;且它属于基础设施而非应用代码。 + +### 裁决:补一份可复现的调度说明 + 幂等建置脚本 + +新增 `docs/06-cron-schedule.md`:列出每个 job 的 URL/频率/参数/鉴权,并给出**幂等的 `gcloud scheduler` 建置命令** +(密钥用占位符从 Secret Manager 取,不硬编码)。不碰 stale 的 cloudbuild.yaml。 + +--- + +## 8. 实施优先级与顺序 + +| 优先级 | 问题 | 动作 | 风险 | +|--------|------|------|------| +| P0 | #1 import 劫持 | 所有权 + 同源守卫,异源降级 PENDING | 低(只加守卫) | +| P0 | #2 撤销治理 | 墓碑集合跳过 + update 不碰 status | 低 | +| P1 | #4 SSRF | 手动逐跳重定向 + DNS/IP 私网校验 | 中(需测正常 card 仍可拉) | +| P2 | #3 可观测 | 三 cron 加 `durationMs`;注释澄清 maxDuration | 极低 | +| P2 | #6 health | `maxDuration` + 错误计数不吞 | 极低 | +| P3 | #7 调度 | 新增 cron 调度文档 + gcloud 幂等脚本 | 无(纯文档) | +| — | #5 阈值 | 仅补注释,不改逻辑 | 无 | + +验证:每步后 `npx tsc --noEmit` + `npm run lint`;SSRF 改动额外跑一次真实 card 拉取回归。 diff --git a/docs/06-cron-schedule.md b/docs/06-cron-schedule.md new file mode 100644 index 00000000..91dea8ec --- /dev/null +++ b/docs/06-cron-schedule.md @@ -0,0 +1,76 @@ +# Cron 调度 — 权威清单与幂等建置 + +> 三个自动摄取端点全靠 **Cloud Scheduler** 触发,但调度配置此前不在代码库里 +> (无 `vercel.json`,`cloudbuild.yaml` 是 stale 模板)。本文件是它们的**唯一权威来源**: +> 频率、参数、鉴权,以及可复现的 `gcloud` 建置命令。改调度请改这里并同步执行。 +> 关联:`docs/05-automation-ingestion-review.md` §7。 + +## 端点一览 + +| Job | 端点 | 建议频率 | 参数 | 落库策略 | +|-----|------|----------|------|----------| +| `tako-scrape-agents` | `GET /api/cron/scrape-agents` | 每日 04:00 UTC | `?pages=1&minStars=200&max=2000` | PROJECT 直接 APPROVED;admin REJECT/DISABLE 的 slug 会被跳过 | +| `tako-import-hosted` | `GET /api/cron/import-hosted` | 每日 04:30 UTC | `?max=0`(不限) | HOSTED 落 PENDING,待 `/admin/agents` 审核;跨域变更也降级 PENDING | +| `tako-health` | `GET /api/cron/health` | 每 6 小时 | —— | 更新 `healthStatus`/`healthCheckedAt` | + +所有端点鉴权:`Authorization: Bearer `(见 `src/lib/cron-auth.ts`; +`CRON_SECRET` 未配置时端点 fail-closed 返回 401)。响应含 `durationMs` + `ranAt`, +调度日志据此判断是否接近超时 / 被截断。 + +## 建置(幂等 create-or-update) + +```bash +set -euo pipefail +PROJECT=takoapi-491505 +REGION=us-central1 +LOCATION=us-central1 # Cloud Scheduler location +BASE=https://takoapi.com # 或 run URL: https://takoapi-429522911261.us-central1.run.app + +# CRON_SECRET 来源(二选一,取决于是否启用 Secret Manager — 见 docs/00-infrastructure.md): +# a) Secret Manager: +SECRET=$(gcloud secrets versions access latest --secret=tako-cron-secret --project="$PROJECT") +# b) 若 Secret Manager 未启用,从 Cloud Run 服务 env 读取当前值: +# SECRET=$(gcloud run services describe takoapi --region "$REGION" --project="$PROJECT" \ +# --format='value(spec.template.spec.containers[0].env)' | tr ',' '\n' | grep -A1 CRON_SECRET | tail -1) + +# 幂等 upsert:create 失败(已存在)即改 update。 +upsert_job () { + local name="$1" schedule="$2" uri="$3" + local verb=create + gcloud scheduler jobs describe "$name" --location="$LOCATION" --project="$PROJECT" >/dev/null 2>&1 && verb=update + gcloud scheduler jobs "$verb" http "$name" \ + --location="$LOCATION" --project="$PROJECT" \ + --schedule="$schedule" --time-zone="Etc/UTC" \ + --uri="$uri" --http-method=GET \ + --headers="Authorization=Bearer ${SECRET}" \ + --attempt-deadline=320s \ + --max-retry-attempts=1 +} + +upsert_job tako-scrape-agents "0 4 * * *" "${BASE}/api/cron/scrape-agents?pages=1&minStars=200&max=2000" +upsert_job tako-import-hosted "30 4 * * *" "${BASE}/api/cron/import-hosted?max=0" +upsert_job tako-health "0 */6 * * *" "${BASE}/api/cron/health" +``` + +> `--attempt-deadline=320s` 略高于 scrape-agents 的 `maxDuration=300`,以免 Scheduler 先于端点超时。 +> **Cloud Run 服务的 request timeout 必须 ≥ 320s**(否则那才是真正的截断点);核对: +> `gcloud run services describe takoapi --region us-central1 --project=takoapi-491505 --format='value(spec.template.spec.timeoutSeconds)'`。 + +## 运维 + +```bash +# 列出 / 查看 +gcloud scheduler jobs list --location=us-central1 --project=takoapi-491505 +# 手动触发一次(冒烟) +gcloud scheduler jobs run tako-health --location=us-central1 --project=takoapi-491505 +# 轮换 CRON_SECRET 后,重跑上面的 upsert_job 即可刷新 header +# 暂停 / 恢复 +gcloud scheduler jobs pause tako-scrape-agents --location=us-central1 --project=takoapi-491505 +gcloud scheduler jobs resume tako-scrape-agents --location=us-central1 --project=takoapi-491505 +``` + +## 安全备注 + +- header 里带的是明文 bearer secret;`gcloud scheduler jobs describe` 可见,权限同项目其他密钥。 + 若日后启用 OIDC(service account),可改用 `--oidc-service-account-email` 并把 `cron-auth.ts` 换成校验 OIDC token。 +- **别把 `SECRET` 硬编码进本文件或提交历史**;上面的命令都从 Secret Manager / 运行时 env 现取。 diff --git a/src/app/api/cron/health/route.ts b/src/app/api/cron/health/route.ts index 5610c5a0..7f755a70 100644 --- a/src/app/api/cron/health/route.ts +++ b/src/app/api/cron/health/route.ts @@ -6,13 +6,16 @@ import { isAuthorizedCron } from "@/lib/cron-auth"; // healthStatus / healthCheckedAt. Wire to Cloud Scheduler, which sends // `Authorization: Bearer ` (the tako-cron-secret value). export const dynamic = "force-dynamic"; +// Bounded probes (8-wide, 8s each) — give the run headroom as HOSTED count grows. +export const maxDuration = 120; async function handle(req: NextRequest) { if (!isAuthorizedCron(req)) { return NextResponse.json({ error: "Unauthorized" }, { status: 401 }); } + const startedAt = Date.now(); const summary = await runHealthChecks(); - return NextResponse.json(summary); + return NextResponse.json({ ...summary, durationMs: Date.now() - startedAt }); } export const GET = handle; diff --git a/src/app/api/cron/import-hosted/route.ts b/src/app/api/cron/import-hosted/route.ts index 3d5c650a..7d083041 100644 --- a/src/app/api/cron/import-hosted/route.ts +++ b/src/app/api/cron/import-hosted/route.ts @@ -21,13 +21,14 @@ async function handle(req: NextRequest) { if (!isAuthorizedCron(req)) { return NextResponse.json({ error: "Unauthorized" }, { status: 401 }); } + const startedAt = Date.now(); const sp = new URL(req.url).searchParams; // 0 = no cap. Bounded to protect the request from an over-large registry. const max = intParam(sp.get("max"), 0, 0, 1000); try { // Default status PENDING — third-party agents await admin review. const result = await importHostedAgents({ max, status: "PENDING" }, prisma); - return NextResponse.json({ ...result, max, ranAt: new Date().toISOString() }); + return NextResponse.json({ ...result, max, durationMs: Date.now() - startedAt, ranAt: new Date().toISOString() }); } catch (e) { return NextResponse.json({ error: e instanceof Error ? e.message : "import failed" }, { status: 500 }); } diff --git a/src/app/api/cron/scrape-agents/route.ts b/src/app/api/cron/scrape-agents/route.ts index a25f5a43..39de644d 100644 --- a/src/app/api/cron/scrape-agents/route.ts +++ b/src/app/api/cron/scrape-agents/route.ts @@ -8,6 +8,10 @@ import { scrapeGithubAgents } from "@/lib/scrape-agents"; // catalog fresh and adds newly-popular repos. Wire to Cloud Scheduler with // `Authorization: Bearer `. Tunable via ?pages=&minStars=&max=. export const dynamic = "force-dynamic"; +// NOTE: on Cloud Run (output: "standalone") the effective ceiling is the +// service's request timeout, not this Vercel-style export — keep the Cloud Run +// timeout ≥ this. `durationMs` in the response lets the scheduler log spot runs +// that approach the limit (a truncated run leaves partial, non-transactional writes). export const maxDuration = 300; function intParam(v: string | null, def: number, min: number, max: number): number { @@ -20,6 +24,7 @@ async function handle(req: NextRequest) { if (!isAuthorizedCron(req)) { return NextResponse.json({ error: "Unauthorized" }, { status: 401 }); } + const startedAt = Date.now(); const sp = new URL(req.url).searchParams; // Defaults match the live catalog (PAGES=1 keeps it fast enough for a sync // request; raise pages for a deeper backfill). Bounded to protect the request. @@ -31,7 +36,14 @@ async function handle(req: NextRequest) { { token: process.env.GITHUB_TOKEN, pages, minStars, maxAgents, reqDelayMs: 900 }, prisma ); - return NextResponse.json({ ...result, pages, minStars, maxAgents, ranAt: new Date().toISOString() }); + return NextResponse.json({ + ...result, + pages, + minStars, + maxAgents, + durationMs: Date.now() - startedAt, + ranAt: new Date().toISOString(), + }); } catch (e) { return NextResponse.json({ error: e instanceof Error ? e.message : "scrape failed" }, { status: 500 }); } diff --git a/src/lib/agentcard.ts b/src/lib/agentcard.ts index da109906..5c1e96e8 100644 --- a/src/lib/agentcard.ts +++ b/src/lib/agentcard.ts @@ -1,4 +1,5 @@ import { z } from "zod"; +import { lookup } from "node:dns/promises"; // ── A2A AgentCard parsing ──────────────────────────────────────────────── // Fetches and validates an A2A AgentCard (the canonical machine-readable @@ -13,6 +14,7 @@ const WELL_KNOWN_LEGACY = "/.well-known/agent.json"; // pre-v1 path, still in th const MAX_BYTES = 256 * 1024; const TIMEOUT_MS = 8000; +const MAX_REDIRECTS = 3; const AgentSkillSchema = z.object({ id: z.string().min(1).max(200), @@ -57,8 +59,12 @@ export type ParsedAgentCard = { }[]; }; -// Best-effort SSRF guard. Full protection needs DNS-resolution checks -// (a hardening TODO, see docs/agent-marketplace/03-technical-architecture.md §11). +// SSRF guard — first line: reject non-http(s) and literal private/loopback +// hosts from the URL string. Paired with assertPublicHost() (DNS/IP resolution) +// and manual per-hop redirect revalidation in fetchOne(), this closes the +// redirect-to-metadata and DNS-rebinding vectors. Residual: a connect-level +// TOCTOU (rebind between resolve and connect) still needs socket-level pinning — +// see docs/agent-marketplace/03-technical-architecture.md §11. function assertSafeUrl(raw: string): URL { let u: URL; try { @@ -85,6 +91,55 @@ function assertSafeUrl(raw: string): URL { return u; } +// True if an IPv4 literal is in a private / loopback / link-local / CGNAT range. +// Unparseable input is treated as unsafe (fail closed). +function ipv4IsPrivate(ip: string): boolean { + const p = ip.split(".").map((n) => Number(n)); + if (p.length !== 4 || p.some((n) => !Number.isInteger(n) || n < 0 || n > 255)) return true; + const [a, b] = p; + if (a === 0 || a === 127) return true; // 0.0.0.0/8, loopback + if (a === 10) return true; // 10/8 + if (a === 172 && b >= 16 && b <= 31) return true; // 172.16/12 + if (a === 192 && b === 168) return true; // 192.168/16 + if (a === 169 && b === 254) return true; // 169.254/16 link-local (cloud metadata) + if (a === 100 && b >= 64 && b <= 127) return true; // 100.64/10 CGNAT + if (a === 198 && (b === 18 || b === 19)) return true; // 198.18/15 benchmarking, non-routable + return false; +} + +// True if a resolved address (v4 or v6) is one we must not fetch from. +function ipIsPrivate(address: string, family: number): boolean { + if (family === 4) return ipv4IsPrivate(address); + const ip = address.toLowerCase(); + // IPv4-mapped / -embedded (e.g. ::ffff:169.254.169.254) → check the v4 part. + const embedded = ip.match(/((?:\d{1,3}\.){3}\d{1,3})$/); + if (embedded) return ipv4IsPrivate(embedded[1]); + if (ip === "::1" || ip === "::") return true; // loopback / unspecified + if (ip.startsWith("fc") || ip.startsWith("fd")) return true; // fc00::/7 unique-local + if (/^fe[89ab]/.test(ip)) return true; // fe80::/10 link-local + return false; +} + +// Resolve a hostname and refuse if ANY resolved address is private/loopback/ +// link-local. Closes the DNS-rebinding gap that the string-only host check in +// assertSafeUrl can't catch (e.g. a public name pointing at 169.254.169.254). +// Prod-only, matching assertSafeUrl's guard, so localhost cards still work in dev. +async function assertPublicHost(hostname: string): Promise { + if (process.env.NODE_ENV !== "production") return; + let addrs: { address: string; family: number }[]; + try { + addrs = await lookup(hostname, { all: true }); + } catch { + throw new AgentCardError(`Could not resolve ${hostname}`); + } + if (!addrs.length) throw new AgentCardError(`Host ${hostname} did not resolve`); + for (const a of addrs) { + if (ipIsPrivate(a.address, a.family)) { + throw new AgentCardError("Refusing to fetch a private/loopback address"); + } + } +} + function originOf(u: string): string { try { const x = new URL(u); @@ -123,33 +178,55 @@ function normalize(card: z.infer, cardUrl: string): Pars } async function fetchOne(cardUrl: string): Promise { - assertSafeUrl(cardUrl); + // Store the URL we were asked to fetch as the card's canonical URL, even if a + // redirect serves the bytes from elsewhere (keeps stored cardUrl stable). + const requested = assertSafeUrl(cardUrl).toString(); + let current = requested; + // One timer bounds the whole redirect chain, not each hop. const ctrl = new AbortController(); const timer = setTimeout(() => ctrl.abort(), TIMEOUT_MS); - let res: Response; try { - res = await fetch(cardUrl, { - signal: ctrl.signal, - headers: { Accept: "application/json" }, - redirect: "follow", - }); - } catch { - throw new AgentCardError(`Could not reach ${cardUrl}`); + for (let hop = 0; ; hop++) { + // Revalidate every hop: string checks + DNS/IP resolution. A "follow" + // fetch would let a public URL redirect into a private address unchecked. + const u = assertSafeUrl(current); + await assertPublicHost(u.hostname); + + let res: Response; + try { + res = await fetch(current, { + signal: ctrl.signal, + headers: { Accept: "application/json" }, + redirect: "manual", + }); + } catch { + throw new AgentCardError(`Could not reach ${current}`); + } + + if (res.status >= 300 && res.status < 400) { + const loc = res.headers.get("location"); + if (!loc) throw new AgentCardError("Redirect without a Location header"); + if (hop >= MAX_REDIRECTS) throw new AgentCardError("Too many redirects"); + current = new URL(loc, current).toString(); // resolve relative redirects + continue; + } + + if (!res.ok) throw new AgentCardError(`AgentCard fetch returned HTTP ${res.status}`); + const text = await res.text(); + if (text.length > MAX_BYTES) throw new AgentCardError("AgentCard response too large"); + let json: unknown; + try { + json = JSON.parse(text); + } catch { + throw new AgentCardError("AgentCard is not valid JSON"); + } + const parsed = AgentCardSchema.safeParse(json); + if (!parsed.success) throw new AgentCardError("Not a valid A2A AgentCard"); + return normalize(parsed.data, requested); + } } finally { clearTimeout(timer); } - if (!res.ok) throw new AgentCardError(`AgentCard fetch returned HTTP ${res.status}`); - const text = await res.text(); - if (text.length > MAX_BYTES) throw new AgentCardError("AgentCard response too large"); - let json: unknown; - try { - json = JSON.parse(text); - } catch { - throw new AgentCardError("AgentCard is not valid JSON"); - } - const parsed = AgentCardSchema.safeParse(json); - if (!parsed.success) throw new AgentCardError("Not a valid A2A AgentCard"); - return normalize(parsed.data, cardUrl); } /** diff --git a/src/lib/health.ts b/src/lib/health.ts index b9a7dcc3..7881bf7a 100644 --- a/src/lib/health.ts +++ b/src/lib/health.ts @@ -30,7 +30,15 @@ export async function probeAgentHealth(agent: { cardUrl: string | null; endpoint } } -export type HealthSummary = { checked: number; ok: number; degraded: number; down: number; checkedAt: string }; +export type HealthSummary = { + checked: number; + ok: number; + degraded: number; + down: number; + /** Persist failures — surfaced instead of being silently swallowed. */ + writeErrors: number; + checkedAt: string; +}; /** Probe all APPROVED HOSTED agents (bounded concurrency) and persist their health. */ export async function runHealthChecks(): Promise { @@ -40,7 +48,7 @@ export async function runHealthChecks(): Promise { }); const now = new Date(); - const summary: HealthSummary = { checked: agents.length, ok: 0, degraded: 0, down: 0, checkedAt: now.toISOString() }; + const summary: HealthSummary = { checked: agents.length, ok: 0, degraded: 0, down: 0, writeErrors: 0, checkedAt: now.toISOString() }; for (let i = 0; i < agents.length; i += PROBE_CONCURRENCY) { const batch = agents.slice(i, i + PROBE_CONCURRENCY); @@ -48,9 +56,13 @@ export async function runHealthChecks(): Promise { await Promise.all( results.map((r) => { summary[r.health]++; + // Don't let one failed write abort the batch, but do count it so a + // systemic DB problem shows up in the summary rather than vanishing. return prisma.agent .update({ where: { id: r.id }, data: { healthStatus: r.health, healthCheckedAt: now } }) - .catch(() => {}); + .catch(() => { + summary.writeErrors++; + }); }) ); } diff --git a/src/lib/import-hosted.ts b/src/lib/import-hosted.ts index d0ed5d03..f7acec3c 100644 --- a/src/lib/import-hosted.ts +++ b/src/lib/import-hosted.ts @@ -20,6 +20,18 @@ function slugify(s: string): string { return s.toLowerCase().replace(/[^a-z0-9]+/g, "-").replace(/(^-|-$)/g, "").slice(0, 80); } +// Origin (scheme://host) of a URL, for comparing an existing card's host with an +// incoming one. Returns null when the input can't be parsed. +function originOf(u: string | null | undefined): string | null { + if (!u) return null; + try { + const x = new URL(u); + return `${x.protocol}//${x.host}`; + } catch { + return null; + } +} + const dedupe = (a: string[]) => Array.from(new Set(a)); // Resolve the list of AgentCard URLs from a local file, a registry index URL, @@ -94,6 +106,35 @@ export async function importHostedAgents( failed++; continue; } + + // Ownership + same-origin guards. The registry importer upserts by + // slugify(card.name); without these, a card whose name collides with an + // existing entry would silently overwrite that entry's endpoint (and keep + // its APPROVED status). Since the default registry is a community-editable + // list, that's a takeover vector. Rules: + // 1. Only refresh rows this importer owns (publisherId === pub.id) — never + // touch a user's submitted agent. + // 2. A hosted card must not overwrite a non-HOSTED (PROJECT) entry. + // 3. If the card's host differs from the stored one, treat it as a + // possible takeover / migration: refresh fields but drop to PENDING so + // an admin re-approves the new endpoint before it serves traffic. + const existing = await db.agent.findUnique({ + where: { slug }, + select: { kind: true, publisherId: true, cardUrl: true }, + }); + if (existing && existing.publisherId !== pub.id) { + log(`↷ ${slug} — owned by another publisher; skipping`); + failed++; + continue; + } + if (existing && existing.kind !== "HOSTED") { + log(`↷ ${slug} — existing ${existing.kind} entry; a hosted card must not overwrite it; skipping`); + failed++; + continue; + } + const existingOrigin = originOf(existing?.cardUrl); + const crossOrigin = !!existingOrigin && existingOrigin !== originOf(card.cardUrl); + const scenarios = classifyScenarios( [card.name, card.description, card.skills.map((s) => s.name).join(" ")].join(" ") ); @@ -112,6 +153,8 @@ export async function importHostedAgents( securitySchemes: security, cardFetchedAt: new Date(), scenarios, + // Endpoint moved to a new host → force re-review before it's public. + ...(crossOrigin ? { status: "PENDING" as AgentStatus } : {}), }, create: { slug, @@ -151,7 +194,10 @@ export async function importHostedAgents( } } - log(`✓ ${slug} — ${card.skills.length} skill(s), scenarios: ${scenarios.join("/") || "none"}`); + log( + `✓ ${slug} — ${card.skills.length} skill(s), scenarios: ${scenarios.join("/") || "none"}` + + (crossOrigin ? " [host changed → status reset to PENDING for re-review]" : "") + ); ok++; } catch (e) { const msg = e instanceof AgentCardError ? e.message : (e as Error).message; diff --git a/src/lib/scrape-agents.ts b/src/lib/scrape-agents.ts index 87e32360..0ae07de6 100644 --- a/src/lib/scrape-agents.ts +++ b/src/lib/scrape-agents.ts @@ -1,4 +1,4 @@ -import type { PrismaClient } from "@prisma/client"; +import type { PrismaClient, AgentStatus } from "@prisma/client"; import { classifyScenarios } from "./scenarios"; // Import top open-source agent PROJECTS from GitHub (by stars) into the registry @@ -51,6 +51,11 @@ type Repo = { // the star-farmed band polluting the newest-by-recency slice of the catalog. function looksStarFarmed(r: Repo): boolean { const stars = r.stargazers_count; + // The 150 floor is deliberately permissive to avoid false positives on small + // real projects. Note: the observed star-farm band sits higher (~340★), so + // some farmed repos in 150–400★ can slip past the fork-ratio test. That's an + // accepted trade-off — the durable backstop is admin REJECT/DISABLE, which a + // scheduled re-scrape now honors (see scenarios loop below), not this heuristic. if (stars < 150) return false; // small repos: don't judge on fork ratio return (r.forks_count ?? 0) / stars < 0.01; // < 1 fork per 100 stars → suspicious } @@ -148,17 +153,33 @@ export async function scrapeGithubAgents(opts: ScrapeOpts, db: PrismaClient): Pr const eligible = starred.filter((r) => !looksStarFarmed(r)); const skipped = starred.length - eligible.length; const repos = eligible.slice(0, max); - log(`found ${all.length} unique repos${minStars ? `, ${starred.length} with >=${minStars} stars` : ""}${skipped ? `, skipped ${skipped} likely star-farmed` : ""}; importing top ${repos.length} by stars`); + + // Respect human governance: never resurrect a repo an admin has REJECTED or + // DISABLED. Load those slugs up front (one query) so the loop can skip them — + // otherwise the update path below would flip them back to APPROVED, and a prior + // deletion would be re-created as APPROVED, silently undoing the moderation. + const repoSlugs = repos.map((r) => slugify(`${r.owner.login}-${r.name}`)); + const governed = await db.agent.findMany({ + where: { slug: { in: repoSlugs }, status: { in: ["REJECTED", "DISABLED"] as AgentStatus[] } }, + select: { slug: true }, + }); + const tombstoned = new Set(governed.map((g) => g.slug)); + + log(`found ${all.length} unique repos${minStars ? `, ${starred.length} with >=${minStars} stars` : ""}${skipped ? `, skipped ${skipped} likely star-farmed` : ""}${tombstoned.size ? `, skipping ${tombstoned.size} admin-rejected/disabled` : ""}; importing top ${repos.length} by stars`); let imported = 0; for (const r of repos) { const slug = slugify(`${r.owner.login}-${r.name}`); + if (tombstoned.has(slug)) continue; // admin REJECTED/DISABLED — leave it alone const description = (r.description || r.name).slice(0, 500); // Classify by name + description + GitHub topics into use-case scenarios. const scenarios = classifyScenarios([r.name, r.description ?? "", (r.topics ?? []).join(" ")].join(" ")); + // NOTE: the update branch deliberately does NOT set `status` — a scheduled + // refresh must not override an admin's moderation decision. Only newly + // discovered repos (the create branch) land as APPROVED. await db.agent.upsert({ where: { slug }, - update: { stars: r.stargazers_count, description, status: "APPROVED", scenarios }, + update: { stars: r.stargazers_count, description, scenarios }, create: { slug, name: r.name,