fix: bind hwpod node ops to target node
Prevent direct hwpod-node URL dispatch from executing plans on the wrong node and expose spawn diagnostics for cmd.run failures.
This commit is contained in:
@@ -134,7 +134,7 @@ test("cloud-api returns blocked payload when hwpod-node is not wired yet", async
|
||||
});
|
||||
|
||||
test("cloud-api forwards hwpod-node-ops to configured thin hwpod-node URL", async () => {
|
||||
const node = createHwpodNodeServer();
|
||||
const node = createHwpodNodeServer({ nodeId: "pc-host-1" });
|
||||
await listen(node);
|
||||
const server = createCloudApiServer({
|
||||
env: {
|
||||
@@ -163,6 +163,74 @@ test("cloud-api forwards hwpod-node-ops to configured thin hwpod-node URL", asyn
|
||||
}
|
||||
});
|
||||
|
||||
test("cloud-api blocks direct hwpod-node URL when target node id does not match", async () => {
|
||||
const node = createHwpodNodeServer({ nodeId: "g14-host-hwpod-node" });
|
||||
await listen(node);
|
||||
const server = createCloudApiServer({
|
||||
env: {
|
||||
PATH: process.env.PATH,
|
||||
HWPOD_NODE_ID: "g14-host-hwpod-node",
|
||||
HWLAB_HWPOD_NODE_OPS_URL: `${serverUrl(node)}/v1/hwpod-node-ops`
|
||||
}
|
||||
});
|
||||
await listen(server);
|
||||
try {
|
||||
const plan = samplePlan();
|
||||
plan.nodeId = "node-d601-f103-v2";
|
||||
plan.ops[0].args.workspacePath = "F:\\Work\\HWLAB-CASE-F103";
|
||||
const response = await fetch(`${serverUrl(server)}/v1/hwpod-node-ops`, {
|
||||
method: "POST",
|
||||
headers: { "content-type": "application/json" },
|
||||
body: JSON.stringify(plan)
|
||||
});
|
||||
const payload = await response.json();
|
||||
assert.equal(response.status, 200);
|
||||
assert.equal(payload.ok, false);
|
||||
assert.equal(payload.status, "blocked");
|
||||
assert.equal(payload.blocker.code, "hwpod_node_id_mismatch");
|
||||
assert.equal(payload.blocker.details.requestedNodeId, "node-d601-f103-v2");
|
||||
assert.equal(payload.blocker.details.targetNodeId, "g14-host-hwpod-node");
|
||||
assert.equal(payload.results[0].blocker.code, "hwpod_node_id_mismatch");
|
||||
} finally {
|
||||
await close(server);
|
||||
await close(node);
|
||||
}
|
||||
});
|
||||
|
||||
test("cloud-api blocks direct hwpod-node URL when node identity is not visible", async () => {
|
||||
const target = createCloudApiServer({
|
||||
hwpodNodeOpsHandler: async (plan: any) => ({
|
||||
ok: true,
|
||||
status: "completed",
|
||||
results: plan.ops.map((op: any) => ({ opId: op.opId, op: op.op, ok: true, status: "completed", output: { shouldNotRun: true } }))
|
||||
})
|
||||
});
|
||||
await listen(target);
|
||||
const server = createCloudApiServer({
|
||||
env: {
|
||||
PATH: process.env.PATH,
|
||||
HWLAB_HWPOD_NODE_OPS_URL: `${serverUrl(target)}/v1/hwpod-node-ops`
|
||||
}
|
||||
});
|
||||
await listen(server);
|
||||
try {
|
||||
const response = await fetch(`${serverUrl(server)}/v1/hwpod-node-ops`, {
|
||||
method: "POST",
|
||||
headers: { "content-type": "application/json" },
|
||||
body: JSON.stringify(samplePlan())
|
||||
});
|
||||
const payload = await response.json();
|
||||
assert.equal(response.status, 200);
|
||||
assert.equal(payload.ok, false);
|
||||
assert.equal(payload.status, "blocked");
|
||||
assert.equal(payload.blocker.code, "hwpod_node_identity_unverified");
|
||||
assert.equal(payload.results[0].output, undefined);
|
||||
} finally {
|
||||
await close(server);
|
||||
await close(target);
|
||||
}
|
||||
});
|
||||
|
||||
test("cloud-api dispatches hwpod-node-ops to outbound WebSocket hwpod-node by nodeId", async () => {
|
||||
const root = await mkdtemp(path.join(os.tmpdir(), "hwlab-hwpod-node-ws-"));
|
||||
const runtime = await createCloudApiBunServer({ host: "127.0.0.1", port: 0, env: { PATH: process.env.PATH } });
|
||||
|
||||
@@ -713,7 +713,8 @@ async function probeDiscoveredHwpodSpec(spec, { request, options, observedAt })
|
||||
async function dispatchHwpodNodeOpsPlan(plan, requestMeta, options, context = {}) {
|
||||
const hwpodNodeOpsUrl = normalizedHwpodNodeOpsUrl(options.env ?? process.env);
|
||||
const hwpodNodeWsRegistry = options.hwpodNodeWsRegistry;
|
||||
if (typeof options.hwpodNodeOpsHandler !== "function" && !hwpodNodeWsRegistry?.hasNode?.(plan.nodeId) && !hwpodNodeOpsUrl) {
|
||||
const hasWsNode = hwpodNodeWsRegistry?.hasNode?.(plan.nodeId);
|
||||
if (typeof options.hwpodNodeOpsHandler !== "function" && !hasWsNode && !hwpodNodeOpsUrl) {
|
||||
return {
|
||||
httpStatus: 200,
|
||||
payload: hwpodNodeOpsBlockedPayload(plan, requestMeta, "no outbound WebSocket hwpod-node is connected and HWLAB_HWPOD_NODE_OPS_URL is not configured; cloud-api is only validating the hwpod-node-ops contract")
|
||||
@@ -721,7 +722,7 @@ async function dispatchHwpodNodeOpsPlan(plan, requestMeta, options, context = {}
|
||||
}
|
||||
const handled = typeof options.hwpodNodeOpsHandler === "function"
|
||||
? await options.hwpodNodeOpsHandler(plan, { request: context.request, requestMeta, env: options.env ?? process.env })
|
||||
: hwpodNodeWsRegistry?.hasNode?.(plan.nodeId)
|
||||
: hasWsNode
|
||||
? await hwpodNodeWsRegistry.dispatch(plan, requestMeta, { timeoutMs: parsePositiveInteger(options.env?.HWLAB_HWPOD_NODE_OPS_TIMEOUT_MS, 30000) })
|
||||
: await forwardHwpodNodeOpsPlan(hwpodNodeOpsUrl, plan, requestMeta, options);
|
||||
const results = Array.isArray(handled?.results) ? handled.results : [];
|
||||
@@ -742,6 +743,24 @@ async function dispatchHwpodNodeOpsPlan(plan, requestMeta, options, context = {}
|
||||
}
|
||||
|
||||
async function forwardHwpodNodeOpsPlan(targetUrl, plan, requestMeta, options) {
|
||||
const target = await describeDirectHwpodNode(targetUrl, options);
|
||||
if (!target.ok || !target.nodeId) {
|
||||
return directHwpodNodeBlocked(plan, "hwpod_node_identity_unverified", "configured hwpod-node URL did not expose a verifiable nodeId", {
|
||||
requestedNodeId: plan.nodeId,
|
||||
targetUrl: redactNodeOpsUrl(targetUrl),
|
||||
dispatchMode: "direct-url",
|
||||
targetStatus: target.status ?? null,
|
||||
error: target.error ?? null
|
||||
});
|
||||
}
|
||||
if (target.ok && target.nodeId && target.nodeId !== plan.nodeId) {
|
||||
return directHwpodNodeBlocked(plan, "hwpod_node_id_mismatch", `configured hwpod-node URL is ${target.nodeId}, but plan targets ${plan.nodeId}`, {
|
||||
requestedNodeId: plan.nodeId,
|
||||
targetNodeId: target.nodeId,
|
||||
targetUrl: redactNodeOpsUrl(targetUrl),
|
||||
dispatchMode: "direct-url"
|
||||
});
|
||||
}
|
||||
const response = await fetch(targetUrl, {
|
||||
method: "POST",
|
||||
headers: {
|
||||
@@ -777,6 +796,42 @@ async function forwardHwpodNodeOpsPlan(targetUrl, plan, requestMeta, options) {
|
||||
};
|
||||
}
|
||||
|
||||
function directHwpodNodeBlocked(plan, code, summary, details) {
|
||||
const blocker = { code, layer: "hwpod-node", retryable: true, summary, details };
|
||||
return {
|
||||
ok: false,
|
||||
status: "blocked",
|
||||
httpStatus: 200,
|
||||
results: plan.ops.map((op) => ({ opId: op.opId, op: op.op, ok: false, status: "blocked", blocker })),
|
||||
blocker
|
||||
};
|
||||
}
|
||||
|
||||
async function describeDirectHwpodNode(targetUrl, options) {
|
||||
try {
|
||||
const response = await fetch(targetUrl, {
|
||||
method: "GET",
|
||||
signal: AbortSignal.timeout(parsePositiveInteger(options.env?.HWLAB_HWPOD_NODE_OPS_TIMEOUT_MS, 30000))
|
||||
});
|
||||
const body = await response.json().catch(() => null);
|
||||
return { ok: response.ok && body && typeof body === "object", status: response.status, nodeId: safeOpaqueId(body?.nodeId), body };
|
||||
} catch (error) {
|
||||
return { ok: false, error: error instanceof Error ? error.message : String(error), nodeId: "" };
|
||||
}
|
||||
}
|
||||
|
||||
function redactNodeOpsUrl(value) {
|
||||
try {
|
||||
const parsed = new URL(value);
|
||||
parsed.username = "";
|
||||
parsed.password = "";
|
||||
parsed.search = parsed.search ? "?redacted=1" : "";
|
||||
return parsed.toString();
|
||||
} catch {
|
||||
return "<invalid-url>";
|
||||
}
|
||||
}
|
||||
|
||||
function normalizedHwpodNodeOpsUrl(env = process.env) {
|
||||
const direct = String(env.HWLAB_HWPOD_NODE_OPS_URL ?? "").trim();
|
||||
if (direct) return direct;
|
||||
|
||||
@@ -31,7 +31,7 @@ test("hwpod-node executes minimal workspace node ops", async () => {
|
||||
}
|
||||
});
|
||||
test("hwpod-node HTTP endpoint accepts hwpod-node-ops plans", async () => {
|
||||
const server = createHwpodNodeServer({ now: () => "2026-06-05T00:00:00.000Z" });
|
||||
const server = createHwpodNodeServer({ now: () => "2026-06-05T00:00:00.000Z", nodeId: "pc-host-1" });
|
||||
await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve));
|
||||
try {
|
||||
const response = await fetch(`http://127.0.0.1:${server.address().port}/v1/hwpod-node-ops`, {
|
||||
@@ -119,6 +119,49 @@ test("hwpod-node reports cmd.run non-zero exits as failed results", async () =>
|
||||
}
|
||||
});
|
||||
|
||||
test("hwpod-node refuses plans targeting another node id", async () => {
|
||||
const root = await mkdtemp(path.join(os.tmpdir(), "hwlab-hwpod-node-id-mismatch-"));
|
||||
try {
|
||||
const result = await executeHwpodNodeOpsPlan({
|
||||
contractVersion: "hwpod-node-ops-v1",
|
||||
planId: "hwpod_plan_node_mismatch",
|
||||
hwpodId: "hwpod-local",
|
||||
nodeId: "node-d601-f103-v2",
|
||||
ops: [{ opId: "op_cmd", op: "cmd.run", args: { workspacePath: root, command: "sh", argv: ["-c", "echo must-not-run"] } }]
|
||||
}, { now: () => "2026-06-05T00:00:00.000Z", nodeId: "g14-host-hwpod-node" });
|
||||
|
||||
assert.equal(result.ok, false);
|
||||
assert.equal(result.status, "blocked");
|
||||
assert.equal(result.localNodeId, "g14-host-hwpod-node");
|
||||
assert.equal(result.results[0].blocker.code, "hwpod_node_id_mismatch");
|
||||
assert.equal(result.results[0].blocker.details.requestedNodeId, "node-d601-f103-v2");
|
||||
assert.equal(result.results[0].blocker.details.localNodeId, "g14-host-hwpod-node");
|
||||
} finally {
|
||||
await rm(root, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
test("hwpod-node exposes command spawn diagnostics when executable is missing", async () => {
|
||||
const root = await mkdtemp(path.join(os.tmpdir(), "hwlab-hwpod-node-spawn-missing-"));
|
||||
try {
|
||||
const result = await executeHwpodNodeOpsPlan({
|
||||
contractVersion: "hwpod-node-ops-v1",
|
||||
planId: "hwpod_plan_spawn_missing",
|
||||
hwpodId: "hwpod-local",
|
||||
nodeId: "pc-host-1",
|
||||
ops: [{ opId: "op_cmd", op: "cmd.run", args: { workspacePath: root, command: "hwlab-command-that-does-not-exist-1018", argv: ["--version"] } }]
|
||||
}, { now: () => "2026-06-05T00:00:00.000Z", nodeId: "pc-host-1" });
|
||||
|
||||
assert.equal(result.ok, false);
|
||||
assert.equal(result.results[0].status, "blocked");
|
||||
assert.equal(result.results[0].blocker.code, "hwpod_node_command_spawn_failed");
|
||||
assert.equal(result.results[0].blocker.details.output.commandResolution.requested, "hwlab-command-that-does-not-exist-1018");
|
||||
assert.equal(result.results[0].blocker.details.output.spawnDiagnostics.platform, process.platform);
|
||||
} finally {
|
||||
await rm(root, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
test("hwpod-node includes structured UART diagnostics when platform blocks serial access", async () => {
|
||||
const root = await mkdtemp(path.join(os.tmpdir(), "hwlab-hwpod-node-uart-"));
|
||||
try {
|
||||
|
||||
@@ -78,7 +78,7 @@ export function connectHwpodNodeWs(config: any, options: any = {}) {
|
||||
const reconnectBaseMs = Math.max(250, numberValue(config.reconnectBaseMs) ?? 1000);
|
||||
const reconnectMaxMs = Math.max(reconnectBaseMs, numberValue(config.reconnectMaxMs) ?? 30000);
|
||||
const reconnect = config.reconnect !== false;
|
||||
const executePlan = options.executePlan ?? ((plan: any) => executeHwpodNodeOpsPlan(plan, { now }));
|
||||
const executePlan = options.executePlan ?? ((plan: any) => executeHwpodNodeOpsPlan(plan, { now, nodeId }));
|
||||
let socket: WebSocket | null = null;
|
||||
let heartbeatTimer: ReturnType<typeof setInterval> | null = null;
|
||||
let reconnectTimer: ReturnType<typeof setTimeout> | null = null;
|
||||
@@ -202,6 +202,7 @@ function appendToken(urlValue: string, token: string) {
|
||||
|
||||
export function createHwpodNodeServer(options: any = {}) {
|
||||
const now = options.now ?? (() => new Date().toISOString());
|
||||
const localNodeId = localHwpodNodeId(options);
|
||||
return createServer(async (request, response) => {
|
||||
try {
|
||||
const url = new URL(request.url || "/", "http://hwpod-node.local");
|
||||
@@ -215,14 +216,14 @@ export function createHwpodNodeServer(options: any = {}) {
|
||||
nodeRole: "single-host-hwpod-node",
|
||||
acceptedInput: "hwpod-node-ops",
|
||||
specAuthority: "none",
|
||||
nodeId: process.env.HWPOD_NODE_ID || os.hostname(),
|
||||
nodeId: localNodeId,
|
||||
observedAt: now()
|
||||
});
|
||||
}
|
||||
if (request.method === "POST" && url.pathname === "/v1/hwpod-node-ops") {
|
||||
const body = await readBody(request, options.bodyLimitBytes);
|
||||
const plan = body ? JSON.parse(body) : {};
|
||||
return sendJson(response, 200, await executeHwpodNodeOpsPlan(plan, { now }));
|
||||
return sendJson(response, 200, await executeHwpodNodeOpsPlan(plan, { now, nodeId: localNodeId }));
|
||||
}
|
||||
return sendJson(response, 404, { ok: false, error: { code: "not_found", message: "hwpod-node route not found" } });
|
||||
} catch (error) {
|
||||
@@ -236,6 +237,17 @@ export async function executeHwpodNodeOpsPlan(plan: any, options: any = {}) {
|
||||
if (!plan || typeof plan !== "object" || Array.isArray(plan)) throw cliError("invalid_hwpod_node_ops_plan", "plan must be a JSON object");
|
||||
if (plan.contractVersion !== HWPOD_NODE_OPS_CONTRACT_VERSION) throw cliError("invalid_hwpod_node_ops_contract", `contractVersion must be ${HWPOD_NODE_OPS_CONTRACT_VERSION}`);
|
||||
if (!Array.isArray(plan.ops) || plan.ops.length === 0) throw cliError("invalid_hwpod_node_ops", "ops must be a non-empty array");
|
||||
const localNodeId = localHwpodNodeId(options);
|
||||
const requestedNodeId = text(plan.nodeId);
|
||||
const enforceLocalNodeId = Boolean(text(options.nodeId) || text(process.env.HWPOD_NODE_ID));
|
||||
if (enforceLocalNodeId && requestedNodeId && requestedNodeId !== localNodeId) {
|
||||
return hwpodNodeMismatchPayload(plan, {
|
||||
localNodeId,
|
||||
requestedNodeId,
|
||||
now,
|
||||
summary: `hwpod-node ${localNodeId} cannot execute plan for ${requestedNodeId}`
|
||||
});
|
||||
}
|
||||
const results = [];
|
||||
for (const op of plan.ops) {
|
||||
results.push(await executeOp(op, { plan, now }));
|
||||
@@ -252,12 +264,58 @@ export async function executeHwpodNodeOpsPlan(plan: any, options: any = {}) {
|
||||
specAuthority: "none",
|
||||
planId: plan.planId ?? null,
|
||||
hwpodId: plan.hwpodId ?? null,
|
||||
nodeId: plan.nodeId ?? process.env.HWPOD_NODE_ID ?? os.hostname(),
|
||||
nodeId: requestedNodeId || localNodeId,
|
||||
localNodeId,
|
||||
results,
|
||||
observedAt: now()
|
||||
};
|
||||
}
|
||||
|
||||
function localHwpodNodeId(options: any = {}) {
|
||||
return text(options.nodeId) || text(process.env.HWPOD_NODE_ID) || os.hostname();
|
||||
}
|
||||
|
||||
function hwpodNodeMismatchPayload(plan: any, input: { localNodeId: string; requestedNodeId: string; now: () => string; summary: string }) {
|
||||
const ops = Array.isArray(plan.ops) ? plan.ops : [];
|
||||
const blocker = {
|
||||
code: "hwpod_node_id_mismatch",
|
||||
layer: "hwpod-node",
|
||||
retryable: true,
|
||||
summary: input.summary,
|
||||
details: {
|
||||
localNodeId: input.localNodeId,
|
||||
requestedNodeId: input.requestedNodeId,
|
||||
platform: process.platform,
|
||||
hostname: os.hostname(),
|
||||
cwd: process.cwd(),
|
||||
nodeVersion: NODE_VERSION
|
||||
}
|
||||
};
|
||||
return {
|
||||
ok: false,
|
||||
status: "blocked",
|
||||
contractVersion: HWPOD_NODE_OPS_CONTRACT_VERSION,
|
||||
serviceId: "hwpod-node",
|
||||
nodeVersion: NODE_VERSION,
|
||||
nodeRole: "single-host-hwpod-node",
|
||||
acceptedInput: "hwpod-node-ops",
|
||||
specAuthority: "none",
|
||||
planId: plan.planId ?? null,
|
||||
hwpodId: plan.hwpodId ?? null,
|
||||
nodeId: input.requestedNodeId,
|
||||
localNodeId: input.localNodeId,
|
||||
results: ops.map((op: any, index: number) => ({
|
||||
opId: text(op?.opId) || `op_${index + 1}`,
|
||||
op: text(op?.op) || "unknown",
|
||||
ok: false,
|
||||
status: "blocked",
|
||||
blocker
|
||||
})),
|
||||
blocker,
|
||||
observedAt: input.now()
|
||||
};
|
||||
}
|
||||
|
||||
async function executeOp(op: any, context: any) {
|
||||
const opId = text(op?.opId) || "op_unknown";
|
||||
const name = text(op?.op);
|
||||
@@ -275,6 +333,7 @@ async function executeOp(op: any, context: any) {
|
||||
if (name === "workspace.insert-after") return opOk(opId, name, await workspaceInsertAfter(args));
|
||||
if (name === "cmd.run") {
|
||||
const output = await cmdRun(args);
|
||||
if (output.blockerCode) return opBlocked(opId, name, output.blockerCode, output.stderr || "cmd.run failed before process start", { output });
|
||||
return output.ok ? opOk(opId, name, output) : opFailed(opId, name, output, "hwpod_node_command_failed", `cmd.run node-side command exited with ${output.exitCode}`);
|
||||
}
|
||||
if (["debug.build", "debug.download", "debug.reset"].includes(name)) {
|
||||
@@ -790,7 +849,20 @@ async function spawnShellOutput(command: string, { cwd, timeoutMs }: any) {
|
||||
async function spawnOutput(command: string[], { cwd, stdinText = "", timeoutMs }: any) {
|
||||
const resolution = await resolveHwpodNodeCommand(command[0]);
|
||||
const resolvedCommand = [resolution.command, ...command.slice(1)];
|
||||
const proc = Bun.spawn(resolvedCommand, { cwd, stdin: stdinText ? "pipe" : "ignore", stdout: "pipe", stderr: "pipe" });
|
||||
let proc;
|
||||
try {
|
||||
proc = Bun.spawn(resolvedCommand, { cwd, stdin: stdinText ? "pipe" : "ignore", stdout: "pipe", stderr: "pipe" });
|
||||
} catch (error) {
|
||||
return {
|
||||
exitCode: null,
|
||||
stdout: "",
|
||||
stderr: error instanceof Error ? error.message : String(error),
|
||||
ok: false,
|
||||
blockerCode: "hwpod_node_command_spawn_failed",
|
||||
commandResolution: resolution,
|
||||
spawnDiagnostics: commandSpawnDiagnostics(command, resolvedCommand, cwd, resolution, error)
|
||||
};
|
||||
}
|
||||
if (stdinText && proc.stdin) {
|
||||
proc.stdin.write(stdinText);
|
||||
proc.stdin.end();
|
||||
@@ -804,6 +876,24 @@ async function spawnOutput(command: string[], { cwd, stdinText = "", timeoutMs }
|
||||
}
|
||||
}
|
||||
|
||||
function commandSpawnDiagnostics(command: string[], resolvedCommand: string[], cwd: string, resolution: any, error: unknown) {
|
||||
return {
|
||||
requestedCommand: command,
|
||||
resolvedCommand,
|
||||
commandResolution: resolution,
|
||||
cwd,
|
||||
platform: process.platform,
|
||||
hostname: os.hostname(),
|
||||
nodeVersion: NODE_VERSION,
|
||||
pathPreview: pathPreview(process.env.PATH),
|
||||
error: error instanceof Error ? { name: error.name, message: error.message, code: (error as any).code ?? null } : { message: String(error) }
|
||||
};
|
||||
}
|
||||
|
||||
function pathPreview(value: unknown) {
|
||||
return text(value).split(path.delimiter).filter(Boolean).slice(0, 12);
|
||||
}
|
||||
|
||||
export async function resolveHwpodNodeCommand(command: string, options: any = {}) {
|
||||
const requested = requiredText(command, "command");
|
||||
const platform = text(options.platform) || process.platform;
|
||||
|
||||
Reference in New Issue
Block a user