From 927b833ef46843ba6da4a7f6cdb004d5d5eb4784 Mon Sep 17 00:00:00 2001 From: sqsge <74439428+sqsge@users.noreply.github.com> Date: Tue, 7 Jul 2026 08:05:26 +0800 Subject: [PATCH] fix(guardrails): propagate internal owner attribution to guardrail context (#3839) * fix guardrail attribution for internal owner runs * fix guardrail owner attribution mismatch --- backend/app/gateway/services.py | 41 ++++++++- backend/docs/GUARDRAILS.md | 4 +- backend/tests/test_gateway_services.py | 114 +++++++++++++++++++++++++ 3 files changed, 154 insertions(+), 5 deletions(-) diff --git a/backend/app/gateway/services.py b/backend/app/gateway/services.py index 762dd3ac1..a93c23c6e 100644 --- a/backend/app/gateway/services.py +++ b/backend/app/gateway/services.py @@ -21,7 +21,7 @@ from langchain_core.messages.utils import convert_to_messages from langgraph.types import Command from app.gateway.auth_disabled import AUTH_SOURCE_INTERNAL -from app.gateway.deps import get_checkpointer, get_run_context, get_run_manager, get_stream_bridge +from app.gateway.deps import get_checkpointer, get_local_provider, get_run_context, get_run_manager, get_stream_bridge from app.gateway.internal_auth import ( INTERNAL_OWNER_USER_ID_HEADER_NAME, INTERNAL_SYSTEM_ROLE, @@ -259,7 +259,27 @@ def merge_run_context_overrides(config: dict[str, Any], context: Mapping[str, An runtime_context.setdefault("channel_user_id", context["channel_user_id"]) -def inject_authenticated_user_context(config: dict[str, Any], request: Request) -> None: +async def resolve_trusted_internal_owner_for_attribution(request: Request, owner_user_id: str | None) -> Any | None: + """Resolve the DeerFlow user used only for trusted internal attribution.""" + + if not owner_user_id: + return None + user = getattr(request.state, "user", None) + if getattr(user, "system_role", None) != INTERNAL_SYSTEM_ROLE: + return None + try: + return await get_local_provider().get_user(owner_user_id) + except Exception: + logger.exception("Failed to resolve trusted internal owner %s", sanitize_log_param(owner_user_id)) + return None + + +def inject_authenticated_user_context( + config: dict[str, Any], + request: Request, + *, + internal_owner_user: Any | None = None, +) -> None: """Stamp the authenticated user into the run context for background tools. Tool execution may happen after the request handler has returned, so tools @@ -273,6 +293,20 @@ def inject_authenticated_user_context(config: dict[str, Any], request: Request) return if getattr(user, "system_role", None) == INTERNAL_SYSTEM_ROLE: + runtime_context = config.setdefault("context", {}) + if not isinstance(runtime_context, dict): + return + if internal_owner_user is None: + runtime_context.pop("user_role", None) + runtime_context.pop("oauth_provider", None) + runtime_context.pop("oauth_id", None) + return + owner_user_id = getattr(internal_owner_user, "id", None) + if owner_user_id is not None: + runtime_context["user_id"] = str(owner_user_id) + runtime_context["user_role"] = getattr(internal_owner_user, "system_role", None) + runtime_context["oauth_provider"] = getattr(internal_owner_user, "oauth_provider", None) + runtime_context["oauth_id"] = getattr(internal_owner_user, "oauth_id", None) return runtime_context = config.setdefault("context", {}) @@ -638,7 +672,8 @@ async def start_run( # ``body.config`` is free-form and copied verbatim by # ``build_run_config``; scrub internal-only keys smuggled there. strip_internal_context_keys(config) - inject_authenticated_user_context(config, request) + internal_owner_user = await resolve_trusted_internal_owner_for_attribution(request, owner_user_id) + inject_authenticated_user_context(config, request, internal_owner_user=internal_owner_user) stream_modes = normalize_stream_modes(body.stream_mode) diff --git a/backend/docs/GUARDRAILS.md b/backend/docs/GUARDRAILS.md index 4648b18d1..b41207829 100644 --- a/backend/docs/GUARDRAILS.md +++ b/backend/docs/GUARDRAILS.md @@ -267,9 +267,9 @@ Runtime attribution fields are optional. Providers that need richer policy conte | `run_id` | Link a decision back to one execution run | | `tool_call_id` | Identify the exact tool call that was allowed or denied | -These fields are populated by the Gateway from server-side auth state (`inject_authenticated_user_context` writes `user_id`/`user_role`/`oauth_provider`/`oauth_id` from `request.state.user`; the run worker always sets `thread_id`/`run_id`), and propagated into subagent runs so delegated tool calls are evaluated with the same identity as the lead agent. Client-supplied values cannot override them — the server-side assignment wins. +These fields are populated by the Gateway from server-side auth state (the run worker always sets `thread_id`/`run_id`). For web-authenticated runs, `inject_authenticated_user_context` writes `user_id`/`user_role`/`oauth_provider`/`oauth_id` from `request.state.user`. For trusted IM / internal-auth runs (Slack, Discord, Telegram, Feishu, DingTalk, and other internal callers that provide a trusted owner header), the Gateway resolves the owner user server-side and writes the same attribution fields from that owner. Client-supplied values cannot override them — the server-side assignment wins. -**Known limitation — IM / internal-auth runs.** `inject_authenticated_user_context` early-returns for the internal system role that IM channel workers (Slack, Discord, Telegram, Feishu, DingTalk) and other internal-auth callers are stamped with, so on those runs `user_role`/`oauth_provider`/`oauth_id` stay `None`. `user_id` still resolves (via the channel-bound owner), but role-based policy cannot be applied to channel-delivered work in this release. Web-authenticated runs are unaffected. Threaded alongside the owner `system_role` in a follow-up if channel-originated role policy is needed. +If a trusted internal caller does not resolve to an owner user, the Gateway strips client-supplied `user_role`/`oauth_provider`/`oauth_id` from the run context instead of treating them as authoritative. Any `user_id` already present is left in place for legacy channel storage behavior, but role/oauth-based policy is only applied when the owner user was resolved server-side. For example, if your deployment has user-scoped policy requirements, you can opt into a context-aware provider that passes the runtime fields into an external policy file. This keeps business policy out of Python code and `config.yaml`; the provider only normalizes context, evaluates a configured policy, and maps the result back to `GuardrailDecision`. diff --git a/backend/tests/test_gateway_services.py b/backend/tests/test_gateway_services.py index 10ffc3db2..f21ecf473 100644 --- a/backend/tests/test_gateway_services.py +++ b/backend/tests/test_gateway_services.py @@ -791,6 +791,36 @@ def test_inject_authenticated_user_context_skips_internal_role(): assert config["context"]["user_id"] == "channel-user-7" +def test_inject_authenticated_user_context_strips_internal_spoofed_attribution(): + """Internal callers must not carry role/oauth attribution from request config + unless the gateway resolved a trusted owner user server-side. + """ + from types import SimpleNamespace + + from app.gateway.services import build_run_config, inject_authenticated_user_context + + config = build_run_config( + "thread-1", + { + "context": { + "user_id": "channel-user-7", + "user_role": "admin", + "oauth_provider": "spoofed-provider", + "oauth_id": "spoofed-subject", + } + }, + None, + ) + request = SimpleNamespace(state=SimpleNamespace(user=SimpleNamespace(id="internal-bot", system_role="internal"))) + + inject_authenticated_user_context(config, request) + + assert config["context"]["user_id"] == "channel-user-7" + assert "user_role" not in config["context"] + assert "oauth_provider" not in config["context"] + assert "oauth_id" not in config["context"] + + async def _capture_start_run_graph_input(body): from types import SimpleNamespace from unittest.mock import patch @@ -951,6 +981,90 @@ def test_start_run_uses_internal_owner_header_for_persistence(_stub_app_config): assert task_context["user_id"] == "owner-1" +def test_start_run_stamps_internal_owner_guardrail_attribution(_stub_app_config): + import asyncio + from types import SimpleNamespace + from unittest.mock import patch + + from langgraph.checkpoint.memory import InMemorySaver + from langgraph.store.memory import InMemoryStore + + from app.gateway.internal_auth import INTERNAL_OWNER_USER_ID_HEADER_NAME, INTERNAL_SYSTEM_ROLE + from app.gateway.services import start_run + from deerflow.persistence.thread_meta.memory import MemoryThreadMetaStore + from deerflow.runtime import RunManager + from deerflow.runtime.runs.store.memory import MemoryRunStore + + class _Provider: + async def get_user(self, user_id: str): + assert user_id == "owner-1" + return SimpleNamespace( + id="owner-1", + system_role="user", + oauth_provider="keycloak", + oauth_id="subject-123", + ) + + async def _scenario(): + thread_store = MemoryThreadMetaStore(InMemoryStore()) + await thread_store.create("channel-thread", user_id="owner-1", metadata={}) + run_manager = RunManager(store=MemoryRunStore()) + state = SimpleNamespace( + stream_bridge=SimpleNamespace(), + run_manager=run_manager, + checkpointer=InMemorySaver(), + store=InMemoryStore(), + run_event_store=SimpleNamespace(), + run_events_config=None, + thread_store=thread_store, + ) + request = SimpleNamespace( + headers={INTERNAL_OWNER_USER_ID_HEADER_NAME: "owner-1"}, + state=SimpleNamespace(user=SimpleNamespace(id="default", system_role=INTERNAL_SYSTEM_ROLE)), + app=SimpleNamespace(state=state), + ) + body = SimpleNamespace( + assistant_id="lead_agent", + input={"messages": [{"role": "human", "content": "hi"}]}, + metadata={}, + config={ + "context": { + "user_role": "admin", + "oauth_provider": "spoofed-provider", + "oauth_id": "spoofed-subject", + } + }, + context={"user_id": "spoofed-client"}, + on_disconnect="cancel", + multitask_strategy="reject", + stream_mode=None, + stream_subgraphs=False, + interrupt_before=None, + interrupt_after=None, + ) + captured_context: dict[str, object] = {} + + async def fake_run_agent(*args, **kwargs): + captured_context.update(kwargs["config"]["context"]) + + with ( + patch("app.gateway.services.get_local_provider", return_value=_Provider()), + patch("app.gateway.services.resolve_agent_factory", return_value=object()), + patch("app.gateway.services.run_agent", side_effect=fake_run_agent), + ): + record = await start_run(body, "channel-thread", request) + await record.task + + return captured_context + + context = asyncio.run(_scenario()) + + assert context["user_id"] == "owner-1" + assert context["user_role"] == "user" + assert context["oauth_provider"] == "keycloak" + assert context["oauth_id"] == "subject-123" + + def test_launch_scheduled_thread_run_marks_context_non_interactive(_stub_app_config): import asyncio from types import SimpleNamespace