From e356c5cd5a797a34953cb75078e845a35f4ed0b3 Mon Sep 17 00:00:00 2001 From: Steven Enamakel <31011319+senamakel@users.noreply.github.com> Date: Sat, 6 Jun 2026 19:04:52 -0400 Subject: [PATCH] feat(ios): complete QR tunnel pairing flow (#3452) --- app/src/lib/tunnel/crypto.test.ts | 14 + app/src/lib/tunnel/crypto.ts | 10 + .../transport/TransportManager.test.ts | 17 + .../services/transport/TransportManager.ts | 7 +- .../transport/TunnelTransport.test.ts | 236 +++++++++++-- app/src/services/transport/TunnelTransport.ts | 126 ++++++- src/openhuman/devices/bus.rs | 334 ++++++++++++++++-- src/openhuman/devices/rpc.rs | 10 +- 8 files changed, 672 insertions(+), 82 deletions(-) diff --git a/app/src/lib/tunnel/crypto.test.ts b/app/src/lib/tunnel/crypto.test.ts index 51037d9d8..1f3e98a62 100644 --- a/app/src/lib/tunnel/crypto.test.ts +++ b/app/src/lib/tunnel/crypto.test.ts @@ -12,6 +12,7 @@ import { generateKeypair, HKDF_INFO_C2S, HKDF_INFO_S2C, + keypairFromSecretKey, LEGACY_FRAME_VERSION_V1, open, openHandshake, @@ -61,6 +62,19 @@ describe('generateKeypair', () => { }); }); +describe('keypairFromSecretKey', () => { + it('restores the public key from a private key', () => { + const original = generateKeypair(); + const restored = keypairFromSecretKey(original.secretKey); + expect(restored.secretKey).toEqual(original.secretKey); + expect(restored.publicKey).toEqual(original.publicKey); + }); + + it('rejects non-X25519 private key lengths', () => { + expect(() => keypairFromSecretKey(new Uint8Array(31))).toThrow(/32 bytes/i); + }); +}); + describe('deriveSharedSecret', () => { it('both sides derive the same secret', () => { const alice = generateKeypair(); diff --git a/app/src/lib/tunnel/crypto.ts b/app/src/lib/tunnel/crypto.ts index 9aac31130..7c9c3656a 100644 --- a/app/src/lib/tunnel/crypto.ts +++ b/app/src/lib/tunnel/crypto.ts @@ -77,6 +77,16 @@ export function generateKeypair(): TunnelKeypair { return { publicKey, secretKey }; } +/** Reconstruct an X25519 keypair from a 32-byte private key. */ +export function keypairFromSecretKey(secretKey: Uint8Array): TunnelKeypair { + if (secretKey.length !== 32) { + throw new Error(`[crypto] X25519 private key must be 32 bytes, got ${secretKey.length}`); + } + const publicKey = x25519.getPublicKey(secretKey); + cryptoLog('[crypto] keypair restored pubkey_len=%d', publicKey.length); + return { publicKey, secretKey }; +} + /** Derive a 32-byte X25519 shared secret. */ export function deriveSharedSecret(myPriv: Uint8Array, theirPub: Uint8Array): Uint8Array { const shared = x25519.getSharedSecret(myPriv, theirPub); diff --git a/app/src/services/transport/TransportManager.test.ts b/app/src/services/transport/TransportManager.test.ts index 59444dfff..0828a897d 100644 --- a/app/src/services/transport/TransportManager.test.ts +++ b/app/src/services/transport/TransportManager.test.ts @@ -84,6 +84,23 @@ describe('TransportManager', () => { await manager.close(); }); + it('tunnel profile can use a pairing token before a session token exists', async () => { + const profile = makeProfile('tunnel', { + rpcUrl: undefined, + sessionToken: undefined, + pairingToken: 'pair123', + }); + const manager = new TransportManager( + profile, + () => Promise.resolve(''), + () => Promise.resolve(null), + 'http://backend:3000' + ); + const t = await manager.getTransport(); + expect(t.kind).toBe('tunnel'); + await manager.close(); + }); + it('throws when tunnel profile missing channelId', async () => { const profile = makeProfile('tunnel', { channelId: undefined }); const manager = new TransportManager( diff --git a/app/src/services/transport/TransportManager.ts b/app/src/services/transport/TransportManager.ts index 20bad2c41..d2d861736 100644 --- a/app/src/services/transport/TransportManager.ts +++ b/app/src/services/transport/TransportManager.ts @@ -107,7 +107,8 @@ export class TransportManager { * If LAN wins but later fails, caller should call reset() to re-race. */ private async raceLanAndTunnel(): Promise { - const { rpcUrl, channelId, corePubkey, sessionToken, pairingToken } = this.profile; + const { rpcUrl, channelId, corePubkey, sessionToken, pairingToken, devicePrivkey } = + this.profile; if (!channelId || !corePubkey) { throw new Error('[transport:manager] tunnel profile missing channelId or corePubkey'); @@ -122,7 +123,9 @@ export class TransportManager { this.backendSocketUrl, channelId, corePubkey, - tunnelToken + tunnelToken, + devicePrivkey, + sessionToken ? 'session' : 'pairing' ); if (!rpcUrl) { diff --git a/app/src/services/transport/TunnelTransport.test.ts b/app/src/services/transport/TunnelTransport.test.ts index 772baad63..a15f49b67 100644 --- a/app/src/services/transport/TunnelTransport.test.ts +++ b/app/src/services/transport/TunnelTransport.test.ts @@ -4,15 +4,18 @@ * We mock socket.io-client so no real network connection is made. * Each test gets a fresh socket mock via the module factory pattern. */ -import { beforeEach, describe, expect, it, vi } from 'vitest'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import { base64urlEncode, + deriveSessionKeys, deriveSharedSecret, generateKeypair, open, ReplayTracker, seal, + TunnelCipher, + type TunnelKeypair, } from '../../lib/tunnel/crypto'; // -- socket mock factory ------------------------------------------------------- @@ -51,14 +54,41 @@ function fire(event: string, ...args: unknown[]) { _handlers.get(event)?.(...args); } -async function connectTransport(transport: InstanceType): Promise { +async function connectTransport( + transport: InstanceType +): Promise { const connectP = (transport as unknown as { ensureConnected(): Promise }).ensureConnected(); // Flush: give socket.on a chance to register. await Promise.resolve(); fire('connect'); await Promise.resolve(); fire('tunnel:connected'); + await Promise.resolve(); + + type HandshakeInternals = { + staticDhKey: Uint8Array | null; + clientEphemeralKeypair: TunnelKeypair | null; + }; + const internals = transport as unknown as HandshakeInternals; + expect(internals.staticDhKey).toBeTruthy(); + expect(internals.clientEphemeralKeypair).toBeTruthy(); + + const serverEphemeral = generateKeypair(); + const keys = deriveSessionKeys( + internals.staticDhKey!, + deriveSharedSecret(serverEphemeral.secretKey, internals.clientEphemeralKeypair!.publicKey), + internals.clientEphemeralKeypair!.publicKey, + serverEphemeral.publicKey + ); + const ack = new TextEncoder().encode( + JSON.stringify({ + kind: 'handshake_ack', + server_ephemeral_pubkey: base64urlEncode(serverEphemeral.publicKey), + }) + ); + fire('tunnel:frame', { payload: base64urlEncode(seal(internals.staticDhKey!, ack)) }); await connectP; + return new TunnelCipher('server', keys); } function coreB64(kp: ReturnType) { @@ -71,6 +101,10 @@ beforeEach(() => { resetSocket(); }); +afterEach(() => { + vi.useRealTimers(); +}); + describe('TunnelTransport', () => { it('emits tunnel:connect with channelId + role on connect', async () => { const coreKp = generateKeypair(); @@ -90,6 +124,30 @@ describe('TunnelTransport', () => { await transport.close(); }); + it('emits a session token field for reconnect authentication', async () => { + const coreKp = generateKeypair(); + const transport = new TunnelTransport( + 'http://backend', + 'CHAN_SESSION', + coreB64(coreKp), + 'sess_tok', + undefined, + 'session' + ); + + await connectTransport(transport); + + const connectCall = _emitSpy.mock.calls.find(([ev]) => ev === 'tunnel:connect'); + expect(connectCall![1]).toMatchObject({ + channelId: 'CHAN_SESSION', + token: 'sess_tok', + sessionToken: 'sess_tok', + }); + expect((connectCall![1] as { pairingToken?: string }).pairingToken).toBeUndefined(); + + await transport.close(); + }); + it('rejects pending calls when close() is called', async () => { const coreKp = generateKeypair(); const transport = new TunnelTransport('http://backend', 'CHAN_002', coreB64(coreKp), 'tok'); @@ -140,11 +198,140 @@ describe('TunnelTransport', () => { await expect(connectP).rejects.toThrow(/server error|unauthorized/i); }, 5000); + it('rejects the connect promise when the handshake ack times out', async () => { + vi.useFakeTimers(); + const coreKp = generateKeypair(); + const transport = new TunnelTransport('http://backend', 'CHAN_TIMEOUT', coreB64(coreKp), 'tok'); + + const connectP = ( + transport as unknown as { ensureConnected(): Promise } + ).ensureConnected(); + await Promise.resolve(); + fire('connect'); + await Promise.resolve(); + fire('tunnel:connected'); + await Promise.resolve(); + + vi.advanceTimersByTime(10_000); + await expect(connectP).rejects.toThrow(/handshake ack timed out/i); + + await transport.close(); + }); + + it('rejects the connect promise when the handshake ack cannot be opened', async () => { + const coreKp = generateKeypair(); + const transport = new TunnelTransport( + 'http://backend', + 'CHAN_BAD_ACK_OPEN', + coreB64(coreKp), + 'tok' + ); + + const connectP = ( + transport as unknown as { ensureConnected(): Promise } + ).ensureConnected(); + await Promise.resolve(); + fire('connect'); + await Promise.resolve(); + fire('tunnel:connected'); + await Promise.resolve(); + + fire('tunnel:frame', { payload: base64urlEncode(new Uint8Array([1, 2, 3, 4])) }); + await expect(connectP).rejects.toThrow(); + + await transport.close(); + }); + + it('rejects the connect promise when the handshake ack is not JSON', async () => { + const coreKp = generateKeypair(); + const transport = new TunnelTransport( + 'http://backend', + 'CHAN_BAD_ACK_JSON', + coreB64(coreKp), + 'tok' + ); + + const connectP = ( + transport as unknown as { ensureConnected(): Promise } + ).ensureConnected(); + await Promise.resolve(); + fire('connect'); + await Promise.resolve(); + fire('tunnel:connected'); + await Promise.resolve(); + + type HandshakeInternals = { staticDhKey: Uint8Array | null }; + const internals = transport as unknown as HandshakeInternals; + expect(internals.staticDhKey).toBeTruthy(); + + fire('tunnel:frame', { + payload: base64urlEncode(seal(internals.staticDhKey!, new TextEncoder().encode('not json'))), + }); + await expect(connectP).rejects.toThrow(); + + await transport.close(); + }); + + it('rejects the connect promise when the handshake ack kind is invalid', async () => { + const coreKp = generateKeypair(); + const transport = new TunnelTransport( + 'http://backend', + 'CHAN_BAD_ACK_KIND', + coreB64(coreKp), + 'tok' + ); + + const connectP = ( + transport as unknown as { ensureConnected(): Promise } + ).ensureConnected(); + await Promise.resolve(); + fire('connect'); + await Promise.resolve(); + fire('tunnel:connected'); + await Promise.resolve(); + + type HandshakeInternals = { staticDhKey: Uint8Array | null }; + const internals = transport as unknown as HandshakeInternals; + expect(internals.staticDhKey).toBeTruthy(); + + fire('tunnel:frame', { + payload: base64urlEncode( + seal(internals.staticDhKey!, new TextEncoder().encode(JSON.stringify({ kind: 'nope' }))) + ), + }); + await expect(connectP).rejects.toThrow(/invalid handshake ack/i); + + await transport.close(); + }); + + it('rejects an in-flight handshake when close() is called', async () => { + const coreKp = generateKeypair(); + const transport = new TunnelTransport( + 'http://backend', + 'CHAN_CLOSE_HANDSHAKE', + coreB64(coreKp), + 'tok' + ); + + const connectP = ( + transport as unknown as { ensureConnected(): Promise } + ).ensureConnected(); + await Promise.resolve(); + fire('connect'); + await Promise.resolve(); + fire('tunnel:connected'); + await Promise.resolve(); + + await transport.close(); + + await expect(connectP).rejects.toThrow(/transport closed/i); + }); + it('resolves call() when a matching encrypted response frame arrives', async () => { const coreKp = generateKeypair(); const transport = new TunnelTransport('http://backend', 'CHAN_004', coreB64(coreKp), 'tok'); - await connectTransport(transport); + const serverCipher = await connectTransport(transport); const callP = transport.call<{ pong: number }>('openhuman.ping', { who: 'me' }); @@ -152,31 +339,19 @@ describe('TunnelTransport', () => { await Promise.resolve(); await Promise.resolve(); - // Extract requestId from the chunk envelope the client just emitted. - // Since chunks are encrypted we can't decode them — instead simulate the - // server response by re-using the same session key derivation in reverse. - // The transport derives sessionKey from (device.secret, core.public). The - // server side derives the same key from (core.secret, device.public). We - // mimic that by importing the same helpers. - const { deriveSharedSecret, seal, base64urlEncode } = await import('../../lib/tunnel/crypto'); const { chunk } = await import('../../lib/tunnel/framing'); // Pull the device pubkey out of the handshake frame the client sent. const handshakeCall = _emitSpy.mock.calls.find(([ev]) => ev === 'tunnel:frame'); expect(handshakeCall).toBeTruthy(); - // We can't decode the handshake without the core's secret key, but the - // transport exposes its sessionKey on the instance (derived from the - // device keypair). Reach in to get it for the test. - type Internals = { sessionKey: Uint8Array | null; pending: Map }; + type Internals = { pending: Map }; const internals = transport as unknown as Internals; - // Wait until sessionKey is populated and pending request is registered. - for (let i = 0; i < 10 && (!internals.sessionKey || internals.pending.size === 0); i++) { + // Wait until the pending request is registered. + for (let i = 0; i < 10 && internals.pending.size === 0; i++) { await Promise.resolve(); } - expect(internals.sessionKey).toBeTruthy(); - const sessionKey = internals.sessionKey!; const [requestId] = Array.from(internals.pending.keys()) as string[]; expect(requestId).toBeTruthy(); @@ -184,13 +359,11 @@ describe('TunnelTransport', () => { // via the tunnel:frame handler. const envelope = { requestId, kind: 'response' as const, seq: 0, payload: { pong: 42 } }; for (const raw of chunk(envelope)) { - const encrypted = seal(sessionKey, raw); + const encrypted = serverCipher.seal(raw); fire('tunnel:frame', { payload: base64urlEncode(encrypted) }); } await expect(callP).resolves.toEqual({ pong: 42 }); - // unused helper in this test, satisfy linter - void deriveSharedSecret; await transport.close(); }, 10000); @@ -198,24 +371,23 @@ describe('TunnelTransport', () => { it('routes error envelopes back to the matching pending call', async () => { const coreKp = generateKeypair(); const transport = new TunnelTransport('http://backend', 'CHAN_005', coreB64(coreKp), 'tok'); - await connectTransport(transport); + const serverCipher = await connectTransport(transport); const callP = transport.call('openhuman.fail', {}); await Promise.resolve(); await Promise.resolve(); - const { seal, base64urlEncode } = await import('../../lib/tunnel/crypto'); const { chunk } = await import('../../lib/tunnel/framing'); - type Internals = { sessionKey: Uint8Array | null; pending: Map }; + type Internals = { pending: Map }; const internals = transport as unknown as Internals; - for (let i = 0; i < 10 && (!internals.sessionKey || internals.pending.size === 0); i++) { + for (let i = 0; i < 10 && internals.pending.size === 0; i++) { await Promise.resolve(); } const [requestId] = Array.from(internals.pending.keys()) as string[]; const envelope = { requestId, kind: 'error' as const, seq: 0, payload: 'tunnel exploded' }; for (const raw of chunk(envelope)) { - fire('tunnel:frame', { payload: base64urlEncode(seal(internals.sessionKey!, raw)) }); + fire('tunnel:frame', { payload: base64urlEncode(serverCipher.seal(raw)) }); } await expect(callP).rejects.toThrow('tunnel exploded'); @@ -235,7 +407,7 @@ describe('TunnelTransport', () => { await transport.close(); }); - it('ignores frames that arrive before the session key is set', async () => { + it('ignores frames that arrive before the session cipher is set', async () => { const coreKp = generateKeypair(); const transport = new TunnelTransport('http://backend', 'CHAN_007', coreB64(coreKp), 'tok'); @@ -243,7 +415,7 @@ describe('TunnelTransport', () => { void (transport as unknown as { ensureConnected(): Promise }).ensureConnected(); await Promise.resolve(); fire('connect'); - // (no tunnel:connected → no handshake → sessionKey stays null) + // (no tunnel:connected → no handshake → session cipher stays null) // Frame arrives early — should be silently dropped. fire('tunnel:frame', { payload: 'AAAAAAA' }); @@ -263,17 +435,17 @@ describe('TunnelTransport', () => { await expect(healthyP).resolves.toBe(false); }); - it('disconnect resets the session key and connect promise', async () => { + it('disconnect resets the session cipher and connect promise', async () => { const coreKp = generateKeypair(); const transport = new TunnelTransport('http://backend', 'CHAN_009', coreB64(coreKp), 'tok'); await connectTransport(transport); - type Internals = { sessionKey: Uint8Array | null; _connectPromise: Promise | null }; + type Internals = { cipher: TunnelCipher | null; _connectPromise: Promise | null }; const internals = transport as unknown as Internals; - expect(internals.sessionKey).toBeTruthy(); + expect(internals.cipher).toBeTruthy(); fire('disconnect', 'transport close'); - expect(internals.sessionKey).toBeNull(); + expect(internals.cipher).toBeNull(); expect(internals._connectPromise).toBeNull(); await transport.close(); diff --git a/app/src/services/transport/TunnelTransport.ts b/app/src/services/transport/TunnelTransport.ts index b383d58ea..d7e60325f 100644 --- a/app/src/services/transport/TunnelTransport.ts +++ b/app/src/services/transport/TunnelTransport.ts @@ -16,12 +16,14 @@ import { io, Socket } from 'socket.io-client'; import { base64urlDecode, base64urlEncode, + deriveSessionKeys, deriveSharedSecret, generateKeypair, + keypairFromSecretKey, open, ReplayTracker, - seal, sealHandshake, + TunnelCipher, type TunnelKeypair, } from '../../lib/tunnel/crypto'; import { chunk, Envelope, Reassembler, TokenBucket } from '../../lib/tunnel/framing'; @@ -50,8 +52,15 @@ export class TunnelTransport implements CoreTransport { readonly kind = 'tunnel' as const; private socket: Socket | null = null; - private sessionKey: Uint8Array | null = null; // derived after handshake + private staticDhKey: Uint8Array | null = null; // bootstrap key for handshake ack + private clientEphemeralKeypair: TunnelKeypair | null = null; + private cipher: TunnelCipher | null = null; // derived after handshake ack private deviceKeypair: TunnelKeypair | null = null; + private handshakeAck: { + resolve: () => void; + reject: (err: Error) => void; + timeoutId: ReturnType; + } | null = null; private readonly replayTracker = new ReplayTracker(); private readonly reassembler = new Reassembler(); private readonly rateLimiter = new TokenBucket(100, 100); @@ -66,11 +75,14 @@ export class TunnelTransport implements CoreTransport { private readonly channelId: string, private readonly corePubkeyB64: string, private readonly authToken: string, // sessionToken (reconnect) or pairingToken (first) + devicePrivkeyB64?: string | null, + private readonly authTokenKind: 'pairing' | 'session' = 'pairing', private readonly role: 'client' = 'client', private readonly callTimeoutMs: number = 30_000 ) { - // Generate device keypair on construction. - this.deviceKeypair = generateKeypair(); + this.deviceKeypair = devicePrivkeyB64 + ? keypairFromSecretKey(base64urlDecode(devicePrivkeyB64)) + : generateKeypair(); log('[tunnel] created channelId=%s corePubkey=%s…', channelId, corePubkeyB64.slice(0, 4)); } @@ -98,6 +110,9 @@ export class TunnelTransport implements CoreTransport { channelId: this.channelId, role: this.role, token: this.authToken, + ...(this.authTokenKind === 'session' + ? { sessionToken: this.authToken } + : { pairingToken: this.authToken }), }); }); @@ -120,7 +135,9 @@ export class TunnelTransport implements CoreTransport { socket.on('disconnect', (reason: string) => { log('[tunnel] disconnected reason=%s', reason); - this.sessionKey = null; + this.staticDhKey = null; + this.clientEphemeralKeypair = null; + this.cipher = null; this._connectPromise = null; }); @@ -141,9 +158,15 @@ export class TunnelTransport implements CoreTransport { const corePubkey = base64urlDecode(this.corePubkeyB64); const devicePubkeyB64 = base64urlEncode(this.deviceKeypair.publicKey); + const clientEphemeral = generateKeypair(); + const clientEphemeralPubkeyB64 = base64urlEncode(clientEphemeral.publicKey); - // Device pubkey payload (base64url-encoded, UTF-8). - const payload = new TextEncoder().encode(devicePubkeyB64); + const payload = new TextEncoder().encode( + JSON.stringify({ + device_pubkey: devicePubkeyB64, + client_ephemeral_pubkey: clientEphemeralPubkeyB64, + }) + ); // Seal the handshake payload to the core's public key. const handshakeFrame = sealHandshake(corePubkey, payload); @@ -152,10 +175,18 @@ export class TunnelTransport implements CoreTransport { log('[tunnel] sending sealed handshake frame_len=%d', handshakeFrame.length); this.socket!.emit('tunnel:frame', { channelId: this.channelId, payload: frameB64 }); - // Derive session key from static keys (both sides derive the same key). - this.sessionKey = deriveSharedSecret(this.deviceKeypair.secretKey, corePubkey); + this.staticDhKey = deriveSharedSecret(this.deviceKeypair.secretKey, corePubkey); + this.clientEphemeralKeypair = clientEphemeral; - log('[tunnel] handshake complete, session key derived'); + log('[tunnel] handshake sent, waiting for server ephemeral ack'); + + return new Promise((resolve, reject) => { + const timeoutId = setTimeout(() => { + this.handshakeAck = null; + reject(new Error('[tunnel] handshake ack timed out')); + }, 10_000); + this.handshakeAck = { resolve, reject, timeoutId }; + }); } // -- incoming frames ------------------------------------------------------- @@ -168,11 +199,6 @@ export class TunnelTransport implements CoreTransport { return; } - if (!this.sessionKey) { - log('[tunnel] frame received before session key — ignoring'); - return; - } - let frameBytes: Uint8Array; try { frameBytes = base64urlDecode(payloadB64); @@ -183,7 +209,11 @@ export class TunnelTransport implements CoreTransport { let plaintext: Uint8Array; try { - plaintext = open(this.sessionKey, frameBytes, this.replayTracker); + if (!this.cipher) { + this.handleHandshakeAck(frameBytes); + return; + } + plaintext = this.cipher.open(frameBytes); } catch (err) { logErr('[tunnel] frame decryption failed: %s', (err as Error).message); return; @@ -195,6 +225,57 @@ export class TunnelTransport implements CoreTransport { this.dispatchEnvelope(envelope); } + private handleHandshakeAck(frameBytes: Uint8Array): void { + if (!this.staticDhKey || !this.clientEphemeralKeypair) { + log('[tunnel] handshake ack received before handshake state — ignoring'); + return; + } + + let plaintext: Uint8Array; + try { + plaintext = open(this.staticDhKey, frameBytes, this.replayTracker); + } catch (err) { + this.handshakeAck?.reject(err as Error); + logErr('[tunnel] handshake ack open failed: %s', (err as Error).message); + return; + } + + let ack: { kind?: string; server_ephemeral_pubkey?: string }; + try { + ack = JSON.parse(new TextDecoder().decode(plaintext)) as { + kind?: string; + server_ephemeral_pubkey?: string; + }; + } catch (err) { + this.handshakeAck?.reject(err as Error); + logErr('[tunnel] handshake ack JSON parse failed: %o', err); + return; + } + + if (ack.kind !== 'handshake_ack' || !ack.server_ephemeral_pubkey) { + const err = new Error(`[tunnel] invalid handshake ack kind=${ack.kind ?? 'missing'}`); + this.handshakeAck?.reject(err); + logErr(err.message); + return; + } + + const serverEphemeralPubkey = base64urlDecode(ack.server_ephemeral_pubkey); + const ephDh = deriveSharedSecret(this.clientEphemeralKeypair.secretKey, serverEphemeralPubkey); + const keys = deriveSessionKeys( + this.staticDhKey, + ephDh, + this.clientEphemeralKeypair.publicKey, + serverEphemeralPubkey + ); + this.cipher = new TunnelCipher('client', keys, this.replayTracker); + if (this.handshakeAck) { + clearTimeout(this.handshakeAck.timeoutId); + this.handshakeAck.resolve(); + this.handshakeAck = null; + } + log('[tunnel] handshake ack complete server_eph_len=%d', serverEphemeralPubkey.length); + } + private dispatchEnvelope(envelope: Envelope): void { const { requestId, kind } = envelope; @@ -238,13 +319,13 @@ export class TunnelTransport implements CoreTransport { // -- send ------------------------------------------------------------------ private async sendEnvelope(envelope: Envelope): Promise { - if (!this.sessionKey) throw new Error('[tunnel] no session key — handshake incomplete'); + if (!this.cipher) throw new Error('[tunnel] no session cipher — handshake incomplete'); await this.rateLimiter.consume(); const chunks = chunk(envelope); for (const raw of chunks) { - const encrypted = seal(this.sessionKey, raw); + const encrypted = this.cipher.seal(raw); const frameB64 = base64urlEncode(encrypted); this.socket!.emit('tunnel:frame', { channelId: this.channelId, payload: frameB64 }); } @@ -363,7 +444,14 @@ export class TunnelTransport implements CoreTransport { this.socket?.disconnect(); this.socket = null; this._connectPromise = null; - this.sessionKey = null; + this.staticDhKey = null; + this.clientEphemeralKeypair = null; + this.cipher = null; + if (this.handshakeAck) { + clearTimeout(this.handshakeAck.timeoutId); + this.handshakeAck.reject(new Error('[tunnel] transport closed')); + this.handshakeAck = null; + } } private rejectAllPending(err: Error): void { diff --git a/src/openhuman/devices/bus.rs b/src/openhuman/devices/bus.rs index 3330e4fab..f2647f72b 100644 --- a/src/openhuman/devices/bus.rs +++ b/src/openhuman/devices/bus.rs @@ -11,10 +11,20 @@ use std::sync::{Arc, OnceLock}; use crate::core::event_bus::{publish_global, DomainEvent, EventHandler, SubscriptionHandle}; -use crate::openhuman::devices::rpc::{PEER_STATUS, PENDING_KEYPAIRS, PENDING_SESSIONS}; +use crate::openhuman::devices::crypto::{ + base64url_decode, base64url_encode, derive_session_keys, TunnelCipher, TunnelRole, +}; +use crate::openhuman::devices::rpc::{ + ACTIVE_CIPHERS, PEER_STATUS, PENDING_KEYPAIRS, PENDING_SESSIONS, +}; use crate::openhuman::devices::store; -use crate::openhuman::devices::tunnel_client::{resolve_register_ack, TunnelRegisterResponse}; +use crate::openhuman::devices::tunnel_client::{ + emit_frame, resolve_register_ack, TunnelRegisterResponse, +}; use async_trait::async_trait; +use serde::{Deserialize, Serialize}; +use serde_json::{json, Value}; +use x25519_dalek::{PublicKey, StaticSecret}; static DEVICE_TUNNEL_HANDLE: OnceLock = OnceLock::new(); @@ -92,6 +102,34 @@ impl EventHandler for DeviceTunnelSubscriber { // Handlers // --------------------------------------------------------------------------- +#[derive(Debug)] +struct HandshakePayload { + device_pubkey: String, + client_ephemeral_pubkey: Option, +} + +#[derive(Debug, Deserialize)] +struct JsonHandshakePayload { + device_pubkey: String, + client_ephemeral_pubkey: Option, +} + +#[derive(Debug, Deserialize, Serialize)] +struct TunnelEnvelope { + #[serde(rename = "requestId")] + request_id: String, + kind: String, + seq: u64, + payload: Value, +} + +#[derive(Debug, Deserialize)] +struct TunnelRpcPayload { + method: String, + #[serde(default)] + params: Value, +} + async fn handle_peer_online(channel_id: &str) { log::info!("[devices/bus] peer online channel_id={}", channel_id); PEER_STATUS @@ -120,6 +158,23 @@ async fn handle_tunnel_frame(channel_id: &str, payload_b64: &str) { payload_b64.len() ); + // Decode the outer base64url envelope. + let frame_bytes = match crate::openhuman::devices::crypto::base64url_decode(payload_b64) { + Ok(b) => b, + Err(e) => { + log::warn!( + "[devices/bus] bad base64url in tunnel:frame channel_id={}: {e}", + channel_id + ); + return; + } + }; + + if frame_bytes.first() == Some(&crate::openhuman::devices::crypto::FRAME_VERSION) { + handle_encrypted_rpc_frame(channel_id, &frame_bytes).await; + return; + } + // Look up the pending keypair for this channel. let keypair = { let map = PENDING_KEYPAIRS.lock().unwrap(); @@ -134,18 +189,6 @@ async fn handle_tunnel_frame(channel_id: &str, payload_b64: &str) { return; }; - // Decode the outer base64url envelope. - let frame_bytes = match crate::openhuman::devices::crypto::base64url_decode(payload_b64) { - Ok(b) => b, - Err(e) => { - log::warn!( - "[devices/bus] bad base64url in tunnel:frame channel_id={}: {e}", - channel_id - ); - return; - } - }; - // Wire format for the handshake frame: // // 0x01 || eph_pub(32) || nonce(24) || ciphertext+tag @@ -161,7 +204,7 @@ async fn handle_tunnel_frame(channel_id: &str, payload_b64: &str) { // Fallback: if the frame begins with a printable ASCII character other than // 0x01/0x02, treat the entire payload as a base64url(device_pubkey) string // for backward compat with any pre-Layer-2 devices. - let device_pubkey_b64 = if frame_bytes.first() == Some(&0x01) { + let handshake_payload = if frame_bytes.first() == Some(&0x01) { // Sealed handshake: eph_pub(32) || nonce(24) || ciphertext+tag if frame_bytes.len() < 1 + 32 + 24 + 16 { log::warn!( @@ -207,11 +250,6 @@ async fn handle_tunnel_frame(channel_id: &str, payload_b64: &str) { }; // Decrypt: nonce(24) || ciphertext+tag at offset 33. let inner_frame = &frame_bytes[33..]; - let cipher = crate::openhuman::devices::crypto::TunnelCipher::new(&dh_key); - // Reconstruct frame with version byte 0x01 so TunnelCipher::open can - // validate the version — prepend it back. - let mut framed = vec![0x01u8]; - framed.extend_from_slice(inner_frame); match { // TunnelCipher::open expects version(1)||nonce(24)||ct+tag, but we already // stripped the eph_pub prefix. Reconstruct a plain open call by using @@ -230,7 +268,7 @@ async fn handle_tunnel_frame(channel_id: &str, payload_b64: &str) { } } { Ok(plaintext_bytes) => match String::from_utf8(plaintext_bytes) { - Ok(s) => s.trim().to_string(), + Ok(s) => parse_handshake_payload(&s), Err(_) => { log::warn!( "[devices/bus] decrypted handshake payload is not UTF-8 channel_id={}", @@ -254,7 +292,7 @@ async fn handle_tunnel_frame(channel_id: &str, payload_b64: &str) { channel_id ); match String::from_utf8(frame_bytes) { - Ok(s) => s.trim().to_string(), + Ok(s) => parse_handshake_payload(&s), Err(_) => { log::warn!( "[devices/bus] tunnel:frame payload not valid UTF-8 for channel_id={}", @@ -264,6 +302,7 @@ async fn handle_tunnel_frame(channel_id: &str, payload_b64: &str) { } } }; + let device_pubkey_b64 = handshake_payload.device_pubkey; log::info!( "[devices/bus] handshake frame received channel_id={} device_pubkey_len={}", @@ -272,12 +311,26 @@ async fn handle_tunnel_frame(channel_id: &str, payload_b64: &str) { ); // Derive shared secret — if this fails the device sent a bad pubkey. - if let Err(e) = keypair.derive_shared_secret(&device_pubkey_b64) { - log::error!( - "[devices/bus] X25519 key agreement failed channel_id={}: {e}", - channel_id - ); - return; + let static_dh = match keypair.derive_shared_secret(&device_pubkey_b64) { + Ok(secret) => secret, + Err(e) => { + log::error!( + "[devices/bus] X25519 key agreement failed channel_id={}: {e}", + channel_id + ); + return; + } + }; + + if let Some(client_eph_pubkey) = handshake_payload.client_ephemeral_pubkey { + if let Err(e) = install_v2_cipher_and_ack(channel_id, &static_dh, &client_eph_pubkey).await + { + log::error!( + "[devices/bus] v2 handshake ack failed channel_id={}: {e}", + channel_id + ); + return; + } } // Persist the paired device. @@ -334,6 +387,231 @@ async fn handle_tunnel_frame(channel_id: &str, payload_b64: &str) { } } +fn parse_handshake_payload(raw: &str) -> HandshakePayload { + let trimmed = raw.trim(); + match serde_json::from_str::(trimmed) { + Ok(payload) => HandshakePayload { + device_pubkey: payload.device_pubkey, + client_ephemeral_pubkey: payload.client_ephemeral_pubkey, + }, + Err(_) => HandshakePayload { + device_pubkey: trimmed.to_string(), + client_ephemeral_pubkey: None, + }, + } +} + +async fn install_v2_cipher_and_ack( + channel_id: &str, + static_dh: &[u8; 32], + client_eph_pubkey_b64: &str, +) -> Result<(), String> { + let client_eph_bytes = base64url_decode(client_eph_pubkey_b64) + .map_err(|e| format!("[devices/bus] bad client ephemeral pubkey: {e}"))?; + if client_eph_bytes.len() != 32 { + return Err(format!( + "[devices/bus] client ephemeral pubkey must be 32 bytes, got {}", + client_eph_bytes.len() + )); + } + let client_eph_arr: [u8; 32] = client_eph_bytes + .try_into() + .map_err(|_| "[devices/bus] client ephemeral pubkey slice error".to_string())?; + let client_eph_pub = PublicKey::from(client_eph_arr); + + let server_eph_secret = StaticSecret::from(rand::random::<[u8; 32]>()); + let server_eph_pub = PublicKey::from(&server_eph_secret); + let eph_dh = server_eph_secret.diffie_hellman(&client_eph_pub); + let server_eph_pub_bytes = *server_eph_pub.as_bytes(); + let keys = derive_session_keys( + static_dh, + eph_dh.as_bytes(), + &client_eph_arr, + &server_eph_pub_bytes, + ); + + ACTIVE_CIPHERS.lock().unwrap().insert( + channel_id.to_string(), + Arc::new(std::sync::Mutex::new(TunnelCipher::for_role( + TunnelRole::Server, + &keys, + ))), + ); + + let ack = json!({ + "kind": "handshake_ack", + "server_ephemeral_pubkey": base64url_encode(&server_eph_pub_bytes), + }); + let ack_bytes = serde_json::to_vec(&ack) + .map_err(|e| format!("[devices/bus] handshake ack serialize failed: {e}"))?; + let bootstrap_cipher = TunnelCipher::new(static_dh); + let ack_frame = bootstrap_cipher + .seal(&ack_bytes) + .map_err(|e| format!("[devices/bus] handshake ack seal failed: {e}"))?; + let ack_b64 = base64url_encode(&ack_frame); + emit_frame(channel_id, &ack_b64).await?; + log::info!( + "[devices/bus] v2 tunnel cipher installed channel_id={} server_eph_len={}", + channel_id, + server_eph_pub_bytes.len() + ); + Ok(()) +} + +async fn handle_encrypted_rpc_frame(channel_id: &str, frame_bytes: &[u8]) { + let cipher = { + let map = ACTIVE_CIPHERS.lock().unwrap(); + map.get(channel_id).cloned() + }; + let Some(cipher) = cipher else { + log::warn!( + "[devices/bus] encrypted frame with no active cipher channel_id={}", + channel_id + ); + return; + }; + + let plaintext = { + let mut guard = cipher.lock().unwrap(); + match guard.open(frame_bytes) { + Ok(bytes) => bytes, + Err(e) => { + log::warn!( + "[devices/bus] encrypted frame open failed channel_id={}: {e}", + channel_id + ); + return; + } + } + }; + + let envelope = match serde_json::from_slice::(&plaintext) { + Ok(envelope) => envelope, + Err(e) => { + log::warn!( + "[devices/bus] tunnel envelope parse failed channel_id={}: {e}", + channel_id + ); + return; + } + }; + + if envelope.kind != "request" { + log::debug!( + "[devices/bus] ignoring non-request tunnel envelope kind={} channel_id={}", + envelope.kind, + channel_id + ); + return; + } + + let request = match serde_json::from_value::(envelope.payload) { + Ok(request) => request, + Err(e) => { + emit_tunnel_error( + channel_id, + &cipher, + &envelope.request_id, + format!("invalid tunnel RPC payload: {e}"), + ) + .await; + return; + } + }; + + log::debug!( + "[devices/bus] tunnel RPC request channel_id={} method={} request_id={}", + channel_id, + request.method, + envelope.request_id + ); + + let result = crate::core::jsonrpc::invoke_method( + crate::core::jsonrpc::default_state(), + &request.method, + request.params, + ) + .await; + + match result { + Ok(value) => { + emit_tunnel_response(channel_id, &cipher, &envelope.request_id, "response", value) + .await; + } + Err(message) => { + emit_tunnel_response( + channel_id, + &cipher, + &envelope.request_id, + "error", + Value::String(message), + ) + .await; + } + } +} + +async fn emit_tunnel_error( + channel_id: &str, + cipher: &Arc>, + request_id: &str, + message: String, +) { + emit_tunnel_response( + channel_id, + cipher, + request_id, + "error", + Value::String(message), + ) + .await; +} + +async fn emit_tunnel_response( + channel_id: &str, + cipher: &Arc>, + request_id: &str, + kind: &str, + payload: Value, +) { + let response = TunnelEnvelope { + request_id: request_id.to_string(), + kind: kind.to_string(), + seq: 0, + payload, + }; + let plaintext = match serde_json::to_vec(&response) { + Ok(bytes) => bytes, + Err(e) => { + log::error!( + "[devices/bus] tunnel response serialize failed channel_id={}: {e}", + channel_id + ); + return; + } + }; + let encrypted = { + let guard = cipher.lock().unwrap(); + match guard.seal(&plaintext) { + Ok(frame) => frame, + Err(e) => { + log::error!( + "[devices/bus] tunnel response seal failed channel_id={}: {e}", + channel_id + ); + return; + } + } + }; + let payload_b64 = base64url_encode(&encrypted); + if let Err(e) = emit_frame(channel_id, &payload_b64).await { + log::error!( + "[devices/bus] tunnel response emit failed channel_id={}: {e}", + channel_id + ); + } +} + /// Resolve the pending `tunnel:register` ack in `tunnel_client`. fn handle_registered(channel_id: &str, pairing_token: &str, session_token: &str) { log::debug!( diff --git a/src/openhuman/devices/rpc.rs b/src/openhuman/devices/rpc.rs index ffb86b8f2..f4c5e9757 100644 --- a/src/openhuman/devices/rpc.rs +++ b/src/openhuman/devices/rpc.rs @@ -16,7 +16,9 @@ use std::sync::{Arc, Mutex}; use chrono::Utc; use crate::openhuman::config::Config; -use crate::openhuman::devices::crypto::{base64url_decode, base64url_encode, DeviceKeypair}; +use crate::openhuman::devices::crypto::{ + base64url_decode, base64url_encode, DeviceKeypair, TunnelCipher, +}; use crate::openhuman::devices::store; use crate::openhuman::devices::tunnel_client; use crate::openhuman::devices::types::{ @@ -49,6 +51,11 @@ pub(crate) static PENDING_SESSIONS: once_cell::sync::Lazy>> = once_cell::sync::Lazy::new(|| Mutex::new(HashMap::new())); +/// Active post-handshake tunnel ciphers (keyed by channel_id). +pub(crate) static ACTIVE_CIPHERS: once_cell::sync::Lazy< + Mutex>>>, +> = once_cell::sync::Lazy::new(|| Mutex::new(HashMap::new())); + // --------------------------------------------------------------------------- // create_pairing // --------------------------------------------------------------------------- @@ -210,6 +217,7 @@ pub async fn devices_revoke( PENDING_SESSIONS.lock().unwrap().remove(&channel_id); PEER_STATUS.lock().unwrap().remove(&channel_id); PERSISTED_KEYPAIRS.lock().unwrap().remove(&channel_id); + ACTIVE_CIPHERS.lock().unwrap().remove(&channel_id); // Publish DeviceRevoked so UI and other subscribers are notified. crate::core::event_bus::publish_global(crate::core::event_bus::DomainEvent::DeviceRevoked {