DeerFlow 授权身份链路实战：从 Gateway 防伪到 Guardrail 的可信 Principal 传播机制

DeerFlow 的 Phase 1A-1 阶段要解决一个具体问题：让"谁在执行工具"这一身份信息从 Gateway 认证状态一路可信地传播到执行时刻的 Guardrail 授权判断中，并在链路上清除一切客户端伪造的身份字段。读完本文，你能完整掌握 DeerFlow 可信授权身份链的设计契约、Gateway 防伪边界的实现细节、Principal builder 的统一构造规则，以及 subagent 身份继承与 Guardrail adapter 的落点——这套链路正是 Phase 1A-1 实施计划所固定的端到端方案。

1. 目标链路：一次请求中的身份生命周期

本 PR 完成一条可信、可测试的授权身份链路，覆盖从 Gateway 到 Guardrail 的全部关键节点：

Gateway 认证状态
  → 清除客户端伪造字段
  → 服务端注入 is_internal
  → lead runtime context
  → task_tool 捕获
  → SubagentExecutor 复制
  → subagent runtime context
  → GuardrailRequest
  → GuardrailAuthorizationAdapter
  → build_principal_from_context()
  → Principal

这条链路的设计意图是：身份的唯一权威来源是服务端认证状态（request.state），而不是客户端请求体。客户端可以在 body.config["context"]、body.config["configurable"] 或顶层 body.context 中伪造任何字段，但只有经过 Gateway 清理与重新盖章（restamp）后的值才会进入下游。

合并后的兼容性承诺有三条：

• authorization.enabled: false 时，工具集合和工具执行决策不变；
• runtime context 新增服务端身份字段是有意的可观察变化；
• 现有 adapter 异常传播语义保持不变。

2. 固定安全契约

2.1 Principal：7 个字段的显式构造

Principal 是授权判定中"执行者"的数据结构，定义在 provider.py 中：

@dataclass
class Principal:
    user_id: str | None = None
    role: str | None = None
    oauth_provider: str | None = None
    oauth_id: str | None = None
    channel_user_id: str | None = None
    is_internal: bool = False
    attributes: dict[str, Any] = field(default_factory=dict)

build_principal_from_context() 必须显式构造全部 7 个字段（user_id / role / oauth_provider / oauth_id / channel_user_id / is_internal / attributes），规则如下：

• user_role 为 None 或空字符串时使用 default_role；
• 未知但非空的角色原样保留，不能回退到默认角色（避免把 admin 之外的合法自定义角色误降级）；
• is_internal 只有原值严格等于 True 时才为 True（1、"true" 均不合法）；
• authz_attributes 为 None 或缺失时解析为 {}；
• 非空 authz_attributes 必须实现 Mapping，否则抛 TypeError；
• attributes 每次都复制为新字典，不共享可变引用；
• builder 是纯函数：不读全局 config、不缓存、不修改输入。

2.2 Gateway 可信边界：服务端拥有字段

is_internal、authz_attributes 和 channel_user_id 是服务端拥有（server-owned）字段，安全规则为：

• 必须从 config["context"] 和 config["configurable"] 清除客户端值；
• is_internal 只能由 request.state.auth_source == AUTH_SOURCE_INTERNAL 产生；
• channel_user_id 只能来自内部认证 IM 调用方的顶层 body.context；普通 session 请求和 body.config 两个 section 中的值必须删除；
• 必须使用直接赋值（runtime_context["is_internal"] = ...），不能使用 setdefault（否则客户端伪造值会"抢先"存活）；
• 必须在 user_id is None 等所有 early return 之前写入，保证任何请求路径都拿到定义明确的 is_internal。

从当前源码看，Phase 1A-1 落地后 services.py 中的服务端拥有键集合已扩展为 _SERVER_OWNED_RUNTIME_CONTEXT_KEYS（user_id 对非 internal 调用方同样是服务端拥有的，客户端伪造值会被清除后仅从 request.state.user 重新盖章），清理与注入逻辑集中在 inject_authenticated_user_context() 函数顶部：

# --- Server-owned authorization and sandbox lifecycle identity fields ---
runtime_context = config.setdefault("context", {})
if not isinstance(runtime_context, dict):
    raise TypeError("run context must be a mapping")
