sscoble/pi-mono
1
0
Fork 0
mirror of https://github.com/badlogic/pi-mono.git synced 2026-11-05 23:37:25 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

fix: make session authentication transport-specific (#7551)

Browse source 
Set the initial protocol version to 1 and remove bearer authentication from framed messages. Require transports to establish authorized connections.
This commit is contained in:
Christian Klotz 2026-08-03 16:40:00 +03:00 • committed by GitHub
parent 0e633790c5
commit 305c014dcc
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
28 changed files with 89 additions and 140 deletions
Download patch file Download diff file Expand all files Collapse all files

7
packages/client/README.md
View file

@ -15,7 +15,7 @@ const transportFactory: ByteTransportFactory = async (handlers) => {
};
};
const client = new PiClient({ token: bearerToken, transportFactory });
const client = new PiClient({ transportFactory });
await client.connect();
const session = await client.createSession({ cwd: "/workspace" });
const unsubscribe = session.subscribe((snapshot) => render(snapshot));
@ -23,7 +23,7 @@ await session.prompt("Inspect this project");
unsubscribe();
```
Call `handlers.onData(chunk)` for inbound bytes, `handlers.onClose()` for an orderly terminal close, and `handlers.onError(error)` for transport failures. A factory must create a fresh transport for every connection attempt.
Call `handlers.onData(chunk)` for inbound bytes, `handlers.onClose()` for an orderly terminal close, and `handlers.onError(error)` for transport failures. A factory must create a fresh transport for every connection attempt and complete any transport-specific authentication before resolving. For example, a WebSocket factory can provide credentials in its upgrade request.
`PiClient` does not reconnect automatically. Call `reconnect()` after disconnection. One connection can attach several sessions. Requests are correlated by ID. Server snapshots and successful response snapshots are authoritative, while progress events do not mutate snapshot state optimistically. Read cached session summaries from `client.snapshot?.sessions`; call `listSessions()` to request a refreshed list from the server.
@ -37,7 +37,7 @@ Calling `dispose()` or `detach()` releases only that lease. A lease rejects comm
`PiClientOptions.maxFrameLength` bounds inbound and outbound CBOR payloads. Configure matching limits on the client and server. Transports should separately bound queued outbound bytes and preserve send order.
Treat peers as untrusted. Use a secure transport where required and protect the protocol bearer token.
Treat peers as untrusted. Use a secure transport with appropriate access controls and authenticate during transport establishment.
Subscriber exceptions are isolated from protocol state. Set `onListenerError` in `PiClientOptions` to report them to application logging or diagnostics.
@ -50,7 +50,6 @@ import { PiClient } from "@earendil-works/pi-client";
import { createUnixTransportFactory } from "@earendil-works/pi-client/unix";
const client = new PiClient({
token: bearerToken,
transportFactory: createUnixTransportFactory({
path: "/tmp/pi.sock",
}),

1
packages/client/src/client.ts
View file

@ -69,7 +69,6 @@ export class PiClient {
this.#options = options;
this.#state = new ClientState(options.onListenerError);
this.#connection = new Connection({
token: options.token,
transportFactory: options.transportFactory,
maxFrameLength: options.maxFrameLength,
onHandshake: (snapshot) => this.#state.applyServerSnapshot(snapshot),

6
packages/client/src/connection.ts
View file

@ -30,7 +30,6 @@ type ConnectionLifecycle =
} & ActiveConnection);
interface ConnectionOptions {
token: string;
transportFactory: ByteTransportFactory;
maxFrameLength?: number;
onHandshake(snapshot: ServerSnapshot): void;
@ -133,10 +132,7 @@ export class Connection {
this.#lifecycle = { ...lifecycle, transport };
try {
await transport.send(
encodeClientMessage(
{ type: "hello", version: PROTOCOL_VERSION, token: this.#options.token },
{ maxFrameLength: this.#maxFrameLength },
),
encodeClientMessage({ type: "hello", version: PROTOCOL_VERSION }, { maxFrameLength: this.#maxFrameLength }),
);
} catch (error) {
if (this.#isCurrent(id)) this.#failAndClose(toDisconnectedError(error));

2
packages/client/src/transport.ts
View file

@ -14,5 +14,5 @@ export interface ByteTransportHandlers {
onError(error: Error): void;
}
/** Creates a fresh connected transport for each PiClient connection attempt. Exactly one terminal handler is expected. */
/** Creates a fresh connected, authenticated transport. Exactly one terminal handler is expected. */
export type ByteTransportFactory = (handlers: ByteTransportHandlers) => ByteTransport | Promise<ByteTransport>;

1
packages/client/src/types.ts
View file

@ -12,7 +12,6 @@ export type Unsubscribe = () => void;
export type ListenerErrorHandler = (error: Error) => void;
export interface PiClientOptions {
token: string;
transportFactory: ByteTransportFactory;
maxFrameLength?: number;
/** Reports subscriber failures without allowing them to corrupt client state. */

23
packages/client/test/connection.test.ts
View file

@ -19,7 +19,7 @@ import {
} from "./support.ts";
describe("PiClient", () => {
test("sends a framed version and bearer token before accepting a fragmented server hello", async () => {
test("sends a framed version before accepting a fragmented server hello", async () => {
const server = new MemoryByteServer();
const received: ClientMessage[] = [];
server.onMessage((message) => {
@ -39,8 +39,7 @@ describe("PiClient", () => {
const client = createClient(server);
await expect(client.connect()).resolves.toEqual(baseServerSnapshot);
expect(received[0]).toEqual({ type: "hello", version: PROTOCOL_VERSION, token: "bearer-secret" });
expect(server.sentByClient[0]).toBeInstanceOf(Uint8Array);
expect(received[0]).toEqual({ type: "hello", version: PROTOCOL_VERSION });
expect(client.connectionState).toBe("connected");
});
@ -48,7 +47,6 @@ describe("PiClient", () => {
let closeCount = 0;
let sendCount = 0;
const client = new PiClient({
token: "bearer-secret",
transportFactory: (handlers) => {
handlers.onData(
encodeServerMessage({
@ -113,7 +111,6 @@ describe("PiClient", () => {
}
});
const client = new PiClient({
token: "bearer-secret",
transportFactory: (handlers) => server.connect(handlers),
onListenerError: (error) => listenerErrors.push(error),
});
@ -161,7 +158,6 @@ describe("PiClient", () => {
});
}
const client = new PiClient({
token: "bearer-secret",
transportFactory: (handlers) => (connection++ === 0 ? first : second).connect(handlers),
});
let reconnect: Promise<ServerSnapshot> | undefined;
@ -180,20 +176,20 @@ describe("PiClient", () => {
expect(first.clientCloseCount).toBe(1);
});
test("rejects a typed handshake authentication error", async () => {
test("rejects a typed handshake version error", async () => {
const server = new MemoryByteServer();
server.onMessage(() => {
server.send({
type: "hello_error",
error: { code: "auth", message: "Invalid token" },
error: { code: "version", message: "Unsupported protocol version" },
});
});
const client = createClient(server, "wrong");
const client = createClient(server);
await expect(client.connect()).rejects.toMatchObject({
name: "PiServerError",
code: "auth",
message: "Invalid token",
code: "version",
message: "Unsupported protocol version",
});
expect(client.connectionState).toBe("disconnected");
expect(server.clientCloseCount).toBe(1);
@ -217,7 +213,7 @@ describe("PiClient", () => {
}
const transportFactory: ByteTransportFactory = (handlers) =>
(connection++ === 0 ? first : second).connect(handlers);
const client = new PiClient({ token: "bearer-secret", transportFactory });
const client = new PiClient({ transportFactory });
const states: string[] = [];
client.onConnectionStateChange(({ state }) => states.push(state));
await client.connect();
@ -247,7 +243,6 @@ describe("PiClient", () => {
});
}
const client = new PiClient({
token: "bearer-secret",
transportFactory: (handlers) => (connection++ === 0 ? first : second).connect(handlers),
});
await client.connect();
@ -285,7 +280,6 @@ describe("PiClient", () => {
}
});
const client = new PiClient({
token: "bearer-secret",
maxFrameLength: 512,
transportFactory: (handlers) => server.connect(handlers),
});
@ -325,7 +319,6 @@ describe("PiClient", () => {
expect(
() =>
new PiClient({
token: "secret",
maxFrameLength: 0x1_0000_0000,
transportFactory: (handlers) => server.connect(handlers),
}),

4
packages/client/test/disposal.test.ts
View file

@ -1,3 +1,4 @@
import { PROTOCOL_VERSION } from "@earendil-works/pi-protocol";
import { describe, expect, test } from "vitest";
import { PiClient, PiClientDisposedError } from "../src/index.ts";
import { attachSession, baseServerSnapshot, connectClient, MemoryByteServer, sessionSnapshot } from "./support.ts";
@ -9,14 +10,13 @@ describe("PiClient disposal", () => {
if (message.type !== "hello") return;
server.send({
type: "hello",
version: 2,
version: PROTOCOL_VERSION,
connectionId: "connection-1",
snapshot: baseServerSnapshot,
});
});
const client = await PiClient.connect({
token: "secret",
transportFactory: (handlers) => server.connect(handlers),
});

7
packages/client/test/support.ts
View file

@ -104,15 +104,14 @@ export function sessionSnapshot(id: string, overrides: Partial<SessionSnapshot>
};
}
export function createClient(server: MemoryByteServer, token = "bearer-secret"): PiClient {
export function createClient(server: MemoryByteServer): PiClient {
return new PiClient({
token,
transportFactory: (handlers) => server.connect(handlers),
});
}
export async function connectClient(server: MemoryByteServer, token = "bearer-secret"): Promise<PiClient> {
const client = createClient(server, token);
export async function connectClient(server: MemoryByteServer): Promise<PiClient> {
const client = createClient(server);
server.onMessage((message) => {
if (message.type === "hello") {
server.send({

2
packages/client/test/unix.test.ts
View file

@ -78,7 +78,6 @@ describe.runIf(process.platform !== "win32")("Unix-domain sockets", () => {
});
await listen(server, socketPath);
const client = new PiClient({
token: "unix-secret",
transportFactory: createUnixTransportFactory({ path: socketPath }),
});
@ -192,7 +191,6 @@ describe.runIf(process.platform !== "win32")("Unix-domain sockets", () => {
});
await listen(server, socketPath);
const client = new PiClient({
token: "unix-secret",
transportFactory: createUnixTransportFactory({ path: socketPath }),
});

2
packages/coding-agent/test/client/support.ts
View file

@ -74,7 +74,7 @@ export function sessionSnapshot(id: string, overrides: Partial<SessionSnapshot>
}
export async function connectClient(server: MemoryServer): Promise<PiClient> {
const client = new PiClient({ token: "secret", transportFactory: (handlers) => server.connect(handlers) });
const client = new PiClient({ transportFactory: (handlers) => server.connect(handlers) });
server.onMessage((message) => {
if (message.type !== "hello") return;
server.send({

7
packages/protocol/README.md
View file

@ -2,12 +2,12 @@
Runtime-neutral schemas, types, CBOR encoding, and byte-stream framing for the experimental pi protocol.
Protocol version `2` uses binary messages with this wire layout:
Protocol version `1` uses binary messages with this wire layout:
1. A four-byte unsigned big-endian payload length.
2. One definite-length CBOR item containing the message.
The first client message is always `hello`, containing `PROTOCOL_VERSION` and a bearer token. Subsequent messages use correlated request/response envelopes and server event envelopes. Session and server snapshots are authoritative. Progress events are transient UI hints and must not be reduced into authoritative state.
The first client message is always `hello`, containing `PROTOCOL_VERSION`. Subsequent messages use correlated request/response envelopes and server event envelopes. Session and server snapshots are authoritative. Progress events are transient UI hints and must not be reduced into authoritative state. Transports complete authentication before protocol bytes are exchanged.
## Validated message API
@ -24,7 +24,6 @@ import {
const hello: ClientHello = {
type: "hello",
version: PROTOCOL_VERSION,
token: bearerToken,
};
transport.send(encodeClientMessage(hello));
@ -46,7 +45,7 @@ Every transport carries the same complete bytes: `[uint32-be CBOR length][CBOR p
This package does not bundle a transport. Consumers provide a byte-stream transport that preserves byte order and reports stream closure. Custom transports must handle arbitrary frame fragmentation and coalescing.
All transports are untrusted. Configure matching frame limits and enforce the authentication and access controls appropriate for the transport.
All transports are untrusted. Configure matching frame limits and enforce access controls appropriate for the transport before exposing a connection to the protocol. Unix sockets can use filesystem permissions, while network transports can authenticate during connection establishment.
## Encoding and framing

4
packages/protocol/src/schemas.ts
View file

@ -1,6 +1,6 @@
import Type, { type Static } from "typebox";
export const PROTOCOL_VERSION = 2 as const;
export const PROTOCOL_VERSION = 1 as const;
const IdSchema = Type.String({ minLength: 1 });
const TimestampSchema = Type.Integer({ minimum: 0 });
@ -264,7 +264,6 @@ export const ServerSnapshotSchema = StrictObject({
export type ServerSnapshot = Static<typeof ServerSnapshotSchema>;
export const ProtocolErrorCodeSchema = Type.Union([
Type.Literal("auth"),
Type.Literal("version"),
Type.Literal("busy"),
Type.Literal("session_locked"),
@ -381,7 +380,6 @@ export type ResultForCommand<TCommand extends Command> = TCommand["command"] ext
export const ClientHelloSchema = StrictObject({
type: Type.Literal("hello"),
version: Type.Integer({ minimum: 0 }),
token: Type.String({ minLength: 1 }),
});
export type ClientHello = Static<typeof ClientHelloSchema>;

52
packages/protocol/test/protocol.test.ts
View file

@ -31,7 +31,6 @@ const emptyServerSnapshot: ServerSnapshot = {
const clientHello: ClientHello = {
type: "hello",
version: PROTOCOL_VERSION,
token: "secret",
};
const serverHello: ServerHello = {
@ -53,24 +52,27 @@ function itemMessage(item: unknown, type: "item_updated" | "item_finished" = "it
}
describe("protocol validation", () => {
test("uses protocol version 2", () => {
expect(PROTOCOL_VERSION).toBe(2);
expect(isSupportedProtocolVersion(2)).toBe(true);
expect(isSupportedProtocolVersion(1)).toBe(false);
test("uses protocol version 1", () => {
expect(PROTOCOL_VERSION).toBe(1);
expect(isSupportedProtocolVersion(1)).toBe(true);
expect(isSupportedProtocolVersion(2)).toBe(false);
expect(isSupportedProtocolVersion(2.5)).toBe(false);
});
test.each([0, 1, PROTOCOL_VERSION])("accepts integer client hello version %s for negotiation", (version) => {
const message = { ...clientHello, version };
expect(parseClientMessage(message)).toEqual(message);
});
test.each([0, PROTOCOL_VERSION, PROTOCOL_VERSION + 1])(
"accepts integer client hello version %s for negotiation",
(version) => {
const message = { ...clientHello, version };
expect(parseClientMessage(message)).toEqual(message);
},
);
test.each([
{ type: "hello", version: "2", token: "secret" },
{ type: "hello", version: 2.5, token: "secret" },
{ type: "hello", version: 2, token: "" },
{ type: "hello", version: 2, token: "secret", extra: true },
])("rejects an invalid strict handshake: $version", (message) => {
["string version", { type: "hello", version: String(PROTOCOL_VERSION) }],
["fractional version", { type: "hello", version: PROTOCOL_VERSION + 0.5 }],
["credential field", { type: "hello", version: PROTOCOL_VERSION, token: "secret" }],
["unknown field", { type: "hello", version: PROTOCOL_VERSION, extra: true }],
] as const)("rejects a handshake with %s", (_label, message) => {
expect(() => parseClientMessage(message)).toThrow(ProtocolValidationError);
});
@ -99,7 +101,13 @@ describe("protocol validation", () => {
});
test.each([
{ type: "hello", version: 1, connectionId: "connection-1", snapshot: emptyServerSnapshot },
{
type: "hello",
version: PROTOCOL_VERSION + 1,
connectionId: "connection-1",
snapshot: emptyServerSnapshot,
},
{ type: "hello_error", error: { code: "auth", message: "Authentication failed" } },
{ type: "response", id: "request-1", ok: true, result: { command: "unknown" } },
{ type: "event", event: { type: "session_removed", sessionId: 42 } },
])("rejects invalid server messages", (wire) => {
@ -257,7 +265,11 @@ describe("protocol validation", () => {
test("validation errors do not retain rejected payloads", () => {
let thrown: unknown;
try {
parseClientMessage({ type: "hello", version: "2", token: "x".repeat(2_000_000) });
parseClientMessage({
type: "hello",
version: String(PROTOCOL_VERSION),
extra: "x".repeat(2_000_000),
});
} catch (error) {
thrown = error;
}
@ -284,7 +296,9 @@ describe("validated framed protocol APIs", () => {
});
test("validates messages before encoding", () => {
expect(() => encodeClientMessage({ type: "hello", version: 2, token: "" })).toThrow(ProtocolValidationError);
expect(() => encodeClientMessage({ type: "hello", version: PROTOCOL_VERSION + 0.5 })).toThrow(
ProtocolValidationError,
);
});
test("omits explicit undefined optional properties on the wire", () => {
@ -324,7 +338,7 @@ describe("validated framed protocol APIs", () => {
test("incrementally decodes server messages", () => {
const errorMessage: ServerMessage = {
type: "hello_error",
error: { code: "auth", message: "Invalid token" },
error: { code: "version", message: "Unsupported protocol version" },
};
const decoder = new ServerMessageDecoder();
expect(decoder.push(encodeServerMessage(errorMessage))).toEqual([errorMessage]);
@ -334,7 +348,7 @@ describe("validated framed protocol APIs", () => {
test.each([
["empty CBOR payload", encodeFrame(new Uint8Array())],
["malformed CBOR", encodeFrame(new Uint8Array([0xff]))],
["schema-invalid CBOR", encodeFrame(encodeCbor({ type: "hello", version: 2, token: "", extra: true }))],
["schema-invalid CBOR", encodeFrame(encodeCbor({ type: "hello", version: PROTOCOL_VERSION, extra: true }))],
] as const)("rejects invalid framed client input: %s", (_label, wire) => {
const decoder = new ClientMessageDecoder();
expect(() => decoder.push(wire)).toThrow(ProtocolValidationError);

3
packages/server/README.md
View file

@ -34,13 +34,12 @@ const backend: PiSessionBackend = {
};
const server = createUnixServer(backend, {
token: process.env.PI_SERVER_TOKEN!,
path: "/tmp/pi/server.sock",
});
await server.start();
```
`PiServer` composes transport listeners through the `PiServerListener` interface. The Unix submodule exports the `createUnixListener()` building block and `createUnixServer()` preset, keeping the common case concise without coupling the primary server to Unix sockets. The listener uses authenticated, length-prefixed CBOR messages from `@earendil-works/pi-protocol`. It does not yet replace the legacy JSONL IPC control plane, child-process supervisor, standalone `server` CLI, or Radius presence integration.
`PiServer` composes transport listeners through the `PiServerListener` interface. Each listener must complete any transport-specific authentication and authorization before passing a connection to `PiServer`. For example, a WebSocket listener can validate credentials during the HTTP upgrade, while the Unix listener relies on socket filesystem permissions. The Unix submodule exports the `createUnixListener()` building block and `createUnixServer()` preset, keeping the common case concise without coupling the primary server to Unix sockets. The listener uses length-prefixed CBOR messages from `@earendil-works/pi-protocol`. It does not yet replace the legacy JSONL IPC control plane, child-process supervisor, standalone `server` CLI, or Radius presence integration.
## Transport testing

2
packages/server/src/connection.ts
View file

@ -2,7 +2,7 @@ import type { ClientMessageDecoder } from "@earendil-works/pi-protocol";
import type { MaybePromise } from "./types.ts";
/** A connected, ordered byte sink used by PiServer's transport-neutral domain core. */
/** An established, authorized ordered byte connection. */
export interface ByteConnection {
readonly closed: boolean;
send(chunk: Uint8Array): Promise<void>;

3
packages/server/src/listener.ts
View file

@ -1,9 +1,10 @@
import type { ByteConnectionAcceptor } from "./connection.ts";
/** A transport listener that supplies ordered byte connections to PiServer. */
/** Supplies established byte connections after any required transport authentication. */
export interface PiServerListener {
/** Human-readable bound address after startup, when the transport has one. */
readonly address?: string;
/** Starts listening and passes authorized connections to accept. */
start(accept: ByteConnectionAcceptor): Promise<void>;
close(): Promise<void>;
}

17
packages/server/src/server.ts
View file

@ -1,4 +1,4 @@
import { createHash, randomUUID, timingSafeEqual } from "node:crypto";
import { randomUUID } from "node:crypto";
import {
type ClientHello,
type ClientMessage,
@ -31,15 +31,10 @@ const DEFAULT_HANDSHAKE_TIMEOUT_MS = 5_000;
const MAX_UINT32 = 0xffff_ffff;
const MAX_TIMER_DELAY_MS = 2_147_483_647;
function tokenDigest(token: string): Buffer {
return createHash("sha256").update(token, "utf8").digest();
}
export class PiServer {
readonly id: string;
private readonly listeners: readonly PiServerListener[];
private readonly expectedTokenDigest: Buffer;
private readonly maxFrameLength: number;
private readonly handshakeTimeoutMs: number;
private readonly onError: ((error: Error) => void) | undefined;
@ -55,7 +50,6 @@ export class PiServer {
const resolved = resolveOptions(options);
this.listeners = options.listeners;
this.id = options.serverId ?? randomUUID();
this.expectedTokenDigest = tokenDigest(options.token);
this.maxFrameLength = resolved.maxFrameLength;
this.handshakeTimeoutMs = resolved.handshakeTimeoutMs;
this.onError = options.onError;
@ -220,10 +214,6 @@ export class PiServer {
}
private async finishHandshake(state: ConnectionState, hello: ClientHello): Promise<void> {
if (!this.authenticate(hello)) {
await this.failProtocol(state, { code: "auth", message: "Authentication failed" });
return;
}
if (!isSupportedProtocolVersion(hello.version)) {
await this.failProtocol(state, {
code: "version",
@ -254,10 +244,6 @@ export class PiServer {
}
}
private authenticate(hello: ClientHello): boolean {
return timingSafeEqual(tokenDigest(hello.token), this.expectedTokenDigest);
}
private async handleRequest(state: ConnectionState, envelope: RequestEnvelope): Promise<void> {
try {
const result = await this.sessions.executeCommand(state, envelope.request);
@ -381,7 +367,6 @@ export class PiServer {
function resolveOptions(options: PiServerOptions): { maxFrameLength: number; handshakeTimeoutMs: number } {
if (!Array.isArray(options.listeners)) throw new TypeError("PiServer listeners must be an array");
if (!options.token) throw new TypeError("PiServer token must not be empty");
if (options.serverId === "") throw new TypeError("PiServer serverId must not be empty");
const maxFrameLength = options.maxFrameLength ?? DEFAULT_MAX_FRAME_LENGTH;
if (!Number.isSafeInteger(maxFrameLength) || maxFrameLength <= 0 || maxFrameLength > MAX_UINT32) {

1
packages/server/src/testing/backend.ts
View file

@ -16,7 +16,6 @@ import type {
PromptInput,
} from "../types.ts";
export const TEST_TOKEN = "server-conformance-token";
export const TEST_MODEL: ModelMetadata = {
provider: "test",
id: "small",

6
packages/server/src/testing/client.ts
View file

@ -9,7 +9,7 @@ import {
type ServerMessage,
ServerMessageDecoder,
} from "@earendil-works/pi-protocol";
import { Deferred, TEST_TOKEN } from "./backend.ts";
import { Deferred } from "./backend.ts";
interface MessageWaiter {
predicate: (message: ServerMessage) => boolean;
@ -40,9 +40,9 @@ export class ProtocolTestClient {
return this.closedValue;
}
hello(token = TEST_TOKEN, version: number = PROTOCOL_VERSION): Promise<ServerMessage> {
hello(version: number = PROTOCOL_VERSION): Promise<ServerMessage> {
const response = this.next((message) => message.type === "hello" || message.type === "hello_error");
void this.sendMessage({ type: "hello", token, version });
void this.sendMessage({ type: "hello", version });
return response;
}

2
packages/server/src/testing/index.ts
View file

@ -1,4 +1,4 @@
export { Deferred, TEST_MODEL, TEST_TOKEN, TestSessionBackend, TestSessionRuntime } from "./backend.ts";
export { Deferred, TEST_MODEL, TestSessionBackend, TestSessionRuntime } from "./backend.ts";
export type { WireChannel } from "./client.ts";
export { connectUnixTestClient, ProtocolTestClient } from "./client.ts";
export type { TestServer, TestServerOptions } from "./server.ts";

6
packages/server/src/testing/server.ts
View file

@ -1,9 +1,8 @@
import { PiServer } from "../server.ts";
import type { PiServerOptions, PiSessionBackend } from "../types.ts";
import { TEST_TOKEN, TestSessionBackend } from "./backend.ts";
import { TestSessionBackend } from "./backend.ts";
export interface TestServerOptions extends Omit<PiServerOptions, "token"> {
token?: string;
export interface TestServerOptions extends PiServerOptions {
backend?: PiSessionBackend;
}
@ -17,7 +16,6 @@ export function createTestServer(options: TestServerOptions): TestServer {
const backend = options.backend ?? new TestSessionBackend();
return {
server: new PiServer(backend, {
token: options.token ?? TEST_TOKEN,
listeners: options.listeners,
maxFrameLength: options.maxFrameLength,
handshakeTimeoutMs: options.handshakeTimeoutMs,

1
packages/server/src/transports/unix/preset.ts
View file

@ -14,7 +14,6 @@ export function createUnixServer(backend: PiSessionBackend, options: UnixServerO
onError: options.onError,
});
return new PiServer(backend, {
token: options.token,
listeners: [listener],
maxFrameLength: options.maxFrameLength,
handshakeTimeoutMs: options.handshakeTimeoutMs,

1
packages/server/src/types.ts
View file

@ -12,7 +12,6 @@ import type { PiServerError } from "./errors.ts";
import type { PiServerListener } from "./listener.ts";
export interface PiServerOptions {
token: string;
listeners: readonly PiServerListener[];
maxFrameLength?: number;
handshakeTimeoutMs?: number;

28
packages/server/test/conformance.test.ts
View file

@ -4,13 +4,7 @@ import { join } from "node:path";
import { encodeClientMessage, encodeFrame, PROTOCOL_VERSION } from "@earendil-works/pi-protocol";
import { afterEach, describe, expect, test } from "vitest";
import { type PiServer, PiServerError } from "../src/index.ts";
import {
connectUnixTestClient,
Deferred,
type ProtocolTestClient,
TEST_TOKEN,
TestSessionBackend,
} from "../src/testing/index.ts";
import { connectUnixTestClient, Deferred, type ProtocolTestClient, TestSessionBackend } from "../src/testing/index.ts";
import { createUnixServer, type UnixServerOptions } from "../src/transports/unix/index.ts";
const servers = new Set<PiServer>();
@ -24,7 +18,6 @@ async function startServer(
const directory = await mkdtemp(join(tmpdir(), "pis-"));
tempDirectories.add(directory);
const server = createUnixServer(backend, {
token: TEST_TOKEN,
path: join(directory, "server.sock"),
...overrides,
});
@ -53,22 +46,15 @@ describe("Unix transport conformance", () => {
const { server } = await startServer();
const client = await connect(server);
const response = client.next((message) => message.type === "hello");
await client.sendFragmentedMessage({ type: "hello", version: PROTOCOL_VERSION, token: TEST_TOKEN }, 2);
await client.sendFragmentedMessage({ type: "hello", version: PROTOCOL_VERSION }, 2);
expect(await response).toMatchObject({ type: "hello", version: PROTOCOL_VERSION });
});
test("enforces authentication, version, and exactly one first-message hello", async () => {
test("enforces version and exactly one first-message hello", async () => {
const { server } = await startServer();
const badToken = await connect(server);
expect(await badToken.hello("wrong-token")).toMatchObject({
type: "hello_error",
error: { code: "auth" },
});
await badToken.waitForClose();
const badVersion = await connect(server);
expect(await badVersion.hello(TEST_TOKEN, PROTOCOL_VERSION + 1)).toMatchObject({
expect(await badVersion.hello(PROTOCOL_VERSION + 1)).toMatchObject({
type: "hello_error",
error: { code: "version" },
});
@ -83,7 +69,7 @@ describe("Unix transport conformance", () => {
const duplicate = await connect(server);
expect(await duplicate.hello()).toMatchObject({ type: "hello" });
const duplicateError = duplicate.next((message) => message.type === "hello_error");
await duplicate.sendMessage({ type: "hello", version: PROTOCOL_VERSION, token: TEST_TOKEN });
await duplicate.sendMessage({ type: "hello", version: PROTOCOL_VERSION });
expect(await duplicateError).toMatchObject({ type: "hello_error", error: { code: "invalid_request" } });
await duplicate.waitForClose();
});
@ -102,7 +88,7 @@ describe("Unix transport conformance", () => {
const delay = backend.delayNextList();
const { server } = await startServer(backend, { handshakeTimeoutMs: 20 });
const client = await connect(server);
await client.sendMessage({ type: "hello", version: PROTOCOL_VERSION, token: TEST_TOKEN });
await client.sendMessage({ type: "hello", version: PROTOCOL_VERSION });
await delay.entered.promise;
await client.waitForClose();
delay.release.resolve(undefined);
@ -132,7 +118,7 @@ describe("Unix transport conformance", () => {
const outboundServer = await startServer(new TestSessionBackend(), { maxFrameLength: 128 });
const outbound = await connect(outboundServer.server);
await outbound.sendMessage({ type: "hello", version: PROTOCOL_VERSION, token: TEST_TOKEN });
await outbound.sendMessage({ type: "hello", version: PROTOCOL_VERSION });
await outbound.waitForClose();
expect(outbound.messages).toEqual([]);
});

4
packages/server/test/listener.test.ts
View file

@ -31,7 +31,7 @@ describe("PiServer listener composition", () => {
test("starts and closes every configured listener", async () => {
const first = new TestListener("first");
const second = new TestListener("second");
const { server } = createTestServer({ token: "secret", listeners: [first, second] });
const { server } = createTestServer({ listeners: [first, second] });
await server.start();
expect(server.addresses).toEqual(["first", "second"]);
@ -48,7 +48,7 @@ describe("PiServer listener composition", () => {
const first = new TestListener("first");
const failure = new Error("listener failed");
const second = new TestListener("second", failure);
const { server } = createTestServer({ token: "secret", listeners: [first, second] });
const { server } = createTestServer({ listeners: [first, second] });
await expect(server.start()).rejects.toBe(failure);
expect(first.closeCount).toBe(1);

25
packages/server/test/server.test.ts
View file

@ -8,8 +8,6 @@ import { PiServer } from "../src/index.ts";
import { TestSessionBackend } from "../src/testing/index.ts";
import { createUnixServer } from "../src/transports/unix/index.ts";
const TOKEN = "transport-smoke-token";
const backend = new TestSessionBackend();
let server: PiServer | undefined;
@ -28,25 +26,25 @@ afterEach(async () => {
});
test("requires explicit listeners", () => {
expect(() => Reflect.construct(PiServer, [backend, { token: TOKEN }])).toThrow(/listeners/);
expect(() => Reflect.construct(PiServer, [backend, {}])).toThrow(/listeners/);
});
test("rejects Unix socket paths that cannot fit in sockaddr_un", () => {
expect(() => createUnixServer(backend, { token: TOKEN, path: `/tmp/${"x".repeat(512)}` })).toThrow(/too long/);
expect(() => createUnixServer(backend, { path: `/tmp/${"x".repeat(512)}` })).toThrow(/too long/);
});
test("rejects an overlong derived private Unix bind path", async () => {
const maxLength = process.platform === "linux" ? 107 : 103;
const suffixLength = Buffer.byteLength("/tmp//s");
const path = `/tmp/${"x".repeat(maxLength - suffixLength)}/s`;
server = createUnixServer(backend, { token: TOKEN, path });
server = createUnixServer(backend, { path });
await expect(server.start()).rejects.toThrow(/private Unix bind path.*too long/);
});
test("rejects concurrent start calls without leaking the Unix listener", async () => {
const path = await makeSocketPath();
server = createUnixServer(backend, { token: TOKEN, path });
server = createUnixServer(backend, { path });
const starting = server.start();
await expect(server.start()).rejects.toThrow(/starting/);
await starting;
@ -71,7 +69,6 @@ test("handshake timeout cleanup does not wait for a blocked output queue", async
}
const core = new PiServer(backend, {
listeners: [],
token: TOKEN,
maxFrameLength: 1024,
handshakeTimeoutMs: 10,
});
@ -87,17 +84,17 @@ test("handshake timeout cleanup does not wait for a blocked output queue", async
test("rejects timeout values above Node's maximum timer delay", () => {
const unix = { path: "/tmp/pi-server-timeout-test.sock" };
expect(() =>
createUnixServer(backend, { token: TOKEN, path: unix.path, handshakeTimeoutMs: 2_147_483_648 }),
).toThrow(/handshakeTimeoutMs/);
expect(() =>
createUnixServer(backend, { token: TOKEN, path: unix.path, gracefulCloseTimeoutMs: 2_147_483_648 }),
).toThrow(/gracefulCloseTimeoutMs/);
expect(() => createUnixServer(backend, { path: unix.path, handshakeTimeoutMs: 2_147_483_648 })).toThrow(
/handshakeTimeoutMs/,
);
expect(() => createUnixServer(backend, { path: unix.path, gracefulCloseTimeoutMs: 2_147_483_648 })).toThrow(
/gracefulCloseTimeoutMs/,
);
});
test("rejects pending-byte limits smaller than one maximum frame", async () => {
const path = await makeSocketPath();
expect(() => createUnixServer(backend, { token: TOKEN, path, maxFrameLength: 128, maxPendingBytes: 131 })).toThrow(
expect(() => createUnixServer(backend, { path, maxFrameLength: 128, maxPendingBytes: 131 })).toThrow(
/maxPendingBytes/,
);
});

3
packages/server/test/sessions.test.ts
View file

@ -9,12 +9,10 @@ import {
Deferred,
type ProtocolTestClient,
TEST_MODEL,
TEST_TOKEN,
TestSessionBackend,
} from "../src/testing/index.ts";
import { createUnixServer, type UnixServerOptions } from "../src/transports/unix/index.ts";
const TOKEN = TEST_TOKEN;
const MODEL = TEST_MODEL;
type Client = ProtocolTestClient;
class MemoryBackend extends TestSessionBackend {}
@ -49,7 +47,6 @@ async function startServer(backend = new MemoryBackend(), options: Partial<UnixS
const directory = await mkdtemp(join(tmpdir(), "pst-"));
tempDirectories.add(directory);
const server = createUnixServer(backend, {
token: TOKEN,
path: join(directory, "server.sock"),
...options,
});

9
packages/server/test/unix.test.ts
View file

@ -5,12 +5,7 @@ import { tmpdir } from "node:os";
import { join } from "node:path";
import { afterEach, describe, expect, test } from "vitest";
import type { PiServer } from "../src/index.ts";
import {
connectUnixTestClient,
type ProtocolTestClient,
TEST_TOKEN,
TestSessionBackend,
} from "../src/testing/index.ts";
import { connectUnixTestClient, type ProtocolTestClient, TestSessionBackend } from "../src/testing/index.ts";
import { createUnixServer } from "../src/transports/unix/index.ts";
const servers = new Set<PiServer>();
@ -25,7 +20,7 @@ async function makeSocketPath(nested = false): Promise<string> {
}
function makeServer(path: string): PiServer {
const server = createUnixServer(new TestSessionBackend(), { token: TEST_TOKEN, path });
const server = createUnixServer(new TestSessionBackend(), { path });
servers.add(server);
return server;
}