carlos / platform Public

provision: a cached policy verification never outlives its key (#328)

The policy cache records the release key its entries were verified with
and forgets them all when a pass brings a different key (set, cleared or
replaced), so a policy verified under one key is never trusted under
another. The verification test now runs a persistent cache through a
nil key and a replacement key, across a backstop with the object
unchanged, and back.

Both provisioning loops build their pass config through
provisionPassConfig, which carries the one cache armPolicyCache builds
after the change index; TestProvisioningLoopsShareThePolicyCache pins it.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

claude 1 day ago 2133c5d5623e5755be0b361adccd94cd6e549b4e parent 02f55fd

4 files changed, +125 −6

internal/host/agent.go +19 −6
internal/host/policycache_wiring_test.go +67 −0
internal/provision/apppolicy_cache.go +12 −0
internal/provision/apppolicy_reconcile_test.go +27 −0

diff --git a/internal/host/agent.go b/internal/host/agent.go

index 147304d9e..00139089b 100644

--- a/internal/host/agent.go

+++ b/internal/host/agent.go

@@ -806,7 +806,7 @@ func runAgent(args []string, buildClaims ClaimFactory) {

// The app policy cache (#328, #329), built once the index is armed

// so it follows the watch just armed (nil on a fleet or disarmed

// box), before any loop that reconciles.

- armedPolicyCache.Store(provision.NewPolicyCache(armedChangeWatch.Load()))

+ armPolicyCache()

go adoptLoop(reg, svc, sd, claims.Writes)

go fastConvergeLoop(reg, svc, sd, claims.Writes, serialClaims, bs, claims.Providers.Provision)

if logSink != nil {

@@ -1588,6 +1588,22 @@ var reconcileMu sync.Mutex

// (tests) it is nil and each pass builds a throwaway.

var armedPolicyCache atomic.Pointer[provision.PolicyCache]

+// armPolicyCache builds the process's policy cache over the armed change

+// watch. RunAgent calls it once, after armChangeIndex.

+func armPolicyCache() {

+ armedPolicyCache.Store(provision.NewPolicyCache(armedChangeWatch.Load()))

+// provisionPassConfig is the config of one instances reconcile pass from

+// either provisioning loop (fastConvergeLoop's full pass, instancesLoop):

+// the environment's, the claim wiring, and the process's one policy cache.

+// Pinned by TestProvisioningLoopsShareThePolicyCache.

+func provisionPassConfig(configure []func(*provision.ReconcileConfig)) provision.ReconcileConfig {

+ cfg := configuredClaimProvision(provisionTickConfig(os.Getenv), configure)

+ cfg.Policies = armedPolicyCache.Load()

+ return cfg

// reconcileSerialized is the one way any agent loop runs the record-family

// reconcile passes — the mutex lives in a shared wrapper, not at the call

// sites, so a future third caller cannot forget it. Pinned by

@@ -3049,9 +3065,7 @@ func fastConvergeLoop(reg *registry.Registry, svc *activator.Service, sd *schedu

legs := &passLegs{store: store, reg: reg, pub: pub, optDir: optDir, restart: restart, opts: opts,

priv: priv, dd: dd, svc: svc, writes: writes, bs: bs,

cfg: func() provision.ReconcileConfig {

- cfg := configuredClaimProvision(provisionTickConfig(os.Getenv), configure)

- cfg.Policies = armedPolicyCache.Load()

- return cfg

+ return provisionPassConfig(configure)

}}

d.fullPass = legs.full

// The change index (S3 change-index design §4.5): an index-only change

@@ -3592,8 +3606,7 @@ func instancesLoop(reg *registry.Registry, writes hostwrite.Config, bs *buildSta

log.Printf("agent: instances loop disabled: %v", err)

return

}

- cfg := configuredClaimProvision(provisionTickConfig(os.Getenv), configure)

- cfg.Policies = armedPolicyCache.Load()

+ cfg := provisionPassConfig(configure)

backfill := newClaimBackfill(writes, store, cfg.ReleasePub, os.Getenv("CARLOS_DEPLOYMENT_BUCKET"))

log.Printf("agent: instances loop armed (5min + jitter), region %q pools %v", cfg.Region, cfg.Pools)

// Platform-only: a fleet box (CARLOS_FLEET set) never reads other

diff --git a/internal/host/policycache_wiring_test.go b/internal/host/policycache_wiring_test.go

new file mode 100644

index 000000000..36815a27d

--- /dev/null

+++ b/internal/host/policycache_wiring_test.go

@@ -0,0 +1,67 @@

+package host

+import (

+ "os"

+ "strings"

+ "testing"

+ "github.com/carlosframework/platform/internal/release"

+// TestProvisioningLoopsShareThePolicyCache pins the app policy cache's

+// wiring (#328, #329; spec §3a): RunAgent builds one cache after arming the

+// change index, and both provisioning loops hand that same cache to every

+// pass. A second cache, or a throwaway one per pass, would multiply the

+// policy reads the budget allows. Not parallel: it swaps the armed watch

+// and the armed cache.

+func TestProvisioningLoopsShareThePolicyCache(t *testing.T) {

+ prevWatch, prevCache := armedChangeWatch.Load(), armedPolicyCache.Load()

+ t.Cleanup(func() {

+ armedChangeWatch.Store(prevWatch)

+ armedPolicyCache.Store(prevCache)

+ })

+ armedChangeWatch.Store(release.NewChangeWatch(release.NewDirStore(t.TempDir())))

+ armedPolicyCache.Store(nil)

+ armPolicyCache()

+ built := armedPolicyCache.Load()

+ if built == nil {

+ t.Fatal("armPolicyCache built no cache")

+ }

+ a, b := provisionPassConfig(nil), provisionPassConfig(nil)

+ if a.Policies != built || b.Policies != built {

+ t.Fatalf("pass configs carry caches %p and %p, want the armed %p in both", a.Policies, b.Policies, built)

+ }

+ // Both loops build their pass config through provisionPassConfig, and

+ // RunAgent arms the cache after the change index (syntactic: a loop

+ // that builds its config another way fails here).

+ src, err := os.ReadFile("agent.go")

+ if err != nil {

+ t.Fatal(err)

+ }

+ text := string(src)

+ if n := strings.Count(text, "configuredClaimProvision(provisionTickConfig("); n != 1 {

+ t.Fatalf("agent.go builds a pass config outside provisionPassConfig (%d sites, want 1)", n)

+ }

+ if n := strings.Count(text, "provisionPassConfig(configure)"); n != 2 {

+ t.Fatalf("agent.go has %d provisionPassConfig(configure) calls, want 2 (fastConvergeLoop and instancesLoop)", n)

+ }

+ for _, loop := range []string{"func fastConvergeLoop(", "func instancesLoop("} {

+ i := strings.Index(text, loop)

+ if i < 0 {

+ t.Fatalf("agent.go has no %s", loop)

+ }

+ body := text[i:]

+ if j := strings.Index(body[1:], "\nfunc "); j >= 0 {

+ body = body[:j+1]

+ }

+ if !strings.Contains(body, "provisionPassConfig(configure)") {

+ t.Fatalf("%s...) does not build its pass config with provisionPassConfig", loop)

+ }

+ arm, cache := strings.Index(text, "armChangeIndex(agentCtx"), strings.Index(text, "\t\tarmPolicyCache()")

+ if arm < 0 || cache < 0 || cache < arm {

+ t.Fatal("RunAgent must call armPolicyCache() after armChangeIndex")

+ }

diff --git a/internal/provision/apppolicy_cache.go b/internal/provision/apppolicy_cache.go

index 2c0e85ea6..149075840 100644

--- a/internal/provision/apppolicy_cache.go

+++ b/internal/provision/apppolicy_cache.go

@@ -1,6 +1,7 @@

package provision

import (

+ "bytes"

"context"

"crypto/ed25519"

"errors"

@@ -61,6 +62,11 @@ import (

// A verified policy's expiry is re-checked against the pass's clock on every

// pass with no read: a cached record past its lifetime is expired.

+// Every entry was verified with one release key, the cache's pub. A pass

+// with a different key (a key set, cleared or replaced) forgets every entry

+// before it decides anything, so no policy verified under the old key is

+// trusted under the new one; each app is then new to the cache (rule 4).

+//

// Safe for concurrent use; a read holds the cache for its whole run, so two

// passes never read the same app twice. Production builds one per agent

// process and hands it to every pass (ReconcileConfig.Policies).

@@ -69,6 +75,8 @@ type PolicyCache struct {

mu sync.Mutex

entries map[[2]string]*policyEntry

+ // pub is the key every entry was verified with.

+ pub ed25519.PublicKey

// next is when the next backstop is due; zero before the first one

// completes. failed says the last backstop's listing failed.

next time.Time

@@ -124,6 +132,10 @@ func (c *PolicyCache) read(ctx context.Context, s release.Store, pub ed25519.Pub

c.mu.Lock()

defer c.mu.Unlock()

out := make(map[[2]string]policyRead, len(apps))

+ if !bytes.Equal(c.pub, pub) {

+ clear(c.entries)

+ c.pub = bytes.Clone(pub)

+ }

if len(apps) == 0 {

return out

}

diff --git a/internal/provision/apppolicy_reconcile_test.go b/internal/provision/apppolicy_reconcile_test.go

index 8563dac05..f1dc43ae0 100644

--- a/internal/provision/apppolicy_reconcile_test.go

+++ b/internal/provision/apppolicy_reconcile_test.go

@@ -240,6 +240,33 @@ func TestReconcileVerifiesBeforeConverging(t *testing.T) {

fx.run()

wantCols(t, fx.row(policyHost), "", "", 0)

fx.wantState("policy unverifiable")

+ // The same through the agent's long-lived cache, where nothing is due

+ // to be read again: a cached verification never outlives its key.

+ fx.cfg.ReleasePub = fx.pub

+ fx.cfg.Policies = NewPolicyCache(nil)

+ converge() // the cache's first pass: its backstop

+ exp := fx.now.Add(apppolicy.Lifetime).Unix()

+ _, otherPub := releaseKeys(t)

+ for _, c := range []struct {

+ name string

+ pub ed25519.PublicKey

+ }{{"no key", nil}, {"a replacement key", otherPub}} {

+ fx.cfg.ReleasePub = c.pub

+ fx.run()

+ wantCols(t, fx.row(policyHost), "", "", 0)

+ fx.wantState("policy unverifiable")

+ // A later backstop with the object unchanged keeps it refused.

+ fx.now = fx.now.Add(34 * time.Minute)

+ fx.run()

+ wantCols(t, fx.row(policyHost), "", "", 0)

+ fx.wantState("policy unverifiable")

+ // The right key again: verified again, not from the stale entry.

+ fx.cfg.ReleasePub = fx.pub

+ fx.run()

+ wantCols(t, fx.row(policyHost), "10.0.5.20", "large", exp)

+ fx.wantState("")

+ }

}

func TestReconcileGivesSidecarMembersNoPolicy(t *testing.T) {