for key in _SERVER_OWNED_RUNTIME_CONTEXT_KEYS:
    runtime_context.pop(key, None)
configurable = config.get("configurable")
if isinstance(configurable, dict):
    for key in _SERVER_OWNED_RUNTIME_CONTEXT_KEYS:
        configurable.pop(key, None)
auth_source = getattr(getattr(request, "state", None), "auth_source", None)
# 外部调用方的 user_id 也是服务端拥有字段，先清除再重新盖章
user = getattr(getattr(request, "state", None), "user", None)
if auth_source != AUTH_SOURCE_INTERNAL and getattr(user, "system_role", None) != INTERNAL_SYSTEM_ROLE:
    runtime_context.pop("user_id", None)
    if isinstance(configurable, dict):
        configurable.pop("user_id", None)
runtime_context["is_internal"] = auth_source == AUTH_SOURCE_INTERNAL
if auth_source == AUTH_SOURCE_INTERNAL and request_context is not None:
    channel_user_id = request_context.get("channel_user_id")
    if channel_user_id is not None:
        runtime_context["channel_user_id"] = channel_user_id

对应实施记录：在 实施笔记中追加 Phase 1A-1 决策日志（internal 来自服务端 auth source、Gateway 清除服务端拥有字段、adapter 复用唯一 builder 等）。

2.3 传递语义：subagent 必须"无差别"继承

• Subagent 必须继承 parent context 的 is_internal 和合法 attributes；
• is_internal=False 也必须显式写回，不能按 truthy 条件省略（省略会导致下游无法区分"未设置"和"明确外部"）；
• attributes 在 task 捕获、executor 构造和 context 写回时均使用副本；
• 非 Mapping attributes 在所有进程内消费边界统一抛 TypeError——不能某些层抛错、另一些层静默变成 {}；
• Guardrail adapter 必须复用 Principal builder，不能维护第二套默认角色或 attributes 解析逻辑。

3. Principal builder：唯一的身份构造入口

实现位于 principal.py，核心是一个共享的规范化函数加一个纯函数 builder：

def normalize_authz_attributes(raw: Any) -> dict[str, Any]:
    """Validate and copy ``authz_attributes`` into a fresh dict."""
    if raw is None:
        return {}
    if isinstance(raw, Mapping):
        return dict(raw)
    raise TypeError(f"authz_attributes must be a Mapping, got {type(raw).__name__}")


def build_principal_from_context(
    context: Mapping[str, Any],
    *,
    default_role: str,
) -> Principal:
    resolved_role = context.get("user_role")
    if resolved_role is None or resolved_role == "":
        resolved_role = default_role

    return Principal(
        user_id=context.get("user_id"),
        role=resolved_role,
        oauth_provider=context.get("oauth_provider"),
        oauth_id=context.get("oauth_id"),
        channel_user_id=context.get("channel_user_id"),
        is_internal=context.get("is_internal") is True,
        attributes=normalize_authz_attributes(context.get("authz_attributes")),
    )

两个实现细节值得注意：

1. normalize_authz_attributes 被抽成共享函数。计划文档中 builder、task_tool、executor、middleware 四层各写一段相同的 Mapping 校验分支，落地时收敛为单一规范化点——所有进程内消费边界（middleware.py、executor.py、task_tool.py）都 import 它，从代码结构上保证了"非 Mapping 一律 TypeError"的契约不会被某一层遗漏。
2. is_internal=context.get("is_internal") is True 用 is 而非真值判断，把 1、"true" 等伪造形态一律视为外部调用方。

行为测试集中在 test_authorization_principal.py，覆盖：空 context、部分 context、全 7 字段映射；None/""/缺失 role 回退 default_role；未知非空 role 原样保留；True/False/1/"true"/None 的严格布尔行为；attributes 缺失或 None 得到 {}；Mapping 被复制（修改输入不影响 Principal）；非 Mapping 抛 TypeError 且错误信息包含实际类型名；oauth 与 channel identity 正确映射。

4. Gateway 防伪：测试驱动的可信边界

