openclaw/extensions/admin-http-rpc/src/handler.ts

/**
 * HTTP handler for the Admin RPC endpoint. It validates JSON requests, enforces
 * the method allowlist, dispatches gateway methods, and maps errors to HTTP.
 */
import { randomUUID } from "node:crypto";
import type { IncomingMessage, ServerResponse } from "node:http";
import { dispatchGatewayMethod } from "openclaw/plugin-sdk/gateway-method-runtime";
import { isRecord } from "openclaw/plugin-sdk/string-coerce-runtime";
import {
  requestBodyErrorToText,
  WEBHOOK_BODY_READ_DEFAULTS,
} from "openclaw/plugin-sdk/webhook-request-guards";
import { isAdminHttpRpcAllowedMethod, listAdminHttpRpcAllowedMethods } from "./methods.js";

const ErrorCodes = {
  AGENT_TIMEOUT: "AGENT_TIMEOUT",
  APPROVAL_NOT_FOUND: "APPROVAL_NOT_FOUND",
  INVALID_REQUEST: "INVALID_REQUEST",
  NOT_LINKED: "NOT_LINKED",
  NOT_PAIRED: "NOT_PAIRED",
  UNAVAILABLE: "UNAVAILABLE",
} as const;

type RpcBody = {
  id?: unknown;
  method?: unknown;
  params?: unknown;
};

type RpcError = {
  code: string;
  message: string;
  details?: unknown;
  retryable?: boolean;
  retryAfterMs?: number;
};

type RpcResponse =
  | { id: string; ok: true; payload: unknown; meta?: Record<string, unknown> }
  | { id: string; ok: false; error: RpcError; meta?: Record<string, unknown> };

type ParsedRequest = {
  id: string;
  method: string;
  params?: unknown;
};

type RequestBodyLimitFailureCode =
  | "PAYLOAD_TOO_LARGE"
  | "REQUEST_BODY_TIMEOUT"
  | "CONNECTION_CLOSED";
type ReadJsonBodyResult =
  | { ok: true; value: unknown }
  | {
      ok: false;
      status: number;
      message: string;
      closeAfterResponse?: boolean;
    };
type ReadRawBodyResult =
  | { ok: true; raw: string }
  | { ok: false; code: RequestBodyLimitFailureCode; closeAfterResponse?: boolean };

function createError(code: string, message: string): RpcError {
  return { code, message };
}

function rpcHttpStatus(response: RpcResponse): number {
  if (response.ok) {
    return 200;
  }
  switch (response.error.code) {
    case ErrorCodes.INVALID_REQUEST:
      return 400;
    case ErrorCodes.APPROVAL_NOT_FOUND:
      return 404;
    case ErrorCodes.UNAVAILABLE:
      return 503;
    case ErrorCodes.AGENT_TIMEOUT:
      return 504;
    case ErrorCodes.NOT_LINKED:
    case ErrorCodes.NOT_PAIRED:
      return 409;
    default:
      return 500;
  }
}

function sendJson(res: ServerResponse, status: number, body: unknown): void {
  res.statusCode = status;
  res.setHeader("Cache-Control", "no-store");
  res.setHeader("Content-Type", "application/json; charset=utf-8");
  res.end(JSON.stringify(body));
}

function sendError(res: ServerResponse, status: number, error: { type: string; message: string }) {
  sendJson(res, status, { ok: false, error });
}

async function readJsonBody(
  req: IncomingMessage,
  maxBytes: number,
  timeoutMs: number,
): Promise<ReadJsonBodyResult> {
  const body = await readRawBodyWithLimit(req, maxBytes, timeoutMs);
  if (!body.ok) {
    return {
      ok: false,
      status: statusForBodyErrorCode(body.code),
      message: requestBodyErrorToText(body.code),
      closeAfterResponse: body.closeAfterResponse,
    };
  }

  const raw = body.raw.trim();
  if (!raw) {
    return { ok: false, status: 400, message: "request body must be JSON" };
  }
  try {
    return { ok: true, value: JSON.parse(raw) as unknown };
  } catch {
    return { ok: false, status: 400, message: "request body must be valid JSON" };
  }
}

function statusForBodyErrorCode(code: RequestBodyLimitFailureCode): number {
  switch (code) {
    case "PAYLOAD_TOO_LARGE":
      return 413;
    case "REQUEST_BODY_TIMEOUT":
      return 408;
    case "CONNECTION_CLOSED":
      return 400;
  }
  return 400;
}

async function readRawBodyWithLimit(
  req: IncomingMessage,
  maxBytes: number,
  timeoutMs: number,
): Promise<ReadRawBodyResult> {
  const declaredLength = readDeclaredContentLength(req);
  if (declaredLength !== null && declaredLength > maxBytes) {
    req.pause();
    return { ok: false, code: "PAYLOAD_TOO_LARGE", closeAfterResponse: true };
  }

  return await new Promise((resolve) => {
    let done = false;
    let ended = false;
    let totalBytes = 0;
    const chunks: Buffer[] = [];

    const cleanup = () => {
      req.removeListener("data", onData);
      req.removeListener("end", onEnd);
      req.removeListener("error", onError);
      req.removeListener("close", onClose);
      clearTimeout(timer);
    };

    const finish = (result: ReadRawBodyResult) => {
      if (done) {
        return;
      }
      done = true;
      cleanup();
      resolve(result);
    };

    const failAfterResponse = (code: RequestBodyLimitFailureCode) => {
      req.pause();
      finish({ ok: false, code, closeAfterResponse: true });
    };

    const timer = setTimeout(() => {
      failAfterResponse("REQUEST_BODY_TIMEOUT");
    }, timeoutMs);

    const onData = (chunk: Buffer | string) => {
      if (done) {
        return;
      }
      const buffer = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk);
      totalBytes += buffer.byteLength;
      if (totalBytes > maxBytes) {
        failAfterResponse("PAYLOAD_TOO_LARGE");
        return;
      }
      chunks.push(buffer);
    };

    const onEnd = () => {
      ended = true;
      finish({ ok: true, raw: Buffer.concat(chunks).toString("utf8") });
    };

    const onError = () => {
      finish({ ok: false, code: "CONNECTION_CLOSED" });
    };

    const onClose = () => {
      if (done || ended) {
        return;
      }
      finish({ ok: false, code: "CONNECTION_CLOSED" });
    };

    req.on("data", onData);
    req.on("end", onEnd);
    req.on("error", onError);
    req.on("close", onClose);
  });
}

