turbo-agent-auth

turbo agent auth manage tool, to support adjust auth

Project description

turbo-agent-auth

# 回滚最后一次迁移
uv run python -m prisma migrate reset
```

旧版信息已归档

技术栈

FastAPI + Uvicorn: 提供 HTTP API
Typer: 命令行入口
Prisma (Python 包名 prisma): 访问 PostgreSQL（尚未定义具体模型）
uv: 依赖与环境管理
Pytest + httpx(testclient): 基础测试

安装依赖

使用 uv 管理，不要手动修改 pyproject.toml dependencies。

uv sync

运行 API (Typer CLI)

支持以下方式：

# 方式1：直接运行 main.py（初始化阶段用）
uv run python main.py run-api --host 0.0.0.0 --port 8000

# 方式2：模块方式（推荐，使用 __main__）
uv run python -m turbo_agent_auth run-api --host 0.0.0.0 --port 8000

# 方式3：控制台脚本（已配置 project.scripts）
# 推荐：分组命令
uv run turbo-agent-auth serve run --host 0.0.0.0 --port 8000

# 兼容：旧版根级子命令（仍可用）
uv run turbo-agent-auth run-api --host 0.0.0.0 --port 8000

访问健康检查：

curl http://127.0.0.1:8000/health
curl http://127.0.0.1:8000/v1/ping

安全：API Key (Authorization Bearer)

通过环境变量或 CLI 启动参数配置一个全局 API Key；配置后所有 /v1/* 路由需要在请求头携带 Authorization: Bearer <key>，未配置时不启用鉴权（/health 永远开放）。

环境变量：AUTH_API_KEY=yourkey ；或 CLI --api-key yourkey。
请求头：Authorization: Bearer yourkey
失败返回：401 {"detail":"Invalid API key"}

示例：

AUTH_API_KEY=abc123 uv run turbo-agent-auth serve run --host 0.0.0.0 --port 8000
# 或 CLI 参数方式（内部会写入环境变量）
uv run turbo-agent-auth serve run --api-key abc123 --host 0.0.0.0 --port 8000
curl -H 'Authorization: Bearer abc123' http://127.0.0.1:8000/v1/ping

日志

依赖：loguru
环境变量：
- LOG_LEVEL（默认 INFO）
- LOG_FORMAT=json|plain（默认 plain）
- LOG_FILE：设置则输出到文件
- LOG_ROTATION（默认 10 MB）、LOG_RETENTION（默认 7 days）
- LOG_SENSITIVE_DEBUG：设置为 1 且 LOG_LEVEL=DEBUG 时允许输出未脱敏的敏感登录字段（如 password/secret/token），否则仅输出脱敏版
CLI 参数（与环境变量等价，优先级更高）：

uv run turbo-agent-auth serve run \
	--log-level DEBUG \
	--json \
	--log-file logs/app.log \
	--rotation "10 MB" \
	--retention "7 days"

应用内日志工具：from turbo_agent_auth import logger, setup_logging

Prisma 初始化

确认已设置数据库环境变量：DATABASE_URL=postgresql://user:pass@host:5432/dbname
生成客户端（首次或修改 schema 后）：
```
uv run prisma generate
```
迁移（首次将创建/应用迁移）：
```
uv run prisma migrate dev --name init
```

（已实现）模型与迁移：请在设置 DATABASE_URL 后执行迁移与生成客户端。

# 1. 生成客户端
uv run prisma generate

# 2. 应用迁移（首次将创建 auth_init 迁移）
uv run prisma migrate dev --name auth_init

# 从旧版本升级到“多秘钥/外部ID/SMTP/POP3”版本：
uv run prisma migrate dev --name auth_update_multi_secrets_external_ids_smtp_pop3
uv run prisma generate

# 生产/部署环境（非交互）
uv run prisma migrate deploy && uv run prisma generate

# 后续新增：用户绑定与凭证保存/提醒列字段
uv run prisma migrate dev --name secret_user_bind_and_credentials
uv run prisma generate

运行测试

uv run pytest -q

架构概览

turbo-agent-auth 采用三层架构设计，实现了业务逻辑与核心机制的解耦：

核心层 (auth_core.py): 纯粹的认证流程执行引擎，无数据库依赖
- AuthFlowExecutor: 基于生成器的多阶段流程编排器
- parse_response_data: 统一的响应解析引擎（支持 JSON/Headers/Cookies/HTML）
- prepare_login_request_config: 请求配置构建器（模板渲染、参数注入）
服务层 (services/auth_service.py): 数据库驱动的业务服务
- 平台/授权方式/秘钥的 CRUD 操作
- 集成 auth_core 执行登录/刷新流程
- 持久化认证结果与执行轨迹
客户端层 (client_helper.py): 独立的 HTTP 客户端工具
- AuthClient / AsyncAuthClient: 同步/异步认证客户端
- 支持 login() / refresh() / get/post/put/patch/delete 等完整 HTTP 方法
- 无需数据库连接，可在脚本/服务间复用

项目结构

main.py                # Typer CLI 入口
src/turbo_agent_auth/
  app.py               # FastAPI 应用与路由聚合
  auth_core.py         # 核心认证引擎（无DB依赖）
  client_helper.py     # 独立HTTP客户端（支持sync/async）
  logging.py           # 日志配置（setup_logging / patch_standard_logging）
  schemas.py           # Pydantic 模型
  api/
    routes.py          # API 路由（平台/授权方式/秘钥/默认schema）
  services/
    auth_service.py    # 业务服务（集成 auth_core + 数据库）
  config/
    auth_schemas.yaml  # 各授权方式默认字段Schema与解析映射
prisma/schema.prisma   # Prisma 数据模型
tests/test_health.py   # 基础健康与 ping 测试

Docker 构建与部署

已提供示例 Dockerfile（基于 uv 官方 python3.12 debian slim 镜像）与 docker-compose.yml。

Dockerfile 关键点：

使用 uv 安装依赖并生成 Prisma Client。
仅单阶段构建，减小复杂度。
通过环境变量注入 DATABASE_URL 与可选 AUTH_API_KEY。

构建与运行：

docker build -t turbo-agent-auth:latest .
docker run --rm -e DATABASE_URL='postgresql://user:pass@host:5432/dbname' -e AUTH_API_KEY=abc123 -p 8000:8000 turbo-agent-auth:latest

docker-compose 示例（见仓库根目录）：

docker compose up -d --build
curl -H 'X-API-Key: changeme' http://127.0.0.1:8000/v1/ping

Compose 中包含 Postgres 服务与示例环境变量；首次启动后可执行迁移：

docker compose exec auth uv run prisma migrate deploy && docker compose exec auth uv run prisma generate

一键清理：

docker compose down -v

变更摘要（2025-11-20 新增）

启动可配置 API Key（AUTH_API_KEY / CLI --api-key），保护所有 /v1/* 路由（Bearer 格式）。
新增 Dockerfile 与 docker-compose.yml 示例，简化部署与本地联调。
访问需鉴权接口使用 Authorization: Bearer <key>；未配置则保持兼容性不拦截。

变更摘要（2025-11-19）

本次迭代围绕“可观测性 / 生命周期 / 安全策略”完善：

多阶段登录：loginFlow.steps 支持步骤级 responseMapping，可解析 HTML/Headers/Body/Cookies；
步骤间注入：requestPlacements 将上一步解析结果注入下一步请求（headers/query/cookies/body）；
统一过期推导：综合 expires_in / expires_at 与 Set-Cookie 取最早到期写入 Secret.expiresAt；
两类放置明确：登录阶段用 requestPlacements，业务阶段用 authFieldPlacements；
登录可观测性：新增 POST /v1/secrets/{id}/login/trace 与 PATCH /v1/secrets/{id}/with-execution，返回步骤快照与最终解析；普通登录仍用 /login；
首次创建即授权：POST /v1/auth-methods/{id}/secret 返回 secret + execution 复合体；
Secret 唯一性调整：取消 (authMethodId,name) 唯一，仅保留 id 唯一；同名取最新；
失效与通知：新增 /invalidate 与 /invalid-notified；登录失败自动置 isValid=false,status=invalid,invalidNotified=false；
有效性字段：isValid + invalidNotified 支持外部事件闭环与告警去重；
隐私策略：autoLoginEnabled=false 不持久化 loginPayload，登录接口必须即时携带；
敏感字段日志：仅在 LOG_LEVEL=DEBUG 且 LOG_SENSITIVE_DEBUG=1 时输出原文，否则 DEBUG 级别输出脱敏版；步骤执行 INFO / 4xx WARNING / 5xx ERROR；
新增日志辅助函数（内部）：统一输出 login.step / login.step.parsed / login.final / login.failure；
文档：补充“秘钥生命周期与接口选择”（详见 docs/manuals.md）。

设计思路

配置驱动：Schema（表单）、Flow（请求模板）、Mapping（解析）、Placements（注入）组合实现通用授权。
两阶段“放置”语义清晰：
- 登录阶段（steps 间）：requestPlacements。
- 业务阶段（最终请求）：authFieldPlacements。
解析统一：步骤级与顶层 responseMapping 共用解析器，支持 body.jsonPath、headers、cookies（含 Set-Cookie 聚合）、html 选择器/regex。
有效期：综合 token 与 cookie 的到期时间，选取最早者写入 Secret.expiresAt。

系统综述（故事版）

把 Auth 看成一个“可编排的流水线”：你先为某个平台声明一套“如何登录”（loginFlow）和“如何读回结果”（responseMapping）的规则，系统就会按照这套规则去请求远端接口、解析响应、计算过期时间，并把得到的“最终凭证”落到 Secret。整个过程无需写代码，像把乐高块拼起来：

你告诉系统“登录请求长这样”（url/method/headers/query/body_template），或者“先 A 再 B”（steps）；
系统把你在前端表单里填的“登录字段”（loginPayload）代入模板发出请求；
收到响应后，它按 responseMapping 去 JSON/Headers/Cookies/HTML 里找值，拼出你要的字段；
如果是多步流，它把上一步解析到的值，按 requestPlacements 放进下一步请求的 Header/Query/Cookie/Body；
全部步骤跑完，再用顶层 responseMapping 产出最终凭证，存到 Secret.data，并推导 expiresAt；
当你真正要调用业务接口时，再用 authFieldPlacements 把 Secret.data 填回到业务请求里。

它像一个“低代码登录机器人”：你给出“表单结构 + 请求配方 + 结果采集 + 值的落位”，它就替你把浏览器里那一套操作流程复刻一遍，但更可靠、更可观察。

使用方法（快速）

创建平台与授权方式：
- 单步（如微信）：loginFlow 直接含 url/method/headers/body_template；
- 多步（如 Superset）：loginFlow.steps[*] 各自定义请求、可解析并注入下一步请求；
- 完整样例见 docs/auth_method.md。
提交 Secret：POST /v1/auth-methods/{id}/secret，可携带 loginPayload 与 autoLoginEnabled: true 触发登录。
解析调试：POST /v1/auth-methods/{id}/parse 校验 responseMapping 效果（支持 body/headers/cookies/html）。
结果查看：GET /v1/secrets/{secret_id} 查看 data、authDataSources、expiresAt。

内部运行流程（login）

构建上下文：以 loginPayload 为基础，自动注入 runtime.callback_url 与 runtime.ip_allow_list；
若配置 steps：逐步请求；在每步 responseMapping 解析后，将产物合并回上下文；
下一步请求前应用 requestPlacements 将上一步产物注入 headers/query/cookies/body；
全部步骤末尾，对最终响应应用“顶层” responseMapping，得到写入 Secret.data 的字段与 expiresAt；
更新 Secret.data、authDataSources、expiresAt、lastRefreshed、status。

端到端演练：Superset 二阶段（叙述）

以 Superset 为例：第 1 步你把用户名/密码交给系统。它先向 /login/ 发一遍表单请求，拿到两样东西：

响应头里的 Set-Cookie: session=...（系统会自动把每一个 Set-Cookie 折叠成 cookies map）；
登录页 HTML 里一个隐藏的 <input id="csrf_token" value="...">。

接着，系统执行第 2 步：仍请求 /login/，但这次会把“上一轮拿到的 session 放到 Cookie 里、csrf_token 放到表单里”。这正是 requestPlacements 的作用：

{
	"requestPlacements": {
		"cookies": {"session": "cookies.session"},
		"body": {"username": "username", "password": "password", "csrf_token": "csrf_token"}
	}
}

系统按这份说明，把“上下文里的 cookies.session / username / password / csrf_token”各就各位地植入请求。最后一跳收到 302/成功状态后，系统再把所有 Cookies 聚合到 Secret.data.cookies，并根据 Max-Age/Expires 与 token 的到期时间取一个最早者作为 expiresAt。这时 Secret 就“活”了：它知道自己什么时候过期、每个字段是从哪儿来的（authDataSources），也能被 authFieldPlacements 投放进你真实的业务请求。

requestPlacements 的心智模型

你可以把它理解成“把上下文里的值搬来搬去”的路由表：

上下文初始只有 loginPayload（外加系统自动注入的 runtime.callback_url 等）；
每一步解析成功后，上下文会追加新键，比如 cookies.session、csrf_token；
下一步请求前，requestPlacements 读取这些键，把它们按你指定的位置写入 headers/query/cookies/body；
如果需要拼接格式，可以写 { "sourceField": "cookies.session", "format": "session={{value}}" }。

这就是“把上一步结果放到第二步请求里”的统一做法，它与最终的 authFieldPlacements 各司其职、互不干扰。

Secret 最终输出字段

data：最终凭证，如 access_token、cookies.session 等；
authDataSources：字段来源（body.* / headers.* / cookies.* / html.* / manual-input）；
expiresAt：根据 token 与 cookie 取最早到期；
lastRefreshed：最后更新时间；status：状态。
usageMapping：等同于关联 AuthMethod.authFieldPlacements，用于客户端把 Secret.data 注入到业务请求的 headers/query/cookies/body。
type：与 AuthMethod.type 一致（如 OAuth1、OAuth2、Cookies 等），客户端可据此选择构造请求的方式（例如 OAuth1Session、Bearer 头或携带 Cookies）。

API 使用说明

平台 Platform:
- POST /v1/platforms
- GET /v1/platforms
- GET /v1/platforms/{platform_id}
- PATCH /v1/platforms/{platform_id}
授权方式 AuthMethod:
- POST /v1/auth-methods
  - Body: {id, platformId, name, type, description?, loginFlow?, loginFieldsSchema?, refreshFlow?, refreshFieldsSchema?, responseMapping?, defaultValiditySeconds?, refreshBeforeSeconds?}
- GET /v1/platforms/{platform_id}/auth-methods
- GET /v1/auth-methods/{auth_method_id}
- PATCH /v1/auth-methods/{auth_method_id}
默认 Schema:
- GET /v1/auth-types/default-schemas: 获取所有类型默认的 description, loginFlow, loginFieldsSchema 等配置。
- GET /v1/auth-types/default-schemas/{auth_type}: 获取指定类型的默认配置。
密钥 Secret:
- POST /v1/auth-methods/{auth_method_id}/secret: 创建或更新一个密钥。
- GET /v1/auth-methods/{auth_method_id}/secret?name=xxx: 按名称查询密钥。
- GET /v1/auth-methods/{auth_method_id}/secrets: 列出该授权方式下的所有密钥。
- GET /v1/secrets/{secret_id}: 按ID查询密钥。
- PATCH /v1/secrets/{secret_id}
- POST /v1/secrets/{secret_id}/remind: 记录提醒（占位）。
- POST /v1/secrets/{secret_id}/login: 使用已保存的登录字段自动登录并刷新密钥（仅返回 Secret）。
- POST /v1/secrets/{secret_id}/login/trace: 触发登录，返回每一步执行快照与最终解析结果。
- PATCH /v1/secrets/{secret_id}/with-execution: 更新秘钥（可带 loginPayload/autoLoginEnabled），并返回登录执行详情。
- DELETE /v1/secrets/{secret_id}: 删除秘钥。
- POST /v1/secrets/{secret_id}/invalidate: 外部将秘钥标记为失效，可选提供 reason。
- POST /v1/secrets/{secret_id}/invalid-notified: 将失效通知状态置为已通知/未通知，Body: { "notified": true }。
辅助接口:
- POST /v1/auth-methods/{auth_method_id}/parse: 根据 responseMapping 提取令牌并计算到期时间，支持 body/headers/cookies/html。

自动登录与 Token 获取

Secret 新增 loginPayload、autoLoginEnabled、lastLoginAt、lastLoginStatus、lastLoginError 等字段，可保存首次登录所需字段并追踪登录状态。
创建或更新秘钥时传入 autoLoginEnabled: true 会在服务端立即调用 loginFlow，并将响应通过 responseMapping 写回 data 与 expiresAt。
新增接口 POST /v1/secrets/{secret_id}/login，可随时触发重登；请求体可指定 loginPayload 重写登录字段以及 storeLoginPayload 是否同步保存。
登录成功会刷新 lastRefreshed、expiresAt（若响应含 expires_in / expires_at 或 Cookie 头包含过期时间）并记录 lastLoginAt=UTC，设置 isValid=true、lastLoginStatus=success。
登录失败会将 isValid=false、status=invalid、lastLoginStatus=failed 并写入 lastLoginError 便于排查。
Cookies 模式下会自动解析 Set-Cookie 的 Max-Age/Expires 计算秘钥过期时间，GET /v1/secrets/{id} 将返回即将过期时间供刷新决策。
AuthMethod 可选定义 authFieldsSchema（JSON Schema）与 authFieldPlacements，用于声明用户可直接填写的授权字段以及这些字段在后续请求中应放置的 header/query/cookie/body 键位；Secret.data 会据此校验并在 authDataSources 中记录字段来源（手动输入/接口解析等）。
parse 解析结果会透出 fieldSources，帮助定位 access_token 等字段的来源路径；当响应提供 expires_in（秒或时间戳）或 expires_at 时，会自动换算为 UTC 时间并存储。

登录流程可观测性（新）

日志输出：
- login.step：步骤索引、方法、URL、状态码、耗时（ms），状态码 4xx→WARNING，5xx→ERROR。
- login.step.parsed：DEBUG 级别；步骤解析出的字段键名、步骤内推导的到期时间（如有）。
- login.final：最终解析后的字段键名与到期时间（INFO）。
- login.failure：异常信息（ERROR），并自动标记秘钥失效。
API 回传：
- POST /v1/secrets/{id}/login/trace 返回结构：
  - execution.steps[*]：每一步的 request{method,url,headers,params,cookies,json,data}、statusCode、durationMs、extracted、fieldSources、ok。
  - execution.finalExtracted/fieldSources/finalExpiresAt/success：最终结果汇总。
- PATCH /v1/secrets/{id}/with-execution：先 PATCH 更新，再按上述方式回传执行详情。

秘钥生命周期（概要）

操作	接口	适用场景	返回执行轨迹	备注
创建/替换	`POST /auth-methods/{id}/secret`	首次建立或更换账号	是（如提供 loginPayload 且有 loginFlow）	`autoLoginEnabled=false` 不持久化登录表单
普通更新	`PATCH /secrets/{id}`	修改名称/标签/策略等	否	可附带 loginPayload 触发登录（若自动登录开启）
更新+调试	`PATCH /secrets/{id}/with-execution`	验证表单改动/排障	是	登录失败自动失效
手动登录	`POST /secrets/{id}/login`	周期刷新	否	可提供 `loginPayload` 覆盖，本次可存储
手动登录（调试）	`POST /secrets/{id}/login/trace`	排障/回归	是	DEBUG 级字段来源更易定位
标记失效	`POST /secrets/{id}/invalidate`	外部吊销/异常检测	否	置 `isValid=false,status=invalid,invalidNotified=false`
标记已通知	`POST /secrets/{id}/invalid-notified`	告警闭环	否	防止重复通知
删除	`DELETE /secrets/{id}`	账号迁移/废弃	否	彻底移除

更多细节与 curl 示例参见 docs/manuals.md。

示例：创建平台与 OAuth2 授权方式

# 1. 创建平台
curl -X POST http://127.0.0.1:8000/v1/platforms \
	-H 'content-type: application/json' \
	-d '{
		"id":"plat_x",
		"orgId":"org_demo",
		"nameId":"x_com",
		"name":"X (Twitter)",
		"code":"x"
	}'

# 2. 查看 OAuth2 的默认配置模板
curl -s http://127.0.0.1:8000/v1/auth-types/default-schemas/OAuth2 | jq

# 3. 创建一个 OAuth2 授权方式
# 注意：我们直接使用了默认模板中的 refreshFlow 和 responseMapping，
# 只需提供 loginFieldsSchema 中定义的字段即可。
curl -X POST http://127.0.0.1:8000/v1/auth-methods \
	-H 'content-type: application/json' \
	-d '{
		"id":"am_x_oauth2",
		"platformId":"plat_x",
		"name":"oauth2_user_context",
		"type":"OAuth2",
		"loginFieldsSchema": {
			"type": "object",
			"properties": {
				"client_id": { "type": "string", "title": "Client ID" },
				"client_secret": { "type": "string", "title": "Client Secret" },
				"authorize_url": { "type": "string", "format": "uri", "title": "Authorize URL" },
				"token_url": { "type": "string", "format": "uri", "title": "Token URL" },
				"scopes": { "type": "array", "items": { "type": "string" }, "title": "Scopes" }
			},
			"required": ["client_id", "client_secret", "authorize_url", "token_url", "scopes"]
		},
		"defaultValiditySeconds": 7200,
		"refreshBeforeSeconds": 300
	}'

# 4. 创建一个密钥 (Secret)
# 此时，用户需要根据 loginFieldsSchema 提供具体的值，
# 这些值将用于后续的认证流程（例如，重定向到 authorize_url）。
# data 字段存储了实际的认证配置。
curl -X POST http://127.0.0.1:8000/v1/auth-methods/am_x_oauth2/secret \
	-H 'content-type: application/json' \
	-d '{
		"id":"sec_x_user_A",
		"name":"user_A_secrets",
		"tags":["test_account"],
		"data": {
			"client_id":"<YOUR_CLIENT_ID>",
			"client_secret":"<YOUR_CLIENT_SECRET>",
			"authorize_url":"https://twitter.com/i/oauth2/authorize",
			"token_url":"https://api.twitter.com/2/oauth2/token",
			"scopes":["tweet.read","users.read","tweet.write","offline.access"]
		}
	}'

示例：创建 JWT 登录授权方式

# 1. 创建一个 JWT 授权方式
# 这里我们自定义了 loginFlow，指向一个具体的登录端点。
curl -X POST http://127.0.0.1:8000/v1/auth-methods \
	-H 'content-type: application/json' \
	-d '{
		"id":"am_my_service_jwt",
		"platformId":"<your_platform_id>",
		"name":"jwt_login",
		"type":"JWT",
		"description": "使用您在 MyService 的用户名和密码登录。",
		"loginFlow": {
			"url": "https://my-service.com/api/auth/login",
			"method": "POST",
			"headers": { "Content-Type": "application/json" },
			"body_template": {
				"login_id": "{{username}}",
				"password": "{{password}}"
			}
		}
	}'

# 2. 创建一个密钥 (Secret)
# 用户只需根据默认的 loginFieldsSchema 提供 username 和 password（通过 loginPayload）。
# 凭证不会被持久化，仅用于服务端登录。
curl -X POST http://127.0.0.1:8000/v1/auth-methods/am_my_service_jwt/secret \
	-H 'content-type: application/json' \
	-d '{
		"id":"sec_my_service_user_B",
		"name":"user_B_creds",
		"data": {},
		"loginPayload": {"username":"user_b","password":"a_very_secure_password"},
		"autoLoginEnabled": true
	}'

示例：Superset Cookies 授权方式（CSRF + 会话）

以下示例基于 http://43.140.209.117:8088/ 的 Superset 实例，演示如何收集两阶段登录得到的 Cookie，并将结果回填到 Cookies 类型的授权方式中。

创建 Cookies 类型授权方式，声明需要存储的字段：

curl -X POST http://127.0.0.1:8000/v1/auth-methods \
		-H 'content-type: application/json' \
		-d '{
			"id":"am_superset_cookies",
			"platformId":"plat_superset",
			"name":"superset_form_login",
			"type":"Cookies",
			"description":"Superset 表单登录，需要先获取 csrf_token 再提交一次表单完成会话初始化。",
			"authFieldsSchema":{
				"type":"object",
				"properties":{
					"cookies":{
						"type":"object",
						"title":"Superset Cookies",
						"properties":{
							"session":{"type":"string","title":"session"},
							"csrf_token":{"type":"string","title":"csrf_token","description":"部分接口需要同步提交"}
						},
						"required":["session"]
					}
				},
				"required":["cookies"]
			}
		}'

运行下面的 Python 片段完成“两次表单提交”并提取 csrf_token 与最终的 session Cookie：

import json
import re
import requests

BASE_URL = "http://43.140.209.117:8088"
USERNAME = "admin"
PASSWORD = "your_password"

session = requests.Session()

# 第一次提交，只为了拿隐藏的 csrf_token 和初始 session
first = session.post(
		f"{BASE_URL}/login/",
		data={"username": USERNAME, "password": PASSWORD},
		allow_redirects=False,
)
csrf_match = re.search(r'name="csrf_token" value="([^"]+)"', first.text)
if not csrf_match:
		raise RuntimeError("未能在登录页解析到 csrf_token")
csrf_token = csrf_match.group(1)
initial_session = session.cookies.get("session")

# 第二次提交，携带 csrf_token 与初始 session 完成登录
second = session.post(
		f"{BASE_URL}/login/",
		data={
				"username": USERNAME,
				"password": PASSWORD,
				"csrf_token": csrf_token,
		},
		headers={"Cookie": f"session={initial_session}"},
		allow_redirects=False,
)
second.raise_for_status()
final_session = session.cookies.get("session")

payload = {
		"body": {"csrf_token": csrf_token},
		"headers": {"Set-Cookie": second.headers.get("Set-Cookie")},
		"cookies": {"session": final_session},
}

print("解析请求示例:")
print(json.dumps(payload, indent=2, ensure_ascii=False))

print("Set-Cookie:", second.headers.get("Set-Cookie"))

将脚本输出填入解析与入库请求：

# 解析登录响应，计算过期时间并写入字段来源
curl -X POST http://127.0.0.1:8000/v1/auth-methods/am_superset_cookies/parse \
		-H 'content-type: application/json' \
		-d '{
			"body": {"csrf_token": "<CSRF_TOKEN>"},
			"headers": {"Set-Cookie": "session=<SESSION_COOKIE>; Expires=Wed, 19 Nov 2025 12:00:00 GMT; Path=/; HttpOnly"},
			"cookies": {"session": "<SESSION_COOKIE>"}
		}'

# 将 cookies 与 csrf_token 保存为 Secret，expiresAt 会根据解析结果写入
curl -X POST http://127.0.0.1:8000/v1/auth-methods/am_superset_cookies/secret \
		-H 'content-type: application/json' \
		-d '{
			"id":"sec_superset_admin",
			"name":"superset_admin",
			"tags":["superset"],
			"data":{
				"cookies":{
					"session":"<SESSION_COOKIE>",
					"csrf_token":"<CSRF_TOKEN>"
				}
			}
		}'

经过上述步骤，Secret 会同时保存 session 与 csrf_token；若后续响应同时返回 Token 与 Cookies，系统会自动选取最早过期的时间更新 expiresAt，避免凭证提前失效。

### 授权配置样例

#### 1. X 平台：OAuth1 授权方法 + 秘钥

```bash
# 创建授权方式（OAuth1）
curl -X POST http://127.0.0.1:8000/v1/auth-methods \
	-H 'content-type: application/json' \
	-d '{
		"id":"am_x_oauth1",
		"platformId":"plat_x",
		"name":"oauth1_bot",
		"type":"OAuth1",
		"description":"X 平台 OAuth1 三阶段授权样例",
		"loginFlow":{
			"url":"https://api.x.com/oauth/request_token",
			"method":"POST",
			"headers":{"Content-Type":"application/json"},
			"body_template":{
				"consumer_key":"{{consumer_key}}",
				"consumer_secret":"{{consumer_secret}}",
				"callback_url":"{{callback_url}}"
			}
		},
		"refreshFlow":{
			"url":"https://api.x.com/oauth/access_token",
			"method":"POST",
			"headers":{"Content-Type":"application/json"},
			"body_template":{
				"oauth_token":"{{oauth_token}}",
				"oauth_verifier":"{{oauth_verifier}}"
			}
		},
		"authFieldsSchema":{
			"type":"object",
			"title":"OAuth1 授权凭证",
			"properties":{
				"oauth_token":{"type":"string","title":"OAuth Token"},
				"oauth_token_secret":{"type":"string","title":"OAuth Token Secret"},
				"access_token":{"type":"string","title":"Access Token"},
				"access_token_secret":{"type":"string","title":"Access Token Secret"},
				"expires_at":{"type":["integer","null"],"title":"过期时间戳"}
			},
			"required":["oauth_token","oauth_token_secret","access_token","access_token_secret"]
		},
		"authFieldPlacements":{
			"headers":{
				"Authorization":{"type":"Bearer","tokenField":"access_token"},
				"oauth_token":{"type":"plain","valueField":"oauth_token"},
				"oauth_token_secret":{"type":"plain","valueField":"oauth_token_secret"},
				"x-token-secret":{"type":"plain","valueField":"access_token_secret"}
			},
			"cookies":{
				"AUTHORIZATION":{"sourceField":"access_token","template":"token={{value}}"}
			},
			"metadata":{
				"tokenType":"OAuth1",
				"defaultExpiresField":"expires_at"
			}
		},
		"responseMapping":{
			"oauth_token":"oauth_token",
			"oauth_token_secret":"oauth_token_secret",
			"access_token":"access_token",
			"access_token_secret":"access_token_secret",
			"expires_in":"expires_in"
		}
	}'

