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

@@ -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)
 

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`.
 

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