amadan

carlos / platform Public

console: re-sign every app policy on the door keys' loop (#328, #329)

StartDoorKeyResign now runs resignAppPolicies after the door keys' pass.
Candidates are the table's live and pending rows on every pass, plus a
listing of control/app-policy/ only on the first pass of a holding or
while holding none. Each app is re-signed from its recorded content at
its recorded revision under one writer holding, bound per attempt;
absent objects are never recreated, unverifiable unrecorded bytes are
never signed, and a lost holding stops the pass.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 3e86e9d379dcbd98c62aa35e4f88590b9f239bbc parent c5632c4
8 files changed, +1231 −23
  • .amadan/test-weights.tsv +12 −10
  • internal/console/apidoorkeys.go +3 −1
  • internal/console/apipolicy.go +311 −7
  • internal/console/apipolicy_resign_test.go +534 −0
  • internal/console/apipolicy_test.go +17 −3
  • internal/console/claim_invariant_fixtures_test.go +4 −0
  • internal/console/policylease_twoconsoles_test.go +345 −0
  • internal/console/server.go +5 −2
diff --git a/.amadan/test-weights.tsv b/.amadan/test-weights.tsv
index 6241b5499..d67600acc 100644
--- a/.amadan/test-weights.tsv
+++ b/.amadan/test-weights.tsv
@@ -1,5 +1,16 @@
-# package test seconds (.amadan/test-weights.sh, 2026-10-21)
+# package test seconds (.amadan/test-weights.sh, 2026-10-25)
+github.com/carlosframework/platform TestDBForwardPrintsNothingForABadAnswer 67.0
+github.com/carlosframework/platform TestDBForwardLongRequestThroughTheEdge 44.5
+github.com/carlosframework/platform TestDBForwardThroughTheBackhaulHop 44.1
+github.com/carlosframework/platform TestDBForwardPullParityThroughTheCLI 4.8
+github.com/carlosframework/platform TestDBForwardParityThroughTheCLI 4.6
+github.com/carlosframework/platform TestDBForwardPullInterruptedLeavesNothing 2.1
+github.com/carlosframework/platform TestDBForwardMixedVersions 1.9
+github.com/carlosframework/platform TestDBForwardPullSavesNothingForABadAnswer 1.8
+github.com/carlosframework/platform TestDBForwardRefusesSQLThatIsNotUTF8 1.0
+github.com/carlosframework/platform/internal/console TestALapsedWriterCannotOverwriteTheNextWritersRevoke 60.5
github.com/carlosframework/platform/internal/console TestClaimConsoleMutationInvariant 60.1
+github.com/carlosframework/platform/internal/console TestPolicyWriterLeaseIsOneConsolePerBucket 30.4
github.com/carlosframework/platform/internal/console TestClaimConsoleMutationInvariantWitnesses 12.3
github.com/carlosframework/platform/internal/console TestRetiringAnInstancePurgesItsMoveLogRows 9.3
github.com/carlosframework/platform/internal/console TestDeclaredLostIsLoggedByTheConsole 7.2
@@ -37,12 +48,3 @@ github.com/carlosframework/platform/internal/console TestAnIncompletePlacementRe
github.com/carlosframework/platform/internal/console TestClaimLossPagesScanProgressSurvivesRestartAndWraps 1.0
github.com/carlosframework/platform/internal/console TestCreateDoorsRefuseWithNoRoomBeforeRegistering 1.0
github.com/carlosframework/platform/internal/console TestDBReadRequestBoundClampsAForgedDeadline 1.0
-github.com/carlosframework/platform TestDBForwardPrintsNothingForABadAnswer 67.0
-github.com/carlosframework/platform TestDBForwardLongRequestThroughTheEdge 44.5
-github.com/carlosframework/platform TestDBForwardThroughTheBackhaulHop 44.1
-github.com/carlosframework/platform TestDBForwardPullParityThroughTheCLI 4.8
-github.com/carlosframework/platform TestDBForwardParityThroughTheCLI 4.6
-github.com/carlosframework/platform TestDBForwardPullInterruptedLeavesNothing 2.1
-github.com/carlosframework/platform TestDBForwardMixedVersions 1.9
-github.com/carlosframework/platform TestDBForwardPullSavesNothingForABadAnswer 1.8
-github.com/carlosframework/platform TestDBForwardRefusesSQLThatIsNotUTF8 1.0
diff --git a/internal/console/apidoorkeys.go b/internal/console/apidoorkeys.go
index 8a4e8ecea..41c1f07df 100644
--- a/internal/console/apidoorkeys.go
+++ b/internal/console/apidoorkeys.go
@@ -343,7 +343,8 @@ func (s *Server) resignDoorKeys(ctx context.Context) {
}
}
-// StartDoorKeyResign re-signs every door-key record now and then every 12
+// StartDoorKeyResign re-signs every door-key record (resignDoorKeys) and then
+// every app policy (resignAppPolicies, apipolicy.go) now and then every 12
// hours until ctx is done. Started from cmd/console beside the other passes,
// and only on a console that holds the signing key: a read-only console has
// nothing to sign with.
@@ -355,6 +356,7 @@ func (s *Server) StartDoorKeyResign(ctx context.Context) {
defer t.Stop()
for {
s.resignDoorKeys(ctx)
+ s.resignAppPolicies(ctx)
select {
case <-ctx.Done():
return
diff --git a/internal/console/apipolicy.go b/internal/console/apipolicy.go
index d09f922ab..ec9fb05ef 100644
--- a/internal/console/apipolicy.go
+++ b/internal/console/apipolicy.go
@@ -188,6 +188,30 @@ func (s *Server) policyBase(ctx context.Context, acct *Account, app string, h po
// policyBaseFor is policyBase that also answers the app's current
// incarnation under claims ("" without), which the door stamps.
func (s *Server) policyBaseFor(ctx context.Context, acct *Account, app string, h policyHolding, commit bool) (base apppolicy.Policy, etag string, next uint64, row AppPolicyRow, cur string, err error) {
+ r, err := s.readPolicyBase(ctx, acct, app, h, commit)
+ return r.base, r.etag, r.next, r.row, r.cur, err
+}
+
+// policyBaseRead is readPolicyBase's whole answer: policyBaseFor's, and the
+// bucket's object as read (bucket nil when it is absent or does not verify),
+// which the re-sign pass compares with the recorded bytes.
+type policyBaseRead struct {
+ base apppolicy.Policy
+ etag string
+ next uint64
+ row AppPolicyRow
+ cur string
+ bucket *apppolicy.Policy
+ wire []byte
+}
+
+func (s *Server) readPolicyBase(ctx context.Context, acct *Account, app string, h policyHolding, commit bool) (policyBaseRead, error) {
+ var (
+ base apppolicy.Policy
+ row AppPolicyRow
+ cur string
+ err error
+ )
pub := s.policyPub()
key := acct.Sqid + "/" + app
b, wire, etag, gerr := apppolicy.GetForUpdate(ctx, s.cfg.Releases, pub, acct.Sqid, app)
@@ -199,11 +223,11 @@ func (s *Server) policyBaseFor(ctx context.Context, acct *Account, app string, h
case errors.Is(gerr, apppolicy.ErrUnverifiable):
b = nil
default:
- return base, "", 0, row, "", &policyRefusal{http.StatusBadGateway, policyReadFailedMsg, gerr}
+ return policyBaseRead{base: base, row: row}, &policyRefusal{http.StatusBadGateway, policyReadFailedMsg, gerr}
}
row, err = s.store.AppPolicyRecordContext(ctx, acct.ID, app)
if err != nil {
- return base, "", 0, row, "", &policyRefusal{http.StatusBadGateway, policyReadFailedMsg, err}
+ return policyBaseRead{base: base, row: row}, &policyRefusal{http.StatusBadGateway, policyReadFailedMsg, err}
}
recorded := row.Revision
var bucketRev uint64
@@ -251,7 +275,7 @@ func (s *Server) policyBaseFor(ctx context.Context, acct *Account, app string, h
break
}
if err := adopt(); err != nil {
- return base, "", 0, row, "", err
+ return policyBaseRead{base: base, row: row}, err
}
base = *b
s.policyLogf("console: app policy %s/%s: new writer holding; building on the bucket's record (revision %d)",
@@ -261,13 +285,13 @@ func (s *Server) policyBaseFor(ctx context.Context, acct *Account, app string, h
// and then lost the row of: only a holder of the release key signs
// one, so it is adopted.
if err := adopt(); err != nil {
- return base, "", 0, row, "", err
+ return policyBaseRead{base: base, row: row}, err
}
base = *b
case recorded > 0:
p, err := apppolicy.OpenKey(pub, row.Wire, acct.Sqid, app)
if err != nil {
- return base, "", 0, row, "", &policyRefusal{http.StatusBadGateway, policyReadFailedMsg, err}
+ return policyBaseRead{base: base, row: row}, &policyRefusal{http.StatusBadGateway, policyReadFailedMsg, err}
}
base = p
switch {
@@ -287,7 +311,7 @@ func (s *Server) policyBaseFor(ctx context.Context, acct *Account, app string, h
if s.cfg.HostClaim {
cur, err = s.store.ClaimAppID(acct.ID, app)
if err != nil {
- return base, "", 0, row, "", &policyRefusal{http.StatusServiceUnavailable, policyReadFailedMsg, err}
+ return policyBaseRead{base: base, row: row}, &policyRefusal{http.StatusServiceUnavailable, policyReadFailedMsg, err}
}
if base.AppID != cur {
// A predecessor's policy that a failed create-time clear left.
@@ -301,7 +325,8 @@ func (s *Server) policyBaseFor(ctx context.Context, acct *Account, app string, h
if commit && durable {
s.markPolicyTouched(key, h.n)
}
- return base, etag, max(recorded, bucketRev) + 1, row, cur, nil
+ return policyBaseRead{base: base, etag: etag, next: max(recorded, bucketRev) + 1, row: row, cur: cur,
+ bucket: b, wire: wire}, nil
}
// samePolicyPayload reports whether two signed records carry the same
@@ -620,6 +645,285 @@ func (s *Server) publishPolicy(ctx context.Context, acct *Account, app string, h
return "", unpublished
}
+// policyResignAction is the re-sign's app mutation.
+const policyResignAction = "app-policy-resign"
+
+// errPolicyResignLost is resignAppPolicy's answer when the pass's writer
+// holding is lost: the pass stops there.
+var errPolicyResignLost = errors.New("console: app policies: the writer lease was lost")
+
+// policyCandidate is one app the re-sign pass reads.
+type policyCandidate struct {
+ acct *Account
+ app string
+}
+
+// resignAppPolicies re-signs every live app policy with a fresh SignedAt, as
+// resignDoorKeys does the door-key records: StartDoorKeyResign runs it at
+// start and every doorKeyResignEvery, well inside apppolicy.Lifetime. The
+// pass at start is also the restart recovery: it finishes any publication a
+// crash or a lost write interrupted.
+//
+// The content and the revision both come from the table, never from the
+// bucket (Task 9; the door keys' database-as-authority rule, Astra plan
+// R1-1): the door records the signed bytes before it publishes them, so the
+// bucket may hold the recorded revision, an interrupted publication's
+// predecessor, a replayed older record or bytes that do not verify, and the
+// pass publishes the recorded one over any of them. A higher revision the
+// bucket holds is adopted first (policyBase's rule), which is safe because
+// only a console with the release key can have signed it.
+//
+// Only present objects are re-signed. An absent one is a purge or a
+// create-time clear, never revived: the row is marked a confirmed absence
+// and leaves the candidates until an operator writes again. And the write is
+// replace-only on the ETag just read, so a purge between the read and the
+// write is never undone.
+//
+// The candidates are the table's live and pending rows, read on every pass,
+// so a publication whose sync mark failed is still re-signed (Astra plan
+// R4-2). The bucket's prefix is listed only on this process's first pass in
+// a holding, and on every pass while it holds none (another console may hold
+// the lease and have written apps this table does not know), so a steady
+// state costs one Get per live policy per pass and no listing (spec §3a).
+//
+// Nothing is signed without the writer lease (ruling 10): with candidates
+// the pass takes one holding, and every app's signing and publication is
+// bound to it through policyAttempt, so a holding lost part way through
+// stops the pass and nothing more is read or written (Astra plan R3-1).
+func (s *Server) resignAppPolicies(ctx context.Context) {
+ if len(s.cfg.Key) == 0 || s.cfg.Releases == nil {
+ return
+ }
+ cands, err := s.policyResignCandidates(ctx)
+ if err != nil {
+ s.policyLogf("console: app policies: re-sign: %v", err)
+ return
+ }
+ if len(cands) == 0 {
+ return
+ }
+ hold, ref := s.policyWriter(ctx)
+ if ref != nil {
+ s.policyLogf("console: app policies: not re-signing; %s", ref.Msg)
+ return
+ }
+ for _, c := range cands {
+ err := s.resignAppPolicy(ctx, c.acct, c.app, hold)
+ if errors.Is(err, errPolicyResignLost) {
+ s.policyLogf("console: app policies: stopped re-signing at %s/%s; the writer lease was lost", c.acct.Sqid, c.app)
+ return
+ }
+ if err != nil {
+ s.policyLogf("console: app policy %s/%s: re-sign: %v", c.acct.Sqid, c.app, err)
+ }
+ }
+}
+
+// policyResignCandidates is the pass's apps, sorted: every live or pending
+// row, and, on a pass that lists, every listed key that has no row.
+func (s *Server) policyResignCandidates(ctx context.Context) ([]policyCandidate, error) {
+ rows, err := s.store.AppPolicyRowsContext(ctx)
+ if err != nil {
+ return nil, err
+ }
+ seen := map[string]bool{}
+ var out []policyCandidate
+ for _, r := range rows {
+ seen[r.Sqid+"/"+r.App] = true
+ out = append(out, policyCandidate{&Account{ID: r.AccountID, Sqid: r.Sqid}, r.App})
+ }
+
+ n := s.policyReadHolding().n
+ s.policyStateMu.Lock()
+ listed := n != 0 && s.policyListedN == n
+ s.policyStateMu.Unlock()
+ if !listed {
+ present, err := apppolicy.Present(ctx, s.cfg.Releases)
+ if err != nil {
+ s.policyLogf("console: app policies: %v", err)
+ } else {
+ s.policyStateMu.Lock()
+ s.policyListedN = n
+ s.policyStateMu.Unlock()
+ for key := range present {
+ sqid, app, ok := policyKeyApp(key)
+ if !ok {
+ s.policyLogf("console: app policies: skipping %s: not an app policy key", key)
+ continue
+ }
+ if seen[sqid+"/"+app] {
+ continue
+ }
+ seen[sqid+"/"+app] = true
+ acct, err := s.store.AccountBySqid(sqid)
+ if err != nil {
+ s.policyLogf("console: app policy %s/%s: account: %v", sqid, app, err)
+ continue
+ }
+ if acct == nil {
+ s.policyLogf("console: app policy %s/%s: no such account; skipped (the purge sweep owns it)", sqid, app)
+ continue
+ }
+ out = append(out, policyCandidate{acct, app})
+ }
+ }
+ }
+ sort.Slice(out, func(i, j int) bool {
+ if out[i].acct.Sqid != out[j].acct.Sqid {
+ return out[i].acct.Sqid < out[j].acct.Sqid
+ }
+ return out[i].app < out[j].app
+ })
+ return out, nil
+}
+
+// policyKeyApp is the account and app a listed key is the policy of: ok only
+// when the key is exactly apppolicy.Key(sqid, app) for a valid account and
+// app.
+func policyKeyApp(key string) (sqid, app string, ok bool) {
+ rest, ok := strings.CutPrefix(key, apppolicy.Prefix)
+ if !ok {
+ return "", "", false
+ }
+ sqid, file, ok := strings.Cut(rest, "/")
+ if !ok {
+ return "", "", false
+ }
+ app, ok = strings.CutSuffix(file, ".json")
+ if !ok || !release.ValidAccount(sqid) || !release.ValidApp(app) || apppolicy.Key(sqid, app) != key {
+ return "", "", false
+ }
+ return sqid, app, true
+}
+
+// resignAppPolicy re-signs one app's policy under its policy lock and as its
+// app mutation, within doorKeyPublishBudget, bound to hold. A lost holding is
+// errPolicyResignLost; any other failure is this app's alone.
+func (s *Server) resignAppPolicy(ctx context.Context, acct *Account, app string, hold policyHolding) error {
+ mu := s.policyLock(acct.Sqid, app)
+ mu.Lock()
+ defer mu.Unlock()
+ // Nothing of a later app is read once the holding is gone.
+ _, done, ref := s.policyAttempt(ctx, hold)
+ done()
+ if ref != nil {
+ return errPolicyResignLost
+ }
+ ctx, cancel := context.WithTimeout(ctx, doorKeyPublishBudget)
+ defer cancel()
+ var (
+ res error
+ ran bool
+ )
+ err := s.appMutation(ctx, acct, app, policyResignAction, func(ctx context.Context) error {
+ ran = true
+ res = s.resignAppPolicyLocked(ctx, acct, app, hold)
+ return res
+ })
+ if ran && res != nil {
+ return res
+ }
+ return err
+}
+
+// resignAppPolicyLocked is the re-sign itself: the base read, then the
+// recorded content signed again at its recorded revision, recorded, and
+// published replace-only on the ETag just read.
+func (s *Server) resignAppPolicyLocked(ctx context.Context, acct *Account, app string, hold policyHolding) error {
+ r, err := s.readPolicyBase(ctx, acct, app, hold, true)
+ if s.policyAfterBase != nil {
+ s.policyAfterBase(acct.Sqid, app)
+ }
+ if err != nil {
+ return err
+ }
+ row := r.row
+ if r.etag == "" {
+ // Absent: a purge or a create-time clear. Never recreated; the row
+ // is a confirmed absence until an operator writes again.
+ if row.Revision == 0 {
+ return nil
+ }
+ if err := s.store.SyncedAppPolicyContext(ctx, acct.ID, app, row.Revision, ""); err != nil {
+ return fmt.Errorf("mark revision %d absent: %w", row.Revision, err)
+ }
+ return nil
+ }
+ if r.base.Revision == 0 {
+ // The base is empty: the object does not verify and there is no row
+ // this holding may build on, or (under claims) the stored policy is
+ // another incarnation's, which policyBase has logged. The console
+ // never signs bytes it neither recorded nor verified.
+ if !s.policyOtherIncarnation(acct, app, row, r.cur) {
+ s.policyLogf("console: app policy %s/%s: not re-signing an object that does not verify; an operator write repairs it",
+ acct.Sqid, app)
+ }
+ return nil
+ }
+ p, err := apppolicy.OpenKey(s.policyPub(), row.Wire, acct.Sqid, app)
+ if err != nil {
+ return fmt.Errorf("open recorded revision %d: %w", row.Revision, err)
+ }
+ if !bytes.Equal(r.wire, row.Wire) {
+ var held uint64
+ if r.bucket != nil {
+ held = r.bucket.Revision
+ }
+ s.policyLogf("console: app policy %s/%s: the bucket held revision %d, not the recorded %d; publishing the recorded one",
+ acct.Sqid, app, held, row.Revision)
+ }
+ rev := row.Revision
+ p.Revision, p.SignedAt = rev, time.Now().UTC()
+
+ _, done, ref := s.policyAttempt(ctx, hold)
+ if ref != nil {
+ return errPolicyResignLost
+ }
+ wire, err := apppolicy.Sign(s.cfg.Key, p)
+ done()
+ if err != nil {
+ return fmt.Errorf("sign revision %d: %w", rev, err)
+ }
+ // The record never lags the bucket.
+ if err := s.store.ResignAppPolicyContext(ctx, acct.ID, app, rev, wire); err != nil {
+ return fmt.Errorf("record re-signed revision %d: %w", rev, err)
+ }
+ actx, done, ref := s.policyAttempt(ctx, hold)
+ if ref != nil {
+ return errPolicyResignLost
+ }
+ // One attempt, replace-only: a lost compare-and-swap is never retried.
+ newETag, err := apppolicy.PutWire(actx, s.cfg.Releases, acct.Sqid, app, wire, r.etag)
+ ended := !hold.sole && actx.Err() != nil
+ done()
+ switch {
+ case ended:
+ // A write that outlived its holding is never reported published;
+ // the row stays pending.
+ s.policyLogf("console: app policy %s/%s: re-sign revision %d: the writer holding ended during the write", acct.Sqid, app, rev)
+ return errPolicyResignLost
+ case errors.Is(err, release.ErrCASConflict) || errors.Is(err, os.ErrNotExist):
+ s.policyLogf("console: app policy %s/%s: the object changed since it was read; not re-signing it in this pass", acct.Sqid, app)
+ return nil
+ case err != nil:
+ return fmt.Errorf("publish re-signed revision %d: %w", rev, err)
+ }
+ if err := s.store.SyncedAppPolicyContext(ctx, acct.ID, app, rev, newETag); err != nil {
+ return fmt.Errorf("mark re-signed revision %d published: %w", rev, err)
+ }
+ return nil
+}
+
+// policyOtherIncarnation reports whether, under claims, the recorded policy
+// belongs to another incarnation of the app than cur.
+func (s *Server) policyOtherIncarnation(acct *Account, app string, row AppPolicyRow, cur string) bool {
+ if !s.cfg.HostClaim || row.Revision == 0 {
+ return false
+ }
+ p, err := apppolicy.OpenKey(s.policyPub(), row.Wire, acct.Sqid, app)
+ return err == nil && p.AppID != cur
+}
+
// applyPolicyChange applies the request to a copy of base (Task 9 step 8):
// the normalized allow, the default size, one host's override captured with
// its instance record's CreatedAt, and every other override whose instance is
diff --git a/internal/console/apipolicy_resign_test.go b/internal/console/apipolicy_resign_test.go
new file mode 100644
index 000000000..262a76218
--- /dev/null
+++ b/internal/console/apipolicy_resign_test.go
@@ -0,0 +1,534 @@
+package console
+
+import (
+ "bytes"
+ "context"
+ "errors"
+ "os"
+ "reflect"
+ "slices"
+ "strings"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/apppolicy"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// The re-sign pass (Task 10): resignAppPolicies on the door keys' loop.
+
+// recordPolicyAt signs p, writes it at its key with PutIfMatch and records it
+// as this console's published revision. It answers the bytes and the ETag.
+func recordPolicyAt(t *testing.T, fx *policyFx, p apppolicy.Policy) ([]byte, string) {
+ t.Helper()
+ ctx := context.Background()
+ _, cur, err := fx.rel.Get(ctx, apppolicy.Key(fx.sqid, p.App))
+ if err != nil && !errors.Is(err, os.ErrNotExist) {
+ t.Fatal(err)
+ }
+ etag, err := apppolicy.PutIfMatchETag(ctx, fx.rel, fx.priv, p, cur)
+ if err != nil {
+ t.Fatal(err)
+ }
+ wire := signPolicy(t, fx.consoleFixture, p)
+ if err := fx.st.RecordAppPolicyContext(ctx, OpsAccountID, p.App, p.Revision, wire); err != nil {
+ t.Fatal(err)
+ }
+ if err := fx.st.SyncedAppPolicyContext(ctx, OpsAccountID, p.App, p.Revision, etag); err != nil {
+ t.Fatal(err)
+ }
+ return wire, etag
+}
+
+// bytesAt is the raw object at the app's key.
+func (fx *policyFx) bytesAt(t *testing.T, app string) []byte {
+ t.Helper()
+ b, _, err := fx.rel.Get(context.Background(), apppolicy.Key(fx.sqid, app))
+ if err != nil {
+ t.Fatalf("object of %s: %v", app, err)
+ }
+ return b
+}
+
+func (fx *policyFx) addApps(t *testing.T, apps ...string) {
+ t.Helper()
+ for _, app := range apps {
+ if _, err := fx.st.CreateAppIn(OpsAccountID, app, "paul@keymail.dev", time.Now()); err != nil {
+ t.Fatal(err)
+ }
+ }
+}
+
+// policyCountingStore counts the store requests under the policy prefix:
+// listings, ETag listings, and Gets and Puts per key. afterPut, when set,
+// runs after each policy Put returns.
+type policyCountingStore struct {
+ release.Store
+ mu sync.Mutex
+ lists int
+ listETags int
+ gets map[string]int
+ puts map[string]int
+ afterPut func(key string)
+}
+
+func newPolicyCountingStore(s release.Store) *policyCountingStore {
+ return &policyCountingStore{Store: s, gets: map[string]int{}, puts: map[string]int{}}
+}
+
+func (c *policyCountingStore) List(ctx context.Context, prefix string) ([]string, error) {
+ if strings.HasPrefix(prefix, apppolicy.Prefix) {
+ c.mu.Lock()
+ c.lists++
+ c.mu.Unlock()
+ }
+ return c.Store.List(ctx, prefix)
+}
+
+func (c *policyCountingStore) ListETags(ctx context.Context, prefix string) ([]release.KeyETag, error) {
+ if strings.HasPrefix(prefix, apppolicy.Prefix) {
+ c.mu.Lock()
+ c.listETags++
+ c.mu.Unlock()
+ }
+ l, ok := c.Store.(release.ETagLister)
+ if !ok {
+ return nil, errors.ErrUnsupported
+ }
+ return l.ListETags(ctx, prefix)
+}
+
+func (c *policyCountingStore) Get(ctx context.Context, key string) ([]byte, string, error) {
+ if strings.HasPrefix(key, apppolicy.Prefix) {
+ c.mu.Lock()
+ c.gets[key]++
+ c.mu.Unlock()
+ }
+ return c.Store.Get(ctx, key)
+}
+
+func (c *policyCountingStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
+ etag, err := c.Store.Put(ctx, key, body, ifMatch)
+ if strings.HasPrefix(key, apppolicy.Prefix) {
+ c.mu.Lock()
+ c.puts[key]++
+ hook := c.afterPut
+ c.mu.Unlock()
+ if hook != nil {
+ hook(key)
+ }
+ }
+ return etag, err
+}
+
+// counts answers the listings and the Gets and Puts summed over every key.
+func (c *policyCountingStore) counts() (lists, listETags, gets, puts int) {
+ c.mu.Lock()
+ defer c.mu.Unlock()
+ for _, n := range c.gets {
+ gets += n
+ }
+ for _, n := range c.puts {
+ puts += n
+ }
+ return c.lists, c.listETags, gets, puts
+}
+
+func (c *policyCountingStore) keyCounts(key string) (gets, puts int) {
+ c.mu.Lock()
+ defer c.mu.Unlock()
+ return c.gets[key], c.puts[key]
+}
+
+func (c *policyCountingStore) reset() {
+ c.mu.Lock()
+ defer c.mu.Unlock()
+ c.lists, c.listETags = 0, 0
+ c.gets, c.puts = map[string]int{}, map[string]int{}
+}
+
+// waitForPolicy polls cond until it holds or ten seconds pass.
+func waitForPolicy(t *testing.T, what string, cond func() bool) {
+ t.Helper()
+ deadline := time.Now().Add(10 * time.Second)
+ for !cond() {
+ if time.Now().After(deadline) {
+ t.Fatalf("timed out waiting for %s", what)
+ }
+ time.Sleep(10 * time.Millisecond)
+ }
+}
+
+// runDoorKeyResignOnce runs StartDoorKeyResign until done reports its first
+// pass finished, then cancels it and waits for it to return.
+func runDoorKeyResignOnce(t *testing.T, srv *Server, done func() bool) {
+ t.Helper()
+ ctx, cancel := context.WithCancel(context.Background())
+ returned := make(chan struct{})
+ go func() {
+ defer close(returned)
+ srv.StartDoorKeyResign(ctx)
+ }()
+ waitForPolicy(t, "the first re-sign pass", done)
+ cancel()
+ select {
+ case <-returned:
+ case <-time.After(10 * time.Second):
+ t.Fatal("StartDoorKeyResign did not return after its context was cancelled")
+ }
+}
+
+func TestResignRefreshesSignedAtAndKeepsTheRevision(t *testing.T) {
+ t.Parallel()
+ fx := newPolicyFx(t, "code-resign-refresh")
+ p := apppolicy.Policy{V: 1, Account: fx.sqid, App: "shop", Revision: 3,
+ SignedAt: time.Now().UTC().Add(-50 * time.Hour).Truncate(time.Second), EgressAllow: []string{"10.0.5.20"}, DefaultSize: "small"}
+ recordPolicyAt(t, fx, p)
+
+ before := time.Now()
+ fx.srv.resignAppPolicies(context.Background())
+ after := time.Now()
+
+ wire := fx.bytesAt(t, "shop")
+ got, err := apppolicy.VerifyKey(fx.pub, wire, time.Now().Add(71*time.Hour), fx.sqid, "shop")
+ if err != nil {
+ t.Fatalf("the re-signed object at now+71h: %v", err)
+ }
+ if got.SignedAt.Before(before) || got.SignedAt.After(after) {
+ t.Fatalf("signed_at %s, want within [%s, %s]", got.SignedAt, before, after)
+ }
+ want := p
+ want.SignedAt = got.SignedAt
+ if !reflect.DeepEqual(got, want) {
+ t.Fatalf("re-signed %+v, want %+v", got, want)
+ }
+ _, _, etag := fx.stored(t, "shop")
+ if row := fx.row(t, "shop"); row.Revision != 3 || !bytes.Equal(row.Wire, wire) || row.SyncedETag != etag || row.Pending {
+ t.Fatalf("row %+v, want revision 3 with the stored bytes, synced", row)
+ }
+}
+
+func TestResignRevivesAnExpiredRecord(t *testing.T) {
+ t.Parallel()
+ fx := newPolicyFx(t, "code-resign-expired")
+ p := apppolicy.Policy{V: 1, Account: fx.sqid, App: "shop", Revision: 2,
+ SignedAt: time.Now().UTC().Add(-80 * time.Hour), EgressAllow: []string{"10.0.5.20"}}
+ wire, _ := recordPolicyAt(t, fx, p)
+ if _, err := apppolicy.Verify(fx.pub, wire, time.Now()); !errors.Is(err, apppolicy.ErrExpired) {
+ t.Fatalf("the record is not expired: %v", err)
+ }
+ fx.srv.resignAppPolicies(context.Background())
+ got, err := apppolicy.VerifyKey(fx.pub, fx.bytesAt(t, "shop"), time.Now(), fx.sqid, "shop")
+ if err != nil || got.Revision != 2 || !slices.Equal(got.EgressAllow, []string{"10.0.5.20"}) {
+ t.Fatalf("after the pass %+v %v", got, err)
+ }
+}
+
+func TestResignPublishesOnlyTheRecordedContent(t *testing.T) {
+ t.Parallel()
+ fx := newPolicyFx(t, "code-resign-recorded")
+ fx.addApps(t, "shop-c", "shop-d", "shop-e")
+ now := time.Now().UTC()
+ pol := func(app string, rev uint64, allow []string, size string) apppolicy.Policy {
+ return apppolicy.Policy{V: 1, Account: fx.sqid, App: app, Revision: rev, SignedAt: now, EgressAllow: allow, DefaultSize: size}
+ }
+ granting := []string{"10.0.5.20"}
+
+ // a: unverifiable bytes, nothing recorded.
+ putAtKey(t, fx.rel, fx.sqid, "shop", []byte(`{`))
+ // b: revision 4 recorded with no allow, a replayed revision 1 granting at the key.
+ recordPolicyAt(t, fx, pol("shop-b", 4, nil, "small"))
+ putAtKey(t, fx.rel, fx.sqid, "shop-b", signPolicy(t, fx.consoleFixture, pol("shop-b", 1, granting, "")))
+ // c: revision 2 recorded, unverifiable bytes at the key.
+ recordPolicyAt(t, fx, pol("shop-c", 2, nil, "large"))
+ putAtKey(t, fx.rel, fx.sqid, "shop-c", []byte(`{`))
+ // d: revision 2 recorded, a revision 6 record at the key.
+ recordPolicyAt(t, fx, pol("shop-d", 2, granting, ""))
+ putAtKey(t, fx.rel, fx.sqid, "shop-d", signPolicy(t, fx.consoleFixture, pol("shop-d", 6, nil, "xlarge")))
+ // e: revision 5 recorded with an empty allow, another revision 5 granting at the key.
+ recordPolicyAt(t, fx, pol("shop-e", 5, []string{}, "small"))
+ putAtKey(t, fx.rel, fx.sqid, "shop-e", signPolicy(t, fx.consoleFixture, pol("shop-e", 5, granting, "small")))
+
+ fx.srv.resignAppPolicies(context.Background())
+
+ if got := fx.bytesAt(t, "shop"); string(got) != `{` {
+ t.Fatalf("a: the unverifiable object was rewritten: %q", got)
+ }
+ fx.logs.has(t, "console: app policy bab/shop: not re-signing an object that does not verify; an operator write repairs it")
+ check := func(app string, rev uint64, allow []string, size string) {
+ t.Helper()
+ wire := fx.bytesAt(t, app)
+ got, err := apppolicy.VerifyKey(fx.pub, wire, time.Now(), fx.sqid, app)
+ if err != nil {
+ t.Fatalf("%s: %v", app, err)
+ }
+ if got.Revision != rev || len(got.EgressAllow) != len(allow) || !slices.Equal(got.EgressAllow, allow) || got.DefaultSize != size {
+ t.Fatalf("%s: stored %+v, want revision %d allow %v size %q", app, got, rev, allow, size)
+ }
+ row := fx.row(t, app)
+ if row.Revision != rev || !bytes.Equal(row.Wire, wire) || row.Pending {
+ t.Fatalf("%s: row %+v, want revision %d with the stored bytes", app, row, rev)
+ }
+ }
+ check("shop-b", 4, nil, "small")
+ fx.logs.has(t, "console: app policy bab/shop-b: the bucket held revision 1, not the recorded 4; publishing the recorded one")
+ check("shop-c", 2, nil, "large")
+ fx.logs.has(t, "console: app policy bab/shop-c: the bucket held revision 0, not the recorded 2; publishing the recorded one")
+ check("shop-d", 6, nil, "xlarge")
+ check("shop-e", 5, nil, "small")
+ fx.logs.has(t, "console: app policy bab/shop-e: the bucket held revision 5, not the recorded 5; publishing the recorded one")
+}
+
+func TestResignFinishesAnInterruptedPublication(t *testing.T) {
+ t.Parallel()
+ fx := newPolicyFx(t, "code-resign-interrupted")
+ allow := apppolicy.Policy{V: 1, Account: fx.sqid, App: "shop", Revision: 1, SignedAt: time.Now().UTC(), EgressAllow: []string{"10.0.5.20"}}
+ recordPolicyAt(t, fx, allow)
+ // A crash between RecordAppPolicyContext and the write.
+ revoke := allow
+ revoke.Revision, revoke.EgressAllow = 2, nil
+ if err := fx.st.RecordAppPolicyContext(context.Background(), OpsAccountID, "shop", 2, signPolicy(t, fx.consoleFixture, revoke)); err != nil {
+ t.Fatal(err)
+ }
+
+ runDoorKeyResignOnce(t, fx.srv, func() bool { return !fx.row(t, "shop").Pending })
+
+ stored, wire, _ := fx.stored(t, "shop")
+ if stored.Revision != 2 || len(stored.EgressAllow) != 0 {
+ t.Fatalf("stored %+v, want the recorded revocation at revision 2", stored)
+ }
+ if row := fx.row(t, "shop"); !bytes.Equal(row.Wire, wire) {
+ t.Fatalf("row %+v does not hold the stored bytes", row)
+ }
+ fx.logs.has(t, "console: app policy bab/shop: the bucket held revision 1, not the recorded 2; publishing the recorded one")
+}
+
+func TestResignNeverRecreatesAPurgedPolicy(t *testing.T) {
+ t.Parallel()
+ fx := newPolicyFx(t, "code-resign-purged")
+ recordPolicyAt(t, fx, apppolicy.Policy{V: 1, Account: fx.sqid, App: "shop", Revision: 1,
+ SignedAt: time.Now().UTC(), EgressAllow: []string{"10.0.5.20"}})
+ key := apppolicy.Key(fx.sqid, "shop")
+ hs := &appPolicyHookStore{Store: fx.rel, key: key}
+ hs.hook = func(ctx context.Context, n int, body []byte, ifMatch string) (bool, string, error) {
+ // The purge, between the pass's read and its write.
+ if err := fx.rel.Delete(ctx, key); err != nil {
+ t.Error(err)
+ }
+ return false, "", nil
+ }
+ fx.srv.cfg.Releases = hs
+
+ fx.srv.resignAppPolicies(context.Background())
+
+ if hs.putCount() != 1 {
+ t.Fatalf("%d Puts of the key, want 1", hs.putCount())
+ }
+ fx.noObject(t, "shop")
+ fx.logs.has(t, "console: app policy bab/shop: the object changed since it was read; not re-signing it in this pass")
+}
+
+func TestStartDoorKeyResignAlsoResignsPolicies(t *testing.T) {
+ t.Parallel()
+ t.Run("with a signing key", func(t *testing.T) {
+ t.Parallel()
+ fx := newPolicyFx(t, "code-resign-loop")
+ signed := time.Now().UTC().Add(-50 * time.Hour).Truncate(time.Second)
+ recordPolicyAt(t, fx, apppolicy.Policy{V: 1, Account: fx.sqid, App: "shop", Revision: 1, SignedAt: signed, DefaultSize: "small"})
+ runDoorKeyResignOnce(t, fx.srv, func() bool {
+ p, _, _, err := apppolicy.GetForUpdate(context.Background(), fx.rel, fx.pub, fx.sqid, "shop")
+ return err == nil && p.SignedAt.After(signed)
+ })
+ })
+ t.Run("without a signing key", func(t *testing.T) {
+ t.Parallel()
+ fx := newPolicyFxFrom(t, newConsoleFixtureKeyed(t, false), "code-resign-loop-nokey")
+ wire := signPolicy(t, fx.consoleFixture, apppolicy.Policy{V: 1, Account: fx.sqid, App: "shop", Revision: 1,
+ SignedAt: time.Now().UTC().Add(-50 * time.Hour), DefaultSize: "small"})
+ putAtKey(t, fx.rel, fx.sqid, "shop", wire)
+ returned := make(chan struct{})
+ go func() {
+ defer close(returned)
+ fx.srv.StartDoorKeyResign(context.Background())
+ }()
+ select {
+ case <-returned:
+ case <-time.After(10 * time.Second):
+ t.Fatal("StartDoorKeyResign without a signing key did not return")
+ }
+ if got := fx.bytesAt(t, "shop"); !bytes.Equal(got, wire) {
+ t.Fatal("the object changed on a console with no signing key")
+ }
+ })
+}
+
+// Astra plan R3-1.
+func TestResignStopsWhenTheHoldingIsLost(t *testing.T) {
+ t.Parallel()
+ fx := newPolicyFx(t, "code-resign-lost")
+ f := fx.leased()
+ fx.addApps(t, "shop-c")
+ apps := []string{"shop", "shop-b", "shop-c"}
+ before := map[string][]byte{}
+ for _, app := range apps {
+ recordPolicyAt(t, fx, apppolicy.Policy{V: 1, Account: fx.sqid, App: app, Revision: 1,
+ SignedAt: time.Now().UTC().Add(-50 * time.Hour), DefaultSize: "small"})
+ before[app] = fx.bytesAt(t, app)
+ }
+ cs := newPolicyCountingStore(fx.rel)
+ var lostAt struct{ gets, puts int }
+ first := true
+ cs.afterPut = func(string) {
+ if !first {
+ return
+ }
+ first = false
+ lastPolicyLease(t, f).set(errors.New("renewal lost"), time.Now().Add(10*time.Minute), nil)
+ _, _, lostAt.gets, lostAt.puts = cs.counts()
+ }
+ fx.srv.cfg.Releases = cs
+
+ fx.srv.resignAppPolicies(context.Background())
+
+ resigned := 0
+ for _, app := range apps {
+ if !bytes.Equal(fx.bytesAt(t, app), before[app]) {
+ resigned++
+ }
+ }
+ if resigned != 1 {
+ t.Fatalf("%d policies re-signed, want 1", resigned)
+ }
+ for _, app := range apps[1:] {
+ if !bytes.Equal(fx.bytesAt(t, app), before[app]) {
+ t.Fatalf("%s was re-signed after the holding was lost", app)
+ }
+ }
+ if _, _, gets, puts := cs.counts(); gets != lostAt.gets || puts != lostAt.puts {
+ t.Fatalf("after the holding was lost: %d Gets and %d Puts of the prefix, want %d and %d", gets, puts, lostAt.gets, lostAt.puts)
+ }
+ fx.logs.has(t, "console: app policies: stopped re-signing at bab/shop-b; the writer lease was lost")
+}
+
+// Astra plan R4-2. Fails if pending rows are left out of the candidates.
+func TestResignPicksUpAFirstPublicationWhoseSyncMarkFailed(t *testing.T) {
+ t.Parallel()
+ fx := newPolicyFx(t, "code-resign-pending")
+ ctx := context.Background()
+ fx.mustPut(t, "shop", `{"default_size":"small"}`, policyCreate)
+ fx.srv.resignAppPolicies(ctx) // this holding's listing has run
+ cs := newPolicyCountingStore(fx.rel)
+ fx.srv.cfg.Releases = cs
+
+ if _, err := fx.st.db.Exec(`CREATE TRIGGER test_sync_mark_fails BEFORE UPDATE OF synced_etag ON app_policy_revisions
+WHEN NEW.synced_etag != '' BEGIN SELECT RAISE(ABORT, 'test: the sync mark fails'); END`); err != nil {
+ t.Fatal(err)
+ }
+ fx.mustPut(t, "shop-b", `{"egress_allow":["10.0.5.20"]}`, policyCreate)
+ first, _, _ := fx.stored(t, "shop-b")
+ if first.Revision != 1 {
+ t.Fatalf("stored %+v, want revision 1", first)
+ }
+ row := fx.row(t, "shop-b")
+ if row.SyncedETag != "" || !row.Pending {
+ t.Fatalf("row %+v, want pending with no sync mark", row)
+ }
+ rows, err := fx.st.AppPolicyRowsContext(ctx)
+ if err != nil || !slices.ContainsFunc(rows, func(r AppPolicyRow) bool { return r.App == "shop-b" }) {
+ t.Fatalf("AppPolicyRowsContext %+v %v, want shop-b listed", rows, err)
+ }
+
+ fx.srv.resignAppPolicies(ctx)
+ second, _, _ := fx.stored(t, "shop-b")
+ if !second.SignedAt.After(first.SignedAt) {
+ t.Fatalf("signed_at %s did not move past %s", second.SignedAt, first.SignedAt)
+ }
+ if row := fx.row(t, "shop-b"); !row.Pending {
+ t.Fatalf("row %+v, want still pending (its mark failed again)", row)
+ }
+
+ if _, err := fx.st.db.Exec(`DROP TRIGGER test_sync_mark_fails`); err != nil {
+ t.Fatal(err)
+ }
+ fx.srv.resignAppPolicies(ctx)
+ third, _, etag := fx.stored(t, "shop-b")
+ if !third.SignedAt.After(second.SignedAt) || third.Revision != 1 {
+ t.Fatalf("stored %+v, want revision 1 signed after %s", third, second.SignedAt)
+ }
+ if row := fx.row(t, "shop-b"); row.Pending || row.SyncedETag != etag {
+ t.Fatalf("row %+v, want synced at %q", row, etag)
+ }
+ if lists, listETags, _, _ := cs.counts(); lists != 0 || listETags != 0 {
+ t.Fatalf("%d listings and %d ETag listings of the prefix, want none", lists, listETags)
+ }
+
+ key := apppolicy.Key(fx.sqid, "shop-b")
+ if err := fx.rel.Delete(ctx, key); err != nil {
+ t.Fatal(err)
+ }
+ cs.reset()
+ fx.srv.resignAppPolicies(ctx)
+ if _, puts := cs.keyCounts(key); puts != 0 {
+ t.Fatalf("%d Puts of the deleted key", puts)
+ }
+ fx.noObject(t, "shop-b")
+ if row := fx.row(t, "shop-b"); row.Pending || row.SyncedETag != "" {
+ t.Fatalf("row %+v, want a confirmed absence", row)
+ }
+ cs.reset()
+ fx.srv.resignAppPolicies(ctx)
+ if gets, _ := cs.keyCounts(key); gets != 0 {
+ t.Fatalf("%d Gets of the absent policy's key, want 0", gets)
+ }
+}
+
+func TestResignWithNoPoliciesTakesNoLease(t *testing.T) {
+ t.Parallel()
+ fx := newPolicyFx(t, "code-resign-none")
+ f := fx.leased()
+ cs := newPolicyCountingStore(fx.rel)
+ fx.srv.cfg.Releases = cs
+ fx.srv.resignAppPolicies(context.Background())
+ if lists, _, _, _ := cs.counts(); lists != 1 {
+ t.Fatalf("%d listings, want 1", lists)
+ }
+ if n := f.callCount(); n != 0 {
+ t.Fatalf("%d leaser calls, want 0", n)
+ }
+}
+
+// The §3a console budget for the re-sign, pinned (ruling 11). Fails on any
+// increase.
+func TestPolicyResignKeepsTheRequestBudget(t *testing.T) {
+ t.Parallel()
+ fx := newPolicyFx(t, "code-resign-budget")
+ f := fx.leased()
+ fx.addApps(t, "shop-c")
+ cs := newPolicyCountingStore(fx.rel)
+ fx.srv.cfg.Releases = cs
+ for _, app := range []string{"shop", "shop-b", "shop-c"} {
+ fx.mustPut(t, app, `{"egress_allow":["10.0.5.20"]}`, policyCreate)
+ }
+ ctx := context.Background()
+ fx.srv.resignAppPolicies(ctx) // the start pass
+ cs.reset()
+
+ // One day: two passes.
+ fx.srv.resignAppPolicies(ctx)
+ fx.srv.resignAppPolicies(ctx)
+ if lists, listETags, gets, _ := cs.counts(); lists != 0 || listETags != 0 || gets != 6 {
+ t.Fatalf("a day of re-signing: %d List, %d ListETags, %d Get under the prefix; want 0, 0, 6", lists, listETags, gets)
+ }
+
+ // A new holding lists once.
+ lastPolicyLease(t, f).set(errors.New("renewal lost"), time.Now().Add(10*time.Minute), nil)
+ cs.reset()
+ fx.srv.resignAppPolicies(ctx)
+ if lists, listETags, _, _ := cs.counts(); lists != 1 || listETags != 0 {
+ t.Fatalf("the first pass of a new holding: %d List, %d ListETags; want 1, 0", lists, listETags)
+ }
+ if n := f.callCount(); n != 2 {
+ t.Fatalf("%d leaser calls, want 2", n)
+ }
+}
diff --git a/internal/console/apipolicy_test.go b/internal/console/apipolicy_test.go
index df8725367..906a39933 100644
--- a/internal/console/apipolicy_test.go
+++ b/internal/console/apipolicy_test.go
@@ -1089,9 +1089,9 @@ func TestPolicyRefusedTakeoverKeepsTheEmptyBase(t *testing.T) {
}
}
-// Astra plan R1-1, the door's half (review fix round 1, finding 3). Task 10
-// adds the re-sign continuation: a pass between the replay and the GET
-// finishes the interrupted publication.
+// Astra plan R1-1 (review fix round 1, finding 3), the door's half and the
+// re-sign's: a pass after the replay finishes the interrupted publication
+// with the recorded revocation.
func TestPolicyRevokeSurvivesALostWriteAndAReplay(t *testing.T) {
t.Parallel()
fx := newPolicyFx(t, "code-policy-lost-write")
@@ -1124,6 +1124,20 @@ func TestPolicyRevokeSurvivesALostWriteAndAReplay(t *testing.T) {
if len(g.Policy.EgressAllow) != 0 || g.Policy.Revision != 2 || g.ETag != etag {
t.Fatalf("GET over the replayed allow = %+v etag %q, want the recorded revocation", g.Policy, g.ETag)
}
+
+ // The re-sign pass publishes the recorded revocation, never the replayed
+ // allow.
+ fx.srv.resignAppPolicies(context.Background())
+ stored, wire, etag := fx.stored(t, "shop")
+ if stored.Revision != 2 || len(stored.EgressAllow) != 0 {
+ t.Fatalf("stored after the re-sign %+v, want the recorded revocation at revision 2", stored)
+ }
+ if row := fx.row(t, "shop"); !bytes.Equal(row.Wire, wire) || row.SyncedETag != etag || row.Pending {
+ t.Fatalf("row after the re-sign %+v, want the stored bytes, synced", row)
+ }
+ if g := fx.get(t, "shop"); len(g.Policy.EgressAllow) != 0 || g.Policy.Revision != 2 || g.ETag != etag {
+ t.Fatalf("GET after the re-sign = %+v etag %q", g.Policy, g.ETag)
+ }
fx.mustPut(t, "shop", `{"default_size":"small"}`, policyIfMatch(etag))
if stored, _, _ := fx.stored(t, "shop"); stored.Revision != 3 || len(stored.EgressAllow) != 0 || stored.DefaultSize != "small" {
t.Fatalf("stored %+v, want revision 3 with no egress allow", stored)
diff --git a/internal/console/claim_invariant_fixtures_test.go b/internal/console/claim_invariant_fixtures_test.go
index 1c233d65a..315672dd8 100644
--- a/internal/console/claim_invariant_fixtures_test.go
+++ b/internal/console/claim_invariant_fixtures_test.go
@@ -1983,6 +1983,10 @@ var claimFixtures = map[string]claimFixture{
apps: claimRefs("bab/titogo"),
setup: func(t *testing.T, cx *claimCase) {
cx.mustAPI(t, http.MethodPost, "/api/cli/apps/bab/titogo/door-keys", `{"label":"router"}`, http.StatusCreated)
+ // The pass also re-signs app policies (apipolicy.go): one staged
+ // through its door, with Task 8's fake writer lease set on the
+ // server's own field, so the case stays parallel.
+ claimWorldPolicy(t, cx)
},
do: claimWorker("StartDoorKeyResign"),
}},
diff --git a/internal/console/policylease_twoconsoles_test.go b/internal/console/policylease_twoconsoles_test.go
new file mode 100644
index 000000000..7569cf8da
--- /dev/null
+++ b/internal/console/policylease_twoconsoles_test.go
@@ -0,0 +1,345 @@
+package console
+
+import (
+ "bytes"
+ "context"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "net/http"
+ "net/http/httptest"
+ "slices"
+ "sync/atomic"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/apppolicy"
+ "github.com/carlosframework/platform/internal/lease"
+ "github.com/carlosframework/platform/internal/lease/leasetest"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// Ruling 10's witnesses: two consoles, each with its own database, over one
+// deployment bucket, with the production leaser and the production
+// policyAttempt checks. Only the TTL is the test's (Astra plan R4-3): at 30 s
+// internal/lease's margin is lease.RenewMargin and it renews every 10 s, so a
+// live holding's SafeUntil is never less than 15 s ahead, above
+// policyAttemptWindow; a lapse is the next renewal failing (within 10 s) and
+// the lock's ExpiresAt passing (at most 30 s after the last renewal), so 45 s
+// covers it with RenewMargin and 10 s to spare.
+const (
+ policyWitnessTTL = 30 * time.Second
+ policyLapseDeadline = 45 * time.Second
+)
+
+const policyLeaseLockKey = appPolicyLeasePrefix + "/lock.json"
+
+// twoConsoles is the witnesses' world: one versioned bucket "deploy" behind
+// an S3 server, and consoles A and B over it with the same release key.
+type twoConsoles struct {
+ bucket *leasetest.VersionedBucket
+ failLock atomic.Bool
+ a, b *policyFx
+}
+
+func newTwoConsoles(t *testing.T) *twoConsoles {
+ t.Helper()
+ if policyWitnessTTL < 3*lease.RenewMargin {
+ t.Fatalf("policyWitnessTTL %s is below 3*lease.RenewMargin: the margin would be TTL/10", policyWitnessTTL)
+ }
+ if policyWitnessTTL-policyWitnessTTL/3-lease.RenewMargin < policyAttemptWindow {
+ t.Fatalf("policyWitnessTTL %s leaves a live holding less than policyAttemptWindow %s", policyWitnessTTL, policyAttemptWindow)
+ }
+ w := &twoConsoles{bucket: leasetest.NewVersionedBucket()}
+ w.bucket.AddBucket("deploy")
+ // Set once, before the server serves: the server reads it under the
+ // bucket's own lock, and the test flips failLock.
+ w.bucket.FailOn = func(op, key string) error {
+ if op == "put" && key == policyLeaseLockKey && w.failLock.Load() {
+ return errors.New("test: the lock key refuses puts")
+ }
+ return nil
+ }
+ handler, _ := leasetest.NewS3Server(w.bucket)
+ srv := httptest.NewServer(handler)
+ t.Cleanup(srv.Close)
+ for k, v := range map[string]string{"AWS_ACCESS_KEY_ID": "test-key", "AWS_SECRET_ACCESS_KEY": "test-secret",
+ "AWS_SESSION_TOKEN": "", "AWS_REGION": "eu-west-1", "AWS_EC2_METADATA_DISABLED": "true",
+ "AWS_CONFIG_FILE": "/nonexistent", "AWS_SHARED_CREDENTIALS_FILE": "/nonexistent", "CARLOS_S3_ENDPOINT": srv.URL} {
+ t.Setenv(k, v)
+ }
+ console := func() *consoleFixture {
+ rel, err := release.NewS3Store(context.Background(), "deploy")
+ if err != nil {
+ t.Fatal(err)
+ }
+ rel.EnableChangeIndex()
+ fx := newConsoleFixtureOn(t, true, rel)
+ fx.srv.cfg.Bucket = "deploy"
+ fx.srv.cfg.PolicyLease = ProductionPolicyLeaser
+ fx.srv.cfg.PolicyLeaseTTL = policyWitnessTTL
+ return fx
+ }
+ a, b := console(), console()
+ b.srv.cfg.Key, b.srv.cfg.ReleasePub, b.priv, b.pub = a.priv, a.pub, a.priv, a.pub
+ w.a, w.b = newPolicyFxFrom(t, a, "code-two-a"), newPolicyFxFrom(t, b, "code-two-b")
+ t.Cleanup(func() {
+ w.a.srv.ReleasePolicyWriter(context.Background())
+ w.b.srv.ReleasePolicyWriter(context.Background())
+ })
+ return w
+}
+
+// lock is the writer lease's lock object as the bucket holds it now.
+func (w *twoConsoles) lock(t *testing.T) (owner string, expires time.Time, ok bool) {
+ t.Helper()
+ body, ok := w.bucket.Current("deploy", policyLeaseLockKey)
+ if !ok {
+ return "", time.Time{}, false
+ }
+ var l struct {
+ Owner string `json:"owner"`
+ ExpiresAt time.Time `json:"expires_at"`
+ }
+ if err := json.Unmarshal(body, &l); err != nil {
+ t.Fatalf("lock %q: %v", body, err)
+ }
+ return l.Owner, l.ExpiresAt, true
+}
+
+func (w *twoConsoles) heldMsg(t *testing.T) string {
+ t.Helper()
+ owner, expires, ok := w.lock(t)
+ if !ok {
+ t.Fatal("no lock object")
+ }
+ return fmt.Sprintf(policyLeaseHeldMsg, owner, expires.UTC().Format(time.RFC3339))
+}
+
+// putHeld is a PUT that another console's lease refuses: 503 naming the
+// lock's owner and expiry (read on both sides of the request, since the
+// holder renews meanwhile).
+func (w *twoConsoles) putHeld(t *testing.T, fx *policyFx, body string, hdr map[string]string) {
+ t.Helper()
+ before := w.heldMsg(t)
+ st, _, out := fx.put(t, "shop", body, hdr)
+ after := w.heldMsg(t)
+ if st != http.StatusServiceUnavailable {
+ t.Fatalf("PUT %s = %d %s, want 503 naming the holder", body, st, out)
+ }
+ if got := errorOf(t, out); got != before && got != after {
+ t.Fatalf("PUT %s refused with %q, want %q", body, got, after)
+ }
+}
+
+// lapse makes fx's lease lapse: renewals of the lock fail until fx no
+// longer holds it and the lock's expiry has passed.
+func (w *twoConsoles) lapse(t *testing.T, fx *policyFx) {
+ t.Helper()
+ w.failLock.Store(true)
+ defer w.failLock.Store(false)
+ deadline := time.Now().Add(policyLapseDeadline)
+ for {
+ if !fx.srv.policyHeld() {
+ if _, expires, ok := w.lock(t); !ok || time.Now().After(expires) {
+ return
+ }
+ }
+ if time.Now().After(deadline) {
+ t.Fatalf("the lease did not lapse within %s", policyLapseDeadline)
+ }
+ time.Sleep(250 * time.Millisecond)
+ }
+}
+
+// restartPolicyWriter is a console process restarted over the same
+// database: ReleasePolicyWriter stops a writer for good, as a shutdown
+// does, and a new process starts with no holding marks and no listing.
+func restartPolicyWriter(s *Server) {
+ s.policyStateMu.Lock()
+ s.policyStopping = false
+ s.policyListedN = 0
+ s.policyStateMu.Unlock()
+ s.policySyncedMu.Lock()
+ s.policySynced = nil
+ s.policySyncedMu.Unlock()
+}
+
+func (fx *policyFx) bucketBytes(t *testing.T) []byte {
+ t.Helper()
+ b, _, err := fx.rel.Get(context.Background(), apppolicy.Key(fx.sqid, "shop"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ return b
+}
+
+// Ruling 10's witness. Not parallel: it sets the S3 endpoint and the AWS
+// variables.
+func TestPolicyWriterLeaseIsOneConsolePerBucket(t *testing.T) {
+ w := newTwoConsoles(t)
+ A, B := w.a, w.b
+
+ // 1. B writes, then stands down gracefully.
+ r := B.mustPut(t, "shop", `{"egress_allow":["10.0.5.20"]}`, policyCreate)
+ if r.Policy.Revision != 1 {
+ t.Fatalf("B's first write = %+v", r.Policy)
+ }
+ B.srv.ReleasePolicyWriter(context.Background())
+ restartPolicyWriter(B.srv)
+
+ // 2. A takes the free lease and builds on B's revision 1.
+ g := A.get(t, "shop")
+ if !slices.Equal(g.Policy.EgressAllow, []string{"10.0.5.20"}) {
+ t.Fatalf("A's GET = %+v", g.Policy)
+ }
+ r = A.mustPut(t, "shop", `{"egress_allow":[]}`, policyIfMatch(g.ETag))
+ if r.Policy.Revision != 2 || len(r.Policy.EgressAllow) != 0 {
+ t.Fatalf("A's write = %+v", r.Policy)
+ }
+
+ // 3. B is refused, naming A's lock, and changes nothing.
+ obj, row, events := B.bucketBytes(t), B.row(t, "shop"), len(B.policyEvents(t, "shop"))
+ w.putHeld(t, B, `{"default_size":"small"}`, policyIfMatch(r.ETag))
+ if !bytes.Equal(B.bucketBytes(t), obj) {
+ t.Fatal("B's refused PUT changed the object")
+ }
+ if got := B.row(t, "shop"); got.Revision != row.Revision || !bytes.Equal(got.Wire, row.Wire) || got.SyncedETag != row.SyncedETag {
+ t.Fatalf("B's refused PUT changed its row: revision %d etag %q, was revision %d etag %q",
+ got.Revision, got.SyncedETag, row.Revision, row.SyncedETag)
+ }
+ if n := len(B.policyEvents(t, "shop")); n != events {
+ t.Fatalf("B's refused PUT wrote an event: %d, was %d", n, events)
+ }
+
+ // 4. B's re-sign is refused too: nothing signed.
+ B.srv.resignAppPolicies(context.Background())
+ if !bytes.Equal(B.bucketBytes(t), obj) {
+ t.Fatal("B's refused re-sign changed the object")
+ }
+ B.logs.has(t, "console: app policies: not re-signing; another console (")
+
+ // 5. A's lease lapses.
+ w.lapse(t, A)
+
+ // 6. B builds on the bucket's revision 2, not its own stale table.
+ g = B.get(t, "shop")
+ if len(g.Policy.EgressAllow) != 0 || g.Policy.Revision != 2 {
+ t.Fatalf("B's GET after the lapse = %+v", g.Policy)
+ }
+ r = B.mustPut(t, "shop", `{"default_size":"small"}`, policyIfMatch(g.ETag))
+ stored, wire, etag := storedPolicy(t, B.rel, B.pub, B.sqid, "shop")
+ if stored.Revision != 3 || len(stored.EgressAllow) != 0 || stored.DefaultSize != "small" || r.ETag != etag {
+ t.Fatalf("B's write stored %+v", stored)
+ }
+ if row := B.row(t, "shop"); row.Revision != 3 || !bytes.Equal(row.Wire, wire) || row.SyncedETag != etag {
+ t.Fatalf("B's row %+v", row)
+ }
+
+ // 7. A is refused, naming B's lock; A's re-sign signs nothing.
+ w.putHeld(t, A, `{"default_size":"large"}`, policyIfMatch(etag))
+ A.srv.resignAppPolicies(context.Background())
+ if !bytes.Equal(A.bucketBytes(t), wire) {
+ t.Fatal("A's refused re-sign changed the object")
+ }
+}
+
+// Astra plan R3-1: a writer paused inside a write must not overwrite the
+// next writer's revoke. Not parallel, as above.
+func TestALapsedWriterCannotOverwriteTheNextWritersRevoke(t *testing.T) {
+ w := newTwoConsoles(t)
+ A, B := w.a, w.b
+
+ // 1. A grants.
+ r1 := A.mustPut(t, "shop", `{"egress_allow":["10.0.5.20"]}`, policyCreate)
+ rowA, eventsA := A.row(t, "shop"), len(A.policyEvents(t, "shop"))
+
+ // 2. A pauses inside a write, its base read.
+ enteredA, releaseA := make(chan struct{}), make(chan struct{})
+ var pausedA atomic.Bool
+ A.srv.policyAfterBase = func(string, string) {
+ if pausedA.CompareAndSwap(false, true) {
+ close(enteredA)
+ <-releaseA
+ }
+ }
+ type answer struct {
+ status int
+ out string
+ }
+ doneA := make(chan answer, 1)
+ go func() {
+ st, _, out := policyReq(t, A.consoleFixture, http.MethodPut, "shop", A.owner, `{"default_size":"small"}`, policyIfMatch(r1.ETag))
+ doneA <- answer{st, out}
+ }()
+ <-enteredA
+
+ // 3. A's lease lapses.
+ w.lapse(t, A)
+
+ // 4. B revokes.
+ g := B.get(t, "shop")
+ if !slices.Equal(g.Policy.EgressAllow, []string{"10.0.5.20"}) {
+ t.Fatalf("B's GET = %+v", g.Policy)
+ }
+ rB := B.mustPut(t, "shop", `{"egress_allow":[]}`, policyIfMatch(g.ETag))
+ if rB.Policy.Revision != 2 {
+ t.Fatalf("B's revoke = %+v", rB.Policy)
+ }
+ wB := B.bucketBytes(t)
+
+ // 5. A resumes: refused, and the revoke stands byte for byte.
+ close(releaseA)
+ a := <-doneA
+ if a.status != http.StatusServiceUnavailable || errorOf(t, a.out) != policyLeaseLostMsg {
+ t.Fatalf("A's paused PUT = %d %s, want 503 %q", a.status, a.out, policyLeaseLostMsg)
+ }
+ if !bytes.Equal(A.bucketBytes(t), wB) {
+ t.Fatal("A's lapsed write replaced B's revoke")
+ }
+ // Refused before anything was audited, signed or recorded: the bucket's
+ // compare-and-swap alone would also keep wB, but only after A had
+ // recorded its own revision 2.
+ if got := A.row(t, "shop"); got.Revision != rowA.Revision || !bytes.Equal(got.Wire, rowA.Wire) {
+ t.Fatalf("A's refused write changed its row: revision %d pending %v, was revision %d", got.Revision, got.Pending, rowA.Revision)
+ }
+ if n := len(A.policyEvents(t, "shop")); n != eventsA {
+ t.Fatalf("A's refused write audited a change: %d events, was %d", n, eventsA)
+ }
+
+ // 6. B's re-sign pauses after its base read; B's lease lapses; A writes.
+ enteredB, releaseB := make(chan struct{}), make(chan struct{})
+ var pausedB atomic.Bool
+ B.srv.policyAfterBase = func(string, string) {
+ if pausedB.CompareAndSwap(false, true) {
+ close(enteredB)
+ <-releaseB
+ }
+ }
+ rowB := B.row(t, "shop")
+ doneB := make(chan struct{})
+ go func() {
+ defer close(doneB)
+ B.srv.resignAppPolicies(context.Background())
+ }()
+ <-enteredB
+ w.lapse(t, B)
+ g = A.get(t, "shop")
+ rA := A.mustPut(t, "shop", `{"default_size":"large"}`, policyIfMatch(g.ETag))
+ stored, wA, _ := storedPolicy(t, A.rel, A.pub, A.sqid, "shop")
+ if rA.Policy.Revision != 3 || stored.Revision != 3 || len(stored.EgressAllow) != 0 || stored.DefaultSize != "large" {
+ t.Fatalf("A's write stored %+v", stored)
+ }
+
+ // 7. B resumes: its pass signs and writes nothing.
+ close(releaseB)
+ <-doneB
+ if !bytes.Equal(B.bucketBytes(t), wA) {
+ t.Fatal("B's lapsed re-sign replaced A's write")
+ }
+ if got := B.row(t, "shop"); got.Revision != rowB.Revision || !bytes.Equal(got.Wire, rowB.Wire) || got.Pending != rowB.Pending {
+ t.Fatalf("B's lapsed re-sign signed and recorded: revision %d pending %v, was revision %d pending %v",
+ got.Revision, got.Pending, rowB.Revision, rowB.Pending)
+ }
+ B.logs.has(t, "console: app policies: stopped re-signing at bab/shop; the writer lease was lost")
+}
diff --git a/internal/console/server.go b/internal/console/server.go
index cb3e7c8e4..b0369d2d4 100644
--- a/internal/console/server.go
+++ b/internal/console/server.go
@@ -1002,13 +1002,16 @@ type Server struct {
// for it, under policySyncedMu: an app not yet touched in the current
// holding builds on the bucket's record when another console may have
// written it (ruling 10). policyAfterBase is a TEST SEAM (nil in
- // production), called right after policyBase returns in a write, so a
- // test can pause a writer that has read its base.
+ // production), called right after policyBase returns in a write or a
+ // re-sign, so a test can pause a writer that has read its base.
+ // policyListedN, under policyStateMu, is the holding the re-sign pass
+ // last listed control/app-policy/ in (0: not holding, never counted).
policyLocksMu sync.Mutex
policyLocks map[string]*sync.Mutex
policySyncedMu sync.Mutex
policySynced map[string]uint64
policyAfterBase func(sqid, app string)
+ policyListedN uint64
// sendApp is how the alert watcher (alertwatch.go) delivers keymail:
// km.SendApp in production, a capture in tests. A field rather than a
// direct call so an alertPass assertion needs no keymail server.