diff --git a/CURRENT_STATE.md b/CURRENT_STATE.md index ecb9bc7..e61ca50 100644 --- a/CURRENT_STATE.md +++ b/CURRENT_STATE.md @@ -176,6 +176,6 @@ live timeline with OS notifications on critical failures. | `docs` (docs, ADRs) | ✅ done for S0 | | `ocpp` (engine) | ✅ S2 + ingestion (#29) + reports (#41) + anonymize (#42) + diff (#43) + replay core (#44); O(n) detection pending (#36) | | `ui` (native views) | ✅ S3 inspector (#27–#32) + replay transport (#44) | -| `capture` (live proxy) | 🔨 S5 in progress: WS transport (#54) + frame decode (#55) + pump pipeline (#56 pt1) | +| `capture` (live proxy) | 🔨 S5 in progress: WS transport (#54) + frame decode (#55) + MITM proxy (#56) | | `cli` (headless) | ✅ inspect/report/diff/anonymize/ci/scenario (#45) | | `conformance` | ✅ done for S2 (15/15, `contract-v1`) | diff --git a/docs/adr/0010-live-proxy-transport-concurrency.md b/docs/adr/0010-live-proxy-transport-concurrency.md new file mode 100644 index 0000000..b62431c --- /dev/null +++ b/docs/adr/0010-live-proxy-transport-concurrency.md @@ -0,0 +1,66 @@ +# ADR-0010 — Live-proxy transport & concurrency + +- **Status:** Accepted +- **Date:** 2026-07-12 + +## Context + +The live-capture proxy (#56) sits between a charge point and its CSMS and must, +concurrently and for the life of a session: + +- accept the downstream CP connection and dial the upstream CSMS, +- relay every WebSocket frame in **both** directions at once, +- decode a copy of each frame, record it, and run detection. + +Two questions had no obvious default in Zig 0.16: + +1. **Sockets.** The classic `std.net` is gone; networking moved into the async + `std.Io` interface (`std.Io.net`), reached through the `io` the runtime and + the CLI already thread (`init.io`). +2. **Concurrency.** Both pump directions block on reads, so they must make + progress independently. `std.Io` offers `io.async` (which *permits* + single-threaded blocking execution) and `io.concurrent` (which *guarantees* + independent progress). + +## Decision + +- **Transport: `std.Io.net`, threaded through `io`.** `run` does + `IpAddress.listen` → `Server.accept` (CP) and `IpAddress.connect` (CSMS), then + drives the relay over each stream's reader/writer. +- **Concurrency: `io.concurrent`, not `io.async`.** One direction runs + concurrently and the other inline; when the inline side ends (its peer closed), + the concurrent one is cancelled so the session tears down cleanly. Two blocking + pumps under `io.async` could deadlock on a single-threaded backend; + `io.concurrent` rules that out. +- **Shared state guarded by `std.Io.Mutex`.** Both directions feed one `Sink` + (event list, sequence counter, recorder, detection). Ingest is serialized with + an `Io.Mutex` (cooperates with the scheduler on any backend); `detect`/`count` + are called only when pumping is quiesced. +- **Verbatim relay.** Frames are forwarded byte-for-byte. The masking polarity is + preserved hop-for-hop (CP→CSMS stays client→server/masked; CSMS→CP stays + server→client/unmasked), so raw bytes are valid on the second hop; the tap + decodes a *copy*. +- **A testable seam.** `relayStreams` is generic over `*Io.Reader`/`*Io.Writer`, + so the whole relay — MITM handshake, concurrent pump, record, detect — is + driven in-memory under test (with a `std.Io.Threaded` io for real concurrency + and the `Io.Mutex`), while `run` supplies socket reader/writers. No OS + socketpair is used: `socketpair(2)` here supports only `AF_INET`, which the OS + rejects, and no `AF_UNIX` pair is exposed. +- **Wall-clock via `TimeSource`.** Time in 0.16 is read from `io` + (`Clock.now(.real, io)`), so a plain function can't supply it. A small + `TimeSource` union provides `fixed` (deterministic, tests), `wall` (the io + clock, live), or `none`. +- **Deterministic session key.** The upstream handshake key is derived from the + CP's own key (not a secret — the peer only echoes it), which also lets tests + reproduce the proxy's key and precompute the matching accept token. + +## Consequences + +- The relay logic is fully unit-tested in-memory with genuine concurrency; the + thin socket glue in `run` is compiled (referenced from a test) and exercised + end-to-end when the `studio capture` CLI wires it (#57). +- No new dependency; `std.Io` covers sockets, concurrency, mutex, and clock. +- Single session per `run` for now (accept one CP, relay, return). A multi-session + accept loop is a later refinement, behind the same relay core. +- TLS (secure profiles) remains post-0.5, behind the same transport boundary + (ADR-0008). diff --git a/docs/adr/README.md b/docs/adr/README.md index 0bf8df9..051e72d 100644 --- a/docs/adr/README.md +++ b/docs/adr/README.md @@ -18,3 +18,4 @@ supersedes the old one rather than editing history. | [0006](0006-inspector-builder-view.md) | Inspector as a Zig builder view (not `.native` markup) | Accepted | | [0007](0007-trusted-ingestion.md) | Trusted-ingestion limits for user-opened traces | Accepted | | [0008](0008-websocket-transport.md) | WebSocket transport: a hand-rolled RFC 6455 subset | Accepted | +| [0010](0010-live-proxy-transport-concurrency.md) | Live-proxy transport & concurrency | Accepted | diff --git a/src/capture/proxy.zig b/src/capture/proxy.zig index 36b799e..d1096ed 100644 --- a/src/capture/proxy.zig +++ b/src/capture/proxy.zig @@ -24,6 +24,7 @@ const ocpp = @import("../ocpp/ocpp.zig"); const ws = @import("ws.zig"); const decode = @import("decode.zig"); +const net = std.Io.net; const types = ocpp.types; const detection = ocpp.detection; const timeline = ocpp.timeline; @@ -39,12 +40,27 @@ pub const max_frame_bytes: usize = 1 * 1024 * 1024; // 1 MiB /// Default per-session cap on retained decoded events. pub const default_max_events: usize = 1_000_000; -/// Wall-clock receipt time in epoch milliseconds — the default `now` for a live -/// session. Always ≥ 10^12, so recordings survive the offline normalizer's -/// seconds/ms threshold unchanged (see `decode`). -pub fn wallClockMs() ?i64 { - return std.time.milliTimestamp(); -} +/// Where a live session's receipt timestamps come from. Wall-clock time in Zig +/// 0.16 is read through `io`, so a plain function can't supply it — this small +/// source does, while staying deterministic under test. +pub const TimeSource = union(enum) { + /// A fixed epoch-ms value — deterministic, for tests. + fixed: i64, + /// The real wall clock, read from `io` — live capture. Real wall-clock is + /// always ≥ 10^12 ms, so recordings survive the offline normalizer's + /// seconds/ms threshold unchanged (see `decode`). + wall: std.Io, + /// No clock; events are recorded without a timestamp. + none, + + pub fn nowMs(self: TimeSource) ?i64 { + return switch (self) { + .fixed => |v| v, + .none => null, + .wall => |io| @intCast(@divTrunc(std.Io.Clock.now(.real, io).nanoseconds, 1_000_000)), + }; + } +}; // --------------------------------------------------------------------- sink @@ -61,6 +77,12 @@ pub const Sink = struct { /// Optional JSONL recorder. Each ingested message appends one line. record: ?*std.Io.Writer = null, max_events: usize = default_max_events, + /// Guards `ingest` when both pump directions feed this sink concurrently + /// (the real-socket path). Null for the single-writer in-memory tests. + /// `detect`/`count` are not guarded — call them when pumping is quiesced. + mutex: ?*std.Io.Mutex = null, + /// The `io` the guard cooperates with; set together with `mutex`. + io: ?std.Io = null, pub fn deinit(self: *Sink) void { self.events.deinit(self.gpa); @@ -72,6 +94,8 @@ pub const Sink = struct { /// message that fails to decode is counted in `dropped` and skipped — one /// bad frame never aborts the session. pub fn ingest(self: *Sink, text: []const u8, origin: Direction, received_ms: ?i64) !void { + if (self.mutex) |m| m.lockUncancelable(self.io.?); + defer if (self.mutex) |m| m.unlock(self.io.?); if (self.events.items.len >= self.max_events) return; const seq = self.seq + 1; @@ -167,7 +191,7 @@ pub fn pumpDirection( dst: *std.Io.Writer, origin: Direction, sink: *Sink, - now: *const fn () ?i64, + time: TimeSource, ) !void { // Client→server frames are masked; server→client are not (RFC 6455 §5.1). const expect_masked = origin == .cs_to_csms; @@ -190,18 +214,166 @@ pub fn pumpDirection( continue; // ping / pong: relayed, nothing to tap } if (try assembler.push(gpa, dec.frame)) |msg| { - try sink.ingest(msg.payload, origin, now()); + try sink.ingest(msg.payload, origin, time.nowMs()); } } } +// ------------------------------------------------------------- sockets + +/// Default WebSocket subprotocol offered upstream. OCPP-J stations use `ocpp1.6`. +pub const default_subprotocol: []const u8 = "ocpp1.6"; + +/// Network read buffer, sized to hold the largest relayable frame since +/// `readFrameBytes` peeks a whole frame at once. +const read_buffer_bytes = max_frame_bytes; +/// Outgoing buffer; writes larger than this drain straight through. +const write_buffer_bytes = 64 * 1024; + +/// Read one HTTP head (request/status line + headers) up to and including the +/// terminating blank line. Bounded by `ws.max_handshake_bytes`. +fn readHead(gpa: std.mem.Allocator, r: *std.Io.Reader) ![]u8 { + var head: std.ArrayList(u8) = .empty; + errdefer head.deinit(gpa); + while (true) { + const line = try r.takeDelimiterInclusive('\n'); + try head.appendSlice(gpa, line); + if (head.items.len > ws.max_handshake_bytes) return error.HandshakeTooLarge; + if (std.mem.eql(u8, line, "\r\n") or std.mem.eql(u8, line, "\n")) break; + } + return head.toOwnedSlice(gpa); +} + +/// The request-target from an HTTP request line (`GET HTTP/1.1`). +fn requestPath(req: []const u8) ?[]const u8 { + const eol = std.mem.indexOf(u8, req, "\r\n") orelse return null; + var it = std.mem.tokenizeScalar(u8, req[0..eol], ' '); + _ = it.next() orelse return null; // method + return it.next(); +} + +/// A deterministic per-session client nonce, derived from the CP's own key. Not +/// a secret (see `relayHandshake`); deterministic so tests can reproduce the +/// proxy's upstream key and precompute the matching accept token. +fn sessionNonce(cp_key: []const u8) [16]u8 { + var nonce: [16]u8 = undefined; + var prng = std.Random.DefaultPrng.init(std.hash.Wyhash.hash(0, cp_key)); + prng.random().bytes(&nonce); + return nonce; +} + +/// The MITM opening handshake: two independent RFC 6455 handshakes, one per peer. +/// Studio is the *server* to the CP (accepts its key) and the *client* to the +/// CSMS (sends its own key, validates the reply), mirroring the CP's request path +/// upstream. Reuses the #54 codec's both-halves API. +fn relayHandshake( + gpa: std.mem.Allocator, + cp_r: *std.Io.Reader, + cp_w: *std.Io.Writer, + csms_r: *std.Io.Reader, + csms_w: *std.Io.Writer, + upstream_host: []const u8, + subprotocol: ?[]const u8, +) !void { + const cp_req = try readHead(gpa, cp_r); + defer gpa.free(cp_req); + const cp_hs = try ws.parseClientHandshake(cp_req); + const path = requestPath(cp_req) orelse "/"; + + // The client-handshake key is fresh per session but not a secret (the peer + // only echoes it in the accept token); it is derived deterministically from + // the CP's own key (`sessionNonce`), which also lets tests reproduce it. + const our_key = ws.clientKey(sessionNonce(cp_hs.key)); + const upstream_req = try ws.writeClientHandshake(gpa, upstream_host, path, our_key, subprotocol); + defer gpa.free(upstream_req); + try csms_w.writeAll(upstream_req); + try csms_w.flush(); + + const csms_resp = try readHead(gpa, csms_r); + defer gpa.free(csms_resp); + try ws.parseServerHandshake(csms_resp, our_key); + + const accept = try ws.writeServerAccept(gpa, cp_hs.key); + defer gpa.free(accept); + try cp_w.writeAll(accept); + try cp_w.flush(); +} + +/// Relay a full session over two already-connected streams: the MITM handshake, +/// then both frame-pump directions concurrently. Each direction blocks on reads, +/// so they must truly run in parallel — `io.concurrent`, not `io.async`. The +/// shared `sink` is mutex-guarded for the duration; when one side closes, the +/// other direction is cancelled so the session tears down cleanly. +pub fn relayStreams( + io: std.Io, + gpa: std.mem.Allocator, + cp_r: *std.Io.Reader, + cp_w: *std.Io.Writer, + cs_r: *std.Io.Reader, + cs_w: *std.Io.Writer, + upstream_host: []const u8, + sink: *Sink, + time: TimeSource, +) !void { + try relayHandshake(gpa, cp_r, cp_w, cs_r, cs_w, upstream_host, default_subprotocol); + + var mutex: std.Io.Mutex = .init; + sink.mutex = &mutex; + sink.io = io; + defer { + sink.mutex = null; + sink.io = null; + } + + // CP→CSMS runs concurrently; CSMS→CP runs inline. When the inline side ends + // (its peer closed), cancel unblocks the other's pending read. + var up = try io.concurrent(pumpDirection, .{ gpa, cp_r, cs_w, Direction.cs_to_csms, sink, time }); + pumpDirection(gpa, cs_r, cp_w, Direction.csms_to_cs, sink, time) catch {}; + up.cancel(io) catch {}; +} + +/// Listen for one downstream charge point, dial the upstream CSMS, and relay the +/// session over real sockets. Single-session: returns when the session ends. +/// `listen_addr` / `upstream_addr` are resolved addresses; `upstream_host` is the +/// Host header sent upstream. The relay logic itself is `relayStreams`, driven +/// here over socket reader/writers (and in-memory ones under test). +pub fn run( + io: std.Io, + gpa: std.mem.Allocator, + listen_addr: net.IpAddress, + upstream_addr: net.IpAddress, + upstream_host: []const u8, + sink: *Sink, + time: TimeSource, +) !void { + var server = try listen_addr.listen(io, .{}); + defer server.deinit(io); + const cp = try server.accept(io); + defer cp.close(io); + const csms = try upstream_addr.connect(io, .{ .mode = .stream }); + defer csms.close(io); + + const rbuf_cp = try gpa.alloc(u8, read_buffer_bytes); + defer gpa.free(rbuf_cp); + const rbuf_cs = try gpa.alloc(u8, read_buffer_bytes); + defer gpa.free(rbuf_cs); + const wbuf_cp = try gpa.alloc(u8, write_buffer_bytes); + defer gpa.free(wbuf_cp); + const wbuf_cs = try gpa.alloc(u8, write_buffer_bytes); + defer gpa.free(wbuf_cs); + + var cp_reader = cp.reader(io, rbuf_cp); + var cp_writer = cp.writer(io, wbuf_cp); + var cs_reader = csms.reader(io, rbuf_cs); + var cs_writer = csms.writer(io, wbuf_cs); + try relayStreams(io, gpa, &cp_reader.interface, &cp_writer.interface, &cs_reader.interface, &cs_writer.interface, upstream_host, sink, time); +} + // ------------------------------------------------------------------- tests const testing = std.testing; -fn testNow() ?i64 { - return 1_705_312_800_000; // fixed epoch-ms ≥ 10^12 -} +const test_ms: i64 = 1_705_312_800_000; // fixed epoch-ms ≥ 10^12 test "pump relays frames verbatim and taps decoded messages" { const gpa = testing.allocator; @@ -229,7 +401,7 @@ test "pump relays frames verbatim and taps decoded messages" { var sink = Sink{ .gpa = arena_state.allocator(), .record = &rec }; defer sink.deinit(); - try pumpDirection(gpa, &src, &dst, .cs_to_csms, &sink, testNow); + try pumpDirection(gpa, &src, &dst, .cs_to_csms, &sink, .{ .fixed = test_ms }); // Relay is byte-exact (still masked — Studio is transparent). try testing.expectEqualSlices(u8, wire.items, dst.buffered()); @@ -281,7 +453,7 @@ test "pump reassembles a fragmented message and stops on close" { var sink = Sink{ .gpa = arena_state.allocator() }; defer sink.deinit(); - try pumpDirection(gpa, &src, &dst, .csms_to_cs, &sink, testNow); + try pumpDirection(gpa, &src, &dst, .csms_to_cs, &sink, .{ .fixed = test_ms }); // Exactly one reassembled message; the post-close frame was not tapped. try testing.expectEqual(@as(usize, 1), sink.events.items.len); @@ -311,7 +483,7 @@ test "sink detection matches an offline pass over the recording" { for (msgs, 0..) |m, i| { const origin: Direction = if (i % 2 == 0) .cs_to_csms else .csms_to_cs; - try sink.ingest(m, origin, testNow()); + try sink.ingest(m, origin, test_ms); } // Streaming detection over the live events… @@ -327,3 +499,86 @@ test "sink detection matches an offline pass over the recording" { try testing.expectEqual(off.code, lf.code); } } + +test "in-process relay: MITM handshake, concurrent pump, record re-parses and re-detects identically" { + const gpa = testing.allocator; + var threaded = std.Io.Threaded.init(gpa, .{}); + defer threaded.deinit(); + const io = threaded.io(); + + var arena_state = std.heap.ArenaAllocator.init(gpa); + defer arena_state.deinit(); + const arena = arena_state.allocator(); + + const cp_key = ws.clientKey([_]u8{ 3, 1, 4, 1, 5, 9, 2, 6, 5, 3, 5, 8, 9, 7, 9, 3 }); + const mask = [4]u8{ 0xa1, 0xb2, 0xc3, 0xd4 }; + + // The CP's wire: opening request, two masked Calls, a close. + var cp_wire: std.ArrayList(u8) = .empty; + defer cp_wire.deinit(gpa); + try cp_wire.appendSlice(gpa, try ws.writeClientHandshake(arena, "studio.local", "/ocpp/CP1", cp_key, "ocpp1.6")); + inline for (.{ "[2,\"m1\",\"BootNotification\",{\"chargePointVendor\":\"Acme\"}]", "[2,\"m2\",\"Heartbeat\",{}]" }) |call| { + try cp_wire.appendSlice(gpa, try ws.encodeFrameAlloc(arena, true, .text, call, mask)); + } + try cp_wire.appendSlice(gpa, try ws.encodeFrameAlloc(arena, true, .close, &[_]u8{ 0x03, 0xe8 }, mask)); + + // The CSMS's wire: a 101 carrying accept() for the proxy's *deterministic* + // upstream key (sessionNonce), two unmasked Results, a close. + const our_key = ws.clientKey(sessionNonce(&cp_key)); + const our_accept = ws.acceptToken(&our_key); + var cs_wire: std.ArrayList(u8) = .empty; + defer cs_wire.deinit(gpa); + try cs_wire.appendSlice(gpa, try std.fmt.allocPrint(arena, "HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: {s}\r\n\r\n", .{our_accept[0..]})); + inline for (.{ "[3,\"m1\",{\"status\":\"Accepted\"}]", "[3,\"m2\",{}]" }) |res| { + try cs_wire.appendSlice(gpa, try ws.encodeFrameAlloc(arena, true, .text, res, null)); + } + try cs_wire.appendSlice(gpa, try ws.encodeFrameAlloc(arena, true, .close, &[_]u8{ 0x03, 0xe8 }, null)); + + var cp_r = std.Io.Reader.fixed(cp_wire.items); + var cs_r = std.Io.Reader.fixed(cs_wire.items); + var cp_out: [8192]u8 = undefined; + var cs_out: [8192]u8 = undefined; + var cp_w = std.Io.Writer.fixed(&cp_out); + var cs_w = std.Io.Writer.fixed(&cs_out); + + var rec_buf: [16384]u8 = undefined; + var rec = std.Io.Writer.fixed(&rec_buf); + var sink = Sink{ .gpa = arena, .record = &rec }; + defer sink.deinit(); + + try relayStreams(io, arena, &cp_r, &cp_w, &cs_r, &cs_w, "csms.local", &sink, .{ .fixed = test_ms }); + + // Relay correctness: the CSMS-facing side carries the proxy's client handshake + // (mirroring the CP's path); the CP-facing side got a 101 accept. + try testing.expect(std.mem.indexOf(u8, cs_w.buffered(), "GET /ocpp/CP1 HTTP/1.1") != null); + try testing.expect(std.mem.startsWith(u8, cp_w.buffered(), "HTTP/1.1 101")); + + // The tap captured all four messages; the recording re-parses to four events. + try testing.expectEqual(@as(usize, 4), sink.count()); + const reparsed = try ocpp.parser.parseTrace(arena, rec.buffered()); + try testing.expectEqual(@as(usize, 4), reparsed.events.len); + + // Streaming detection equals a cold offline pass over the recording. + const live = try sink.detect(arena); + const sessions = try timeline.buildSessionTimeline(arena, reparsed.events); + const offline = try detection.detectFailures(arena, reparsed.events, sessions); + try testing.expectEqual(offline.len, live.len); + for (live, offline) |lf, off| try testing.expectEqual(off.code, lf.code); +} + +test "run() socket wiring is analyzed" { + // `run` is only reached via itself (the CLI wires it in #57), so reference it + // to force Zig to compile its socket accept/dial path — lazy analysis would + // otherwise skip it in both the test and the app build. + _ = &run; +} + +test "TimeSource.wall reads a plausible epoch-ms from the io clock" { + var threaded = std.Io.Threaded.init(testing.allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + const ms = (TimeSource{ .wall = io }).nowMs().?; + try testing.expect(ms > 1_600_000_000_000); // after 2020-09 + try testing.expectEqual(@as(?i64, null), (TimeSource{ .none = {} }).nowMs()); + try testing.expectEqual(@as(?i64, 42), (TimeSource{ .fixed = 42 }).nowMs()); +}