计划文档要求 Gateway 防伪测试复用真实的配置装配顺序，而不能只测孤立函数：

build_run_config
  → merge_run_context_overrides
  → strip_internal_context_keys（非 internal 请求）
  → inject_authenticated_user_context

测试必须分别覆盖三个伪造入口，且断言新增覆盖逻辑是唯一通过原因（不能用本来就被 body.context 白名单过滤的空路径）：

1. body.config["context"] 伪造 is_internal=True；
2. body.config["configurable"] 伪造 is_internal=True；
3. 两个入口分别伪造 authz_attributes。

断言要点：普通/session 请求最终 context.is_internal is False 且 configurable.is_internal 被删除；两个 section 中的 authz_attributes 都被删除；internal 请求最终 is True，但客户端伪造的 attributes 同样被删除；user=None、auth source 缺失或 session 时仍写入 False；user=None + auth_source=internal 时仍写入 True；非 Mapping runtime context 显式抛 TypeError 而不静默跳过。相关测试位于 test_gateway_services.py。

5. 身份跨进程内边界的复制：task_tool 与 SubagentExecutor

身份进入 lead agent 的 runtime context 后，第一次跨边界传播发生在 task_tool 派生子代理时。task_tool.py 中的 parent identity capture 块现在无条件携带两个新字段：

# Propagate authorization identity: is_internal (strict bool) and
# authz_attributes (validated Mapping, copied).
is_internal = parent_context.get("is_internal") is True
authz_attributes = normalize_authz_attributes(parent_context.get("authz_attributes"))
# ... 无条件加入 executor_kwargs

随后 SubagentExecutor 在构造参数中接收这两个字段，构造时严格校验并复制，context 写回时无条件执行（包括 False）：

# 构造时
self.is_internal = is_internal
self.authz_attributes = normalize_authz_attributes(authz_attributes)

# context write-back（无条件，含 False）
context["is_internal"] = self.is_internal
context["authz_attributes"] = dict(self.authz_attributes)

测试方面：test_task_tool_core_logic.py 覆盖 parent is_internal=True/False 显式传递、Mapping 复制、捕获后修改 parent 不影响 executor kwargs、非 Mapping 抛 TypeError；test_subagent_executor.py 覆盖构造参数写回、构造与写回两次复制的隔离性、以及 executor 直接收到非 Mapping attributes 时的 TypeError。

6. Guardrail 层：GuardrailRequest 扩展与 adapter 复用

执行时刻的身份最终收敛到 Guardrail 中间件。provider.py 中 GuardrailRequest 新增的三个字段全部带默认值，保证向后兼容：

channel_user_id: str | None = None
is_internal: bool = False
authz_attributes: dict[str, Any] = field(default_factory=dict)

middleware.py 在构造 request 前用普通分支完成 attributes 校验并做严格布尔映射：

channel_user_id=context.get("channel_user_id"),
is_internal=context.get("is_internal") is True,
authz_attributes=normalize_authz_attributes(context.get("authz_attributes")),

而 GuardrailAuthorizationAdapter 的职责是把 AuthorizationProvider 适配为 GuardrailProvider，从而复用现有 GuardrailMiddleware 的执行期拦截，不新增中间件类。关键约束在 _to_authz() 中体现：

principal = build_principal_from_context(
    {
        "user_id": gr.user_id,
        "user_role": gr.user_role,
        "oauth_provider": gr.oauth_provider,
        "oauth_id": gr.oauth_id,
        "channel_user_id": gr.channel_user_id,
        "is_internal": gr.is_internal,
        "authz_attributes": gr.authz_attributes,
    },
    default_role=self._default_role,
)

• 构造函数增加 default_role: str = "user" 并保存；Phase 1B 自动装配 adapter 时必须显式传入 AuthorizationConfig.default_role；
• 不得手工构造 Principal，必须走 builder，保证 Layer 1（工具装配）与 Layer 2（执行期 adapter）共享同一套 role/attributes 语义；
• 不捕获 provider 异常、不实现 fail_closed——异常继续向上传播，由 GuardrailMiddleware 基于其 fail_closed 参数统一处理，避免两层逻辑分叉；
• 同步 evaluate() 与异步 aevaluate() 通过同一条 _to_authz() 路径，产生语义一致的完整 Principal。