# 写入秘钥并启用自动登录获取 request token
curl -X POST http://127.0.0.1:8000/v1/auth-methods/am_x_oauth1/secret \
	-H 'content-type: application/json' \
	-d '{
		"id":"sec_x_oauth1_bot",
		"name":"bot_account",
		"data":{},
		"loginPayload":{
			"consumer_key":"<X_CONSUMER_KEY>",
			"consumer_secret":"<X_CONSUMER_SECRET>",
			"callback_url":"https://your-app.example.com/oauth/callback"
		},
		"autoLoginEnabled":true
	}'

# 随时可通过 login 接口发起 refresh，完成 access token 替换
curl -X POST http://127.0.0.1:8000/v1/secrets/sec_x_oauth1_bot/login \
	-H 'content-type: application/json' \
	-d '{"storeLoginPayload":false,
			 "loginPayload":{"oauth_token":"<OAUTH_TOKEN>","oauth_verifier":"<OAUTH_VERIFIER>"}}'
```

#### 2. 微信公众平台：OAuth1（稳定版接口）示例

- 第一阶段：管理员在前端表单中根据 `loginFieldsSchema` 填写 `AppID`/`AppSecret`，系统自动注入环境变量提供的回调地址；
- 第二阶段：服务端根据 `loginFlow` 调用微信接口 `POST https://api.weixin.qq.com/cgi-bin/stable_token`，按要求传入 `grant_type=client_credential` 等字段获取稳定版 `access_token`；
- 接口返回的 `access_token` 与 `expires_in` 会通过 `responseMapping` 保存到 `Secret.data` 中，并折算出到期时间。

