From 0b712962eb5da0df4f15c8b4da31e67c01ed3de7 Mon Sep 17 00:00:00 2001 From: sepehr-safari Date: Sun, 12 Jul 2026 16:26:17 +0300 Subject: [PATCH] =?UTF-8?q?feat(cli):=20live-capture=20subcommand=20?= =?UTF-8?q?=E2=80=94=20studio=20capture=20(--ndjson)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Wires the S5 flagship proxy into the headless CLI as `studio capture` — the second face's live command. - maybeRun dispatches `capture`; cmdCapture parses --listen host:port / --upstream ws://host:port / --ndjson, resolves the addresses, and calls proxy.run(io, ..., .{ .wall = init.io }) for one CP<->CSMS session. - --ndjson streams each captured event as a JSONL line to stdout (redirect to save a trace); otherwise a one-line summary is printed. The Sink now flushes each record so the stream is live (a no-op for the fixed-buffer test writers). - A pure parseCaptureArgs (unit-tested) does the parsing; wss:// is rejected (TLS is post-0.5), IP addresses only for now (hostname resolution deferred). - docs/cli-parity.md gains the Studio-only `capture` row + notes. Wiring run() into the CLI also brings its socket path into the app build's analysis (it was previously test-only). Verified on the built binary: usage / error / help paths, no regression on existing commands, and run() binds its listen socket and waits on accept. Closes #57. --- CURRENT_STATE.md | 2 +- docs/cli-parity.md | 8 +++ src/capture/proxy.zig | 7 +- src/cli.zig | 155 ++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 170 insertions(+), 2 deletions(-) diff --git a/CURRENT_STATE.md b/CURRENT_STATE.md index e61ca50..179809f 100644 --- a/CURRENT_STATE.md +++ b/CURRENT_STATE.md @@ -177,5 +177,5 @@ live timeline with OS notifications on critical failures. | `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) + MITM proxy (#56) | -| `cli` (headless) | ✅ inspect/report/diff/anonymize/ci/scenario (#45) | +| `cli` (headless) | ✅ inspect/report/diff/anonymize/ci/scenario (#45) + live `capture` (#57) | | `conformance` | ✅ done for S2 (15/15, `contract-v1`) | diff --git a/docs/cli-parity.md b/docs/cli-parity.md index 794364d..7f70e99 100644 --- a/docs/cli-parity.md +++ b/docs/cli-parity.md @@ -17,6 +17,7 @@ intentional difference. | `scenario list` | `studio scenario list` | ✅ Full. | | `scenario run ` | `studio scenario run ` | ✅ Built-in scenarios. | | `scenario run --file ` | — | ⚠️ Deferred — Studio runs only the built-in contract scenarios. | +| — | `studio capture --listen H:P --upstream ws://H:P [--ndjson]` | ➕ **Studio-only** — a live WebSocket MITM proxy; the toolkit is offline by charter. Relays a CP↔CSMS session, decodes + records it, runs detection. `--ndjson` streams each captured event as a JSONL line to stdout (redirect to save a trace). | ## Intentional differences @@ -31,6 +32,13 @@ intentional difference. - **One binary, two faces.** The CLI shares the exact pure engine the GUI uses (`src/ocpp/`); there is no separate code path to drift. A bare trace path opens the GUI: `studio path/to/trace.json`. +- **`capture` is Studio-only.** The toolkit is offline (parse / detect / report) + by charter; live capture is the territory Studio was built to own (ADR-0001). + It's a plaintext `ws://` MITM proxy today — `wss://` (TLS) is post-0.5 + (ADR-0008), hostname resolution is a follow-up (IP addresses for now), and it + handles one session per invocation. Without `--ndjson` it prints a one-line + summary; with `--ndjson`, stdout is the trace stream and the summary goes to + stderr. ## Exit codes diff --git a/src/capture/proxy.zig b/src/capture/proxy.zig index d1096ed..f128d70 100644 --- a/src/capture/proxy.zig +++ b/src/capture/proxy.zig @@ -105,7 +105,12 @@ pub const Sink = struct { }; self.seq = seq; try self.events.append(self.gpa, ev); - if (self.record) |w| try writeRecord(w, received_ms, origin, text); + if (self.record) |w| { + try writeRecord(w, received_ms, origin, text); + // Flush each record so `--ndjson` streams live (a no-op for the + // fixed-buffer writers used in tests). + w.flush() catch {}; + } } /// Run detection over everything captured so far. Streaming callers invoke diff --git a/src/cli.zig b/src/cli.zig index 3d2aa10..c6615a5 100644 --- a/src/cli.zig +++ b/src/cli.zig @@ -25,6 +25,8 @@ const report = ocpp.report; const anonymize = ocpp.anonymize; const diff = ocpp.diff; const conformance = ocpp.conformance; +const proxy = @import("capture/capture.zig").proxy; +const net = std.Io.net; const Allocator = std.mem.Allocator; @@ -55,6 +57,7 @@ pub fn maybeRun(init: std.process.Init) ?u8 { if (eql(cmd, "anonymize")) return cmdAnonymize(gpa, init, rest); if (eql(cmd, "ci")) return cmdCi(gpa, init.io); if (eql(cmd, "scenario")) return cmdScenario(gpa, init.io, rest); + if (eql(cmd, "capture")) return cmdCapture(gpa, init, rest); if (eql(cmd, "help") or eql(cmd, "--help") or eql(cmd, "-h")) { _ = emit(init.io, help_text); return 0; @@ -73,6 +76,8 @@ const help_text = \\ anonymize Strip sensitive fields (stdout). \\ ci Run the conformance scenarios; exit 0/1. \\ scenario list | run List or run a conformance scenario. + \\ capture --listen H:P --upstream ws://H:P [--ndjson] + \\ Live WS proxy: relay + record a CP<->CSMS session. \\ \\With no command, a trace path opens the GUI: studio path/to/trace.json \\ @@ -339,6 +344,125 @@ fn cmdScenario(gpa: Allocator, io: std.Io, args: []const []const u8) u8 { return usageErr("scenario: expected 'list' or 'run '"); } +const CaptureOptions = struct { + listen_host: []const u8, + listen_port: u16, + upstream_host: []const u8, + upstream_port: u16, + /// `host:port` used for the upstream `Host` header. + upstream_authority: []const u8, + ndjson: bool, +}; + +const CaptureArgError = error{ MissingListen, MissingUpstream, BadListen, BadUpstream, TlsUnsupported, Unexpected }; + +const HostPort = struct { host: []const u8, port: u16 }; + +/// Split `host:port` on the last colon. An empty host uses `default_host`; a +/// bare value with no `:port` uses `default_port` (null → a port is required). +fn splitHostPort(s: []const u8, default_host: []const u8, default_port: ?u16) ?HostPort { + if (std.mem.lastIndexOfScalar(u8, s, ':')) |idx| { + const host = if (idx == 0) default_host else s[0..idx]; + const port = std.fmt.parseInt(u16, s[idx + 1 ..], 10) catch return null; + return .{ .host = host, .port = port }; + } + if (default_port) |dp| return .{ .host = s, .port = dp }; + return null; +} + +/// Parse a `ws://host:port/path` upstream: scheme and path are optional and the +/// path is ignored (the proxy mirrors the CP's request path). `wss://` is +/// rejected — TLS is post-0.5 (ADR-0008). +fn parseUpstream(s: []const u8) CaptureArgError!HostPort { + var rest = s; + if (std.mem.startsWith(u8, rest, "wss://")) return error.TlsUnsupported; + if (std.mem.startsWith(u8, rest, "ws://")) rest = rest["ws://".len..]; + const authority = if (std.mem.indexOfScalar(u8, rest, '/')) |slash| rest[0..slash] else rest; + return splitHostPort(authority, "127.0.0.1", 80) orelse error.BadUpstream; +} + +/// Pure argument parser for `capture` — unit-tested, no I/O. +fn parseCaptureArgs(args: []const []const u8) CaptureArgError!CaptureOptions { + var listen: ?[]const u8 = null; + var upstream: ?[]const u8 = null; + var ndjson = false; + var i: usize = 0; + while (i < args.len) : (i += 1) { + const arg = args[i]; + if (eql(arg, "--listen")) { + i += 1; + if (i >= args.len) return error.BadListen; + listen = args[i]; + } else if (eql(arg, "--upstream")) { + i += 1; + if (i >= args.len) return error.BadUpstream; + upstream = args[i]; + } else if (eql(arg, "--ndjson")) { + ndjson = true; + } else return error.Unexpected; + } + const l = listen orelse return error.MissingListen; + const u = upstream orelse return error.MissingUpstream; + const la = splitHostPort(l, "127.0.0.1", null) orelse return error.BadListen; + const authority = if (std.mem.startsWith(u8, u, "ws://")) u["ws://".len..] else u; + const ua = try parseUpstream(u); + return .{ + .listen_host = la.host, + .listen_port = la.port, + .upstream_host = ua.host, + .upstream_port = ua.port, + .upstream_authority = if (std.mem.indexOfScalar(u8, authority, '/')) |slash| authority[0..slash] else authority, + .ndjson = ndjson, + }; +} + +/// `capture` — a live WebSocket MITM proxy: relay a CP<->CSMS session, decode and +/// record it, and run detection. One session, then exit. With `--ndjson`, each +/// captured event streams to stdout as a JSONL line (redirect to save a trace — +/// `studio capture … --ndjson > session.jsonl`); otherwise a summary is printed. +fn cmdCapture(gpa: Allocator, init: std.process.Init, args: []const []const u8) u8 { + const opts = parseCaptureArgs(args) catch |e| return usageErr(switch (e) { + error.MissingListen => "capture: missing --listen ", + error.MissingUpstream => "capture: missing --upstream ws://", + error.BadListen => "capture: invalid --listen (want host:port)", + error.BadUpstream => "capture: invalid --upstream (want ws://host:port)", + error.TlsUnsupported => "capture: wss:// (TLS) is not supported yet", + error.Unexpected => "capture: unexpected argument", + }); + + const io = init.io; + const listen_addr = net.IpAddress.parse(opts.listen_host, opts.listen_port) catch + return usageErr("capture: --listen host must be an IP address"); + const upstream_addr = net.IpAddress.parse(opts.upstream_host, opts.upstream_port) catch + return usageErr("capture: --upstream host must be an IP address"); + + var arena = std.heap.ArenaAllocator.init(gpa); + defer arena.deinit(); + const a = arena.allocator(); + + var stdout_buf: [64 * 1024]u8 = undefined; + var fw = std.Io.File.stdout().writer(io, &stdout_buf); + const w = &fw.interface; + + var sink = proxy.Sink{ .gpa = a }; + defer sink.deinit(); + if (opts.ndjson) sink.record = w; + + proxy.run(io, a, listen_addr, upstream_addr, opts.upstream_authority, &sink, .{ .wall = io }) catch |e| + return renderErr("capture", e); + if (opts.ndjson) w.flush() catch return 1; + + // Session summary. With --ndjson stdout is the trace stream, so it goes to + // stderr; otherwise to stdout. + const failures: []const types.Failure = sink.detect(a) catch &.{}; + if (opts.ndjson) { + std.debug.print("captured {d} events, {d} failures\n", .{ sink.count(), failures.len }); + return 0; + } + const summary = std.fmt.allocPrint(a, "Captured {d} events, {d} failures.\n", .{ sink.count(), failures.len }) catch return 1; + return emit(io, summary); +} + // --------------------------------------------------------------------------- // Small helpers // --------------------------------------------------------------------------- @@ -469,3 +593,34 @@ test "runNamed distinguishes a known scenario from an unknown one" { var w2 = std.Io.Writer.fixed(&out2); try testing.expectEqual(@as(?bool, null), try conformance.runNamed(a, &w2, "does-not-exist")); } + +test "parseCaptureArgs parses listen, upstream, and ndjson" { + const args = [_][]const u8{ "--listen", "127.0.0.1:8080", "--upstream", "ws://10.0.0.5:9000/ocpp", "--ndjson" }; + const opts = try parseCaptureArgs(&args); + try testing.expectEqualStrings("127.0.0.1", opts.listen_host); + try testing.expectEqual(@as(u16, 8080), opts.listen_port); + try testing.expectEqualStrings("10.0.0.5", opts.upstream_host); + try testing.expectEqual(@as(u16, 9000), opts.upstream_port); + try testing.expectEqualStrings("10.0.0.5:9000", opts.upstream_authority); + try testing.expect(opts.ndjson); +} + +test "parseCaptureArgs: defaults, missing args, and rejections" { + // Empty host defaults to loopback; upstream without a port defaults to 80. + const ok = [_][]const u8{ "--listen", ":8080", "--upstream", "ws://10.0.0.5" }; + const opts = try parseCaptureArgs(&ok); + try testing.expectEqualStrings("127.0.0.1", opts.listen_host); + try testing.expectEqual(@as(u16, 80), opts.upstream_port); + try testing.expect(!opts.ndjson); + + const missing_up = [_][]const u8{ "--listen", "127.0.0.1:8080" }; + try testing.expectError(error.MissingUpstream, parseCaptureArgs(&missing_up)); + const missing_listen = [_][]const u8{ "--upstream", "ws://x:1" }; + try testing.expectError(error.MissingListen, parseCaptureArgs(&missing_listen)); + const no_port = [_][]const u8{ "--listen", "127.0.0.1", "--upstream", "ws://10.0.0.5:9000" }; + try testing.expectError(error.BadListen, parseCaptureArgs(&no_port)); + const tls = [_][]const u8{ "--listen", "127.0.0.1:8080", "--upstream", "wss://secure:443" }; + try testing.expectError(error.TlsUnsupported, parseCaptureArgs(&tls)); + const bogus = [_][]const u8{"--frobnicate"}; + try testing.expectError(error.Unexpected, parseCaptureArgs(&bogus)); +}