feat(relay-run): Phase 0 single-host relay — real data-plane wiring + integration test
New relay-run/ package boots the REAL term-relay relay-node between a browser WSS
listener and an agent mTLS listener, delegating every authz verdict to the real
relay-auth onUpgrade (Origin/CSWSH + capability verify + DPoP PoP + single-use jti),
with the real P4 E2E crypto. No audited package modified.
- P1 (done): in-process integration test round-trips a sealed payload both ways
through the real createRelayNode splice + real createMuxSession; INV2 asserted
(relay sees only ciphertext); negative controls (foreign Origin / missing DPoP -> 401).
- P2 (boots): main.ts brings up self-signed TLS + 3 listeners + in-memory control-plane.
- P3 (not landed): no agent dial / node-pty / relay-web serve yet.
- 4 tests green, tsc clean. node_modules via symlinks (gitignored).
Two real seam drifts surfaced (documented in PLAN_RELAY_RUN_PHASE0, not hacked):
(1) term-relay authz-port DpopContext shape vs relay-auth — reconciled (server-derived htu/htm);
(2) control-plane CapabilityVerifier.verify is sync but relay-auth verify is async — blocks
HTTP account/pairing seeding until a 1-line async widen.
This commit is contained in:
104
relay-run/src/wiring/socket-pipe.ts
Normal file
104
relay-run/src/wiring/socket-pipe.ts
Normal file
@@ -0,0 +1,104 @@
|
||||
/**
|
||||
* `WebSocketLike` adapters. Two of them:
|
||||
* - `makeSocketPair()` — an in-process duplex pair used to drive the REAL relay-node/mux in tests
|
||||
* (no sockets, deterministic). Delivery is synchronous but non-reentrant (a per-endpoint
|
||||
* trampoline) so a send emitted while draining the peer's inbox can't recurse unboundedly.
|
||||
* - `wsToWebSocketLike()` — wraps a real `ws` WebSocket to the same interface for `main.ts`.
|
||||
* Bytes are COPIED on send (the mux slices/retains buffers) so no caller can alias another's memory.
|
||||
*/
|
||||
import type { WebSocket as WsWebSocket } from 'ws'
|
||||
import type { WebSocketLike } from 'term-relay/data-plane/ws-like.js'
|
||||
|
||||
interface Endpoint extends WebSocketLike {
|
||||
/** Internal: deliver bytes that the peer sent to THIS endpoint. */
|
||||
_deliver(data: Uint8Array): void
|
||||
_closed(): void
|
||||
}
|
||||
|
||||
function makeEndpoint(): Endpoint {
|
||||
let onMessage: ((data: Uint8Array) => void) | null = null
|
||||
let onClose: (() => void) | null = null
|
||||
let peer: Endpoint | null = null
|
||||
const inbox: Uint8Array[] = []
|
||||
let draining = false
|
||||
let closed = false
|
||||
|
||||
const ep: Endpoint & { _link(p: Endpoint): void } = {
|
||||
_link(p: Endpoint) {
|
||||
peer = p
|
||||
},
|
||||
send(data: Uint8Array) {
|
||||
if (closed || peer === null) return
|
||||
peer._deliver(new Uint8Array(data)) // copy: never alias the sender's buffer
|
||||
},
|
||||
close(_code?: number) {
|
||||
if (closed) return
|
||||
closed = true
|
||||
onClose?.()
|
||||
peer?._closed()
|
||||
},
|
||||
onMessage(handler) {
|
||||
onMessage = handler
|
||||
},
|
||||
onClose(handler) {
|
||||
onClose = handler
|
||||
},
|
||||
_deliver(data: Uint8Array) {
|
||||
inbox.push(data)
|
||||
if (draining) return
|
||||
draining = true
|
||||
try {
|
||||
while (inbox.length > 0) {
|
||||
const next = inbox.shift()!
|
||||
onMessage?.(next)
|
||||
}
|
||||
} finally {
|
||||
draining = false
|
||||
}
|
||||
},
|
||||
_closed() {
|
||||
if (closed) return
|
||||
closed = true
|
||||
onClose?.()
|
||||
},
|
||||
}
|
||||
return ep as Endpoint & { _link(p: Endpoint): void }
|
||||
}
|
||||
|
||||
/** An in-process duplex `WebSocketLike` pair: `a.send` → `b`'s message handler and vice versa. */
|
||||
export function makeSocketPair(): { a: WebSocketLike; b: WebSocketLike } {
|
||||
const a = makeEndpoint() as Endpoint & { _link(p: Endpoint): void }
|
||||
const b = makeEndpoint() as Endpoint & { _link(p: Endpoint): void }
|
||||
a._link(b)
|
||||
b._link(a)
|
||||
return { a, b }
|
||||
}
|
||||
|
||||
function toUint8(data: unknown): Uint8Array | null {
|
||||
if (data instanceof Uint8Array) return data
|
||||
if (data instanceof ArrayBuffer) return new Uint8Array(data)
|
||||
if (Array.isArray(data)) return Buffer.concat(data.map((d) => (d instanceof Uint8Array ? d : Buffer.from(d))))
|
||||
if (Buffer.isBuffer(data)) return new Uint8Array(data)
|
||||
return null
|
||||
}
|
||||
|
||||
/** Adapt a live `ws` WebSocket to `WebSocketLike` (binary frames only). */
|
||||
export function wsToWebSocketLike(ws: WsWebSocket): WebSocketLike {
|
||||
return {
|
||||
send(data: Uint8Array) {
|
||||
ws.send(data, { binary: true })
|
||||
},
|
||||
close(code?: number) {
|
||||
ws.close(code)
|
||||
},
|
||||
onMessage(handler) {
|
||||
ws.on('message', (data: unknown) => {
|
||||
const bytes = toUint8(data)
|
||||
if (bytes !== null) handler(bytes)
|
||||
})
|
||||
},
|
||||
onClose(handler) {
|
||||
ws.on('close', () => handler())
|
||||
},
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user