amadan

carlos / platform Public

console: the app policy door (#328, #329)

GET and PUT /api/cli/apps/{account}/{app}/policy, operator-only
(apiSystemRoller, the zero role Set). The PUT is compare-and-swap on the
object's ETag, under the app's policy lock and the writer lease; it builds
on the recorded content (or, in a new writer holding, on the bucket's
record), audits before anything is signed and refuses when the audit
fails, signs one above both the bucket's and the table's revision,
records the signed bytes before publishing, and binds every signing and
publication attempt to its holding. The app_policy_revisions table is
classified as gated, app-scoped, with its writes behind withAutoAppSQL
adapters. checkEgressGrant is plan 2's stub.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude e131b5e56f544dc24f681062cfd031916d7e949b parent e0cd222
11 files changed, +2095 −5
  • internal/console/apipolicy.go +764 −0
  • internal/console/apipolicy_test.go +1034 −0
  • internal/console/apppolicy_store.go +216 −0
  • internal/console/authcli.go +9 −0
  • internal/console/claim_invariant_fixtures_test.go +25 −0
  • internal/console/claim_invariant_probes_read_test.go +12 −0
  • internal/console/server.go +14 −0
  • internal/console/store.go +6 −1
  • internal/console/store_gate.go +1 −0
  • internal/console/store_gate_test.go +1 −1
  • internal/console/svcroutes_test.go +13 −3
diff --git a/internal/console/apipolicy.go b/internal/console/apipolicy.go
new file mode 100644
index 000000000..553fd0bce
--- /dev/null
+++ b/internal/console/apipolicy.go
@@ -0,0 +1,764 @@
+// The app policy door (instance sizes and egress spec §6, #328, #329):
+// GET and PUT /api/cli/apps/{account}/{app}/policy, the operator's view and
+// edit of one app's policy object (internal/apppolicy): the private
+// addresses its instances may reach and the sizes they run at.
+//
+// GATE: system operators only, both doors: apiSystemRoller (an owner of the
+// ops account, or an address on CARLOS_SYSTEM_OPERATORS) in the handler, and
+// the ZERO role Set at the mux (authcli.go). An egress allow opens private
+// addresses to an app's instances, so no credential an app holds may reach
+// the door that writes it, and no service role is the right holder of an
+// operator's decision.
+//
+// The audit is not best-effort here, unlike s.audit: spec §6 says every
+// change is recorded, so the event is written first, inside the write, and a
+// change whose event could not be written is refused before anything is
+// signed, recorded or published.
+//
+// Revisions. A box keeps the highest revision it has verified, so every write
+// is signed one above both the revision the bucket's verified record carries
+// and the one this console's table last recorded (apppolicy_store.go). The
+// table is the authority for content once a revision is recorded: the door
+// records the signed bytes BEFORE it writes them to the bucket, so a write
+// whose reply is lost, or a bucket writer that puts an older record back, can
+// never leave the console building on, or re-signing, content older than
+// what it last signed (Astra plan R1-1). The change stands once recorded; a
+// failed publication is finished by the re-sign pass or the next write.
+//
+// One writer per bucket (ruling 10, policylease.go). Nothing is signed
+// without the writer lease, and every signing and publication attempt is
+// bound to the holding the write began under (policyAttempt). A NEW holding
+// means another console may have written since this one last did, so the
+// first write of an app in a new holding builds on the bucket's verified
+// record rather than on this console's table, unless the bucket still holds
+// exactly what this console last published (then a publication a crash
+// interrupted is still finished). A directory deployment's console is the
+// only writer by construction and never rebases.
+package console
+
+import (
+ "bytes"
+ "context"
+ "crypto/ed25519"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "maps"
+ "net/http"
+ "os"
+ "slices"
+ "sort"
+ "strings"
+ "sync"
+ "time"
+
+ "github.com/carlosframework/platform/internal/apppolicy"
+ "github.com/carlosframework/platform/internal/cgroup"
+ "github.com/carlosframework/platform/internal/egressaddr"
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// Door copy (collected for review in plan Task 15).
+const (
+ policyNothingMsg = "nothing to set: send egress_allow, default_size or size"
+ policyDefaultSizeMsg = "default_size must be small, standard, large or xlarge, or empty for the build's size"
+ policySizeMsg = "size must be small, standard, large, xlarge or default"
+ policyNoInstanceMsg = "no instance %s in this app"
+ policyPreconditionMsg = "send If-Match with the policy's etag, or If-None-Match: * when there is none"
+ policyChangedMsg = "the policy changed since it was read. Read it again and retry"
+ policyRecordFailedMsg = "the change could not be recorded, so it was not made. Try again"
+ policyAuditFailedMsg = "the change could not be audited, so it was not made. Try again"
+ policyUnpublishedMsg = "the change is recorded but not yet published. Run the command again to publish it now"
+ policyReadFailedMsg = "could not read the policy. Try again"
+ policyInvalidHostMsg = "invalid host"
+ policyAuditAction = "app-policy"
+ policyBodyLimit = 8 << 10
+ policySizeValueDefault = "default"
+)
+
+// policyPutReq is the PUT body. null or absent leaves a field as it is; []
+// clears the allow; "" clears the default size; a size value of "default"
+// removes that host's override.
+type policyPutReq struct {
+ EgressAllow *[]string `json:"egress_allow"`
+ DefaultSize *string `json:"default_size"`
+ Size *policySizeReq `json:"size"`
+}
+
+type policySizeReq struct {
+ Host string `json:"host"`
+ Value string `json:"value"`
+}
+
+// policyReply is both doors' answer. ETag is the object's ("" when there is
+// none) and is also the ETag header. NotChecked is checkEgressGrant's.
+type policyReply struct {
+ Policy apppolicy.Policy `json:"policy"`
+ ETag string `json:"etag"`
+ NotChecked []string `json:"not_checked"`
+ Stale []string `json:"stale_overrides,omitempty"`
+}
+
+// policyLock is the app's policy mutex, created on first use as
+// doorPublishLock is. The door holds it from before its base read until its
+// reply; the re-sign pass shares it.
+func (s *Server) policyLock(sqid, app string) *sync.Mutex {
+ s.policyLocksMu.Lock()
+ defer s.policyLocksMu.Unlock()
+ if s.policyLocks == nil {
+ s.policyLocks = map[string]*sync.Mutex{}
+ }
+ k := sqid + "/" + app
+ mu := s.policyLocks[k]
+ if mu == nil {
+ mu = &sync.Mutex{}
+ s.policyLocks[k] = mu
+ }
+ return mu
+}
+
+// policyTouched reports whether holding n already chose a base for key.
+// Holdings are numbered from 1, so n 0 (a reader holding nothing) is never
+// touched.
+func (s *Server) policyTouched(key string, n uint64) bool {
+ s.policySyncedMu.Lock()
+ defer s.policySyncedMu.Unlock()
+ got, ok := s.policySynced[key]
+ return ok && got == n
+}
+
+func (s *Server) markPolicyTouched(key string, n uint64) {
+ s.policySyncedMu.Lock()
+ defer s.policySyncedMu.Unlock()
+ if s.policySynced == nil {
+ s.policySynced = map[string]uint64{}
+ }
+ s.policySynced[key] = n
+}
+
+// policyReadHolding is the holding a read evaluates the base rules under:
+// the current one while its lease is healthy, else holding 0, which no write
+// ever marks, so a console not holding the lease shows what a takeover would
+// build on. It takes no lease.
+func (s *Server) policyReadHolding() policyHolding {
+ if s.cfg.Bucket == "" {
+ return policyHolding{n: 1, sole: true}
+ }
+ s.policyStateMu.Lock()
+ cur := s.policyCur
+ s.policyStateMu.Unlock()
+ if cur != nil && cur.h.lease.Err() == nil {
+ return cur.h
+ }
+ return policyHolding{}
+}
+
+// policyPub is the key policies verify against: the signing key's own half
+// when this console signs, else the pinned release key.
+func (s *Server) policyPub() ed25519.PublicKey {
+ if len(s.cfg.Key) == ed25519.PrivateKeySize {
+ if pub, ok := s.cfg.Key.Public().(ed25519.PublicKey); ok {
+ return pub
+ }
+ }
+ return s.cfg.ReleasePub
+}
+
+// policyRefusal is a policyBase failure with the answer the door gives.
+type policyRefusal struct {
+ status int
+ msg string
+ err error
+}
+
+func (e *policyRefusal) Error() string { return e.err.Error() }
+func (e *policyRefusal) Unwrap() error { return e.err }
+
+// policyBase reads the bucket's object and the table's row and chooses the
+// policy a write builds on (Task 9 step 7): the base, the bucket's ETag ("" for
+// no object), the revision the next write signs, and the row (as adopted, when
+// commit adopted the bucket's record). commit is false for a read: nothing is
+// recorded and no holding is marked.
+func (s *Server) policyBase(ctx context.Context, acct *Account, app string, h policyHolding, commit bool) (base apppolicy.Policy, etag string, next uint64, row AppPolicyRow, err error) {
+ base, etag, next, row, _, err = s.policyBaseFor(ctx, acct, app, h, commit)
+ return base, etag, next, row, err
+}
+
+// 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) {
+ pub := s.policyPub()
+ key := acct.Sqid + "/" + app
+ b, wire, etag, gerr := apppolicy.GetForUpdate(ctx, s.cfg.Releases, pub, acct.Sqid, app)
+ absent := false
+ switch {
+ case gerr == nil:
+ case errors.Is(gerr, os.ErrNotExist):
+ absent, etag = true, ""
+ case errors.Is(gerr, apppolicy.ErrUnverifiable):
+ b = nil
+ default:
+ return base, "", 0, 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}
+ }
+ recorded := row.Revision
+ var bucketRev uint64
+ if b != nil {
+ bucketRev = b.Revision
+ }
+ adopt := func() error {
+ if !commit {
+ return nil
+ }
+ if err := s.store.AdoptAppPolicyContext(ctx, acct.ID, app, b.Revision, wire, etag); err != nil {
+ return &policyRefusal{http.StatusServiceUnavailable, policyRecordFailedMsg, err}
+ }
+ row = AppPolicyRow{AccountID: acct.ID, Sqid: acct.Sqid, App: app, Revision: max(recorded, b.Revision),
+ Wire: wire, SyncedETag: etag}
+ return nil
+ }
+ notBuilding := func(why error) {
+ s.policyLogf("console: app policy %s/%s: not building on the stored object: %v", acct.Sqid, app, why)
+ }
+
+ switch {
+ case absent:
+ // An absent object is a purge, a create-time clear or a first write
+ // that never landed. The re-sign pass never revives one, so neither
+ // does the door: the recorded content is not used.
+ base = apppolicy.Policy{V: 1}
+ case !h.sole && !s.policyTouched(key, h.n) && etag != row.SyncedETag && !bytes.Equal(wire, row.Wire):
+ // A new holding, and the bucket holds something this console did not
+ // last publish: another console may have written it.
+ if b == nil {
+ notBuilding(gerr)
+ base = apppolicy.Policy{V: 1}
+ break
+ }
+ if err := adopt(); err != nil {
+ return base, "", 0, row, "", err
+ }
+ base = *b
+ s.policyLogf("console: app policy %s/%s: new writer holding; building on the bucket's record (revision %d)",
+ acct.Sqid, app, b.Revision)
+ case b != nil && b.Revision > recorded:
+ // A database restored from backup, or a record this console signed
+ // 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
+ }
+ 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}
+ }
+ base = p
+ switch {
+ case b == nil:
+ notBuilding(gerr)
+ case b.Revision < recorded:
+ notBuilding(fmt.Errorf("it is revision %d, below the recorded %d", b.Revision, recorded))
+ case !samePolicyPayload(wire, row.Wire):
+ notBuilding(fmt.Errorf("it is revision %d with other content than the recorded one", b.Revision))
+ }
+ default:
+ // No row, and an object that does not verify.
+ notBuilding(gerr)
+ base = apppolicy.Policy{V: 1}
+ }
+
+ if s.cfg.HostClaim {
+ cur, err = s.store.ClaimAppID(acct.ID, app)
+ if err != nil {
+ return base, "", 0, row, "", &policyRefusal{http.StatusServiceUnavailable, policyReadFailedMsg, err}
+ }
+ if base.AppID != cur {
+ // A predecessor's policy that a failed create-time clear left.
+ if base.Revision != 0 {
+ s.policyLogf("console: app policy %s/%s: stored policy is for incarnation %q, not %q; starting empty",
+ acct.Sqid, app, base.AppID, cur)
+ }
+ base = apppolicy.Policy{V: 1}
+ }
+ }
+ if commit {
+ s.markPolicyTouched(key, h.n)
+ }
+ return base, etag, max(recorded, bucketRev) + 1, row, cur, nil
+}
+
+// samePolicyPayload reports whether two signed records carry the same
+// payload, whatever their signatures.
+func samePolicyPayload(a, b []byte) bool {
+ var wa, wb struct {
+ Payload string `json:"payload"`
+ }
+ if json.Unmarshal(a, &wa) != nil || json.Unmarshal(b, &wb) != nil {
+ return false
+ }
+ return wa.Payload != "" && wa.Payload == wb.Payload
+}
+
+// policyAccountApp resolves the path's account and app for an operator: a
+// missing account or app is the same bare 404 as the operator gate's.
+func (s *Server) policyAccountApp(w http.ResponseWriter, r *http.Request) (*Account, string, bool) {
+ acct, err := s.store.AccountBySqid(r.PathValue("account"))
+ if err != nil {
+ s.policyLogf("console: app policy: account: %v", err)
+ jsonError(w, http.StatusServiceUnavailable, policyReadFailedMsg)
+ return nil, "", false
+ }
+ if acct == nil {
+ http.NotFound(w, r)
+ return nil, "", false
+ }
+ app := r.PathValue("app")
+ ok, err := s.store.HasAppIn(acct.ID, app)
+ if err != nil {
+ s.policyLogf("console: app policy %s/%s: app: %v", acct.Sqid, app, err)
+ jsonError(w, http.StatusServiceUnavailable, policyReadFailedMsg)
+ return nil, "", false
+ }
+ if !ok {
+ http.NotFound(w, r)
+ return nil, "", false
+ }
+ return acct, app, true
+}
+
+func (s *Server) policyRefuse(w http.ResponseWriter, acct *Account, app string, err error) {
+ s.policyLogf("console: app policy %s/%s: %v", acct.Sqid, app, err)
+ var ref *policyRefusal
+ if errors.As(err, &ref) {
+ jsonError(w, ref.status, ref.msg)
+ return
+ }
+ jsonError(w, http.StatusBadGateway, policyReadFailedMsg)
+}
+
+// handleAPIPolicyGet answers the policy a write would build on now (the
+// recorded content, {v:1} for an absent or unusable object, {v:1} for another
+// incarnation's policy under claims), the bucket's ETag, and the overrides
+// whose instance is gone or was recreated. It records nothing and takes no
+// lease. An expired record is shown as stored; its signed_at says when.
+func (s *Server) handleAPIPolicyGet(w http.ResponseWriter, r *http.Request) {
+ if _, _, ok := s.apiSystemRoller(w, r); !ok {
+ return
+ }
+ acct, app, ok := s.policyAccountApp(w, r)
+ if !ok {
+ return
+ }
+ if s.cfg.Releases == nil {
+ jsonError(w, http.StatusServiceUnavailable, doorKeyNoBucketMsg)
+ return
+ }
+ ctx, cancel := context.WithTimeout(r.Context(), doorKeyPublishBudget)
+ defer cancel()
+ base, etag, _, _, err := s.policyBase(ctx, acct, app, s.policyReadHolding(), false)
+ if err != nil {
+ s.policyRefuse(w, acct, app, err)
+ return
+ }
+ w.Header().Set("ETag", etag)
+ writeJSON(w, http.StatusOK, policyReply{Policy: base, ETag: etag, NotChecked: []string{},
+ Stale: s.staleOverrides(ctx, acct, app, base)})
+}
+
+// policyPrecondition is the write's compare-and-swap target: If-Match's
+// ETag, or "" for If-None-Match: * (there is no object). ok is false when
+// the request carries neither.
+func policyPrecondition(r *http.Request) (string, bool) {
+ if m := strings.TrimSpace(r.Header.Get("If-Match")); m != "" {
+ return m, true
+ }
+ if strings.TrimSpace(r.Header.Get("If-None-Match")) == "*" {
+ return "", true
+ }
+ return "", false
+}
+
+// handleAPIPolicyPut changes the app's policy: compare-and-swap on the
+// object's ETag, under the app's policy lock and the writer lease, audited
+// before anything is signed, recorded before it is published. Every refusal
+// before the write writes nothing.
+func (s *Server) handleAPIPolicyPut(w http.ResponseWriter, r *http.Request) {
+ addr, _, ok := s.apiSystemRoller(w, r)
+ if !ok {
+ return
+ }
+ acct, app, ok := s.policyAccountApp(w, r)
+ if !ok {
+ return
+ }
+ p, _ := principalFrom(r.Context())
+ if !s.allowWrite(p, time.Now()) {
+ jsonError(w, http.StatusTooManyRequests, apiSwapLimited)
+ return
+ }
+ if !s.doorKeyWritable(w) {
+ return
+ }
+ var req policyPutReq
+ if !readJSONStrictLimit(w, r, &req, policyBodyLimit) {
+ return
+ }
+ if req.EgressAllow == nil && req.DefaultSize == nil && req.Size == nil {
+ jsonError(w, http.StatusBadRequest, policyNothingMsg)
+ return
+ }
+ want, ok := policyPrecondition(r)
+ if !ok {
+ jsonError(w, http.StatusPreconditionRequired, policyPreconditionMsg)
+ return
+ }
+
+ // Detached from the client: once a change is recorded it stands, and a
+ // client that hangs up must not cut its publication short.
+ ctx, cancel := context.WithTimeout(context.WithoutCancel(r.Context()), doorKeyPublishBudget)
+ defer cancel()
+ mu := s.policyLock(acct.Sqid, app)
+ mu.Lock()
+ defer mu.Unlock()
+ hold, ref := s.policyWriter(ctx)
+ if ref != nil {
+ jsonError(w, ref.Status, ref.Msg)
+ return
+ }
+ old, etag, next, _, cur, err := s.policyBaseFor(ctx, acct, app, hold, true)
+ if s.policyAfterBase != nil {
+ s.policyAfterBase(acct.Sqid, app)
+ }
+ if err != nil {
+ s.policyRefuse(w, acct, app, err)
+ return
+ }
+ if want != etag {
+ jsonError(w, http.StatusPreconditionFailed, policyChangedMsg)
+ return
+ }
+ pol, status, msg := s.applyPolicyChange(ctx, acct, app, old, req)
+ if status != 0 {
+ jsonError(w, status, msg)
+ return
+ }
+ if s.cfg.HostClaim {
+ pol.AppID = cur
+ }
+ notChecked := []string{}
+ if req.EgressAllow != nil {
+ nc, status, msg := s.checkEgressGrant(ctx, acct, app, old.EgressAllow, pol.EgressAllow)
+ if status != 0 {
+ jsonError(w, status, msg)
+ return
+ }
+ if nc != nil {
+ notChecked = nc
+ }
+ }
+ detail := policyAuditDetail(old, pol, req)
+
+ var (
+ refusal *doorRefusal
+ reply policyReply
+ ran bool
+ )
+ errRefused := errors.New("console: app policy: refused")
+ err = s.appMutation(ctx, acct, app, policyAuditAction, func(ctx context.Context) error {
+ ran = true
+ reply, refusal = s.writePolicy(ctx, acct, app, addr, hold, pol, next, etag, detail)
+ if refusal != nil {
+ return errRefused
+ }
+ return nil
+ })
+ switch {
+ case refusal != nil:
+ jsonError(w, refusal.Status, refusal.Msg)
+ return
+ case !ran:
+ s.policyLogf("console: app policy %s/%s: %v", acct.Sqid, app, err)
+ jsonError(w, http.StatusServiceUnavailable, policyRecordFailedMsg)
+ return
+ case err != nil:
+ // Recorded and published; only the mutation's own close failed.
+ s.policyLogf("console: app policy %s/%s: after the write: %v", acct.Sqid, app, err)
+ }
+ s.bumpSerial(ctx)
+ reply.NotChecked = notChecked
+ w.Header().Set("ETag", reply.ETag)
+ writeJSON(w, http.StatusOK, reply)
+}
+
+// writePolicy is the write itself, inside the app mutation: the holding
+// check, the mandatory audit, the check before signing, the signing, the
+// record, and the publication. A refusal says what the caller answers.
+func (s *Server) writePolicy(ctx context.Context, acct *Account, app, addr string, hold policyHolding,
+ p apppolicy.Policy, next uint64, etag, detail string) (policyReply, *doorRefusal) {
+ _, done, ref := s.policyAttempt(ctx, hold)
+ done()
+ if ref != nil {
+ return policyReply{}, ref
+ }
+ if err := s.auditRequired(ctx, acct, app, addr, detail); err != nil {
+ s.policyLogf("console: app policy %s/%s: audit: %v", acct.Sqid, app, err)
+ return policyReply{}, &doorRefusal{http.StatusServiceUnavailable, policyAuditFailedMsg}
+ }
+ p.Account, p.App, p.Revision, p.SignedAt = acct.Sqid, app, next, time.Now().UTC()
+ _, done, ref = s.policyAttempt(ctx, hold)
+ if ref != nil {
+ return policyReply{}, ref
+ }
+ wire, err := apppolicy.Sign(s.cfg.Key, p)
+ done()
+ if err != nil {
+ s.policyLogf("console: app policy %s/%s: sign: %v", acct.Sqid, app, err)
+ return policyReply{}, &doorRefusal{http.StatusServiceUnavailable, policyRecordFailedMsg}
+ }
+ if err := s.store.RecordAppPolicyContext(ctx, acct.ID, app, next, wire); err != nil {
+ s.policyLogf("console: app policy %s/%s: record revision %d: %v", acct.Sqid, app, next, err)
+ return policyReply{}, &doorRefusal{http.StatusServiceUnavailable, policyRecordFailedMsg}
+ }
+ newETag, ref := s.publishPolicy(ctx, acct, app, hold, wire, next, etag)
+ if ref != nil {
+ return policyReply{}, ref
+ }
+ if err := s.store.SyncedAppPolicyContext(ctx, acct.ID, app, next, newETag); err != nil {
+ // The row stays pending, so the re-sign pass keeps it a candidate,
+ // and the next write finds the bytes equal to the recorded ones.
+ s.policyLogf("console: app policy %s/%s: mark revision %d published: %v", acct.Sqid, app, next, err)
+ }
+ return policyReply{Policy: p, ETag: newETag}, nil
+}
+
+// publishPolicy writes the recorded bytes, compare-and-swap on etag, each
+// attempt bound to the holding (Astra plan R3-1): a refused check, or an
+// attempt cut off by its holding, is the lease-lost refusal and no further
+// attempt is made. A lost compare-and-swap re-reads under that attempt's
+// check: a record at or above this revision with other bytes is another
+// writer's and is never overwritten; otherwise the next attempt writes the
+// same bytes against the new ETag, up to doorKeyPublishAttempts. An object
+// that existed at the base read and has disappeared since (a purge) is never
+// recreated; a create whose object is still absent is tried again as a create.
+func (s *Server) publishPolicy(ctx context.Context, acct *Account, app string, hold policyHolding,
+ wire []byte, next uint64, etag string) (string, *doorRefusal) {
+ lost := &doorRefusal{http.StatusServiceUnavailable, policyLeaseLostMsg}
+ unpublished := &doorRefusal{http.StatusBadGateway, policyUnpublishedMsg}
+ pub := s.policyPub()
+ created := etag == ""
+ for attempt := 0; attempt < doorKeyPublishAttempts; attempt++ {
+ actx, done, ref := s.policyAttempt(ctx, hold)
+ if ref != nil {
+ return "", ref
+ }
+ newETag, err := apppolicy.PutWire(actx, s.cfg.Releases, acct.Sqid, app, wire, etag)
+ if err == nil {
+ done()
+ return newETag, nil
+ }
+ if !hold.sole && actx.Err() != nil {
+ done()
+ s.policyLogf("console: app policy %s/%s: publish revision %d: the writer holding ended: %v", acct.Sqid, app, next, err)
+ return "", lost
+ }
+ if !errors.Is(err, release.ErrCASConflict) {
+ done()
+ s.policyLogf("console: app policy %s/%s: publish revision %d: %v", acct.Sqid, app, next, err)
+ return "", unpublished
+ }
+ b, got, gotETag, gerr := apppolicy.GetForUpdate(actx, s.cfg.Releases, pub, acct.Sqid, app)
+ ended := !hold.sole && actx.Err() != nil
+ done()
+ switch {
+ case ended:
+ s.policyLogf("console: app policy %s/%s: publish revision %d: the writer holding ended", acct.Sqid, app, next)
+ return "", lost
+ case gerr == nil && bytes.Equal(got, wire):
+ return gotETag, nil
+ case gerr == nil && b.Revision >= next:
+ s.policyLogf("console: app policy %s/%s: publish revision %d: the bucket holds another writer's revision %d",
+ acct.Sqid, app, next, b.Revision)
+ return "", lost
+ case gerr == nil || errors.Is(gerr, apppolicy.ErrUnverifiable):
+ etag = gotETag
+ case errors.Is(gerr, os.ErrNotExist) && created:
+ // Still no object: the next attempt is the same create.
+ default:
+ s.policyLogf("console: app policy %s/%s: publish revision %d: re-read: %v", acct.Sqid, app, next, gerr)
+ return "", unpublished
+ }
+ }
+ s.policyLogf("console: app policy %s/%s: publish revision %d: %d attempts lost the compare-and-swap",
+ acct.Sqid, app, next, doorKeyPublishAttempts)
+ return "", unpublished
+}
+
+// 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
+// gone or was recreated dropped, so no member door ever writes the object. A
+// non-zero status refuses with msg.
+func (s *Server) applyPolicyChange(ctx context.Context, acct *Account, app string, base apppolicy.Policy, req policyPutReq) (apppolicy.Policy, int, string) {
+ p := base
+ p.EgressAllow = slices.Clone(base.EgressAllow)
+ p.Sizes = maps.Clone(base.Sizes)
+ if req.EgressAllow != nil {
+ norm, err := egressaddr.Normalize(*req.EgressAllow)
+ if err != nil {
+ return p, http.StatusBadRequest, err.Error()
+ }
+ p.EgressAllow = norm
+ }
+ if req.DefaultSize != nil {
+ if v := *req.DefaultSize; v != "" {
+ if _, ok := cgroup.SizeByName(v); !ok {
+ return p, http.StatusBadRequest, policyDefaultSizeMsg
+ }
+ }
+ p.DefaultSize = *req.DefaultSize
+ }
+ setHost := ""
+ if req.Size != nil {
+ v := req.Size.Value
+ if v != policySizeValueDefault {
+ if _, ok := cgroup.SizeByName(v); !ok {
+ return p, http.StatusBadRequest, policySizeMsg
+ }
+ }
+ host := strings.ToLower(req.Size.Host)
+ if !provision.ValidHostShape(host) {
+ return p, http.StatusBadRequest, policyInvalidHostMsg
+ }
+ rec, err := provision.GetRecord(ctx, s.cfg.Releases, acct.Sqid, app, host)
+ if errors.Is(err, os.ErrNotExist) {
+ return p, http.StatusNotFound, fmt.Sprintf(policyNoInstanceMsg, host)
+ }
+ if err != nil {
+ s.policyLogf("console: app policy %s/%s: record %s: %v", acct.Sqid, app, host, err)
+ return p, http.StatusBadGateway, policyReadFailedMsg
+ }
+ if v == policySizeValueDefault {
+ delete(p.Sizes, host)
+ } else {
+ if p.Sizes == nil {
+ p.Sizes = map[string]apppolicy.SizeOverride{}
+ }
+ p.Sizes[host] = apppolicy.SizeOverride{Size: v, CreatedAt: rec.CreatedAt}
+ }
+ setHost = host
+ }
+ for host, o := range p.Sizes {
+ if host == setHost {
+ continue
+ }
+ if current, err := s.overrideCurrent(ctx, acct, app, host, o); err != nil {
+ s.policyLogf("console: app policy %s/%s: record %s: %v", acct.Sqid, app, host, err)
+ } else if !current {
+ delete(p.Sizes, host)
+ }
+ }
+ if len(p.Sizes) == 0 {
+ p.Sizes = nil
+ }
+ return p, 0, ""
+}
+
+// overrideCurrent reports whether host's override was set on the instance
+// that exists now: its record is present and has the override's CreatedAt.
+func (s *Server) overrideCurrent(ctx context.Context, acct *Account, app, host string, o apppolicy.SizeOverride) (bool, error) {
+ rec, err := provision.GetRecord(ctx, s.cfg.Releases, acct.Sqid, app, host)
+ if errors.Is(err, os.ErrNotExist) {
+ return false, nil
+ }
+ if err != nil {
+ return false, err
+ }
+ return rec.CreatedAt.Equal(o.CreatedAt), nil
+}
+
+// staleOverrides lists, sorted, the override hosts whose instance record is
+// absent or has another CreatedAt. A record that cannot be read is left out.
+func (s *Server) staleOverrides(ctx context.Context, acct *Account, app string, p apppolicy.Policy) []string {
+ var stale []string
+ for host, o := range p.Sizes {
+ current, err := s.overrideCurrent(ctx, acct, app, host, o)
+ if err != nil {
+ s.policyLogf("console: app policy %s/%s: record %s: %v", acct.Sqid, app, host, err)
+ continue
+ }
+ if !current {
+ stale = append(stale, host)
+ }
+ }
+ sort.Strings(stale)
+ return stale
+}
+
+// checkEgressGrant is the fleet-address check on a new egress allow (spec
+// §4.2), called whenever the request carries egress_allow; old is the stored
+// list, next the normalized new one. A non-zero status refuses with msg
+// before the audit; notChecked is the reply's not_checked. Plan 1's body is
+// a stub: plan 2 fills it.
+func (s *Server) checkEgressGrant(ctx context.Context, acct *Account, app string, old, next []string) (notChecked []string, status int, msg string) {
+ return []string{}, 0, ""
+}
+
+// auditRequired records the change's event and returns its failure: for the
+// policy door the audit is part of the write, never best-effort (spec §6).
+func (s *Server) auditRequired(ctx context.Context, acct *Account, app, addr, detail string) error {
+ return s.store.RecordEventContext(ctx, acct.ID, app, addr, policyAuditAction, detail, time.Now())
+}
+
+// policyAuditDetail names what the request changed, old -> new, ending
+// " (cli)". A request that changed nothing names what it sent.
+func policyAuditDetail(old, next apppolicy.Policy, req policyPutReq) string {
+ var sent, changed []string
+ add := func(part string, differs bool) {
+ sent = append(sent, part)
+ if differs {
+ changed = append(changed, part)
+ }
+ }
+ allow := func(l []string) string {
+ if len(l) == 0 {
+ return "none"
+ }
+ return strings.Join(l, ", ")
+ }
+ defaultSize := func(v string) string {
+ if v == "" {
+ return "the build's size"
+ }
+ return v
+ }
+ if req.EgressAllow != nil {
+ a, b := allow(old.EgressAllow), allow(next.EgressAllow)
+ add("egress allow: "+a+" -> "+b, a != b)
+ }
+ if req.DefaultSize != nil {
+ a, b := defaultSize(old.DefaultSize), defaultSize(next.DefaultSize)
+ add("default size: "+a+" -> "+b, a != b)
+ }
+ if req.Size != nil {
+ host := strings.ToLower(req.Size.Host)
+ size := func(p apppolicy.Policy) string {
+ if o, ok := p.Sizes[host]; ok {
+ return o.Size
+ }
+ return "the app default"
+ }
+ a, b := size(old), size(next)
+ add("size of "+host+": "+a+" -> "+b, a != b)
+ }
+ parts := changed
+ if len(parts) == 0 {
+ parts = sent
+ }
+ return strings.Join(parts, "; ") + " (cli)"
+}
diff --git a/internal/console/apipolicy_test.go b/internal/console/apipolicy_test.go
new file mode 100644
index 000000000..498dcdd80
--- /dev/null
+++ b/internal/console/apipolicy_test.go
@@ -0,0 +1,1034 @@
+package console
+
+import (
+ "bytes"
+ "context"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "io"
+ "net/http"
+ "os"
+ "reflect"
+ "slices"
+ "strings"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/apppolicy"
+ "github.com/carlosframework/platform/internal/lease"
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/release"
+ "github.com/carlosframework/platform/internal/svcrole"
+)
+
+// rawReqHeader is rawReq with request headers: the policy door's write
+// carries its precondition in If-Match or If-None-Match.
+func rawReqHeader(t *testing.T, method, u, bearer, body string, hdr map[string]string) (int, http.Header, string) {
+ t.Helper()
+ var rdr io.Reader
+ if body != "" {
+ rdr = strings.NewReader(body)
+ }
+ req, err := http.NewRequest(method, u, rdr)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if body != "" {
+ req.Header.Set("Content-Type", "application/json")
+ }
+ if bearer != "" {
+ req.Header.Set("Authorization", "Bearer "+bearer)
+ }
+ for k, v := range hdr {
+ req.Header.Set(k, v)
+ }
+ res, err := http.DefaultClient.Do(req)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer res.Body.Close()
+ out, err := io.ReadAll(res.Body)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return res.StatusCode, res.Header, string(out)
+}
+
+// policyReq is one request to an app's policy door on the fixture's account.
+func policyReq(t *testing.T, fx *consoleFixture, method, app, token, body string, hdr map[string]string) (int, http.Header, string) {
+ t.Helper()
+ return rawReqHeader(t, method, fx.ts.URL+"/api/cli/apps/"+fx.sqid+"/"+app+"/policy", token, body, hdr)
+}
+
+var policyCreate = map[string]string{"If-None-Match": "*"}
+
+func policyIfMatch(etag string) map[string]string { return map[string]string{"If-Match": etag} }
+
+// policyFx is the console fixture with an app "shop" in the ops account,
+// whose owner (paul@keymail.dev) is the system operator, and the owner's
+// CLI token. The door's log lines go to logs.
+type policyFx struct {
+ *consoleFixture
+ owner string
+ logs *policyLogRecorder
+}
+
+func newPolicyFxFrom(t *testing.T, fx *consoleFixture, code string) *policyFx {
+ t.Helper()
+ for _, app := range []string{"shop", "shop-b"} {
+ if _, err := fx.st.CreateAppIn(OpsAccountID, app, "paul@keymail.dev", time.Now()); err != nil {
+ t.Fatal(err)
+ }
+ }
+ logs := &policyLogRecorder{}
+ fx.srv.policyLog = logs.logf
+ return &policyFx{consoleFixture: fx, owner: cliToken(t, fx, "paul@keymail.dev", code), logs: logs}
+}
+
+func newPolicyFx(t *testing.T, code string) *policyFx {
+ t.Helper()
+ return newPolicyFxFrom(t, newConsoleFixture(t), code)
+}
+
+// leased gives the fixture a deployment bucket, so its door takes the
+// writer lease from a fake leaser instead of running sole.
+func (fx *policyFx) leased() *fakePolicyLeaser {
+ f := &fakePolicyLeaser{}
+ fx.srv.cfg.Bucket = "deploy"
+ fx.srv.cfg.PolicyLease = f.leaser
+ return f
+}
+
+func lastPolicyLease(t *testing.T, f *fakePolicyLeaser) *fakePolicyLease {
+ t.Helper()
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ if len(f.leases) == 0 {
+ t.Fatal("no lease has been acquired")
+ }
+ return f.leases[len(f.leases)-1]
+}
+
+func (fx *policyFx) put(t *testing.T, app, body string, hdr map[string]string) (int, policyReply, string) {
+ t.Helper()
+ status, h, out := policyReq(t, fx.consoleFixture, http.MethodPut, app, fx.owner, body, hdr)
+ var r policyReply
+ if status == http.StatusOK {
+ if err := json.Unmarshal([]byte(out), &r); err != nil {
+ t.Fatalf("PUT reply %q: %v", out, err)
+ }
+ if h.Get("ETag") != r.ETag {
+ t.Fatalf("ETag header %q, reply etag %q", h.Get("ETag"), r.ETag)
+ }
+ }
+ return status, r, out
+}
+
+// mustPut is put for a write that must land.
+func (fx *policyFx) mustPut(t *testing.T, app, body string, hdr map[string]string) policyReply {
+ t.Helper()
+ status, r, out := fx.put(t, app, body, hdr)
+ if status != http.StatusOK {
+ t.Fatalf("PUT %s %s = %d %s", app, body, status, out)
+ }
+ return r
+}
+
+func (fx *policyFx) get(t *testing.T, app string) policyReply {
+ t.Helper()
+ status, h, out := policyReq(t, fx.consoleFixture, http.MethodGet, app, fx.owner, "", nil)
+ if status != http.StatusOK {
+ t.Fatalf("GET %s = %d %s", app, status, out)
+ }
+ var r policyReply
+ if err := json.Unmarshal([]byte(out), &r); err != nil {
+ t.Fatalf("GET reply %q: %v", out, err)
+ }
+ if h.Get("ETag") != r.ETag {
+ t.Fatalf("ETag header %q, reply etag %q", h.Get("ETag"), r.ETag)
+ }
+ return r
+}
+
+// stored is the object at the app's key, opened with the fixture's key.
+func storedPolicy(t *testing.T, rel release.Store, pub []byte, sqid, app string) (apppolicy.Policy, []byte, string) {
+ t.Helper()
+ p, wire, etag, err := apppolicy.GetForUpdate(context.Background(), rel, pub, sqid, app)
+ if err != nil {
+ t.Fatalf("stored policy %s/%s: %v", sqid, app, err)
+ }
+ return *p, wire, etag
+}
+
+func (fx *policyFx) stored(t *testing.T, app string) (apppolicy.Policy, []byte, string) {
+ t.Helper()
+ return storedPolicy(t, fx.rel, fx.pub, fx.sqid, app)
+}
+
+func (fx *policyFx) noObject(t *testing.T, app string) {
+ t.Helper()
+ if _, _, err := fx.rel.Get(context.Background(), apppolicy.Key(fx.sqid, app)); !errors.Is(err, os.ErrNotExist) {
+ t.Fatalf("the policy object of %s exists (%v)", app, err)
+ }
+}
+
+func (fx *policyFx) row(t *testing.T, app string) AppPolicyRow {
+ t.Helper()
+ row, err := fx.st.AppPolicyRecordContext(context.Background(), OpsAccountID, app)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return row
+}
+
+func (fx *policyFx) policyEvents(t *testing.T, app string) []Event {
+ t.Helper()
+ evs, err := fx.st.Events(OpsAccountID, app, 100)
+ if err != nil {
+ t.Fatal(err)
+ }
+ var out []Event
+ for _, e := range evs {
+ if e.Action == "app-policy" {
+ out = append(out, e)
+ }
+ }
+ return out
+}
+
+func (fx *policyFx) serial(t *testing.T) int64 {
+ t.Helper()
+ doc, _, err := release.ReadSerial(context.Background(), fx.rel)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return doc.Serial
+}
+
+// signPolicy signs p as the fixture's console would, at the key it names.
+func signPolicy(t *testing.T, fx *consoleFixture, p apppolicy.Policy) []byte {
+ t.Helper()
+ wire, err := apppolicy.Sign(fx.priv, p)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return wire
+}
+
+// putAtKey writes bytes at the app's key directly, as a bucket writer or
+// another console would, unconditionally.
+func putAtKey(t *testing.T, rel release.Store, sqid, app string, wire []byte) string {
+ t.Helper()
+ ctx := context.Background()
+ _, etag, err := rel.Get(ctx, apppolicy.Key(sqid, app))
+ if err != nil && !errors.Is(err, os.ErrNotExist) {
+ t.Fatal(err)
+ }
+ newETag, err := rel.Put(ctx, apppolicy.Key(sqid, app), wire, etag)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return newETag
+}
+
+func policyRecord(t *testing.T, rel release.Store, sqid, app, host string, created time.Time) {
+ t.Helper()
+ rec := provision.Record{Schema: 1, Host: host, Slug: strings.Split(host, ".")[0], App: app, Account: sqid,
+ Channel: "stable", CreatedAt: created}
+ if err := provision.CreateRecord(context.Background(), rel, rec); err != nil {
+ t.Fatal(err)
+ }
+}
+
+func TestPolicyDoorsAreForSystemOperatorsOnly(t *testing.T) {
+ t.Parallel()
+ fx := newPolicyFx(t, "code-policy-op")
+ if err := fx.st.AddMember(OpsAccountID, "member@keymail.dev", RoleMember, time.Now()); err != nil {
+ t.Fatal(err)
+ }
+ member := cliToken(t, fx.consoleFixture, "member@keymail.dev", "code-policy-member")
+ _, want := getRaw(t, fx.ts.URL+"/api/cli/system/no-such-door", member)
+ body := `{"default_size":"small"}`
+ for _, method := range []string{http.MethodGet, http.MethodPut} {
+ if st, _, out := policyReq(t, fx.consoleFixture, method, "shop", member, body, policyCreate); st != http.StatusNotFound || out != want {
+ t.Errorf("%s as a member = %d %q, want the bare 404", method, st, out)
+ }
+ }
+ for _, svc := range []string{
+ serviceToken(t, fx.st, OpsAccountID, "svc-admin", "shop", svcrole.Admin),
+ serviceToken(t, fx.st, OpsAccountID, "svc-instance", "", svcrole.Instance),
+ } {
+ for _, method := range []string{http.MethodGet, http.MethodPut} {
+ if st, _, out := policyReq(t, fx.consoleFixture, method, "shop", svc, body, policyCreate); st != http.StatusForbidden {
+ t.Errorf("%s as a service credential = %d %q, want 403", method, st, out)
+ }
+ }
+ }
+ for _, path := range []string{"/api/cli/apps/bab/nosuch/policy", "/api/cli/apps/zzz/shop/policy"} {
+ for _, method := range []string{http.MethodGet, http.MethodPut} {
+ if st, _, out := rawReqHeader(t, method, fx.ts.URL+path, fx.owner, body, policyCreate); st != http.StatusNotFound || out != want {
+ t.Errorf("%s %s as the operator = %d %q, want the bare 404", method, path, st, out)
+ }
+ }
+ }
+ fx.noObject(t, "shop")
+ if evs := fx.policyEvents(t, "shop"); len(evs) != 0 {
+ t.Fatalf("refused callers recorded %d events", len(evs))
+ }
+}
+
+func TestPolicyPutSetsAuditsAndBumps(t *testing.T) {
+ t.Parallel()
+ fx := newPolicyFx(t, "code-policy-set")
+ before := fx.serial(t)
+ status, h, out := policyReq(t, fx.consoleFixture, http.MethodPut, "shop", fx.owner,
+ `{"egress_allow":["fd00:5::20","10.0.5.20","10.0.5.20"],"default_size":"large"}`, policyCreate)
+ if status != http.StatusOK {
+ t.Fatalf("PUT = %d %s", status, out)
+ }
+ if !strings.Contains(out, `"not_checked":[]`) {
+ t.Fatalf("reply %s has no empty not_checked", out)
+ }
+ var r policyReply
+ if err := json.Unmarshal([]byte(out), &r); err != nil {
+ t.Fatal(err)
+ }
+ if want := []string{"10.0.5.20", "fd00:5::20"}; !slices.Equal(r.Policy.EgressAllow, want) || r.Policy.DefaultSize != "large" {
+ t.Fatalf("reply policy %+v", r.Policy)
+ }
+ stored, _, etag := fx.stored(t, "shop")
+ if !reflect.DeepEqual(stored, r.Policy) {
+ t.Fatalf("stored %+v, reply %+v", stored, r.Policy)
+ }
+ if r.ETag == "" || r.ETag != h.Get("ETag") || r.ETag != etag {
+ t.Fatalf("reply etag %q, header %q, stored %q", r.ETag, h.Get("ETag"), etag)
+ }
+ if got := fx.serial(t); got != before+1 {
+ t.Fatalf("serial %d, want %d", got, before+1)
+ }
+ evs := fx.policyEvents(t, "shop")
+ if len(evs) != 1 || !strings.Contains(evs[0].Detail, "egress allow: none -> 10.0.5.20, fd00:5::20") ||
+ !strings.Contains(evs[0].Detail, "default size: the build's size -> large") || !strings.HasSuffix(evs[0].Detail, " (cli)") {
+ t.Fatalf("events %+v", evs)
+ }
+}
+
+func TestPolicyPutIsCompareAndSwap(t *testing.T) {
+ t.Parallel()
+ fx := newPolicyFx(t, "code-policy-cas")
+ body := `{"default_size":"small"}`
+ if st, _, out := fx.put(t, "shop", body, nil); st != http.StatusPreconditionRequired ||
+ errorOf(t, out) != "send If-Match with the policy's etag, or If-None-Match: * when there is none" {
+ t.Fatalf("no precondition = %d %s", st, out)
+ }
+ if st, _, out := fx.put(t, "shop", body, policyIfMatch("stale")); st != http.StatusPreconditionFailed ||
+ errorOf(t, out) != "the policy changed since it was read. Read it again and retry" {
+ t.Fatalf("If-Match stale = %d %s", st, out)
+ }
+ fx.noObject(t, "shop")
+ first := fx.mustPut(t, "shop", body, policyCreate)
+ if st, _, out := fx.put(t, "shop", `{"default_size":"large"}`, policyCreate); st != http.StatusPreconditionFailed {
+ t.Fatalf("If-None-Match * over an object = %d %s", st, out)
+ }
+ fx.mustPut(t, "shop", `{"egress_allow":["10.0.5.20"]}`, policyIfMatch(first.ETag))
+ if st, _, out := fx.put(t, "shop", `{"default_size":"large"}`, policyIfMatch(first.ETag)); st != http.StatusPreconditionFailed {
+ t.Fatalf("the second PUT on one etag = %d %s", st, out)
+ }
+ if got, _, _ := fx.stored(t, "shop"); got.DefaultSize != "small" || !slices.Equal(got.EgressAllow, []string{"10.0.5.20"}) || got.Revision != 2 {
+ t.Fatalf("stored %+v: the losing PUT's change landed", got)
+ }
+}
+
+func TestPolicyPutRefusesBeforeWriting(t *testing.T) {
+ t.Parallel()
+ fx := newPolicyFx(t, "code-policy-refuse")
+ nine := make([]string, 9)
+ for i := range nine {
+ nine[i] = fmt.Sprintf(`"10.0.5.%d"`, i+1)
+ }
+ for _, c := range []struct {
+ body string
+ status int
+ msg string
+ }{
+ {`{}`, http.StatusBadRequest, "nothing to set: send egress_allow, default_size or size"},
+ {`{"egress_allow":null}`, http.StatusBadRequest, "nothing to set: send egress_allow, default_size or size"},
+ {`{"egress_allow":["10.0.5.20:5432"]}`, http.StatusBadRequest, ""},
+ {`{"egress_allow":["8.8.8.8"]}`, http.StatusBadRequest, ""},
+ {`{"egress_allow":[` + strings.Join(nine, ",") + `]}`, http.StatusBadRequest, ""},
+ {`{"default_size":"huge"}`, http.StatusBadRequest, "default_size must be small, standard, large or xlarge, or empty for the build's size"},
+ {`{"size":{"host":"shop-a.example.com","value":"huge"}}`, http.StatusBadRequest, "size must be small, standard, large, xlarge or default"},
+ {`{"size":{"host":"../x","value":"large"}}`, http.StatusBadRequest, "invalid host"},
+ {`{"egress_alow":[]}`, http.StatusBadRequest, ""},
+ {`{"size":{"host":"shop-a.example.com","value":"large"}}`, http.StatusNotFound, "no instance shop-a.example.com in this app"},
+ } {
+ st, _, out := fx.put(t, "shop", c.body, policyCreate)
+ if st != c.status || (c.msg != "" && errorOf(t, out) != c.msg) {
+ t.Errorf("%s = %d %s, want %d %q", c.body, st, out, c.status, c.msg)
+ }
+ }
+ fx.noObject(t, "shop")
+ if evs := fx.policyEvents(t, "shop"); len(evs) != 0 {
+ t.Fatalf("refused writes recorded events: %+v", evs)
+ }
+ if row := fx.row(t, "shop"); row.Revision != 0 {
+ t.Fatalf("refused writes recorded a row: %+v", row)
+ }
+}
+
+func TestPolicyOverrideCapturesTheRecord(t *testing.T) {
+ t.Parallel()
+ fx := newPolicyFx(t, "code-policy-override")
+ t0 := time.Now().UTC().Add(-time.Hour).Truncate(time.Second)
+ policyRecord(t, fx.rel, fx.sqid, "shop", "shop-a.example.com", t0)
+ r := fx.mustPut(t, "shop", `{"size":{"host":"SHOP-A.example.com","value":"xlarge"}}`, policyCreate)
+ got, ok := r.Policy.Sizes["shop-a.example.com"]
+ if !ok || got.Size != "xlarge" || !got.CreatedAt.Equal(t0) || len(r.Policy.Sizes) != 1 {
+ t.Fatalf("sizes %+v, want shop-a.example.com {xlarge %s}", r.Policy.Sizes, t0)
+ }
+ if stored, _, _ := fx.stored(t, "shop"); !reflect.DeepEqual(stored.Sizes, r.Policy.Sizes) {
+ t.Fatalf("stored sizes %+v", stored.Sizes)
+ }
+ r = fx.mustPut(t, "shop", `{"size":{"host":"shop-a.example.com","value":"default"}}`, policyIfMatch(r.ETag))
+ if len(r.Policy.Sizes) != 0 {
+ t.Fatalf("sizes %+v after default, want none", r.Policy.Sizes)
+ }
+ if evs := fx.policyEvents(t, "shop"); len(evs) != 2 || !strings.Contains(evs[0].Detail+evs[1].Detail, "size of shop-a.example.com: xlarge -> the app default") {
+ t.Fatalf("events %+v", evs)
+ }
+}
+
+func TestPolicyGetMarksAndPutDropsStaleOverrides(t *testing.T) {
+ t.Parallel()
+ fx := newPolicyFx(t, "code-policy-stale")
+ ctx := context.Background()
+ t0 := time.Now().UTC().Add(-time.Hour).Truncate(time.Second)
+ policyRecord(t, fx.rel, fx.sqid, "shop", "shop-a.example.com", t0)
+ fx.mustPut(t, "shop", `{"size":{"host":"shop-a.example.com","value":"large"}}`, policyCreate)
+ if r := fx.get(t, "shop"); len(r.Stale) != 0 {
+ t.Fatalf("stale %v while the record is the one", r.Stale)
+ }
+ if err := provision.DeleteRecord(ctx, fx.rel, fx.sqid, "shop", "shop-a.example.com"); err != nil {
+ t.Fatal(err)
+ }
+ policyRecord(t, fx.rel, fx.sqid, "shop", "shop-a.example.com", t0.Add(time.Minute))
+ r := fx.get(t, "shop")
+ if !slices.Equal(r.Stale, []string{"shop-a.example.com"}) {
+ t.Fatalf("stale_overrides %v", r.Stale)
+ }
+ fx.mustPut(t, "shop", `{"default_size":"small"}`, policyIfMatch(r.ETag))
+ if stored, _, _ := fx.stored(t, "shop"); len(stored.Sizes) != 0 || stored.DefaultSize != "small" {
+ t.Fatalf("stored %+v, want the stale entry gone", stored)
+ }
+}
+
+// Astra plan R8-3: only the audit insertion fails. The reads before it
+// (apiSystemRoller's Role, AccountBySqid, HasAppIn) use the same database,
+// so closing it would never reach the audit.
+func TestPolicyPutRefusesWhenTheAuditFails(t *testing.T) {
+ t.Parallel()
+ fx := newPolicyFx(t, "code-policy-audit")
+ first := fx.mustPut(t, "shop", `{"default_size":"small"}`, policyCreate)
+ _, wire, etag := fx.stored(t, "shop")
+ row := fx.row(t, "shop")
+ if row.Revision != 1 || !bytes.Equal(row.Wire, wire) || row.SyncedETag != etag {
+ t.Fatalf("row %+v", row)
+ }
+ serial := fx.serial(t)
+ if _, err := fx.st.db.Exec(`CREATE TRIGGER fail_policy_audit BEFORE INSERT ON audit WHEN NEW.action = 'app-policy'
+BEGIN SELECT RAISE(ABORT, 'test audit failure'); END`); err != nil {
+ t.Fatal(err)
+ }
+ st, _, out := fx.put(t, "shop", `{"default_size":"large"}`, policyIfMatch(first.ETag))
+ if st != http.StatusServiceUnavailable || errorOf(t, out) != policyAuditFailedMsg {
+ t.Fatalf("PUT with the audit failing = %d %s, want 503 %q", st, out, policyAuditFailedMsg)
+ }
+ if got, gotWire, gotETag := fx.stored(t, "shop"); got.DefaultSize != "small" || !bytes.Equal(gotWire, wire) || gotETag != etag {
+ t.Fatalf("the object changed: %+v", got)
+ }
+ if got := fx.row(t, "shop"); got.Revision != 1 || !bytes.Equal(got.Wire, wire) || got.SyncedETag != etag {
+ t.Fatalf("the row changed: %+v", got)
+ }
+ if got := fx.serial(t); got != serial {
+ t.Fatalf("serial %d, want %d", got, serial)
+ }
+ if evs := fx.policyEvents(t, "shop"); len(evs) != 1 {
+ t.Fatalf("%d app-policy events, want 1", len(evs))
+ }
+ if _, err := fx.st.db.Exec(`DROP TRIGGER fail_policy_audit`); err != nil {
+ t.Fatal(err)
+ }
+ r := fx.mustPut(t, "shop", `{"default_size":"large"}`, policyIfMatch(first.ETag))
+ if r.Policy.Revision != 2 {
+ t.Fatalf("revision %d after the refused write, want 2", r.Policy.Revision)
+ }
+}
+
+func TestPolicyPutSignsOneRevisionAboveTheCurrent(t *testing.T) {
+ t.Parallel()
+ fx := newPolicyFx(t, "code-policy-revision")
+ ctx := context.Background()
+ start := time.Now()
+ r := fx.mustPut(t, "shop", `{"egress_allow":["10.0.5.20"]}`, policyCreate)
+ end := time.Now()
+ _, wire, etag := fx.stored(t, "shop")
+ if _, err := apppolicy.Verify(fx.pub, wire, time.Now()); err != nil {
+ t.Fatal(err)
+ }
+ p, err := apppolicy.VerifyKey(fx.pub, wire, time.Now(), fx.sqid, "shop")
+ if err != nil {
+ t.Fatal(err)
+ }
+ if p.Revision != 1 || p.SignedAt.Before(start.Add(-time.Second)) || p.SignedAt.After(end.Add(time.Second)) {
+ t.Fatalf("revision %d signed %s, want 1 within [%s, %s]", p.Revision, p.SignedAt, start, end)
+ }
+ if row := fx.row(t, "shop"); row.Revision != 1 || !bytes.Equal(row.Wire, wire) || row.SyncedETag != etag || row.Pending {
+ t.Fatalf("row %+v", row)
+ }
+ r = fx.mustPut(t, "shop", `{"default_size":"small"}`, policyIfMatch(r.ETag))
+ if r.Policy.Revision != 2 {
+ t.Fatalf("second revision %d", r.Policy.Revision)
+ }
+
+ at7 := r.Policy
+ at7.Revision = 7
+ if err := fx.st.RecordAppPolicyContext(ctx, OpsAccountID, "shop", 7, signPolicy(t, fx.consoleFixture, at7)); err != nil {
+ t.Fatal(err)
+ }
+ r = fx.mustPut(t, "shop", `{"default_size":"large"}`, policyIfMatch(r.ETag))
+ if r.Policy.Revision != 8 {
+ t.Fatalf("after a recorded 7, revision %d, want 8", r.Policy.Revision)
+ }
+
+ at9 := r.Policy
+ at9.Revision, at9.SignedAt, at9.EgressAllow = 9, time.Now().UTC(), nil
+ etag9 := putAtKey(t, fx.rel, fx.sqid, "shop", signPolicy(t, fx.consoleFixture, at9))
+ r = fx.mustPut(t, "shop", `{"default_size":"xlarge"}`, policyIfMatch(etag9))
+ if r.Policy.Revision != 10 || len(r.Policy.EgressAllow) != 0 {
+ t.Fatalf("after a stored 9, policy %+v, want revision 10 built on it", r.Policy)
+ }
+ if row := fx.row(t, "shop"); row.Revision != 10 {
+ t.Fatalf("row revision %d, want 10", row.Revision)
+ }
+}
+
+func TestAppPolicyRevisionsAreNeverReused(t *testing.T) {
+ t.Parallel()
+ fx := newConsoleFixture(t)
+ st, ctx := fx.st, context.Background()
+ want := func(app string, rev uint64, wire, etag string, pending bool) {
+ t.Helper()
+ row, err := st.AppPolicyRecordContext(ctx, OpsAccountID, app)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if row.Revision != rev || string(row.Wire) != wire || row.SyncedETag != etag || row.Pending != pending {
+ t.Fatalf("row %+v, want (%d, %q, %q, pending %v)", row, rev, wire, etag, pending)
+ }
+ }
+ listed := func(app string) bool {
+ t.Helper()
+ rows, err := st.AppPolicyRowsContext(ctx)
+ if err != nil {
+ t.Fatal(err)
+ }
+ for _, r := range rows {
+ if r.App == app {
+ if r.Sqid != release.OpsAccount || r.AccountID != OpsAccountID {
+ t.Fatalf("listed row %+v without its account's sqid", r)
+ }
+ return true
+ }
+ }
+ return false
+ }
+ if row, err := st.AppPolicyRecordContext(ctx, OpsAccountID, "shop"); err != nil || row.Revision != 0 {
+ t.Fatalf("no row: %+v %v", row, err)
+ }
+ if err := st.RecordAppPolicyContext(ctx, OpsAccountID, "shop", 2, []byte("w2")); err != nil {
+ t.Fatal(err)
+ }
+ for _, c := range []struct {
+ rev uint64
+ wire string
+ }{{2, "w2'"}, {1, "w1"}} {
+ if err := st.RecordAppPolicyContext(ctx, OpsAccountID, "shop", c.rev, []byte(c.wire)); !errors.Is(err, errAppPolicyRevisionUsed) {
+ t.Fatalf("record %d: %v, want errAppPolicyRevisionUsed", c.rev, err)
+ }
+ }
+ want("shop", 2, "w2", "", true)
+ if err := st.ResignAppPolicyContext(ctx, OpsAccountID, "shop", 2, []byte("w2 re-signed")); err != nil {
+ t.Fatal(err)
+ }
+ if err := st.ResignAppPolicyContext(ctx, OpsAccountID, "shop", 1, []byte("x")); !errors.Is(err, errAppPolicyRevisionMoved) {
+ t.Fatalf("resign 1: %v", err)
+ }
+ want("shop", 2, "w2 re-signed", "", true)
+ if err := st.AdoptAppPolicyContext(ctx, OpsAccountID, "shop", 1, []byte("w1''"), "e9"); err != nil {
+ t.Fatal(err)
+ }
+ want("shop", 2, "w1''", "e9", false)
+ if err := st.SyncedAppPolicyContext(ctx, OpsAccountID, "shop", 2, "e10"); err != nil {
+ t.Fatal(err)
+ }
+ if err := st.SyncedAppPolicyContext(ctx, OpsAccountID, "shop", 1, "e11"); !errors.Is(err, errAppPolicyRevisionMoved) {
+ t.Fatalf("synced 1: %v", err)
+ }
+ want("shop", 2, "w1''", "e10", false)
+ if !listed("shop") {
+ t.Fatal("a synced row is not listed")
+ }
+
+ // Astra plan R4-2: pending tells an unconfirmed publication from a
+ // confirmed absence.
+ if err := st.RecordAppPolicyContext(ctx, OpsAccountID, "shop-b", 1, []byte("b1")); err != nil {
+ t.Fatal(err)
+ }
+ want("shop-b", 1, "b1", "", true)
+ if !listed("shop-b") {
+ t.Fatal("a pending row is not listed")
+ }
+ if err := st.SyncedAppPolicyContext(ctx, OpsAccountID, "shop-b", 1, "e1"); err != nil {
+ t.Fatal(err)
+ }
+ want("shop-b", 1, "b1", "e1", false)
+ if err := st.ResignAppPolicyContext(ctx, OpsAccountID, "shop-b", 1, []byte("b1'")); err != nil {
+ t.Fatal(err)
+ }
+ want("shop-b", 1, "b1'", "e1", true)
+ if err := st.SyncedAppPolicyContext(ctx, OpsAccountID, "shop-b", 1, ""); err != nil {
+ t.Fatal(err)
+ }
+ want("shop-b", 1, "b1'", "", false)
+ if listed("shop-b") {
+ t.Fatal("a row found absent is still listed")
+ }
+ if err := st.AdoptAppPolicyContext(ctx, OpsAccountID, "shop-c", 4, []byte("c4"), "e4"); err != nil {
+ t.Fatal(err)
+ }
+ want("shop-c", 4, "c4", "e4", false)
+}
+
+// Ruling 10's takeover, with Task 8's fake leaser.
+func TestPolicyBaseRebasesOnANewHolding(t *testing.T) {
+ t.Parallel()
+ fx := newPolicyFx(t, "code-policy-rebase")
+ f := fx.leased()
+ ctx := context.Background()
+
+ // Holding 1.
+ r := fx.mustPut(t, "shop", `{"egress_allow":["10.0.5.20"]}`, policyCreate)
+ r = fx.mustPut(t, "shop", `{"default_size":"small"}`, policyIfMatch(r.ETag))
+ if r.Policy.Revision != 2 || fx.row(t, "shop").SyncedETag != r.ETag {
+ t.Fatalf("holding 1 left %+v, row %+v", r.Policy, fx.row(t, "shop"))
+ }
+ // Another console's record, at the same revision, with other content.
+ other := apppolicy.Policy{V: 1, Account: fx.sqid, App: "shop", Revision: 2, SignedAt: time.Now().UTC(), DefaultSize: "large"}
+ otherETag := putAtKey(t, fx.rel, fx.sqid, "shop", signPolicy(t, fx.consoleFixture, other))
+
+ // Still holding 1: the recorded content, not the bucket's.
+ g := fx.get(t, "shop")
+ if g.Policy.Revision != 2 || !slices.Equal(g.Policy.EgressAllow, []string{"10.0.5.20"}) || g.Policy.DefaultSize != "small" || g.ETag != otherETag {
+ t.Fatalf("GET within holding 1 = %+v etag %q", g.Policy, g.ETag)
+ }
+ fx.logs.has(t, "console: app policy bab/shop: not building on the stored object")
+
+ // The lease is lost: the next holding builds on the bucket's record.
+ lastPolicyLease(t, f).set(errors.New("renewal lost"), time.Now().Add(10*time.Minute), nil)
+ g = fx.get(t, "shop")
+ if len(g.Policy.EgressAllow) != 0 || g.Policy.DefaultSize != "large" || g.ETag != otherETag {
+ t.Fatalf("GET as a new holding = %+v", g.Policy)
+ }
+ if row := fx.row(t, "shop"); row.SyncedETag == otherETag {
+ t.Fatal("a GET adopted the bucket's record")
+ }
+ r = fx.mustPut(t, "shop", `{"default_size":"xlarge"}`, policyIfMatch(g.ETag))
+ if f.callCount() != 2 {
+ t.Fatalf("%d lease acquisitions, want 2", f.callCount())
+ }
+ fx.logs.has(t, "console: app policy bab/shop: new writer holding; building on the bucket's record (revision 2)")
+ stored, wire, etag := fx.stored(t, "shop")
+ if stored.Revision != 3 || len(stored.EgressAllow) != 0 || stored.DefaultSize != "xlarge" {
+ t.Fatalf("stored %+v, want revision 3 built on the other console's record", stored)
+ }
+ if row := fx.row(t, "shop"); row.Revision != 3 || !bytes.Equal(row.Wire, wire) || row.SyncedETag != etag {
+ t.Fatalf("row %+v", row)
+ }
+
+ // A publication interrupted across holdings is finished, never rebased:
+ // revision 4 recorded, the bucket still at this console's revision 3.
+ at4 := stored
+ at4.Revision, at4.SignedAt, at4.EgressAllow = 4, time.Now().UTC(), []string{"10.0.5.20"}
+ if err := fx.st.RecordAppPolicyContext(ctx, OpsAccountID, "shop", 4, signPolicy(t, fx.consoleFixture, at4)); err != nil {
+ t.Fatal(err)
+ }
+ lastPolicyLease(t, f).set(errors.New("renewal lost"), time.Now().Add(10*time.Minute), nil)
+ g = fx.get(t, "shop")
+ if g.Policy.Revision != 4 || !slices.Equal(g.Policy.EgressAllow, []string{"10.0.5.20"}) {
+ t.Fatalf("GET across holdings = %+v, want the recorded revision 4", g.Policy)
+ }
+
+ // A sole console never rebases, even with nothing marked (a restart).
+ sole := newPolicyFx(t, "code-policy-rebase-sole")
+ r = sole.mustPut(t, "shop", `{"egress_allow":["10.0.5.20"]}`, policyCreate)
+ sole.mustPut(t, "shop", `{"default_size":"small"}`, policyIfMatch(r.ETag))
+ putAtKey(t, sole.rel, sole.sqid, "shop", signPolicy(t, sole.consoleFixture, other))
+ sole.srv.policySyncedMu.Lock()
+ sole.srv.policySynced = nil
+ sole.srv.policySyncedMu.Unlock()
+ g = sole.get(t, "shop")
+ if g.Policy.Revision != 2 || !slices.Equal(g.Policy.EgressAllow, []string{"10.0.5.20"}) || g.Policy.DefaultSize != "small" {
+ t.Fatalf("sole GET = %+v, want the recorded revision 2", g.Policy)
+ }
+}
+
+func TestPolicyDoorRefusesWhileAnotherConsoleHoldsTheLease(t *testing.T) {
+ t.Parallel()
+ fx := newPolicyFx(t, "code-policy-held")
+ fx.srv.cfg.Bucket = "deploy"
+ f := &fakePolicyLeaser{err: fmt.Errorf("acquire lease: %w",
+ &lease.HeldError{Owner: "console-b:4242", ExpiresAt: time.Now().Add(10 * time.Minute)})}
+ fx.srv.cfg.PolicyLease = f.leaser
+ st, _, out := fx.put(t, "shop", `{"egress_allow":["10.0.5.20"]}`, policyCreate)
+ if st != http.StatusServiceUnavailable || !strings.Contains(errorOf(t, out), "another console (console-b:4242)") {
+ t.Fatalf("PUT = %d %s, want 503 naming console-b:4242", st, out)
+ }
+ fx.noObject(t, "shop")
+ if evs := fx.policyEvents(t, "shop"); len(evs) != 0 {
+ t.Fatalf("events %+v", evs)
+ }
+ if row := fx.row(t, "shop"); row.Revision != 0 {
+ t.Fatalf("row %+v", row)
+ }
+ if g := fx.get(t, "shop"); g.Policy.V != 1 || g.ETag != "" || g.Policy.Revision != 0 {
+ t.Fatalf("GET = %+v", g)
+ }
+}
+
+// appPolicyHookStore runs hook on each Put of one policy key, before the
+// write: a hook that returns handled answers the Put itself.
+type appPolicyHookStore struct {
+ release.Store
+ key string
+ mu sync.Mutex
+ puts int
+ hook func(n int, body []byte, ifMatch string) (handled bool, etag string, err error)
+}
+
+func (s *appPolicyHookStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
+ if key != s.key {
+ return s.Store.Put(ctx, key, body, ifMatch)
+ }
+ s.mu.Lock()
+ s.puts++
+ n, hook := s.puts, s.hook
+ s.mu.Unlock()
+ if hook != nil {
+ if handled, etag, err := hook(n, body, ifMatch); handled {
+ return etag, err
+ }
+ }
+ return s.Store.Put(ctx, key, body, ifMatch)
+}
+
+func (s *appPolicyHookStore) putCount() int {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ return s.puts
+}
+
+// Astra plan R3-1, the retry half.
+func TestPolicyPublishRechecksTheHoldingOnEveryAttempt(t *testing.T) {
+ t.Parallel()
+ conflict := fmt.Errorf("test: %w", release.ErrCASConflict)
+ setup := func(t *testing.T, code string) (*policyFx, *fakePolicyLeaser, *appPolicyHookStore) {
+ fx := newPolicyFx(t, code)
+ f := fx.leased()
+ hs := &appPolicyHookStore{Store: fx.rel, key: apppolicy.Key(fx.sqid, "shop")}
+ fx.srv.cfg.Releases = hs
+ return fx, f, hs
+ }
+ wantLost := func(t *testing.T, st int, out string) {
+ t.Helper()
+ if st != http.StatusServiceUnavailable || errorOf(t, out) != policyLeaseLostMsg {
+ t.Fatalf("PUT = %d %s, want 503 %q", st, out, policyLeaseLostMsg)
+ }
+ }
+
+ t.Run("the lease fails during a conflicting attempt", func(t *testing.T) {
+ t.Parallel()
+ fx, f, hs := setup(t, "code-policy-attempt-err")
+ hs.hook = func(n int, _ []byte, _ string) (bool, string, error) {
+ if n == 1 {
+ lastPolicyLease(t, f).set(errors.New("renewal lost"), time.Now().Add(10*time.Minute), nil)
+ return true, "", conflict
+ }
+ return false, "", nil
+ }
+ st, _, out := fx.put(t, "shop", `{"egress_allow":["10.0.5.20"]}`, policyCreate)
+ wantLost(t, st, out)
+ if hs.putCount() != 1 {
+ t.Fatalf("%d Puts reached the store, want 1", hs.putCount())
+ }
+ if row := fx.row(t, "shop"); row.Revision != 1 || !row.Pending {
+ t.Fatalf("row %+v, want the recorded revision 1, pending", row)
+ }
+ fx.noObject(t, "shop")
+ })
+
+ t.Run("the safe deadline is too near before the write", func(t *testing.T) {
+ t.Parallel()
+ fx, f, hs := setup(t, "code-policy-attempt-near")
+ r := fx.mustPut(t, "shop", `{"default_size":"small"}`, policyCreate)
+ lastPolicyLease(t, f).set(nil, time.Now().Add(8*time.Second), nil)
+ st, _, out := fx.put(t, "shop", `{"default_size":"large"}`, policyIfMatch(r.ETag))
+ wantLost(t, st, out)
+ if hs.putCount() != 1 {
+ t.Fatalf("%d Puts, want only the first PUT's", hs.putCount())
+ }
+ if row := fx.row(t, "shop"); row.Revision != 1 {
+ t.Fatalf("row %+v: the refused write recorded", row)
+ }
+ if evs := fx.policyEvents(t, "shop"); len(evs) != 1 {
+ t.Fatalf("%d events, want only the first PUT's", len(evs))
+ }
+ })
+
+ t.Run("the safe deadline moves in during a conflicting attempt", func(t *testing.T) {
+ t.Parallel()
+ fx, f, hs := setup(t, "code-policy-attempt-moved")
+ hs.hook = func(n int, _ []byte, _ string) (bool, string, error) {
+ if n == 1 {
+ lastPolicyLease(t, f).set(nil, time.Now().Add(8*time.Second), nil)
+ return true, "", conflict
+ }
+ return false, "", nil
+ }
+ st, _, out := fx.put(t, "shop", `{"egress_allow":["10.0.5.20"]}`, policyCreate)
+ wantLost(t, st, out)
+ if hs.putCount() != 1 {
+ t.Fatalf("%d Puts, want 1", hs.putCount())
+ }
+ })
+
+ t.Run("another writer's record is never overwritten", func(t *testing.T) {
+ t.Parallel()
+ fx, _, hs := setup(t, "code-policy-attempt-other")
+ theirs := signPolicy(t, fx.consoleFixture, apppolicy.Policy{V: 1, Account: fx.sqid, App: "shop", Revision: 1,
+ SignedAt: time.Now().UTC(), DefaultSize: "large"})
+ hs.hook = func(n int, _ []byte, _ string) (bool, string, error) {
+ if n == 1 {
+ if _, err := fx.rel.Put(context.Background(), apppolicy.Key(fx.sqid, "shop"), theirs, ""); err != nil {
+ t.Error(err)
+ }
+ return true, "", conflict
+ }
+ return false, "", nil
+ }
+ st, _, out := fx.put(t, "shop", `{"egress_allow":["10.0.5.20"]}`, policyCreate)
+ wantLost(t, st, out)
+ if hs.putCount() != 1 {
+ t.Fatalf("%d Puts, want 1", hs.putCount())
+ }
+ if _, wire, _ := fx.stored(t, "shop"); !bytes.Equal(wire, theirs) {
+ t.Fatal("the other writer's record was overwritten")
+ }
+ })
+
+ t.Run("control: a plain conflict is retried", func(t *testing.T) {
+ t.Parallel()
+ fx, _, hs := setup(t, "code-policy-attempt-control")
+ hs.hook = func(n int, _ []byte, _ string) (bool, string, error) {
+ if n == 1 {
+ return true, "", conflict
+ }
+ return false, "", nil
+ }
+ r := fx.mustPut(t, "shop", `{"egress_allow":["10.0.5.20"]}`, policyCreate)
+ if hs.putCount() != 2 || r.Policy.Revision != 1 {
+ t.Fatalf("%d Puts, revision %d; want 2 and 1", hs.putCount(), r.Policy.Revision)
+ }
+ if row := fx.row(t, "shop"); row.Pending || row.SyncedETag != r.ETag {
+ t.Fatalf("row %+v", row)
+ }
+ })
+}
+
+// Astra plan R1-4.
+func TestOpenGatedClassifiesAppPolicyRevisions(t *testing.T) {
+ t.Parallel()
+ s := openGatedStore(t)
+ present, err := claimSQLCensus(s.bootDB(), claimSQLTables)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if spec, ok := present["app_policy_revisions"]; !ok || spec.NonApp != "" || spec.Account != "account_id" || spec.App != "app" {
+ t.Fatalf("app_policy_revisions classified %+v (present %v), want gated by account_id and app", spec, ok)
+ }
+ _, err = s.db.Exec(`INSERT INTO app_policy_revisions (account_id, app, revision, wire) VALUES (7, 'shop', 1, x'00')`)
+ requireDenied(t, "outer insert into app_policy_revisions", err)
+
+ cx := newClaimCase(t)
+ if err := cx.st.RecordAppPolicyContext(context.Background(), OpsAccountID, "titogo", 1, []byte("w1")); err != nil {
+ t.Fatalf("RecordAppPolicyContext through the admission seam: %v", err)
+ }
+ if row, err := cx.st.AppPolicyRecordContext(context.Background(), OpsAccountID, "titogo"); err != nil || row.Revision != 1 {
+ t.Fatalf("row %+v %v", row, err)
+ }
+}
+
+func TestPolicyPutDoesNotBuildOnAReplayedRecord(t *testing.T) {
+ t.Parallel()
+ fx := newPolicyFx(t, "code-policy-replay")
+ r := fx.mustPut(t, "shop", `{"egress_allow":["10.0.5.20"]}`, policyCreate)
+ _, w1, _ := fx.stored(t, "shop")
+ fx.mustPut(t, "shop", `{"egress_allow":[]}`, policyIfMatch(r.ETag))
+ etag := putAtKey(t, fx.rel, fx.sqid, "shop", w1)
+ g := fx.get(t, "shop")
+ if len(g.Policy.EgressAllow) != 0 || g.ETag != etag {
+ t.Fatalf("GET over a replayed record = %+v", g.Policy)
+ }
+ 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)
+ }
+}
+
+func TestPolicyPutRepairsAnInvalidObject(t *testing.T) {
+ t.Parallel()
+ fx := newPolicyFx(t, "code-policy-repair")
+ for app, seed := range map[string]string{"shop": `{`, "shop-b": `{"v":1,"egress_allow":["10.0.5.20"]}`} {
+ etag := putAtKey(t, fx.rel, fx.sqid, app, []byte(seed))
+ g := fx.get(t, app)
+ if g.ETag != etag || g.Policy.V != 1 || len(g.Policy.EgressAllow) != 0 || g.Policy.Revision != 0 {
+ t.Fatalf("%s: GET over %q = %+v etag %q", app, seed, g.Policy, g.ETag)
+ }
+ fx.mustPut(t, app, `{"default_size":"small"}`, policyIfMatch(etag))
+ _, wire, _ := fx.stored(t, app)
+ p, err := apppolicy.VerifyKey(fx.pub, wire, time.Now(), fx.sqid, app)
+ if err != nil || len(p.EgressAllow) != 0 || p.Revision != 1 {
+ t.Fatalf("%s: repaired object %+v %v", app, p, err)
+ }
+ }
+}
+
+func TestPolicyDoorRefusesWithoutASigningKey(t *testing.T) {
+ t.Parallel()
+ fx := newPolicyFxFrom(t, newConsoleFixtureKeyed(t, false), "code-policy-nokey")
+ st, _, out := fx.put(t, "shop", `{"default_size":"small"}`, policyCreate)
+ if st != http.StatusServiceUnavailable || errorOf(t, out) != doorKeyReadOnlyMsg {
+ t.Fatalf("PUT = %d %s, want 503 %q", st, out, doorKeyReadOnlyMsg)
+ }
+ fx.noObject(t, "shop")
+ if evs := fx.policyEvents(t, "shop"); len(evs) != 0 {
+ t.Fatalf("events %+v", evs)
+ }
+}
+
+func TestPolicyGetShowsAnExpiredRecord(t *testing.T) {
+ t.Parallel()
+ fx := newPolicyFx(t, "code-policy-expired")
+ signed := time.Now().UTC().Add(-73 * time.Hour).Truncate(time.Second)
+ wire := signPolicy(t, fx.consoleFixture, apppolicy.Policy{V: 1, Account: fx.sqid, App: "shop", Revision: 3,
+ SignedAt: signed, EgressAllow: []string{"10.0.5.20"}})
+ if _, err := apppolicy.Verify(fx.pub, wire, time.Now()); !errors.Is(err, apppolicy.ErrExpired) {
+ t.Fatalf("the record is not expired: %v", err)
+ }
+ putAtKey(t, fx.rel, fx.sqid, "shop", wire)
+ g := fx.get(t, "shop")
+ if !slices.Equal(g.Policy.EgressAllow, []string{"10.0.5.20"}) || !g.Policy.SignedAt.Equal(signed) || g.Policy.Revision != 3 {
+ t.Fatalf("GET = %+v", g.Policy)
+ }
+}
+
+// claimPolicyLease gives a claim case's console the fake writer lease (the
+// case carries a Bucket) and keeps the door's log lines.
+func claimPolicyLease(cx *claimCase) *policyLogRecorder {
+ cx.srv.cfg.PolicyLease = (&fakePolicyLeaser{}).leaser
+ logs := &policyLogRecorder{}
+ cx.srv.policyLog = logs.logf
+ return logs
+}
+
+func claimPolicyReq(t *testing.T, cx *claimCase, method, body string, hdr map[string]string) (int, policyReply, string) {
+ t.Helper()
+ st, _, out := rawReqHeader(t, method, cx.ts.URL+"/api/cli/apps/bab/titogo/policy", cx.bearer, body, hdr)
+ var r policyReply
+ if st == http.StatusOK {
+ if err := json.Unmarshal([]byte(out), &r); err != nil {
+ t.Fatal(err)
+ }
+ }
+ return st, r, out
+}
+
+func TestPolicyPutStampsTheAppIDUnderClaims(t *testing.T) {
+ t.Parallel()
+ cx := newClaimCase(t)
+ claimPolicyLease(cx)
+ st, _, out := claimPolicyReq(t, cx, http.MethodPut, `{"default_size":"small"}`, policyCreate)
+ if st != http.StatusOK {
+ t.Fatalf("PUT = %d %s", st, out)
+ }
+ cur, err := cx.st.ClaimAppID(OpsAccountID, "titogo")
+ if err != nil || cur == "" {
+ t.Fatalf("ClaimAppID %q %v", cur, err)
+ }
+ if stored, _, _ := storedPolicy(t, cx.rel, cx.pub, "bab", "titogo"); stored.AppID != cur || stored.Revision != 1 {
+ t.Fatalf("stored %+v, want app_id %q", stored, cur)
+ }
+}
+
+// Astra plan R1-2. The state a purge and a recreate leave when the
+// create-time clear failed (or an older console created the app): the
+// table's row and the object at the key both carry the predecessor
+// incarnation's policy, recorded at revision 1. Staged directly: the
+// create-time clear is Task 11's.
+func TestPolicyPutStartsEmptyOverAnotherIncarnationsPolicy(t *testing.T) {
+ t.Parallel()
+ cx := newClaimCase(t)
+ logs := claimPolicyLease(cx)
+ ctx := context.Background()
+ cur, err := cx.st.ClaimAppID(OpsAccountID, "titogo")
+ if err != nil {
+ t.Fatal(err)
+ }
+ w1 := signPolicy(t, cx.consoleFixture, apppolicy.Policy{V: 1, Account: "bab", App: "titogo", Revision: 1,
+ SignedAt: time.Now().UTC(), AppID: cur + "-predecessor", EgressAllow: []string{"10.0.5.20"}})
+ if err := cx.st.RecordAppPolicyContext(ctx, OpsAccountID, "titogo", 1, w1); err != nil {
+ t.Fatal(err)
+ }
+ etag := putAtKey(t, cx.rel, "bab", "titogo", w1)
+
+ st, g, out := claimPolicyReq(t, cx, http.MethodGet, "", nil)
+ if st != http.StatusOK || g.Policy.V != 1 || len(g.Policy.EgressAllow) != 0 || g.Policy.Revision != 0 || g.ETag != etag {
+ t.Fatalf("GET = %d %s", st, out)
+ }
+ logs.has(t, fmt.Sprintf("console: app policy bab/titogo: stored policy is for incarnation %q, not %q; starting empty", cur+"-predecessor", cur))
+ st, _, out = claimPolicyReq(t, cx, http.MethodPut, `{"default_size":"small"}`, policyIfMatch(etag))
+ if st != http.StatusOK {
+ t.Fatalf("PUT = %d %s", st, out)
+ }
+ stored, _, _ := storedPolicy(t, cx.rel, cx.pub, "bab", "titogo")
+ if stored.AppID != cur || stored.Revision != 2 || len(stored.EgressAllow) != 0 || stored.DefaultSize != "small" {
+ t.Fatalf("stored %+v, want app_id %q, revision 2, no egress allow, default small", stored, cur)
+ }
+ evs, err := cx.st.Events(OpsAccountID, "titogo", 100)
+ if err != nil {
+ t.Fatal(err)
+ }
+ found := false
+ for _, e := range evs {
+ if e.Action == "app-policy" {
+ found = true
+ if strings.Contains(e.Detail, "egress allow") {
+ t.Fatalf("audit detail %q names the predecessor's egress allow", e.Detail)
+ }
+ }
+ }
+ if !found {
+ t.Fatal("no app-policy event")
+ }
+}
diff --git a/internal/console/apppolicy_store.go b/internal/console/apppolicy_store.go
new file mode 100644
index 000000000..d63022c1b
--- /dev/null
+++ b/internal/console/apppolicy_store.go
@@ -0,0 +1,216 @@
+package console
+
+// The app policy, storage half (#328, #329). Once a revision is recorded the
+// console's table is the authority for its content, as the door keys'
+// database is (doorkeys_store.go): the door records the signed bytes before
+// it writes them to the bucket, so a write whose reply is lost, or a bucket
+// writer that puts an older record back, can never leave the console
+// building on, or re-signing, content older than what it last signed.
+//
+// One row per app: the last recorded revision, its signed bytes, the bucket
+// ETag the row last matched (synced_etag) and whether a publication of the
+// recorded bytes is still unconfirmed (pending). Revisions only rise. A
+// recorded revision is never recorded again, so a revision number never
+// names two contents. The purge leaves the row, so a recreated app's first
+// policy is above every record its predecessor had.
+
+import (
+ "context"
+ "database/sql"
+ "errors"
+ "fmt"
+)
+
+var (
+ // errAppPolicyRevisionUsed is RecordAppPolicy's refusal of a revision at
+ // or below the one already recorded.
+ errAppPolicyRevisionUsed = errors.New("console: app policy revision already used")
+ // errAppPolicyRevisionMoved is SyncedAppPolicy's and ResignAppPolicy's
+ // refusal when the recorded revision is no longer the one named.
+ errAppPolicyRevisionMoved = errors.New("console: app policy revision moved")
+)
+
+// migrateAppPolicy creates the revision table. Additive and idempotent,
+// called from Store.migrate.
+func (s *Store) migrateAppPolicy() error {
+ _, err := s.db.Exec(`CREATE TABLE IF NOT EXISTS app_policy_revisions (
+ account_id INTEGER NOT NULL,
+ app TEXT NOT NULL,
+ revision INTEGER NOT NULL,
+ wire BLOB NOT NULL,
+ synced_etag TEXT NOT NULL DEFAULT '',
+ pending INTEGER NOT NULL DEFAULT 0,
+ PRIMARY KEY (account_id, app)
+);`)
+ return err
+}
+
+// AppPolicyRow is one app's recorded policy. SyncedETag is the bucket ETag
+// the row last matched: the ETag this console's publication of Wire got
+// back, or that of a bucket record it adopted; "" when no publication of
+// this row is confirmed. Pending is true while signed bytes are recorded
+// whose publication is not yet confirmed; false with SyncedETag "" is a
+// confirmed absence.
+type AppPolicyRow struct {
+ AccountID int64
+ Sqid, App string
+ Revision uint64
+ Wire []byte
+ SyncedETag string
+ Pending bool
+}
+
+// appPolicyRevisionOK refuses a revision SQLite's signed INTEGER cannot hold,
+// and revision 0, which no signed record carries.
+func appPolicyRevisionOK(rev uint64) error {
+ if rev == 0 || rev > 1<<63-1 {
+ return fmt.Errorf("console: app policy revision %d out of range", rev)
+ }
+ return nil
+}
+
+// AppPolicyRecordContext is the app's row: the zero row (Revision 0) and a
+// nil error when there is none. A read: it takes no app mutation.
+func (s *Store) AppPolicyRecordContext(ctx context.Context, accountID int64, app string) (AppPolicyRow, error) {
+ row := AppPolicyRow{AccountID: accountID, App: app}
+ var rev int64
+ var pending int
+ var sqid sql.NullString
+ err := s.db.QueryRowContext(ctx, `SELECT a.sqid, r.revision, r.wire, r.synced_etag, r.pending
+FROM app_policy_revisions r LEFT JOIN accounts a ON a.id = r.account_id
+WHERE r.account_id = ? AND r.app = ?`, accountID, app).Scan(&sqid, &rev, &row.Wire, &row.SyncedETag, &pending)
+ if errors.Is(err, sql.ErrNoRows) {
+ return AppPolicyRow{AccountID: accountID, App: app}, nil
+ }
+ if err != nil {
+ return AppPolicyRow{}, err
+ }
+ row.Sqid, row.Revision, row.Pending = sqid.String, uint64(rev), pending != 0
+ return row, nil
+}
+
+// AppPolicyRowsContext is every row whose publication was confirmed or is
+// still pending, each with its account's sqid: the live policies as far as
+// this console knows, without listing the bucket. A read.
+func (s *Store) AppPolicyRowsContext(ctx context.Context) ([]AppPolicyRow, error) {
+ rows, err := s.db.QueryContext(ctx, `SELECT r.account_id, a.sqid, r.app, r.revision, r.wire, r.synced_etag, r.pending
+FROM app_policy_revisions r JOIN accounts a ON a.id = r.account_id
+WHERE r.synced_etag != '' OR r.pending = 1 ORDER BY a.sqid, r.app`)
+ if err != nil {
+ return nil, err
+ }
+ defer rows.Close()
+ var out []AppPolicyRow
+ for rows.Next() {
+ var row AppPolicyRow
+ var rev int64
+ var pending int
+ if err := rows.Scan(&row.AccountID, &row.Sqid, &row.App, &rev, &row.Wire, &row.SyncedETag, &pending); err != nil {
+ return nil, err
+ }
+ row.Revision, row.Pending = uint64(rev), pending != 0
+ out = append(out, row)
+ }
+ return out, rows.Err()
+}
+
+// RecordAppPolicy records rev and its signed bytes when rev is above the
+// row's revision (inserting the row when there is none), else
+// errAppPolicyRevisionUsed. It is both the counter bump and the content
+// record. synced_etag is left as it was; pending is set.
+func (s *Store) claimRawRecordAppPolicy(accountID int64, app string, rev uint64, wire []byte) error {
+ if err := appPolicyRevisionOK(rev); err != nil {
+ return err
+ }
+ res, err := s.db.Exec(`INSERT INTO app_policy_revisions (account_id, app, revision, wire, pending) VALUES (?, ?, ?, ?, 1)
+ON CONFLICT (account_id, app) DO UPDATE SET revision = excluded.revision, wire = excluded.wire, pending = 1
+WHERE app_policy_revisions.revision < excluded.revision`, accountID, app, int64(rev), wire)
+ if err != nil {
+ return err
+ }
+ if n, err := res.RowsAffected(); err != nil {
+ return err
+ } else if n == 0 {
+ return errAppPolicyRevisionUsed
+ }
+ return nil
+}
+
+// AdoptAppPolicy makes the bucket's verified record the row's content:
+// revision = max(revision, rev), its bytes, its ETag, nothing pending,
+// inserting the row when there is none. Revisions still only rise.
+func (s *Store) claimRawAdoptAppPolicy(accountID int64, app string, rev uint64, wire []byte, etag string) error {
+ if err := appPolicyRevisionOK(rev); err != nil {
+ return err
+ }
+ _, err := s.db.Exec(`INSERT INTO app_policy_revisions (account_id, app, revision, wire, synced_etag, pending) VALUES (?, ?, ?, ?, ?, 0)
+ON CONFLICT (account_id, app) DO UPDATE SET revision = MAX(app_policy_revisions.revision, excluded.revision),
+ wire = excluded.wire, synced_etag = excluded.synced_etag, pending = 0`, accountID, app, int64(rev), wire, etag)
+ return err
+}
+
+// SyncedAppPolicy marks a publication of revision rev confirmed (etag the
+// ETag it got back, or "" when a read found the object absent), else
+// errAppPolicyRevisionMoved. It is the only write that clears pending for a
+// recorded revision.
+func (s *Store) claimRawSyncedAppPolicy(accountID int64, app string, rev uint64, etag string) error {
+ if err := appPolicyRevisionOK(rev); err != nil {
+ return err
+ }
+ return appPolicyUpdateAt(s.db.Exec(`UPDATE app_policy_revisions SET synced_etag = ?, pending = 0
+WHERE account_id = ? AND app = ? AND revision = ?`, etag, accountID, app, int64(rev)))
+}
+
+// ResignAppPolicy replaces the recorded bytes of revision rev (the re-sign:
+// the same content with a later signed_at) and marks them pending, else
+// errAppPolicyRevisionMoved.
+func (s *Store) claimRawResignAppPolicy(accountID int64, app string, rev uint64, wire []byte) error {
+ if err := appPolicyRevisionOK(rev); err != nil {
+ return err
+ }
+ return appPolicyUpdateAt(s.db.Exec(`UPDATE app_policy_revisions SET wire = ?, pending = 1
+WHERE account_id = ? AND app = ? AND revision = ?`, wire, accountID, app, int64(rev)))
+}
+
+// appPolicyUpdateAt is errAppPolicyRevisionMoved for an UPDATE that matched
+// no row at the named revision.
+func appPolicyUpdateAt(res sql.Result, err error) error {
+ if err != nil {
+ return err
+ }
+ if n, err := res.RowsAffected(); err != nil {
+ return err
+ } else if n == 0 {
+ return errAppPolicyRevisionMoved
+ }
+ return nil
+}
+
+func (s *Store) RecordAppPolicyContext(claimContext context.Context, accountID int64, app string, rev uint64, wire []byte) (claimResult0 error) {
+ claimResult0 = s.withAutoAppSQL(claimContext, accountID, app, "sql.RecordAppPolicy", func(claimView *Store) error {
+ claimResult0 = claimView.claimRawRecordAppPolicy(accountID, app, rev, wire)
+ return claimResult0
+ })
+ return
+}
+func (s *Store) AdoptAppPolicyContext(claimContext context.Context, accountID int64, app string, rev uint64, wire []byte, etag string) (claimResult0 error) {
+ claimResult0 = s.withAutoAppSQL(claimContext, accountID, app, "sql.AdoptAppPolicy", func(claimView *Store) error {
+ claimResult0 = claimView.claimRawAdoptAppPolicy(accountID, app, rev, wire, etag)
+ return claimResult0
+ })
+ return
+}
+func (s *Store) SyncedAppPolicyContext(claimContext context.Context, accountID int64, app string, rev uint64, etag string) (claimResult0 error) {
+ claimResult0 = s.withAutoAppSQL(claimContext, accountID, app, "sql.SyncedAppPolicy", func(claimView *Store) error {
+ claimResult0 = claimView.claimRawSyncedAppPolicy(accountID, app, rev, etag)
+ return claimResult0
+ })
+ return
+}
+func (s *Store) ResignAppPolicyContext(claimContext context.Context, accountID int64, app string, rev uint64, wire []byte) (claimResult0 error) {
+ claimResult0 = s.withAutoAppSQL(claimContext, accountID, app, "sql.ResignAppPolicy", func(claimView *Store) error {
+ claimResult0 = claimView.claimRawResignAppPolicy(accountID, app, rev, wire)
+ return claimResult0
+ })
+ return
+}
diff --git a/internal/console/authcli.go b/internal/console/authcli.go
index 87ecc1b47..800c9f072 100644
--- a/internal/console/authcli.go
+++ b/internal/console/authcli.go
@@ -917,6 +917,15 @@ func (s *Server) registerCLIRoutes() {
// §11 says only app members change it, and a credential the app holds
// must not choose what its own migrations run beside.
s.cliRoute("PUT /api/cli/apps/{account}/{app}/swap", svcrole.Set{}, s.handleAPISwapPut)
+ // The app policy (instance sizes and egress spec §6, apipolicy.go): the
+ // private addresses an app's instances may reach and the sizes they run
+ // at. The ZERO Set at the mux, so no service credential reaches either
+ // door, an app's own included: a credential must not widen what its own
+ // instances may reach. The operator gate (apiSystemRoller: an ops account
+ // owner or CARLOS_SYSTEM_OPERATORS) is in the handlers, because a person
+ // always passes cliRoute's Set.
+ s.cliRoute("GET /api/cli/apps/{account}/{app}/policy", svcrole.Set{}, s.handleAPIPolicyGet)
+ s.cliRoute("PUT /api/cli/apps/{account}/{app}/policy", svcrole.Set{}, s.handleAPIPolicyPut)
// The object-storage API (apistore.go, issue #55) — declare a store, read
// what became of it, and drive the walks that make it real. Beside the
// config routes on purpose: a store's credentials are DELIVERED as config,
diff --git a/internal/console/claim_invariant_fixtures_test.go b/internal/console/claim_invariant_fixtures_test.go
index 92749cd45..1c233d65a 100644
--- a/internal/console/claim_invariant_fixtures_test.go
+++ b/internal/console/claim_invariant_fixtures_test.go
@@ -63,6 +63,15 @@ func claimAPI(method, path, body string) func(*testing.T, *claimCase) int {
return func(t *testing.T, cx *claimCase) int { return cx.api(t, method, path, body) }
}
+// claimAPIWith is claimAPI with request headers, for a door whose request
+// carries a precondition (the app policy door's If-None-Match).
+func claimAPIWith(method, path, body string, header map[string]string) func(*testing.T, *claimCase) int {
+ return func(t *testing.T, cx *claimCase) int {
+ status, _, _ := rawReqHeader(t, method, cx.ts.URL+cx.expand(path), cx.bearer, cx.expand(body), header)
+ return status
+ }
+}
+
// claimAPIVar is claimAPI with "<var>" in path or body replaced from cx.vars,
// for names a setup only learns at runtime (a door key id, a proposal id).
func claimAPIVar(method, path, body string) func(*testing.T, *claimCase) int {
@@ -827,6 +836,7 @@ var claimFixtures = map[string]claimFixture{
"route GET /api/cli/apps/{account}/{app}/instances/{host}": claimRead("reads one instance record"),
"route GET /api/cli/apps/{account}/{app}/instances/{host}/move": claimRead("reads one instance's move state"),
"route GET /api/cli/apps/{account}/{app}/ledgers": claimRead("lists ledgers"),
+ "route GET /api/cli/apps/{account}/{app}/policy": claimRead("reads the app policy object and the console's recorded revision; records nothing and takes no lease"),
"route GET /api/cli/apps/{account}/{app}/ledgers/{ledger}/head": claimRead("reads a ledger head"),
"route GET /api/cli/apps/{account}/{app}/logs": claimRead("reads log chunks"),
"route GET /api/cli/apps/{account}/{app}/pending": claimRead("lists pending proposals"),
@@ -1905,6 +1915,21 @@ var claimFixtures = map[string]claimFixture{
apps: claimRefs("bab/titogo"),
do: claimAPI(http.MethodPut, "/api/cli/apps/bab/titogo/swap", `{"swap":"hold"}`),
}},
+ // The app policy door (apipolicy.go). The claim case carries a Bucket, so
+ // its console takes the writer lease: the setup gives it Task 8's fake
+ // leaser, the server's own field and not process-wide state, so the door
+ // runs in parallel. A real lease's lock object would be written by
+ // internal/lease's own client, outside the boundary observed here, as the
+ // housekeeper's is.
+ "route PUT /api/cli/apps/{account}/{app}/policy": {door: &claimDoor{
+ apps: claimRefs("bab/titogo"),
+ setup: func(t *testing.T, cx *claimCase) {
+ t.Helper()
+ claimPolicyLease(cx)
+ },
+ do: claimAPIWith(http.MethodPut, "/api/cli/apps/bab/titogo/policy", `{"default_size":"small"}`,
+ map[string]string{"If-None-Match": "*"}),
+ }},
"route PUT /api/cli/apps/{account}/{app}/telemetry": {door: &claimDoor{
apps: claimRefs("bab/titogo"),
do: claimAPI(http.MethodPut, "/api/cli/apps/bab/titogo/telemetry", `{"errors":{"log_level":"warn"}}`),
diff --git a/internal/console/claim_invariant_probes_read_test.go b/internal/console/claim_invariant_probes_read_test.go
index 5c814388d..55049d178 100644
--- a/internal/console/claim_invariant_probes_read_test.go
+++ b/internal/console/claim_invariant_probes_read_test.go
@@ -56,6 +56,17 @@ func claimWorldDoorKey(t *testing.T, cx *claimCase) {
cx.mustAPI(t, http.MethodPost, "/api/cli/apps/bab/titogo/door-keys", `{"label":"router"}`, http.StatusCreated)
}
+// claimWorldPolicy stages an app policy for bab/titogo through its door,
+// with the fake writer lease the claim case's Bucket needs.
+func claimWorldPolicy(t *testing.T, cx *claimCase) {
+ t.Helper()
+ claimPolicyLease(cx)
+ if st, _, out := rawReqHeader(t, http.MethodPut, cx.ts.URL+"/api/cli/apps/bab/titogo/policy", cx.bearer,
+ `{"default_size":"small"}`, map[string]string{"If-None-Match": "*"}); st != http.StatusOK {
+ t.Fatalf("stage the app policy = %d %s", st, out)
+ }
+}
+
func claimWorldLedger(t *testing.T, cx *claimCase) {
t.Helper()
cx.mustAPI(t, http.MethodPost, "/api/cli/apps/bab/titogo/ledgers/carbon/append", `{"payload":{"kg_central":1}}`, http.StatusOK)
@@ -168,6 +179,7 @@ var claimReadProbes = map[string]claimProbe{
"route GET /api/cli/apps/{account}/{app}/domains": claimAPIRead(claimWorldAlias, "/api/cli/apps/bab/titogo/domains", "tickets.example.com"),
"route GET /api/cli/apps/{account}/{app}/domains/{host}": claimAPIRead(claimWorldAlias, "/api/cli/apps/bab/titogo/domains/tickets.example.com", "tickets.example.com"),
"route GET /api/cli/apps/{account}/{app}/door-keys": claimAPIRead(claimWorldDoorKey, "/api/cli/apps/bab/titogo/door-keys", `"label":"router"`),
+ "route GET /api/cli/apps/{account}/{app}/policy": claimAPIRead(claimWorldPolicy, "/api/cli/apps/bab/titogo/policy", `"default_size":"small"`),
"route GET /api/cli/apps/{account}/{app}/email": claimAPIRead(func(t *testing.T, cx *claimCase) { cx.emailDeclared(t) },
"/api/cli/apps/bab/titogo/email", claimMailDomain),
"route GET /api/cli/apps/{account}/{app}/email/events": claimAPIRead(func(t *testing.T, cx *claimCase) { cx.emailProvisioned(t) },
diff --git a/internal/console/server.go b/internal/console/server.go
index 1fdec2e29..cb3e7c8e4 100644
--- a/internal/console/server.go
+++ b/internal/console/server.go
@@ -995,6 +995,20 @@ type Server struct {
policyStopping bool
policyLog func(format string, args ...any)
policyAttemptEnding func()
+ // The app policy door (apipolicy.go). policyLocks is each app's policy
+ // mutex, keyed "<sqid>/<app>" and built lazily under policyLocksMu as
+ // doorPublish is; the door and the re-sign pass share it. policySynced
+ // is, per app, the number of the last writer holding that chose a base
+ // 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.
+ policyLocksMu sync.Mutex
+ policyLocks map[string]*sync.Mutex
+ policySyncedMu sync.Mutex
+ policySynced map[string]uint64
+ policyAfterBase func(sqid, app string)
// 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.
diff --git a/internal/console/store.go b/internal/console/store.go
index 2f83cc562..9ea01e066 100644
--- a/internal/console/store.go
+++ b/internal/console/store.go
@@ -545,7 +545,12 @@ func (s *Store) migrate() error {
return err
}
// Door keys' own tables (doorkeys_store.go), for the same reason.
- return s.migrateDoorKeys()
+ if err := s.migrateDoorKeys(); err != nil {
+ return err
+ }
+ // The app policy's revision and recorded content (apppolicy_store.go),
+ // for the same reason.
+ return s.migrateAppPolicy()
}
// mailCountsBackfillVersion is the user_version this migration stamps. The
diff --git a/internal/console/store_gate.go b/internal/console/store_gate.go
index 679c3a742..da95ba489 100644
--- a/internal/console/store_gate.go
+++ b/internal/console/store_gate.go
@@ -191,6 +191,7 @@ var claimSQLTables = map[string]claimSQLTable{
"cli_tokens": {Account: "account_id", App: "app", FreeWhenNoApp: true}, // user tokens carry app ''
"door_keys": {Account: "account_id", App: "app", PurgeMonotone: "revoked_at"},
"door_key_revisions": {Account: "account_id", App: "app", PurgeMonotone: "revision"},
+ "app_policy_revisions": {Account: "account_id", App: "app"}, // apppolicy_store.go; the purge never writes it (revisions only rise)
"pending_changes": {Account: "account_id", App: "app"},
"pending_confirms": {Account: "account_id", App: "app", Via: &claimSQLVia{Parent: "pending_changes", ChildCol: "change_id", ParentCol: "id"}},
"mail_counts": {Sqid: "account", App: "app"},
diff --git a/internal/console/store_gate_test.go b/internal/console/store_gate_test.go
index 614d05979..1dd83ca60 100644
--- a/internal/console/store_gate_test.go
+++ b/internal/console/store_gate_test.go
@@ -717,7 +717,7 @@ func TestClaimSQLTableCensus(t *testing.T) {
t.Errorf("NonApp table %q has app column %q", x.table, x.col)
}
}
- for _, name := range []string{"apps", "owners", "audit", "cli_tokens", "door_keys", "door_key_revisions", "pending_changes", "pending_confirms",
+ for _, name := range []string{"apps", "owners", "audit", "cli_tokens", "door_keys", "door_key_revisions", "app_policy_revisions", "pending_changes", "pending_confirms",
"mail_counts", "mail_sender_counts", "mail_events", "mail_event_hooks", "mail_event_deliveries"} {
if sp, ok := present[name]; !ok || sp.NonApp != "" {
t.Errorf("%s must be present and gated", name)
diff --git a/internal/console/svcroutes_test.go b/internal/console/svcroutes_test.go
index 497b2060e..ef323cde9 100644
--- a/internal/console/svcroutes_test.go
+++ b/internal/console/svcroutes_test.go
@@ -51,6 +51,10 @@ import (
// /{app}/updates is the same manifest's update policy (settings spec §3.2),
// including which settings an instance may change itself: the bound on the
// instance credential, which that credential must never widen.
+// /{app}/policy is the operator's app policy (instance sizes and egress spec
+// §6): the private addresses an app's instances may reach and the sizes they
+// run at. The same fact about its subject: an app's credential must never
+// widen what its own instances may reach.
//
// The email family adds one entry to each list, and both are facts about the
// door's subject rather than about any role:
@@ -78,7 +82,7 @@ import (
var (
mintDoors = []string{"/instance-token", "/door-keys", "/api/cli/services/", "/email/credentials"}
fleetDoors = []string{"/api/cli/fleets/"}
- policyDoors = []string{"/instances-manifest", "/{app}/updates", "/{app}/routes", "/email/pause", "/build-probe", "/{app}/swap"}
+ policyDoors = []string{"/instances-manifest", "/{app}/updates", "/{app}/routes", "/email/pause", "/build-probe", "/{app}/swap", "/{app}/policy"}
systemDoors = []string{"/api/cli/system/"}
)
@@ -237,6 +241,12 @@ var closedDoors = []string{
// app members": a credential the app holds must not choose what its own
// migrations run beside.
"PUT /api/cli/apps/{account}/{app}/swap",
+ // The app policy (instance sizes and egress spec §6): operator-only, both
+ // the read and the write. Its egress allow opens private addresses to an
+ // app's instances, so a credential the app holds must never widen it, and
+ // no service role is the right holder of an operator's decision.
+ "GET /api/cli/apps/{account}/{app}/policy",
+ "PUT /api/cli/apps/{account}/{app}/policy",
// The sidecar manifest, shut for the instances manifest's reason and one
// of its own. It is standing policy — which domain, pool, regions and
// channel a sidecar may ever come up on — and it also carries `args`,
@@ -1050,8 +1060,8 @@ func TestEveryClosedDoorRefusesAServiceCredential(t *testing.T) {
}
seen++
}
- if seen != len(closedDoors) || seen != 59 {
- t.Errorf("walked %d closed doors, want all %d (and 59 today: instance-token, three door-key doors, the routes door, instances-manifest, the app update policy, sidecars-manifest, the route-channel move, three instance update settings doors, two db replica doors, the forwarded db read, six fleets, four services, two email credentials, email pause, the build probe, the swap policy, the live tenancy read, the layout read, the activation-cancel door, nine system doors, three deployment-wide antivirus, three platform geoip doors, two pointer digest doors, the channel resign, two release authorize doors, four store-entitlement doors, and two account plan doors)",
+ if seen != len(closedDoors) || seen != 61 {
+ t.Errorf("walked %d closed doors, want all %d (and 61 today: instance-token, three door-key doors, the routes door, instances-manifest, the app update policy, sidecars-manifest, the route-channel move, three instance update settings doors, two db replica doors, the forwarded db read, six fleets, four services, two email credentials, email pause, the build probe, the swap policy, two app policy doors, the live tenancy read, the layout read, the activation-cancel door, nine system doors, three deployment-wide antivirus, three platform geoip doors, two pointer digest doors, the channel resign, two release authorize doors, four store-entitlement doors, and two account plan doors)",
seen, len(closedDoors))
}
}