1// SPDX-License-Identifier: MIT
2
3// Package tcprelay forwards TCP connections accepted on a sandbox's gateway
4// address to one fixed target. It carries the guest forge access when the
5// forge does not listen on the runner's own network namespace: a runner in a
6// container reaches the forge by name on its container network, and the
7// sandboxes, which only reach their gateway, are relayed there.
8package tcprelay
9
10import (
11 "context"
12 "errors"
13 "fmt"
14 "io"
15 "net"
16 "net/netip"
17 "strconv"
18 "sync"
19 "syscall"
20 "time"
21
22 "github.com/bnema/zerowrap"
23 "golang.org/x/sys/unix"
24)
25
26// MaxConns bounds the connections relayed at once for one listener: a guest
27// cannot exhaust the runner's file descriptors.
28const MaxConns = 64
29
30// dialTimeout bounds the connection to the target.
31const dialTimeout = 10 * time.Second
32
33// saturationLogEvery bounds the warnings about refused connections: a guest
34// that floods its gateway must not flood the runner's log.
35const saturationLogEvery = 10 * time.Second
36
37// Relay forwards every connection accepted on its listeners to Target.
38type Relay struct {
39 // Target is host:port, resolved again for every connection.
40 Target string
41 // Dial connects to Target; nil means a net.Dialer.
42 Dial func(ctx context.Context, network, addr string) (net.Conn, error)
43 // FreeBind binds with IP_FREEBIND, so that the gateway address may be
44 // configured after Listen (podman's netavark creates the bridge address
45 // when the first container starts).
46 FreeBind bool
47
48 mu sync.Mutex
49 listeners map[netip.Addr]*listener
50}
51
52type listener struct {
53 ln net.Listener
54 cancel context.CancelFunc
55 done sync.WaitGroup
56 slots chan struct{}
57}
58
59// New returns a relay to target (host:port).
60func New(target string) *Relay {
61 return &Relay{Target: target, listeners: map[netip.Addr]*listener{}}
62}
63
64// Listen starts relaying connections to addr:port. It returns once the socket
65// is bound. Listening twice on the same address is an error.
66func (r *Relay) Listen(ctx context.Context, addr netip.Addr, port uint16) error {
67 r.mu.Lock()
68 defer r.mu.Unlock()
69 if r.listeners == nil {
70 r.listeners = map[netip.Addr]*listener{}
71 }
72 if _, ok := r.listeners[addr]; ok {
73 return fmt.Errorf("tcprelay: already listening on %s", addr)
74 }
75 var lc net.ListenConfig
76 if r.FreeBind {
77 lc.Control = func(_, _ string, c syscall.RawConn) error {
78 var serr error
79 if err := c.Control(func(fd uintptr) { serr = unix.SetsockoptInt(int(fd), unix.IPPROTO_IP, unix.IP_FREEBIND, 1) }); err != nil {
80 return err
81 }
82 return serr
83 }
84 }
85 ln, err := lc.Listen(ctx, "tcp", net.JoinHostPort(addr.String(), strconv.Itoa(int(port))))
86 if err != nil {
87 return fmt.Errorf("tcprelay: listen %s:%d: %w", addr, port, err)
88 }
89 lctx, cancel := context.WithCancel(context.WithoutCancel(ctx))
90 l := &listener{ln: ln, cancel: cancel, slots: make(chan struct{}, MaxConns)}
91 r.listeners[addr] = l
92 l.done.Add(1)
93 go func() {
94 defer l.done.Done()
95 r.serve(lctx, l)
96 }()
97 return nil
98}
99
100// Addr returns the bound address for addr, useful when port is 0.
101func (r *Relay) Addr(addr netip.Addr) (net.Addr, bool) {
102 r.mu.Lock()
103 defer r.mu.Unlock()
104 l, ok := r.listeners[addr]
105 if !ok {
106 return nil, false
107 }
108 return l.ln.Addr(), true
109}
110
111// Close stops the listener of addr and its relayed connections. It is a no-op
112// when there is none.
113func (r *Relay) Close(addr netip.Addr) {
114 r.mu.Lock()
115 l, ok := r.listeners[addr]
116 delete(r.listeners, addr)
117 r.mu.Unlock()
118 if ok {
119 l.stop()
120 }
121}
122
123// CloseAll stops every listener.
124func (r *Relay) CloseAll() {
125 r.mu.Lock()
126 ls := r.listeners
127 r.listeners = map[netip.Addr]*listener{}
128 r.mu.Unlock()
129 for _, l := range ls {
130 l.stop()
131 }
132}
133
134func (l *listener) stop() {
135 l.cancel()
136 _ = l.ln.Close()
137 l.done.Wait()
138}
139
140func (r *Relay) serve(ctx context.Context, l *listener) {
141 log := zerowrap.FromCtx(ctx)
142 var (
143 refused int
144 lastWarn time.Time
145 )
146 for {
147 c, err := l.ln.Accept()
148 if err != nil {
149 if ctx.Err() != nil || errors.Is(err, net.ErrClosed) {
150 return
151 }
152 log.Warn().Err(err).Msg("tcprelay: accept")
153 continue
154 }
155 select {
156 case l.slots <- struct{}{}:
157 default:
158 _ = c.Close()
159 refused++
160 if now := time.Now(); now.Sub(lastWarn) >= saturationLogEvery {
161 log.Warn().Str("target", r.Target).Int("refused", refused).Int("max", MaxConns).Msg("tcprelay: too many relayed connections, refusing")
162 lastWarn, refused = now, 0
163 }
164 continue
165 }
166 l.done.Add(1)
167 go func() {
168 defer l.done.Done()
169 defer func() { <-l.slots }()
170 r.relay(ctx, c)
171 }()
172 }
173}
174
175func (r *Relay) relay(ctx context.Context, in net.Conn) {
176 defer func() { _ = in.Close() }()
177 dial := r.Dial
178 if dial == nil {
179 d := net.Dialer{}
180 dial = d.DialContext
181 }
182 dctx, cancel := context.WithTimeout(ctx, dialTimeout)
183 out, err := dial(dctx, "tcp", r.Target)
184 cancel()
185 if err != nil {
186 log := zerowrap.FromCtx(ctx)
187 log.Warn().Err(err).Str("target", r.Target).Msg("tcprelay: dial")
188 return
189 }
190 defer func() { _ = out.Close() }()
191 // Closing both ends when the listener stops ends the copies.
192 stop := context.AfterFunc(ctx, func() { _ = in.Close(); _ = out.Close() })
193 defer stop()
194 var wg sync.WaitGroup
195 wg.Add(2)
196 go func() { defer wg.Done(); pipe(out, in) }()
197 go func() { defer wg.Done(); pipe(in, out) }()
198 wg.Wait()
199}
200
201// pipe copies src to dst, then half-closes dst so the other side sees the end
202// of the stream while the reverse direction may still be sending.
203func pipe(dst, src net.Conn) {
204 _, _ = io.Copy(dst, src)
205 if cw, ok := dst.(interface{ CloseWrite() error }); ok {
206 _ = cw.CloseWrite()
207 } else {
208 _ = dst.Close()
209 }
210}