1package dgram
2
3import (
4 "bufio"
5 "context"
6 "encoding/base64"
7 "fmt"
8 "io"
9 "log/slog"
10 "net"
11 "os/exec"
12 "strconv"
13 "strings"
14 "time"
15
16 "github.com/bnema/vev/internal/adapters/sshstdio"
17 "github.com/bnema/vev/internal/ports"
18 pdgram "github.com/bnema/vev/pkg/dgram"
19)
20
21const (
22 maxBootstrapStderr = 64 * 1024
23 defaultBootstrapTimeout = 10 * time.Second
24 defaultProbeTimeout = 3 * time.Second
25)
26
27type bootstrapProcess interface {
28 StdoutPipe() (io.ReadCloser, error)
29 Start() error
30 Kill() error
31 Wait() error
32}
33
34type execBootstrapProcess struct{ cmd *exec.Cmd }
35
36func (p execBootstrapProcess) StdoutPipe() (io.ReadCloser, error) { return p.cmd.StdoutPipe() }
37func (p execBootstrapProcess) Start() error { return p.cmd.Start() }
38func (p execBootstrapProcess) Kill() error { return p.cmd.Process.Kill() }
39func (p execBootstrapProcess) Wait() error { return p.cmd.Wait() }
40
41var (
42 startUDPBootstrap = func(ctx context.Context, target, session string, stderr io.Writer) bootstrapProcess {
43 spec := sshstdio.BuildCommandForMode(target, "_udp-bootstrap", session)
44 cmd := exec.CommandContext(ctx, spec.Path, spec.Args...)
45 cmd.Stderr = stderr
46 return execBootstrapProcess{cmd: cmd}
47 }
48 listenUDPPacket = func(ctx context.Context) (net.PacketConn, error) {
49 var lc net.ListenConfig
50 return lc.ListenPacket(ctx, "udp", ":0")
51 }
52 probeUDPTransport = func(ctx context.Context, t *Transport) error { return t.Probe(ctx) }
53)
54
55// RemoteDialFailureKind classifies which part of UDP remote bootstrap failed
56// so client-side recovery policy and tests do not have to parse human text.
57type RemoteDialFailureKind uint8
58
59const (
60 RemoteDialBootstrapUnavailable RemoteDialFailureKind = iota + 1
61 RemoteDialProbeUnreachable
62)
63
64// RemoteDialError wraps a UDP remote bootstrap failure with a stable machine
65// classification while preserving the existing human-facing error text.
66type RemoteDialError struct {
67 Kind RemoteDialFailureKind
68 Action string
69 Err error
70 Hint string
71}
72
73func (e *RemoteDialError) Error() string {
74 base := fmt.Sprintf("remote UDP transport unavailable: %s: %v", e.Action, e.Err)
75 if e.Hint != "" {
76 return base + e.Hint
77 }
78 return base
79}
80
81func (e *RemoteDialError) Unwrap() error { return e.Err }
82
83type limitedBuffer struct {
84 buf []byte
85}
86
87func (b *limitedBuffer) Write(p []byte) (int, error) {
88 if len(b.buf) < maxBootstrapStderr {
89 keep := min(len(p), maxBootstrapStderr-len(b.buf))
90 b.buf = append(b.buf, p[:keep]...)
91 }
92 return len(p), nil
93}
94
95func (b *limitedBuffer) String() string { return string(b.buf) }
96
97// RemoteDialer connects to the authenticated datagram bootstrap for UDP remote
98// attach. SSH stdio is selected only through the explicit stdio transport mode.
99type RemoteDialer struct {
100 Target string
101 Session string
102 BootstrapTimeout time.Duration
103 ProbeTimeout time.Duration
104 Log *slog.Logger
105 RuntimeObserver ports.SerializedRuntimeObserver
106}
107
108func NewRemoteDialer(target, session string) RemoteDialer {
109 return RemoteDialer{Target: target, Session: session, ProbeTimeout: defaultProbeTimeout}
110}
111
112func NewRemoteDialerWithLogger(target, session string, log *slog.Logger) RemoteDialer {
113 return RemoteDialer{Target: target, Session: session, ProbeTimeout: defaultProbeTimeout, Log: log}
114}
115
116func (d RemoteDialer) Dial(ctx context.Context) (ports.Transport, error) {
117 if err := ctx.Err(); err != nil {
118 return nil, err
119 }
120 bootstrapTimeout := d.BootstrapTimeout
121 if bootstrapTimeout <= 0 {
122 bootstrapTimeout = defaultBootstrapTimeout
123 }
124 probeTimeout := d.ProbeTimeout
125 if probeTimeout <= 0 {
126 probeTimeout = defaultProbeTimeout
127 }
128 bootstrapCtx, bootstrapCancel := context.WithTimeout(ctx, bootstrapTimeout)
129 defer bootstrapCancel()
130
131 var stderr limitedBuffer
132 proc := startUDPBootstrap(bootstrapCtx, d.Target, d.Session, &stderr)
133 stdout, err := proc.StdoutPipe()
134 if err != nil {
135 return nil, udpUnavailable("bootstrap stdout", err, &stderr)
136 }
137 if err := proc.Start(); err != nil {
138 _ = stdout.Close()
139 return nil, udpUnavailable("start bootstrap", err, &stderr)
140 }
141 cleanup := true
142 waited := false
143 defer func() {
144 if cleanup {
145 _ = proc.Kill()
146 }
147 if !waited {
148 _ = proc.Wait()
149 }
150 _ = stdout.Close()
151 }()
152
153 ready, err := readUDPReadyContext(bootstrapCtx, stdout)
154 if err != nil {
155 _ = proc.Kill()
156 _ = proc.Wait()
157 waited = true
158 cleanup = false
159 return nil, udpUnavailable("read bootstrap readiness", err, &stderr)
160 }
161 defer pdgram.Erase(ready.key)
162 if err := waitBootstrapContext(bootstrapCtx, proc); err != nil {
163 waited = true
164 cleanup = false
165 return nil, udpUnavailable("wait bootstrap", err, &stderr)
166 }
167 waited = true
168 cleanup = false
169 peer, err := resolveUDPPeer(bootstrapCtx, d.Target, ready.port)
170 if err != nil {
171 return nil, udpUnavailable("resolve UDP peer", err, &stderr)
172 }
173 pc, err := listenUDPPacket(ctx)
174 if err != nil {
175 return nil, udpUnavailable("listen UDP", err, &stderr)
176 }
177 // RebindPacketConn heals NAT rebinds / stale conntrack paths by hopping to a
178 // fresh local UDP socket once the link is silent long enough to probe. The
179 // bind reuses listenUDPPacket so tests can stub it.
180 t, err := NewTransportWithOptions(pc, peer, ready.key, 1, 2, Options{
181 Observe: DiagnosticLogObserver(d.Log),
182 RebindPacketConn: func(net.PacketConn) (net.PacketConn, error) {
183 return listenUDPPacket(ctx)
184 },
185 }, WithRuntimeObserver(d.RuntimeObserver))
186 pdgram.Erase(ready.key)
187 if err != nil {
188 _ = pc.Close()
189 return nil, udpUnavailable("create UDP transport", err, &stderr)
190 }
191 probeCtx, cancel := context.WithTimeout(ctx, probeTimeout)
192 defer cancel()
193 if err := probeUDPTransport(probeCtx, t); err != nil {
194 _ = t.Close()
195 return nil, udpProbeUnreachable(d.Target, err, &stderr)
196 }
197 cleanup = false
198 return t, nil
199}
200
201type udpReadyResult struct {
202 ready udpReady
203 err error
204}
205
206func deliverUDPReady(ctx context.Context, ch chan<- udpReadyResult, res udpReadyResult) {
207 select {
208 case ch <- res:
209 case <-ctx.Done():
210 pdgram.Erase(res.ready.key)
211 }
212}
213
214func readUDPReadyContext(ctx context.Context, r io.Reader) (udpReady, error) {
215 ch := make(chan udpReadyResult)
216 go func() {
217 ready, err := readUDPReady(r)
218 deliverUDPReady(ctx, ch, udpReadyResult{ready: ready, err: err})
219 }()
220 select {
221 case res := <-ch:
222 return res.ready, res.err
223 case <-ctx.Done():
224 return udpReady{}, ctx.Err()
225 }
226}
227
228func waitBootstrapContext(ctx context.Context, proc bootstrapProcess) error {
229 ch := make(chan error, 1)
230 go func() { ch <- proc.Wait() }()
231 select {
232 case err := <-ch:
233 return err
234 case <-ctx.Done():
235 _ = proc.Kill()
236 return ctx.Err()
237 }
238}
239
240func resolveUDPPeer(ctx context.Context, target string, port int) (*net.UDPAddr, error) {
241 host := sshTargetHost(target)
242 if ip := net.ParseIP(host); ip != nil {
243 return &net.UDPAddr{IP: ip, Port: port}, nil
244 }
245 addrs, err := net.DefaultResolver.LookupIPAddr(ctx, host)
246 if err != nil {
247 return nil, err
248 }
249 if len(addrs) == 0 {
250 return nil, fmt.Errorf("no UDP peer addresses for %q", host)
251 }
252 return &net.UDPAddr{IP: addrs[0].IP, Zone: addrs[0].Zone, Port: port}, nil
253}
254
255type udpReady struct {
256 port int
257 key []byte
258}
259
260func readUDPReady(r io.Reader) (udpReady, error) {
261 out := make([]byte, pdgram.KeySize)
262 var (
263 port int
264 keyLen int
265 parseErr error
266 )
267 pdgram.SecretDo(func() {
268 line, err := bufio.NewReader(r).ReadString('\n')
269 if err != nil && line == "" {
270 parseErr = err
271 return
272 }
273
274 fields := strings.Fields(line)
275 if len(fields) != 3 || fields[0] != "VEV-UDP" {
276 parseErr = fmt.Errorf("malformed UDP readiness line")
277 return
278 }
279 port, parseErr = strconv.Atoi(fields[1])
280 if parseErr != nil || port < 1 || port > 65535 {
281 parseErr = fmt.Errorf("invalid UDP port %q", fields[1])
282 return
283 }
284 key, decodeErr := base64.StdEncoding.DecodeString(fields[2])
285 defer pdgram.Erase(key)
286 if decodeErr != nil {
287 parseErr = fmt.Errorf("invalid bootstrap key: %w", decodeErr)
288 return
289 }
290 keyLen = len(key)
291 copy(out, key)
292 })
293 if parseErr != nil {
294 pdgram.Erase(out)
295 return udpReady{}, parseErr
296 }
297 if keyLen != pdgram.KeySize {
298 pdgram.Erase(out)
299 return udpReady{}, fmt.Errorf("invalid bootstrap key length %d", keyLen)
300 }
301 return udpReady{port: port, key: out}, nil
302}
303
304func udpUnavailable(action string, err error, stderr fmt.Stringer) error {
305 return udpDialFailure(RemoteDialBootstrapUnavailable, action, err, stderr, "")
306}
307
308func udpDialFailure(kind RemoteDialFailureKind, action string, err error, stderr fmt.Stringer, hint string) error {
309 if msg := strings.TrimSpace(stderr.String()); msg != "" {
310 hint = fmt.Sprintf(" (stderr: %s)%s", msg, hint)
311 }
312 return &RemoteDialError{Kind: kind, Action: action, Err: err, Hint: hint}
313}
314
315// udpProbeUnreachable wraps a probe failure with actionable guidance. A failed
316// probe almost always means a firewall on the remote is dropping datagrams to
317// vev's UDP proxy rather than a vev bug, so name the two real fixes instead of
318// surfacing a bare deadline error.
319func udpProbeUnreachable(target string, err error, stderr fmt.Stringer) error {
320 return udpDialFailure(RemoteDialProbeUnreachable, "probe UDP transport", err, stderr, fmt.Sprintf("\n"+
321 " the remote host is not reachable over UDP; a firewall is likely dropping the datagrams.\n"+
322 " open UDP on the remote (e.g. `sudo ufw allow in on tailscale0`) or attach over SSH with `VEV_REMOTE_TRANSPORT=stdio vev attach %s`", target))
323}
324
325func sshTargetHost(target string) string {
326 if at := strings.LastIndexByte(target, '@'); at >= 0 {
327 target = target[at+1:]
328 }
329 if h, _, err := net.SplitHostPort(target); err == nil {
330 return h
331 }
332 if strings.HasPrefix(target, "[") && strings.HasSuffix(target, "]") {
333 return strings.TrimPrefix(strings.TrimSuffix(target, "]"), "[")
334 }
335 return target
336}