refactor(app): remove pre-broker stdio and quic commands
- commit
- 03285845d6886767b38b6d05d2b7a73b1db32866
- author
- brice · · 2026-09-23 13:45 UTC
- parent
- 22a960f
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
9 files · +21 -385
M docs/protocol.md +6 -6
| @@ -222,7 +222,7 @@ than `KillResult` and `CommandResult`. | |||
| 222 | 222 | | Encoding buffer | bounded per-connection framing buffers owned by `streamframe` | `adapters/streamframe` | | |
| 223 | 223 | | Output states in flight | 8 (`MaxOutputWindow`) preferred for QUIC/IPC/SSH | daemon, negotiated down | | |
| 224 | 224 | | QUIC unauthenticated peers | 4 connections, 3 s auth deadline, one bidirectional stream each; extra/unidirectional streams rejected | `adapters/quic` | | |
| 225 | - | | QUIC bootstrap | 32-byte token + 16-byte nonce (base64), ≤4 KiB readiness (schema v1, host-independent port, SHA-256 pin, ≤15 s expiry), one bounded (512 B) auth record, atomic one-time consumption, token erasure on close | `adapters/quic` bootstrap, `app/_quic-bootstrap` + `_quic-proxy` | | |
| 225 | + | | QUIC bootstrap | 32-byte token + 16-byte nonce (base64), ≤4 KiB readiness (schema v1, host-independent port, SHA-256 pin, ≤15 s expiry), one bounded (512 B) auth record, atomic one-time consumption, token erasure on close | `adapters/quic` bootstrap, `app/_broker-mux-quic-bootstrap` + `_broker-mux-quic-proxy` | | |
| 226 | 226 | | Handshake, dial → first committed publication | 15 s, single absolute deadline propagated unchanged, never reset | both ends | | |
| 227 | 227 | | Command result | 10 s per request | `daemon/command_tracker.go` | | |
| 228 | 228 | | `RemotePreview` viewport | 256×128 cells, 1 MiB | `protocol/preview.go` | | |
| @@ -241,17 +241,17 @@ fields before generated unmarshal. | |||
| 241 | 241 | No session bytes are accepted before authentication and version | |
| 242 | 242 | acceptance. Local carriage requires the preamble + `Hello`/`Welcome` | |
| 243 | 243 | inside the 15-second budget; a failed handshake closes the exact | |
| 244 | - | connection. Direct remote carriage is QUIC (`internal/adapters/quic`, | |
| 244 | + | connection. Broker-owned remote carriage uses QUIC (`internal/adapters/quic`, | |
| 245 | 245 | TLS 1.3, ALPN `vev/1`, exact SHA-256 pin, one bidirectional stream, | |
| 246 | 246 | no 0-RTT); `VEV_REMOTE_TRANSPORT=stdio` selects SSH stdio instead. | |
| 247 | - | Each remote attach starts one short-lived `_quic-bootstrap` over SSH | |
| 248 | - | which spawns a detached `_quic-proxy` (ephemeral listener + | |
| 247 | + | For the QUIC route the broker starts one short-lived `_broker-mux-quic-bootstrap --production` over SSH | |
| 248 | + | which spawns a detached `_broker-mux-quic-proxy` (ephemeral listener + | |
| 249 | 249 | certificate, 32-byte token, ≤4 KiB readiness with host-independent | |
| 250 | 250 | port/pin/expiry), prints the single readiness line, and exits. The | |
| 251 | - | client resolves the SSH target host, composes `host:port`, pins the | |
| 251 | + | broker resolves the SSH target host, composes `host:port`, pins the | |
| 252 | 252 | certificate, and sends one bounded auth record before any preamble. | |
| 253 | 253 | The token is consumed atomically on first use; concurrent, replayed, | |
| 254 | - | or expired auth fails closed without IPC access, and the proxy exits | |
| 254 | + | or expired auth fails closed without private daemonmux access, and the proxy exits | |
| 255 | 255 | after session termination, expiry, or setup failure with token | |
| 256 | 256 | material erased. | |
| 257 | 257 | ||
| @@ -222,7 +222,7 @@ than `KillResult` and `CommandResult`. | |||||
| 222 | | Encoding buffer | bounded per-connection framing buffers owned by `streamframe` | `adapters/streamframe` | | 222 | | Encoding buffer | bounded per-connection framing buffers owned by `streamframe` | `adapters/streamframe` | | ||
| 223 | | Output states in flight | 8 (`MaxOutputWindow`) preferred for QUIC/IPC/SSH | daemon, negotiated down | | 223 | | Output states in flight | 8 (`MaxOutputWindow`) preferred for QUIC/IPC/SSH | daemon, negotiated down | | ||
| 224 | | QUIC unauthenticated peers | 4 connections, 3 s auth deadline, one bidirectional stream each; extra/unidirectional streams rejected | `adapters/quic` | | 224 | | QUIC unauthenticated peers | 4 connections, 3 s auth deadline, one bidirectional stream each; extra/unidirectional streams rejected | `adapters/quic` | | ||
| 225 | - | | QUIC bootstrap | 32-byte token + 16-byte nonce (base64), ≤4 KiB readiness (schema v1, host-independent port, SHA-256 pin, ≤15 s expiry), one bounded (512 B) auth record, atomic one-time consumption, token erasure on close | `adapters/quic` bootstrap, `app/_quic-bootstrap` + `_quic-proxy` | | 225 | + | | QUIC bootstrap | 32-byte token + 16-byte nonce (base64), ≤4 KiB readiness (schema v1, host-independent port, SHA-256 pin, ≤15 s expiry), one bounded (512 B) auth record, atomic one-time consumption, token erasure on close | `adapters/quic` bootstrap, `app/_broker-mux-quic-bootstrap` + `_broker-mux-quic-proxy` | |
| 226 | | Handshake, dial → first committed publication | 15 s, single absolute deadline propagated unchanged, never reset | both ends | | 226 | | Handshake, dial → first committed publication | 15 s, single absolute deadline propagated unchanged, never reset | both ends | | ||
| 227 | | Command result | 10 s per request | `daemon/command_tracker.go` | | 227 | | Command result | 10 s per request | `daemon/command_tracker.go` | | ||
| 228 | | `RemotePreview` viewport | 256×128 cells, 1 MiB | `protocol/preview.go` | | 228 | | `RemotePreview` viewport | 256×128 cells, 1 MiB | `protocol/preview.go` | | ||
| @@ -241,17 +241,17 @@ fields before generated unmarshal. | |||||
| 241 | No session bytes are accepted before authentication and version | 241 | No session bytes are accepted before authentication and version | ||
| 242 | acceptance. Local carriage requires the preamble + `Hello`/`Welcome` | 242 | acceptance. Local carriage requires the preamble + `Hello`/`Welcome` | ||
| 243 | inside the 15-second budget; a failed handshake closes the exact | 243 | inside the 15-second budget; a failed handshake closes the exact | ||
| 244 | - | connection. Direct remote carriage is QUIC (`internal/adapters/quic`, | 244 | + | connection. Broker-owned remote carriage uses QUIC (`internal/adapters/quic`, |
| 245 | TLS 1.3, ALPN `vev/1`, exact SHA-256 pin, one bidirectional stream, | 245 | TLS 1.3, ALPN `vev/1`, exact SHA-256 pin, one bidirectional stream, | ||
| 246 | no 0-RTT); `VEV_REMOTE_TRANSPORT=stdio` selects SSH stdio instead. | 246 | no 0-RTT); `VEV_REMOTE_TRANSPORT=stdio` selects SSH stdio instead. | ||
| 247 | - | Each remote attach starts one short-lived `_quic-bootstrap` over SSH | 247 | + | For the QUIC route the broker starts one short-lived `_broker-mux-quic-bootstrap --production` over SSH |
| 248 | - | which spawns a detached `_quic-proxy` (ephemeral listener + | 248 | + | which spawns a detached `_broker-mux-quic-proxy` (ephemeral listener + |
| 249 | certificate, 32-byte token, ≤4 KiB readiness with host-independent | 249 | certificate, 32-byte token, ≤4 KiB readiness with host-independent | ||
| 250 | port/pin/expiry), prints the single readiness line, and exits. The | 250 | port/pin/expiry), prints the single readiness line, and exits. The | ||
| 251 | - | client resolves the SSH target host, composes `host:port`, pins the | 251 | + | broker resolves the SSH target host, composes `host:port`, pins the |
| 252 | certificate, and sends one bounded auth record before any preamble. | 252 | certificate, and sends one bounded auth record before any preamble. | ||
| 253 | The token is consumed atomically on first use; concurrent, replayed, | 253 | The token is consumed atomically on first use; concurrent, replayed, | ||
| 254 | - | or expired auth fails closed without IPC access, and the proxy exits | 254 | + | or expired auth fails closed without private daemonmux access, and the proxy exits |
| 255 | after session termination, expiry, or setup failure with token | 255 | after session termination, expiry, or setup failure with token | ||
| 256 | material erased. | 256 | material erased. | ||
| 257 | 257 | ||||
M internal/adapters/daemonmux/ssh_carriage_test.go +1 -1
| @@ -173,7 +173,7 @@ func sshMuxSilenceStdout() { | |||
| 173 | 173 | ||
| 174 | 174 | // sshMuxHelperCommand builds an explicit command specification that re-executes | |
| 175 | 175 | // this test binary in one helper mode. The command is never the session | |
| 176 | - | // `_stdio` command: the mux carriage takes its whole command from the caller. | |
| 176 | + | // `_broker-mux-stdio` command: the mux carriage takes its whole command from the caller. | |
| 177 | 177 | func sshMuxHelperCommand(mode string, extra ...string) sshstdio.CommandSpec { | |
| 178 | 178 | args := append([]string{"-test.run=^TestSSHMuxHelperProcess$", sshMuxHelperMarker, mode}, extra...) | |
| 179 | 179 | return sshstdio.CommandSpec{Path: os.Args[0], Args: args} | |
| @@ -173,7 +173,7 @@ func sshMuxSilenceStdout() { | |||||
| 173 | 173 | ||||
| 174 | // sshMuxHelperCommand builds an explicit command specification that re-executes | 174 | // sshMuxHelperCommand builds an explicit command specification that re-executes | ||
| 175 | // this test binary in one helper mode. The command is never the session | 175 | // this test binary in one helper mode. The command is never the session | ||
| 176 | - | // `_stdio` command: the mux carriage takes its whole command from the caller. | 176 | + | // `_broker-mux-stdio` command: the mux carriage takes its whole command from the caller. |
| 177 | func sshMuxHelperCommand(mode string, extra ...string) sshstdio.CommandSpec { | 177 | func sshMuxHelperCommand(mode string, extra ...string) sshstdio.CommandSpec { | ||
| 178 | args := append([]string{"-test.run=^TestSSHMuxHelperProcess$", sshMuxHelperMarker, mode}, extra...) | 178 | args := append([]string{"-test.run=^TestSSHMuxHelperProcess$", sshMuxHelperMarker, mode}, extra...) | ||
| 179 | return sshstdio.CommandSpec{Path: os.Args[0], Args: args} | 179 | return sshstdio.CommandSpec{Path: os.Args[0], Args: args} | ||
M internal/adapters/quic/quic.go +1 -1
| @@ -459,7 +459,7 @@ func (t *Transport) Close() error { | |||
| 459 | 459 | ||
| 460 | 460 | // WaitGracefulTeardown blocks until the deferred hard connection close that | |
| 461 | 461 | // follows an orderly Close has run: the graceful window elapsed or the peer | |
| 462 | - | // closed first. A process-owning caller (the _quic-proxy) waits after Close so | |
| 462 | + | // closed first. A process-owning caller (the broker mux QUIC proxy) waits after Close so | |
| 463 | 463 | // the final synchronous envelope is not discarded by process exit; ctx bounds | |
| 464 | 464 | // the wait. It returns ctx.Err() when the wait is cut short, and blocks until | |
| 465 | 465 | // ctx expires when Close was never called. | |
| @@ -459,7 +459,7 @@ func (t *Transport) Close() error { | |||||
| 459 | 459 | ||||
| 460 | // WaitGracefulTeardown blocks until the deferred hard connection close that | 460 | // WaitGracefulTeardown blocks until the deferred hard connection close that | ||
| 461 | // follows an orderly Close has run: the graceful window elapsed or the peer | 461 | // follows an orderly Close has run: the graceful window elapsed or the peer | ||
| 462 | - | // closed first. A process-owning caller (the _quic-proxy) waits after Close so | 462 | + | // closed first. A process-owning caller (the broker mux QUIC proxy) waits after Close so |
| 463 | // the final synchronous envelope is not discarded by process exit; ctx bounds | 463 | // the final synchronous envelope is not discarded by process exit; ctx bounds | ||
| 464 | // the wait. It returns ctx.Err() when the wait is cut short, and blocks until | 464 | // the wait. It returns ctx.Err() when the wait is cut short, and blocks until | ||
| 465 | // ctx expires when Close was never called. | 465 | // ctx expires when Close was never called. | ||
M internal/adapters/sshstdio/mux.go +2 -2
| @@ -4,7 +4,7 @@ | |||
| 4 | 4 | // multiplexer (internal/adapters/daemonmux) runs over. It is deliberately | |
| 5 | 5 | // separate from the ordinary Dial/DialContext pair: DialMuxContext starts an | |
| 6 | 6 | // explicit, caller-supplied command specification and never builds or reuses | |
| 7 | - | // the `vev _stdio` session command, so a mux carriage can never become a | |
| 7 | + | // an implicit legacy session command, so a mux carriage can never become a | |
| 8 | 8 | // second navigation path to a session. Session selection stays owned by the | |
| 9 | 9 | // typed Hello handshake over the ordinary carriage; the two endpoints cannot be | |
| 10 | 10 | // confused because the mux dial takes the whole command from its caller. | |
| @@ -90,7 +90,7 @@ func BuildCommandForMux(target string, command ...string) CommandSpec { | |||
| 90 | 90 | // DialMuxContext starts spec as one explicit remote command over a local ssh | |
| 91 | 91 | // child process and returns its stdin/stdout as the raw bounded carriage the | |
| 92 | 92 | // daemonmux multiplexer runs over. spec is executed verbatim as argv, never | |
| 93 | - | // through a shell, and is never the `vev _stdio` session command: the caller | |
| 93 | + | // through a shell, and never selects a session itself: the caller | |
| 94 | 94 | // supplies the whole command. | |
| 95 | 95 | // | |
| 96 | 96 | // ctx (or its deadline) bounds subprocess start. The returned carriage is | |
| @@ -4,7 +4,7 @@ | |||||
| 4 | // multiplexer (internal/adapters/daemonmux) runs over. It is deliberately | 4 | // multiplexer (internal/adapters/daemonmux) runs over. It is deliberately | ||
| 5 | // separate from the ordinary Dial/DialContext pair: DialMuxContext starts an | 5 | // separate from the ordinary Dial/DialContext pair: DialMuxContext starts an | ||
| 6 | // explicit, caller-supplied command specification and never builds or reuses | 6 | // explicit, caller-supplied command specification and never builds or reuses | ||
| 7 | - | // the `vev _stdio` session command, so a mux carriage can never become a | 7 | + | // an implicit legacy session command, so a mux carriage can never become a |
| 8 | // second navigation path to a session. Session selection stays owned by the | 8 | // second navigation path to a session. Session selection stays owned by the | ||
| 9 | // typed Hello handshake over the ordinary carriage; the two endpoints cannot be | 9 | // typed Hello handshake over the ordinary carriage; the two endpoints cannot be | ||
| 10 | // confused because the mux dial takes the whole command from its caller. | 10 | // confused because the mux dial takes the whole command from its caller. | ||
| @@ -90,7 +90,7 @@ func BuildCommandForMux(target string, command ...string) CommandSpec { | |||||
| 90 | // DialMuxContext starts spec as one explicit remote command over a local ssh | 90 | // DialMuxContext starts spec as one explicit remote command over a local ssh | ||
| 91 | // child process and returns its stdin/stdout as the raw bounded carriage the | 91 | // child process and returns its stdin/stdout as the raw bounded carriage the | ||
| 92 | // daemonmux multiplexer runs over. spec is executed verbatim as argv, never | 92 | // daemonmux multiplexer runs over. spec is executed verbatim as argv, never | ||
| 93 | - | // through a shell, and is never the `vev _stdio` session command: the caller | 93 | + | // through a shell, and never selects a session itself: the caller |
| 94 | // supplies the whole command. | 94 | // supplies the whole command. | ||
| 95 | // | 95 | // | ||
| 96 | // ctx (or its deadline) bounds subprocess start. The returned carriage is | 96 | // ctx (or its deadline) bounds subprocess start. The returned carriage is | ||
M internal/app/connectivity_matrix.go +5 -26
| @@ -113,14 +113,11 @@ var connectivityMatrix = []connectivityEntry{ | |||
| 113 | 113 | Notes: "Cutover complete: runKill and requestDaemonStop reach the per-user broker through the private connectBroker seam (run.go) and drive BrokerOperations.Kill/KillAll/StopDaemon. Daemon-stop uses ExistingOnly, so it can never start what it is stopping; kill-all purges sessions and leaves the daemon running. The offline persisted-mutation path and the direct force-stop fallback are removed, so no request is ever bypassed with a direct process signal.", | |
| 114 | 114 | }, | |
| 115 | 115 | { | |
| 116 | - | Kind: "kindCmd", | |
| 117 | - | Summary: "scriptable command requests, incl. remote-catalog", | |
| 118 | - | Owner: connectivityBrokerOnly, | |
| 119 | - | DirectDialDebt: []string{ | |
| 120 | - | "realDial command path (cmd.go)", | |
| 121 | - | "ensureDaemonWithLifecycle remote-catalog path (cmd.go)", | |
| 122 | - | }, | |
| 123 | - | Notes: "Command tracker semantics unchanged; only the carriage moves behind the broker.", | |
| 116 | + | Kind: "kindCmd", | |
| 117 | + | Summary: "scriptable command requests, incl. remote-catalog", | |
| 118 | + | Owner: connectivityBrokerOnly, | |
| 119 | + | Migrated: true, | |
| 120 | + | Notes: "Command requests use broker operations in production; the injected direct dial seam remains for focused legacy tracker tests.", | |
| 124 | 121 | }, | |
| 125 | 122 | { | |
| 126 | 123 | Kind: "kindUIDriver", | |
| @@ -162,24 +159,6 @@ var connectivityMatrix = []connectivityEntry{ | |||
| 162 | 159 | Owner: connectivityTransportInfra, | |
| 163 | 160 | Notes: "Process mechanics only; never a connectivity façade.", | |
| 164 | 161 | }, | |
| 165 | - | { | |
| 166 | - | Kind: "kindStdio", | |
| 167 | - | Summary: "_stdio SSH-side carriage proxy", | |
| 168 | - | Owner: connectivityTransportInfra, | |
| 169 | - | Notes: "Terminates SSH stdio transport on the far side; never resolves navigation or owns sessions.", | |
| 170 | - | }, | |
| 171 | - | { | |
| 172 | - | Kind: "kindQUICBootstrap", | |
| 173 | - | Summary: "_quic-bootstrap SSH-side bootstrap", | |
| 174 | - | Owner: connectivityTransportInfra, | |
| 175 | - | Notes: "One-time authenticated bootstrap; terminates transport, never navigation authority.", | |
| 176 | - | }, | |
| 177 | - | { | |
| 178 | - | Kind: "kindQUICProxy", | |
| 179 | - | Summary: "_quic-proxy carriage-neutral bridge", | |
| 180 | - | Owner: connectivityTransportInfra, | |
| 181 | - | Notes: "Blind proxy may forward raw envelopes; exposes no bytes to use cases and owns no sessions.", | |
| 182 | - | }, | |
| 183 | 162 | { | |
| 184 | 163 | Kind: "kindWebRenew", | |
| 185 | 164 | Summary: "browser token renewal", | |
| @@ -113,14 +113,11 @@ var connectivityMatrix = []connectivityEntry{ | |||||
| 113 | Notes: "Cutover complete: runKill and requestDaemonStop reach the per-user broker through the private connectBroker seam (run.go) and drive BrokerOperations.Kill/KillAll/StopDaemon. Daemon-stop uses ExistingOnly, so it can never start what it is stopping; kill-all purges sessions and leaves the daemon running. The offline persisted-mutation path and the direct force-stop fallback are removed, so no request is ever bypassed with a direct process signal.", | 113 | Notes: "Cutover complete: runKill and requestDaemonStop reach the per-user broker through the private connectBroker seam (run.go) and drive BrokerOperations.Kill/KillAll/StopDaemon. Daemon-stop uses ExistingOnly, so it can never start what it is stopping; kill-all purges sessions and leaves the daemon running. The offline persisted-mutation path and the direct force-stop fallback are removed, so no request is ever bypassed with a direct process signal.", | ||
| 114 | }, | 114 | }, | ||
| 115 | { | 115 | { | ||
| 116 | - | Kind: "kindCmd", | 116 | + | Kind: "kindCmd", |
| 117 | - | Summary: "scriptable command requests, incl. remote-catalog", | 117 | + | Summary: "scriptable command requests, incl. remote-catalog", |
| 118 | - | Owner: connectivityBrokerOnly, | 118 | + | Owner: connectivityBrokerOnly, |
| 119 | - | DirectDialDebt: []string{ | 119 | + | Migrated: true, |
| 120 | - | "realDial command path (cmd.go)", | 120 | + | Notes: "Command requests use broker operations in production; the injected direct dial seam remains for focused legacy tracker tests.", |
| 121 | - | "ensureDaemonWithLifecycle remote-catalog path (cmd.go)", | |||
| 122 | - | }, | |||
| 123 | - | Notes: "Command tracker semantics unchanged; only the carriage moves behind the broker.", | |||
| 124 | }, | 121 | }, | ||
| 125 | { | 122 | { | ||
| 126 | Kind: "kindUIDriver", | 123 | Kind: "kindUIDriver", | ||
| @@ -162,24 +159,6 @@ var connectivityMatrix = []connectivityEntry{ | |||||
| 162 | Owner: connectivityTransportInfra, | 159 | Owner: connectivityTransportInfra, | ||
| 163 | Notes: "Process mechanics only; never a connectivity façade.", | 160 | Notes: "Process mechanics only; never a connectivity façade.", | ||
| 164 | }, | 161 | }, | ||
| 165 | - | { | |||
| 166 | - | Kind: "kindStdio", | |||
| 167 | - | Summary: "_stdio SSH-side carriage proxy", | |||
| 168 | - | Owner: connectivityTransportInfra, | |||
| 169 | - | Notes: "Terminates SSH stdio transport on the far side; never resolves navigation or owns sessions.", | |||
| 170 | - | }, | |||
| 171 | - | { | |||
| 172 | - | Kind: "kindQUICBootstrap", | |||
| 173 | - | Summary: "_quic-bootstrap SSH-side bootstrap", | |||
| 174 | - | Owner: connectivityTransportInfra, | |||
| 175 | - | Notes: "One-time authenticated bootstrap; terminates transport, never navigation authority.", | |||
| 176 | - | }, | |||
| 177 | - | { | |||
| 178 | - | Kind: "kindQUICProxy", | |||
| 179 | - | Summary: "_quic-proxy carriage-neutral bridge", | |||
| 180 | - | Owner: connectivityTransportInfra, | |||
| 181 | - | Notes: "Blind proxy may forward raw envelopes; exposes no bytes to use cases and owns no sessions.", | |||
| 182 | - | }, | |||
| 183 | { | 162 | { | ||
| 184 | Kind: "kindWebRenew", | 163 | Kind: "kindWebRenew", | ||
| 185 | Summary: "browser token renewal", | 164 | Summary: "browser token renewal", | ||
M internal/app/run.go +1 -239
| @@ -5,7 +5,6 @@ | |||
| 5 | 5 | package app | |
| 6 | 6 | ||
| 7 | 7 | import ( | |
| 8 | - | "bufio" | |
| 9 | 8 | "context" | |
| 10 | 9 | "crypto/rand" | |
| 11 | 10 | "errors" | |
| @@ -16,7 +15,6 @@ import ( | |||
| 16 | 15 | "net/http" | |
| 17 | 16 | _ "net/http/pprof" | |
| 18 | 17 | "os" | |
| 19 | - | "os/exec" | |
| 20 | 18 | "os/signal" | |
| 21 | 19 | "path/filepath" | |
| 22 | 20 | "strconv" | |
| @@ -35,11 +33,9 @@ import ( | |||
| 35 | 33 | "github.com/bnema/vev/internal/adapters/noticefile" | |
| 36 | 34 | "github.com/bnema/vev/internal/adapters/observability" | |
| 37 | 35 | "github.com/bnema/vev/internal/adapters/pty" | |
| 38 | - | "github.com/bnema/vev/internal/adapters/quic" | |
| 39 | 36 | "github.com/bnema/vev/internal/adapters/sessionwire" | |
| 40 | 37 | "github.com/bnema/vev/internal/adapters/shellcmd" | |
| 41 | 38 | snapshotadapter "github.com/bnema/vev/internal/adapters/snapshot" | |
| 42 | - | "github.com/bnema/vev/internal/adapters/sshstdio" | |
| 43 | 39 | "github.com/bnema/vev/internal/domain" | |
| 44 | 40 | "github.com/bnema/vev/internal/logging" | |
| 45 | 41 | "github.com/bnema/vev/internal/persist" | |
| @@ -64,9 +60,6 @@ const ( | |||
| 64 | 60 | kindCmd | |
| 65 | 61 | kindDaemon | |
| 66 | 62 | kindDaemonLauncher | |
| 67 | - | kindStdio | |
| 68 | - | kindQUICBootstrap | |
| 69 | - | kindQUICProxy | |
| 70 | 63 | kindUIDriver | |
| 71 | 64 | kindUIRemoteCleanup | |
| 72 | 65 | kindWebDaemon | |
| @@ -266,21 +259,6 @@ parsedUIFlags: | |||
| 266 | 259 | return parseBrokerMuxArgs(brokerMuxQUICBootstrapCommand, kindBrokerMuxQUICBootstrap, args[1:]) | |
| 267 | 260 | case brokerMuxQUICProxyCommand: | |
| 268 | 261 | return parseBrokerMuxArgs(brokerMuxQUICProxyCommand, kindBrokerMuxQUICProxy, args[1:]) | |
| 269 | - | case "_stdio": | |
| 270 | - | if len(args) != 1 { | |
| 271 | - | return command{}, usagef("`_stdio` does not accept a session name") | |
| 272 | - | } | |
| 273 | - | return command{kind: kindStdio}, nil | |
| 274 | - | case "_quic-bootstrap": | |
| 275 | - | if len(args) != 1 { | |
| 276 | - | return command{}, usagef("`_quic-bootstrap` does not accept a session name") | |
| 277 | - | } | |
| 278 | - | return command{kind: kindQUICBootstrap}, nil | |
| 279 | - | case "_quic-proxy": | |
| 280 | - | if len(args) != 1 { | |
| 281 | - | return command{}, usagef("`_quic-proxy` does not accept a session name") | |
| 282 | - | } | |
| 283 | - | return command{kind: kindQUICProxy}, nil | |
| 284 | 262 | case uiRemoteCleanupCommand: | |
| 285 | 263 | if len(args) != 1 { | |
| 286 | 264 | return command{}, usagef("`%s` does not accept arguments", uiRemoteCleanupCommand) | |
| @@ -417,12 +395,6 @@ func dispatch(ctx context.Context, cmd command) error { | |||
| 417 | 395 | return runProductionBrokerLauncherCommand(ctx) | |
| 418 | 396 | case kindBrokerReady: | |
| 419 | 397 | return runBrokerReady(ctx, cmd.brokerReady, os.Stdout) | |
| 420 | - | case kindStdio: | |
| 421 | - | return runStdio(ctx) | |
| 422 | - | case kindQUICBootstrap: | |
| 423 | - | return runQUICBootstrap(ctx) | |
| 424 | - | case kindQUICProxy: | |
| 425 | - | return runQUICProxy(ctx) | |
| 426 | 398 | case kindUIRemoteCleanup: | |
| 427 | 399 | return runUIRemoteCleanup(ctx) | |
| 428 | 400 | case kindUIDriver: | |
| @@ -1069,162 +1041,8 @@ func brokerKillResultError(result protocol.KillResult) error { | |||
| 1069 | 1041 | } | |
| 1070 | 1042 | } | |
| 1071 | 1043 | ||
| 1072 | - | // runStdio is the hidden remote-side mode used by `ssh host vev _stdio`: it | |
| 1073 | - | // connects to the per-user daemon (auto-spawning it if needed) and proxies the | |
| 1074 | - | // framed protocol between process stdio and the daemon socket. | |
| 1075 | - | func runStdio(ctx context.Context) (retErr error) { | |
| 1076 | - | log, logCloser, err := configureLogging(logging.Stdio, false) | |
| 1077 | - | if err != nil { | |
| 1078 | - | return err | |
| 1079 | - | } | |
| 1080 | - | defer func() { _ = logCloser.Close() }() | |
| 1081 | - | log.Debug("stdio proxy starting") | |
| 1082 | - | ||
| 1083 | - | clk := clock.New() | |
| 1084 | - | observer, observerCloser, err := newPerformanceTrace(clk) | |
| 1085 | - | if err != nil { | |
| 1086 | - | return fmt.Errorf("vev: performance trace: %w", err) | |
| 1087 | - | } | |
| 1088 | - | if observerCloser != nil { | |
| 1089 | - | defer func() { retErr = errors.Join(retErr, observerCloser.Close()) }() | |
| 1090 | - | } | |
| 1091 | - | transport, err := ensureDaemonWithLifecycle(ctx, ipc.SocketDir(), func(ctx context.Context, dir string) (wire.Transport, error) { | |
| 1092 | - | return ipc.DialContext(ctx, dir, ipc.WithRuntimeObserver(observer)) | |
| 1093 | - | }, realSpawn, defaultBackoff) | |
| 1094 | - | if err != nil { | |
| 1095 | - | return err | |
| 1096 | - | } | |
| 1097 | - | defer func() { _ = transport.Close() }() | |
| 1098 | - | ||
| 1099 | - | // sshstdio.Close owns and closes writers that implement io.Closer so a | |
| 1100 | - | // blocked Send can unwind. Keep process-global stdin/stdout outside that | |
| 1101 | - | // ownership boundary: sshstdio recognizes process stdin as a shared | |
| 1102 | - | // cancellable pump, while output is an owned pipe drained by a forwarding | |
| 1103 | - | // goroutine. | |
| 1104 | - | stdin, stdout := os.Stdin, os.Stdout | |
| 1105 | - | stdoutReader, stdoutWriter := io.Pipe() | |
| 1106 | - | forwardDone := make(chan struct{}) | |
| 1107 | - | go func() { | |
| 1108 | - | defer close(forwardDone) | |
| 1109 | - | if _, err := io.Copy(stdout, stdoutReader); err != nil { | |
| 1110 | - | _ = stdoutReader.CloseWithError(err) | |
| 1111 | - | } | |
| 1112 | - | }() | |
| 1113 | - | defer func() { | |
| 1114 | - | _ = stdoutWriter.Close() | |
| 1115 | - | <-forwardDone | |
| 1116 | - | _ = stdoutReader.Close() | |
| 1117 | - | }() | |
| 1118 | - | stdio := sshstdio.NewTransport(stdin, stdoutWriter, nil, sshstdio.WithRuntimeObserver(observer)) | |
| 1119 | - | return proxyTransports(ctx, stdio, transport, log) | |
| 1120 | - | } | |
| 1121 | - | ||
| 1122 | - | // runQUICBootstrap starts a detached QUIC proxy, forwards its single | |
| 1123 | - | // readiness line, and exits so the SSH channel closes without owning | |
| 1124 | - | // the session lifetime. The detached proxy serves exactly one | |
| 1125 | - | // authenticated stream, then exits deterministically. | |
| 1126 | - | func runQUICBootstrap(ctx context.Context) error { | |
| 1127 | - | r, w, err := os.Pipe() | |
| 1128 | - | if err != nil { | |
| 1129 | - | return err | |
| 1130 | - | } | |
| 1131 | - | defer func() { _ = r.Close() }() | |
| 1132 | - | ||
| 1133 | - | exe, err := os.Executable() | |
| 1134 | - | if err != nil { | |
| 1135 | - | _ = w.Close() | |
| 1136 | - | return err | |
| 1137 | - | } | |
| 1138 | - | devNull, err := os.OpenFile(os.DevNull, os.O_RDWR, 0) | |
| 1139 | - | if err != nil { | |
| 1140 | - | _ = w.Close() | |
| 1141 | - | return err | |
| 1142 | - | } | |
| 1143 | - | defer func() { _ = devNull.Close() }() | |
| 1144 | - | ||
| 1145 | - | cmd := exec.CommandContext(ctx, exe, "_quic-proxy") | |
| 1146 | - | // The proxy writes diagnostics through configureLogging; stdio is | |
| 1147 | - | // detached here so the bootstrap SSH channel can close. The proxy | |
| 1148 | - | // owns the session lifetime in its own session (Setsid) and is | |
| 1149 | - | // released: the bootstrap child exits after forwarding readiness. | |
| 1150 | - | cmd.Stdin = devNull | |
| 1151 | - | cmd.Stdout = w | |
| 1152 | - | cmd.Stderr = devNull | |
| 1153 | - | cmd.SysProcAttr = &syscall.SysProcAttr{Setsid: true} | |
| 1154 | - | if err := cmd.Start(); err != nil { | |
| 1155 | - | _ = w.Close() | |
| 1156 | - | return err | |
| 1157 | - | } | |
| 1158 | - | _ = w.Close() | |
| 1159 | - | ||
| 1160 | - | readyCtx, cancel := context.WithTimeout(ctx, 5*time.Second) | |
| 1161 | - | defer cancel() | |
| 1162 | - | lineCh := make(chan string, 1) | |
| 1163 | - | errCh := make(chan error, 1) | |
| 1164 | - | go func() { | |
| 1165 | - | line, err := bufio.NewReader(r).ReadString('\n') | |
| 1166 | - | if err != nil { | |
| 1167 | - | errCh <- err | |
| 1168 | - | return | |
| 1169 | - | } | |
| 1170 | - | lineCh <- line | |
| 1171 | - | }() | |
| 1172 | - | select { | |
| 1173 | - | case line := <-lineCh: | |
| 1174 | - | if err := cmd.Process.Release(); err != nil { | |
| 1175 | - | return err | |
| 1176 | - | } | |
| 1177 | - | _, err := fmt.Fprint(os.Stdout, line) | |
| 1178 | - | return err | |
| 1179 | - | case err := <-errCh: | |
| 1180 | - | _ = cmd.Wait() | |
| 1181 | - | return err | |
| 1182 | - | case <-readyCtx.Done(): | |
| 1183 | - | _ = cmd.Process.Kill() | |
| 1184 | - | _ = cmd.Wait() | |
| 1185 | - | return fmt.Errorf("vev: quic bootstrap readiness: %w", readyCtx.Err()) | |
| 1186 | - | } | |
| 1187 | - | } | |
| 1188 | - | ||
| 1189 | - | // runQUICProxy is the detached long-lived remote-side QUIC proxy: one | |
| 1190 | - | // ephemeral listener/certificate, one atomic token, private IPC only | |
| 1191 | - | // after the first authenticated stream. It exits after session | |
| 1192 | - | // termination, token expiry, or setup failure. | |
| 1193 | - | func runQUICProxy(ctx context.Context) (retErr error) { | |
| 1194 | - | log, logCloser, err := configureLogging(logging.Stdio, false) | |
| 1195 | - | if err != nil { | |
| 1196 | - | return err | |
| 1197 | - | } | |
| 1198 | - | defer func() { _ = logCloser.Close() }() | |
| 1199 | - | log.Debug("quic proxy starting") | |
| 1200 | - | server, readiness, err := quic.NewServer() | |
| 1201 | - | if err != nil { | |
| 1202 | - | return err | |
| 1203 | - | } | |
| 1204 | - | defer func() { _ = server.Close() }() | |
| 1205 | - | raw, err := quic.EncodeReadiness(readiness) | |
| 1206 | - | if err != nil { | |
| 1207 | - | return err | |
| 1208 | - | } | |
| 1209 | - | if _, err := fmt.Fprintf(os.Stdout, "%s\n", raw); err != nil { | |
| 1210 | - | return err | |
| 1211 | - | } | |
| 1212 | - | // Serve exactly one authenticated session, then exit | |
| 1213 | - | // deterministically. Accept fails closed on token expiry, so the | |
| 1214 | - | // proxy cannot outlive its advertised credential. | |
| 1215 | - | transport, err := server.Accept(ctx) | |
| 1216 | - | if err != nil { | |
| 1217 | - | return err | |
| 1218 | - | } | |
| 1219 | - | defer func() { | |
| 1220 | - | _ = transport.Close() | |
| 1221 | - | _ = waitGracefulTeardown(transport) | |
| 1222 | - | }() | |
| 1223 | - | return proxyQUICBootstrap(ctx, transport) | |
| 1224 | - | } | |
| 1225 | - | ||
| 1226 | 1044 | // gracefulTeardownWaiter is implemented by carriages (QUIC) whose Close defers | |
| 1227 | - | // a bounded connection close past its return. The _quic-proxy owns its | |
| 1045 | + | // a bounded connection close past its return. The broker mux QUIC proxy owns its | |
| 1228 | 1046 | // process, so it must wait for that close before returning: exiting first | |
| 1229 | 1047 | // would discard the final synchronous envelope Close already handed to the | |
| 1230 | 1048 | // stream. | |
| @@ -1247,62 +1065,6 @@ func waitGracefulTeardown(transport wire.Transport) error { | |||
| 1247 | 1065 | return waiter.WaitGracefulTeardown(ctx) | |
| 1248 | 1066 | } | |
| 1249 | 1067 | ||
| 1250 | - | // proxyQUICBootstrap bridges the single authenticated QUIC stream to the | |
| 1251 | - | // local daemon over private IPC. The daemon side is wrapped as a typed | |
| 1252 | - | // server connection by the daemon's own accept path; here the bridge | |
| 1253 | - | // copies raw envelopes both ways until either side closes. No session | |
| 1254 | - | // bytes flow before auth: Accept returned only the authenticated stream. | |
| 1255 | - | func proxyQUICBootstrap(ctx context.Context, transport wire.Transport) error { | |
| 1256 | - | daemonTr, err := ensureDaemonWithLifecycle(ctx, ipc.SocketDir(), func(ctx context.Context, dir string) (wire.Transport, error) { | |
| 1257 | - | return ipc.DialContext(ctx, dir) | |
| 1258 | - | }, realSpawn, defaultBackoff) | |
| 1259 | - | if err != nil { | |
| 1260 | - | return err | |
| 1261 | - | } | |
| 1262 | - | defer func() { _ = daemonTr.Close() }() | |
| 1263 | - | return proxyTransports(ctx, transport, daemonTr, slog.Default()) | |
| 1264 | - | } | |
| 1265 | - | ||
| 1266 | - | // proxyTransports copies raw envelopes both ways until either side | |
| 1267 | - | // closes. It is carriage-neutral (IPC, SSH, QUIC): the only datagram | |
| 1268 | - | // special case (hello output-window clamping) lived in the removed | |
| 1269 | - | // datagram proxy, not here. | |
| 1270 | - | func proxyTransports(ctx context.Context, a, b wire.Transport, _ *slog.Logger) error { | |
| 1271 | - | ctx, cancel := context.WithCancel(ctx) | |
| 1272 | - | defer cancel() | |
| 1273 | - | // Own both carriages: closing them releases the copy goroutines' | |
| 1274 | - | // blocked Recv calls (with serialized end marks) instead of | |
| 1275 | - | // abandoning them mid-Recv when the first direction fails. | |
| 1276 | - | defer func() { _ = a.Close() }() | |
| 1277 | - | defer func() { _ = b.Close() }() | |
| 1278 | - | errCh := make(chan error, 2) | |
| 1279 | - | go func() { errCh <- copyTransport(ctx, a, b) }() | |
| 1280 | - | go func() { errCh <- copyTransport(ctx, b, a) }() | |
| 1281 | - | select { | |
| 1282 | - | case err := <-errCh: | |
| 1283 | - | return err | |
| 1284 | - | case <-ctx.Done(): | |
| 1285 | - | return ctx.Err() | |
| 1286 | - | } | |
| 1287 | - | } | |
| 1288 | - | ||
| 1289 | - | func copyTransport(ctx context.Context, src, dst wire.Transport) error { | |
| 1290 | - | for { | |
| 1291 | - | envelope, err := src.Recv() | |
| 1292 | - | if err != nil { | |
| 1293 | - | return err | |
| 1294 | - | } | |
| 1295 | - | if err := dst.Send(envelope); err != nil { | |
| 1296 | - | return err | |
| 1297 | - | } | |
| 1298 | - | select { | |
| 1299 | - | case <-ctx.Done(): | |
| 1300 | - | return ctx.Err() | |
| 1301 | - | default: | |
| 1302 | - | } | |
| 1303 | - | } | |
| 1304 | - | } | |
| 1305 | - | ||
| 1306 | 1068 | // runList reads existing state without starting either background process. | |
| 1307 | 1069 | func runList(ctx context.Context, cmd command) error { | |
| 1308 | 1070 | service, err := connectListBroker(ctx) | |
| @@ -5,7 +5,6 @@ | |||||
| 5 | package app | 5 | package app | ||
| 6 | 6 | ||||
| 7 | import ( | 7 | import ( | ||
| 8 | - | "bufio" | |||
| 9 | "context" | 8 | "context" | ||
| 10 | "crypto/rand" | 9 | "crypto/rand" | ||
| 11 | "errors" | 10 | "errors" | ||
| @@ -16,7 +15,6 @@ import ( | |||||
| 16 | "net/http" | 15 | "net/http" | ||
| 17 | _ "net/http/pprof" | 16 | _ "net/http/pprof" | ||
| 18 | "os" | 17 | "os" | ||
| 19 | - | "os/exec" | |||
| 20 | "os/signal" | 18 | "os/signal" | ||
| 21 | "path/filepath" | 19 | "path/filepath" | ||
| 22 | "strconv" | 20 | "strconv" | ||
| @@ -35,11 +33,9 @@ import ( | |||||
| 35 | "github.com/bnema/vev/internal/adapters/noticefile" | 33 | "github.com/bnema/vev/internal/adapters/noticefile" | ||
| 36 | "github.com/bnema/vev/internal/adapters/observability" | 34 | "github.com/bnema/vev/internal/adapters/observability" | ||
| 37 | "github.com/bnema/vev/internal/adapters/pty" | 35 | "github.com/bnema/vev/internal/adapters/pty" | ||
| 38 | - | "github.com/bnema/vev/internal/adapters/quic" | |||
| 39 | "github.com/bnema/vev/internal/adapters/sessionwire" | 36 | "github.com/bnema/vev/internal/adapters/sessionwire" | ||
| 40 | "github.com/bnema/vev/internal/adapters/shellcmd" | 37 | "github.com/bnema/vev/internal/adapters/shellcmd" | ||
| 41 | snapshotadapter "github.com/bnema/vev/internal/adapters/snapshot" | 38 | snapshotadapter "github.com/bnema/vev/internal/adapters/snapshot" | ||
| 42 | - | "github.com/bnema/vev/internal/adapters/sshstdio" | |||
| 43 | "github.com/bnema/vev/internal/domain" | 39 | "github.com/bnema/vev/internal/domain" | ||
| 44 | "github.com/bnema/vev/internal/logging" | 40 | "github.com/bnema/vev/internal/logging" | ||
| 45 | "github.com/bnema/vev/internal/persist" | 41 | "github.com/bnema/vev/internal/persist" | ||
| @@ -64,9 +60,6 @@ const ( | |||||
| 64 | kindCmd | 60 | kindCmd | ||
| 65 | kindDaemon | 61 | kindDaemon | ||
| 66 | kindDaemonLauncher | 62 | kindDaemonLauncher | ||
| 67 | - | kindStdio | |||
| 68 | - | kindQUICBootstrap | |||
| 69 | - | kindQUICProxy | |||
| 70 | kindUIDriver | 63 | kindUIDriver | ||
| 71 | kindUIRemoteCleanup | 64 | kindUIRemoteCleanup | ||
| 72 | kindWebDaemon | 65 | kindWebDaemon | ||
| @@ -266,21 +259,6 @@ parsedUIFlags: | |||||
| 266 | return parseBrokerMuxArgs(brokerMuxQUICBootstrapCommand, kindBrokerMuxQUICBootstrap, args[1:]) | 259 | return parseBrokerMuxArgs(brokerMuxQUICBootstrapCommand, kindBrokerMuxQUICBootstrap, args[1:]) | ||
| 267 | case brokerMuxQUICProxyCommand: | 260 | case brokerMuxQUICProxyCommand: | ||
| 268 | return parseBrokerMuxArgs(brokerMuxQUICProxyCommand, kindBrokerMuxQUICProxy, args[1:]) | 261 | return parseBrokerMuxArgs(brokerMuxQUICProxyCommand, kindBrokerMuxQUICProxy, args[1:]) | ||
| 269 | - | case "_stdio": | |||
| 270 | - | if len(args) != 1 { | |||
| 271 | - | return command{}, usagef("`_stdio` does not accept a session name") | |||
| 272 | - | } | |||
| 273 | - | return command{kind: kindStdio}, nil | |||
| 274 | - | case "_quic-bootstrap": | |||
| 275 | - | if len(args) != 1 { | |||
| 276 | - | return command{}, usagef("`_quic-bootstrap` does not accept a session name") | |||
| 277 | - | } | |||
| 278 | - | return command{kind: kindQUICBootstrap}, nil | |||
| 279 | - | case "_quic-proxy": | |||
| 280 | - | if len(args) != 1 { | |||
| 281 | - | return command{}, usagef("`_quic-proxy` does not accept a session name") | |||
| 282 | - | } | |||
| 283 | - | return command{kind: kindQUICProxy}, nil | |||
| 284 | case uiRemoteCleanupCommand: | 262 | case uiRemoteCleanupCommand: | ||
| 285 | if len(args) != 1 { | 263 | if len(args) != 1 { | ||
| 286 | return command{}, usagef("`%s` does not accept arguments", uiRemoteCleanupCommand) | 264 | return command{}, usagef("`%s` does not accept arguments", uiRemoteCleanupCommand) | ||
| @@ -417,12 +395,6 @@ func dispatch(ctx context.Context, cmd command) error { | |||||
| 417 | return runProductionBrokerLauncherCommand(ctx) | 395 | return runProductionBrokerLauncherCommand(ctx) | ||
| 418 | case kindBrokerReady: | 396 | case kindBrokerReady: | ||
| 419 | return runBrokerReady(ctx, cmd.brokerReady, os.Stdout) | 397 | return runBrokerReady(ctx, cmd.brokerReady, os.Stdout) | ||
| 420 | - | case kindStdio: | |||
| 421 | - | return runStdio(ctx) | |||
| 422 | - | case kindQUICBootstrap: | |||
| 423 | - | return runQUICBootstrap(ctx) | |||
| 424 | - | case kindQUICProxy: | |||
| 425 | - | return runQUICProxy(ctx) | |||
| 426 | case kindUIRemoteCleanup: | 398 | case kindUIRemoteCleanup: | ||
| 427 | return runUIRemoteCleanup(ctx) | 399 | return runUIRemoteCleanup(ctx) | ||
| 428 | case kindUIDriver: | 400 | case kindUIDriver: | ||
| @@ -1069,162 +1041,8 @@ func brokerKillResultError(result protocol.KillResult) error { | |||||
| 1069 | } | 1041 | } | ||
| 1070 | } | 1042 | } | ||
| 1071 | 1043 | ||||
| 1072 | - | // runStdio is the hidden remote-side mode used by `ssh host vev _stdio`: it | |||
| 1073 | - | // connects to the per-user daemon (auto-spawning it if needed) and proxies the | |||
| 1074 | - | // framed protocol between process stdio and the daemon socket. | |||
| 1075 | - | func runStdio(ctx context.Context) (retErr error) { | |||
| 1076 | - | log, logCloser, err := configureLogging(logging.Stdio, false) | |||
| 1077 | - | if err != nil { | |||
| 1078 | - | return err | |||
| 1079 | - | } | |||
| 1080 | - | defer func() { _ = logCloser.Close() }() | |||
| 1081 | - | log.Debug("stdio proxy starting") | |||
| 1082 | - | ||||
| 1083 | - | clk := clock.New() | |||
| 1084 | - | observer, observerCloser, err := newPerformanceTrace(clk) | |||
| 1085 | - | if err != nil { | |||
| 1086 | - | return fmt.Errorf("vev: performance trace: %w", err) | |||
| 1087 | - | } | |||
| 1088 | - | if observerCloser != nil { | |||
| 1089 | - | defer func() { retErr = errors.Join(retErr, observerCloser.Close()) }() | |||
| 1090 | - | } | |||
| 1091 | - | transport, err := ensureDaemonWithLifecycle(ctx, ipc.SocketDir(), func(ctx context.Context, dir string) (wire.Transport, error) { | |||
| 1092 | - | return ipc.DialContext(ctx, dir, ipc.WithRuntimeObserver(observer)) | |||
| 1093 | - | }, realSpawn, defaultBackoff) | |||
| 1094 | - | if err != nil { | |||
| 1095 | - | return err | |||
| 1096 | - | } | |||
| 1097 | - | defer func() { _ = transport.Close() }() | |||
| 1098 | - | ||||
| 1099 | - | // sshstdio.Close owns and closes writers that implement io.Closer so a | |||
| 1100 | - | // blocked Send can unwind. Keep process-global stdin/stdout outside that | |||
| 1101 | - | // ownership boundary: sshstdio recognizes process stdin as a shared | |||
| 1102 | - | // cancellable pump, while output is an owned pipe drained by a forwarding | |||
| 1103 | - | // goroutine. | |||
| 1104 | - | stdin, stdout := os.Stdin, os.Stdout | |||
| 1105 | - | stdoutReader, stdoutWriter := io.Pipe() | |||
| 1106 | - | forwardDone := make(chan struct{}) | |||
| 1107 | - | go func() { | |||
| 1108 | - | defer close(forwardDone) | |||
| 1109 | - | if _, err := io.Copy(stdout, stdoutReader); err != nil { | |||
| 1110 | - | _ = stdoutReader.CloseWithError(err) | |||
| 1111 | - | } | |||
| 1112 | - | }() | |||
| 1113 | - | defer func() { | |||
| 1114 | - | _ = stdoutWriter.Close() | |||
| 1115 | - | <-forwardDone | |||
| 1116 | - | _ = stdoutReader.Close() | |||
| 1117 | - | }() | |||
| 1118 | - | stdio := sshstdio.NewTransport(stdin, stdoutWriter, nil, sshstdio.WithRuntimeObserver(observer)) | |||
| 1119 | - | return proxyTransports(ctx, stdio, transport, log) | |||
| 1120 | - | } | |||
| 1121 | - | ||||
| 1122 | - | // runQUICBootstrap starts a detached QUIC proxy, forwards its single | |||
| 1123 | - | // readiness line, and exits so the SSH channel closes without owning | |||
| 1124 | - | // the session lifetime. The detached proxy serves exactly one | |||
| 1125 | - | // authenticated stream, then exits deterministically. | |||
| 1126 | - | func runQUICBootstrap(ctx context.Context) error { | |||
| 1127 | - | r, w, err := os.Pipe() | |||
| 1128 | - | if err != nil { | |||
| 1129 | - | return err | |||
| 1130 | - | } | |||
| 1131 | - | defer func() { _ = r.Close() }() | |||
| 1132 | - | ||||
| 1133 | - | exe, err := os.Executable() | |||
| 1134 | - | if err != nil { | |||
| 1135 | - | _ = w.Close() | |||
| 1136 | - | return err | |||
| 1137 | - | } | |||
| 1138 | - | devNull, err := os.OpenFile(os.DevNull, os.O_RDWR, 0) | |||
| 1139 | - | if err != nil { | |||
| 1140 | - | _ = w.Close() | |||
| 1141 | - | return err | |||
| 1142 | - | } | |||
| 1143 | - | defer func() { _ = devNull.Close() }() | |||
| 1144 | - | ||||
| 1145 | - | cmd := exec.CommandContext(ctx, exe, "_quic-proxy") | |||
| 1146 | - | // The proxy writes diagnostics through configureLogging; stdio is | |||
| 1147 | - | // detached here so the bootstrap SSH channel can close. The proxy | |||
| 1148 | - | // owns the session lifetime in its own session (Setsid) and is | |||
| 1149 | - | // released: the bootstrap child exits after forwarding readiness. | |||
| 1150 | - | cmd.Stdin = devNull | |||
| 1151 | - | cmd.Stdout = w | |||
| 1152 | - | cmd.Stderr = devNull | |||
| 1153 | - | cmd.SysProcAttr = &syscall.SysProcAttr{Setsid: true} | |||
| 1154 | - | if err := cmd.Start(); err != nil { | |||
| 1155 | - | _ = w.Close() | |||
| 1156 | - | return err | |||
| 1157 | - | } | |||
| 1158 | - | _ = w.Close() | |||
| 1159 | - | ||||
| 1160 | - | readyCtx, cancel := context.WithTimeout(ctx, 5*time.Second) | |||
| 1161 | - | defer cancel() | |||
| 1162 | - | lineCh := make(chan string, 1) | |||
| 1163 | - | errCh := make(chan error, 1) | |||
| 1164 | - | go func() { | |||
| 1165 | - | line, err := bufio.NewReader(r).ReadString('\n') | |||
| 1166 | - | if err != nil { | |||
| 1167 | - | errCh <- err | |||
| 1168 | - | return | |||
| 1169 | - | } | |||
| 1170 | - | lineCh <- line | |||
| 1171 | - | }() | |||
| 1172 | - | select { | |||
| 1173 | - | case line := <-lineCh: | |||
| 1174 | - | if err := cmd.Process.Release(); err != nil { | |||
| 1175 | - | return err | |||
| 1176 | - | } | |||
| 1177 | - | _, err := fmt.Fprint(os.Stdout, line) | |||
| 1178 | - | return err | |||
| 1179 | - | case err := <-errCh: | |||
| 1180 | - | _ = cmd.Wait() | |||
| 1181 | - | return err | |||
| 1182 | - | case <-readyCtx.Done(): | |||
| 1183 | - | _ = cmd.Process.Kill() | |||
| 1184 | - | _ = cmd.Wait() | |||
| 1185 | - | return fmt.Errorf("vev: quic bootstrap readiness: %w", readyCtx.Err()) | |||
| 1186 | - | } | |||
| 1187 | - | } | |||
| 1188 | - | ||||
| 1189 | - | // runQUICProxy is the detached long-lived remote-side QUIC proxy: one | |||
| 1190 | - | // ephemeral listener/certificate, one atomic token, private IPC only | |||
| 1191 | - | // after the first authenticated stream. It exits after session | |||
| 1192 | - | // termination, token expiry, or setup failure. | |||
| 1193 | - | func runQUICProxy(ctx context.Context) (retErr error) { | |||
| 1194 | - | log, logCloser, err := configureLogging(logging.Stdio, false) | |||
| 1195 | - | if err != nil { | |||
| 1196 | - | return err | |||
| 1197 | - | } | |||
| 1198 | - | defer func() { _ = logCloser.Close() }() | |||
| 1199 | - | log.Debug("quic proxy starting") | |||
| 1200 | - | server, readiness, err := quic.NewServer() | |||
| 1201 | - | if err != nil { | |||
| 1202 | - | return err | |||
| 1203 | - | } | |||
| 1204 | - | defer func() { _ = server.Close() }() | |||
| 1205 | - | raw, err := quic.EncodeReadiness(readiness) | |||
| 1206 | - | if err != nil { | |||
| 1207 | - | return err | |||
| 1208 | - | } | |||
| 1209 | - | if _, err := fmt.Fprintf(os.Stdout, "%s\n", raw); err != nil { | |||
| 1210 | - | return err | |||
| 1211 | - | } | |||
| 1212 | - | // Serve exactly one authenticated session, then exit | |||
| 1213 | - | // deterministically. Accept fails closed on token expiry, so the | |||
| 1214 | - | // proxy cannot outlive its advertised credential. | |||
| 1215 | - | transport, err := server.Accept(ctx) | |||
| 1216 | - | if err != nil { | |||
| 1217 | - | return err | |||
| 1218 | - | } | |||
| 1219 | - | defer func() { | |||
| 1220 | - | _ = transport.Close() | |||
| 1221 | - | _ = waitGracefulTeardown(transport) | |||
| 1222 | - | }() | |||
| 1223 | - | return proxyQUICBootstrap(ctx, transport) | |||
| 1224 | - | } | |||
| 1225 | - | ||||
| 1226 | // gracefulTeardownWaiter is implemented by carriages (QUIC) whose Close defers | 1044 | // gracefulTeardownWaiter is implemented by carriages (QUIC) whose Close defers | ||
| 1227 | - | // a bounded connection close past its return. The _quic-proxy owns its | 1045 | + | // a bounded connection close past its return. The broker mux QUIC proxy owns its |
| 1228 | // process, so it must wait for that close before returning: exiting first | 1046 | // process, so it must wait for that close before returning: exiting first | ||
| 1229 | // would discard the final synchronous envelope Close already handed to the | 1047 | // would discard the final synchronous envelope Close already handed to the | ||
| 1230 | // stream. | 1048 | // stream. | ||
| @@ -1247,62 +1065,6 @@ func waitGracefulTeardown(transport wire.Transport) error { | |||||
| 1247 | return waiter.WaitGracefulTeardown(ctx) | 1065 | return waiter.WaitGracefulTeardown(ctx) | ||
| 1248 | } | 1066 | } | ||
| 1249 | 1067 | ||||
| 1250 | - | // proxyQUICBootstrap bridges the single authenticated QUIC stream to the | |||
| 1251 | - | // local daemon over private IPC. The daemon side is wrapped as a typed | |||
| 1252 | - | // server connection by the daemon's own accept path; here the bridge | |||
| 1253 | - | // copies raw envelopes both ways until either side closes. No session | |||
| 1254 | - | // bytes flow before auth: Accept returned only the authenticated stream. | |||
| 1255 | - | func proxyQUICBootstrap(ctx context.Context, transport wire.Transport) error { | |||
| 1256 | - | daemonTr, err := ensureDaemonWithLifecycle(ctx, ipc.SocketDir(), func(ctx context.Context, dir string) (wire.Transport, error) { | |||
| 1257 | - | return ipc.DialContext(ctx, dir) | |||
| 1258 | - | }, realSpawn, defaultBackoff) | |||
| 1259 | - | if err != nil { | |||
| 1260 | - | return err | |||
| 1261 | - | } | |||
| 1262 | - | defer func() { _ = daemonTr.Close() }() | |||
| 1263 | - | return proxyTransports(ctx, transport, daemonTr, slog.Default()) | |||
| 1264 | - | } | |||
| 1265 | - | ||||
| 1266 | - | // proxyTransports copies raw envelopes both ways until either side | |||
| 1267 | - | // closes. It is carriage-neutral (IPC, SSH, QUIC): the only datagram | |||
| 1268 | - | // special case (hello output-window clamping) lived in the removed | |||
| 1269 | - | // datagram proxy, not here. | |||
| 1270 | - | func proxyTransports(ctx context.Context, a, b wire.Transport, _ *slog.Logger) error { | |||
| 1271 | - | ctx, cancel := context.WithCancel(ctx) | |||
| 1272 | - | defer cancel() | |||
| 1273 | - | // Own both carriages: closing them releases the copy goroutines' | |||
| 1274 | - | // blocked Recv calls (with serialized end marks) instead of | |||
| 1275 | - | // abandoning them mid-Recv when the first direction fails. | |||
| 1276 | - | defer func() { _ = a.Close() }() | |||
| 1277 | - | defer func() { _ = b.Close() }() | |||
| 1278 | - | errCh := make(chan error, 2) | |||
| 1279 | - | go func() { errCh <- copyTransport(ctx, a, b) }() | |||
| 1280 | - | go func() { errCh <- copyTransport(ctx, b, a) }() | |||
| 1281 | - | select { | |||
| 1282 | - | case err := <-errCh: | |||
| 1283 | - | return err | |||
| 1284 | - | case <-ctx.Done(): | |||
| 1285 | - | return ctx.Err() | |||
| 1286 | - | } | |||
| 1287 | - | } | |||
| 1288 | - | ||||
| 1289 | - | func copyTransport(ctx context.Context, src, dst wire.Transport) error { | |||
| 1290 | - | for { | |||
| 1291 | - | envelope, err := src.Recv() | |||
| 1292 | - | if err != nil { | |||
| 1293 | - | return err | |||
| 1294 | - | } | |||
| 1295 | - | if err := dst.Send(envelope); err != nil { | |||
| 1296 | - | return err | |||
| 1297 | - | } | |||
| 1298 | - | select { | |||
| 1299 | - | case <-ctx.Done(): | |||
| 1300 | - | return ctx.Err() | |||
| 1301 | - | default: | |||
| 1302 | - | } | |||
| 1303 | - | } | |||
| 1304 | - | } | |||
| 1305 | - | ||||
| 1306 | // runList reads existing state without starting either background process. | 1068 | // runList reads existing state without starting either background process. | ||
| 1307 | func runList(ctx context.Context, cmd command) error { | 1069 | func runList(ctx context.Context, cmd command) error { | ||
| 1308 | service, err := connectListBroker(ctx) | 1070 | service, err := connectListBroker(ctx) | ||
M internal/app/run_test.go +3 -108
| @@ -8,7 +8,6 @@ import ( | |||
| 8 | 8 | "errors" | |
| 9 | 9 | "io" | |
| 10 | 10 | "log/slog" | |
| 11 | - | "net" | |
| 12 | 11 | "os" | |
| 13 | 12 | "path/filepath" | |
| 14 | 13 | "strings" | |
| @@ -16,9 +15,7 @@ import ( | |||
| 16 | 15 | "time" | |
| 17 | 16 | ||
| 18 | 17 | "github.com/bnema/vev/internal/adapters/daemonmux" | |
| 19 | - | "github.com/bnema/vev/internal/adapters/ipc" | |
| 20 | 18 | "github.com/bnema/vev/internal/adapters/lifecycle" | |
| 21 | - | "github.com/bnema/vev/internal/adapters/quic" | |
| 22 | 19 | "github.com/bnema/vev/internal/domain" | |
| 23 | 20 | "github.com/bnema/vev/internal/persist" | |
| 24 | 21 | "github.com/bnema/vev/internal/ports" | |
| @@ -343,104 +340,6 @@ func TestLifecycleOwnershipPrecedesDaemonStartup(t *testing.T) { | |||
| 343 | 340 | }) | |
| 344 | 341 | } | |
| 345 | 342 | ||
| 346 | - | // scriptRecv makes a transport yield frames in order, then wait for done | |
| 347 | - | // before returning EOF. The shared done channel coordinates proxy readers | |
| 348 | - | // without relying on scheduling or sleeps. | |
| 349 | - | func TestProxyTransportsCopiesBothDirections(t *testing.T) { | |
| 350 | - | left, right := net.Pipe() | |
| 351 | - | defer func() { _ = left.Close() }() | |
| 352 | - | defer func() { _ = right.Close() }() | |
| 353 | - | client := ipc.NewTransport(left) | |
| 354 | - | daemon := ipc.NewTransport(right) | |
| 355 | - | done := make(chan error, 1) | |
| 356 | - | go func() { done <- proxyTransports(context.Background(), client, daemon, nil) }() | |
| 357 | - | payload := []byte("proxy stays carriage-neutral") | |
| 358 | - | require.NoError(t, client.Send(wire.Envelope{Payload: payload})) | |
| 359 | - | got, err := daemon.Recv() | |
| 360 | - | require.NoError(t, err) | |
| 361 | - | require.Equal(t, payload, got.Payload) | |
| 362 | - | _ = client.Close() | |
| 363 | - | _ = daemon.Close() | |
| 364 | - | } | |
| 365 | - | ||
| 366 | - | // TestQUICProxyLifecycleDeliversFinalEnvelope exercises the production proxy | |
| 367 | - | // bridge instead of a bare same-process Transport.Close. It verifies that a | |
| 368 | - | // client already reading, as production clients are, receives the daemon's | |
| 369 | - | // final synchronous envelope while the proxy tears both carriages down. | |
| 370 | - | func TestQUICProxyLifecycleDeliversFinalEnvelope(t *testing.T) { | |
| 371 | - | cert, fingerprint, err := quic.GenerateEphemeralCert() | |
| 372 | - | require.NoError(t, err) | |
| 373 | - | listener, err := quic.ListenConfig("127.0.0.1:0", cert, quic.Config{}, 4) | |
| 374 | - | require.NoError(t, err) | |
| 375 | - | defer func() { _ = listener.Close() }() | |
| 376 | - | ||
| 377 | - | accepted := make(chan wire.Transport, 1) | |
| 378 | - | go func() { | |
| 379 | - | if transport, err := listener.Accept(); err == nil { | |
| 380 | - | accepted <- transport | |
| 381 | - | } | |
| 382 | - | }() | |
| 383 | - | ||
| 384 | - | ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) | |
| 385 | - | defer cancel() | |
| 386 | - | client, err := quic.DialConfig(listener.Addr(), "vev-bootstrap", fingerprint, quic.Config{}, 5*time.Second).Dial(ctx) | |
| 387 | - | require.NoError(t, err) | |
| 388 | - | defer func() { _ = client.Close() }() | |
| 389 | - | ||
| 390 | - | var proxySide wire.Transport | |
| 391 | - | select { | |
| 392 | - | case proxySide = <-accepted: | |
| 393 | - | case <-time.After(10 * time.Second): | |
| 394 | - | t.Fatal("listener did not accept") | |
| 395 | - | } | |
| 396 | - | ||
| 397 | - | daemonPipe, peerPipe := net.Pipe() | |
| 398 | - | daemon := ipc.NewTransport(daemonPipe) | |
| 399 | - | peer := ipc.NewTransport(peerPipe) | |
| 400 | - | defer func() { _ = peer.Close() }() | |
| 401 | - | ||
| 402 | - | proxyDone := make(chan error, 1) | |
| 403 | - | go func() { proxyDone <- proxyTransports(ctx, proxySide, daemon, nil) }() | |
| 404 | - | ||
| 405 | - | final := []byte("final envelope survives the proxy lifecycle") | |
| 406 | - | require.NoError(t, peer.Send(wire.Envelope{Payload: final})) | |
| 407 | - | _ = peer.Close() | |
| 408 | - | ||
| 409 | - | // Receive while the proxy performs its bounded graceful teardown. Waiting | |
| 410 | - | // for teardown first lets QUIC's eventual CONNECTION_CLOSE discard unread | |
| 411 | - | // stream data, a property the adapter does not promise to prevent. | |
| 412 | - | type recvResult struct { | |
| 413 | - | envelope wire.Envelope | |
| 414 | - | err error | |
| 415 | - | } | |
| 416 | - | received := make(chan recvResult, 1) | |
| 417 | - | go func() { | |
| 418 | - | envelope, err := client.Recv() | |
| 419 | - | received <- recvResult{envelope: envelope, err: err} | |
| 420 | - | }() | |
| 421 | - | var got wire.Envelope | |
| 422 | - | select { | |
| 423 | - | case result := <-received: | |
| 424 | - | require.NoError(t, result.err, "a reading client must receive the final envelope during proxy teardown") | |
| 425 | - | got = result.envelope | |
| 426 | - | case <-ctx.Done(): | |
| 427 | - | t.Fatal("timed out receiving the final envelope during proxy teardown") | |
| 428 | - | } | |
| 429 | - | require.Equal(t, final, got.Payload) | |
| 430 | - | ||
| 431 | - | select { | |
| 432 | - | case <-proxyDone: | |
| 433 | - | case <-time.After(10 * time.Second): | |
| 434 | - | t.Fatal("proxy did not stop after the daemon side closed") | |
| 435 | - | } | |
| 436 | - | ||
| 437 | - | // Verify the bounded teardown phase completed after the proxy closed its | |
| 438 | - | // QUIC carriage; runQUICProxy performs the same wait before process exit. | |
| 439 | - | require.NoError(t, waitGracefulTeardown(proxySide)) | |
| 440 | - | _, err = client.Recv() | |
| 441 | - | require.ErrorIs(t, err, io.EOF) | |
| 442 | - | } | |
| 443 | - | ||
| 444 | 343 | func TestDevelopmentTempDirOption(t *testing.T) { | |
| 445 | 344 | absoluteBase := filepath.Join(t.TempDir(), ".dev") | |
| 446 | 345 | creatorErr := errors.New("cannot secure directory") | |
| @@ -587,13 +486,9 @@ func TestParseArgs(t *testing.T) { | |||
| 587 | 486 | {name: "kill daemon rejects extra arg", args: []string{"kill", "--daemon", "extra"}, wantErr: true}, | |
| 588 | 487 | {name: "kill extra arg", args: []string{"kill", "work", "extra"}, wantErr: true}, | |
| 589 | 488 | {name: "daemon", args: []string{"--daemon"}, wantKind: kindDaemon}, | |
| 590 | - | {name: "stdio", args: []string{"_stdio"}, wantKind: kindStdio}, | |
| 591 | - | {name: "stdio rejects session", args: []string{"_stdio", "work"}, wantErr: true}, | |
| 592 | - | {name: "stdio rejects extra args", args: []string{"_stdio", "work", "extra"}, wantErr: true}, | |
| 593 | - | {name: "quic bootstrap", args: []string{"_quic-bootstrap"}, wantKind: kindQUICBootstrap}, | |
| 594 | - | {name: "quic bootstrap rejects session", args: []string{"_quic-bootstrap", "work"}, wantErr: true}, | |
| 595 | - | {name: "quic proxy", args: []string{"_quic-proxy"}, wantKind: kindQUICProxy}, | |
| 596 | - | {name: "quic proxy rejects session", args: []string{"_quic-proxy", "work"}, wantErr: true}, | |
| 489 | + | {name: "removed stdio", args: []string{"_stdio"}, wantErr: true}, | |
| 490 | + | {name: "removed quic bootstrap", args: []string{"_quic-bootstrap"}, wantErr: true}, | |
| 491 | + | {name: "removed quic proxy", args: []string{"_quic-proxy"}, wantErr: true}, | |
| 597 | 492 | {name: "removed udp bootstrap", args: []string{"_udp-bootstrap"}, wantErr: true}, | |
| 598 | 493 | {name: "removed udp proxy", args: []string{"_udp-proxy"}, wantErr: true}, | |
| 599 | 494 | {name: "removed udp proxy rejects session", args: []string{"_udp-proxy", "work"}, wantErr: true}, | |
| @@ -8,7 +8,6 @@ import ( | |||||
| 8 | "errors" | 8 | "errors" | ||
| 9 | "io" | 9 | "io" | ||
| 10 | "log/slog" | 10 | "log/slog" | ||
| 11 | - | "net" | |||
| 12 | "os" | 11 | "os" | ||
| 13 | "path/filepath" | 12 | "path/filepath" | ||
| 14 | "strings" | 13 | "strings" | ||
| @@ -16,9 +15,7 @@ import ( | |||||
| 16 | "time" | 15 | "time" | ||
| 17 | 16 | ||||
| 18 | "github.com/bnema/vev/internal/adapters/daemonmux" | 17 | "github.com/bnema/vev/internal/adapters/daemonmux" | ||
| 19 | - | "github.com/bnema/vev/internal/adapters/ipc" | |||
| 20 | "github.com/bnema/vev/internal/adapters/lifecycle" | 18 | "github.com/bnema/vev/internal/adapters/lifecycle" | ||
| 21 | - | "github.com/bnema/vev/internal/adapters/quic" | |||
| 22 | "github.com/bnema/vev/internal/domain" | 19 | "github.com/bnema/vev/internal/domain" | ||
| 23 | "github.com/bnema/vev/internal/persist" | 20 | "github.com/bnema/vev/internal/persist" | ||
| 24 | "github.com/bnema/vev/internal/ports" | 21 | "github.com/bnema/vev/internal/ports" | ||
| @@ -343,104 +340,6 @@ func TestLifecycleOwnershipPrecedesDaemonStartup(t *testing.T) { | |||||
| 343 | }) | 340 | }) | ||
| 344 | } | 341 | } | ||
| 345 | 342 | ||||
| 346 | - | // scriptRecv makes a transport yield frames in order, then wait for done | |||
| 347 | - | // before returning EOF. The shared done channel coordinates proxy readers | |||
| 348 | - | // without relying on scheduling or sleeps. | |||
| 349 | - | func TestProxyTransportsCopiesBothDirections(t *testing.T) { | |||
| 350 | - | left, right := net.Pipe() | |||
| 351 | - | defer func() { _ = left.Close() }() | |||
| 352 | - | defer func() { _ = right.Close() }() | |||
| 353 | - | client := ipc.NewTransport(left) | |||
| 354 | - | daemon := ipc.NewTransport(right) | |||
| 355 | - | done := make(chan error, 1) | |||
| 356 | - | go func() { done <- proxyTransports(context.Background(), client, daemon, nil) }() | |||
| 357 | - | payload := []byte("proxy stays carriage-neutral") | |||
| 358 | - | require.NoError(t, client.Send(wire.Envelope{Payload: payload})) | |||
| 359 | - | got, err := daemon.Recv() | |||
| 360 | - | require.NoError(t, err) | |||
| 361 | - | require.Equal(t, payload, got.Payload) | |||
| 362 | - | _ = client.Close() | |||
| 363 | - | _ = daemon.Close() | |||
| 364 | - | } | |||
| 365 | - | ||||
| 366 | - | // TestQUICProxyLifecycleDeliversFinalEnvelope exercises the production proxy | |||
| 367 | - | // bridge instead of a bare same-process Transport.Close. It verifies that a | |||
| 368 | - | // client already reading, as production clients are, receives the daemon's | |||
| 369 | - | // final synchronous envelope while the proxy tears both carriages down. | |||
| 370 | - | func TestQUICProxyLifecycleDeliversFinalEnvelope(t *testing.T) { | |||
| 371 | - | cert, fingerprint, err := quic.GenerateEphemeralCert() | |||
| 372 | - | require.NoError(t, err) | |||
| 373 | - | listener, err := quic.ListenConfig("127.0.0.1:0", cert, quic.Config{}, 4) | |||
| 374 | - | require.NoError(t, err) | |||
| 375 | - | defer func() { _ = listener.Close() }() | |||
| 376 | - | ||||
| 377 | - | accepted := make(chan wire.Transport, 1) | |||
| 378 | - | go func() { | |||
| 379 | - | if transport, err := listener.Accept(); err == nil { | |||
| 380 | - | accepted <- transport | |||
| 381 | - | } | |||
| 382 | - | }() | |||
| 383 | - | ||||
| 384 | - | ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) | |||
| 385 | - | defer cancel() | |||
| 386 | - | client, err := quic.DialConfig(listener.Addr(), "vev-bootstrap", fingerprint, quic.Config{}, 5*time.Second).Dial(ctx) | |||
| 387 | - | require.NoError(t, err) | |||
| 388 | - | defer func() { _ = client.Close() }() | |||
| 389 | - | ||||
| 390 | - | var proxySide wire.Transport | |||
| 391 | - | select { | |||
| 392 | - | case proxySide = <-accepted: | |||
| 393 | - | case <-time.After(10 * time.Second): | |||
| 394 | - | t.Fatal("listener did not accept") | |||
| 395 | - | } | |||
| 396 | - | ||||
| 397 | - | daemonPipe, peerPipe := net.Pipe() | |||
| 398 | - | daemon := ipc.NewTransport(daemonPipe) | |||
| 399 | - | peer := ipc.NewTransport(peerPipe) | |||
| 400 | - | defer func() { _ = peer.Close() }() | |||
| 401 | - | ||||
| 402 | - | proxyDone := make(chan error, 1) | |||
| 403 | - | go func() { proxyDone <- proxyTransports(ctx, proxySide, daemon, nil) }() | |||
| 404 | - | ||||
| 405 | - | final := []byte("final envelope survives the proxy lifecycle") | |||
| 406 | - | require.NoError(t, peer.Send(wire.Envelope{Payload: final})) | |||
| 407 | - | _ = peer.Close() | |||
| 408 | - | ||||
| 409 | - | // Receive while the proxy performs its bounded graceful teardown. Waiting | |||
| 410 | - | // for teardown first lets QUIC's eventual CONNECTION_CLOSE discard unread | |||
| 411 | - | // stream data, a property the adapter does not promise to prevent. | |||
| 412 | - | type recvResult struct { | |||
| 413 | - | envelope wire.Envelope | |||
| 414 | - | err error | |||
| 415 | - | } | |||
| 416 | - | received := make(chan recvResult, 1) | |||
| 417 | - | go func() { | |||
| 418 | - | envelope, err := client.Recv() | |||
| 419 | - | received <- recvResult{envelope: envelope, err: err} | |||
| 420 | - | }() | |||
| 421 | - | var got wire.Envelope | |||
| 422 | - | select { | |||
| 423 | - | case result := <-received: | |||
| 424 | - | require.NoError(t, result.err, "a reading client must receive the final envelope during proxy teardown") | |||
| 425 | - | got = result.envelope | |||
| 426 | - | case <-ctx.Done(): | |||
| 427 | - | t.Fatal("timed out receiving the final envelope during proxy teardown") | |||
| 428 | - | } | |||
| 429 | - | require.Equal(t, final, got.Payload) | |||
| 430 | - | ||||
| 431 | - | select { | |||
| 432 | - | case <-proxyDone: | |||
| 433 | - | case <-time.After(10 * time.Second): | |||
| 434 | - | t.Fatal("proxy did not stop after the daemon side closed") | |||
| 435 | - | } | |||
| 436 | - | ||||
| 437 | - | // Verify the bounded teardown phase completed after the proxy closed its | |||
| 438 | - | // QUIC carriage; runQUICProxy performs the same wait before process exit. | |||
| 439 | - | require.NoError(t, waitGracefulTeardown(proxySide)) | |||
| 440 | - | _, err = client.Recv() | |||
| 441 | - | require.ErrorIs(t, err, io.EOF) | |||
| 442 | - | } | |||
| 443 | - | ||||
| 444 | func TestDevelopmentTempDirOption(t *testing.T) { | 343 | func TestDevelopmentTempDirOption(t *testing.T) { | ||
| 445 | absoluteBase := filepath.Join(t.TempDir(), ".dev") | 344 | absoluteBase := filepath.Join(t.TempDir(), ".dev") | ||
| 446 | creatorErr := errors.New("cannot secure directory") | 345 | creatorErr := errors.New("cannot secure directory") | ||
| @@ -587,13 +486,9 @@ func TestParseArgs(t *testing.T) { | |||||
| 587 | {name: "kill daemon rejects extra arg", args: []string{"kill", "--daemon", "extra"}, wantErr: true}, | 486 | {name: "kill daemon rejects extra arg", args: []string{"kill", "--daemon", "extra"}, wantErr: true}, | ||
| 588 | {name: "kill extra arg", args: []string{"kill", "work", "extra"}, wantErr: true}, | 487 | {name: "kill extra arg", args: []string{"kill", "work", "extra"}, wantErr: true}, | ||
| 589 | {name: "daemon", args: []string{"--daemon"}, wantKind: kindDaemon}, | 488 | {name: "daemon", args: []string{"--daemon"}, wantKind: kindDaemon}, | ||
| 590 | - | {name: "stdio", args: []string{"_stdio"}, wantKind: kindStdio}, | 489 | + | {name: "removed stdio", args: []string{"_stdio"}, wantErr: true}, |
| 591 | - | {name: "stdio rejects session", args: []string{"_stdio", "work"}, wantErr: true}, | 490 | + | {name: "removed quic bootstrap", args: []string{"_quic-bootstrap"}, wantErr: true}, |
| 592 | - | {name: "stdio rejects extra args", args: []string{"_stdio", "work", "extra"}, wantErr: true}, | 491 | + | {name: "removed quic proxy", args: []string{"_quic-proxy"}, wantErr: true}, |
| 593 | - | {name: "quic bootstrap", args: []string{"_quic-bootstrap"}, wantKind: kindQUICBootstrap}, | |||
| 594 | - | {name: "quic bootstrap rejects session", args: []string{"_quic-bootstrap", "work"}, wantErr: true}, | |||
| 595 | - | {name: "quic proxy", args: []string{"_quic-proxy"}, wantKind: kindQUICProxy}, | |||
| 596 | - | {name: "quic proxy rejects session", args: []string{"_quic-proxy", "work"}, wantErr: true}, | |||
| 597 | {name: "removed udp bootstrap", args: []string{"_udp-bootstrap"}, wantErr: true}, | 492 | {name: "removed udp bootstrap", args: []string{"_udp-bootstrap"}, wantErr: true}, | ||
| 598 | {name: "removed udp proxy", args: []string{"_udp-proxy"}, wantErr: true}, | 493 | {name: "removed udp proxy", args: []string{"_udp-proxy"}, wantErr: true}, | ||
| 599 | {name: "removed udp proxy rejects session", args: []string{"_udp-proxy", "work"}, wantErr: true}, | 494 | {name: "removed udp proxy rejects session", args: []string{"_udp-proxy", "work"}, wantErr: true}, | ||
M scripts/remote-picker-harness/README.md +1 -1
| @@ -30,7 +30,7 @@ container. The client needs a pinned `remote` SSH alias and | |||
| 30 | 30 | their clients; the container owner handles session cleanup. | |
| 31 | 31 | ||
| 32 | 32 | Client and daemon debug logging is on (`VEV_LOG=debug`); the remote accepts | |
| 33 | - | `VEV_LOG` over SSH so `_stdio`/`_quic-proxy` and the daemon they spawn log at | |
| 33 | + | `VEV_LOG` over SSH so `_broker-mux-stdio`/`_broker-mux-quic-proxy` and the daemon they spawn log at | |
| 34 | 34 | debug too. After every run, sanitized per-run logs and scenario evidence are | |
| 35 | 35 | copied out of the containers to `/tmp/vev-acceptance-artifacts-*` (override | |
| 36 | 36 | with `VEV_ACCEPTANCE_ARTIFACTS_DIR`). Only structured vev state logs and | |
| @@ -30,7 +30,7 @@ container. The client needs a pinned `remote` SSH alias and | |||||
| 30 | their clients; the container owner handles session cleanup. | 30 | their clients; the container owner handles session cleanup. | ||
| 31 | 31 | ||||
| 32 | Client and daemon debug logging is on (`VEV_LOG=debug`); the remote accepts | 32 | Client and daemon debug logging is on (`VEV_LOG=debug`); the remote accepts | ||
| 33 | - | `VEV_LOG` over SSH so `_stdio`/`_quic-proxy` and the daemon they spawn log at | 33 | + | `VEV_LOG` over SSH so `_broker-mux-stdio`/`_broker-mux-quic-proxy` and the daemon they spawn log at |
| 34 | debug too. After every run, sanitized per-run logs and scenario evidence are | 34 | debug too. After every run, sanitized per-run logs and scenario evidence are | ||
| 35 | copied out of the containers to `/tmp/vev-acceptance-artifacts-*` (override | 35 | copied out of the containers to `/tmp/vev-acceptance-artifacts-*` (override | ||
| 36 | with `VEV_ACCEPTANCE_ARTIFACTS_DIR`). Only structured vev state logs and | 36 | with `VEV_ACCEPTANCE_ARTIFACTS_DIR`). Only structured vev state logs and | ||
M scripts/remote-picker-harness/run.sh +1 -1
| @@ -76,7 +76,7 @@ docker exec --user root "$remote_container" sh -c 'rm -f /etc/ssh/ssh_host_* /ho | |||
| 76 | 76 | docker cp "$context_dir/host" "$remote_container:/etc/ssh/ssh_host_ed25519_key" | |
| 77 | 77 | docker cp "$context_dir/host.pub" "$remote_container:/etc/ssh/ssh_host_ed25519_key.pub" | |
| 78 | 78 | docker cp "$context_dir/client.pub" "$remote_container:/home/demo/.ssh/authorized_keys" | |
| 79 | - | # Accept the debug level the client forwards so remote _stdio/_quic-proxy and | |
| 79 | + | # Accept the debug level the client forwards so remote _broker-mux-stdio/_broker-mux-quic-proxy and | |
| 80 | 80 | # the daemon they spawn log at debug too. | |
| 81 | 81 | docker exec --user root "$remote_container" sh -c 'printf "\nAcceptEnv VEV_LOG\n" >> /etc/ssh/sshd_config; chmod 600 /etc/ssh/ssh_host_ed25519_key /home/demo/.ssh/authorized_keys; chown demo:demo /home/demo/.ssh/authorized_keys; /usr/sbin/sshd' | |
| 82 | 82 | docker cp "$context_dir/client" "$local_container:/home/demo/.ssh/id_ed25519" | |
| @@ -76,7 +76,7 @@ docker exec --user root "$remote_container" sh -c 'rm -f /etc/ssh/ssh_host_* /ho | |||||
| 76 | docker cp "$context_dir/host" "$remote_container:/etc/ssh/ssh_host_ed25519_key" | 76 | docker cp "$context_dir/host" "$remote_container:/etc/ssh/ssh_host_ed25519_key" | ||
| 77 | docker cp "$context_dir/host.pub" "$remote_container:/etc/ssh/ssh_host_ed25519_key.pub" | 77 | docker cp "$context_dir/host.pub" "$remote_container:/etc/ssh/ssh_host_ed25519_key.pub" | ||
| 78 | docker cp "$context_dir/client.pub" "$remote_container:/home/demo/.ssh/authorized_keys" | 78 | docker cp "$context_dir/client.pub" "$remote_container:/home/demo/.ssh/authorized_keys" | ||
| 79 | - | # Accept the debug level the client forwards so remote _stdio/_quic-proxy and | 79 | + | # Accept the debug level the client forwards so remote _broker-mux-stdio/_broker-mux-quic-proxy and |
| 80 | # the daemon they spawn log at debug too. | 80 | # the daemon they spawn log at debug too. | ||
| 81 | docker exec --user root "$remote_container" sh -c 'printf "\nAcceptEnv VEV_LOG\n" >> /etc/ssh/sshd_config; chmod 600 /etc/ssh/ssh_host_ed25519_key /home/demo/.ssh/authorized_keys; chown demo:demo /home/demo/.ssh/authorized_keys; /usr/sbin/sshd' | 81 | docker exec --user root "$remote_container" sh -c 'printf "\nAcceptEnv VEV_LOG\n" >> /etc/ssh/sshd_config; chmod 600 /etc/ssh/ssh_host_ed25519_key /home/demo/.ssh/authorized_keys; chown demo:demo /home/demo/.ssh/authorized_keys; /usr/sbin/sshd' | ||
| 82 | docker cp "$context_dir/client" "$local_container:/home/demo/.ssh/id_ed25519" | 82 | docker cp "$context_dir/client" "$local_container:/home/demo/.ssh/id_ed25519" | ||