fix(remote): return actionable udp bootstrap errors
- commit
- 380112a9545e76c3cd53c3026e3d8f59db4363ea
- author
- brice · · 2026-07-06 21:08 UTC
- parent
- f310c98
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
2 files · +244 -5
M internal/adapters/dgram/ssh_dialer.go +110 -5
| @@ -1,12 +1,19 @@ | |||
| 1 | 1 | package dgram | |
| 2 | 2 | ||
| 3 | 3 | import ( | |
| 4 | + | "bufio" | |
| 4 | 5 | "context" | |
| 6 | + | "encoding/base64" | |
| 7 | + | "fmt" | |
| 8 | + | "io" | |
| 5 | 9 | "log/slog" | |
| 6 | 10 | "net" | |
| 11 | + | "os/exec" | |
| 12 | + | "strconv" | |
| 7 | 13 | "strings" | |
| 8 | 14 | "time" | |
| 9 | 15 | ||
| 16 | + | "github.com/bnema/vev/internal/adapters/sshstdio" | |
| 10 | 17 | "github.com/bnema/vev/internal/ports" | |
| 11 | 18 | ) | |
| 12 | 19 | ||
| @@ -15,6 +22,28 @@ const ( | |||
| 15 | 22 | defaultProbeTimeout = 3 * time.Second | |
| 16 | 23 | ) | |
| 17 | 24 | ||
| 25 | + | type bootstrapProcess interface { | |
| 26 | + | StdoutPipe() (io.ReadCloser, error) | |
| 27 | + | Start() error | |
| 28 | + | Kill() error | |
| 29 | + | } | |
| 30 | + | ||
| 31 | + | type execBootstrapProcess struct{ cmd *exec.Cmd } | |
| 32 | + | ||
| 33 | + | func (p execBootstrapProcess) StdoutPipe() (io.ReadCloser, error) { return p.cmd.StdoutPipe() } | |
| 34 | + | func (p execBootstrapProcess) Start() error { return p.cmd.Start() } | |
| 35 | + | func (p execBootstrapProcess) Kill() error { return p.cmd.Process.Kill() } | |
| 36 | + | ||
| 37 | + | var ( | |
| 38 | + | startUDPBootstrap = func(ctx context.Context, target, session string, stderr io.Writer) bootstrapProcess { | |
| 39 | + | spec := sshstdio.BuildCommandForMode(target, "_udp-bootstrap", session) | |
| 40 | + | cmd := exec.CommandContext(ctx, spec.Path, spec.Args...) | |
| 41 | + | cmd.Stderr = stderr | |
| 42 | + | return execBootstrapProcess{cmd: cmd} | |
| 43 | + | } | |
| 44 | + | listenUDPPacket = func() (net.PacketConn, error) { return net.ListenPacket("udp", ":0") } | |
| 45 | + | ) | |
| 46 | + | ||
| 18 | 47 | type limitedBuffer struct { | |
| 19 | 48 | buf []byte | |
| 20 | 49 | } | |
| @@ -47,12 +76,88 @@ func NewRemoteDialerWithLogger(target, session string, log *slog.Logger) RemoteD | |||
| 47 | 76 | } | |
| 48 | 77 | ||
| 49 | 78 | func (d RemoteDialer) Dial(ctx context.Context) (ports.Transport, error) { | |
| 50 | - | select { | |
| 51 | - | case <-ctx.Done(): | |
| 52 | - | return nil, ctx.Err() | |
| 53 | - | default: | |
| 79 | + | if err := ctx.Err(); err != nil { | |
| 80 | + | return nil, err | |
| 81 | + | } | |
| 82 | + | probeTimeout := d.ProbeTimeout | |
| 83 | + | if probeTimeout <= 0 { | |
| 84 | + | probeTimeout = defaultProbeTimeout | |
| 85 | + | } | |
| 86 | + | var stderr limitedBuffer | |
| 87 | + | proc := startUDPBootstrap(ctx, d.Target, d.Session, &stderr) | |
| 88 | + | stdout, err := proc.StdoutPipe() | |
| 89 | + | if err != nil { | |
| 90 | + | return nil, udpUnavailable("bootstrap stdout", err, &stderr) | |
| 91 | + | } | |
| 92 | + | if err := proc.Start(); err != nil { | |
| 93 | + | return nil, udpUnavailable("start bootstrap", err, &stderr) | |
| 94 | + | } | |
| 95 | + | cleanup := true | |
| 96 | + | defer func() { | |
| 97 | + | if cleanup { | |
| 98 | + | _ = proc.Kill() | |
| 99 | + | } | |
| 100 | + | _ = stdout.Close() | |
| 101 | + | }() | |
| 102 | + | ||
| 103 | + | ready, err := readUDPReady(stdout) | |
| 104 | + | if err != nil { | |
| 105 | + | return nil, udpUnavailable("read bootstrap readiness", err, &stderr) | |
| 106 | + | } | |
| 107 | + | peer, err := net.ResolveUDPAddr("udp", net.JoinHostPort(sshTargetHost(d.Target), strconv.Itoa(ready.port))) | |
| 108 | + | if err != nil { | |
| 109 | + | return nil, udpUnavailable("resolve UDP peer", err, &stderr) | |
| 110 | + | } | |
| 111 | + | pc, err := listenUDPPacket() | |
| 112 | + | if err != nil { | |
| 113 | + | return nil, udpUnavailable("listen UDP", err, &stderr) | |
| 114 | + | } | |
| 115 | + | t, err := NewTransport(pc, peer, ready.key, 1, 2) | |
| 116 | + | if err != nil { | |
| 117 | + | _ = pc.Close() | |
| 118 | + | return nil, udpUnavailable("create UDP transport", err, &stderr) | |
| 119 | + | } | |
| 120 | + | probeCtx, cancel := context.WithTimeout(ctx, probeTimeout) | |
| 121 | + | defer cancel() | |
| 122 | + | if err := t.Probe(probeCtx); err != nil { | |
| 123 | + | _ = t.Close() | |
| 124 | + | return nil, udpUnavailable("probe UDP transport", err, &stderr) | |
| 125 | + | } | |
| 126 | + | cleanup = false | |
| 127 | + | return t, nil | |
| 128 | + | } | |
| 129 | + | ||
| 130 | + | type udpReady struct { | |
| 131 | + | port int | |
| 132 | + | key []byte | |
| 133 | + | } | |
| 134 | + | ||
| 135 | + | func readUDPReady(r io.Reader) (udpReady, error) { | |
| 136 | + | line, err := bufio.NewReader(r).ReadString('\n') | |
| 137 | + | if err != nil && line == "" { | |
| 138 | + | return udpReady{}, err | |
| 139 | + | } | |
| 140 | + | fields := strings.Fields(line) | |
| 141 | + | if len(fields) != 3 || fields[0] != "VEV-UDP" { | |
| 142 | + | return udpReady{}, fmt.Errorf("malformed readiness line %q", strings.TrimSpace(line)) | |
| 143 | + | } | |
| 144 | + | port, err := strconv.Atoi(fields[1]) | |
| 145 | + | if err != nil || port < 1 || port > 65535 { | |
| 146 | + | return udpReady{}, fmt.Errorf("invalid UDP port %q", fields[1]) | |
| 147 | + | } | |
| 148 | + | key, err := base64.StdEncoding.DecodeString(fields[2]) | |
| 149 | + | if err != nil { | |
| 150 | + | return udpReady{}, fmt.Errorf("invalid bootstrap key: %w", err) | |
| 151 | + | } | |
| 152 | + | return udpReady{port: port, key: key}, nil | |
| 153 | + | } | |
| 154 | + | ||
| 155 | + | func udpUnavailable(action string, err error, stderr fmt.Stringer) error { | |
| 156 | + | msg := strings.TrimSpace(stderr.String()) | |
| 157 | + | if msg != "" { | |
| 158 | + | return fmt.Errorf("vev: remote UDP transport unavailable: %s: %w (stderr: %s)", action, err, msg) | |
| 54 | 159 | } | |
| 55 | - | return nil, ErrLinkDead | |
| 160 | + | return fmt.Errorf("vev: remote UDP transport unavailable: %s: %w", action, err) | |
| 56 | 161 | } | |
| 57 | 162 | ||
| 58 | 163 | func sshTargetHost(target string) string { | |
| @@ -1,12 +1,19 @@ | |||||
| 1 | package dgram | 1 | package dgram | ||
| 2 | 2 | ||||
| 3 | import ( | 3 | import ( | ||
| 4 | + | "bufio" | |||
| 4 | "context" | 5 | "context" | ||
| 6 | + | "encoding/base64" | |||
| 7 | + | "fmt" | |||
| 8 | + | "io" | |||
| 5 | "log/slog" | 9 | "log/slog" | ||
| 6 | "net" | 10 | "net" | ||
| 11 | + | "os/exec" | |||
| 12 | + | "strconv" | |||
| 7 | "strings" | 13 | "strings" | ||
| 8 | "time" | 14 | "time" | ||
| 9 | 15 | ||||
| 16 | + | "github.com/bnema/vev/internal/adapters/sshstdio" | |||
| 10 | "github.com/bnema/vev/internal/ports" | 17 | "github.com/bnema/vev/internal/ports" | ||
| 11 | ) | 18 | ) | ||
| 12 | 19 | ||||
| @@ -15,6 +22,28 @@ const ( | |||||
| 15 | defaultProbeTimeout = 3 * time.Second | 22 | defaultProbeTimeout = 3 * time.Second | ||
| 16 | ) | 23 | ) | ||
| 17 | 24 | ||||
| 25 | + | type bootstrapProcess interface { | |||
| 26 | + | StdoutPipe() (io.ReadCloser, error) | |||
| 27 | + | Start() error | |||
| 28 | + | Kill() error | |||
| 29 | + | } | |||
| 30 | + | ||||
| 31 | + | type execBootstrapProcess struct{ cmd *exec.Cmd } | |||
| 32 | + | ||||
| 33 | + | func (p execBootstrapProcess) StdoutPipe() (io.ReadCloser, error) { return p.cmd.StdoutPipe() } | |||
| 34 | + | func (p execBootstrapProcess) Start() error { return p.cmd.Start() } | |||
| 35 | + | func (p execBootstrapProcess) Kill() error { return p.cmd.Process.Kill() } | |||
| 36 | + | ||||
| 37 | + | var ( | |||
| 38 | + | startUDPBootstrap = func(ctx context.Context, target, session string, stderr io.Writer) bootstrapProcess { | |||
| 39 | + | spec := sshstdio.BuildCommandForMode(target, "_udp-bootstrap", session) | |||
| 40 | + | cmd := exec.CommandContext(ctx, spec.Path, spec.Args...) | |||
| 41 | + | cmd.Stderr = stderr | |||
| 42 | + | return execBootstrapProcess{cmd: cmd} | |||
| 43 | + | } | |||
| 44 | + | listenUDPPacket = func() (net.PacketConn, error) { return net.ListenPacket("udp", ":0") } | |||
| 45 | + | ) | |||
| 46 | + | ||||
| 18 | type limitedBuffer struct { | 47 | type limitedBuffer struct { | ||
| 19 | buf []byte | 48 | buf []byte | ||
| 20 | } | 49 | } | ||
| @@ -47,12 +76,88 @@ func NewRemoteDialerWithLogger(target, session string, log *slog.Logger) RemoteD | |||||
| 47 | } | 76 | } | ||
| 48 | 77 | ||||
| 49 | func (d RemoteDialer) Dial(ctx context.Context) (ports.Transport, error) { | 78 | func (d RemoteDialer) Dial(ctx context.Context) (ports.Transport, error) { | ||
| 50 | - | select { | 79 | + | if err := ctx.Err(); err != nil { |
| 51 | - | case <-ctx.Done(): | 80 | + | return nil, err |
| 52 | - | return nil, ctx.Err() | 81 | + | } |
| 53 | - | default: | 82 | + | probeTimeout := d.ProbeTimeout |
| 83 | + | if probeTimeout <= 0 { | |||
| 84 | + | probeTimeout = defaultProbeTimeout | |||
| 85 | + | } | |||
| 86 | + | var stderr limitedBuffer | |||
| 87 | + | proc := startUDPBootstrap(ctx, d.Target, d.Session, &stderr) | |||
| 88 | + | stdout, err := proc.StdoutPipe() | |||
| 89 | + | if err != nil { | |||
| 90 | + | return nil, udpUnavailable("bootstrap stdout", err, &stderr) | |||
| 91 | + | } | |||
| 92 | + | if err := proc.Start(); err != nil { | |||
| 93 | + | return nil, udpUnavailable("start bootstrap", err, &stderr) | |||
| 94 | + | } | |||
| 95 | + | cleanup := true | |||
| 96 | + | defer func() { | |||
| 97 | + | if cleanup { | |||
| 98 | + | _ = proc.Kill() | |||
| 99 | + | } | |||
| 100 | + | _ = stdout.Close() | |||
| 101 | + | }() | |||
| 102 | + | ||||
| 103 | + | ready, err := readUDPReady(stdout) | |||
| 104 | + | if err != nil { | |||
| 105 | + | return nil, udpUnavailable("read bootstrap readiness", err, &stderr) | |||
| 106 | + | } | |||
| 107 | + | peer, err := net.ResolveUDPAddr("udp", net.JoinHostPort(sshTargetHost(d.Target), strconv.Itoa(ready.port))) | |||
| 108 | + | if err != nil { | |||
| 109 | + | return nil, udpUnavailable("resolve UDP peer", err, &stderr) | |||
| 110 | + | } | |||
| 111 | + | pc, err := listenUDPPacket() | |||
| 112 | + | if err != nil { | |||
| 113 | + | return nil, udpUnavailable("listen UDP", err, &stderr) | |||
| 114 | + | } | |||
| 115 | + | t, err := NewTransport(pc, peer, ready.key, 1, 2) | |||
| 116 | + | if err != nil { | |||
| 117 | + | _ = pc.Close() | |||
| 118 | + | return nil, udpUnavailable("create UDP transport", err, &stderr) | |||
| 119 | + | } | |||
| 120 | + | probeCtx, cancel := context.WithTimeout(ctx, probeTimeout) | |||
| 121 | + | defer cancel() | |||
| 122 | + | if err := t.Probe(probeCtx); err != nil { | |||
| 123 | + | _ = t.Close() | |||
| 124 | + | return nil, udpUnavailable("probe UDP transport", err, &stderr) | |||
| 125 | + | } | |||
| 126 | + | cleanup = false | |||
| 127 | + | return t, nil | |||
| 128 | + | } | |||
| 129 | + | ||||
| 130 | + | type udpReady struct { | |||
| 131 | + | port int | |||
| 132 | + | key []byte | |||
| 133 | + | } | |||
| 134 | + | ||||
| 135 | + | func readUDPReady(r io.Reader) (udpReady, error) { | |||
| 136 | + | line, err := bufio.NewReader(r).ReadString('\n') | |||
| 137 | + | if err != nil && line == "" { | |||
| 138 | + | return udpReady{}, err | |||
| 139 | + | } | |||
| 140 | + | fields := strings.Fields(line) | |||
| 141 | + | if len(fields) != 3 || fields[0] != "VEV-UDP" { | |||
| 142 | + | return udpReady{}, fmt.Errorf("malformed readiness line %q", strings.TrimSpace(line)) | |||
| 143 | + | } | |||
| 144 | + | port, err := strconv.Atoi(fields[1]) | |||
| 145 | + | if err != nil || port < 1 || port > 65535 { | |||
| 146 | + | return udpReady{}, fmt.Errorf("invalid UDP port %q", fields[1]) | |||
| 147 | + | } | |||
| 148 | + | key, err := base64.StdEncoding.DecodeString(fields[2]) | |||
| 149 | + | if err != nil { | |||
| 150 | + | return udpReady{}, fmt.Errorf("invalid bootstrap key: %w", err) | |||
| 151 | + | } | |||
| 152 | + | return udpReady{port: port, key: key}, nil | |||
| 153 | + | } | |||
| 154 | + | ||||
| 155 | + | func udpUnavailable(action string, err error, stderr fmt.Stringer) error { | |||
| 156 | + | msg := strings.TrimSpace(stderr.String()) | |||
| 157 | + | if msg != "" { | |||
| 158 | + | return fmt.Errorf("vev: remote UDP transport unavailable: %s: %w (stderr: %s)", action, err, msg) | |||
| 54 | } | 159 | } | ||
| 55 | - | return nil, ErrLinkDead | 160 | + | return fmt.Errorf("vev: remote UDP transport unavailable: %s: %w", action, err) |
| 56 | } | 161 | } | ||
| 57 | 162 | ||||
| 58 | func sshTargetHost(target string) string { | 163 | func sshTargetHost(target string) string { | ||
M internal/adapters/dgram/ssh_dialer_test.go +134 -0
| @@ -2,11 +2,16 @@ package dgram | |||
| 2 | 2 | ||
| 3 | 3 | import ( | |
| 4 | 4 | "context" | |
| 5 | + | "encoding/base64" | |
| 5 | 6 | "errors" | |
| 7 | + | "io" | |
| 8 | + | "net" | |
| 6 | 9 | "strings" | |
| 7 | 10 | "testing" | |
| 11 | + | "time" | |
| 8 | 12 | ||
| 9 | 13 | "github.com/bnema/vev/internal/ports" | |
| 14 | + | pdgram "github.com/bnema/vev/pkg/dgram" | |
| 10 | 15 | ) | |
| 11 | 16 | ||
| 12 | 17 | func TestRemoteDialerFactorySelectsExplicitModes(t *testing.T) { | |
| @@ -77,6 +82,135 @@ func TestStdioDialerUsesContextBeforeStartingSSH(t *testing.T) { | |||
| 77 | 82 | } | |
| 78 | 83 | } | |
| 79 | 84 | ||
| 85 | + | type fakeBootstrapProcess struct { | |
| 86 | + | stdout io.ReadCloser | |
| 87 | + | startErr error | |
| 88 | + | killed bool | |
| 89 | + | } | |
| 90 | + | ||
| 91 | + | func (p *fakeBootstrapProcess) StdoutPipe() (io.ReadCloser, error) { return p.stdout, nil } | |
| 92 | + | func (p *fakeBootstrapProcess) Start() error { return p.startErr } | |
| 93 | + | func (p *fakeBootstrapProcess) Kill() error { p.killed = true; return nil } | |
| 94 | + | ||
| 95 | + | func withBootstrapStarter(t *testing.T, fn func(context.Context, string, string, io.Writer) bootstrapProcess) { | |
| 96 | + | t.Helper() | |
| 97 | + | old := startUDPBootstrap | |
| 98 | + | startUDPBootstrap = fn | |
| 99 | + | t.Cleanup(func() { startUDPBootstrap = old }) | |
| 100 | + | } | |
| 101 | + | ||
| 102 | + | func withListenUDP(t *testing.T, fn func() (net.PacketConn, error)) { | |
| 103 | + | t.Helper() | |
| 104 | + | old := listenUDPPacket | |
| 105 | + | listenUDPPacket = fn | |
| 106 | + | t.Cleanup(func() { listenUDPPacket = old }) | |
| 107 | + | } | |
| 108 | + | ||
| 109 | + | func TestRemoteDialerUDPBootstrapFailures(t *testing.T) { | |
| 110 | + | key := base64.StdEncoding.EncodeToString(make([]byte, pdgram.KeySize)) | |
| 111 | + | tests := []struct { | |
| 112 | + | name string | |
| 113 | + | stdout string | |
| 114 | + | start error | |
| 115 | + | listen error | |
| 116 | + | want string | |
| 117 | + | killed bool | |
| 118 | + | }{ | |
| 119 | + | {name: "start failure", start: errors.New("ssh missing"), want: "vev: remote UDP transport unavailable: start bootstrap: ssh missing"}, | |
| 120 | + | {name: "malformed readiness", stdout: "hello\n", want: "malformed readiness line", killed: true}, | |
| 121 | + | {name: "packet listen failure", stdout: "VEV-UDP 4444 " + key + "\n", listen: errors.New("bind denied"), want: "listen UDP: bind denied", killed: true}, | |
| 122 | + | } | |
| 123 | + | for _, tt := range tests { | |
| 124 | + | t.Run(tt.name, func(t *testing.T) { | |
| 125 | + | proc := &fakeBootstrapProcess{stdout: io.NopCloser(strings.NewReader(tt.stdout)), startErr: tt.start} | |
| 126 | + | withBootstrapStarter(t, func(context.Context, string, string, io.Writer) bootstrapProcess { return proc }) | |
| 127 | + | if tt.listen != nil { | |
| 128 | + | withListenUDP(t, func() (net.PacketConn, error) { return nil, tt.listen }) | |
| 129 | + | } | |
| 130 | + | tr, err := NewRemoteDialer("127.0.0.1", "work").Dial(context.Background()) | |
| 131 | + | if err == nil || !strings.Contains(err.Error(), tt.want) { | |
| 132 | + | t.Fatalf("Dial() error = %v, want containing %q", err, tt.want) | |
| 133 | + | } | |
| 134 | + | if tr != nil { | |
| 135 | + | t.Fatalf("Dial() transport = %T, want nil", tr) | |
| 136 | + | } | |
| 137 | + | if proc.killed != tt.killed { | |
| 138 | + | t.Fatalf("killed=%v, want %v", proc.killed, tt.killed) | |
| 139 | + | } | |
| 140 | + | }) | |
| 141 | + | } | |
| 142 | + | } | |
| 143 | + | ||
| 144 | + | func TestRemoteDialerProbeFailureCleansUpWithoutStdioFallback(t *testing.T) { | |
| 145 | + | key := base64.StdEncoding.EncodeToString(make([]byte, pdgram.KeySize)) | |
| 146 | + | proc := &fakeBootstrapProcess{stdout: io.NopCloser(strings.NewReader("VEV-UDP 4444 " + key + "\n"))} | |
| 147 | + | withBootstrapStarter(t, func(_ context.Context, target, session string, _ io.Writer) bootstrapProcess { | |
| 148 | + | if target != "127.0.0.1" || session != "work" { | |
| 149 | + | t.Fatalf("bootstrap target/session = %q/%q", target, session) | |
| 150 | + | } | |
| 151 | + | return proc | |
| 152 | + | }) | |
| 153 | + | pc, err := net.ListenPacket("udp", "127.0.0.1:0") | |
| 154 | + | if err != nil { | |
| 155 | + | t.Fatal(err) | |
| 156 | + | } | |
| 157 | + | closed := false | |
| 158 | + | withListenUDP(t, func() (net.PacketConn, error) { return closeTrackPacketConn{PacketConn: pc, closed: &closed}, nil }) | |
| 159 | + | d := NewRemoteDialer("127.0.0.1", "work") | |
| 160 | + | d.ProbeTimeout = time.Nanosecond | |
| 161 | + | tr, err := d.Dial(context.Background()) | |
| 162 | + | if err == nil || !strings.Contains(err.Error(), "vev: remote UDP transport unavailable: probe UDP transport") { | |
| 163 | + | t.Fatalf("Dial() error = %v, want probe unavailable", err) | |
| 164 | + | } | |
| 165 | + | if tr != nil { | |
| 166 | + | t.Fatalf("transport=%T, want nil", tr) | |
| 167 | + | } | |
| 168 | + | if !closed { | |
| 169 | + | t.Fatal("packet conn not closed after probe failure") | |
| 170 | + | } | |
| 171 | + | if !proc.killed { | |
| 172 | + | t.Fatal("bootstrap process not killed after probe failure") | |
| 173 | + | } | |
| 174 | + | } | |
| 175 | + | ||
| 176 | + | type closeTrackPacketConn struct { | |
| 177 | + | net.PacketConn | |
| 178 | + | closed *bool | |
| 179 | + | } | |
| 180 | + | ||
| 181 | + | func (c closeTrackPacketConn) Close() error { *c.closed = true; return c.PacketConn.Close() } | |
| 182 | + | ||
| 183 | + | func TestReadUDPReady(t *testing.T) { | |
| 184 | + | key := base64.StdEncoding.EncodeToString(make([]byte, pdgram.KeySize)) | |
| 185 | + | tests := []struct { | |
| 186 | + | name, line string | |
| 187 | + | wantPort int | |
| 188 | + | wantErr string | |
| 189 | + | }{ | |
| 190 | + | {name: "valid", line: "VEV-UDP 1234 " + key + "\n", wantPort: 1234}, | |
| 191 | + | {name: "bad prefix", line: "NOPE 1234 " + key + "\n", wantErr: "malformed readiness"}, | |
| 192 | + | {name: "bad port", line: "VEV-UDP 99999 " + key + "\n", wantErr: "invalid UDP port"}, | |
| 193 | + | {name: "bad key", line: "VEV-UDP 1234 !!!\n", wantErr: "invalid bootstrap key"}, | |
| 194 | + | } | |
| 195 | + | for _, tt := range tests { | |
| 196 | + | t.Run(tt.name, func(t *testing.T) { | |
| 197 | + | got, err := readUDPReady(strings.NewReader(tt.line)) | |
| 198 | + | if tt.wantErr != "" { | |
| 199 | + | if err == nil || !strings.Contains(err.Error(), tt.wantErr) { | |
| 200 | + | t.Fatalf("err=%v, want %q", err, tt.wantErr) | |
| 201 | + | } | |
| 202 | + | return | |
| 203 | + | } | |
| 204 | + | if err != nil { | |
| 205 | + | t.Fatal(err) | |
| 206 | + | } | |
| 207 | + | if got.port != tt.wantPort || len(got.key) != pdgram.KeySize { | |
| 208 | + | t.Fatalf("ready=%+v keylen=%d", got, len(got.key)) | |
| 209 | + | } | |
| 210 | + | }) | |
| 211 | + | } | |
| 212 | + | } | |
| 213 | + | ||
| 80 | 214 | func TestLimitedBufferCapsBootstrapStderr(t *testing.T) { | |
| 81 | 215 | var b limitedBuffer | |
| 82 | 216 | chunk := strings.Repeat("x", maxBootstrapStderr+1024) | |
| @@ -2,11 +2,16 @@ package dgram | |||||
| 2 | 2 | ||||
| 3 | import ( | 3 | import ( | ||
| 4 | "context" | 4 | "context" | ||
| 5 | + | "encoding/base64" | |||
| 5 | "errors" | 6 | "errors" | ||
| 7 | + | "io" | |||
| 8 | + | "net" | |||
| 6 | "strings" | 9 | "strings" | ||
| 7 | "testing" | 10 | "testing" | ||
| 11 | + | "time" | |||
| 8 | 12 | ||||
| 9 | "github.com/bnema/vev/internal/ports" | 13 | "github.com/bnema/vev/internal/ports" | ||
| 14 | + | pdgram "github.com/bnema/vev/pkg/dgram" | |||
| 10 | ) | 15 | ) | ||
| 11 | 16 | ||||
| 12 | func TestRemoteDialerFactorySelectsExplicitModes(t *testing.T) { | 17 | func TestRemoteDialerFactorySelectsExplicitModes(t *testing.T) { | ||
| @@ -77,6 +82,135 @@ func TestStdioDialerUsesContextBeforeStartingSSH(t *testing.T) { | |||||
| 77 | } | 82 | } | ||
| 78 | } | 83 | } | ||
| 79 | 84 | ||||
| 85 | + | type fakeBootstrapProcess struct { | |||
| 86 | + | stdout io.ReadCloser | |||
| 87 | + | startErr error | |||
| 88 | + | killed bool | |||
| 89 | + | } | |||
| 90 | + | ||||
| 91 | + | func (p *fakeBootstrapProcess) StdoutPipe() (io.ReadCloser, error) { return p.stdout, nil } | |||
| 92 | + | func (p *fakeBootstrapProcess) Start() error { return p.startErr } | |||
| 93 | + | func (p *fakeBootstrapProcess) Kill() error { p.killed = true; return nil } | |||
| 94 | + | ||||
| 95 | + | func withBootstrapStarter(t *testing.T, fn func(context.Context, string, string, io.Writer) bootstrapProcess) { | |||
| 96 | + | t.Helper() | |||
| 97 | + | old := startUDPBootstrap | |||
| 98 | + | startUDPBootstrap = fn | |||
| 99 | + | t.Cleanup(func() { startUDPBootstrap = old }) | |||
| 100 | + | } | |||
| 101 | + | ||||
| 102 | + | func withListenUDP(t *testing.T, fn func() (net.PacketConn, error)) { | |||
| 103 | + | t.Helper() | |||
| 104 | + | old := listenUDPPacket | |||
| 105 | + | listenUDPPacket = fn | |||
| 106 | + | t.Cleanup(func() { listenUDPPacket = old }) | |||
| 107 | + | } | |||
| 108 | + | ||||
| 109 | + | func TestRemoteDialerUDPBootstrapFailures(t *testing.T) { | |||
| 110 | + | key := base64.StdEncoding.EncodeToString(make([]byte, pdgram.KeySize)) | |||
| 111 | + | tests := []struct { | |||
| 112 | + | name string | |||
| 113 | + | stdout string | |||
| 114 | + | start error | |||
| 115 | + | listen error | |||
| 116 | + | want string | |||
| 117 | + | killed bool | |||
| 118 | + | }{ | |||
| 119 | + | {name: "start failure", start: errors.New("ssh missing"), want: "vev: remote UDP transport unavailable: start bootstrap: ssh missing"}, | |||
| 120 | + | {name: "malformed readiness", stdout: "hello\n", want: "malformed readiness line", killed: true}, | |||
| 121 | + | {name: "packet listen failure", stdout: "VEV-UDP 4444 " + key + "\n", listen: errors.New("bind denied"), want: "listen UDP: bind denied", killed: true}, | |||
| 122 | + | } | |||
| 123 | + | for _, tt := range tests { | |||
| 124 | + | t.Run(tt.name, func(t *testing.T) { | |||
| 125 | + | proc := &fakeBootstrapProcess{stdout: io.NopCloser(strings.NewReader(tt.stdout)), startErr: tt.start} | |||
| 126 | + | withBootstrapStarter(t, func(context.Context, string, string, io.Writer) bootstrapProcess { return proc }) | |||
| 127 | + | if tt.listen != nil { | |||
| 128 | + | withListenUDP(t, func() (net.PacketConn, error) { return nil, tt.listen }) | |||
| 129 | + | } | |||
| 130 | + | tr, err := NewRemoteDialer("127.0.0.1", "work").Dial(context.Background()) | |||
| 131 | + | if err == nil || !strings.Contains(err.Error(), tt.want) { | |||
| 132 | + | t.Fatalf("Dial() error = %v, want containing %q", err, tt.want) | |||
| 133 | + | } | |||
| 134 | + | if tr != nil { | |||
| 135 | + | t.Fatalf("Dial() transport = %T, want nil", tr) | |||
| 136 | + | } | |||
| 137 | + | if proc.killed != tt.killed { | |||
| 138 | + | t.Fatalf("killed=%v, want %v", proc.killed, tt.killed) | |||
| 139 | + | } | |||
| 140 | + | }) | |||
| 141 | + | } | |||
| 142 | + | } | |||
| 143 | + | ||||
| 144 | + | func TestRemoteDialerProbeFailureCleansUpWithoutStdioFallback(t *testing.T) { | |||
| 145 | + | key := base64.StdEncoding.EncodeToString(make([]byte, pdgram.KeySize)) | |||
| 146 | + | proc := &fakeBootstrapProcess{stdout: io.NopCloser(strings.NewReader("VEV-UDP 4444 " + key + "\n"))} | |||
| 147 | + | withBootstrapStarter(t, func(_ context.Context, target, session string, _ io.Writer) bootstrapProcess { | |||
| 148 | + | if target != "127.0.0.1" || session != "work" { | |||
| 149 | + | t.Fatalf("bootstrap target/session = %q/%q", target, session) | |||
| 150 | + | } | |||
| 151 | + | return proc | |||
| 152 | + | }) | |||
| 153 | + | pc, err := net.ListenPacket("udp", "127.0.0.1:0") | |||
| 154 | + | if err != nil { | |||
| 155 | + | t.Fatal(err) | |||
| 156 | + | } | |||
| 157 | + | closed := false | |||
| 158 | + | withListenUDP(t, func() (net.PacketConn, error) { return closeTrackPacketConn{PacketConn: pc, closed: &closed}, nil }) | |||
| 159 | + | d := NewRemoteDialer("127.0.0.1", "work") | |||
| 160 | + | d.ProbeTimeout = time.Nanosecond | |||
| 161 | + | tr, err := d.Dial(context.Background()) | |||
| 162 | + | if err == nil || !strings.Contains(err.Error(), "vev: remote UDP transport unavailable: probe UDP transport") { | |||
| 163 | + | t.Fatalf("Dial() error = %v, want probe unavailable", err) | |||
| 164 | + | } | |||
| 165 | + | if tr != nil { | |||
| 166 | + | t.Fatalf("transport=%T, want nil", tr) | |||
| 167 | + | } | |||
| 168 | + | if !closed { | |||
| 169 | + | t.Fatal("packet conn not closed after probe failure") | |||
| 170 | + | } | |||
| 171 | + | if !proc.killed { | |||
| 172 | + | t.Fatal("bootstrap process not killed after probe failure") | |||
| 173 | + | } | |||
| 174 | + | } | |||
| 175 | + | ||||
| 176 | + | type closeTrackPacketConn struct { | |||
| 177 | + | net.PacketConn | |||
| 178 | + | closed *bool | |||
| 179 | + | } | |||
| 180 | + | ||||
| 181 | + | func (c closeTrackPacketConn) Close() error { *c.closed = true; return c.PacketConn.Close() } | |||
| 182 | + | ||||
| 183 | + | func TestReadUDPReady(t *testing.T) { | |||
| 184 | + | key := base64.StdEncoding.EncodeToString(make([]byte, pdgram.KeySize)) | |||
| 185 | + | tests := []struct { | |||
| 186 | + | name, line string | |||
| 187 | + | wantPort int | |||
| 188 | + | wantErr string | |||
| 189 | + | }{ | |||
| 190 | + | {name: "valid", line: "VEV-UDP 1234 " + key + "\n", wantPort: 1234}, | |||
| 191 | + | {name: "bad prefix", line: "NOPE 1234 " + key + "\n", wantErr: "malformed readiness"}, | |||
| 192 | + | {name: "bad port", line: "VEV-UDP 99999 " + key + "\n", wantErr: "invalid UDP port"}, | |||
| 193 | + | {name: "bad key", line: "VEV-UDP 1234 !!!\n", wantErr: "invalid bootstrap key"}, | |||
| 194 | + | } | |||
| 195 | + | for _, tt := range tests { | |||
| 196 | + | t.Run(tt.name, func(t *testing.T) { | |||
| 197 | + | got, err := readUDPReady(strings.NewReader(tt.line)) | |||
| 198 | + | if tt.wantErr != "" { | |||
| 199 | + | if err == nil || !strings.Contains(err.Error(), tt.wantErr) { | |||
| 200 | + | t.Fatalf("err=%v, want %q", err, tt.wantErr) | |||
| 201 | + | } | |||
| 202 | + | return | |||
| 203 | + | } | |||
| 204 | + | if err != nil { | |||
| 205 | + | t.Fatal(err) | |||
| 206 | + | } | |||
| 207 | + | if got.port != tt.wantPort || len(got.key) != pdgram.KeySize { | |||
| 208 | + | t.Fatalf("ready=%+v keylen=%d", got, len(got.key)) | |||
| 209 | + | } | |||
| 210 | + | }) | |||
| 211 | + | } | |||
| 212 | + | } | |||
| 213 | + | ||||
| 80 | func TestLimitedBufferCapsBootstrapStderr(t *testing.T) { | 214 | func TestLimitedBufferCapsBootstrapStderr(t *testing.T) { | ||
| 81 | var b limitedBuffer | 215 | var b limitedBuffer | ||
| 82 | chunk := strings.Repeat("x", maxBootstrapStderr+1024) | 216 | chunk := strings.Repeat("x", maxBootstrapStderr+1024) | ||