amadan

carlos / platform Public

edge: fleet tunnel channels authenticate against registry token hashes

TunnelHub gains a second credential source (dynamicAuth: channel
"<fleet>/<label>" -> sha256 token hash) alongside the existing static
env clients. ServeHTTP tries the static (plaintext) path first, falls
through to hashing the presented token and constant-time-comparing it
against the dynamic set, and funnels both into one 401 so a caller
cannot distinguish which source (or neither) missed. SetDynamicAuth
replaces the whole dynamic set and force-closes (CloseChannel, mirrors
pingLoop's delete-if-still-live guard) any live conn whose channel just
vanished, so a rotated/detached box's old token is refused on its very
next dial rather than surviving until its next failed ping.

Edge.Reload wires this from Registry.FleetChannelAuth (guarded on
e.tunnels != nil) — no new polling, the existing 2s Watch->Reload loop
is the whole rotate/detach convergence bound, since Fingerprint already
moves on every fleet mutation. enableTunnelsFromEnv no longer requires
CARLOS_TUNNEL_CLIENTS to be non-empty: a box that only serves fleet
channels legitimately has no static clients configured, but still needs
a hub for the dynamic set to land on.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Paul Campbell 3b4a4a26fa0523e986a2c2c2cf7ef16c72a8aee5 parent 703eb09
3 files changed, +354 −19
  • internal/edge/edge.go +39 −14
  • internal/edge/tunnel.go +83 −2
  • internal/edge/tunnel_test.go +232 −3
diff --git a/internal/edge/edge.go b/internal/edge/edge.go
index 88308d2d9..8f30a9973 100644
--- a/internal/edge/edge.go
+++ b/internal/edge/edge.go
@@ -315,6 +315,23 @@ func (e *Edge) Reload() error {
}
m[t.Host] = upstream{proxy: proxy, route: t, hibernating: hib}
}
+ // Fleet channel auth: registry-backed, so this refreshes on the SAME poll
+ // that rebuilds routes — Task 1's Fingerprint moves on every fleet
+ // mutation (including token rotation), and the edge's existing 2s
+ // Watch->Reload loop (below, Watch) is the whole convergence bound for a
+ // rotated or detached box. No hub configured (e.tunnels == nil, the vast
+ // majority of boxes today) means no fleet channels either — same
+ // nil-means-disabled convention as act/cache/staticStore. Errors here
+ // abort the WHOLE reload (routes included), matching e.reg.List() above:
+ // a transient registry read failure must leave both tables exactly as
+ // they were, not blank out fleet auth while routes move on.
+ if e.tunnels != nil {
+ auth, err := e.reg.FleetChannelAuth()
+ if err != nil {
+ return err
+ }
+ e.tunnels.SetDynamicAuth(auth)
+ }
e.upstreams.Store(&m)
// Reconcile the activator with the new route set: a host un-flagged (or
// removed) since the last reload may still have a live child holding its
@@ -1176,19 +1193,29 @@ func enableStaticFromEnv(e *Edge, dataDir string) error {
}
// enableTunnelsFromEnv arms the reverse-channel hub when the environment
-// names a reserved hostname (CARLOS_TUNNEL_HOST) and its authorized clients
-// (CARLOS_TUNNEL_CLIENTS, "name=token[,name=token]"). No CARLOS_TUNNEL_HOST is
-// not an error: the box simply serves no tunnels, and routes with Tunnel set
-// keep whatever local backend they have (Reload's tunnel case is guarded on
-// the hub). A host WITH no usable clients IS an error — it would open a
-// registration endpoint no one can ever authenticate against, which reads
-// like a working tunnel edge right up until a host tries to dial it.
+// names a reserved hostname (CARLOS_TUNNEL_HOST); CARLOS_TUNNEL_CLIENTS
+// ("name=token[,name=token]") names its STATIC authorized clients, which may
+// legitimately be empty. No CARLOS_TUNNEL_HOST is not an error: the box
+// simply serves no tunnels at all, and routes with Tunnel set keep whatever
+// local backend they have (Reload's tunnel case is guarded on the hub) — no
+// fleet channels either, since those need a hub to land on just as much as a
+// static client does.
+//
+// A host with no static clients configured IS still valid, unlike before
+// fleet channels existed: the hub also serves customer-fleet boxes, whose
+// credentials come from the registry (Reload -> SetDynamicAuth), not this
+// env var — a box that only ever hosts fleet channels legitimately runs with
+// CARLOS_TUNNEL_CLIENTS unset. The registration endpoint this arms is not
+// "no one can ever authenticate against" in that case; it is "no STATIC
+// client can," which the operator chose on purpose.
//
-// Channel names are validated here, at the only place a name enters the
+// Static channel names are validated here, at the only place one enters the
// process from config: a name the CLI would refuse (registry.ValidTunnelName,
// what `carlos route -tunnel` enforces) can never match a route's Tunnel
// column, so such a client would authenticate, connect, and silently carry no
-// traffic at all. Fail the boot instead, naming the entry.
+// traffic at all. Fail the boot instead, naming the entry. Dynamic (fleet)
+// channel names are validated separately, at the registry writer
+// (AddFleetBox) — they are never parsed from this env var.
func enableTunnelsFromEnv(e *Edge) error {
host := os.Getenv("CARLOS_TUNNEL_HOST")
if host == "" {
@@ -1198,9 +1225,6 @@ func enableTunnelsFromEnv(e *Edge) error {
if err != nil {
return err
}
- if len(clients) == 0 {
- return fmt.Errorf("CARLOS_TUNNEL_HOST=%s set but CARLOS_TUNNEL_CLIENTS is empty", host)
- }
for _, name := range slices.Sorted(maps.Keys(clients)) {
if !registry.ValidTunnelName(name) {
return fmt.Errorf("CARLOS_TUNNEL_CLIENTS: %q is not a valid tunnel name — want [a-z0-9-]{1,32}, and it must match a route's -tunnel to carry anything", name)
@@ -1208,11 +1232,12 @@ func enableTunnelsFromEnv(e *Edge) error {
}
e.EnableTunnels(NewTunnelHub(host, clients))
// Rebuild the route table so tunnel-backed routes that were skipped (or
- // left on their old socket/addr) by the startup reload pick up the hub.
+ // left on their old socket/addr) by the startup reload pick up the hub,
+ // and so fleet channel auth (SetDynamicAuth) gets its first load.
if err := e.Reload(); err != nil {
return fmt.Errorf("reload after EnableTunnels: %w", err)
}
- log.Printf("edge: tunnel hub armed on %s (%d client(s))", host, len(clients))
+ log.Printf("edge: tunnel hub armed on %s (%d static client(s))", host, len(clients))
return nil
}
diff --git a/internal/edge/tunnel.go b/internal/edge/tunnel.go
index afd6459d0..8730105fd 100644
--- a/internal/edge/tunnel.go
+++ b/internal/edge/tunnel.go
@@ -10,7 +10,9 @@ package edge
import (
"context"
+ "crypto/sha256"
"crypto/subtle"
+ "encoding/hex"
"fmt"
"log"
"net"
@@ -54,9 +56,17 @@ type tunnelConn struct {
type TunnelHub struct {
host string // reserved hostname this hub answers on
- clients map[string]string // channel name -> bearer token
+ clients map[string]string // channel name -> bearer token (static env config, plaintext)
mu sync.Mutex
conns map[string]*tunnelConn
+ // dynamicAuth is the second, registry-backed credential source: channel
+ // "<fleet>/<label>" -> sha256 hex of its bearer token. Guarded by mu
+ // alongside conns since SetDynamicAuth's rotation path (below) touches
+ // both. Static clients keep authenticating by plaintext compare,
+ // unchanged; this is consulted only when a name misses h.clients, which
+ // is always true for a dynamic name — ValidTunnelName-checked static
+ // names never contain "/".
+ dynamicAuth map[string]string
}
func NewTunnelHub(host string, clients map[string]string) *TunnelHub {
@@ -103,7 +113,29 @@ func (h *TunnelHub) ServeHTTP(w http.ResponseWriter, r *http.Request) {
// An empty configured token never authenticates anything: parseTunnelClients
// rejects one, and a hub built some other way must not turn a blank config
// value into a channel that answers to "Authorization: Bearer ".
- if name == "" || !ok || want == "" || !hasBearer || subtle.ConstantTimeCompare([]byte(token), []byte(want)) != 1 {
+ authOK := name != "" && ok && want != "" && hasBearer &&
+ subtle.ConstantTimeCompare([]byte(token), []byte(want)) == 1
+ // Second credential source: a name that misses the static map (every
+ // "<fleet>/<label>" dynamic channel does, since ValidTunnelName-checked
+ // static names never contain "/") falls through to the registry-backed
+ // set SetDynamicAuth maintains. The presented token is never compared in
+ // the clear here — only its sha256 hex is, against the stored hash — so
+ // a leaked auth log or memory dump of this table never yields a usable
+ // bearer token. Folded into the SAME authOK the static path feeds, so
+ // there is exactly one 401 below regardless of which source (or neither)
+ // missed; a caller cannot distinguish "wrong static token" from "unknown
+ // dynamic channel" from "wrong dynamic token" by response shape.
+ if !authOK && name != "" && hasBearer {
+ h.mu.Lock()
+ wantHash, dynOK := h.dynamicAuth[name]
+ h.mu.Unlock()
+ if dynOK && wantHash != "" {
+ sum := sha256.Sum256([]byte(token))
+ gotHash := hex.EncodeToString(sum[:])
+ authOK = subtle.ConstantTimeCompare([]byte(gotHash), []byte(wantHash)) == 1
+ }
+ }
+ if !authOK {
http.Error(w, "unauthorized", http.StatusUnauthorized)
return
}
@@ -188,6 +220,55 @@ func (h *TunnelHub) pingLoop(name string, cc *http2.ClientConn, raw net.Conn) {
}
}
+// SetDynamicAuth replaces the entire registry-backed credential set —
+// channel "<fleet>/<label>" -> sha256 hex of its bearer token — with auth.
+// The whole-map REPLACE (not an incremental merge) is what makes a rotated
+// or detached box stop authenticating the instant this returns: the next
+// ServeHTTP call for a vanished name simply misses the map. Called by
+// Edge.Reload on every registry poll (edge.go), so this is the whole
+// rotate/detach convergence path — no separate polling of its own.
+//
+// A name present in the OLD set but missing from auth has just been
+// revoked; any conn already established under its (now-invalid) token must
+// be torn down too, or an already-authenticated h2 session would keep
+// carrying traffic past the revocation until its next failed ping
+// (tunnelPingEvery, up to 15s) instead of closing on this call. CloseChannel
+// is called for each such name AFTER the map swap, so the swap (which
+// blocks any NEW dial with the old token) always lands first.
+func (h *TunnelHub) SetDynamicAuth(auth map[string]string) {
+ h.mu.Lock()
+ old := h.dynamicAuth
+ h.dynamicAuth = auth
+ h.mu.Unlock()
+ for name := range old {
+ if _, stillAuthorized := auth[name]; !stillAuthorized {
+ h.CloseChannel(name)
+ }
+ }
+}
+
+// CloseChannel force-closes and unregisters name's live conn, if any — the
+// detach/rotate path (SetDynamicAuth above), also usable directly. Mirrors
+// pingLoop's delete-if-still-live guard: the raw conn is closed OUTSIDE the
+// lock (Close can block briefly on the wire), then a second lock re-checks
+// that this exact conn is STILL the one registered before deleting it — a
+// re-register that has already replaced the entry in between must not have
+// ITS fresh conn deleted out from under it.
+func (h *TunnelHub) CloseChannel(name string) {
+ h.mu.Lock()
+ cur := h.conns[name]
+ h.mu.Unlock()
+ if cur == nil {
+ return
+ }
+ cur.raw.Close()
+ h.mu.Lock()
+ if c := h.conns[name]; c != nil && c.raw == cur.raw {
+ delete(h.conns, name)
+ }
+ h.mu.Unlock()
+}
+
func (h *TunnelHub) conn(name string) *http2.ClientConn {
h.mu.Lock()
defer h.mu.Unlock()
diff --git a/internal/edge/tunnel_test.go b/internal/edge/tunnel_test.go
index cc02d65c0..b2d90a116 100644
--- a/internal/edge/tunnel_test.go
+++ b/internal/edge/tunnel_test.go
@@ -1,6 +1,9 @@
package edge
import (
+ "crypto/sha256"
+ "encoding/hex"
+ "net"
"net/http"
"net/http/httptest"
"strings"
@@ -9,6 +12,14 @@ import (
"github.com/carlosframework/platform/internal/registry"
)
+// sha256Hex is the test-side mirror of what FleetChannelAuth stores and
+// SetDynamicAuth consumes: the sha256 hex of a bearer token, never the
+// plaintext.
+func sha256Hex(s string) string {
+ sum := sha256.Sum256([]byte(s))
+ return hex.EncodeToString(sum[:])
+}
+
// tunnelReq builds a request that is well-formed EXCEPT for the credentials
// under test: valid upgrade headers, so the auth cases below can only fail on
// auth. Pass an empty token to send no Authorization header at all.
@@ -131,6 +142,133 @@ func TestTunnelRegisterRejectsUnknownPath(t *testing.T) {
}
}
+// TestTunnelRegisterAcceptsDynamicToken proves the second credential source:
+// a channel name containing "/" (the "<fleet>/<label>" shape ValidTunnelName
+// would refuse) authenticates against a token hash set via SetDynamicAuth,
+// hashing the presented token and comparing against the stored hash rather
+// than the plaintext.
+func TestTunnelRegisterAcceptsDynamicToken(t *testing.T) {
+ hub := NewTunnelHub("tunnel.test", nil)
+ hub.SetDynamicAuth(map[string]string{"acme/pi-1": sha256Hex("tok")})
+ rec := httptest.NewRecorder()
+ hub.ServeHTTP(rec, tunnelReq("acme/pi-1", "tok"))
+ // Same tell as TestTunnelRegisterAcceptsTokenWiseConnectionHeader: past
+ // auth and the upgrade gate, stopped at the hijack httptest can't do.
+ if rec.Code != http.StatusInternalServerError {
+ t.Fatalf("got %d, want 500 (past auth, stopped at the hijack)", rec.Code)
+ }
+}
+
+func TestTunnelRegisterRejectsWrongDynamicToken(t *testing.T) {
+ hub := NewTunnelHub("tunnel.test", nil)
+ hub.SetDynamicAuth(map[string]string{"acme/pi-1": sha256Hex("tok")})
+ rec := httptest.NewRecorder()
+ hub.ServeHTTP(rec, tunnelReq("acme/pi-1", "wrong"))
+ if rec.Code != http.StatusUnauthorized {
+ t.Fatalf("got %d, want 401", rec.Code)
+ }
+}
+
+func TestTunnelRegisterRejectsUnknownDynamicChannel(t *testing.T) {
+ hub := NewTunnelHub("tunnel.test", nil)
+ hub.SetDynamicAuth(map[string]string{"acme/pi-1": sha256Hex("tok")})
+ rec := httptest.NewRecorder()
+ hub.ServeHTTP(rec, tunnelReq("acme/pi-2", "tok"))
+ if rec.Code != http.StatusUnauthorized {
+ t.Fatalf("got %d, want 401", rec.Code)
+ }
+}
+
+// TestStaticClientStillAuthenticatesAlongsideDynamicSet is the regression: a
+// static env client must keep authenticating unchanged even once a dynamic
+// set is also configured on the same hub (tito hq's plaintext-compare path
+// is untouched by fleet channels landing).
+func TestStaticClientStillAuthenticatesAlongsideDynamicSet(t *testing.T) {
+ hub := NewTunnelHub("tunnel.test", map[string]string{"hq": "sekrit"})
+ hub.SetDynamicAuth(map[string]string{"acme/pi-1": sha256Hex("tok")})
+ rec := httptest.NewRecorder()
+ hub.ServeHTTP(rec, tunnelReq("hq", "sekrit"))
+ if rec.Code != http.StatusInternalServerError {
+ t.Fatalf("static client with a dynamic set also configured: got %d, want 500", rec.Code)
+ }
+}
+
+// TestCloseChannelClosesAndRemovesLiveConn: CloseChannel is the detach/rotate
+// primitive — force-close a live conn and drop it from the map. No real h2
+// handshake needed: CloseChannel only ever touches .raw.
+func TestCloseChannelClosesAndRemovesLiveConn(t *testing.T) {
+ hub := NewTunnelHub("tunnel.test", nil)
+ c1, c2 := net.Pipe()
+ defer c2.Close()
+ hub.mu.Lock()
+ hub.conns["acme/pi-1"] = &tunnelConn{raw: c1}
+ hub.mu.Unlock()
+
+ hub.CloseChannel("acme/pi-1")
+
+ hub.mu.Lock()
+ _, stillThere := hub.conns["acme/pi-1"]
+ hub.mu.Unlock()
+ if stillThere {
+ t.Fatal("CloseChannel left the channel registered")
+ }
+ if _, err := c1.Write([]byte("x")); err == nil {
+ t.Fatal("CloseChannel did not close the raw conn")
+ }
+}
+
+// TestCloseChannelOnUnknownNameIsNoop: closing a name with no live conn must
+// not panic or otherwise misbehave (Reload calls this for every vanished
+// registry entry, whether or not a channel was ever actually connected).
+func TestCloseChannelOnUnknownNameIsNoop(t *testing.T) {
+ hub := NewTunnelHub("tunnel.test", nil)
+ hub.CloseChannel("nosuch/label")
+}
+
+// TestSetDynamicAuthRotationClosesVanishedChannelAndRefusesOldToken pins the
+// brief's exact rotation contract: a token authenticates while its channel is
+// in the dynamic set; once SetDynamicAuth is called WITHOUT that entry (the
+// shape a rotate/detach registry poll produces), any live conn for it is
+// force-closed and the old token is refused on the very next dial — not left
+// to die on its own on the next failed ping.
+func TestSetDynamicAuthRotationClosesVanishedChannelAndRefusesOldToken(t *testing.T) {
+ hub := NewTunnelHub("tunnel.test", nil)
+ hub.SetDynamicAuth(map[string]string{"acme/pi-1": sha256Hex("tok")})
+
+ // Valid while present.
+ rec := httptest.NewRecorder()
+ hub.ServeHTTP(rec, tunnelReq("acme/pi-1", "tok"))
+ if rec.Code != http.StatusInternalServerError {
+ t.Fatalf("initial dynamic dial: got %d, want 500", rec.Code)
+ }
+
+ // A live established channel for that name.
+ c1, c2 := net.Pipe()
+ defer c2.Close()
+ hub.mu.Lock()
+ hub.conns["acme/pi-1"] = &tunnelConn{raw: c1}
+ hub.mu.Unlock()
+
+ // Rotate: the entry vanishes from the dynamic set entirely.
+ hub.SetDynamicAuth(map[string]string{})
+
+ hub.mu.Lock()
+ _, stillThere := hub.conns["acme/pi-1"]
+ hub.mu.Unlock()
+ if stillThere {
+ t.Fatal("SetDynamicAuth left the vanished channel's live conn registered")
+ }
+ if _, err := c1.Write([]byte("x")); err == nil {
+ t.Fatal("SetDynamicAuth did not close the vanished channel's raw conn")
+ }
+
+ rec2 := httptest.NewRecorder()
+ hub.ServeHTTP(rec2, tunnelReq("acme/pi-1", "tok"))
+ if rec2.Code != http.StatusUnauthorized {
+ t.Fatalf("old token after rotation: got %d, want 401", rec2.Code)
+ }
+}
+
func TestParseTunnelClients(t *testing.T) {
got, err := parseTunnelClients("pi-box=sekrit,hq=other")
if err != nil || got["pi-box"] != "sekrit" || got["hq"] != "other" {
@@ -258,11 +396,18 @@ func TestEnableTunnelsFromEnv(t *testing.T) {
t.Fatal("hub armed with no CARLOS_TUNNEL_HOST")
}
})
- t.Run("host with no clients", func(t *testing.T) {
+ // A host with no STATIC clients is valid: fleet channels (SetDynamicAuth,
+ // wired by Reload from the registry) can populate the hub with no static
+ // env config at all — the hub still has to exist for those to land on.
+ t.Run("host with no static clients is valid", func(t *testing.T) {
t.Setenv("CARLOS_TUNNEL_HOST", "tunnel.test")
t.Setenv("CARLOS_TUNNEL_CLIENTS", "")
- if err := enableTunnelsFromEnv(newEdge()); err == nil {
- t.Fatal("want an error: a registration endpoint nobody can authenticate against")
+ e := newEdge()
+ if err := enableTunnelsFromEnv(e); err != nil {
+ t.Fatalf("CARLOS_TUNNEL_HOST with no static clients: %v", err)
+ }
+ if e.tunnels == nil {
+ t.Fatal("hub not armed with no static clients configured")
}
})
t.Run("invalid channel name", func(t *testing.T) {
@@ -286,6 +431,90 @@ func TestEnableTunnelsFromEnv(t *testing.T) {
})
}
+// TestReloadWiresFleetChannelAuthIntoTunnelHub is Task 3's core integration
+// point: Edge.Reload reads Registry.FleetChannelAuth (Task 1) and pushes it
+// into the hub, with no new polling — the existing 2s Watch->Reload loop is
+// the whole convergence path.
+func TestReloadWiresFleetChannelAuthIntoTunnelHub(t *testing.T) {
+ reg := openReg(t)
+ if err := reg.UpsertFleet(registry.Fleet{Name: "acme", Account: "amadan", Bucket: "acme-data"}); err != nil {
+ t.Fatal(err)
+ }
+ if err := reg.AddFleetBox(registry.FleetBox{Fleet: "acme", Label: "pi-1", TokenHash: sha256Hex("tok"), Status: "active"}); err != nil {
+ t.Fatal(err)
+ }
+ e, err := New(reg, nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ e.EnableTunnels(NewTunnelHub("tunnel.test", nil))
+ if err := e.Reload(); err != nil {
+ t.Fatal(err)
+ }
+ rec := httptest.NewRecorder()
+ e.tunnels.ServeHTTP(rec, tunnelReq("acme/pi-1", "tok"))
+ if rec.Code != http.StatusInternalServerError {
+ t.Fatalf("fleet channel auth not wired from the registry: got %d, want 500", rec.Code)
+ }
+}
+
+// TestReloadWithNilTunnelHubIsSafe: most boxes run with no CARLOS_TUNNEL_HOST
+// at all, so e.tunnels is nil on every Reload — the fleet-auth wiring must be
+// guarded, not assume a hub exists.
+func TestReloadWithNilTunnelHubIsSafe(t *testing.T) {
+ e, err := New(openReg(t), nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if err := e.Reload(); err != nil {
+ t.Fatalf("Reload with no tunnel hub configured: %v", err)
+ }
+}
+
+// TestReloadDetachedFleetBoxClosesLiveChannel proves the rotate/detach
+// convergence bound end to end through Reload: a box detached in the
+// registry has its live conn force-closed on the very next reload, the same
+// poll Watch drives every ~2s.
+func TestReloadDetachedFleetBoxClosesLiveChannel(t *testing.T) {
+ reg := openReg(t)
+ if err := reg.UpsertFleet(registry.Fleet{Name: "acme", Account: "amadan", Bucket: "acme-data"}); err != nil {
+ t.Fatal(err)
+ }
+ if err := reg.AddFleetBox(registry.FleetBox{Fleet: "acme", Label: "pi-1", TokenHash: sha256Hex("tok"), Status: "active"}); err != nil {
+ t.Fatal(err)
+ }
+ e, err := New(reg, nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ hub := NewTunnelHub("tunnel.test", nil)
+ e.EnableTunnels(hub)
+ if err := e.Reload(); err != nil {
+ t.Fatal(err)
+ }
+
+ // A live channel, as if the box had already dialed in successfully.
+ c1, c2 := net.Pipe()
+ defer c2.Close()
+ hub.mu.Lock()
+ hub.conns["acme/pi-1"] = &tunnelConn{raw: c1}
+ hub.mu.Unlock()
+
+ if err := reg.SetFleetBoxStatus("acme", "pi-1", "detached"); err != nil {
+ t.Fatal(err)
+ }
+ if err := e.Reload(); err != nil {
+ t.Fatal(err)
+ }
+
+ hub.mu.Lock()
+ _, stillThere := hub.conns["acme/pi-1"]
+ hub.mu.Unlock()
+ if stillThere {
+ t.Fatal("Reload did not close the detached fleet box's live channel")
+ }
+}
+
// A tunnel-served host is never activator-managed: the process lives on the
// far side of the channel, so a wake here would start a second, local copy.
// The retire pass must also drop it, killing the child a repointed route left