1// Command relay is disposable acceptance infrastructure, never a production proxy.
2package main
3
4import (
5 "bufio"
6 "bytes"
7 "context"
8 "encoding/json"
9 "errors"
10 "flag"
11 "fmt"
12 "io"
13 "net"
14 "os"
15 "os/signal"
16 "path/filepath"
17 "sync"
18 "sync/atomic"
19 "syscall"
20 "time"
21
22 "github.com/bnema/vev/internal/adapters/quicnettest"
23)
24
25type command struct {
26 ID uint64 `json:"id"`
27 Op string `json:"op"`
28 Port int `json:"port,omitempty"`
29 UDP bool `json:"udp,omitempty"`
30 TCP bool `json:"tcp,omitempty"`
31 Active bool `json:"active,omitempty"`
32}
33
34type relay struct {
35 ctx context.Context
36 cancel context.CancelFunc
37 remote string
38 listen net.IP
39 degraded bool
40 blackout atomic.Bool
41 bytes atomic.Uint64
42 mu sync.Mutex
43 connections map[net.Conn]struct{}
44 proxies []*quicnettest.Proxy
45 udpBlackout bool
46 wg sync.WaitGroup
47}
48
49func main() {
50 remote := flag.String("remote", "remote", "fixture upstream host")
51 control := flag.String("control", "", "required private Unix control path")
52 // The relay forwards without authentication, so it listens on loopback
53 // unless the disposable fixture asks for every interface.
54 tcp := flag.String("tcp", "127.0.0.1:2222", "SSH listener")
55 listen := flag.String("listen", "127.0.0.1", "UDP proxy listen IP")
56 degraded := flag.Bool("degraded", false, "enable seeded degraded profile")
57 flag.Parse()
58 if *control == "" {
59 fmt.Fprintln(os.Stderr, "-control is required")
60 os.Exit(2)
61 }
62 listenIP := net.ParseIP(*listen)
63 if listenIP == nil {
64 fmt.Fprintln(os.Stderr, "-listen must be an IP address")
65 os.Exit(2)
66 }
67 ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
68 r := &relay{ctx: ctx, cancel: cancel, remote: *remote, listen: listenIP, degraded: *degraded, connections: make(map[net.Conn]struct{})}
69 if err := r.serve(*tcp, *control); err != nil {
70 fmt.Fprintln(os.Stderr, err)
71 os.Exit(1)
72 }
73}
74
75// listenIP is the UDP proxy bind address, loopback when unset.
76func (r *relay) listenIP() net.IP {
77 if r.listen == nil {
78 return net.IPv4(127, 0, 0, 1)
79 }
80 return r.listen
81}
82
83func (r *relay) serve(addr, path string) error {
84 listener, err := net.Listen("tcp", addr)
85 if err != nil {
86 return err
87 }
88 defer listener.Close()
89 parent, err := os.Stat(filepath.Dir(path))
90 if err != nil {
91 return err
92 }
93 if !parent.IsDir() || parent.Mode().Perm() != 0700 {
94 return errors.New("control socket requires a private 0700 directory")
95 }
96 owner, ok := parent.Sys().(*syscall.Stat_t)
97 if !ok || owner.Uid != uint32(os.Getuid()) {
98 return errors.New("control directory must belong to the current user")
99 }
100 control, err := net.Listen("unix", path)
101 if err != nil {
102 return err
103 }
104 defer control.Close()
105 watchDone := make(chan struct{})
106 defer close(watchDone)
107 go func() {
108 select {
109 case <-r.ctx.Done():
110 listener.Close()
111 control.Close()
112 case <-watchDone:
113 }
114 }()
115 defer os.Remove(path)
116 if err := os.Chmod(path, 0600); err != nil {
117 return err
118 }
119 defer func() { listener.Close(); r.close() }()
120 r.wg.Add(1)
121 go func() {
122 defer r.wg.Done()
123 for {
124 conn, err := listener.Accept()
125 if err != nil {
126 return
127 }
128 r.mu.Lock()
129 if len(r.connections) >= 64 {
130 r.mu.Unlock()
131 conn.Close()
132 continue
133 }
134 r.connections[conn] = struct{}{}
135 r.mu.Unlock()
136 r.wg.Add(1)
137 go func() { defer r.wg.Done(); r.forward(conn) }()
138 }
139 }()
140 for {
141 conn, err := control.Accept()
142 if err != nil {
143 if r.ctx.Err() != nil {
144 return nil
145 }
146 return err
147 }
148 // Commands are bounded and handled serially; a stalled caller gets 2s.
149 conn.SetDeadline(time.Now().Add(2 * time.Second))
150 scanner := bufio.NewScanner(conn)
151 scanner.Buffer(make([]byte, 4096), 4096)
152 if scanner.Scan() {
153 var c command
154 err := json.Unmarshal(scanner.Bytes(), &c)
155 var result any
156 if err == nil {
157 result, err = r.apply(c)
158 }
159 reply := map[string]any{"id": c.ID, "result": result}
160 if err != nil {
161 reply["error"] = err.Error()
162 }
163 json.NewEncoder(conn).Encode(reply)
164 }
165 conn.Close()
166 if r.ctx.Err() != nil {
167 listener.Close()
168 return nil
169 }
170 }
171}
172
173func (r *relay) apply(c command) (any, error) {
174 switch c.Op {
175 case "udp":
176 if c.Port <= 0 || c.Port > 65535 {
177 return nil, errors.New("invalid port")
178 }
179 upstream, err := net.ResolveUDPAddr("udp", net.JoinHostPort(r.remote, fmt.Sprint(c.Port)))
180 if err != nil {
181 return nil, err
182 }
183 link := quicnettest.LinkConfig{}
184 if r.degraded {
185 link = quicnettest.LinkConfig{BaseLatency: 150 * time.Millisecond, Jitter: 50 * time.Millisecond, LossPercent: 5}
186 }
187 r.mu.Lock()
188 defer r.mu.Unlock()
189 if len(r.proxies) >= 32 {
190 return nil, errors.New("UDP listener limit")
191 }
192 p, err := quicnettest.New(quicnettest.Config{ServerAddr: upstream, ListenAddr: &net.UDPAddr{IP: r.listenIP()}, Seed: 317, ToServer: link, ToClient: link})
193 if err != nil {
194 return nil, err
195 }
196 p.SetBlackout(r.udpBlackout)
197 r.proxies = append(r.proxies, p)
198 return p.Addr().Port, nil
199 case "blackout":
200 if c.TCP {
201 r.blackout.Store(c.Active)
202 }
203 if c.UDP {
204 r.mu.Lock()
205 r.udpBlackout = c.Active
206 for _, p := range r.proxies {
207 p.SetBlackout(c.Active)
208 }
209 r.mu.Unlock()
210 }
211 return nil, nil
212 case "disconnect":
213 r.mu.Lock()
214 for conn := range r.connections {
215 conn.Close()
216 }
217 r.mu.Unlock()
218 return nil, nil
219 case "stats":
220 r.mu.Lock()
221 defer r.mu.Unlock()
222 stats := make([]quicnettest.Stats, 0, len(r.proxies))
223 for _, p := range r.proxies {
224 stats = append(stats, p.Stats())
225 }
226 return map[string]any{"udp": stats, "tcp_bytes": r.bytes.Load()}, nil
227 case "stop":
228 r.cancel()
229 return nil, nil
230 default:
231 return nil, errors.New("unknown command")
232 }
233}
234
235func (r *relay) forward(client net.Conn) {
236 defer func() { client.Close(); r.mu.Lock(); delete(r.connections, client); r.mu.Unlock() }()
237 server, err := (&net.Dialer{Timeout: 5 * time.Second}).DialContext(r.ctx, "tcp", net.JoinHostPort(r.remote, "22"))
238 if err != nil {
239 return
240 }
241 defer server.Close()
242 r.mu.Lock()
243 r.connections[server] = struct{}{}
244 r.mu.Unlock()
245 defer func() { r.mu.Lock(); delete(r.connections, server); r.mu.Unlock() }()
246 done := make(chan struct{})
247 go func() { r.copy(server, client); server.Close(); client.Close(); close(done) }()
248 r.copy(client, server)
249 server.Close()
250 client.Close()
251 <-done
252}
253
254func (r *relay) wait(delay time.Duration) bool {
255 timer := time.NewTimer(delay)
256 defer timer.Stop()
257 select {
258 case <-timer.C:
259 return true
260 case <-r.ctx.Done():
261 return false
262 }
263}
264
265func (r *relay) copy(dst, src net.Conn) {
266 buf := make([]byte, 32<<10)
267 for {
268 for r.blackout.Load() {
269 if !r.wait(10 * time.Millisecond) {
270 return
271 }
272 }
273 n, err := src.Read(buf)
274 if n > 0 {
275 for r.blackout.Load() {
276 if !r.wait(10 * time.Millisecond) {
277 return
278 }
279 }
280 if r.degraded && !r.wait(150*time.Millisecond+time.Duration(n)*time.Second/(64<<10)) {
281 return
282 }
283 written, writeErr := io.Copy(dst, bytes.NewReader(buf[:n]))
284 r.bytes.Add(uint64(written))
285 if writeErr != nil {
286 return
287 }
288 }
289 if err != nil {
290 return
291 }
292 }
293}
294
295func (r *relay) close() {
296 r.cancel()
297 r.mu.Lock()
298 for conn := range r.connections {
299 conn.Close()
300 }
301 for _, p := range r.proxies {
302 p.Close()
303 }
304 r.mu.Unlock()
305 r.wg.Wait()
306}