From c626b92bcd5f9a454c7fc5e8ee44105aa67eaa6f Mon Sep 17 00:00:00 2001 From: LeonOstrez Date: Thu, 26 Feb 2026 12:11:28 +0100 Subject: [PATCH] Implement OpenClaw text chat proxy --- src/gateway/pazi-context-http.ts | 77 +++++++++++++++++++++ src/gateway/pazi-proxy.ts | 113 +++++++++++++++++++++++++++++++ src/gateway/server-http.ts | 4 ++ src/gateway/server.impl.ts | 6 ++ 4 files changed, 200 insertions(+) create mode 100644 src/gateway/pazi-context-http.ts create mode 100644 src/gateway/pazi-proxy.ts diff --git a/src/gateway/pazi-context-http.ts b/src/gateway/pazi-context-http.ts new file mode 100644 index 0000000000000..2a079fc814f1d --- /dev/null +++ b/src/gateway/pazi-context-http.ts @@ -0,0 +1,77 @@ +import type { IncomingMessage, ServerResponse } from "node:http"; +import { createSubsystemLogger } from "../logging/subsystem.js"; +import { setProxyContext } from "./pazi-proxy.js"; + +type PaziContextRequest = { + userId?: string; + proxyToken?: string; +}; + +type PaziContextOpts = { + gatewayToken?: string; +}; + +const log = createSubsystemLogger("gateway/pazi-context"); + +function writeJson(res: ServerResponse, status: number, body: unknown) { + res.writeHead(status, { "Content-Type": "application/json; charset=utf-8" }); + res.end(JSON.stringify(body)); +} + +async function readJson(req: IncomingMessage): Promise { + const chunks: Buffer[] = []; + for await (const chunk of req) { + if (typeof chunk === "string") { + chunks.push(Buffer.from(chunk)); + } else { + chunks.push(chunk); + } + } + try { + const body = JSON.parse(Buffer.concat(chunks).toString()); + if (body && typeof body === "object") { + return body as PaziContextRequest; + } + } catch { + return null; + } + return null; +} + +export async function handlePaziContextRequest( + req: IncomingMessage, + res: ServerResponse, + opts: PaziContextOpts, +): Promise { + if (req.method !== "POST" || req.url !== "/pazi/context") { + return false; + } + + if (!opts.gatewayToken) { + log.warn("pazi context request rejected: gateway token missing"); + writeJson(res, 500, { error: "gateway_token_missing" }); + return true; + } + + const authHeader = req.headers.authorization; + if (authHeader !== `Bearer ${opts.gatewayToken}`) { + writeJson(res, 401, { error: "unauthorized" }); + return true; + } + + const body = await readJson(req); + if (!body) { + writeJson(res, 400, { error: "invalid JSON" }); + return true; + } + + const { userId, proxyToken } = body; + if (!userId || !proxyToken) { + writeJson(res, 400, { error: "missing userId or proxyToken" }); + return true; + } + + setProxyContext({ userId, proxyToken }); + writeJson(res, 200, { ok: true }); + return true; +} diff --git a/src/gateway/pazi-proxy.ts b/src/gateway/pazi-proxy.ts new file mode 100644 index 0000000000000..80eac932108ce --- /dev/null +++ b/src/gateway/pazi-proxy.ts @@ -0,0 +1,113 @@ +import http from "node:http"; +import https from "node:https"; +import type { IncomingHttpHeaders } from "node:http"; +import { createSubsystemLogger } from "../logging/subsystem.js"; + +type ProxyContext = { userId: string; proxyToken: string }; + +type HttpRequest = typeof http.request; + +type ProxyError = { error: string; message?: string }; + +const log = createSubsystemLogger("gateway/pazi-proxy"); + +let currentContext: ProxyContext | null = null; + +export function setProxyContext(ctx: ProxyContext): void { + currentContext = ctx; +} + +export function clearProxyContext(): void { + currentContext = null; +} + +function requestForUrl(url: URL): HttpRequest { + return url.protocol === "https:" ? https.request : http.request; +} + +function pickAnthropicHeaders(incoming: IncomingHttpHeaders): Record { + const forward: Record = {}; + const passthrough = ["anthropic-version", "anthropic-beta", "accept", "content-type"] as const; + for (const key of passthrough) { + const value = incoming[key]; + if (typeof value === "string") { + forward[key] = value; + } + } + return forward; +} + +function writeJson(res: http.ServerResponse, status: number, body: ProxyError) { + res.writeHead(status, { "Content-Type": "application/json; charset=utf-8" }); + res.end(JSON.stringify(body)); +} + +export async function startPaziProxy(port: number): Promise { + const paziApiUrl = process.env.PAZI_API_URL; + if (!paziApiUrl) { + log.info("pazi proxy disabled (PAZI_API_URL not set)"); + return; + } + + const server = http.createServer(async (req, res) => { + if (req.method !== "POST" || !req.url?.startsWith("/v1/messages")) { + res.writeHead(404, { "Content-Type": "text/plain; charset=utf-8" }); + res.end("Not Found"); + return; + } + + const context = currentContext; + if (!context) { + writeJson(res, 503, { error: "no billing context set" }); + return; + } + + const chunks: Buffer[] = []; + for await (const chunk of req) { + if (typeof chunk === "string") { + chunks.push(Buffer.from(chunk)); + } else { + chunks.push(chunk); + } + } + const body = Buffer.concat(chunks); + + const target = new URL("/anthropic/v1/messages", paziApiUrl); + const doRequest = requestForUrl(target); + + const proxyReq = doRequest( + target, + { + method: "POST", + headers: { + ...pickAnthropicHeaders(req.headers), + "X-Proxy-Token": context.proxyToken, + "X-User-Id": context.userId, + }, + }, + (proxyRes) => { + res.writeHead(proxyRes.statusCode || 500, proxyRes.headers); + proxyRes.pipe(res); + }, + ); + + proxyReq.on("error", (err: Error) => { + log.warn(`pazi proxy error: ${String(err)}`); + if (!res.headersSent) { + writeJson(res, 502, { error: "proxy_error", message: err.message }); + } + }); + + proxyReq.write(body); + proxyReq.end(); + }); + + server.on("clientError", (err, socket) => { + log.warn(`pazi proxy client error: ${String(err)}`); + socket.end("HTTP/1.1 400 Bad Request\r\n\r\n"); + }); + + server.listen(port, "127.0.0.1", () => { + log.info(`pazi proxy listening on 127.0.0.1:${port}`); + }); +} diff --git a/src/gateway/server-http.ts b/src/gateway/server-http.ts index 41d04d5d3ac89..199cec804a350 100644 --- a/src/gateway/server-http.ts +++ b/src/gateway/server-http.ts @@ -58,6 +58,7 @@ import { sendGatewayAuthFailure, setDefaultSecurityHeaders } from "./http-common import { getBearerToken } from "./http-utils.js"; import { handleOpenAiHttpRequest } from "./openai-http.js"; import { handleOpenResponsesHttpRequest } from "./openresponses-http.js"; +import { handlePaziContextRequest } from "./pazi-context-http.js"; import { GATEWAY_CLIENT_MODES, normalizeGatewayClientMode } from "./protocol/client-info.js"; import type { GatewayWsClient } from "./server/ws-types.js"; import { handleToolsInvokeHttpRequest } from "./tools-invoke-http.js"; @@ -476,6 +477,9 @@ export function createGatewayHttpServer(opts: { if (await handleHooksRequest(req, res)) { return; } + if (await handlePaziContextRequest(req, res, { gatewayToken: resolvedAuth.token })) { + return; + } if ( await handleToolsInvokeHttpRequest(req, res, { auth: resolvedAuth, diff --git a/src/gateway/server.impl.ts b/src/gateway/server.impl.ts index 3dbd86e1e5ee3..829e8ab3d1833 100644 --- a/src/gateway/server.impl.ts +++ b/src/gateway/server.impl.ts @@ -78,6 +78,7 @@ import { createGatewayRuntimeState } from "./server-runtime-state.js"; import { resolveSessionKeyForRun } from "./server-session-key.js"; import { logGatewayStartup } from "./server-startup-log.js"; import { startGatewaySidecars } from "./server-startup.js"; +import { startPaziProxy } from "./pazi-proxy.js"; import { startGatewayTailscaleExposure } from "./server-tailscale.js"; import { createWizardSessionTracker } from "./server-wizard-sessions.js"; import { attachGatewayWsHandlers } from "./server-ws-runtime.js"; @@ -268,6 +269,11 @@ export async function startGatewayServer( ); } } + if (!minimalTestGateway) { + const proxyPortRaw = process.env.PAZI_PROXY_PORT; + const proxyPort = proxyPortRaw ? Number.parseInt(proxyPortRaw, 10) : 8765; + await startPaziProxy(Number.isFinite(proxyPort) ? proxyPort : 8765); + } const diagnosticsEnabled = isDiagnosticsEnabled(cfgAtStart); if (diagnosticsEnabled) { startDiagnosticHeartbeat();