amadan

carlos / platform Public

Bound worker assignment storage versions and reject enabled no-op writes

Paul Campbell pushed by paul@keymail.dev b3de19705be8794703021a9ebb858171d0b94111 parent c1d40ee
2 files changed, +128 −3
  • internal/release/workerassignment.go +11 −3
  • internal/release/workerassignment_bounds_test.go +117 −0

CI failed — run details

diff --git a/internal/release/workerassignment.go b/internal/release/workerassignment.go
index 3d2678805..e1d2fdb2a 100644
--- a/internal/release/workerassignment.go
+++ b/internal/release/workerassignment.go
@@ -8,6 +8,12 @@ import (
"github.com/carlosframework/platform/internal/maintenanceguard"
)
+const maxWorkerControlETagBytes = 1024
+
+func validWorkerControlETag(etag string) bool {
+ return len(etag) > 0 && len(etag) <= maxWorkerControlETagBytes
+}
+
// WorkerAssignmentSnapshot binds verified bytes to the store version read.
// It proves neither worker authentication nor the evidence behind a receipt.
// Callers must check current enrollment, registration, expiry and the complete
@@ -20,7 +26,7 @@ type WorkerAssignmentSnapshot struct {
func (s WorkerAssignmentSnapshot) State(pub ed25519.PublicKey) (WorkerAssignmentState, error) {
state, err := VerifyWorkerAssignment(pub, s.raw)
- if err != nil || state.Identity != s.identity || s.etag == "" {
+ if err != nil || state.Identity != s.identity || !validWorkerControlETag(s.etag) {
return WorkerAssignmentState{}, ErrWorkerControl
}
return state, nil
@@ -58,6 +64,7 @@ func ReadWorkerAssignment(ctx context.Context, store WorkerControlStore, pub ed2
// must register its immutable assignment reference on an active worker head
// before it can authorize work. A conflict requires reading existing state;
// creating again never resets accepted receipts or revives a disabled state.
+// An invalid returned ETag is an uncertain write: re-read the same identity.
func CreateWorkerAssignment(ctx context.Context, store WorkerControlStore, sg Signer, state WorkerAssignmentState) (WorkerAssignmentSnapshot, error) {
key, err := workerAssignmentKey(state.Identity)
if err != nil || store == nil || ctx == nil || ctx.Err() != nil || state.Revision != 1 || state.Disabled || state.Preflight != nil || state.Compatibility != nil || state.Execution != nil || state.Resume != nil {
@@ -71,7 +78,7 @@ func CreateWorkerAssignment(ctx context.Context, store WorkerControlStore, sg Si
if err != nil {
return WorkerAssignmentSnapshot{}, err
}
- if etag == "" {
+ if !validWorkerControlETag(etag) {
return WorkerAssignmentSnapshot{}, ErrWorkerControl
}
return WorkerAssignmentSnapshot{raw: raw, etag: etag, identity: state.Identity}, nil
@@ -81,6 +88,7 @@ func CreateWorkerAssignment(ctx context.Context, store WorkerControlStore, sg Si
// publication and revocation therefore compete on the same storage version.
// The storage principal must prevent rollback of these objects: signatures
// alone do not stop a writer replaying an older signed enabled state.
+// An invalid returned ETag is an uncertain write, never evidence of rollback.
func AdvanceWorkerAssignment(ctx context.Context, store WorkerControlStore, sg Signer, before WorkerAssignmentSnapshot, after WorkerAssignmentState) (WorkerAssignmentSnapshot, error) {
if store == nil || ctx == nil || ctx.Err() != nil || sg.usable() != nil || sg.scope != nil || sg.grant != nil {
return WorkerAssignmentSnapshot{}, ErrWorkerControl
@@ -101,7 +109,7 @@ func AdvanceWorkerAssignment(ctx context.Context, store WorkerControlStore, sg S
if err != nil {
return WorkerAssignmentSnapshot{}, err
}
- if etag == "" {
+ if !validWorkerControlETag(etag) {
return WorkerAssignmentSnapshot{}, ErrWorkerControl
}
return WorkerAssignmentSnapshot{raw: raw, etag: etag, identity: after.Identity}, nil
diff --git a/internal/release/workerassignment_bounds_test.go b/internal/release/workerassignment_bounds_test.go
new file mode 100644
index 000000000..6291f2a5b
--- /dev/null
+++ b/internal/release/workerassignment_bounds_test.go
@@ -0,0 +1,117 @@
+package release
+
+import (
+ "context"
+ "errors"
+ "strings"
+ "testing"
+)
+
+type taggedWorkerStore struct {
+ WorkerControlStore
+ readTag, writeTag *string
+}
+
+func (s taggedWorkerStore) Get(ctx context.Context, key string) ([]byte, string, error) {
+ raw, tag, err := s.WorkerControlStore.Get(ctx, key)
+ if s.readTag != nil {
+ tag = *s.readTag
+ }
+ return raw, tag, err
+}
+func (s taggedWorkerStore) Put(ctx context.Context, key string, raw []byte, etag string) (string, error) {
+ tag, err := s.WorkerControlStore.Put(ctx, key, raw, etag)
+ if s.writeTag != nil {
+ tag = *s.writeTag
+ }
+ return tag, err
+}
+func TestWorkerAssignmentBoundsEveryReturnedETag(t *testing.T) {
+ for _, length := range []int{0, maxWorkerControlETagBytes, maxWorkerControlETagBytes + 1} {
+ priv, pub := testKeys(t)
+ sg := RootSigner(priv)
+ ctx := t.Context()
+ store := NewDirStore(t.TempDir())
+ initial := workerTestAssignment()
+ snap, err := CreateWorkerAssignment(ctx, store, sg, initial)
+ if err != nil {
+ t.Fatal(err)
+ }
+ tag := strings.Repeat("a", length)
+ bad := taggedWorkerStore{WorkerControlStore: store, readTag: &tag}
+ _, err = ReadWorkerAssignment(ctx, bad, pub, initial.Identity)
+ if (length == maxWorkerControlETagBytes) != (err == nil) {
+ t.Fatalf("read length %d: %v", length, err)
+ }
+ snap.etag = tag
+ if _, err := snap.State(pub); (length == maxWorkerControlETagBytes) != (err == nil) {
+ t.Fatalf("snapshot length %d: %v", length, err)
+ }
+ if length == maxWorkerControlETagBytes {
+ continue
+ }
+ counted := &countedWorkerStore{WorkerControlStore: store}
+ after := initial
+ after.Revision++
+ after.Disabled = true
+ if _, err := AdvanceWorkerAssignment(ctx, counted, sg, snap, after); !errors.Is(err, ErrWorkerControl) {
+ t.Fatal("invalid snapshot sent to store", err)
+ }
+ if counted.puts.Load() != 0 {
+ t.Fatal("invalid ETag reached conditional write")
+ }
+ }
+}
+func TestWorkerAssignmentInvalidWriteETagIsUncertain(t *testing.T) {
+ for _, create := range []bool{false, true} {
+ priv, pub := testKeys(t)
+ sg := RootSigner(priv)
+ ctx := t.Context()
+ store := NewDirStore(t.TempDir())
+ initial := workerTestAssignment()
+ tag := strings.Repeat("a", maxWorkerControlETagBytes+1)
+ bad := taggedWorkerStore{WorkerControlStore: store, writeTag: &tag}
+ expected := initial
+ var err error
+ if create {
+ _, err = CreateWorkerAssignment(ctx, bad, sg, initial)
+ } else {
+ before, e := CreateWorkerAssignment(ctx, store, sg, initial)
+ if e != nil {
+ t.Fatal(e)
+ }
+ expected.Revision++
+ expected.Disabled = true
+ _, err = AdvanceWorkerAssignment(ctx, bad, sg, before, expected)
+ }
+ if !errors.Is(err, ErrWorkerControl) {
+ t.Fatal("invalid write ETag accepted", err)
+ }
+ recovered, e := ReadWorkerAssignment(ctx, store, pub, initial.Identity)
+ if e != nil {
+ t.Fatal(e)
+ }
+ got, e := recovered.State(pub)
+ if e != nil || got.Revision != expected.Revision || got.Disabled != expected.Disabled {
+ t.Fatal("uncertain write was lost", e)
+ }
+ }
+}
+func TestWorkerAssignmentEnabledNoSlotChangeNeverWrites(t *testing.T) {
+ priv, _ := testKeys(t)
+ sg := RootSigner(priv)
+ store := &countedWorkerStore{WorkerControlStore: NewDirStore(t.TempDir())}
+ state := workerTestAssignment()
+ before, err := CreateWorkerAssignment(t.Context(), store, sg, state)
+ if err != nil {
+ t.Fatal(err)
+ }
+ writes := store.puts.Load()
+ state.Revision++
+ if _, err := AdvanceWorkerAssignment(t.Context(), store, sg, before, state); !errors.Is(err, ErrWorkerControl) {
+ t.Fatal("enabled no-op accepted", err)
+ }
+ if store.puts.Load() != writes {
+ t.Fatal("enabled no-op reached storage")
+ }
+}