| 1 | // Package-internal reverse-channel hub: remote hosts dial OUT to the edge's |
| 2 | // existing 443 listener on a reserved hostname, authenticate with a bearer |
| 3 | // token bound to their channel name, and the connection inverts — the edge |
| 4 | // becomes an HTTP/2 client over the hijacked conn, the host an HTTP/2 server. |
| 5 | // Nothing ever connects TO a remote host (issue #53's locked-down property); |
| 6 | // this file is the edge half of that bargain. Transport judgment call |
| 7 | // (spec 2026-08-04): reverse h2 rather than a websocket port of titogo's |
| 8 | // tunnel — x/net/http2 already supplies mux, flow control, and pings. |
| 9 | package edge |
| 10 | |
| 11 | import ( |
| 12 | "context" |
| 13 | "crypto/sha256" |
| 14 | "crypto/subtle" |
| 15 | "encoding/hex" |
| 16 | "fmt" |
| 17 | "log" |
| 18 | "net" |
| 19 | "net/http" |
| 20 | "slices" |
| 21 | "strings" |
| 22 | "sync" |
| 23 | "time" |
| 24 | |
| 25 | "golang.org/x/net/http/httpguts" |
| 26 | "golang.org/x/net/http2" |
| 27 | ) |
| 28 | |
| 29 | const ( |
| 30 | tunnelRegisterPath = "/register" |
| 31 | // tunnelUpgradeProto is the wire contract the host dialer codes against: |
| 32 | // |
| 33 | // GET /register |
| 34 | // Authorization: Bearer <token> |
| 35 | // Carlos-Tunnel-Name: <name> |
| 36 | // Connection: Upgrade |
| 37 | // Upgrade: carlos-tunnel/1 |
| 38 | // |
| 39 | // The dialer MUST also pin NextProtos to {"http/1.1"} on its TLS config. |
| 40 | // The edge's 443 listener advertises h2, and a dialer that lets ALPN pick |
| 41 | // it authenticates fine and then dies on "hijack unsupported": HTTP/2 |
| 42 | // response writers are not http.Hijacker, and hijacking is how this |
| 43 | // handshake gets the raw conn. The h2 session begins only AFTER the 101, |
| 44 | // on that raw conn, with the roles inverted. |
| 45 | tunnelUpgradeProto = "carlos-tunnel/1" |
| 46 | // tunnelPingEvery / tunnelPingTimeout bound how long a half-dead channel |
| 47 | // can look connected: a failed or timed-out ping tears it down, the host |
| 48 | // notices its conn die and re-dials (seconds — the bound Task 10 asserts). |
| 49 | tunnelPingEvery = 15 * time.Second |
| 50 | tunnelPingTimeout = 10 * time.Second |
| 51 | // Fleet (dynamic) channels ping tighter (fleets-console spec D7): a |
| 52 | // customer fleet's failover is "requests fall back to a surviving box once |
| 53 | // the dead channel is reaped", so the reap bound IS the user-visible |
| 54 | // outage window — ≤10s (one 5s sleep + one 5s deadline) instead of the |
| 55 | // static channels' ~25s. Static platform channels (tito hq) keep the wider |
| 56 | // cadence: their routes have no fallback box, so a tighter ping only adds |
| 57 | // churn risk for no availability gain. |
| 58 | fleetPingEvery = 5 * time.Second |
| 59 | fleetPingTimeout = 5 * time.Second |
| 60 | ) |
| 61 | |
| 62 | // pingCadence picks a channel's liveness cadence from which credential source |
| 63 | // admitted it. Split out from pingLoop so the choice is testable without |
| 64 | // waiting on wall-clock pings. |
| 65 | func pingCadence(dynamic bool) (every, timeout time.Duration) { |
| 66 | if dynamic { |
| 67 | return fleetPingEvery, fleetPingTimeout |
| 68 | } |
| 69 | return tunnelPingEvery, tunnelPingTimeout |
| 70 | } |
| 71 | |
| 72 | type tunnelConn struct { |
| 73 | cc *http2.ClientConn |
| 74 | raw net.Conn |
| 75 | // dynamic records that this conn authenticated against the registry-backed |
| 76 | // (fleet) credential source rather than the static env map — the one fact |
| 77 | // ServeHTTP knows and pingLoop needs for pingCadence. |
| 78 | dynamic bool |
| 79 | } |
| 80 | |
| 81 | type TunnelHub struct { |
| 82 | host string // reserved hostname this hub answers on |
| 83 | clients map[string]string // channel name -> bearer token (static env config, plaintext) |
| 84 | mu sync.Mutex |
| 85 | conns map[string]*tunnelConn |
| 86 | // dynamicAuth is the second, registry-backed credential source: channel |
| 87 | // "<fleet>/<label>" -> sha256 hex of its bearer token. Guarded by mu |
| 88 | // alongside conns since SetDynamicAuth's rotation path (below) touches |
| 89 | // both. Static clients keep authenticating by plaintext compare, |
| 90 | // unchanged; this is consulted only when a name misses h.clients, which |
| 91 | // is always true for a dynamic name — ValidTunnelName-checked static |
| 92 | // names never contain "/". |
| 93 | dynamicAuth map[string]string |
| 94 | } |
| 95 | |
| 96 | func NewTunnelHub(host string, clients map[string]string) *TunnelHub { |
| 97 | return &TunnelHub{host: host, clients: clients, conns: make(map[string]*tunnelConn)} |
| 98 | } |
| 99 | |
| 100 | func (h *TunnelHub) Host() string { return h.host } |
| 101 | |
| 102 | // parseTunnelClients parses CARLOS_TUNNEL_CLIENTS ("name=token[,name=token]"). |
| 103 | func parseTunnelClients(s string) (map[string]string, error) { |
| 104 | out := make(map[string]string) |
| 105 | for _, pair := range strings.Split(s, ",") { |
| 106 | pair = strings.TrimSpace(pair) |
| 107 | if pair == "" { |
| 108 | continue |
| 109 | } |
| 110 | name, token, ok := strings.Cut(pair, "=") |
| 111 | if !ok || name == "" || token == "" { |
| 112 | return nil, fmt.Errorf("tunnel clients: malformed entry %q (want name=token)", pair) |
| 113 | } |
| 114 | out[name] = token |
| 115 | } |
| 116 | return out, nil |
| 117 | } |
| 118 | |
| 119 | // ServeHTTP handles ONLY the registration handshake; the Edge routes requests |
| 120 | // here when r.Host equals the reserved hostname, so ordinary app traffic can |
| 121 | // never reach it and this handler never sees ordinary traffic. |
| 122 | func (h *TunnelHub) ServeHTTP(w http.ResponseWriter, r *http.Request) { |
| 123 | if r.URL.Path != tunnelRegisterPath { |
| 124 | http.NotFound(w, r) |
| 125 | return |
| 126 | } |
| 127 | name := r.Header.Get("Carlos-Tunnel-Name") |
| 128 | // CutPrefix's ok is load-bearing: TrimPrefix would return a scheme-less |
| 129 | // header (e.g. "Authorization: sekrit") unchanged, letting a credential |
| 130 | // that never claimed to be a bearer token compare equal to a configured |
| 131 | // one. No "Bearer " prefix is an outright 401, not a token of "". |
| 132 | token, hasBearer := strings.CutPrefix(r.Header.Get("Authorization"), "Bearer ") |
| 133 | want, ok := h.clients[name] |
| 134 | // Constant-time compare; the name lookup happening first is fine — the |
| 135 | // name is not the secret, the token is. A token is bound to exactly one |
| 136 | // channel name, so a leaked token cannot impersonate a different host. |
| 137 | // An empty configured token never authenticates anything: parseTunnelClients |
| 138 | // rejects one, and a hub built some other way must not turn a blank config |
| 139 | // value into a channel that answers to "Authorization: Bearer ". |
| 140 | authOK := name != "" && ok && want != "" && hasBearer && |
| 141 | subtle.ConstantTimeCompare([]byte(token), []byte(want)) == 1 |
| 142 | // Second credential source: a name that misses the static map (every |
| 143 | // "<fleet>/<label>" dynamic channel does, since ValidTunnelName-checked |
| 144 | // static names never contain "/") falls through to the registry-backed |
| 145 | // set SetDynamicAuth maintains. The presented token is never compared in |
| 146 | // the clear here — only its sha256 hex is, against the stored hash — so |
| 147 | // a leaked auth log or memory dump of this table never yields a usable |
| 148 | // bearer token. Folded into the SAME authOK the static path feeds, so |
| 149 | // there is exactly one 401 below regardless of which source (or neither) |
| 150 | // missed; a caller cannot distinguish "wrong static token" from "unknown |
| 151 | // dynamic channel" from "wrong dynamic token" by response shape. |
| 152 | // viaDynamic is that branch's other output: which source admitted this |
| 153 | // conn decides its ping cadence (pingCadence), and this is the only place |
| 154 | // that fact is known. |
| 155 | viaDynamic := false |
| 156 | if !authOK && name != "" && hasBearer { |
| 157 | _, viaDynamic = h.VerifyDynamic(name, token) |
| 158 | authOK = viaDynamic |
| 159 | } |
| 160 | if !authOK { |
| 161 | http.Error(w, "unauthorized", http.StatusUnauthorized) |
| 162 | return |
| 163 | } |
| 164 | // Wire-contract gate, deliberately AFTER auth: an unauthenticated caller |
| 165 | // gets 401 whatever it sends, so probing cannot discover which protocols |
| 166 | // this endpoint speaks. Checking it at all is what keeps a version skew |
| 167 | // diagnosable — without it we would answer "101 carlos-tunnel/1" to a |
| 168 | // client that never asked, and a future carlos-tunnel/2 host would see a |
| 169 | // mystery h2-preface failure instead of a refusal naming the version. |
| 170 | // Connection is matched token-wise and Upgrade case-insensitively, both |
| 171 | // per RFC 7230; the version suffix is what must match exactly. |
| 172 | if !httpguts.HeaderValuesContainsToken(r.Header["Connection"], "Upgrade") || |
| 173 | !strings.EqualFold(strings.TrimSpace(r.Header.Get("Upgrade")), tunnelUpgradeProto) { |
| 174 | http.Error(w, "expected Connection: Upgrade with Upgrade: "+tunnelUpgradeProto, http.StatusBadRequest) |
| 175 | return |
| 176 | } |
| 177 | hj, ok := w.(http.Hijacker) |
| 178 | if !ok { |
| 179 | http.Error(w, "hijack unsupported", http.StatusInternalServerError) |
| 180 | return |
| 181 | } |
| 182 | conn, bufrw, err := hj.Hijack() |
| 183 | if err != nil { |
| 184 | log.Printf("edge: tunnel %s: hijack: %v", name, err) |
| 185 | return |
| 186 | } |
| 187 | // The h2 session runs on the RAW conn, so anything the host pipelined |
| 188 | // ahead of our 101 is sitting unreadable in bufrw and would corrupt the |
| 189 | // handshake. Fail loudly here rather than hang on a broken preface: the |
| 190 | // host-side dialer's contract is to write nothing until it reads the 101. |
| 191 | if n := bufrw.Reader.Buffered(); n > 0 { |
| 192 | log.Printf("edge: tunnel %s: %d bytes sent before 101 — dropping", name, n) |
| 193 | conn.Close() |
| 194 | return |
| 195 | } |
| 196 | // Manual 101 on the hijacked conn (the ResponseWriter is spent once hijacked). |
| 197 | fmt.Fprintf(bufrw, "HTTP/1.1 101 Switching Protocols\r\nUpgrade: %s\r\nConnection: Upgrade\r\n\r\n", tunnelUpgradeProto) |
| 198 | if err := bufrw.Flush(); err != nil { |
| 199 | conn.Close() |
| 200 | return |
| 201 | } |
| 202 | tr := &http2.Transport{} |
| 203 | cc, err := tr.NewClientConn(conn) |
| 204 | if err != nil { |
| 205 | log.Printf("edge: tunnel %s: h2 client conn: %v", name, err) |
| 206 | conn.Close() |
| 207 | return |
| 208 | } |
| 209 | h.mu.Lock() |
| 210 | if old := h.conns[name]; old != nil { |
| 211 | old.raw.Close() // a re-register replaces the previous channel |
| 212 | } |
| 213 | h.conns[name] = &tunnelConn{cc: cc, raw: conn, dynamic: viaDynamic} |
| 214 | h.mu.Unlock() |
| 215 | log.Printf("edge: tunnel %s: connected from %s", name, conn.RemoteAddr()) |
| 216 | go h.pingLoop(name, cc, conn, viaDynamic) |
| 217 | } |
| 218 | |
| 219 | // pingLoop is the edge-side liveness check: a dead ping closes the raw conn |
| 220 | // and unregisters the channel, so requests fail fast to 502 instead of |
| 221 | // hanging on a black-holed TCP session (the venue-Wi-Fi failure mode). |
| 222 | // dynamic carries the credential source through from ServeHTTP: fleet |
| 223 | // channels reap on a tighter bound because that bound is their failover |
| 224 | // window (see pingCadence). |
| 225 | func (h *TunnelHub) pingLoop(name string, cc *http2.ClientConn, raw net.Conn, dynamic bool) { |
| 226 | every, timeout := pingCadence(dynamic) |
| 227 | for { |
| 228 | time.Sleep(every) |
| 229 | ctx, cancel := context.WithTimeout(context.Background(), timeout) |
| 230 | err := cc.Ping(ctx) |
| 231 | cancel() |
| 232 | if err != nil { |
| 233 | log.Printf("edge: tunnel %s: ping failed (%v) — dropping channel", name, err) |
| 234 | raw.Close() |
| 235 | // Only unregister if WE are still the live channel: a re-register |
| 236 | // may already have replaced this entry (and closed our conn, which |
| 237 | // is what failed the ping), and deleting then would blank out a |
| 238 | // perfectly healthy successor. |
| 239 | h.mu.Lock() |
| 240 | if cur := h.conns[name]; cur != nil && cur.raw == raw { |
| 241 | delete(h.conns, name) |
| 242 | } |
| 243 | h.mu.Unlock() |
| 244 | return |
| 245 | } |
| 246 | } |
| 247 | } |
| 248 | |
| 249 | // VerifyDynamic reports whether (name, token) authenticates as a dynamic |
| 250 | // (registry-backed fleet) channel, and which fleet owns it. It is the single |
| 251 | // answer to that question: ServeHTTP's dynamic branch is one caller, and |
| 252 | // surfaces with no ResponseWriter to 401 with are the others. |
| 253 | // |
| 254 | // The "/" requirement is a gate, not a parse convenience: static channel names |
| 255 | // are ValidTunnelName-checked and never contain "/", so refusing slash-less |
| 256 | // names here means a static client's plaintext token can never be admitted by |
| 257 | // this path, whatever h.clients (or a hand-built dynamicAuth) holds. An empty |
| 258 | // fleet segment ("/box-a") is refused for the same reason: it would otherwise |
| 259 | // authenticate into fleet "", a name no real fleet has, and every ownership |
| 260 | // check downstream compares against the fleet this returns. |
| 261 | // |
| 262 | // The presented token is never compared in the clear — only its sha256 hex is, |
| 263 | // against the stored hash — so a leaked dump of this table yields no usable |
| 264 | // bearer token, and the compare is constant-time so a wrong token leaks no |
| 265 | // prefix information by timing. |
| 266 | func (h *TunnelHub) VerifyDynamic(name, token string) (fleet string, ok bool) { |
| 267 | fleet, _, hasSlash := strings.Cut(name, "/") |
| 268 | if !hasSlash || fleet == "" { |
| 269 | return "", false |
| 270 | } |
| 271 | h.mu.Lock() |
| 272 | wantHash, known := h.dynamicAuth[name] |
| 273 | h.mu.Unlock() |
| 274 | if !known || wantHash == "" { |
| 275 | return "", false |
| 276 | } |
| 277 | sum := sha256.Sum256([]byte(token)) |
| 278 | gotHash := hex.EncodeToString(sum[:]) |
| 279 | if subtle.ConstantTimeCompare([]byte(gotHash), []byte(wantHash)) != 1 { |
| 280 | return "", false |
| 281 | } |
| 282 | return fleet, true |
| 283 | } |
| 284 | |
| 285 | // SetDynamicAuth replaces the entire registry-backed credential set — |
| 286 | // channel "<fleet>/<label>" -> sha256 hex of its bearer token — with auth. |
| 287 | // The whole-map REPLACE (not an incremental merge) is what makes a NEW dial |
| 288 | // with a stale token get refused the instant this returns, for both |
| 289 | // rotation and detach: the next ServeHTTP call either hashes the presented |
| 290 | // token against a DIFFERENT stored hash (rotation: the name is still |
| 291 | // present, just under a new hash) or misses the map entirely (detach: the |
| 292 | // name is gone). Either way, a new dial with the old token fails now. |
| 293 | // |
| 294 | // ROTATE and DETACH are deliberately asymmetric on an ALREADY-LIVE session, |
| 295 | // and that asymmetry is load-bearing, not an oversight: |
| 296 | // |
| 297 | // - Detach: the name vanishes from auth entirely, so it is in the OLD set |
| 298 | // but missing from the new one — the loop below calls CloseChannel on |
| 299 | // it, force-closing any live conn right now. Detach is the hard |
| 300 | // revocation path; a box taken out of the fleet must not keep a |
| 301 | // channel open on its old credential for even one more request. |
| 302 | // - Rotate: the name STAYS present in auth, just under a new hash — the |
| 303 | // loop below sees it as still-authorized (present in both old and new) |
| 304 | // and does NOT close it. An already-established h2 session therefore |
| 305 | // keeps carrying traffic on its OLD token's connection until the |
| 306 | // legitimate box notices (its own ping/reconnect cycle) and re-dials |
| 307 | // with the NEW token — not until this call returns. This is |
| 308 | // deliberate: closing on every rotation would take the box's channel |
| 309 | // down for the gap between the registry write and the box learning its |
| 310 | // new token, and rotation (unlike detach) is not "this box must be cut |
| 311 | // off now" — it is "this box's credential changed." A caller that wants |
| 312 | // rotation to also force an immediate reconnect must close explicitly |
| 313 | // (CloseChannel) rather than relying on SetDynamicAuth to do it. |
| 314 | func (h *TunnelHub) SetDynamicAuth(auth map[string]string) { |
| 315 | h.mu.Lock() |
| 316 | old := h.dynamicAuth |
| 317 | h.dynamicAuth = auth |
| 318 | h.mu.Unlock() |
| 319 | for name := range old { |
| 320 | if _, stillAuthorized := auth[name]; !stillAuthorized { |
| 321 | h.CloseChannel(name) |
| 322 | } |
| 323 | } |
| 324 | } |
| 325 | |
| 326 | // CloseChannel force-closes and unregisters name's live conn, if any — the |
| 327 | // detach/rotate path (SetDynamicAuth above), also usable directly. Mirrors |
| 328 | // pingLoop's delete-if-still-live guard: the raw conn is closed OUTSIDE the |
| 329 | // lock (Close can block briefly on the wire), then a second lock re-checks |
| 330 | // that this exact conn is STILL the one registered before deleting it — a |
| 331 | // re-register that has already replaced the entry in between must not have |
| 332 | // ITS fresh conn deleted out from under it. |
| 333 | func (h *TunnelHub) CloseChannel(name string) { |
| 334 | h.mu.Lock() |
| 335 | cur := h.conns[name] |
| 336 | h.mu.Unlock() |
| 337 | if cur == nil { |
| 338 | return |
| 339 | } |
| 340 | cur.raw.Close() |
| 341 | h.mu.Lock() |
| 342 | if c := h.conns[name]; c != nil && c.raw == cur.raw { |
| 343 | delete(h.conns, name) |
| 344 | } |
| 345 | h.mu.Unlock() |
| 346 | } |
| 347 | |
| 348 | func (h *TunnelHub) conn(name string) *http2.ClientConn { |
| 349 | h.mu.Lock() |
| 350 | defer h.mu.Unlock() |
| 351 | if c := h.conns[name]; c != nil { |
| 352 | return c.cc |
| 353 | } |
| 354 | return nil |
| 355 | } |
| 356 | |
| 357 | // State reports "connected" or "". The heartbeat's tunnel-state field comes |
| 358 | // from the host-side dial loop, not from here; today this is test-consumed |
| 359 | // only — no production caller reads it. |
| 360 | func (h *TunnelHub) State(name string) string { |
| 361 | if h.conn(name) != nil { |
| 362 | return "connected" |
| 363 | } |
| 364 | return "" |
| 365 | } |
| 366 | |
| 367 | // ConnectedOfFleet returns every currently-connected channel name belonging |
| 368 | // to fleet — the "<fleet>/<label>" names whose prefix matches — sorted. This |
| 369 | // is what makes the edge's fleet-route fallback (edge.go's fleetProxy) |
| 370 | // deterministic: "any connected box of this fleet" always picks the same one |
| 371 | // for a given live set, first entry, rather than whatever order Go's map |
| 372 | // iteration happens to produce. |
| 373 | func (h *TunnelHub) ConnectedOfFleet(fleet string) []string { |
| 374 | prefix := fleet + "/" |
| 375 | h.mu.Lock() |
| 376 | defer h.mu.Unlock() |
| 377 | var out []string |
| 378 | for name := range h.conns { |
| 379 | if strings.HasPrefix(name, prefix) { |
| 380 | out = append(out, name) |
| 381 | } |
| 382 | } |
| 383 | slices.Sort(out) |
| 384 | return out |
| 385 | } |
| 386 | |
| 387 | type roundTripperFunc func(*http.Request) (*http.Response, error) |
| 388 | |
| 389 | func (f roundTripperFunc) RoundTrip(r *http.Request) (*http.Response, error) { return f(r) } |
| 390 | |
| 391 | // transportFor resolves the channel PER REQUEST, so a proxy built before a |
| 392 | // reconnect uses the fresh channel afterwards with no reload needed. |
| 393 | func (h *TunnelHub) transportFor(name string) http.RoundTripper { |
| 394 | return roundTripperFunc(func(req *http.Request) (*http.Response, error) { |
| 395 | cc := h.conn(name) |
| 396 | if cc == nil { |
| 397 | return nil, fmt.Errorf("tunnel %q not connected", name) |
| 398 | } |
| 399 | return cc.RoundTrip(req) |
| 400 | }) |
| 401 | } |
| 402 | |
amadan