相应地，provider.py 中 Principal 的生命周期说明也被更新为"adapter 每次请求实时构建，不缓存过期身份"，替代此前"每个 run 只构建一次"的描述。adapter 与 middleware 的测试分别落在 test_authorization_provider.py 和 test_guardrail_middleware.py：新字段默认值向后兼容、middleware 的非 Mapping TypeError、adapter 同步/异步一致、7 字段完整映射、role 缺失应用 default_role、未知非空 role 不回退、provider 异常穿过 adapter。

7. TDD 实施顺序与文件边界

整个链路按"先失败测试、后实现"的 10 个 Step 推进：builder 测试 → builder 实现 → Gateway 防伪测试 → Gateway 实现 → task_tool 捕获测试 → 捕获实现 → executor 写回测试 → 写回实现 → guardrail/adapter 测试 → 扩展 GuardrailRequest 与 adapter，最后追加实施记录。计划刻意把 22 个文件（5 新增 + 17 修改）放进单个 PR，原因是它固定的是一条跨 Gateway、lead、subagent 和 guardrail 的端到端身份链；继续拆分会产生"可合并但身份链不完整"的中间状态。

明确不修改与不实现的范围同样重要，它界定了 Phase 1A-1 的承诺：authorization_config.py、config.example.yaml、lead agent 与 client 均不动；RBAC provider 与 factory（Phase 1A-2）、Layer 1 工具过滤与 deferred catalog 防提升（Phase 1B）、Layer 2 自动接线（Phase 1B）、route/model/skill/sandbox/MCP server 授权、RBAC 配置示例与 config_version 变更全部延期。

8. 验证与验收门槛

提交前按 TDD 步骤逐段运行目标测试，再跑边界与 lint 检查：

cd backend

uv run pytest tests/test_authorization_principal.py -q
uv run pytest tests/test_gateway_services.py -q
uv run pytest tests/test_task_tool_core_logic.py -q
uv run pytest tests/test_subagent_executor.py -q
uv run pytest tests/test_authorization_provider.py -q
uv run pytest tests/test_guardrail_middleware.py -q
uv run pytest tests/test_channel_user_id_env.py -q

uv run pytest tests/test_harness_boundary.py -q
uv run ruff check packages/harness/deerflow/authz packages/harness/deerflow/guardrails \
  packages/harness/deerflow/tools/builtins/task_tool.py \
  packages/harness/deerflow/subagents/executor.py app/gateway/services.py \
  tests/test_authorization_principal.py

make test
make lint
make format

验收门槛本质上是一组破坏性测试——每一项都要求删除某段实现后对应测试必须失败，以此证明测试真正锚定了安全契约：

• 删除 is_internal 服务端直接赋值后，Gateway 防伪测试失败；
• 删除 configurable 清理后，对应伪造测试失败；
• 普通请求不能通过 body.context 或 body.config 注入 channel_user_id，internal 请求只接受顶层 body.context 的值；
• 删除任意一段 task/executor identity wiring 后，subagent 测试失败；
• builder、task、executor、middleware 对 attributes 非 Mapping 的行为一致；
• user=None 的 internal 与 non-internal 分支都被固定；
• authorization.enabled: false 时工具集合与执行决策不变；
• 没有 Layer 1、Layer 2、RBAC 或 client 额外改动混入。

9. 小结

DeerFlow Phase 1A-1 的核心贡献是一条有测试锚点的身份不变量链：服务端认证状态是唯一权威来源，Gateway 在 early return 之前清除并重新盖章服务端拥有字段，is_internal 与 authz_attributes 在 task_tool、executor、middleware 每一跳都显式复制且严格校验，最终由唯一 builder 收敛为 Principal 供授权判定。这套"单点规范化 + 无条件写回 + 每层失败测试"的组合，为后续 Phase 1A-2 的 RBAC provider 与 Phase 1B 的 Layer 1/Layer 2 接线奠定了可信任的身份基础——理解这条链路，也就理解了 DeerFlow 细粒度授权体系中"执行者是谁"这个问题的完整答案。