function readDeclaredContentLength(req: IncomingMessage): number | null {
  const header = req.headers["content-length"];
  const raw = Array.isArray(header) ? header[0] : header;
  if (typeof raw !== "string" || !/^\d+$/.test(raw)) {
    return null;
  }
  const value = Number(raw);
  return Number.isSafeInteger(value) ? value : Number.MAX_SAFE_INTEGER;
}

function closeRequestAfterResponse(req: IncomingMessage, res: ServerResponse): void {
  const once = (res as { once?: ServerResponse["once"] }).once;
  if (typeof once !== "function") {
    return;
  }
  res.setHeader("Connection", "close");
  once.call(res, "finish", () => {
    // Timeout/size failures must flush JSON first; destroying before finish drops
    // the HTTP response on real partial-body sockets.
    if (!req.destroyed) {
      req.destroy();
    }
  });
}

function readRpcRequestBody(body: unknown):
  | { ok: true; request: ParsedRequest }
  | {
      ok: false;
      message: string;
    } {
  if (!isRecord(body)) {
    return { ok: false, message: "request body must be an object" };
  }
  const rpcBody = body as RpcBody;
  if (typeof rpcBody.method !== "string" || rpcBody.method.trim().length === 0) {
    return { ok: false, message: "method must be a non-empty string" };
  }
  const id =
    typeof rpcBody.id === "string" && rpcBody.id.trim().length > 0
      ? rpcBody.id.trim()
      : randomUUID();
  return {
    ok: true,
    request: {
      id,
      method: rpcBody.method.trim(),
      ...(Object.hasOwn(rpcBody, "params") ? { params: rpcBody.params } : {}),
    },
  };
}

function methodNotAllowed(id: string, method: string): RpcResponse {
  return {
    id,
    ok: false,
    error: createError(
      ErrorCodes.INVALID_REQUEST,
      `admin HTTP RPC method is not supported: ${method}`,
    ),
  };
}

function commandsList(id: string): RpcResponse {
  return {
    id,
    ok: true,
    payload: {
      methods: listAdminHttpRpcAllowedMethods(),
    },
  };
}

async function dispatchAdminRpc(request: ParsedRequest): Promise<RpcResponse> {
  try {
    const response = await dispatchGatewayMethod(request.method, request.params);
    if (response.ok) {
      return {
        id: request.id,
        ok: true,
        payload: response.payload,
        ...(response.meta ? { meta: response.meta } : {}),
      };
    }
    return {
      id: request.id,
      ok: false,
      error:
        response.error ??
        createError(ErrorCodes.UNAVAILABLE, "gateway method failed before returning a response"),
      ...(response.meta ? { meta: response.meta } : {}),
    };
  } catch {
    return {
      id: request.id,
      ok: false,
      error: createError(
        ErrorCodes.UNAVAILABLE,
        "gateway method failed before returning a response",
      ),
    };
  }
}

/** Handle one gateway-authenticated Admin HTTP RPC request. */
export async function handleAdminHttpRpcRequest(
  req: IncomingMessage,
  res: ServerResponse,
): Promise<boolean> {
  if ((req.method ?? "GET").toUpperCase() !== "POST") {
    res.setHeader("Allow", "POST");
    sendError(res, 405, {
      type: "method_not_allowed",
      message: "Method Not Allowed",
    });
    return true;
  }

  const body = await readJsonBody(
    req,
    WEBHOOK_BODY_READ_DEFAULTS.postAuth.maxBytes,
    WEBHOOK_BODY_READ_DEFAULTS.postAuth.timeoutMs,
  );
  if (!body.ok) {
    if (body.closeAfterResponse) {
      closeRequestAfterResponse(req, res);
    }
    sendError(res, body.status, {
      type: "invalid_request",
      message: body.message,
    });
    return true;
  }

  const parsed = readRpcRequestBody(body.value);
  if (!parsed.ok) {
    sendError(res, 400, {
      type: "invalid_request",
      message: parsed.message,
    });
    return true;
  }

  if (!isAdminHttpRpcAllowedMethod(parsed.request.method)) {
    const response = methodNotAllowed(parsed.request.id, parsed.request.method);
    sendJson(res, rpcHttpStatus(response), response);
    return true;
  }

  if (parsed.request.method === "commands.list") {
    const response = commandsList(parsed.request.id);
    sendJson(res, 200, response);
    return true;
  }

  const response = await dispatchAdminRpc(parsed.request);
  sendJson(res, rpcHttpStatus(response), response);
  return true;
}