示例授权方式（仅展示关键字段）：

```bash
curl -X POST http://127.0.0.1:8000/v1/auth-methods \
  -H 'content-type: application/json' \
  -d '{
    "id": "am_wechat_official_oauth1",
    "platformId": "plat_wechat_official",
    "name": "wechat_official_oauth1",
    "type": "OAuth1",
    "description": "微信公众平台稳定版 access_token 获取流程，两步：配置账号与调用 stable_token 接口。",
    "loginFieldsSchema": {
      "type": "object",
      "required": ["appid", "secret"],
      "properties": {
        "appid": {"type": "string", "title": "AppID"},
        "secret": {"type": "string", "title": "AppSecret"},
        "grant_type": {
          "type": "string",
          "title": "grant_type",
          "default": "client_credential"
        },
        "force_refresh": {
          "type": "boolean",
          "title": "force_refresh",
          "default": false
        },
        "callback_url": {
          "type": "string",
          "title": "回调地址(只读)",
          "readOnly": true,
          "description": "由 turbo-agent-auth 环境变量提供，禁止人工填写"
        }
      }
    },
    "loginFlow": {
      "url": "https://api.weixin.qq.com/cgi-bin/stable_token",
      "method": "POST",
      "headers": {"Content-Type": "application/json"},
      "body_template": {
        "grant_type": "{{grant_type}}",
        "appid": "{{appid}}",
        "secret": "{{secret}}",
        "force_refresh": "{{force_refresh}}"
      }
    },
    "responseMapping": {
      "fields": {
        "access_token": "access_token",
        "expires_in": "expires_in"
      }
    },
    "defaultValiditySeconds": 7200,
    "refreshBeforeSeconds": 300
  }'
```

