fix(gateway): preserve verified owner identity for OpenAI HTTP (#113638)

Co-authored-by: Peter Steinberger <steipete@golden-gate.local>

src/gateway/openai-http.test.ts

@@ -18,6 +18,7 @@ import { emitAgentEvent } from "../infra/agent-events.js";
 import { enqueueCommandInLane } from "../process/command-queue.js";
 import { getActiveGatewayRootWorkCount } from "../process/gateway-work-admission.js";
 import { createDeferred } from "../test-utils/deferred.js";
+import { withEnvAsync } from "../test-utils/env.js";
 import { buildAssistantDeltaResult } from "./test-helpers.agent-results.js";
 import {
   agentCommand,
@@ -62,10 +63,17 @@ async function startServer(port: number, opts?: { openAiChatCompletionsEnabled?:
   });
 }
 
-async function startTokenServer(port: number, opts?: { openAiChatCompletionsEnabled?: boolean }) {
+async function startSharedSecretServer(
+  port: number,
+  mode: "token" | "password",
+  opts?: { openAiChatCompletionsEnabled?: boolean },
+) {
   return await startGatewayServer(port, {
     host: "127.0.0.1",
-    auth: { mode: "token", token: "secret" },
+    auth:
+      mode === "token"
+        ? { mode: "token", token: "secret" }
+        : { mode: "password", password: "secret" },
     controlUiEnabled: false,
     openAiChatCompletionsEnabled: opts?.openAiChatCompletionsEnabled ?? true,
   });
@@ -116,6 +124,7 @@ type FirstAgentCommandOptions = {
   message?: string;
   messageChannel?: string;
   model?: string;
+  senderIsOwner?: boolean;
   sessionKey?: string;
   streamParams?: {
     frequencyPenalty?: number;
@@ -2658,33 +2667,165 @@ describe("OpenAI-compatible HTTP API (e2e)", () => {
     expect(usageChunks).toHaveLength(0);
   });
 
-  it("accepts shared-secret bearer callers", async () => {
-    const port = await getFreePort();
-    const server = await startTokenServer(port);
-    try {
-      agentCommand.mockClear();
-      agentCommand.mockResolvedValueOnce({ payloads: [{ text: "hello" }] } as never);
+  it("preserves declared owner identity for streaming and non-streaming private callers", async () => {
+    for (const stream of [false, true]) {
+      for (const { scopes, senderIsOwner } of [
+        { scopes: "operator.write", senderIsOwner: false },
+        { scopes: "operator.admin, operator.write", senderIsOwner: true },
+      ]) {
+        agentCommand.mockClear();
+        agentCommand.mockResolvedValueOnce({ payloads: [{ text: "hello" }] } as never);
 
-      const res = await fetch(`http://127.0.0.1:${port}/v1/chat/completions`, {
-        method: "POST",
-        headers: {
-          authorization: "Bearer secret",
-          "content-type": "application/json",
-          "x-openclaw-scopes": "operator.approvals",
-        },
-        body: JSON.stringify({
-          model: "openclaw",
-          messages: [{ role: "user", content: "hi" }],
-        }),
-      });
+        const res = await postChatCompletions(
+          enabledPort,
+          {
+            stream,
+            model: "openclaw",
+            messages: [{ role: "user", content: "hi" }],
+          },
+          {
+            "x-openclaw-scopes": scopes,
+            "x-openclaw-sender-is-owner": "true",
+          },
+        );
 
-      expect(res.status).toBe(200);
-      await res.text();
-    } finally {
-      await server.close({ reason: "openai token auth owner test done" });
+        expect(res.status).toBe(200);
+        await res.text();
+        expect(agentCommand).toHaveBeenCalledTimes(1);
+        expect(firstAgentCommandOptions()?.senderIsOwner).toBe(senderIsOwner);
+      }
     }
   });
 
+  it("preserves verified trusted-proxy owner identity for both response modes", async () => {
+    await withEnvAsync(
+      { OPENCLAW_GATEWAY_TOKEN: undefined, OPENCLAW_GATEWAY_PASSWORD: undefined },
+      async () => {
+        const port = await getFreePort();
+        let server: Awaited<ReturnType<typeof startGatewayServer>> | undefined;
+        const previousGatewayAuth = testState.gatewayAuth;
+        const trustedProxyAuth = {
+          mode: "trusted-proxy" as const,
+          trustedProxy: {
+            userHeader: "x-forwarded-user",
+            requiredHeaders: ["x-forwarded-proto"],
+            allowLoopback: true,
+          },
+        };
+        testState.gatewayAuth = trustedProxyAuth;
+        try {
+          await writeGatewayConfig({
+            gateway: {
+              auth: trustedProxyAuth,
+              trustedProxies: ["127.0.0.1"],
+            },
+          });
+          resetConfigRuntimeState();
+          server = await startGatewayServer(port, {
+            host: "127.0.0.1",
+            auth: trustedProxyAuth,
+            controlUiEnabled: false,
+            openAiChatCompletionsEnabled: true,
+          });
+
+          for (const stream of [false, true]) {
+            for (const { scopes, senderIsOwner } of [
+              { scopes: "operator.write", senderIsOwner: false },
+              { scopes: "operator.admin, operator.write", senderIsOwner: true },
+            ]) {
+              agentCommand.mockClear();
+              agentCommand.mockResolvedValueOnce({ payloads: [{ text: "hello" }] } as never);
+
+              const res = await postChatCompletions(
+                port,
+                {
+                  stream,
+                  model: "openclaw",
+                  messages: [{ role: "user", content: "hi" }],
+                },
+                {
+                  "x-forwarded-proto": "https",
+                  "x-forwarded-user": "operator@example.com",
+                  "x-openclaw-scopes": scopes,
+                  "x-openclaw-sender-is-owner": "true",
+                },
+              );
+
+              expect(res.status).toBe(200);
+              await res.text();
+              expect(agentCommand).toHaveBeenCalledTimes(1);
+              expect(firstAgentCommandOptions()?.senderIsOwner).toBe(senderIsOwner);
+            }
+          }
+
+          agentCommand.mockClear();
+          const unauthorized = await postChatCompletions(
+            port,
+            { model: "openclaw", messages: [{ role: "user", content: "hi" }] },
+            {
+              "x-forwarded-proto": "https",
+              "x-openclaw-scopes": "operator.admin, operator.write",
+              "x-openclaw-sender-is-owner": "true",
+            },
+          );
+          expect(unauthorized.status).toBe(401);
+          await unauthorized.text();
+          expect(agentCommand).not.toHaveBeenCalled();
+        } finally {
+          await server?.close({ reason: "openai trusted-proxy auth owner test done" });
+          testState.gatewayAuth = previousGatewayAuth;
+          await writeGatewayConfig({});
+          resetConfigRuntimeState();
+        }
+      },
+    );
+  });
+
+  it.each(["token", "password"] as const)(
+    "preserves owner identity for streaming and non-streaming %s-authenticated callers",
+    async (mode) => {
+      const port = await getFreePort();
+      const server = await startSharedSecretServer(port, mode);
+      try {
+        for (const stream of [false, true]) {
+          agentCommand.mockClear();
+          agentCommand.mockResolvedValueOnce({ payloads: [{ text: "hello" }] } as never);
+
+          const res = await postChatCompletions(
+            port,
+            {
+              stream,
+              model: "openclaw",
+              messages: [{ role: "user", content: "hi" }],
+            },
+            {
+              authorization: "Bearer secret",
+              "x-openclaw-scopes": "operator.approvals",
+              "x-openclaw-sender-is-owner": "false",
+            },
+          );
+
+          expect(res.status).toBe(200);
+          await res.text();
+          expect(agentCommand).toHaveBeenCalledTimes(1);
+          expect(firstAgentCommandOptions()?.senderIsOwner).toBe(true);
+        }
+
+        agentCommand.mockClear();
+        const unauthorized = await postChatCompletions(
+          port,
+          { model: "openclaw", messages: [{ role: "user", content: "hi" }] },
+          { authorization: "Bearer wrong", "x-openclaw-sender-is-owner": "true" },
+        );
+        expect(unauthorized.status).toBe(401);
+        await unauthorized.text();
+        expect(agentCommand).not.toHaveBeenCalled();
+      } finally {
+        await server.close({ reason: `openai ${mode} auth owner test done` });
+      }
+    },
+  );
+
   it("aborts agent command when streaming client disconnects", { timeout: 15_000 }, async () => {
     const port = enabledPort;
     let serverAbortSignal: AbortSignal | undefined;

src/gateway/openai-http.ts

@@ -62,6 +62,7 @@ import {
   resolveGatewayRequestContext,
   resolveOpenAiCompatModelOverride,
   resolveOpenAiCompatibleHttpOperatorScopes,
+  resolveOpenAiCompatibleHttpSenderIsOwner,
 } from "./http-utils.js";
 import { normalizeInputHostnameAllowlist } from "./input-allowlist.js";
 import { resolveOpenAiCompatError, validateOpenAiSamplingParams } from "./openai-compat-errors.js";
@@ -158,6 +159,7 @@ function buildAgentCommandInput(params: {
   sessionKey: string;
   runId: string;
   messageChannel: string;
+  senderIsOwner: boolean;
   abortSignal?: AbortSignal;
   streamParams?: AgentStreamParams;
 }) {
@@ -171,6 +173,7 @@ function buildAgentCommandInput(params: {
     runId: params.runId,
     deliver: false as const,
     messageChannel: params.messageChannel,
+    senderIsOwner: params.senderIsOwner,
     bestEffortDeliver: false as const,
     allowModelOverride: params.modelOverride !== undefined,
     abortSignal: params.abortSignal,
@@ -897,6 +900,7 @@ export async function handleOpenAiHttpRequest(
     sendMissingScopeForbidden(res, modelOverrideAuth.missingScope);
     return true;
   }
+  const senderIsOwner = resolveOpenAiCompatibleHttpSenderIsOwner(req, handled.requestAuth);
   const payload = coerceRequest(handled.body);
   const stream = Boolean(payload.stream);
   const streamIncludeUsage = stream && resolveIncludeUsageForStreaming(payload);
@@ -1069,6 +1073,7 @@ export async function handleOpenAiHttpRequest(
     sessionKey,
     runId,
     messageChannel,
+    senderIsOwner,
     abortSignal: abortController.signal,
     streamParams,
   });

src/gateway/openresponses-http.test.ts

@@ -13,6 +13,7 @@ import { emitAgentEvent } from "../infra/agent-events.js";
 import { enqueueCommandInLane } from "../process/command-queue.js";
 import { getActiveGatewayRootWorkCount } from "../process/gateway-work-admission.js";
 import { createDeferred } from "../test-utils/deferred.js";
+import { withEnvAsync } from "../test-utils/env.js";
 import { IMAGE_ONLY_USER_MESSAGE } from "./agent-prompt.js";
 import { buildAssistantDeltaResult } from "./test-helpers.agent-results.js";
 import {
@@ -82,11 +83,18 @@ async function startServer(port: number, opts?: { openResponsesEnabled?: boolean
   );
 }
 
-async function startTokenServer(port: number, opts?: { openResponsesEnabled?: boolean }) {
+async function startSharedSecretServer(
+  port: number,
+  mode: "token" | "password",
+  opts?: { openResponsesEnabled?: boolean },
+) {
   const { startGatewayServer } = await import("./server.js");
   const serverOpts = {
     host: "127.0.0.1",
-    auth: { mode: "token" as const, token: "secret" },
+    auth:
+      mode === "token"
+        ? { mode: "token" as const, token: "secret" }
+        : { mode: "password" as const, password: "secret" },
     controlUiEnabled: false,
   } as const;
   return await startGatewayServer(
@@ -1125,76 +1133,158 @@ describe("OpenResponses HTTP API (e2e)", () => {
     },
   );
 
-  it("accepts write-scoped and admin-scoped HTTP callers", async () => {
+  it("preserves declared owner identity for streaming and non-streaming private callers", async () => {
     const port = enabledPort;
+    for (const stream of [false, true]) {
+      for (const { scopes, senderIsOwner } of [
+        { scopes: "operator.write", senderIsOwner: false },
+        { scopes: "operator.admin, operator.write", senderIsOwner: true },
+      ]) {
+        agentCommand.mockClear();
+        agentCommand.mockResolvedValueOnce({ payloads: [{ text: "hello" }] } as never);
 
-    agentCommand.mockClear();
-    agentCommand.mockResolvedValueOnce({ payloads: [{ text: "hello" }] } as never);
+        const res = await postResponses(
+          port,
+          { stream, model: "openclaw", input: "hi" },
+          {
+            "x-openclaw-scopes": scopes,
+            "x-openclaw-sender-is-owner": "true",
+          },
+        );
 
-    const writeScopeResponse = await postResponses(port, {
-      model: "openclaw",
-      input: "hi",
-    });
-    expect(writeScopeResponse.status).toBe(200);
-    await ensureResponseConsumed(writeScopeResponse);
-
-    agentCommand.mockClear();
-    agentCommand.mockResolvedValueOnce({ payloads: [{ text: "hello" }] } as never);
-
-    const adminScopeResponse = await postResponses(
-      port,
-      { model: "openclaw", input: "hi" },
-      { "x-openclaw-scopes": "operator.admin, operator.write" },
-    );
-    expect(adminScopeResponse.status).toBe(200);
-    await ensureResponseConsumed(adminScopeResponse);
-
-    agentCommand.mockClear();
-    agentCommand.mockImplementationOnce((async (opts: unknown) =>
-      buildAssistantDeltaResult({
-        opts,
-        emit: emitAgentEvent,
-        deltas: ["he", "llo"],
-        text: "hello",
-      })) as never);
-
-    const streamingResponse = await postResponses(
-      port,
-      { stream: true, model: "openclaw", input: "hi" },
-      { "x-openclaw-scopes": "operator.admin, operator.write" },
-    );
-    expect(streamingResponse.status).toBe(200);
-    const streamingEvents = parseSseEvents(await streamingResponse.text());
-    expect(streamingEvents.map((event) => event.event)).toContain("response.completed");
-  });
-
-  it("accepts shared-secret bearer callers", async () => {
-    const port = await getFreePort();
-    const server = await startTokenServer(port);
-    try {
-      agentCommand.mockClear();
-      agentCommand.mockResolvedValueOnce({ payloads: [{ text: "hello" }] } as never);
-
-      const res = await fetch(`http://127.0.0.1:${port}/v1/responses`, {
-        method: "POST",
-        headers: {
-          authorization: "Bearer secret",
-          "content-type": "application/json",
-          "x-openclaw-scopes": "operator.approvals",
-        },
-        body: JSON.stringify({
-          model: "openclaw",
-          input: "hi",
-        }),
-      });
-
-      expect(res.status).toBe(200);
-      await ensureResponseConsumed(res);
-    } finally {
-      await server.close({ reason: "openresponses token auth owner test done" });
+        expect(res.status).toBe(200);
+        const body = await res.text();
+        if (stream) {
+          expect(parseSseEvents(body).map((event) => event.event)).toContain("response.completed");
+        }
+        expect(agentCommand).toHaveBeenCalledTimes(1);
+        expect(firstAgentOpts().senderIsOwner).toBe(senderIsOwner);
+      }
     }
   });
 
+  it("preserves verified trusted-proxy owner identity for both response modes", async () => {
+    await withEnvAsync(
+      { OPENCLAW_GATEWAY_TOKEN: undefined, OPENCLAW_GATEWAY_PASSWORD: undefined },
+      async () => {
+        const port = await getFreePort();
+        const { startGatewayServer } = await import("./server.js");
+        let server: Awaited<ReturnType<typeof startGatewayServer>> | undefined;
+        const previousGatewayAuth = testState.gatewayAuth;
+        const trustedProxyAuth = {
+          mode: "trusted-proxy" as const,
+          trustedProxy: {
+            userHeader: "x-forwarded-user",
+            requiredHeaders: ["x-forwarded-proto"],
+            allowLoopback: true,
+          },
+        };
+        testState.gatewayAuth = trustedProxyAuth;
+        try {
+          await writeGatewayConfig({
+            gateway: {
+              auth: trustedProxyAuth,
+              trustedProxies: ["127.0.0.1"],
+            },
+          });
+          resetConfigRuntimeState();
+          server = await startGatewayServer(port, {
+            host: "127.0.0.1",
+            auth: trustedProxyAuth,
+            controlUiEnabled: false,
+            openResponsesEnabled: true,
+          });
+
+          for (const stream of [false, true]) {
+            for (const { scopes, senderIsOwner } of [
+              { scopes: "operator.write", senderIsOwner: false },
+              { scopes: "operator.admin, operator.write", senderIsOwner: true },
+            ]) {
+              agentCommand.mockClear();
+              agentCommand.mockResolvedValueOnce({ payloads: [{ text: "hello" }] } as never);
+
+              const res = await postResponses(
+                port,
+                { stream, model: "openclaw", input: "hi" },
+                {
+                  "x-forwarded-proto": "https",
+                  "x-forwarded-user": "operator@example.com",
+                  "x-openclaw-scopes": scopes,
+                  "x-openclaw-sender-is-owner": "true",
+                },
+              );
+
+              expect(res.status).toBe(200);
+              await ensureResponseConsumed(res);
+              expect(agentCommand).toHaveBeenCalledTimes(1);
+              expect(firstAgentOpts().senderIsOwner).toBe(senderIsOwner);
+            }
+          }
+
+          agentCommand.mockClear();
+          const unauthorized = await postResponses(
+            port,
+            { model: "openclaw", input: "hi" },
+            {
+              "x-forwarded-proto": "https",
+              "x-openclaw-scopes": "operator.admin, operator.write",
+              "x-openclaw-sender-is-owner": "true",
+            },
+          );
+          expect(unauthorized.status).toBe(401);
+          await ensureResponseConsumed(unauthorized);
+          expect(agentCommand).not.toHaveBeenCalled();
+        } finally {
+          await server?.close({ reason: "openresponses trusted-proxy auth owner test done" });
+          testState.gatewayAuth = previousGatewayAuth;
+          await writeGatewayConfig({});
+          resetConfigRuntimeState();
+        }
+      },
+    );
+  });
+
+  it.each(["token", "password"] as const)(
+    "preserves owner identity for streaming and non-streaming %s-authenticated callers",
+    async (mode) => {
+      const port = await getFreePort();
+      const server = await startSharedSecretServer(port, mode);
+      try {
+        for (const stream of [false, true]) {
+          agentCommand.mockClear();
+          agentCommand.mockResolvedValueOnce({ payloads: [{ text: "hello" }] } as never);
+
+          const res = await postResponses(
+            port,
+            { stream, model: "openclaw", input: "hi" },
+            {
+              authorization: "Bearer secret",
+              "x-openclaw-scopes": "operator.approvals",
+              "x-openclaw-sender-is-owner": "false",
+            },
+          );
+
+          expect(res.status).toBe(200);
+          await ensureResponseConsumed(res);
+          expect(agentCommand).toHaveBeenCalledTimes(1);
+          expect(firstAgentOpts().senderIsOwner).toBe(true);
+        }
+
+        agentCommand.mockClear();
+        const unauthorized = await postResponses(
+          port,
+          { model: "openclaw", input: "hi" },
+          { authorization: "Bearer wrong", "x-openclaw-sender-is-owner": "true" },
+        );
+        expect(unauthorized.status).toBe(401);
+        await ensureResponseConsumed(unauthorized);
+        expect(agentCommand).not.toHaveBeenCalled();
+      } finally {
+        await server.close({ reason: `openresponses ${mode} auth owner test done` });
+      }
+    },
+  );
+
   it("keeps streamed agent work admitted after the HTTP handler returns", async () => {
     const idleRootCount = getActiveGatewayRootWorkCount();
     const continueAgent = createDeferred();

src/gateway/openresponses-http.ts

@@ -59,6 +59,7 @@ import {
   resolveGatewayRequestContext,
   resolveOpenAiCompatModelOverride,
   resolveOpenAiCompatibleHttpOperatorScopes,
+  resolveOpenAiCompatibleHttpSenderIsOwner,
 } from "./http-utils.js";
 import { normalizeInputHostnameAllowlist } from "./input-allowlist.js";
 import {
@@ -423,6 +424,7 @@ async function runResponsesAgentCommand(params: {
   sessionKey: string;
   runId: string;
   messageChannel: string;
+  senderIsOwner: boolean;
   deps: CliDeps;
   abortSignal?: AbortSignal;
 }) {
@@ -438,6 +440,7 @@ async function runResponsesAgentCommand(params: {
       runId: params.runId,
       deliver: false,
       messageChannel: params.messageChannel,
+      senderIsOwner: params.senderIsOwner,
       bestEffortDeliver: false,
       allowModelOverride: params.modelOverride !== undefined,
       abortSignal: params.abortSignal,
@@ -479,6 +482,7 @@ export async function handleOpenResponsesHttpRequest(
     sendMissingScopeForbidden(res, modelOverrideAuth.missingScope);
     return true;
   }
+  const senderIsOwner = resolveOpenAiCompatibleHttpSenderIsOwner(req, handled.requestAuth);
   // Validate request body with Zod
   const parseResult = CreateResponseBodySchema.safeParse(handled.body);
   if (!parseResult.success) {
@@ -752,6 +756,7 @@ export async function handleOpenResponsesHttpRequest(
         sessionKey,
         runId: responseId,
         messageChannel,
+        senderIsOwner,
         deps,
         abortSignal: abortController.signal,
       });
@@ -1138,6 +1143,7 @@ export async function handleOpenResponsesHttpRequest(
         sessionKey,
         runId: responseId,
         messageChannel,
+        senderIsOwner,
         deps,
         abortSignal: abortController.signal,
       });