amadan
Keep fleet credentials and account authority in one runtime snapshot
7 files changed,
+592
−110
internal/edge/edge.go+19 −40internal/edge/storedoor.go+5 −18internal/edge/storedoor_test.go+18 −16internal/edge/tunnel.go+102 −32internal/edge/tunnel_test.go+273 −4internal/registry/fleets.go+62 −0internal/registry/fleets_test.go+113 −0
CI passed — run details
CI log
=== gate ===
go: downloading golang.org/x/net v0.55.0
go: downloading golang.org/x/crypto v0.52.0
go: downloading golang.org/x/text v0.37.0
cd launchctx && go vet ./... && go test ./... -count=1
ok amadan.net/carlos/platform.git/launchctx 0.011s
? amadan.net/carlos/platform.git/launchctx/controlname [no test files]
? amadan.net/carlos/platform.git/launchctx/wireid [no test files]
go vet ./...
gofmt -l .: clean
go test ./... github.com/benbjohnson/litestream/s3 -count=1 -timeout 30m
ok github.com/carlosframework/platform 155.405s
ok github.com/carlosframework/platform/cmd/console 2.817s
? github.com/carlosframework/platform/cmd/pricebook-export [no test files]
ok github.com/carlosframework/platform/internal/activator 56.024s
ok github.com/carlosframework/platform/internal/agentsock 0.014s
ok github.com/carlosframework/platform/internal/analytics 0.017s
ok github.com/carlosframework/platform/internal/appconfig 0.723s
ok github.com/carlosframework/platform/internal/apperr 0.072s
ok github.com/carlosframework/platform/internal/applog 0.008s
ok github.com/carlosframework/platform/internal/authority 0.004s
ok github.com/carlosframework/platform/internal/awsbilling 0.006s
ok github.com/carlosframework/platform/internal/awsinv 0.003s
ok github.com/carlosframework/platform/internal/billingstripe 0.193s
ok github.com/carlosframework/platform/internal/caps 0.011s
ok github.com/carlosframework/platform/internal/certwarm 0.002s
ok github.com/carlosframework/platform/internal/cgroup 0.039s
ok github.com/carlosframework/platform/internal/clamd 0.054s
? github.com/carlosframework/platform/internal/cliflag [no test files]
ok github.com/carlosframework/platform/internal/console 221.536s
? github.com/carlosframework/platform/internal/controlname [no test files]
? github.com/carlosframework/platform/internal/controlsock [no test files]
ok github.com/carlosframework/platform/internal/dbverify 0.003s
ok github.com/carlosframework/platform/internal/dnsrec 0.003s
ok github.com/carlosframework/platform/internal/docsite 0.004s
ok github.com/carlosframework/platform/internal/domains 1.074s
ok github.com/carlosframework/platform/internal/economics 0.004s
ok github.com/carlosframework/platform/internal/edge 20.422s
ok github.com/carlosframework/platform/internal/edgesfile 0.005s
ok github.com/carlosframework/platform/internal/entitlement 0.011s
ok github.com/carlosframework/platform/internal/fleetapi 0.093s
ok github.com/carlosframework/platform/internal/gate 1.655s
ok github.com/carlosframework/platform/internal/host 7.072s
ok github.com/carlosframework/platform/internal/infraview 0.002s
ok github.com/carlosframework/platform/internal/instancectl 0.415s
ok github.com/carlosframework/platform/internal/instanceexec 0.007s
ok github.com/carlosframework/platform/internal/instancefs 0.003s
ok github.com/carlosframework/platform/internal/lease 5.537s
ok github.com/carlosframework/platform/internal/ledger 0.019s
ok github.com/carlosframework/platform/internal/mail 0.040s
ok github.com/carlosframework/platform/internal/maintenance 0.078s
ok github.com/carlosframework/platform/internal/maintenanceexec 5.493s
ok github.com/carlosframework/platform/internal/maintenanceguard 0.058s
ok github.com/carlosframework/platform/internal/metrics 0.018s
ok github.com/carlosframework/platform/internal/objstore 7.481s
ok github.com/carlosframework/platform/internal/ops 15.383s
ok github.com/carlosframework/platform/internal/ops/admissionmodel 0.093s
ok github.com/carlosframework/platform/internal/plans 0.069s
ok github.com/carlosframework/platform/internal/privatefile 0.014s
ok github.com/carlosframework/platform/internal/provision 0.814s
ok github.com/carlosframework/platform/internal/publicrelease 0.007s
ok github.com/carlosframework/platform/internal/registry 1.427s
ok github.com/carlosframework/platform/internal/relay 0.177s
ok github.com/carlosframework/platform/internal/release 11.733s
ok github.com/carlosframework/platform/internal/resourcebudget 0.295s
ok github.com/carlosframework/platform/internal/s3compat 0.005s
ok github.com/carlosframework/platform/internal/schedule 1.023s
ok github.com/carlosframework/platform/internal/sigfence 0.272s
ok github.com/carlosframework/platform/internal/status 0.025s
ok github.com/carlosframework/platform/internal/steering 0.007s
ok github.com/carlosframework/platform/internal/svcrole 0.002s
ok github.com/carlosframework/platform/internal/telemetry 0.004s
ok github.com/carlosframework/platform/internal/tunnelclient 0.004s
ok github.com/carlosframework/platform/internal/unixdial 0.005s
ok github.com/carlosframework/platform/internal/usage 3.362s
ok github.com/carlosframework/platform/internal/version 0.005s
ok github.com/carlosframework/platform/internal/vet 1.544s
ok github.com/benbjohnson/litestream/s3 5.406s
=== cross ===
--- shard 0 of 5 ---
cross-compiling darwin/arm64
--- shard 1 of 5 ---
cross-compiling darwin/amd64
--- shard 2 of 5 ---
cross-compiling linux/arm64
--- shard 3 of 5 ---
cross-compiling windows/amd64
--- shard 4 of 5 ---
cross-compiling windows/arm64
=== minio-contract ===
minio contract: arm64, minio RELEASE.2025-09-07T16-13-09Z, mc RELEASE.2025-08-13T08-35-41Z
minio: downloading https://github.com/minio/minio/releases/download/RELEASE.2025-09-07T16-13-09Z/minio.linux-arm64.RELEASE.2025-09-07T16-13-09Z
minio: downloaded, digest verified
mc: downloading https://github.com/minio/mc/releases/download/RELEASE.2025-08-13T08-35-41Z/mc.linux-arm64.RELEASE.2025-08-13T08-35-41Z
mc: downloaded, digest verified
minio healthy on 127.0.0.1:20999
Added `local` successfully.
Bucket created successfully `local/contract-bucket`.
Bucket created successfully `local/denied-bucket`.
Created policy `contract-only` successfully.
Added user `scopeduser` successfully.
Attached Policies: [contract-only]
To User: scopeduser
buckets and scoped user ready
=== RUN TestDeliveredContractMinio
=== RUN TestDeliveredContractMinio/put
=== RUN TestDeliveredContractMinio/get_round-trips_the_body
=== RUN TestDeliveredContractMinio/list_shows_the_key
=== RUN TestDeliveredContractMinio/head_and_list_agree_on_the_version
=== RUN TestDeliveredContractMinio/the_app_cannot_tag_its_own_object
=== RUN TestDeliveredContractMinio/the_app_cannot_tag_its_own_object/not_by_PutObjectTagging
=== RUN TestDeliveredContractMinio/the_app_cannot_tag_its_own_object/and_not_by_x-amz-tagging_on_upload
=== RUN TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route
=== RUN TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/not_by_CreateMultipartUpload
=== RUN TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/not_by_CopyObject_with_a_REPLACE_tagging_directive
=== RUN TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/but_multipart_itself_is_allowed
=== RUN TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/and_copy_itself_is_allowed
=== RUN TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/a_POST-policy_forged_tag_with_a_matching_version_is_still_not_a_verdict
=== RUN TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/and_an_ordinary_tagged-with-something-else_upload_is_allowed
=== RUN TestDeliveredContractMinio/delete_removes_it
=== RUN TestDeliveredContractMinio/cross-bucket_denial
--- PASS: TestDeliveredContractMinio (0.02s)
--- PASS: TestDeliveredContractMinio/put (0.00s)
--- PASS: TestDeliveredContractMinio/get_round-trips_the_body (0.00s)
--- PASS: TestDeliveredContractMinio/list_shows_the_key (0.00s)
--- PASS: TestDeliveredContractMinio/head_and_list_agree_on_the_version (0.00s)
--- PASS: TestDeliveredContractMinio/the_app_cannot_tag_its_own_object (0.00s)
--- PASS: TestDeliveredContractMinio/the_app_cannot_tag_its_own_object/not_by_PutObjectTagging (0.00s)
--- PASS: TestDeliveredContractMinio/the_app_cannot_tag_its_own_object/and_not_by_x-amz-tagging_on_upload (0.00s)
--- PASS: TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route (0.01s)
--- PASS: TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/not_by_CreateMultipartUpload (0.00s)
--- PASS: TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/not_by_CopyObject_with_a_REPLACE_tagging_directive (0.00s)
--- PASS: TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/but_multipart_itself_is_allowed (0.00s)
--- PASS: TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/and_copy_itself_is_allowed (0.00s)
--- PASS: TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/a_POST-policy_forged_tag_with_a_matching_version_is_still_not_a_verdict (0.00s)
--- PASS: TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/and_an_ordinary_tagged-with-something-else_upload_is_allowed (0.00s)
--- PASS: TestDeliveredContractMinio/delete_removes_it (0.00s)
--- PASS: TestDeliveredContractMinio/cross-bucket_denial (0.00s)
PASS
ok github.com/carlosframework/platform/internal/objstore 0.022s
minio contract: PASS
diff --git a/internal/edge/edge.go b/internal/edge/edge.go| index 2045b8eeb..5df6e6f4d 100644 |
| --- a/internal/edge/edge.go |
| +++ b/internal/edge/edge.go |
| @@ -160,12 +160,10 @@ type Edge struct { |
| // lands second. nil = /store/ falls through to the hub's own 404, the |
| // same nil-means-disabled convention as act/cache/staticStore. |
| door *storeDoor |
| - // fleetAccounts is the reload-time fleet→account-sqid snapshot the door's |
| - // ACL is derived from, refreshed on the SAME pass that feeds the hub's |
| - // dynamic auth table (Reload, below) so a box's credentials and its |
| - // tenancy can never come from two different reads of the registry. |
| - // Atomic: written by Reload while requests are being served. |
| - fleetAccounts atomic.Pointer[map[string]string] |
| + // fleetRuntimeMu serializes the registry read through the hub install. |
| + // Reload hooks remain outside it: this lock protects only the order in |
| + // which complete credential/account snapshots become visible. |
| + fleetRuntimeMu sync.Mutex |
| // cache, when non-nil, answers cacheable GET/HEAD requests without ever |
| // invoking the hibernation activator or the backend — the mitigation |
| // this whole feature exists for. nil disables caching entirely. |
| @@ -393,24 +391,23 @@ func (e *Edge) armStoreDoor() { |
| e.door = nil |
| return |
| } |
| - e.door = &storeDoor{hub: e.tunnels, store: e.staticStore, fleetAccount: e.fleetAccount} |
| + e.door = &storeDoor{hub: e.tunnels, store: e.staticStore} |
| } |
| -// fleetAccount resolves an authenticated fleet name to its owning account's |
| -// sqid from the last reload's snapshot. A fleet the snapshot does not carry |
| -// (created since the last reload, or deleted) has no tenancy, so the door |
| -// answers it nothing — the same convergence bound as its channel credentials, |
| -// which come from the same pass. |
| -func (e *Edge) fleetAccount(fleet string) (string, bool) { |
| - m := e.fleetAccounts.Load() |
| - if m == nil { |
| - return "", false |
| - } |
| - sqid, ok := (*m)[fleet] |
| - if !ok || sqid == "" { |
| - return "", false |
| +type fleetRuntimeReader func() ([]registry.FleetRuntimeEntry, error) |
| + |
| +// reloadFleetRuntime keeps concurrent reloads in read/install order. The |
| +// reader seam is intentionally private and narrow so tests can stop one read |
| +// without adding hooks to Registry or the production reload path. |
| +func (e *Edge) reloadFleetRuntime(read fleetRuntimeReader) error { |
| + e.fleetRuntimeMu.Lock() |
| + defer e.fleetRuntimeMu.Unlock() |
| + entries, err := read() |
| + if err != nil { |
| + return err |
| } |
| - return sqid, true |
| + e.tunnels.SetDynamicRuntime(entries) |
| + return nil |
| } |
| // SetStaticRecheck tunes how long a static route trusts a resolved channel |
| @@ -766,27 +763,9 @@ func (e *Edge) Reload() error { |
| // 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 { |
| + if err := e.reloadFleetRuntime(e.reg.FleetRuntimeSnapshot); err != nil { |
| return err |
| } |
| - e.tunnels.SetDynamicAuth(auth) |
| - // The store door's tenancy map, from the SAME pass: a box's |
| - // credentials and the account its reads are scoped to must never come |
| - // from two different reads of the registry. ListFleets goes over a |
| - // fresh connection (like FleetChannelAuth above), so a fleet the |
| - // console created moments ago is visible here rather than stuck |
| - // behind a stale WAL snapshot on the long-lived handle. Same |
| - // all-or-nothing error contract as everything else in Reload. |
| - fleets, err := e.reg.ListFleets() |
| - if err != nil { |
| - return err |
| - } |
| - accounts := make(map[string]string, len(fleets)) |
| - for _, f := range fleets { |
| - accounts[f.Name] = f.Account |
| - } |
| - e.fleetAccounts.Store(&accounts) |
| } |
| // Probers BEFORE the map swap — the opposite order to RetireExcept below, |
| // and deliberately so. multiProxy answers 503 + Retry-After when the |
diff --git a/internal/edge/storedoor.go b/internal/edge/storedoor.go| index 115d39ed8..332f1685c 100644 |
| --- a/internal/edge/storedoor.go |
| +++ b/internal/edge/storedoor.go |
| @@ -58,15 +58,8 @@ const ( |
| // bytes stream through the edge rather than being redirected to a presigned |
| // URL). |
| type storeDoor struct { |
| - hub *TunnelHub // VerifyDynamic: the one auth path, called once per request |
| + hub *TunnelHub // AuthorizeDynamic: the one auth path, called once per request |
| store release.Store // the deployment bucket, edge's own handle |
| - // fleetAccount maps an authenticated fleet name to its owning account's |
| - // sqid — the whole basis of the read ACL. Backed by the edge's |
| - // reload-time snapshot (Edge.fleetAccount), refreshed on the same pass |
| - // that feeds the hub's FleetChannelAuth table, so a fleet's identity and |
| - // its credentials can never come from two different reads of the |
| - // registry. |
| - fleetAccount func(fleet string) (sqid string, ok bool) |
| } |
| // ServeHTTP is the whole door. The order is load-bearing: |
| @@ -98,7 +91,7 @@ type storeDoor struct { |
| // If-Match always carries the same bytes its Get was handed. |
| func (d *storeDoor) ServeHTTP(w http.ResponseWriter, r *http.Request) { |
| // Auth, once. Same two headers as the tunnel dial (F1), same table (F2): |
| - // only fleet channels ("<fleet>/<label>") can pass — VerifyDynamic |
| + // only fleet channels ("<fleet>/<label>") can pass — AuthorizeDynamic |
| // refuses slash-less static names outright, so a CARLOS_TUNNEL_CLIENTS |
| // credential can never open this door. The token is never logged, never |
| // echoed, and never compared in the clear. |
| @@ -108,16 +101,10 @@ func (d *storeDoor) ServeHTTP(w http.ResponseWriter, r *http.Request) { |
| http.NotFound(w, r) |
| return |
| } |
| - fleet, ok := d.hub.VerifyDynamic(name, token) |
| + fleet, sqid, ok := d.hub.AuthorizeDynamic(name, token) |
| if !ok { |
| - http.NotFound(w, r) |
| - return |
| - } |
| - sqid, ok := d.fleetAccount(fleet) |
| - if !ok || sqid == "" { |
| - // Authenticated, but the fleet's registry row is not in the edge's |
| - // current snapshot (a fleet deleted between reloads, or a hub whose |
| - // auth table was set by hand). No identity, no ACL, no answer. |
| + // No current credential/account tuple means no identity, no ACL and |
| + // no answer. |
| http.NotFound(w, r) |
| return |
| } |
diff --git a/internal/edge/storedoor_test.go b/internal/edge/storedoor_test.go| index 8210ea4ef..e79ba189c 100644 |
| --- a/internal/edge/storedoor_test.go |
| +++ b/internal/edge/storedoor_test.go |
| @@ -375,14 +375,11 @@ const ( |
| func newTestDoor(t *testing.T) (*storeDoor, *release.DirStore) { |
| t.Helper() |
| hub := NewTunnelHub("tunnel.test", map[string]string{staticName: staticToken}) |
| - hub.SetDynamicAuth(map[string]string{testChannel: sha256Hex(testToken)}) |
| + hub.SetDynamicRuntime([]registry.FleetRuntimeEntry{{ |
| + Channel: testChannel, Fleet: testFleet, TokenHash: sha256Hex(testToken), Account: testSqid, |
| + }}) |
| store := release.NewDirStore(t.TempDir()) |
| - return &storeDoor{hub: hub, store: store, fleetAccount: func(fleet string) (string, bool) { |
| - if fleet == testFleet { |
| - return testSqid, true |
| - } |
| - return "", false |
| - }}, store |
| + return &storeDoor{hub: hub, store: store}, store |
| } |
| // doorReq builds a request with the given credentials; an empty token sends |
| @@ -774,29 +771,34 @@ func (c *doorOpCounter) Delete(ctx context.Context, key string) error { |
| } |
| // TestStoreDoorAuthRunsExactlyOncePerRequest pins the invariant two ways: the |
| -// handler contains exactly ONE call to VerifyDynamic (no path can skip it, |
| -// none can pay for it twice — and the auth logic itself is never |
| -// re-implemented here), and an unauthenticated request of any shape reaches |
| -// the store zero times. |
| +// handler contains exactly ONE call to AuthorizeDynamic and NO call to |
| +// VerifyDynamic (no split auth/account lookup can return), and an |
| +// unauthenticated request of any shape reaches the store zero times. |
| func TestStoreDoorAuthRunsExactlyOncePerRequest(t *testing.T) { |
| fset := token.NewFileSet() |
| f, err := parser.ParseFile(fset, "storedoor.go", nil, 0) |
| if err != nil { |
| t.Fatalf("parse storedoor.go: %v", err) |
| } |
| - calls := 0 |
| + authorizeCalls := 0 |
| + verifyCalls := 0 |
| ast.Inspect(f, func(n ast.Node) bool { |
| call, ok := n.(*ast.CallExpr) |
| if !ok { |
| return true |
| } |
| - if sel, ok := call.Fun.(*ast.SelectorExpr); ok && sel.Sel.Name == "VerifyDynamic" { |
| - calls++ |
| + if sel, ok := call.Fun.(*ast.SelectorExpr); ok { |
| + switch sel.Sel.Name { |
| + case "AuthorizeDynamic": |
| + authorizeCalls++ |
| + case "VerifyDynamic": |
| + verifyCalls++ |
| + } |
| } |
| return true |
| }) |
| - if calls != 1 { |
| - t.Errorf("storedoor.go calls VerifyDynamic %d times, want exactly 1 (one auth, one place)", calls) |
| + if authorizeCalls != 1 || verifyCalls != 0 { |
| + t.Errorf("storedoor.go calls AuthorizeDynamic %d times and VerifyDynamic %d times, want 1 and 0", authorizeCalls, verifyCalls) |
| } |
| d, _ := newTestDoor(t) |
diff --git a/internal/edge/tunnel.go b/internal/edge/tunnel.go| index 7f619d995..5043326d8 100644 |
| --- a/internal/edge/tunnel.go |
| +++ b/internal/edge/tunnel.go |
| @@ -24,6 +24,8 @@ import ( |
| "golang.org/x/net/http/httpguts" |
| "golang.org/x/net/http2" |
| + |
| + "github.com/carlosframework/platform/internal/registry" |
| ) |
| const ( |
| @@ -83,14 +85,12 @@ type TunnelHub struct { |
| 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 |
| + // dynamicRuntime is the second, registry-backed credential source: |
| + // channel "<fleet>/<label>" -> its token hash AND owning account. The two |
| + // facts are one immutable replacement under mu, alongside conns, so a |
| + // request can never authenticate against one reload and authorize against |
| + // another. Static clients keep their separate plaintext map unchanged. |
| + dynamicRuntime map[string]registry.FleetRuntimeEntry |
| } |
| func NewTunnelHub(host string, clients map[string]string) *TunnelHub { |
| @@ -153,8 +153,9 @@ func (h *TunnelHub) ServeHTTP(w http.ResponseWriter, r *http.Request) { |
| // conn decides its ping cadence (pingCadence), and this is the only place |
| // that fact is known. |
| viaDynamic := false |
| + var dynamicHash string |
| if !authOK && name != "" && hasBearer { |
| - _, viaDynamic = h.VerifyDynamic(name, token) |
| + _, _, dynamicHash, viaDynamic = h.verifyDynamic(name, token) |
| authOK = viaDynamic |
| } |
| if !authOK { |
| @@ -206,12 +207,20 @@ func (h *TunnelHub) ServeHTTP(w http.ResponseWriter, r *http.Request) { |
| conn.Close() |
| return |
| } |
| - h.mu.Lock() |
| - if old := h.conns[name]; old != nil { |
| - old.raw.Close() // a re-register replaces the previous channel |
| + connected := &tunnelConn{cc: cc, raw: conn, dynamic: viaDynamic} |
| + if viaDynamic { |
| + if !h.commitDynamic(name, dynamicHash, connected) { |
| + conn.Close() |
| + return |
| + } |
| + } else { |
| + h.mu.Lock() |
| + if old := h.conns[name]; old != nil { |
| + old.raw.Close() // a re-register replaces the previous channel |
| + } |
| + h.conns[name] = connected |
| + h.mu.Unlock() |
| } |
| - h.conns[name] = &tunnelConn{cc: cc, raw: conn, dynamic: viaDynamic} |
| - h.mu.Unlock() |
| log.Printf("edge: tunnel %s: connected from %s", name, conn.RemoteAddr()) |
| go h.pingLoop(name, cc, conn, viaDynamic) |
| } |
| @@ -263,27 +272,55 @@ func (h *TunnelHub) pingLoop(name string, cc *http2.ClientConn, raw net.Conn, dy |
| // against the stored hash — so a leaked dump of this table yields no usable |
| // bearer token, and the compare is constant-time so a wrong token leaks no |
| // prefix information by timing. |
| -func (h *TunnelHub) VerifyDynamic(name, token string) (fleet string, ok bool) { |
| +func (h *TunnelHub) verifyDynamic(name, token string) (fleet string, entry registry.FleetRuntimeEntry, presentedHash string, ok bool) { |
| fleet, _, hasSlash := strings.Cut(name, "/") |
| if !hasSlash || fleet == "" { |
| - return "", false |
| + return "", registry.FleetRuntimeEntry{}, "", false |
| } |
| + sum := sha256.Sum256([]byte(token)) |
| + presentedHash = hex.EncodeToString(sum[:]) |
| h.mu.Lock() |
| - wantHash, known := h.dynamicAuth[name] |
| + entry, known := h.dynamicRuntime[name] |
| h.mu.Unlock() |
| - if !known || wantHash == "" { |
| - return "", false |
| + if !known || entry.TokenHash == "" { |
| + return "", registry.FleetRuntimeEntry{}, "", false |
| } |
| - sum := sha256.Sum256([]byte(token)) |
| - gotHash := hex.EncodeToString(sum[:]) |
| - if subtle.ConstantTimeCompare([]byte(gotHash), []byte(wantHash)) != 1 { |
| - return "", false |
| + if subtle.ConstantTimeCompare([]byte(presentedHash), []byte(entry.TokenHash)) != 1 { |
| + return "", registry.FleetRuntimeEntry{}, "", false |
| + } |
| + return entry.Fleet, entry, presentedHash, true |
| +} |
| + |
| +func (h *TunnelHub) VerifyDynamic(name, token string) (fleet string, ok bool) { |
| + fleet, _, _, ok = h.verifyDynamic(name, token) |
| + return fleet, ok |
| +} |
| + |
| +// AuthorizeDynamic verifies one fleet credential and returns the account from |
| +// that exact runtime entry. A missing account keeps authentic orphan/hand-built |
| +// entries out of the store door without a second lookup. |
| +func (h *TunnelHub) AuthorizeDynamic(name, token string) (fleet, account string, ok bool) { |
| + fleet, entry, _, ok := h.verifyDynamic(name, token) |
| + if !ok || entry.Account == "" { |
| + return "", "", false |
| + } |
| + return fleet, entry.Account, true |
| +} |
| + |
| +// SetDynamicAuth preserves the hand-built hub contract for callers that have |
| +// credentials but no account snapshot. All publication still passes through |
| +// SetDynamicRuntime; these entries authenticate but cannot open the store door. |
| +func (h *TunnelHub) SetDynamicAuth(auth map[string]string) { |
| + entries := make([]registry.FleetRuntimeEntry, 0, len(auth)) |
| + for channel, tokenHash := range auth { |
| + fleet, _, _ := strings.Cut(channel, "/") |
| + entries = append(entries, registry.FleetRuntimeEntry{Channel: channel, Fleet: fleet, TokenHash: tokenHash}) |
| } |
| - return fleet, true |
| + h.SetDynamicRuntime(entries) |
| } |
| -// SetDynamicAuth replaces the entire registry-backed credential set — |
| -// channel "<fleet>/<label>" -> sha256 hex of its bearer token — with auth. |
| +// SetDynamicRuntime replaces the entire registry-backed credential/account |
| +// set. The whole-map replace makes every later dial use one current entry. |
| // The whole-map REPLACE (not an incremental merge) is what makes a NEW dial |
| // with a stale token get refused the instant this returns, for both |
| // rotation and detach: the next ServeHTTP call either hashes the presented |
| @@ -311,16 +348,49 @@ func (h *TunnelHub) VerifyDynamic(name, token string) (fleet string, ok bool) { |
| // off now" — it is "this box's credential changed." A caller that wants |
| // rotation to also force an immediate reconnect must close explicitly |
| // (CloseChannel) rather than relying on SetDynamicAuth to do it. |
| -func (h *TunnelHub) SetDynamicAuth(auth map[string]string) { |
| +func (h *TunnelHub) SetDynamicRuntime(entries []registry.FleetRuntimeEntry) { |
| + next := make(map[string]registry.FleetRuntimeEntry, len(entries)) |
| + for _, entry := range entries { |
| + next[entry.Channel] = entry |
| + } |
| + var vanished []net.Conn |
| h.mu.Lock() |
| - old := h.dynamicAuth |
| - h.dynamicAuth = auth |
| - h.mu.Unlock() |
| + old := h.dynamicRuntime |
| + h.dynamicRuntime = next |
| for name := range old { |
| - if _, stillAuthorized := auth[name]; !stillAuthorized { |
| - h.CloseChannel(name) |
| + if _, stillAuthorized := next[name]; stillAuthorized { |
| + continue |
| + } |
| + if current := h.conns[name]; current != nil && current.dynamic { |
| + delete(h.conns, name) |
| + if current.raw != nil { |
| + vanished = append(vanished, current.raw) |
| + } |
| } |
| } |
| + h.mu.Unlock() |
| + for _, conn := range vanished { |
| + conn.Close() |
| + } |
| +} |
| + |
| +// commitDynamic is the dynamic registration's linearization point. The |
| +// initial HTTP auth may have happened before a detach or rotation; checking |
| +// the exact current (name, hash) while holding the same lock as runtime swap |
| +// means the connection is either inserted first and then closed by detach, or |
| +// the new runtime wins and this insertion is refused. |
| +func (h *TunnelHub) commitDynamic(name, presentedHash string, conn *tunnelConn) bool { |
| + h.mu.Lock() |
| + defer h.mu.Unlock() |
| + entry, ok := h.dynamicRuntime[name] |
| + if !ok || entry.TokenHash == "" || subtle.ConstantTimeCompare([]byte(presentedHash), []byte(entry.TokenHash)) != 1 { |
| + return false |
| + } |
| + if old := h.conns[name]; old != nil { |
| + old.raw.Close() |
| + } |
| + h.conns[name] = conn |
| + return true |
| } |
| // CloseChannel force-closes and unregisters name's live conn, if any — the |
diff --git a/internal/edge/tunnel_test.go b/internal/edge/tunnel_test.go| index 4ba3c095a..923e6f435 100644 |
| --- a/internal/edge/tunnel_test.go |
| +++ b/internal/edge/tunnel_test.go |
| @@ -1,19 +1,28 @@ |
| package edge |
| import ( |
| + "bufio" |
| "bytes" |
| "crypto/sha256" |
| + "database/sql" |
| "encoding/hex" |
| + "go/ast" |
| + "go/parser" |
| + "go/token" |
| + "io" |
| "log" |
| "net" |
| "net/http" |
| "net/http/httptest" |
| "os" |
| + "path/filepath" |
| "strings" |
| + "sync" |
| "testing" |
| "time" |
| "github.com/carlosframework/platform/internal/registry" |
| + "github.com/carlosframework/platform/internal/release" |
| ) |
| // sha256Hex is the test-side mirror of what FleetChannelAuth stores and |
| @@ -38,6 +47,35 @@ func tunnelReq(name, token string) *http.Request { |
| return req |
| } |
| +// runtimeSwapHijacker lands a registry-driven auth change at the one point |
| +// that used to be unsafe: after ServeHTTP's initial token check, but before it |
| +// receives and registers the hijacked connection. The peer drains the HTTP/2 |
| +// client preface so NewClientConn can complete without a real remote host. |
| +type runtimeSwapHijacker struct { |
| + header http.Header |
| + before func() |
| + peer net.Conn |
| +} |
| + |
| +func (w *runtimeSwapHijacker) Header() http.Header { return w.header } |
| +func (w *runtimeSwapHijacker) Write(p []byte) (int, error) { |
| + return len(p), nil |
| +} |
| +func (w *runtimeSwapHijacker) WriteHeader(int) {} |
| +func (w *runtimeSwapHijacker) Hijack() (net.Conn, *bufio.ReadWriter, error) { |
| + server, peer := net.Pipe() |
| + w.peer = peer |
| + go io.Copy(io.Discard, peer) |
| + w.before() |
| + return server, bufio.NewReadWriter(bufio.NewReader(server), bufio.NewWriter(server)), nil |
| +} |
| + |
| +func (w *runtimeSwapHijacker) close() { |
| + if w.peer != nil { |
| + w.peer.Close() |
| + } |
| +} |
| + |
| func TestTunnelRegisterRejectsBadToken(t *testing.T) { |
| // TWO clients on purpose: with only one configured, no test here can tell |
| // "every token is bound to its own channel" apart from "any known token |
| @@ -183,6 +221,31 @@ func TestTunnelRegisterRejectsUnknownDynamicChannel(t *testing.T) { |
| } |
| } |
| +func TestDynamicRegistrationRechecksRuntimeAfterHijack(t *testing.T) { |
| + for _, tc := range []struct { |
| + name string |
| + reload map[string]string |
| + }{ |
| + {name: "detach", reload: map[string]string{}}, |
| + {name: "rotate", reload: map[string]string{"acme/pi-1": sha256Hex("new-tok")}}, |
| + } { |
| + t.Run(tc.name, func(t *testing.T) { |
| + hub := NewTunnelHub("tunnel.test", nil) |
| + hub.SetDynamicAuth(map[string]string{"acme/pi-1": sha256Hex("old-tok")}) |
| + w := &runtimeSwapHijacker{ |
| + header: make(http.Header), |
| + before: func() { hub.SetDynamicAuth(tc.reload) }, |
| + } |
| + defer w.close() |
| + hub.ServeHTTP(w, tunnelReq("acme/pi-1", "old-tok")) |
| + if got := hub.State("acme/pi-1"); got != "" { |
| + hub.CloseChannel("acme/pi-1") |
| + t.Fatalf("old-token connection committed after %s runtime landed: state=%q", tc.name, got) |
| + } |
| + }) |
| + } |
| +} |
| + |
| // 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 |
| @@ -233,6 +296,56 @@ func TestVerifyDynamic(t *testing.T) { |
| } |
| } |
| +func TestAuthorizeDynamicReturnsOneRuntimeTuple(t *testing.T) { |
| + hub := NewTunnelHub("tunnel.test", nil) |
| + entries := []registry.FleetRuntimeEntry{{ |
| + Channel: "pi/box-a", Fleet: "pi", TokenHash: sha256Hex("tok-a"), Account: "acct-a", |
| + }} |
| + hub.SetDynamicRuntime(entries) |
| + entries[0] = registry.FleetRuntimeEntry{Channel: "pi/box-a", Fleet: "wrong", TokenHash: sha256Hex("wrong"), Account: "wrong"} |
| + if fleet, account, ok := hub.AuthorizeDynamic("pi/box-a", "tok-a"); !ok || fleet != "pi" || account != "acct-a" { |
| + t.Fatalf("AuthorizeDynamic = (%q, %q, %v), want (pi, acct-a, true)", fleet, account, ok) |
| + } |
| + hub.SetDynamicAuth(map[string]string{"pi/box-a": sha256Hex("tok-a")}) |
| + if fleet, ok := hub.VerifyDynamic("pi/box-a", "tok-a"); !ok || fleet != "pi" { |
| + t.Fatalf("SetDynamicAuth compatibility VerifyDynamic = (%q, %v), want (pi, true)", fleet, ok) |
| + } |
| + if fleet, account, ok := hub.AuthorizeDynamic("pi/box-a", "tok-a"); ok || fleet != "" || account != "" { |
| + t.Fatalf("account-less compatibility entry authorized store access: (%q, %q, %v)", fleet, account, ok) |
| + } |
| +} |
| + |
| +func TestDynamicRuntimeSwapNeverMixesTokenAndAccount(t *testing.T) { |
| + hub := NewTunnelHub("tunnel.test", nil) |
| + a := []registry.FleetRuntimeEntry{{Channel: "pi/box-a", Fleet: "pi", TokenHash: sha256Hex("tok-a"), Account: "acct-a"}} |
| + b := []registry.FleetRuntimeEntry{{Channel: "pi/box-a", Fleet: "pi", TokenHash: sha256Hex("tok-b"), Account: "acct-b"}} |
| + hub.SetDynamicRuntime(a) |
| + var wg sync.WaitGroup |
| + wg.Add(1) |
| + go func() { |
| + defer wg.Done() |
| + for i := 0; i < 500; i++ { |
| + if i%2 == 0 { |
| + hub.SetDynamicRuntime(b) |
| + } else { |
| + hub.SetDynamicRuntime(a) |
| + } |
| + } |
| + }() |
| + for i := 0; i < 1000; i++ { |
| + if fleet, ok := hub.VerifyDynamic("pi/box-a", "tok-a"); ok && fleet != "pi" { |
| + t.Fatalf("VerifyDynamic returned mixed fleet %q for token A", fleet) |
| + } |
| + if fleet, account, ok := hub.AuthorizeDynamic("pi/box-a", "tok-a"); ok && (fleet != "pi" || account != "acct-a") { |
| + t.Fatalf("token A authorized against mixed tuple (%q, %q)", fleet, account) |
| + } |
| + if fleet, account, ok := hub.AuthorizeDynamic("pi/box-a", "tok-b"); ok && (fleet != "pi" || account != "acct-b") { |
| + t.Fatalf("token B authorized against mixed tuple (%q, %q)", fleet, account) |
| + } |
| + } |
| + wg.Wait() |
| +} |
| + |
| // TestVerifyDynamicRefusesSlashlessNameInDynamicAuth is the one case only the |
| // "/" gate can refuse, and so the only test that actually pins it: the name is |
| // PRESENT in dynamicAuth under a hash that matches the presented token, so |
| @@ -271,7 +384,7 @@ func TestCloseChannelClosesAndRemovesLiveConn(t *testing.T) { |
| c1, c2 := net.Pipe() |
| defer c2.Close() |
| hub.mu.Lock() |
| - hub.conns["acme/pi-1"] = &tunnelConn{raw: c1} |
| + hub.conns["acme/pi-1"] = &tunnelConn{raw: c1, dynamic: true} |
| hub.mu.Unlock() |
| hub.CloseChannel("acme/pi-1") |
| @@ -316,7 +429,7 @@ func TestSetDynamicAuthRotationClosesVanishedChannelAndRefusesOldToken(t *testin |
| c1, c2 := net.Pipe() |
| defer c2.Close() |
| hub.mu.Lock() |
| - hub.conns["acme/pi-1"] = &tunnelConn{raw: c1} |
| + hub.conns["acme/pi-1"] = &tunnelConn{raw: c1, dynamic: true} |
| hub.mu.Unlock() |
| // Rotate: the entry vanishes from the dynamic set entirely. |
| @@ -339,6 +452,105 @@ func TestSetDynamicAuthRotationClosesVanishedChannelAndRefusesOldToken(t *testin |
| } |
| } |
| +func TestDynamicRuntimeDetachDoesNotCloseStaticConnection(t *testing.T) { |
| + hub := NewTunnelHub("tunnel.test", nil) |
| + hub.SetDynamicAuth(map[string]string{"acme/pi-1": sha256Hex("tok")}) |
| + c1, c2 := net.Pipe() |
| + defer c1.Close() |
| + defer c2.Close() |
| + hub.mu.Lock() |
| + want := &tunnelConn{raw: c1, dynamic: false} |
| + hub.conns["acme/pi-1"] = want |
| + hub.mu.Unlock() |
| + |
| + hub.SetDynamicRuntime(nil) |
| + hub.mu.Lock() |
| + got := hub.conns["acme/pi-1"] |
| + hub.mu.Unlock() |
| + if got != want { |
| + t.Fatal("dynamic detach removed a static connection sharing the name") |
| + } |
| + if err := c1.SetWriteDeadline(time.Now().Add(time.Second)); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := c2.SetReadDeadline(time.Now().Add(time.Second)); err != nil { |
| + t.Fatal(err) |
| + } |
| + wrote := make(chan error, 1) |
| + go func() { _, err := c1.Write([]byte("x")); wrote <- err }() |
| + buf := make([]byte, 1) |
| + if _, err := c2.Read(buf); err != nil { |
| + t.Fatalf("static connection was closed: %v", err) |
| + } |
| + if err := <-wrote; err != nil { |
| + t.Fatalf("static connection write failed: %v", err) |
| + } |
| +} |
| + |
| +func TestFleetRuntimeReloadsSerializeReadThroughInstall(t *testing.T) { |
| + hub := NewTunnelHub("tunnel.test", nil) |
| + e := &Edge{tunnels: hub} |
| + firstEntered := make(chan struct{}) |
| + releaseFirst := make(chan struct{}) |
| + secondEntered := make(chan struct{}) |
| + firstDone := make(chan error, 1) |
| + secondDone := make(chan error, 1) |
| + a := []registry.FleetRuntimeEntry{{Channel: "pi/box", Fleet: "pi", TokenHash: sha256Hex("tok-a"), Account: "acct-a"}} |
| + b := []registry.FleetRuntimeEntry{{Channel: "pi/box", Fleet: "pi", TokenHash: sha256Hex("tok-b"), Account: "acct-b"}} |
| + go func() { |
| + firstDone <- e.reloadFleetRuntime(func() ([]registry.FleetRuntimeEntry, error) { |
| + close(firstEntered) |
| + <-releaseFirst |
| + return a, nil |
| + }) |
| + }() |
| + <-firstEntered |
| + go func() { |
| + secondDone <- e.reloadFleetRuntime(func() ([]registry.FleetRuntimeEntry, error) { |
| + close(secondEntered) |
| + return b, nil |
| + }) |
| + }() |
| + select { |
| + case <-secondEntered: |
| + close(releaseFirst) |
| + t.Fatal("second fleet runtime read entered before the first read/install completed") |
| + case <-time.After(100 * time.Millisecond): |
| + } |
| + close(releaseFirst) |
| + if err := <-firstDone; err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := <-secondDone; err != nil { |
| + t.Fatal(err) |
| + } |
| + if _, account, ok := hub.AuthorizeDynamic("pi/box", "tok-b"); !ok || account != "acct-b" { |
| + t.Fatalf("last serialized snapshot not installed: account=%q ok=%v", account, ok) |
| + } |
| + if _, _, ok := hub.AuthorizeDynamic("pi/box", "tok-a"); ok { |
| + t.Fatal("older snapshot installed after newer serialized reload") |
| + } |
| +} |
| + |
| +func TestEdgeFleetRuntimeWiringUsesOnlyCombinedSnapshot(t *testing.T) { |
| + fset := token.NewFileSet() |
| + f, err := parser.ParseFile(fset, "edge.go", nil, 0) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + counts := map[string]int{} |
| + ast.Inspect(f, func(n ast.Node) bool { |
| + if sel, ok := n.(*ast.SelectorExpr); ok { |
| + counts[sel.Sel.Name]++ |
| + } |
| + return true |
| + }) |
| + if counts["FleetRuntimeSnapshot"] != 1 || counts["SetDynamicRuntime"] != 1 || counts["SetDynamicAuth"] != 0 { |
| + t.Fatalf("edge.go fleet runtime calls = snapshot:%d install:%d old-install:%d, want 1,1,0", |
| + counts["FleetRuntimeSnapshot"], counts["SetDynamicRuntime"], counts["SetDynamicAuth"]) |
| + } |
| +} |
| + |
| func TestParseTunnelClients(t *testing.T) { |
| got, err := parseTunnelClients("pi-box=sekrit,hq=other") |
| if err != nil || got["pi-box"] != "sekrit" || got["hq"] != "other" { |
| @@ -552,6 +764,63 @@ func TestReloadWithNilTunnelHubIsSafe(t *testing.T) { |
| } |
| } |
| +func TestReloadFleetReadFailureLeavesPriorAuthAndAccountTogether(t *testing.T) { |
| + dir := t.TempDir() |
| + path := filepath.Join(dir, "registry.db") |
| + reg, err := registry.Open(path) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := reg.UpsertFleet(registry.Fleet{Name: testFleet, Account: testSqid, Bucket: "pi-data"}); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := reg.AddFleetBox(registry.FleetBox{Fleet: testFleet, Label: "box-a", TokenHash: sha256Hex(testToken), 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) |
| + store := release.NewDirStore(filepath.Join(dir, "store")) |
| + e.EnableStatic(store, nil, filepath.Join(dir, "cache")) |
| + if err := e.Reload(); err != nil { |
| + t.Fatal(err) |
| + } |
| + if _, err := store.Put(t.Context(), ownKey, []byte("old-account"), ""); err != nil { |
| + t.Fatal(err) |
| + } |
| + request := func(token string) *httptest.ResponseRecorder { |
| + w := httptest.NewRecorder() |
| + e.door.ServeHTTP(w, doorReq(http.MethodGet, "/store/o/"+ownKey, testChannel, token, "")) |
| + return w |
| + } |
| + if got := request(testToken); got.Code != http.StatusOK || got.Body.String() != "old-account" { |
| + t.Fatalf("initial old-token/account request = %d %q", got.Code, got.Body.String()) |
| + } |
| + if err := reg.SetFleetBoxToken(testFleet, "box-a", sha256Hex("new-token")); err != nil { |
| + t.Fatal(err) |
| + } |
| + breaker, err := sql.Open("sqlite", path) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + defer breaker.Close() |
| + if _, err := breaker.Exec(`DROP TABLE fleets`); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := e.Reload(); err == nil { |
| + t.Fatal("Reload after dropping fleets succeeded, want a real fleet read failure") |
| + } |
| + if got := request(testToken); got.Code != http.StatusOK || got.Body.String() != "old-account" { |
| + t.Fatalf("failed reload split prior auth/account: old token request = %d %q, want 200 old-account", got.Code, got.Body.String()) |
| + } |
| + if got := request("new-token"); got.Code != http.StatusNotFound { |
| + t.Fatalf("failed reload published new token: status=%d, want hidden 404", got.Code) |
| + } |
| +} |
| + |
| // 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 |
| @@ -578,7 +847,7 @@ func TestReloadDetachedFleetBoxClosesLiveChannel(t *testing.T) { |
| c1, c2 := net.Pipe() |
| defer c2.Close() |
| hub.mu.Lock() |
| - hub.conns["acme/pi-1"] = &tunnelConn{raw: c1} |
| + hub.conns["acme/pi-1"] = &tunnelConn{raw: c1, dynamic: true} |
| hub.mu.Unlock() |
| if err := reg.SetFleetBoxStatus("acme", "pi-1", "detached"); err != nil { |
| @@ -628,7 +897,7 @@ func TestRotateVsDetachAsymmetry(t *testing.T) { |
| c1, c2 := net.Pipe() |
| defer c2.Close() |
| hub.mu.Lock() |
| - hub.conns["acme/pi-1"] = &tunnelConn{raw: c1} |
| + hub.conns["acme/pi-1"] = &tunnelConn{raw: c1, dynamic: true} |
| hub.mu.Unlock() |
| // --- Rotate --- |
diff --git a/internal/registry/fleets.go b/internal/registry/fleets.go| index 3593ddfef..ce5281121 100644 |
| --- a/internal/registry/fleets.go |
| +++ b/internal/registry/fleets.go |
| @@ -8,10 +8,12 @@ |
| package registry |
| import ( |
| + "context" |
| "database/sql" |
| "errors" |
| "fmt" |
| "regexp" |
| + "sort" |
| "strings" |
| "time" |
| ) |
| @@ -46,6 +48,16 @@ type FleetBox struct { |
| Tags string // comma-separated, ValidBoxTags grammar; "" = untagged |
| } |
| +// FleetRuntimeEntry is one active fleet channel's complete edge-runtime |
| +// identity. Channel/token authentication and fleet/account authorization are |
| +// read together so an edge can install them as one immutable snapshot. |
| +type FleetRuntimeEntry struct { |
| + Channel string |
| + Fleet string |
| + TokenHash string |
| + Account string |
| +} |
| + |
| // tokenHashRe: exactly 64 lowercase hex characters — a sha256 digest, |
| // structurally enforcing the "token hashes only, never plaintext" rule at |
| // every writer that accepts one (AddFleetBox, SetFleetBoxToken). A bearer |
| @@ -477,3 +489,53 @@ func (r *Registry) FleetChannelAuth() (map[string]string, error) { |
| } |
| return out, rows.Err() |
| } |
| + |
| +// FleetRuntimeSnapshot returns every active fleet channel with its owning |
| +// account from one SQLite statement. The dedicated fresh connection is the |
| +// same cross-process freshness boundary as FleetChannelAuth/ListFleets, while |
| +// the LEFT JOIN preserves their existing orphan behavior: a manually-created |
| +// box with no fleet row can authenticate, but carries no account for the store |
| +// door to authorize. |
| +func (r *Registry) FleetRuntimeSnapshot() ([]FleetRuntimeEntry, error) { |
| + ctx := context.Background() |
| + db, err := sql.Open("sqlite", r.path) |
| + if err != nil { |
| + return nil, err |
| + } |
| + defer db.Close() |
| + conn, err := db.Conn(ctx) |
| + if err != nil { |
| + return nil, err |
| + } |
| + defer conn.Close() |
| + if _, err := conn.ExecContext(ctx, "PRAGMA busy_timeout=5000"); err != nil { |
| + return nil, err |
| + } |
| + rows, err := conn.QueryContext(ctx, `SELECT b.fleet,b.label,b.token_hash,COALESCE(f.account,'') |
| + FROM fleet_boxes b LEFT JOIN fleets f ON f.name=b.fleet |
| + WHERE b.status='active'`) |
| + if err != nil { |
| + return nil, err |
| + } |
| + out := make([]FleetRuntimeEntry, 0) |
| + for rows.Next() { |
| + var fleet, label string |
| + var entry FleetRuntimeEntry |
| + if err := rows.Scan(&fleet, &label, &entry.TokenHash, &entry.Account); err != nil { |
| + rows.Close() |
| + return nil, err |
| + } |
| + entry.Fleet = fleet |
| + entry.Channel = fleet + "/" + label |
| + out = append(out, entry) |
| + } |
| + if err := rows.Err(); err != nil { |
| + rows.Close() |
| + return nil, err |
| + } |
| + if err := rows.Close(); err != nil { |
| + return nil, err |
| + } |
| + sort.Slice(out, func(i, j int) bool { return out[i].Channel < out[j].Channel }) |
| + return out, nil |
| +} |
diff --git a/internal/registry/fleets_test.go b/internal/registry/fleets_test.go| index ca052b334..fbbc64fc0 100644 |
| --- a/internal/registry/fleets_test.go |
| +++ b/internal/registry/fleets_test.go |
| @@ -225,6 +225,119 @@ func TestFleetChannelAuthOmitsDetachedBoxes(t *testing.T) { |
| } |
| } |
| +func TestFleetRuntimeSnapshotCompleteSortedAndNonNil(t *testing.T) { |
| + reg := openTestRegistry(t) |
| + for _, fleet := range []Fleet{ |
| + {Name: "a", Account: "acct-a", Bucket: "a-data"}, |
| + {Name: "a-b", Account: "acct-ab", Bucket: "ab-data"}, |
| + } { |
| + if err := reg.UpsertFleet(fleet); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + for _, box := range []FleetBox{ |
| + {Fleet: "a", Label: "x", TokenHash: tokenHashA, Status: "active"}, |
| + {Fleet: "a-b", Label: "x", TokenHash: tokenHashB, Status: "active"}, |
| + {Fleet: "a", Label: "gone", TokenHash: tokenHashC, Status: "detached"}, |
| + } { |
| + if err := reg.AddFleetBox(box); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + if _, err := reg.sql.Exec(`INSERT INTO fleet_boxes(fleet,label,token_hash,status,added_at) VALUES(?,?,?,?,?)`, |
| + "orphan", "box", tokenHashC, "active", 1); err != nil { |
| + t.Fatal(err) |
| + } |
| + |
| + got, err := reg.FleetRuntimeSnapshot() |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + want := []FleetRuntimeEntry{ |
| + {Channel: "a-b/x", Fleet: "a-b", TokenHash: tokenHashB, Account: "acct-ab"}, |
| + {Channel: "a/x", Fleet: "a", TokenHash: tokenHashA, Account: "acct-a"}, |
| + {Channel: "orphan/box", Fleet: "orphan", TokenHash: tokenHashC, Account: ""}, |
| + } |
| + if len(got) != len(want) { |
| + t.Fatalf("FleetRuntimeSnapshot = %+v, want %+v", got, want) |
| + } |
| + for i := range want { |
| + if got[i] != want[i] { |
| + t.Fatalf("FleetRuntimeSnapshot[%d] = %+v, want %+v", i, got[i], want[i]) |
| + } |
| + } |
| + |
| + empty := openTestRegistry(t) |
| + got, err = empty.FleetRuntimeSnapshot() |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if got == nil || len(got) != 0 { |
| + t.Fatalf("empty FleetRuntimeSnapshot = %#v, want nonnil empty slice", got) |
| + } |
| +} |
| + |
| +func TestFleetRuntimeSnapshotNeverMixesCommittedPair(t *testing.T) { |
| + reg := openTestRegistry(t) |
| + if err := reg.UpsertFleet(Fleet{Name: "acme", Account: "acct-a", Bucket: "data"}); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := reg.AddFleetBox(FleetBox{Fleet: "acme", Label: "box", TokenHash: tokenHashA, Status: "active"}); err != nil { |
| + t.Fatal(err) |
| + } |
| + writer, err := Open(reg.path) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + defer writer.Close() |
| + |
| + done := make(chan error, 1) |
| + go func() { |
| + for i := 0; i < 100; i++ { |
| + account, hash := "acct-a", tokenHashA |
| + if i%2 == 0 { |
| + account, hash = "acct-b", tokenHashB |
| + } |
| + tx, err := writer.sql.Begin() |
| + if err == nil { |
| + _, err = tx.Exec(`UPDATE fleets SET account=?, rev=rev+1 WHERE name='acme'`, account) |
| + } |
| + if err == nil { |
| + _, err = tx.Exec(`UPDATE fleet_boxes SET token_hash=?, rev=rev+1 WHERE fleet='acme' AND label='box'`, hash) |
| + } |
| + if err != nil { |
| + if tx != nil { |
| + tx.Rollback() |
| + } |
| + done <- err |
| + return |
| + } |
| + if err := tx.Commit(); err != nil { |
| + done <- err |
| + return |
| + } |
| + } |
| + done <- nil |
| + }() |
| + for i := 0; i < 100; i++ { |
| + got, err := reg.FleetRuntimeSnapshot() |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if len(got) != 1 { |
| + t.Fatalf("snapshot = %+v, want one entry", got) |
| + } |
| + entry := got[0] |
| + if !((entry.Account == "acct-a" && entry.TokenHash == tokenHashA) || |
| + (entry.Account == "acct-b" && entry.TokenHash == tokenHashB)) { |
| + t.Fatalf("mixed committed pair: %+v", entry) |
| + } |
| + } |
| + if err := <-done; err != nil { |
| + t.Fatal(err) |
| + } |
| +} |
| + |
| func TestSetFleetBoxTokenMissingRowErrors(t *testing.T) { |
| reg := openTestRegistry(t) |
| if err := reg.SetFleetBoxToken("acme", "nope", tokenHashA); err == nil { |