创建 Secret 时只需提供账号描述与标签，`data` 会由登录响应填充。例如：

```bash
curl -X POST http://127.0.0.1:8000/v1/auth-methods/am_wechat_official_oauth1/secret \
  -H 'content-type: application/json' \
  -d '{
    "id": "sec_wechat_demo",
    "name": "wechat_official_demo",
    "tags": ["wechat", "official"],
    "data": {},
    "loginPayload": {
      "appid": "<你的公众号 AppID>",
      "secret": "<你的公众号 AppSecret>",
      "force_refresh": false
    },
    "autoLoginEnabled": true
  }'
```

> 提示：如需向微信侧备案 IP 白名单，可调用 `/v1/runtime/ip-allowlist` 获取当前出口地址。

> OAuth1 流程通常需要在外部完成用户授权，本示例中 `loginPayload` 第一次写入 `consumer_key/secret` 获取 request token，回调后将 `oauth_token` / `oauth_verifier` 传回，再次调用 `/login` 即可落库 access token。

#### 2. Superset 平台：JWT（登录 + 刷新）

```bash
# 创建 Superset JWT 授权方式
curl -X POST http://127.0.0.1:8000/v1/auth-methods \
	-H 'content-type: application/json' \
	-d '{
		"id":"am_superset_jwt",
		"platformId":"plat_superset",
		"name":"jwt_service_account",
		"type":"JWT",
		"description":"Superset API JWT 登录/刷新",
		"loginFlow":{
			"url":"https://superset.example.com/api/v1/security/login",
			"method":"POST",
			"headers":{"Content-Type":"application/json"},
			"body_template":{
				"username":"{{username}}",
				"password":"{{password}}",
				"provider":"db",
				"refresh":true
			}
		},
		"refreshFlow":{
			"url":"https://superset.example.com/api/v1/security/refresh",
			"method":"POST",
			"headers":{
				"Content-Type":"application/json",
				"Authorization":"Bearer {{access_token}}"
			},
			"body_template":{
				"refresh_token":"{{refresh_token}}"
			}
		},
		"responseMapping":{
			"access_token":"access_token",
			"refresh_token":"refresh_token",
			"expires_in":"expires_in"
		}
	}'

# 上传登录凭证并自动获取 access_token
curl -X POST http://127.0.0.1:8000/v1/auth-methods/am_superset_jwt/secret \
	-H 'content-type: application/json' \
	-d '{
		"id":"sec_superset_jwt_admin",
		"name":"admin_jwt",
		"data":{},
		"loginPayload":{
			"username":"admin",
			"password":"<SUPERSET_PASSWORD>"
		},
		"autoLoginEnabled":true
	}'

# 当 refresh_token 即将过期时再次执行登录
curl -X POST http://127.0.0.1:8000/v1/secrets/sec_superset_jwt_admin/login
```

