diff --git a/packages/core/src/contacts/store.ts b/packages/core/src/contacts/store.ts index d60cdd1fb..d7028038e 100644 --- a/packages/core/src/contacts/store.ts +++ b/packages/core/src/contacts/store.ts @@ -32,6 +32,7 @@ import { joinContact, joinRequest } from './join.ts'; import { evaluatePolicy, TailscaleClient } from './tailscale.ts'; import { attestedTier, checkTier, modelTier, tierAllowed } from './tier.ts'; import { fingerprintWords } from './words.ts'; +import { TextSocket } from '../ws-client.ts'; export { CONTACT_JOIN_ROUTE } from './join.ts'; export const CONTACT_LINK_ROUTE = '/coordination/contact/link'; @@ -51,6 +52,8 @@ export const REPLY_WINDOW_MS = 7 * 24 * 3_600_000; const SKEW_MS = 60_000; const HELLO_INPUT = 'boite-contact-hello\n'; const TICK_MS = 1000; +/** How long the upgrade to a host may take before the link counts as failed. */ +const LINK_DIAL_MS = 8000; /** Every this many ticks, what waits to be told to a contact is offered again. */ const FLUSH_TICKS = 30; const TAILSCALE_CACHE_MS = 60_000; @@ -82,7 +85,7 @@ export class ContactStore { /** The origin an invitation gives: this machine's tailnet name. Tests point it at loopback. */ selfUrl: () => Promise = () => this.tailnetUrl(); /** Guest side: the link this core keeps to each host. */ - private readonly dialed = new Map(); + private readonly dialed = new Map(); /** Host side: the link each guest keeps to this core, by the guest's id. */ private readonly hosted = new Map(); /** Join and hello nonces already used, until they would be stale anyway. */ @@ -784,49 +787,44 @@ export class ContactStore { 'x-boite-hello': Buffer.from(hello).toString('base64url'), 'x-boite-signature': this.core.coordination.signature(Buffer.from(`${HELLO_INPUT}${hello}`)).toString('base64'), }; - let socket: WebSocket; - try { - // Bun's client takes headers; the hello rides in them, before any frame. - socket = new WebSocket(`${contact.url.replace(/^http/, 'ws')}${CONTACT_LINK_ROUTE}`, { headers } as unknown as string[]); - } catch { - // A link that fails counts whatever was heard before: the mark is now. - this.backoff.failed(contact.coreId, 'link', this.backoff.mark()); - return; - } const coreId = contact.coreId; - const link = { socket, state: 'connecting' as 'connecting' | 'connected' }; + const link: { socket: TextSocket | null; state: 'connecting' | 'connected' } = { socket: null, state: 'connecting' }; this.dialed.set(coreId, link); - socket.addEventListener('open', () => { + // A link that fails counts whatever was heard before: the mark is now. + const lost = () => { + if (this.dialed.get(coreId) !== link) return; + this.dialed.delete(coreId); + this.backoff.failed(coreId, 'link', this.backoff.mark()); + if (!this.closed) this.changed(); + }; + // The hello rides in the upgrade's headers, before any frame. + TextSocket.connect(`${contact.url.replace(/^http/, 'ws')}${CONTACT_LINK_ROUTE}`, LINK_DIAL_MS, { headers }).then(socket => { if (this.closed || this.dialed.get(coreId) !== link) { socket.close(1000); return; } + link.socket = socket; link.state = 'connected'; this.backoff.heard(coreId); if (this.has(coreId)) this.mutate(coreId, entry => { entry.lastSeenAt = Date.now(); }); this.flush(coreId); - }); - socket.addEventListener('message', event => { this.background(this.serve(coreId, socket, event.data)); }); - socket.addEventListener('close', () => { - if (this.dialed.get(coreId) !== link) return; - this.dialed.delete(coreId); - this.backoff.failed(coreId, 'link', this.backoff.mark()); - if (!this.closed) this.changed(); - }); + socket.onMessage(text => { this.background(this.serve(coreId, socket, text)); }); + socket.onClose(lost); + }, lost); } /** A request of the host, carried down the link: answered by the same handler as one posted to this core. */ - private async serve(hostId: string, socket: WebSocket, data: unknown): Promise { + private async serve(hostId: string, socket: TextSocket, data: string): Promise { let frame: Partial>; - try { frame = JSON.parse(typeof data === 'string' ? data : Buffer.from(data as ArrayBuffer).toString('utf8')) as typeof frame; } catch { return; } + try { frame = JSON.parse(data) as typeof frame; } catch { return; } if (frame?.t !== 'req' || typeof frame.id !== 'string' || typeof frame.body !== 'string' || typeof frame.signature !== 'string') return; let response: CoordinationBridgeResponse; try { response = await this.core.coordination.forward({ coreId: hostId, body: frame.body, signature: frame.signature }); } catch { response = { status: 400, body: '', signature: '' }; } - if (socket.readyState === WebSocket.OPEN) socket.send(JSON.stringify({ t: 'res', id: frame.id, ...response } satisfies Frame)); + if (socket.open) socket.send(JSON.stringify({ t: 'res', id: frame.id, ...response } satisfies Frame)); } /** Closes this core's side of a contact's link. */ unlink(coreId: string): void { const dialed = this.dialed.get(coreId); - if (dialed) { this.dialed.delete(coreId); dialed.socket.close(1000, 'contact removed'); } + if (dialed) { this.dialed.delete(coreId); dialed.socket?.close(1000, 'contact removed'); } const hosted = this.hosted.get(coreId); if (hosted) { this.hosted.delete(coreId); @@ -881,7 +879,7 @@ export class ContactStore { this.closed = true; if (this.timer !== null) clearInterval(this.timer); this.timer = null; - for (const { socket } of this.dialed.values()) socket.close(1001, 'core stopping'); + for (const { socket } of this.dialed.values()) socket?.close(1001, 'core stopping'); this.dialed.clear(); for (const { socket } of this.hosted.values()) socket.close(1001, 'core stopping'); this.hosted.clear(); diff --git a/packages/core/src/group.ts b/packages/core/src/group.ts index b8b3ce536..98886af69 100644 --- a/packages/core/src/group.ts +++ b/packages/core/src/group.ts @@ -29,18 +29,19 @@ import { hostname } from 'node:os'; import { join as joinPath } from 'node:path'; import { GRANT_TTL_MS, GROUP_MAX_CORES, GROUP_TICKET_TTL_MS, PAIRING_ROLES } from '@boite/contracts'; import { deviceAdmitInput, deviceHelloInput, type CoordinationPeer, type DeviceHello, type DeviceInvite, type Group, type GroupDevice, type GroupInvite, type GroupTicket, type PairingRole, type Principal } from '@boite/contracts'; -import { boundedBody, MAX_BODY, PeerGone } from './coordination-wire.ts'; +import { boundedBody, PeerGone } from './coordination-wire.ts'; import type { Core } from './core.ts'; import { invalidParams, messageOf, refused, RpcFailure, unauthorized, unavailable } from './errors.ts'; import { advertisedAddresses, probeTailnet, tailnetAddress, ticketAddresses, type Tailnet } from './group/addresses.ts'; -import { admissionInput, checkAddress, checkCard, checkRoster, clockOf, CORE_ENTRIES_MAX, DEVICE_ENTRIES_MAX, digestOf, fingerprint, homeOf, liveCores, liveDeviceKeys, liveDevices, memberOrigins, mergeRosters, groupName, peerOf, reachOf } from './group/roster.ts'; +import { admissionInput, checkAddress, checkCard, checkRoster, clockOf, CORE_ENTRIES_MAX, DEVICE_ENTRIES_MAX, digestOf, homeOf, liveCores, liveDeviceKeys, liveDevices, memberOrigins, mergeRosters, groupName, peerOf, reachOf } from './group/roster.ts'; import { deviceAdmissionInput, deviceId, deviceKey, deviceSigned, SERVED_ORIGINS_MAX, type DeviceKeyEntry } from './group/device-keys.ts'; import type { CoreCard, CoreEntry, DeviceEntry, Roster } from './group/roster.ts'; -import { boxPublic, boxSigningInput, newBoxKey, open, openResponse, pack, readBoxKey, seal, SEALED, sealedName, sealResponse, unpack } from './group/seal.ts'; +import { boxPublic, boxSigningInput, newBoxKey, open, pack, readBoxKey, seal, SEALED, sealedName, sealResponse, unpack } from './group/seal.ts'; import type { Sealing } from './group/seal.ts'; import { encodeInvite, encodeTicket, inviteOf, parseInvite, parseTicket, ticketSigningInput, inviteId, invitePsk } from './group/ticket.ts'; import type { Invite } from './group/ticket.ts'; import { DEVICES_MAX, GroupClients } from './group/clients.ts'; +import { askToJoin, JOIN_SIGNING_PREFIX, type JoinReply } from './group/join.ts'; import { SyncBackoff } from './group/backoff.ts'; import { newId, newToken } from './ids.ts'; import { currentOs } from './paths.ts'; @@ -54,11 +55,7 @@ const SPENT_SETTING = 'group:spent'; const JOINS_PER_MINUTE = 30; /** Requests one invitation is asked with before it is taken away: a machine joins once, and asks again a few times at most. */ const JOIN_REQUESTS_MAX = 64; -export const JOIN_ROUTE = '/group/join'; -/** What a join request and its answer are signed with, so neither passes for a coordination message. */ -const JOIN_SIGNING_PREFIX = 'boite-group-join\n'; const JOIN_BODY_MAX = 16_384; -const JOIN_TIMEOUT_MS = 8000; /** A member that missed a change is offered it again this often. */ const TICK_MS = 15_000; /** Every member is asked for its roster this many ticks apart, whatever it acknowledged: five minutes. */ @@ -68,8 +65,7 @@ const NAME_REFRESH_MS = 5 * 60_000; const FAREWELL_MS = 6000; export type { NetworkSink } from './group/addresses.ts'; - -interface JoinReply { nonce: string; publicKey: string; roster?: unknown; error?: string } +export { JOIN_ROUTE } from './group/join.ts'; export class GroupStore { roster: Roster | null = null; @@ -707,7 +703,7 @@ export class GroupStore { const sealed = { ...seal(pack(body, signature), invite.x, context, invitePsk(invite.t)), context }; let reply: JoinReply; try { - reply = await Promise.any(invite.a.map((url) => this.askToJoin(url, invite, sealed, nonce))); + reply = await Promise.any(invite.a.map((url) => askToJoin(url, invite, sealed, nonce))); } catch { throw refused(`no machine accepted the invitation at ${invite.a.join(', ')}: it is off or unreachable from here, or the invitation expired, was already used or was made by another machine`, { field: 'invite' }); } @@ -740,27 +736,6 @@ export class GroupStore { return this.view('owner')!; } - private async askToJoin(url: string, invite: Invite, sealed: { body: string } & Sealing, nonce: string): Promise { - const response = await fetch(`${url}${JOIN_ROUTE}`, { - method: 'POST', - redirect: 'error', - signal: AbortSignal.timeout(JOIN_TIMEOUT_MS), - headers: { 'content-type': 'application/json', 'x-boite-peer': sealed.context.from, 'x-boite-invite': inviteId(invite.t), 'x-boite-signature': SEALED }, - body: sealed.body, - }); - const raw = await boundedBody(response.body, MAX_BODY); - // Only the machine that minted the invitation can answer under this key. - if (response.headers.get('x-boite-signature') !== SEALED) throw new Error('the answer is not sealed'); - const { body, signature } = unpack(openResponse(raw, sealed.responseKey, sealed.context)); - const reply = JSON.parse(body) as JoinReply; - // The invitation names who must answer: a machine with another key is not it. - const key = fingerprint(reply.publicKey); - if (key.coreId !== invite.c) throw new Error('another machine answered'); - if (!verify(null, Buffer.from(JOIN_SIGNING_PREFIX + body), key.publicKey, Buffer.from(signature, 'base64'))) throw new Error('invalid signature'); - if (reply.nonce !== nonce) throw new Error('nonce mismatch'); - return reply; - } - /** * The other end of `join`, on the member that minted the invitation. The * caller is in no roster yet. Its request names an invitation and is sealed diff --git a/packages/core/src/group/join.ts b/packages/core/src/group/join.ts new file mode 100644 index 000000000..908a0cecc --- /dev/null +++ b/packages/core/src/group/join.ts @@ -0,0 +1,38 @@ +/* + * The joining side of a group invitation: one sealed request to the member that + * minted it, and its answer checked against the key the invitation names. + * The member's side is `GroupStore.http` (docs/groups.md). + */ +import { verify } from 'node:crypto'; +import { boundedBody, MAX_BODY } from '../coordination-wire.ts'; +import { fingerprint } from './roster.ts'; +import { openResponse, SEALED, unpack, type Sealing } from './seal.ts'; +import { inviteId, type Invite } from './ticket.ts'; + +export const JOIN_ROUTE = '/group/join'; +/** What a join request and its answer are signed with, so neither passes for a coordination message. */ +export const JOIN_SIGNING_PREFIX = 'boite-group-join\n'; +const JOIN_TIMEOUT_MS = 8000; + +export interface JoinReply { nonce: string; publicKey: string; roster?: unknown; error?: string } + +export async function askToJoin(url: string, invite: Invite, sealed: { body: string } & Sealing, nonce: string): Promise { + const response = await fetch(`${url}${JOIN_ROUTE}`, { + method: 'POST', + redirect: 'error', + signal: AbortSignal.timeout(JOIN_TIMEOUT_MS), + headers: { 'content-type': 'application/json', 'x-boite-peer': sealed.context.from, 'x-boite-invite': inviteId(invite.t), 'x-boite-signature': SEALED }, + body: sealed.body, + }); + const raw = await boundedBody(response.body, MAX_BODY); + // Only the machine that minted the invitation can answer under this key. + if (response.headers.get('x-boite-signature') !== SEALED) throw new Error('the answer is not sealed'); + const { body, signature } = unpack(openResponse(raw, sealed.responseKey, sealed.context)); + const reply = JSON.parse(body) as JoinReply; + // The invitation names who must answer: a machine with another key is not it. + const key = fingerprint(reply.publicKey); + if (key.coreId !== invite.c) throw new Error('another machine answered'); + if (!verify(null, Buffer.from(JOIN_SIGNING_PREFIX + body), key.publicKey, Buffer.from(signature, 'base64'))) throw new Error('invalid signature'); + if (reply.nonce !== nonce) throw new Error('nonce mismatch'); + return reply; +} diff --git a/packages/core/test/group-devices.test.ts b/packages/core/test/group-devices.test.ts index bb99de6d1..1d334fcf2 100644 --- a/packages/core/test/group-devices.test.ts +++ b/packages/core/test/group-devices.test.ts @@ -293,9 +293,9 @@ test('a machine joining a group tells a device of it to come back, not that it i let asked!: () => void; const held = new Promise((resolve) => { release = resolve; }); const asking = new Promise((resolve) => { asked = resolve; }); - const group = b.core.group as unknown as { askToJoin: (...args: unknown[]) => Promise }; - const ask = group.askToJoin.bind(b.core.group); - group.askToJoin = async (...args) => { asked(); await held; return ask(...args); }; + const group = a.core.group as { http: (request: Request) => Promise }; + const answer = group.http.bind(a.core.group); + group.http = async (request) => { asked(); await held; return answer(request); }; const joining = b.core.group.join(invite); await asking; let refusal: { code?: number; data?: unknown } | undefined;