| 1 | // Package tunnelclient is the host half of the reverse channel (issue #53 / |
| 2 | // C6 Step 3): dial OUT to the edge, invert the connection, and serve a local |
| 3 | // http.Handler over HTTP/2 on it. Reconnect fast — hq's dual-WAN failover and |
| 4 | // a Pi's Wi-Fi blip are both "the conn died, dial again", and that latency is |
| 5 | // user-visible. |
| 6 | // |
| 7 | // It is deliberately a LEAF package: stdlib plus golang.org/x/net only, no |
| 8 | // CARLOS imports. internal/host imports internal/edge, so the edge package's |
| 9 | // own tests cannot import internal/host — and the acceptance test for this |
| 10 | // channel (internal/edge/tunnel_e2e_test.go) has to drive the REAL client |
| 11 | // loop, backoff and all, or the reconnect bound it asserts would be measuring |
| 12 | // a test double. Keeping dial/serve/backoff here is what lets both sides use |
| 13 | // one implementation. internal/host/tunnel_client.go is now a thin wrapper: |
| 14 | // it reads the env and stores the state this package reports. |
| 15 | package tunnelclient |
| 16 | |
| 17 | import ( |
| 18 | "bufio" |
| 19 | "context" |
| 20 | "crypto/tls" |
| 21 | "fmt" |
| 22 | "io" |
| 23 | "log" |
| 24 | "math/rand/v2" |
| 25 | "net" |
| 26 | "net/http" |
| 27 | "net/url" |
| 28 | "time" |
| 29 | |
| 30 | "golang.org/x/net/http2" |
| 31 | ) |
| 32 | |
| 33 | // The two states Loop reports through Config.OnState. "" (never reported |
| 34 | // here) means the loop is not running at all — probes must fail toward honest |
| 35 | // values, so the caller's zero value has to be distinguishable from anything |
| 36 | // this package says. |
| 37 | const ( |
| 38 | StateConnected = "connected" |
| 39 | StateReconnecting = "reconnecting" |
| 40 | ) |
| 41 | |
| 42 | const ( |
| 43 | BackoffFloor = 250 * time.Millisecond |
| 44 | BackoffCap = 2 * time.Second |
| 45 | // StableConnected is how long a connection has to have stayed up before |
| 46 | // its loss resets backoff to the floor. Below this, the disconnect is |
| 47 | // treated like a dial failure and backoff keeps escalating: the edge's |
| 48 | // re-register-replaces-the-previous-channel semantics mean two peers |
| 49 | // sharing a channel name (a cloned Pi env, the agent started twice, a |
| 50 | // promote/adopt/restart overlap window) would otherwise evict each other |
| 51 | // in an unbounded zero-sleep loop. |
| 52 | StableConnected = 5 * time.Second |
| 53 | // HandshakeTimeout bounds dial+TLS (via net.Dialer / tls.Dialer) and, |
| 54 | // separately, the /register write+ReadResponse round trip: an edge that |
| 55 | // accepts the TCP conn and then never answers must not park Loop forever |
| 56 | // with no retry and no visible state change. |
| 57 | HandshakeTimeout = 10 * time.Second |
| 58 | ) |
| 59 | |
| 60 | // Config is one channel's worth of dialing instructions. URL, Name, Token and |
| 61 | // Handler are required — Loop does not police them, because the caller is the |
| 62 | // one holding the config source and owes its user the specific complaint (see |
| 63 | // internal/host's env check). A blank or unparseable URL simply fails every |
| 64 | // dial and backs off to BackoffCap, which is a bad config loudly logged, not |
| 65 | // a storm. |
| 66 | type Config struct { |
| 67 | URL string // the edge's registration endpoint, http:// or https:// |
| 68 | Name string // this channel's name, as the edge has it configured |
| 69 | Token string // the bearer token bound to Name |
| 70 | Handler http.Handler // served over the inverted conn |
| 71 | |
| 72 | // OnState, when non-nil, is called with StateConnected / |
| 73 | // StateReconnecting on every transition — synchronously, from Loop's |
| 74 | // goroutine, so it must not block. |
| 75 | OnState func(string) |
| 76 | // Logf defaults to log.Printf. The agent passes one that prefixes |
| 77 | // "agent: " so the journal lines stay exactly what operators grep for. |
| 78 | Logf func(format string, args ...any) |
| 79 | } |
| 80 | |
| 81 | // Loop dials cfg.URL and serves cfg.Handler over the inverted conn, forever, |
| 82 | // with jittered capped backoff. It returns only when ctx is cancelled; a |
| 83 | // cancelled ctx tears down the live conn rather than waiting for it to die on |
| 84 | // its own. State is left at whatever the last transition reported — this |
| 85 | // package never invents a "stopped" state, because the agent's heartbeat |
| 86 | // probe reads "" as "the loop was never started", which is a different fact. |
| 87 | func Loop(ctx context.Context, cfg Config) { |
| 88 | logf := cfg.Logf |
| 89 | if logf == nil { |
| 90 | logf = log.Printf |
| 91 | } |
| 92 | setState := func(s string) { |
| 93 | if cfg.OnState != nil { |
| 94 | cfg.OnState(s) |
| 95 | } |
| 96 | } |
| 97 | backoff := BackoffFloor |
| 98 | for ctx.Err() == nil { |
| 99 | // Set BEFORE dialing, not only on error: Dial can block for up to |
| 100 | // HandshakeTimeout on a black-holed handshake, and a stale "" or |
| 101 | // "connected" during that window would read to the agent's heartbeat |
| 102 | // probe as either "never started" or "healthy" — both wrong while |
| 103 | // we're actually hung waiting on a dead edge. |
| 104 | setState(StateReconnecting) |
| 105 | conn, err := Dial(ctx, cfg.URL, cfg.Name, cfg.Token) |
| 106 | if err != nil { |
| 107 | sleep := backoff + rand.N(backoff/2) |
| 108 | logf("tunnel dial: %v (retry in %s)", err, sleep) |
| 109 | if !wait(ctx, sleep) { |
| 110 | return |
| 111 | } |
| 112 | if backoff *= 2; backoff > BackoffCap { |
| 113 | backoff = BackoffCap |
| 114 | } |
| 115 | continue |
| 116 | } |
| 117 | setState(StateConnected) |
| 118 | logf("tunnel connected as %q", cfg.Name) |
| 119 | connectedAt := time.Now() |
| 120 | serveUntilDead(ctx, conn, Mark(cfg.Handler)) // blocks until the conn dies |
| 121 | setState(StateReconnecting) |
| 122 | logf("tunnel lost — re-dialing") |
| 123 | // A connection that stayed up a while earns a fast retry (the |
| 124 | // dual-WAN-failover / Wi-Fi-blip case); one that dies instantly does |
| 125 | // not — see StableConnected. Either way we always sleep some amount |
| 126 | // before redialing: a zero-delay retry is exactly the storm this |
| 127 | // guards against. |
| 128 | if time.Since(connectedAt) >= StableConnected { |
| 129 | backoff = BackoffFloor |
| 130 | } else if backoff *= 2; backoff > BackoffCap { |
| 131 | backoff = BackoffCap |
| 132 | } |
| 133 | sleep := backoff + rand.N(backoff/2) |
| 134 | logf("tunnel retry in %s", sleep) |
| 135 | if !wait(ctx, sleep) { |
| 136 | return |
| 137 | } |
| 138 | } |
| 139 | } |
| 140 | |
| 141 | // wait sleeps d, or returns false if ctx is cancelled first. Every sleep in |
| 142 | // Loop goes through it: a loop that ignored cancellation would outlive the |
| 143 | // test that started it and keep dialing a closed listener. |
| 144 | func wait(ctx context.Context, d time.Duration) bool { |
| 145 | t := time.NewTimer(d) |
| 146 | defer t.Stop() |
| 147 | select { |
| 148 | case <-ctx.Done(): |
| 149 | return false |
| 150 | case <-t.C: |
| 151 | return true |
| 152 | } |
| 153 | } |
| 154 | |
| 155 | // serveUntilDead is Serve plus cancellation: http2.Server.ServeConn blocks |
| 156 | // until the conn dies and has no other stop signal, so closing the conn is |
| 157 | // how a cancelled ctx gets Loop back. |
| 158 | func serveUntilDead(ctx context.Context, conn net.Conn, handler http.Handler) { |
| 159 | done := make(chan struct{}) |
| 160 | go func() { |
| 161 | select { |
| 162 | case <-ctx.Done(): |
| 163 | conn.Close() |
| 164 | case <-done: |
| 165 | } |
| 166 | }() |
| 167 | defer close(done) |
| 168 | Serve(conn, handler) |
| 169 | } |
| 170 | |
| 171 | // bufferedConn replays bytes the handshake's bufio.Reader over-read (the h2 |
| 172 | // preface can arrive in the same packet as the 101) before the raw conn. |
| 173 | type bufferedConn struct { |
| 174 | r io.Reader |
| 175 | net.Conn |
| 176 | } |
| 177 | |
| 178 | func (b bufferedConn) Read(p []byte) (int, error) { return b.r.Read(p) } |
| 179 | |
| 180 | // Dial performs the registration handshake and returns the raw conn, ready to |
| 181 | // serve h2 on. https URLs use real TLS (the edge's public cert — server auth |
| 182 | // by PKI, client auth by bearer token); http is for tests. |
| 183 | // |
| 184 | // ctx bounds the TCP connect and the TLS handshake. The /register round trip |
| 185 | // after that is bounded by HandshakeTimeout instead, not by ctx: a cancel |
| 186 | // arriving mid-handshake is therefore observed within HandshakeTimeout rather |
| 187 | // than immediately, which is the one place cancellation is not instant and is |
| 188 | // deliberately left simple — the alternative is a watchdog goroutine per dial |
| 189 | // to buy 10 seconds on a shutdown path. |
| 190 | func Dial(ctx context.Context, rawURL, name, token string) (net.Conn, error) { |
| 191 | u, err := url.Parse(rawURL) |
| 192 | if err != nil { |
| 193 | return nil, fmt.Errorf("tunnel url: %w", err) |
| 194 | } |
| 195 | addr := u.Host |
| 196 | dialer := &net.Dialer{Timeout: HandshakeTimeout} |
| 197 | var conn net.Conn |
| 198 | switch u.Scheme { |
| 199 | case "https": |
| 200 | if u.Port() == "" { |
| 201 | addr = net.JoinHostPort(u.Hostname(), "443") |
| 202 | } |
| 203 | // tls.Dialer applies dialer.Timeout to the TCP connect and the TLS |
| 204 | // handshake as a whole (crypto/tls docs) — a black-holed TLS peer |
| 205 | // fails the dial instead of hanging it — and ctx on top of that. |
| 206 | td := &tls.Dialer{NetDialer: dialer, Config: &tls.Config{ |
| 207 | ServerName: u.Hostname(), |
| 208 | NextProtos: []string{"http/1.1"}, // the handshake is 1.1; h2 starts after the 101 |
| 209 | }} |
| 210 | conn, err = td.DialContext(ctx, "tcp", addr) |
| 211 | case "http": |
| 212 | conn, err = dialer.DialContext(ctx, "tcp", addr) |
| 213 | default: |
| 214 | return nil, fmt.Errorf("tunnel url: unsupported scheme %q", u.Scheme) |
| 215 | } |
| 216 | if err != nil { |
| 217 | return nil, err |
| 218 | } |
| 219 | // The dial above only bounds TCP connect + TLS handshake; the /register |
| 220 | // write+read is a separate hang risk (an edge that accepts the conn and |
| 221 | // then never answers). Bound it too, and clear the deadline before |
| 222 | // returning success — the live h2 session that follows must not inherit a |
| 223 | // stale deadline from the handshake. |
| 224 | if err := conn.SetDeadline(time.Now().Add(HandshakeTimeout)); err != nil { |
| 225 | conn.Close() |
| 226 | return nil, fmt.Errorf("tunnel: set handshake deadline: %w", err) |
| 227 | } |
| 228 | req := &http.Request{ |
| 229 | Method: "GET", |
| 230 | URL: &url.URL{Path: "/register"}, |
| 231 | Host: u.Hostname(), |
| 232 | Header: http.Header{ |
| 233 | "Authorization": {"Bearer " + token}, |
| 234 | "Carlos-Tunnel-Name": {name}, |
| 235 | "Connection": {"Upgrade"}, |
| 236 | "Upgrade": {"carlos-tunnel/1"}, |
| 237 | }, |
| 238 | } |
| 239 | if err := req.Write(conn); err != nil { |
| 240 | conn.Close() |
| 241 | return nil, err |
| 242 | } |
| 243 | br := bufio.NewReader(conn) |
| 244 | resp, err := http.ReadResponse(br, req) |
| 245 | if err != nil { |
| 246 | conn.Close() |
| 247 | return nil, err |
| 248 | } |
| 249 | resp.Body.Close() |
| 250 | if resp.StatusCode != http.StatusSwitchingProtocols { |
| 251 | conn.Close() |
| 252 | return nil, fmt.Errorf("tunnel register: %s", resp.Status) |
| 253 | } |
| 254 | if err := conn.SetDeadline(time.Time{}); err != nil { |
| 255 | conn.Close() |
| 256 | return nil, fmt.Errorf("tunnel: clear handshake deadline: %w", err) |
| 257 | } |
| 258 | if br.Buffered() > 0 { |
| 259 | return bufferedConn{r: io.MultiReader(br, conn), Conn: conn}, nil |
| 260 | } |
| 261 | return conn, nil |
| 262 | } |
| 263 | |
| 264 | type tunnelKey struct{} |
| 265 | |
| 266 | // FromTunnel reports whether a request was served over a tunnel this client |
| 267 | // dialled: the hub computed its GeoIP answer, so the edge adopts the hub's |
| 268 | // headers and does not observe the request for analytics (the hub already |
| 269 | // did). Set on every tunnelled request whatever the URL's scheme (geoip spec |
| 270 | // D2: on a plaintext tunnel an on-path attacker can already inject any |
| 271 | // request, so withholding geo would protect nothing and would make the hub's |
| 272 | // cache disagree with what the app saw). Nothing outside this package can |
| 273 | // set it. |
| 274 | func FromTunnel(ctx context.Context) bool { |
| 275 | v, _ := ctx.Value(tunnelKey{}).(bool) |
| 276 | return v |
| 277 | } |
| 278 | |
| 279 | type markedHandler struct{ h http.Handler } |
| 280 | |
| 281 | func (m markedHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { |
| 282 | m.h.ServeHTTP(w, r.WithContext(context.WithValue(r.Context(), tunnelKey{}, true))) |
| 283 | } |
| 284 | |
| 285 | // Mark wraps h so every request it serves carries the tunnel mark. Loop |
| 286 | // applies it; a caller using Serve directly applies it itself if it wants |
| 287 | // the mark. |
| 288 | func Mark(h http.Handler) http.Handler { return markedHandler{h} } |
| 289 | |
| 290 | // Serve serves handler over the inverted conn until it dies. |
| 291 | func Serve(conn net.Conn, handler http.Handler) { |
| 292 | (&http2.Server{}).ServeConn(conn, &http2.ServeConnOpts{Handler: handler}) |
| 293 | } |
| 294 | |
amadan