#### 3. Superset 平台：Cookies （登录 + 自动过期检测）

```bash
# 创建基于 Cookie 的授权方式
curl -X POST http://127.0.0.1:8000/v1/auth-methods \
	-H 'content-type: application/json' \
	-d '{
		"id":"am_superset_cookie",
		"platformId":"plat_superset",
		"name":"cookie_session",
		"type":"Cookies",
		"description":"Superset 会话 Cookie 登录，自动解析 Set-Cookie 的过期时间",
		"loginFlow":{
			"url":"https://superset.example.com/api/v1/security/login",
			"method":"POST",
			"headers":{"Content-Type":"application/json"},
			"body_template":{
				"username":"{{username}}",
				"password":"{{password}}",
				"provider":"db"
			}
		},
		"refreshFlow":{
			"url":"https://superset.example.com/api/v1/security/refresh",
			"method":"POST",
			"headers":{
				"Content-Type":"application/json",
				"Authorization":"Bearer {{access_token}}"
			},
			"body_template":{
				"refresh_token":"{{refresh_token}}"
			}
		},
		"responseMapping":{}
	}'

# 写入账号信息并自动发起登录，Secret 中会保存 cookies 与 expiresAt
curl -X POST http://127.0.0.1:8000/v1/auth-methods/am_superset_cookie/secret \
	-H 'content-type: application/json' \
	-d '{
		"id":"sec_superset_cookie_admin",
		"name":"admin_cookie",
		"data":{},
		"loginPayload":{
			"username":"admin",
			"password":"<SUPERSET_PASSWORD>"
		},
		"autoLoginEnabled":true
	}'

# 查询秘钥即可看到 cookies 及 expiresAt，过期后返回 401 时可调用 login 重新获取
curl http://127.0.0.1:8000/v1/secrets/sec_superset_cookie_admin | jq
```

