1package sshstdio
2
3import (
4 "bytes"
5 "context"
6 "encoding/binary"
7 "errors"
8 "io"
9 "log/slog"
10 "os/exec"
11 "strings"
12 "sync"
13 "testing"
14 "time"
15
16 "github.com/bnema/vev/internal/adapters/sessionwire"
17 "github.com/bnema/vev/internal/domain"
18 "github.com/bnema/vev/internal/protocol"
19 "github.com/bnema/vev/internal/protocol/wire"
20)
21
22// frameHeaderLen is the 4-byte big-endian length prefix; envelopes carry no
23// legacy type byte.
24const frameHeaderLen = 4
25
26func TestBuildCommandForRemoteCommandQuotesEveryWord(t *testing.T) {
27 tests := []struct {
28 name string
29 target string
30 command []string
31 want []string
32 }{
33 {
34 name: "catalog command",
35 target: "arch",
36 command: []string{"vev", "cmd", "remote-catalog", "--json"},
37 want: []string{"--", "arch", "'vev' 'cmd' 'remote-catalog' '--json'"},
38 },
39 {
40 name: "metacharacters in remote words are quoted",
41 target: "user@host; touch /tmp/pwn",
42 command: []string{"vev", "cmd", "remote-catalog; rm -rf /", "--json"},
43 want: []string{"--", "user@host; touch /tmp/pwn", "'vev' 'cmd' 'remote-catalog; rm -rf /' '--json'"},
44 },
45 {
46 name: "single quotes are posix escaped",
47 target: "arch",
48 command: []string{"it's", "fine"},
49 want: []string{"--", "arch", "'it'\\''s' 'fine'"},
50 },
51 }
52
53 for _, tt := range tests {
54 t.Run(tt.name, func(t *testing.T) {
55 got := BuildCommandForRemoteCommand(tt.target, tt.command...)
56 if got.Path != "ssh" {
57 t.Fatalf("Path = %q, want ssh", got.Path)
58 }
59 if len(got.Args) != len(tt.want) {
60 t.Fatalf("Args len = %d, want %d (%q)", len(got.Args), len(tt.want), got.Args)
61 }
62 for i := range tt.want {
63 if got.Args[i] != tt.want[i] {
64 t.Fatalf("Args[%d] = %q, want %q (all args %q)", i, got.Args[i], tt.want[i], got.Args)
65 }
66 }
67 })
68 }
69}
70
71func TestCloseInterruptsBlockedSend(t *testing.T) {
72 writer := &blockedWriter{started: make(chan struct{}), released: make(chan struct{})}
73 transport := NewTransport(nil, writer, nil)
74
75 sendDone := make(chan error, 1)
76 go func() { sendDone <- transport.Send(wire.Envelope{Payload: []byte("blocked")}) }()
77 <-writer.started
78 if err := transport.Close(); err != nil {
79 t.Fatalf("Close() error = %v", err)
80 }
81 if err := <-sendDone; err == nil {
82 t.Fatal("Send() error = nil after Close()")
83 }
84}
85
86type blockedWriter struct {
87 started chan struct{}
88 released chan struct{}
89 once sync.Once
90}
91
92func (w *blockedWriter) Write([]byte) (int, error) {
93 w.once.Do(func() { close(w.started) })
94 <-w.released
95 return 0, io.ErrClosedPipe
96}
97
98func (w *blockedWriter) Close() error {
99 select {
100 case <-w.released:
101 default:
102 close(w.released)
103 }
104 return nil
105}
106
107func TestTransportRoundTripAndVersionMismatchFrame(t *testing.T) {
108 clientRead, serverWrite := io.Pipe()
109 serverRead, clientWrite := io.Pipe()
110 client := NewTransport(clientRead, clientWrite, func() error {
111 _ = clientRead.Close()
112 return clientWrite.Close()
113 })
114 server := NewTransport(serverRead, serverWrite, func() error {
115 _ = serverRead.Close()
116 return serverWrite.Close()
117 })
118 defer func() { _ = client.Close() }()
119 defer func() { _ = server.Close() }()
120
121 hello := protocol.Hello{Version: protocol.Version + 1, Intent: protocol.IntentAttach, Name: "work", Size: domain.Size{Cols: 80, Rows: 24}}
122 go func() {
123 f, err := server.Recv()
124 if err != nil {
125 t.Errorf("server Recv: %v", err)
126 return
127 }
128 decoded, err := sessionwire.DecodeClientEnvelope(f.Payload)
129 if err != nil {
130 t.Errorf("DecodeClientEnvelope: %v", err)
131 return
132 }
133 got, ok := decoded.(protocol.Hello)
134 if !ok {
135 t.Errorf("decoded client message = %T, want protocol.Hello", decoded)
136 return
137 }
138 if got.Version == protocol.Version {
139 t.Errorf("test did not send a mismatched version")
140 }
141 errPayload, err := sessionwire.EncodeServerMessage(protocol.ErrorMsg{Code: protocol.ErrVersionMismatch, Text: "protocol version mismatch"})
142 if err != nil {
143 t.Errorf("EncodeServerMessage: %v", err)
144 return
145 }
146 _ = server.Send(wire.Envelope{Payload: errPayload})
147 }()
148
149 helloPayload, err := sessionwire.EncodeClientMessage(hello)
150 if err != nil {
151 t.Fatalf("EncodeClientMessage: %v", err)
152 }
153 if err := client.Send(wire.Envelope{Payload: helloPayload}); err != nil {
154 t.Fatalf("client Send: %v", err)
155 }
156 reply, err := client.Recv()
157 if err != nil {
158 t.Fatalf("client Recv: %v", err)
159 }
160 decodedReply, err := sessionwire.DecodeServerEnvelope(reply.Payload)
161 if err != nil {
162 t.Fatalf("DecodeServerEnvelope: %v", err)
163 }
164 em, ok := decodedReply.(protocol.ErrorMsg)
165 if !ok {
166 t.Fatalf("decoded server message = %T, want protocol.ErrorMsg", decodedReply)
167 }
168 if em.Code != protocol.ErrVersionMismatch {
169 t.Fatalf("error code = %d, want ErrVersionMismatch", em.Code)
170 }
171}
172
173func TestTransportRejectsZeroLengthFrame(t *testing.T) {
174 tr := NewTransport(strings.NewReader("\x00\x00\x00\x00"), io.Discard, nil)
175 _, err := tr.Recv()
176 if !errors.Is(err, ErrZeroLengthFrame) {
177 t.Fatalf("Recv error = %v, want ErrZeroLengthFrame", err)
178 }
179}
180
181func TestTransportUsesCanonicalFrameMaximum(t *testing.T) {
182 boundaryPayload := make([]byte, wire.AbsoluteEnvelopeLimit-1)
183 boundaryWire := &bytes.Buffer{}
184 boundarySend := NewTransport(nil, boundaryWire, nil)
185 if err := boundarySend.Send(wire.Envelope{Payload: boundaryPayload}); err != nil {
186 t.Fatalf("boundary Send error = %v", err)
187 }
188 if got := binary.BigEndian.Uint32(boundaryWire.Bytes()[:frameHeaderLen]); got != uint32(len(boundaryPayload)) {
189 t.Fatalf("boundary envelope length = %d, want %d", got, len(boundaryPayload))
190 }
191 boundaryRecv := NewTransport(bytes.NewReader(boundaryWire.Bytes()), io.Discard, nil)
192 boundaryFrame, err := boundaryRecv.Recv()
193 if err != nil {
194 t.Fatalf("boundary Recv error = %v", err)
195 }
196 if len(boundaryFrame.Payload) != len(boundaryPayload) {
197 t.Fatalf("boundary envelope = payload %d bytes; want %d bytes", len(boundaryFrame.Payload), len(boundaryPayload))
198 }
199 if !bytes.Equal(boundaryFrame.Payload, boundaryPayload) {
200 t.Fatal("boundary payload was corrupted")
201 }
202
203 send := NewTransport(nil, io.Discard, nil)
204 err = send.Send(wire.Envelope{Payload: make([]byte, wire.AbsoluteEnvelopeLimit+1)})
205 if !errors.Is(err, ErrFrameTooLarge) {
206 t.Fatalf("Send error = %v, want ErrFrameTooLarge", err)
207 }
208
209 var header [frameHeaderLen]byte
210 binary.BigEndian.PutUint32(header[:], wire.AbsoluteEnvelopeLimit+1)
211 recv := NewTransport(bytes.NewReader(header[:]), io.Discard, nil)
212 _, err = recv.Recv()
213 if !errors.Is(err, ErrFrameTooLarge) {
214 t.Fatalf("Recv error = %v, want ErrFrameTooLarge", err)
215 }
216}
217
218func TestRecvMapsUnexpectedEOFDuringHeaderToSSHExit(t *testing.T) {
219 sshErr := errors.New("ssh exited with status 255")
220 tr := newTransport(strings.NewReader("\x00"), io.Discard, nil, func() error { return sshErr })
221
222 _, err := tr.Recv()
223 if !errors.Is(err, sshErr) {
224 t.Fatalf("Recv error = %v, want %v", err, sshErr)
225 }
226}
227
228func TestRecvReportsSSHExitWhenProcessClosesBeforeFrame(t *testing.T) {
229 cmd := exec.CommandContext(context.Background(), "sh", "-c", "echo connection refused >&2; exit 255")
230 stdin, err := cmd.StdinPipe()
231 if err != nil {
232 t.Fatalf("StdinPipe: %v", err)
233 }
234 stdout, err := cmd.StdoutPipe()
235 if err != nil {
236 t.Fatalf("StdoutPipe: %v", err)
237 }
238 var stderr bytes.Buffer
239 cmd.Stderr = &stderr
240 if err := cmd.Start(); err != nil {
241 t.Fatalf("Start: %v", err)
242 }
243
244 waiter := newProcessWaiter(cmd, stdin, &stderr, time.Second, nil, "user@example.com", "work")
245 tr := newTransport(stdout, stdin, waiter.close, waiter.eofErr)
246
247 _, err = tr.Recv()
248 if err == nil {
249 t.Fatal("Recv error = nil, want ssh exit error")
250 }
251 for _, want := range []string{"sshstdio: ssh exited:", "connection refused"} {
252 if !strings.Contains(err.Error(), want) {
253 t.Fatalf("Recv error = %q, want substring %q", err, want)
254 }
255 }
256}
257
258func TestProcessCloserLogsNonCleanExitStderr(t *testing.T) {
259 cmd := exec.Command("sh", "-c", "echo remote failure >&2; exit 7")
260 stdin, err := cmd.StdinPipe()
261 if err != nil {
262 t.Fatalf("StdinPipe: %v", err)
263 }
264 var stderr bytes.Buffer
265 cmd.Stderr = &stderr
266 if err := cmd.Start(); err != nil {
267 t.Fatalf("Start: %v", err)
268 }
269
270 var logBuf bytes.Buffer
271 log := slog.New(slog.NewJSONHandler(&logBuf, nil))
272 err = newProcessCloser(cmd, stdin, &stderr, time.Second, log, "user@example.com", "work")()
273 if err == nil {
274 t.Fatal("Close error = nil, want non-clean ssh exit")
275 }
276 entry := logBuf.String()
277 for _, want := range []string{"ssh exited non-cleanly", "remote failure", "user@example.com", "work"} {
278 if !strings.Contains(entry, want) {
279 t.Fatalf("log entry = %q, want substring %q", entry, want)
280 }
281 }
282 if strings.Contains(entry, "'vev' '_stdio'") || strings.Contains(entry, "-- user@example.com") {
283 t.Fatalf("log entry includes generated command line: %q", entry)
284 }
285}
286
287func TestProcessCloser(t *testing.T) {
288 tests := []struct {
289 name string
290 cmd *exec.Cmd
291 timeout time.Duration
292 wantErrs []string
293 wantBound bool
294 }{
295 {
296 name: "wedged process is killed after timeout and reaped",
297 cmd: exec.Command("sleep", "30"),
298 timeout: 50 * time.Millisecond,
299 wantErrs: []string{"sshstdio: ssh exited:"},
300 wantBound: true,
301 },
302 {
303 name: "stderr from failing shell command is included in error",
304 cmd: exec.Command("sh", "-c", "echo kaboom >&2; exit 7"),
305 timeout: time.Second,
306 wantErrs: []string{"sshstdio: ssh exited:", "kaboom"},
307 },
308 }
309
310 for _, tt := range tests {
311 t.Run(tt.name, func(t *testing.T) {
312 stdin, err := tt.cmd.StdinPipe()
313 if err != nil {
314 t.Fatalf("StdinPipe: %v", err)
315 }
316 var stderr bytes.Buffer
317 tt.cmd.Stderr = &stderr
318 if err := tt.cmd.Start(); err != nil {
319 t.Fatalf("Start: %v", err)
320 }
321
322 started := time.Now()
323 err = newProcessCloser(tt.cmd, stdin, &stderr, tt.timeout, nil, "", "")()
324 elapsed := time.Since(started)
325
326 if err == nil {
327 t.Fatalf("Close error = nil, want error containing %q", tt.wantErrs)
328 }
329 for _, want := range tt.wantErrs {
330 if !strings.Contains(err.Error(), want) {
331 t.Fatalf("Close error = %q, want substring %q", err.Error(), want)
332 }
333 }
334 if tt.wantBound && elapsed > time.Second {
335 t.Fatalf("Close took %s, want bounded below 1s", elapsed)
336 }
337 if tt.cmd.ProcessState == nil {
338 t.Fatalf("ProcessState is nil, want child reaped")
339 }
340 })
341 }
342}
343
344// TestTransportRecvBoundedEnforcesCallerLimit proves the SSH stdio transport
345// exposes the shared bounded receive: an over-limit length prefix is refused
346// before the body is read, a zero limit refuses a non-empty frame, and an
347// in-limit frame round-trips byte for byte.
348func TestTransportRecvBoundedEnforcesCallerLimit(t *testing.T) {
349 var header [frameHeaderLen]byte
350
351 t.Run("over-limit prefix refused before body", func(t *testing.T) {
352 binary.BigEndian.PutUint32(header[:], 8)
353 recv := NewTransport(bytes.NewReader(header[:]), io.Discard, nil).(wire.BoundedTransport)
354 if _, err := recv.RecvBounded(4); !errors.Is(err, ErrFrameTooLarge) {
355 t.Fatalf("RecvBounded error = %v, want ErrFrameTooLarge", err)
356 }
357 })
358
359 t.Run("zero limit refuses non-empty frame", func(t *testing.T) {
360 binary.BigEndian.PutUint32(header[:], 1)
361 recv := NewTransport(bytes.NewReader(header[:]), io.Discard, nil).(wire.BoundedTransport)
362 if _, err := recv.RecvBounded(0); !errors.Is(err, ErrFrameTooLarge) {
363 t.Fatalf("RecvBounded error = %v, want ErrFrameTooLarge", err)
364 }
365 })
366
367 t.Run("in-limit frame round trips", func(t *testing.T) {
368 payload := []byte("bounded ssh envelope")
369 var wireBuf bytes.Buffer
370 binary.BigEndian.PutUint32(header[:], uint32(len(payload)))
371 wireBuf.Write(header[:])
372 wireBuf.Write(payload)
373 recv := NewTransport(bytes.NewReader(wireBuf.Bytes()), io.Discard, nil).(wire.BoundedTransport)
374 envelope, err := recv.RecvBounded(uint64(len(payload)))
375 if err != nil {
376 t.Fatalf("RecvBounded error = %v", err)
377 }
378 if !bytes.Equal(envelope.Payload, payload) {
379 t.Fatalf("RecvBounded payload = %q, want %q", envelope.Payload, payload)
380 }
381 })
382}