1package ipc
2
3import (
4 "context"
5 "errors"
6 "net"
7 "os"
8 "path/filepath"
9 "sync"
10 "testing"
11 "time"
12
13 "github.com/bnema/vev/internal/protocol/wire"
14)
15
16// muxTestPath returns a short, owner-only parent directory and one mux.sock
17// path inside it, so a caller-supplied mux carriage path is exercised at the
18// real AF_UNIX pathname limit rather than a long temporary path.
19func muxTestPath(t *testing.T) string {
20 t.Helper()
21 return filepath.Join(shortSocketDir(t, "vev"), "mux.sock")
22}
23
24// mustListenMux binds one mux carriage at path and closes it at test end.
25func mustListenMux(t *testing.T, path string) MuxListener {
26 t.Helper()
27 listener, err := ListenMux(path)
28 if err != nil {
29 t.Fatalf("ListenMux(%q): %v", path, err)
30 }
31 t.Cleanup(func() { _ = listener.Close() })
32 return listener
33}
34
35// dialMuxAsync dials one mux carriage and delivers the outcome.
36func dialMuxAsync(t *testing.T, ctx context.Context, path string) <-chan struct {
37 transport wire.BoundedTransport
38 err error
39} {
40 t.Helper()
41 out := make(chan struct {
42 transport wire.BoundedTransport
43 err error
44 }, 1)
45 go func() {
46 transport, err := DialMuxContext(ctx, path)
47 out <- struct {
48 transport wire.BoundedTransport
49 err error
50 }{transport: transport, err: err}
51 }()
52 return out
53}
54
55func awaitMuxDial(t *testing.T, out <-chan struct {
56 transport wire.BoundedTransport
57 err error
58}) wire.BoundedTransport {
59 t.Helper()
60 select {
61 case result := <-out:
62 if result.err != nil {
63 t.Fatalf("DialMuxContext: %v", result.err)
64 }
65 if result.transport == nil {
66 t.Fatal("DialMuxContext returned a nil transport")
67 }
68 t.Cleanup(func() { _ = result.transport.Close() })
69 return result.transport
70 case <-time.After(5 * time.Second):
71 t.Fatal("DialMuxContext did not return")
72 return nil
73 }
74}
75
76// TestListenMuxCreatesOwnerOnlyParentAndSocket proves listener setup creates the
77// caller's parent directory owner-only (0700) and tightens the bound socket to
78// 0600, so the private carriage is never exposed through filesystem modes.
79func TestListenMuxCreatesOwnerOnlyParentAndSocket(t *testing.T) {
80 path := muxTestPath(t)
81 parent := filepath.Dir(path)
82
83 listener := mustListenMux(t, path)
84 if listener.Addr() != path {
85 t.Fatalf("Addr() = %q, want %q", listener.Addr(), path)
86 }
87
88 dirInfo, err := os.Stat(parent)
89 if err != nil {
90 t.Fatalf("stat parent: %v", err)
91 }
92 if perm := dirInfo.Mode().Perm(); perm != 0o700 {
93 t.Fatalf("parent perm = %o, want 0700", perm)
94 }
95
96 sockInfo, err := os.Lstat(path)
97 if err != nil {
98 t.Fatalf("stat socket: %v", err)
99 }
100 if sockInfo.Mode()&os.ModeSocket == 0 {
101 t.Fatalf("%s is not a socket: mode %v", path, sockInfo.Mode())
102 }
103 if perm := sockInfo.Mode().Perm(); perm != 0o600 {
104 t.Fatalf("socket perm = %o, want 0600", perm)
105 }
106}
107
108// TestListenMuxSameUserRoundTrip proves a real AF_UNIX mux carriage carries
109// bounded framed envelopes both ways after both ends verify the peer is the
110// current user.
111func TestListenMuxSameUserRoundTrip(t *testing.T) {
112 path := muxTestPath(t)
113 listener := mustListenMux(t, path)
114
115 accepted := make(chan struct {
116 transport wire.BoundedTransport
117 err error
118 }, 1)
119 go func() {
120 transport, err := listener.Accept()
121 accepted <- struct {
122 transport wire.BoundedTransport
123 err error
124 }{transport: transport, err: err}
125 }()
126
127 client := awaitMuxDial(t, dialMuxAsync(t, context.Background(), path))
128
129 serverResult := <-accepted
130 if serverResult.err != nil {
131 t.Fatalf("Accept: %v", serverResult.err)
132 }
133 server := serverResult.transport
134 t.Cleanup(func() { _ = server.Close() })
135
136 request := wire.Envelope{Payload: []byte("mux request stays exact")}
137 if err := client.Send(request); err != nil {
138 t.Fatalf("client Send: %v", err)
139 }
140 got, err := server.RecvBounded(uint64(len(request.Payload)))
141 if err != nil {
142 t.Fatalf("server RecvBounded: %v", err)
143 }
144 if string(got.Payload) != string(request.Payload) {
145 t.Fatalf("server RecvBounded payload = %q, want %q", got.Payload, request.Payload)
146 }
147
148 reply := wire.Envelope{Payload: []byte("mux reply stays exact")}
149 if err := server.Send(reply); err != nil {
150 t.Fatalf("server Send: %v", err)
151 }
152 gotReply, err := client.RecvBounded(uint64(len(reply.Payload)))
153 if err != nil {
154 t.Fatalf("client RecvBounded: %v", err)
155 }
156 if string(gotReply.Payload) != string(reply.Payload) {
157 t.Fatalf("client RecvBounded payload = %q, want %q", gotReply.Payload, reply.Payload)
158 }
159}
160
161// TestListenMuxRejectsUnverifiedAccept proves the accept side verifies peer
162// credentials before handing out a transport: an injected refusal is observable
163// as ErrMuxPeerRejected and the connection is closed.
164func TestListenMuxRejectsUnverifiedAccept(t *testing.T) {
165 path := muxTestPath(t)
166 reject := func(*net.UnixConn) error { return ErrMuxPeerRejected }
167 listener, err := listenMux(path, reject)
168 if err != nil {
169 t.Fatalf("listenMux: %v", err)
170 }
171 t.Cleanup(func() { _ = listener.Close() })
172
173 dialOutcome := dialMuxAsync(t, context.Background(), path)
174
175 transport, err := listener.Accept()
176 if transport != nil {
177 _ = transport.Close()
178 t.Fatal("Accept returned a transport for an unverified peer")
179 }
180 if !errors.Is(err, ErrMuxPeerRejected) {
181 t.Fatalf("Accept error = %v, want ErrMuxPeerRejected", err)
182 }
183
184 result := <-dialOutcome
185 if result.transport != nil {
186 _ = result.transport.Close()
187 }
188}
189
190// TestDialMuxRejectsUnverifiedPeer proves the dial side verifies peer
191// credentials before handing out a transport: an injected refusal fails the
192// dial with ErrMuxPeerRejected and no transport escapes.
193func TestDialMuxRejectsUnverifiedPeer(t *testing.T) {
194 path := muxTestPath(t)
195 permissive := func(*net.UnixConn) error { return nil }
196 listener, err := listenMux(path, permissive)
197 if err != nil {
198 t.Fatalf("listenMux: %v", err)
199 }
200 t.Cleanup(func() { _ = listener.Close() })
201
202 accepted := make(chan wire.BoundedTransport, 1)
203 go func() {
204 transport, acceptErr := listener.Accept()
205 if acceptErr == nil {
206 accepted <- transport
207 return
208 }
209 accepted <- nil
210 }()
211
212 reject := func(*net.UnixConn) error { return ErrMuxPeerRejected }
213 transport, err := dialMux(context.Background(), path, reject)
214 if transport != nil {
215 _ = transport.Close()
216 t.Fatal("dialMux returned a transport for an unverified peer")
217 }
218 if !errors.Is(err, ErrMuxPeerRejected) {
219 t.Fatalf("dialMux error = %v, want ErrMuxPeerRejected", err)
220 }
221
222 select {
223 case server := <-accepted:
224 if server != nil {
225 _ = server.Close()
226 }
227 case <-time.After(5 * time.Second):
228 t.Fatal("Accept did not return")
229 }
230}
231
232// createStaleMuxSocket leaves a socket file at path with nobody listening
233// behind it, simulating an owner that died without unlinking its socket.
234func createStaleMuxSocket(t *testing.T, path string) os.FileInfo {
235 t.Helper()
236 if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil {
237 t.Fatalf("mkdir parent: %v", err)
238 }
239 addr, err := net.ResolveUnixAddr("unix", path)
240 if err != nil {
241 t.Fatalf("ResolveUnixAddr: %v", err)
242 }
243 dead, err := net.ListenUnix("unix", addr)
244 if err != nil {
245 t.Fatalf("ListenUnix: %v", err)
246 }
247 dead.SetUnlinkOnClose(false)
248 if err := dead.Close(); err != nil {
249 t.Fatalf("closing dead listener: %v", err)
250 }
251 info, err := os.Lstat(path)
252 if err != nil {
253 t.Fatalf("expected stale socket file: %v", err)
254 }
255 return info
256}
257
258// TestListenMuxStaleSocketRecovered proves a socket file left by a dead owner is
259// recovered race-safely and the carriage binds and serves.
260func TestListenMuxStaleSocketRecovered(t *testing.T) {
261 path := muxTestPath(t)
262 createStaleMuxSocket(t, path)
263 // Mark the stale file: a filesystem such as ext4 may hand the freed inode
264 // number straight to the recovered socket, so inode identity alone cannot
265 // prove the file was replaced.
266 marked := time.Unix(1, 0)
267 if err := os.Chtimes(path, marked, marked); err != nil {
268 t.Fatalf("mark stale socket: %v", err)
269 }
270
271 mustListenMux(t, path)
272
273 recovered, err := os.Lstat(path)
274 if err != nil {
275 t.Fatalf("stat recovered socket: %v", err)
276 }
277 if recovered.ModTime().Equal(marked) {
278 t.Fatal("stale socket was not replaced by the recovered listener")
279 }
280
281 // The recovered carriage serves a same-user connection.
282 _ = awaitMuxDial(t, dialMuxAsync(t, context.Background(), path))
283}
284
285// TestListenMuxStaleRecoveryRacesToOneWinner proves the stale recovery is
286// race-safe: concurrent listeners racing over one stale socket produce exactly
287// one winner, and every loser observes the winner's live socket instead of
288// unlinking it.
289func TestListenMuxStaleRecoveryRacesToOneWinner(t *testing.T) {
290 path := muxTestPath(t)
291 createStaleMuxSocket(t, path)
292
293 const contenders = 8
294 type outcome struct {
295 listener MuxListener
296 err error
297 }
298 results := make(chan outcome, contenders)
299 var wg sync.WaitGroup
300 for i := 0; i < contenders; i++ {
301 wg.Add(1)
302 go func() {
303 defer wg.Done()
304 listener, err := ListenMux(path)
305 results <- outcome{listener: listener, err: err}
306 }()
307 }
308 wg.Wait()
309 close(results)
310
311 successes, live := 0, 0
312 for result := range results {
313 switch {
314 case result.err == nil:
315 successes++
316 t.Cleanup(func() { _ = result.listener.Close() })
317 case errors.Is(result.err, ErrDaemonRunning):
318 live++
319 default:
320 t.Fatalf("unexpected race outcome: %v", result.err)
321 }
322 }
323 if successes != 1 {
324 t.Fatalf("winners = %d, want exactly 1", successes)
325 }
326 if live != contenders-1 {
327 t.Fatalf("losers observing a live owner = %d, want %d", live, contenders-1)
328 }
329}
330
331// TestListenMuxRefusesLiveEndpoint proves a second listener on a live mux path
332// is refused rather than clobbering it.
333func TestListenMuxRefusesLiveEndpoint(t *testing.T) {
334 path := muxTestPath(t)
335 mustListenMux(t, path)
336
337 second, err := ListenMux(path)
338 if second != nil {
339 _ = second.Close()
340 }
341 if !errors.Is(err, ErrDaemonRunning) {
342 t.Fatalf("second ListenMux error = %v, want ErrDaemonRunning", err)
343 }
344}
345
346// TestListenMuxRefusesForeignPath proves a path that exists and is not a socket
347// is refused without being removed: a regular file and a directory are left
348// intact.
349func TestListenMuxRefusesForeignPath(t *testing.T) {
350 t.Run("regular file", func(t *testing.T) {
351 path := muxTestPath(t)
352 if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil {
353 t.Fatalf("mkdir parent: %v", err)
354 }
355 const contents = "caller data"
356 if err := os.WriteFile(path, []byte(contents), 0o600); err != nil {
357 t.Fatalf("write file: %v", err)
358 }
359
360 listener, err := ListenMux(path)
361 if listener != nil {
362 _ = listener.Close()
363 }
364 if !errors.Is(err, ErrMuxForeignPath) {
365 t.Fatalf("ListenMux error = %v, want ErrMuxForeignPath", err)
366 }
367 got, readErr := os.ReadFile(path)
368 if readErr != nil {
369 t.Fatalf("foreign file was removed: %v", readErr)
370 }
371 if string(got) != contents {
372 t.Fatalf("foreign file contents = %q, want %q", got, contents)
373 }
374 })
375
376 t.Run("directory", func(t *testing.T) {
377 path := muxTestPath(t)
378 if err := os.MkdirAll(path, 0o700); err != nil {
379 t.Fatalf("mkdir foreign directory: %v", err)
380 }
381 listener, err := ListenMux(path)
382 if listener != nil {
383 _ = listener.Close()
384 }
385 if !errors.Is(err, ErrMuxForeignPath) {
386 t.Fatalf("ListenMux error = %v, want ErrMuxForeignPath", err)
387 }
388 if info, statErr := os.Stat(path); statErr != nil || !info.IsDir() {
389 t.Fatalf("foreign directory was removed or changed: info=%v err=%v", info, statErr)
390 }
391 })
392}
393
394// TestMuxPathValidation proves invalid caller paths are refused before any
395// filesystem change.
396func TestMuxPathValidation(t *testing.T) {
397 tests := []struct {
398 name string
399 path string
400 }{
401 {name: "empty", path: ""},
402 {name: "relative", path: "mux.sock"},
403 {name: "unclean", path: "/tmp/a/../mux.sock"},
404 {name: "root", path: "/"},
405 {name: "too long", path: "/" + string(make([]byte, muxSocketPathMax))},
406 }
407 for _, tt := range tests {
408 t.Run(tt.name, func(t *testing.T) {
409 if _, err := ListenMux(tt.path); !errors.Is(err, ErrMuxPath) {
410 t.Fatalf("ListenMux(%q) error = %v, want ErrMuxPath", tt.path, err)
411 }
412 if _, err := DialMuxContext(context.Background(), tt.path); !errors.Is(err, ErrMuxPath) {
413 t.Fatalf("DialMuxContext(%q) error = %v, want ErrMuxPath", tt.path, err)
414 }
415 })
416 }
417}
418
419// TestListenMuxCloseUnblocksAcceptAndCleansSocket proves Close unblocks a
420// blocked Accept, unlinks the socket it created, refuses later dials, and is
421// idempotent.
422func TestListenMuxCloseUnblocksAcceptAndCleansSocket(t *testing.T) {
423 path := muxTestPath(t)
424 listener := mustListenMux(t, path)
425
426 acceptErr := make(chan error, 1)
427 go func() {
428 transport, err := listener.Accept()
429 if transport != nil {
430 _ = transport.Close()
431 }
432 acceptErr <- err
433 }()
434
435 if err := listener.Close(); err != nil {
436 t.Fatalf("Close: %v", err)
437 }
438 select {
439 case err := <-acceptErr:
440 if err == nil {
441 t.Fatal("Accept returned nil error after Close")
442 }
443 case <-time.After(5 * time.Second):
444 t.Fatal("Close did not unblock Accept")
445 }
446
447 if _, err := os.Lstat(path); !os.IsNotExist(err) {
448 t.Fatalf("socket still present after Close: %v", err)
449 }
450 if transport, err := DialMuxContext(context.Background(), path); err == nil {
451 _ = transport.Close()
452 t.Fatal("DialMuxContext succeeded after Close")
453 }
454 if err := listener.Close(); err != nil {
455 t.Fatalf("second Close: %v", err)
456 }
457}
458
459// TestDialMuxCancellationAndCloseUnblocksIO proves a canceled dial fails
460// promptly and that closing one end unblocks a blocked read on the other.
461func TestDialMuxCancellationAndCloseUnblocksIO(t *testing.T) {
462 t.Run("canceled context", func(t *testing.T) {
463 ctx, cancel := context.WithCancel(context.Background())
464 cancel()
465 if transport, err := DialMuxContext(ctx, muxTestPath(t)); !errors.Is(err, context.Canceled) {
466 if transport != nil {
467 _ = transport.Close()
468 }
469 t.Fatalf("DialMuxContext error = %v, want context.Canceled", err)
470 }
471 })
472
473 t.Run("close unblocks read", func(t *testing.T) {
474 path := muxTestPath(t)
475 listener := mustListenMux(t, path)
476
477 accepted := make(chan wire.BoundedTransport, 1)
478 go func() {
479 transport, _ := listener.Accept()
480 accepted <- transport
481 }()
482 client := awaitMuxDial(t, dialMuxAsync(t, context.Background(), path))
483 server := <-accepted
484 if server == nil {
485 t.Fatal("Accept did not return a transport")
486 }
487
488 readErr := make(chan error, 1)
489 go func() {
490 _, err := client.Recv()
491 readErr <- err
492 }()
493 if err := server.Close(); err != nil {
494 t.Fatalf("server Close: %v", err)
495 }
496 select {
497 case err := <-readErr:
498 if err == nil {
499 t.Fatal("blocked Recv returned nil error after peer Close")
500 }
501 case <-time.After(5 * time.Second):
502 t.Fatal("peer Close did not unblock Recv")
503 }
504 })
505}
506
507// TestListenMuxConcurrentCloseAndDial proves a Close racing many dials neither
508// panics nor hangs, and every dial result is closed by the test.
509func TestListenMuxConcurrentCloseAndDial(t *testing.T) {
510 path := muxTestPath(t)
511 listener := mustListenMux(t, path)
512
513 const dialers = 16
514 results := make(chan wire.BoundedTransport, dialers)
515 var wg sync.WaitGroup
516 for i := 0; i < dialers; i++ {
517 wg.Add(1)
518 go func() {
519 defer wg.Done()
520 transport, err := DialMuxContext(context.Background(), path)
521 if err != nil {
522 results <- nil
523 return
524 }
525 results <- transport
526 }()
527 }
528 if err := listener.Close(); err != nil {
529 t.Fatalf("Close: %v", err)
530 }
531 wg.Wait()
532 close(results)
533 for transport := range results {
534 if transport != nil {
535 _ = transport.Close()
536 }
537 }
538}
539
540// TestMuxExportedPeerVerifierSeam proves the exported injection seams the broker
541// IPC adapter consumes behave like the internal ones: the platform default
542// admits a same-user peer, a nil verifier falls back to it, and an injected
543// refusal fails closed without handing out a transport.
544func TestMuxExportedPeerVerifierSeam(t *testing.T) {
545 path := muxTestPath(t)
546 listener, err := ListenMuxWithPeerVerifier(path, SameUserPeerVerifier())
547 if err != nil {
548 t.Fatalf("ListenMuxWithPeerVerifier: %v", err)
549 }
550 t.Cleanup(func() { _ = listener.Close() })
551 if SameUserPeerVerifier() == nil {
552 t.Fatal("SameUserPeerVerifier must return the platform verifier")
553 }
554
555 ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
556 defer cancel()
557 transport, err := DialMuxContextWithPeerVerifier(ctx, path, nil)
558 if err != nil {
559 t.Fatalf("DialMuxContextWithPeerVerifier with a nil verifier: %v", err)
560 }
561 t.Cleanup(func() { _ = transport.Close() })
562 accepted, err := listener.Accept()
563 if err != nil {
564 t.Fatalf("Accept: %v", err)
565 }
566 t.Cleanup(func() { _ = accepted.Close() })
567
568 refused, err := DialMuxContextWithPeerVerifier(ctx, path, func(*net.UnixConn) error { return ErrMuxPeerRejected })
569 if err == nil {
570 t.Fatal("an injected refusal must fail the dial")
571 }
572 if !errors.Is(err, ErrMuxPeerRejected) {
573 t.Fatalf("dial refusal = %v, want ErrMuxPeerRejected", err)
574 }
575 if refused != nil {
576 t.Fatal("a refused peer must never receive a transport")
577 }
578}