> Superset 登录接口会同时返回 Token 与会话 Cookie，本方案使用 Cookie 方式以覆盖需要浏览器态会话的场景；`parse_response` 会读取 `Set-Cookie` 的 `Max-Age/Expires` 并自动推算秘钥过期时间。

Project details

Release history Release notifications | RSS feed

0.1.10rc1 pre-release

Apr 24, 2026

This version

0.1.2

Feb 27, 2026

0.1.1

Feb 26, 2026

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

turbo_agent_auth-0.1.2.tar.gz (108.2 kB view details)

Uploaded Feb 27, 2026 Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

The dropdown lists show the available interpreters, ABIs, and platforms. Enable javascript to be able to filter the list of wheel files.

turbo_agent_auth-0.1.2-py3-none-any.whl (52.0 kB view details)

Uploaded Feb 27, 2026 Python 3

File details

Details for the file turbo_agent_auth-0.1.2.tar.gz.

File metadata

Download URL: turbo_agent_auth-0.1.2.tar.gz
Upload date: Feb 27, 2026
Size: 108.2 kB
Tags: Source
Uploaded using Trusted Publishing? No
Uploaded via: twine/6.2.0 CPython/3.13.7

File hashes

Hashes for turbo_agent_auth-0.1.2.tar.gz
Algorithm	Hash digest
SHA256	`4e0be12659a86d98fa58990640bf5a0f4e2f85613a934a7ec6a9641639b456cb`
MD5	`e25136612002eedbc2b9cefcb72469e3`
BLAKE2b-256	`ce67d97f7e19522fa50527c1e7fe3aad0c8260f65e3d9dd2cb1f71de544308bb`

