mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-25 02:26:12 +02:00
fix(computers): allocate listeners outside guest ephemeral ports
Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
66028796f3
commit
cfa2b81539
6 files changed
+112
-4
No files matched your search
@@ -16,7 +16,7 @@ import { join } from "node:path";
|
||||
import { spawn, spawnSync } from "node:child_process";
|
||||
import { remoteProgram } from "./remote-program.js";
|
||||
import { afterEach, describe, it, expect, vi } from "vitest";
|
||||
import { boatBackend, createBoatTransportAdmission, desktopReadinessProgram, processProgram } from "./boat.js";
|
||||
import { boatBackend, createBoatTransportAdmission, desktopReadinessProgram, processProgram, runnerPortInventoryProgram } from "./boat.js";
|
||||
import { buildGitAuthInvocation } from "../../../services/git-credentials.js";
|
||||
import type { ComputerRecord } from "../domain/ledger.js";
|
||||
const sshFactory = vi.hoisted(() => vi.fn());
|
||||
@@ -44,6 +44,22 @@ const json = (value: unknown, status = 200, headers?: Record<string, string>) =>
|
||||
headers: { "content-type": "application/json", ...headers },
|
||||
});
|
||||
describe("Boat transport validation", () => {
|
||||
it("reads outgoing and listening TCP ports across IPv4 and IPv6", () => {
|
||||
const root = mkdtempSync(join(tmpdir(), "boat-port-inventory-"));
|
||||
try {
|
||||
writeFileSync(join(root, "range"), "32768 60999\n");
|
||||
writeFileSync(join(root, "tcp"), "header\n0: 00000000:3F00 00000000:0000 0A\n");
|
||||
writeFileSync(join(root, "tcp6"), "header\n0: 00000000000000000000000000000000:A87A 00000000:01BB 08\n");
|
||||
const program = runnerPortInventoryProgram
|
||||
.replace("/proc/sys/net/ipv4/ip_local_port_range", join(root, "range"))
|
||||
.replace("/proc/net/tcp6", join(root, "tcp6"))
|
||||
.replace("/proc/net/tcp", join(root, "tcp"));
|
||||
const result = spawnSync("python3", ["-c", program], { encoding: "utf8" });
|
||||
expect(result.status, result.stderr).toBe(0);
|
||||
expect(JSON.parse(result.stdout)).toEqual({ ephemeralStart: 32768, ephemeralEnd: 60999, occupied: [16128, 43130] });
|
||||
} finally { rmSync(root, { recursive: true, force: true }); }
|
||||
});
|
||||
|
||||
it("recognizes idle as an already running computer", async () => {
|
||||
const fetcher = vi.fn(async () =>
|
||||
json({
|
||||
|
||||
@@ -475,6 +475,15 @@ print('{}')
|
||||
});
|
||||
}
|
||||
},
|
||||
async runnerPorts(record) {
|
||||
const value = await execute(record, runnerPortInventoryProgram, {});
|
||||
if (!Number.isInteger(value.ephemeralStart) || !Number.isInteger(value.ephemeralEnd)
|
||||
|| value.ephemeralStart < 1 || value.ephemeralEnd > 65535
|
||||
|| value.ephemeralStart > value.ephemeralEnd || !Array.isArray(value.occupied)
|
||||
|| value.occupied.some((port: unknown) => !Number.isInteger(port) || Number(port) < 1 || Number(port) > 65535))
|
||||
throw new ComputerError("provider_error", "Computer returned invalid listener port availability");
|
||||
return value;
|
||||
},
|
||||
async advance(record, owner) {
|
||||
await execute(record, processProgram, { action: "advance", owner });
|
||||
},
|
||||
@@ -564,3 +573,17 @@ print('{}')
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
/** Read-only Linux socket inventory; all states matter, not only LISTEN. */
|
||||
export const runnerPortInventoryProgram = String.raw`
|
||||
import json
|
||||
with open('/proc/sys/net/ipv4/ip_local_port_range') as source:
|
||||
start,end=map(int,source.read().split())
|
||||
occupied=set()
|
||||
for path in ['/proc/net/tcp','/proc/net/tcp6']:
|
||||
with open(path) as source:
|
||||
for line in source.readlines()[1:]:
|
||||
port=int(line.split()[1].rsplit(':',1)[1],16)
|
||||
if port:occupied.add(port)
|
||||
print(json.dumps({'ephemeralStart':start,'ephemeralEnd':end,'occupied':sorted(occupied)}))
|
||||
`;
|
||||
@@ -4,6 +4,7 @@ import type {
|
||||
Ledger,
|
||||
Owner,
|
||||
ProcessClaim,
|
||||
RunnerPortAvailability,
|
||||
} from "../domain/ledger.js";
|
||||
export interface ComputerRepository {
|
||||
create(record: ComputerRecord): Promise<void>;
|
||||
@@ -31,6 +32,7 @@ export interface ComputerBackend {
|
||||
ready(record: ComputerRecord): Promise<void>;
|
||||
claim(record: ComputerRecord): Promise<void>;
|
||||
advance(record: ComputerRecord, owner: Owner): Promise<void>;
|
||||
runnerPorts(record: ComputerRecord): Promise<RunnerPortAvailability>;
|
||||
runner(record: ComputerRecord, options?: { control?: boolean }): Promise<CommandManagedRuntimeRunner>;
|
||||
launch(
|
||||
record: ComputerRecord,
|
||||
|
||||
@@ -50,6 +50,7 @@ function fixture() {
|
||||
}),
|
||||
claim: vi.fn(async () => {}),
|
||||
advance: vi.fn(async () => {}),
|
||||
runnerPorts: vi.fn(async () => ({ ephemeralStart: 32768, ephemeralEnd: 60999, occupied: [] })),
|
||||
runner: vi.fn(async () => ({ execute: vi.fn() })),
|
||||
launch: vi.fn(async (_record, owner) => ({
|
||||
...owner.process!,
|
||||
@@ -114,6 +115,43 @@ function fixture() {
|
||||
};
|
||||
}
|
||||
describe("computer ownership", () => {
|
||||
it("allocates outside guest ephemeral and occupied ports while reserving live owners", async () => {
|
||||
const f = fixture();
|
||||
await f.attach();
|
||||
vi.mocked(f.backend.runnerPorts).mockResolvedValue({
|
||||
ephemeralStart: 16127, ephemeralEnd: 16128, occupied: [16129, 43130],
|
||||
});
|
||||
const first = await f.admit();
|
||||
const second = await f.admit("other-agent", "other-session");
|
||||
expect(first.listenerPort).toBe(16130);
|
||||
expect(second.listenerPort).toBe(16131);
|
||||
expect(f.backend.advance).toHaveBeenLastCalledWith(expect.anything(), expect.objectContaining({ port: 16131 }));
|
||||
for (const port of [first.listenerPort, 43130]) {
|
||||
await expect(f.service.preview({ ...f.scope, owner: first.owner, port })).rejects.toMatchObject({ code: "forbidden" });
|
||||
}
|
||||
});
|
||||
|
||||
it("fails closed when the guest reserves the entire fixed listener range", async () => {
|
||||
const f = fixture();
|
||||
await f.attach();
|
||||
vi.mocked(f.backend.runnerPorts).mockResolvedValue({ ephemeralStart: 1024, ephemeralEnd: 65535, occupied: [] });
|
||||
await expect(f.admit()).rejects.toMatchObject({ code: "conflict", message: "Computer has no available runner ports" });
|
||||
expect(f.backend.advance).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("preserves historical warm listener identity without reallocating an occupied own port", async () => {
|
||||
const f = fixture();
|
||||
await f.attach();
|
||||
const first = await f.admit();
|
||||
await first.launch({ command: "runnerd" });
|
||||
await f.repository.update(f.scope, record => { record.ledger.owners.find(owner => owner.id === first.owner.ownerId)!.port = 43130; });
|
||||
await f.service.retainWarm({ ...f.scope, owner: first.owner, idleTimeoutMs: 60_000 });
|
||||
vi.mocked(f.backend.runnerPorts).mockClear();
|
||||
const next = await f.admit();
|
||||
expect(next.listenerPort).toBe(43130);
|
||||
expect(f.backend.runnerPorts).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("passes checkout credentials transiently without retaining them in computer state", async () => {
|
||||
const f = fixture();
|
||||
await f.attach();
|
||||
|
||||
@@ -7,6 +7,7 @@ import {
|
||||
expired,
|
||||
liveOwners,
|
||||
nextPort,
|
||||
isRunnerPort,
|
||||
segment,
|
||||
timeout,
|
||||
type ComputerRecord,
|
||||
@@ -455,6 +456,18 @@ finally:
|
||||
});
|
||||
await backend.ready(result.record);
|
||||
await backend.claim(result.record);
|
||||
if (result.owner.generation === 1 && !result.owner.process) {
|
||||
const availability = await backend.runnerPorts(result.record);
|
||||
const allocated = await repository.update(input, (record) => {
|
||||
const owner = exactOwner(record, ref(result.record, result.owner));
|
||||
if (owner.phase !== "starting")
|
||||
throw new ComputerError("conflict", "Computer admission was retired");
|
||||
owner.port = nextPort(record.ledger.owners.filter((other) => other.id !== owner.id), availability);
|
||||
return { record: structuredClone(record), owner: structuredClone(owner) };
|
||||
});
|
||||
result.record = allocated.record;
|
||||
result.owner = allocated.owner;
|
||||
}
|
||||
await backend.advance(result.record, result.owner);
|
||||
await backend.renew(result.record);
|
||||
if (input.probeId) {
|
||||
@@ -661,7 +674,7 @@ finally:
|
||||
expired(owner, now())
|
||||
)
|
||||
throw new ComputerError("conflict", "Preview owner is not live");
|
||||
if (input.port >= 43127 && input.port < 44127)
|
||||
if (isRunnerPort(input.port))
|
||||
throw new ComputerError(
|
||||
"forbidden",
|
||||
"Runner ports are not application previews",
|
||||
|
||||
@@ -87,12 +87,28 @@ export function exactOwner(record: ComputerRecord, ref: OwnerRef): Owner {
|
||||
throw new ComputerError("conflict", "Computer owner has been superseded");
|
||||
return owner;
|
||||
}
|
||||
export function nextPort(owners: Owner[]): number {
|
||||
export interface RunnerPortAvailability {
|
||||
ephemeralStart: number;
|
||||
ephemeralEnd: number;
|
||||
occupied: number[];
|
||||
}
|
||||
// Keep fixed listeners below the usual Linux ephemeral range, and still check
|
||||
// the actual guest policy: custom images can configure a different range.
|
||||
export const RUNNER_PORT_START = 16127;
|
||||
export const RUNNER_PORT_END = 17127;
|
||||
export function isRunnerPort(port: number): boolean {
|
||||
return (port >= RUNNER_PORT_START && port < RUNNER_PORT_END)
|
||||
|| (port >= 43127 && port < 44127); // Existing warm owners retain their port.
|
||||
}
|
||||
export function nextPort(owners: Owner[], availability?: RunnerPortAvailability): number {
|
||||
const occupied = new Set(
|
||||
owners.filter((o) => o.phase !== "retired").map((o) => o.port),
|
||||
);
|
||||
for (let port = 43127; port < 44127; port++)
|
||||
for (const port of availability?.occupied ?? []) occupied.add(port);
|
||||
for (let port = RUNNER_PORT_START; port < RUNNER_PORT_END; port++) {
|
||||
if (availability && port >= availability.ephemeralStart && port <= availability.ephemeralEnd) continue;
|
||||
if (!occupied.has(port)) return port;
|
||||
}
|
||||
throw new ComputerError("conflict", "Computer has no available runner ports");
|
||||
}
|
||||
export function liveOwners(ledger: Ledger): Owner[] {
|
||||
|
||||
Reference in new issue
Block a user