See more details on using hashes here.

File details

Details for the file turbo_agent_auth-0.1.2-py3-none-any.whl.

File metadata

Download URL: turbo_agent_auth-0.1.2-py3-none-any.whl
Upload date: Feb 27, 2026
Size: 52.0 kB
Tags: Python 3
Uploaded using Trusted Publishing? No
Uploaded via: twine/6.2.0 CPython/3.13.7

File hashes

Hashes for turbo_agent_auth-0.1.2-py3-none-any.whl
Algorithm	Hash digest
SHA256	`44ce40b6955f87bce0ec435a2c38ba44afa881c66c5a85c850ef4fcef20ad1c7`
MD5	`b391a15e1dbaedbb6d8793e4b6ecee3b`
BLAKE2b-256	`d95962fde91300a011fa33a3282de3006d6d56421b5bf93275853636e02fe5f5`

See more details on using hashes here.

turbo-agent-auth 0.1.2

Navigation

Verified details

Maintainers

Unverified details

Meta

Project description

turbo-agent-auth

技术栈

安装依赖

运行 API (Typer CLI)

安全：API Key (Authorization Bearer)

日志

Prisma 初始化

运行测试

架构概览

项目结构

Docker 构建与部署

变更摘要（2025-11-20 新增）

变更摘要（2025-11-19）

设计思路

系统综述（故事版）

使用方法（快速）

内部运行流程（login）

端到端演练：Superset 二阶段（叙述）

requestPlacements 的心智模型

Secret 最终输出字段

API 使用说明

自动登录与 Token 获取

登录流程可观测性（新）

秘钥生命周期（概要）

示例：创建平台与 OAuth2 授权方式

示例：创建 JWT 登录授权方式

示例：Superset Cookies 授权方式（CSRF + 会话）

Project details

Verified details

Maintainers

Unverified details

Meta

Release history Release notifications | RSS feed

Download files

Source Distribution

Built Distribution

File details

File metadata

File hashes

File details

File metadata

File hashes