From 0db956dd83447b73102635a3e0fc664e109f71df Mon Sep 17 00:00:00 2001 From: fix-sdk-pas Date: Fri, 16 Oct 2026 19:28:25 +0200 Subject: [PATCH] Skip a bad assignment, honour the plan's credential form, delete only what was read, redact every credential form (issue 825) --- go/provider/provider.go | 138 ++++++++++++++++++++++++---- go/provider/provider_test.go | 169 +++++++++++++++++++++++++++++++---- go/provision.go | 2 +- 3 files changed, 276 insertions(+), 33 deletions(-) diff --git a/go/provider/provider.go b/go/provider/provider.go index b169a49..ce5df6e 100644 --- a/go/provider/provider.go +++ b/go/provider/provider.go @@ -1,5 +1,6 @@ // Package provider is a providing module's half of a binding: it answers the provider's consumers itself, over the -// bus (novox/hq ADR 0349 clause 12, design 57 B1, issue 678). +// bus (novox/hq ADR 0349 clause 12, issue 678): the provider module, not the controller, makes each bound +// consumer's entry. // // **Who binds the provider.** Each node's provisioner client records the needs of the modules handed to it from // their plans, and the controller names in each plan, per required provision and local name, the provider that @@ -35,7 +36,9 @@ package provider import ( "context" + "encoding/base64" "encoding/json" + "errors" "fmt" "regexp" "sort" @@ -63,6 +66,15 @@ type Bus interface { Delete(ctx context.Context, bucket, key string) error } +// ConditionalBus is a Bus that can also delete a key only while it still holds what the caller read. The helper uses +// it when the Bus offers it, so a pass that lets a consumer go never removes an entry another instance of the +// provider wrote after the pass read it; a Bus without it deletes unconditionally. DeleteIf returns an error that +// is stdio.ErrChanged when the key holds something else now. +type ConditionalBus interface { + Bus + DeleteIf(ctx context.Context, bucket, key string, held []byte) error +} + // Vault is how the provider reads a credential the vault keeps, by its name, opened. type Vault interface { Value(ctx context.Context, name string) ([]byte, error) @@ -85,7 +97,8 @@ type Offer struct { // Serves are the values served to each consumer, the provider's config already composed into them; a string // may name the consumer's identity, `${consumer:as}` or `${consumer:as:dns}`. Serves map[string]any - // Credential is the form of the credential its consumers present. + // Credential is the form of the credential its consumers present. A consumer's plan entry that names a + // credential form of its own wins over it (the plan's `credential-form`). Credential CredentialForm // User is the provider's one user name, for a Shared credential. User string @@ -121,6 +134,10 @@ type Entry struct { // internal name, and one of each per local name it contributes under, keyed as its binding file names them. // Written into Values over anything the provider serves under the key; never alone to the bucket. Names map[string]string `json:"-"` + // CredentialForm and Name are from the consumer's plan entry (issue 825): the form of the credential the + // controller chose for this bind (`shared`, `pair`, `issued` or `given`) and the holder's name. When set, they + // win over the offer's form. Never written to the bucket. + CredentialForm, Name string `json:"-"` } // Key is the entry's key in the provider's bucket: `..`, and `.` for a bind under a @@ -197,6 +214,9 @@ const ( Waiting = "waiting" Failed = "failed" Removed = "removed" + // Superseded is an entry no bind asks for any more that another instance of the provider wrote after this pass + // read it: it is left standing. + Superseded = "superseded" ) // Outcome is what a pass did for one consumer's entry. @@ -252,8 +272,13 @@ func (p *Provider) watch(ctx context.Context, bucket string, into map[string][]b } // Pass reads the binds and the provider's entries from the bus and answers once: each bind of this provider gets -// its entry, and each entry of this provider no bind asks for is removed. An assignment that cannot be read stops -// the pass before anything is removed. Run passes by itself; Pass is for a provider that passes when it chooses. +// its entry, and each entry of this provider no bind asks for is removed. An assignment that cannot be read, or that +// is not the one its key names, is skipped and said (an outcome of status Failed under the assignment's key, and a +// line of the log), and the entries of its consumer are left as they are, since a bind missed would read as an +// unbind; the pass goes on for the rest. Run passes by itself; Pass is for a provider that passes when it chooses. +// +// Over StateBus each call to Pass leaves two watches open for the life of the process, silenced once the call has +// ended (see StateBus): a provider that passes often should use Run, which watches once. func (p *Provider) Pass(ctx context.Context) ([]Outcome, error) { watchCtx, cancel := context.WithCancel(ctx) defer cancel() @@ -299,17 +324,25 @@ type planned struct { From string `json:"from"` As string `json:"as"` Names map[string]string `json:"names"` + CredentialForm string `json:"credential-form"` + Name string `json:"name"` } `json:"provisions"` } `json:"provisioning"` } func (p *Provider) pass(ctx context.Context, assignments, entries map[string][]byte) ([]Outcome, error) { var binds []Entry + var out []Outcome + unknown := map[string]bool{} // consumers whose assignment was skipped: their entries are not judged for _, key := range sortedKeys(assignments) { var a planned if err := json.Unmarshal(assignments[key], &a); err != nil || a.Node+"."+a.Module != key { - return nil, fmt.Errorf("the assignment %s is not the controller's, so who binds %s on %s is not known; "+ - "nothing is answered or removed", key, p.opts.Module, p.opts.Node) + o := Outcome{Key: key, Status: Failed, Why: fmt.Sprintf("the assignment %s is not the controller's, so "+ + "who it binds is not known; it is skipped, and its entries are left as they are", key)} + out = append(out, o) + p.say(key, o.Status+" "+o.Why) + unknown[key] = true + continue } if a.Provisioning == nil { continue @@ -320,11 +353,11 @@ func (p *Provider) pass(ctx context.Context, assignments, entries map[string][]b } binds = append(binds, Entry{Provider: p.opts.Node, ProviderModule: p.opts.Module, Provision: b.Provision, Consumer: a.Node, Module: a.Module, Local: b.Local, - Contributes: b.Contributes, At: b.At, From: b.From, As: b.As, Names: b.Names}) + Contributes: b.Contributes, At: b.At, From: b.From, As: b.As, Names: b.Names, + CredentialForm: b.CredentialForm, Name: b.Name}) } } - var out []Outcome want := map[string]bool{} for _, b := range binds { key := b.Key() @@ -334,7 +367,7 @@ func (p *Provider) pass(ctx context.Context, assignments, entries map[string][]b p.say(key, o.Status+" "+o.Why) } for _, key := range sortedKeys(entries) { - if want[key] { + if want[key] || unknown[consumerOf(key)] { continue } var e Entry @@ -351,7 +384,14 @@ func (p *Provider) pass(ctx context.Context, assignments, entries map[string][]b continue } } - if err := p.opts.Bus.Delete(ctx, BucketOf(p.opts.Module), key); err != nil { + if err := p.remove(ctx, key, entries[key]); err != nil { + if errors.Is(err, stdio.ErrChanged) { + o := Outcome{Key: key, Status: Superseded, Why: "another instance of the provider wrote its entry " + + "since this pass read it, so it stands"} + out = append(out, o) + p.say(key, o.Status+" "+o.Why) + continue + } o := Outcome{Key: key, Status: Failed, Why: "no longer bound, and its entry could not be removed: " + err.Error()} out = append(out, o) @@ -367,6 +407,23 @@ func (p *Provider) pass(ctx context.Context, assignments, entries map[string][]b return out, nil } +// consumerOf is the `.` an entry's key starts with: its first two parts. +func consumerOf(key string) string { + parts := strings.SplitN(key, ".", 3) + if len(parts) < 3 { + return key + } + return parts[0] + "." + parts[1] +} + +// remove deletes an entry this instance read as held, and only while it is still as read when the Bus can say so. +func (p *Provider) remove(ctx context.Context, key string, held []byte) error { + if c, ok := p.opts.Bus.(ConditionalBus); ok { + return c.DeleteIf(ctx, BucketOf(p.opts.Module), key, held) + } + return p.opts.Bus.Delete(ctx, BucketOf(p.opts.Module), key) +} + // answer makes one bind's entry and writes it when it differs from what the bucket holds. func (p *Provider) answer(ctx context.Context, e Entry, have []byte) Outcome { key := e.Key() @@ -418,13 +475,22 @@ func (p *Provider) answer(ctx context.Context, e Entry, have []byte) Outcome { } e.Values = values var credential []byte - switch offer.Credential { - case Shared: + // The plan's form wins; the offer's is the default. + form := e.CredentialForm + if form == "" { + form = map[CredentialForm]string{Shared: "shared", PerConsumer: "pair"}[offer.Credential] + } + switch form { + case "shared": e.User = offer.User e.VaultName = SharedName(e.Provider, e.Provision) - case PerConsumer: + case "pair": e.User = e.As e.VaultName = PairName(e.Provider, e.Provision, e.Consumer, e.Module, e.Local) + case "issued", "given": + // A per-holder value, as mesh-vault names it (perholder.go). + e.User = e.As + e.VaultName = HolderName(e.Provider, e.Provision, e.Consumer, e.Module) } if e.VaultName != "" { credential, err = p.opts.Vault.Value(ctx, e.VaultName) @@ -435,15 +501,30 @@ func (p *Provider) answer(ctx context.Context, e Entry, have []byte) Outcome { if p.opts.Make != nil { if err := p.opts.Make(ctx, Consumer{Entry: e, Credential: credential}); err != nil { why := err.Error() - if len(credential) > 0 { - why = strings.ReplaceAll(why, string(credential), "") - } + why = redact(why, credential) return Outcome{Key: key, Status: Failed, Why: "the provider could not make it: " + why} } } return p.write(ctx, e, have, Outcome{Key: key, Status: Answered}) } +// redact takes a credential out of text in the forms it can come back in: as it is, in base64 (standard and URL, +// padded and not) and as a JSON string says it (escapes included). +func redact(text string, credential []byte) string { + if len(credential) == 0 { + return text + } + quoted, _ := json.Marshal(string(credential)) + forms := []string{string(credential), base64.StdEncoding.EncodeToString(credential), + base64.RawStdEncoding.EncodeToString(credential), base64.URLEncoding.EncodeToString(credential), + base64.RawURLEncoding.EncodeToString(credential), string(quoted[1 : len(quoted)-1])} + sort.SliceStable(forms, func(i, j int) bool { return len(forms[i]) > len(forms[j]) }) + for _, f := range forms { + text = strings.ReplaceAll(text, f, "") + } + return text +} + func (p *Provider) write(ctx context.Context, e Entry, have []byte, o Outcome) Outcome { body, err := json.Marshal(e) if err != nil { @@ -472,6 +553,11 @@ func (p *Provider) say(key, what string) { // SharedName is the vault's one value per provider and provision (ADR 0158, issue 677). func SharedName(provider, provision string) string { return "provision/" + provider + "/" + provision } +// HolderName is the vault's value of one holder's own credential, issued or given for it (mesh-vault perholder.go). +func HolderName(provider, provision, node, module string) string { + return "provision/" + provider + "/" + provision + "/" + node + "/" + module +} + // PairName is the vault's value of one consumer's own login (ADR 0048, ADR 0158 §11, issue 677's note 12). func PairName(provider, provision, node, module, local string) string { n := "pair/" + provider + "/" + provision + "/" + node + "/" + module @@ -628,6 +714,26 @@ func (b stateBus) Put(_ context.Context, bucket, key string, value []byte) error return err } +// DeleteIf removes the key only while it still holds what the caller read; it reads the key first, so another +// writer's change in that short while can still be lost. +func (b stateBus) DeleteIf(_ context.Context, bucket, key string, held []byte) error { + s, err := b.state(bucket) + if err != nil { + return err + } + cur, err := s.Get(key) + if err != nil { + return err + } + if cur == nil { + return nil + } + if !sameJSON(cur.Value, held) { + return fmt.Errorf("%s is not as it was read: %w", key, stdio.ErrChanged) + } + return s.Delete(key) +} + func (b stateBus) Delete(_ context.Context, bucket, key string) error { s, err := b.state(bucket) if err != nil { diff --git a/go/provider/provider_test.go b/go/provider/provider_test.go index 5c7ae2f..a36644c 100644 --- a/go/provider/provider_test.go +++ b/go/provider/provider_test.go @@ -3,6 +3,7 @@ package provider import ( "bytes" "context" + "encoding/base64" "encoding/json" "errors" "fmt" @@ -12,6 +13,8 @@ import ( "sync" "testing" "time" + + stdio "git.novox.be/novox/mesh-sdk/go" ) // standIn is a stand-in bus: buckets of keys and values, and watches that hand the current values first and then @@ -65,6 +68,17 @@ func (s *standIn) Delete(_ context.Context, bucket, key string) error { return nil } +// DeleteIf removes the key only while it still holds what the caller read. +func (s *standIn) DeleteIf(ctx context.Context, bucket, key string, held []byte) error { + s.mu.Lock() + cur := s.buckets[bucket][key] + s.mu.Unlock() + if !bytes.Equal(cur, held) { + return fmt.Errorf("%s: %w", key, stdio.ErrChanged) + } + return s.Delete(ctx, bucket, key) +} + func (s *standIn) get(bucket, key string) []byte { s.mu.Lock() defer s.mu.Unlock() @@ -404,22 +418,42 @@ func TestOnlyBindsNamingThisProviderAreAnswered(t *testing.T) { } } -// An assignment that cannot be read stops the pass: a bind missed would read as an unbind and remove an entry its -// consumer still reads. -func TestAnUnreadableAssignmentRemovesNothing(t *testing.T) { - bus, m := newStandIn(), &maker{} - v := &vault{values: map[string]string{"pair/novox/postgres-database/ace/letta": "s"}} - assign(t, bus, "ace", "letta", bind("postgres-database", "", "novox", "postgres")) - p := postgres(bus, v, m) - if _, err := p.Pass(context.Background()); err != nil { - t.Fatal(err) - } - _ = bus.Put(context.Background(), AssignmentsBucket, "ace.letta", []byte("{not json")) - if _, err := p.Pass(context.Background()); err == nil { - t.Fatal("a pass over an unreadable assignment went on") - } - if bus.get(BucketOf("postgres"), "ace.letta.postgres-database") == nil || len(m.gone) != 0 { - t.Fatal("an entry was removed on an unreadable assignment") +// An assignment that cannot be read is skipped and said, by its key and why; the rest of the pass goes on (issue 825, +// finding 3). Its consumer's own entry is left alone, since a bind missed would read as an unbind. +func TestAnUnreadableAssignmentIsSkippedAndSaidAndTheRestGoesOn(t *testing.T) { + for name, bad := range map[string]string{"not json": "{not json", + "another key": `{"node":"g14","module":"umami"}`} { + bus, m := newStandIn(), &maker{} + v := &vault{values: map[string]string{"pair/novox/postgres-database/ace/letta": "s", + "pair/novox/postgres-database/g14/umami": "u"}} + assign(t, bus, "ace", "letta", bind("postgres-database", "", "novox", "postgres")) + p := postgres(bus, v, m) + if _, err := p.Pass(context.Background()); err != nil { + t.Fatal(err) + } + _ = bus.Put(context.Background(), AssignmentsBucket, "ace.letta", []byte(bad)) + assign(t, bus, "g14", "umami", bind("postgres-database", "", "novox", "postgres")) + var logged []string + p.opts.Logf = func(f string, a ...any) { logged = append(logged, fmt.Sprintf(f, a...)) } + out, err := p.Pass(context.Background()) + if err != nil { + t.Fatalf("%s: %v", name, err) + } + if bus.get(BucketOf("postgres"), "g14.umami.postgres-database") == nil { + t.Fatalf("%s: the good consumer was not answered", name) + } + if bus.get(BucketOf("postgres"), "ace.letta.postgres-database") == nil || len(m.gone) != 0 { + t.Fatalf("%s: the entry of the unreadable assignment's consumer was removed", name) + } + said := false + for _, o := range out { + if o.Key == "ace.letta" && o.Status == Failed && o.Why != "" { + said = true + } + } + if !said || !strings.Contains(strings.Join(logged, "\n"), "ace.letta") { + t.Fatalf("%s: the skipped assignment is not said: %+v, %q", name, out, logged) + } } } @@ -647,3 +681,106 @@ func TestAnEntryHasNoKeyTheStateGuardRefuses(t *testing.T) { } } } + +// The plan's credential form and holder name win over the offer's; the offer's form is the default (issue 825, +// finding 4). +func TestThePlansCredentialFormWinsOverTheOffers(t *testing.T) { + run := func(offer CredentialForm, plan map[string]string) Entry { + bus, m := newStandIn(), &maker{} + v := &vault{values: map[string]string{"pair/novox/postgres-database/ace/letta": "p", + "provision/novox/postgres-database": "s", "provision/novox/postgres-database/ace/letta": "h"}} + b := bind("postgres-database", "", "novox", "postgres") + for k, val := range plan { + b[k] = val + } + assign(t, bus, "ace", "letta", b) + p := New(Options{Node: "novox", Module: "postgres", Bus: bus, Vault: v, Make: m.Make, Logf: func(string, ...any) {}, + Offers: map[string]Offer{"postgres-database": {Credential: offer, User: "shared-user"}}}) + out, err := p.Pass(context.Background()) + if err != nil || len(out) != 1 || out[0].Status != Answered { + t.Fatalf("%v %+v", err, out) + } + return entryOf(t, bus, "ace.letta.postgres-database") + } + if e := run(PerConsumer, map[string]string{"credential-form": "shared"}); e.VaultName != + "provision/novox/postgres-database" || e.User != "shared-user" { + t.Fatalf("shared over pair: %+v", e) + } + if e := run(Shared, map[string]string{"credential-form": "pair"}); e.VaultName != + "pair/novox/postgres-database/ace/letta" || e.User != "mesh_ace_letta" { + t.Fatalf("pair over shared: %+v", e) + } + if e := run(NoCredential, map[string]string{"credential-form": "pair"}); e.VaultName != + "pair/novox/postgres-database/ace/letta" { + t.Fatalf("pair over none: %+v", e) + } + for _, form := range []string{"issued", "given"} { + if e := run(PerConsumer, map[string]string{"credential-form": form, "name": "main"}); e.VaultName != + "provision/novox/postgres-database/ace/letta" || e.User != "mesh_ace_letta" { + t.Fatalf("%s holder: %+v", form, e) + } + } + if e := run(Shared, nil); e.VaultName != "provision/novox/postgres-database" { + t.Fatalf("the offer's form is the default: %+v", e) + } +} + +// An instance deletes only an entry it wrote, and only while the entry is as it read it: one that another instance of +// the provider wrote after the consumer moved stands (issue 825, finding 5). +func TestAnEntryAnotherInstanceWroteAfterTheMoveStands(t *testing.T) { + bus, m := newStandIn(), &maker{} + v := &vault{values: map[string]string{"pair/novox/postgres-database/ace/letta": "s"}} + assign(t, bus, "ace", "letta", bind("postgres-database", "", "novox", "postgres")) + a := postgres(bus, v, m) + if _, err := a.Pass(context.Background()); err != nil { + t.Fatal(err) + } + // The consumer moves to the instance on shanks: A's pass reads its own entry and the unbind, and while it lets + // the consumer go, B writes the consumer's entry. + _ = bus.Delete(context.Background(), AssignmentsBucket, "ace.letta") + theirs, _ := json.Marshal(Entry{Provider: "shanks", ProviderModule: "postgres", Provision: "postgres-database", + Consumer: "ace", Module: "letta", VaultName: "pair/shanks/postgres-database/ace/letta"}) + a.opts.Gone = func(ctx context.Context, e Entry) error { + return bus.Put(ctx, BucketOf("postgres"), e.Key(), theirs) + } + out, err := a.Pass(context.Background()) + if err != nil { + t.Fatal(err) + } + if got := bus.get(BucketOf("postgres"), "ace.letta.postgres-database"); !bytes.Equal(got, theirs) { + t.Fatalf("the other instance's entry was removed: %s", got) + } + for _, o := range out { + if o.Status == Removed { + t.Fatalf("outcomes %+v", out) + } + } +} + +// A failed Make is said with the credential taken out in the forms it can come back in: as it is, in base64 (standard +// and URL, with and without padding) and as a JSON string says it (issue 825, finding 6). +func TestAFailedMakeIsSaidWithoutTheCredentialInAnyOfItsForms(t *testing.T) { + cred := "pa/ss+w\"rd?>>x" + b := []byte(cred) + jsonQuoted, _ := json.Marshal(cred) + forms := map[string]string{ + "raw": cred, "std": base64.StdEncoding.EncodeToString(b), "std unpadded": base64.RawStdEncoding.EncodeToString(b), + "url": base64.URLEncoding.EncodeToString(b), "url unpadded": base64.RawURLEncoding.EncodeToString(b), + "json": string(jsonQuoted[1 : len(jsonQuoted)-1]), + } + if forms["std"] == forms["url"] || forms["json"] == forms["raw"] { + t.Fatal("the test credential does not tell the forms apart") + } + for name, form := range forms { + bus, m := newStandIn(), &maker{fail: errors.New("login failed for " + form + " at the server")} + v := &vault{values: map[string]string{"pair/novox/postgres-database/ace/letta": cred}} + assign(t, bus, "ace", "letta", bind("postgres-database", "", "novox", "postgres")) + out, err := postgres(bus, v, m).Pass(context.Background()) + if err != nil || len(out) != 1 { + t.Fatalf("%v %+v", err, out) + } + if strings.Contains(out[0].Why, form) || !strings.Contains(out[0].Why, "") { + t.Fatalf("%s form: %q", name, out[0].Why) + } + } +} diff --git a/go/provision.go b/go/provision.go index 03a7adc..e4d65c7 100644 --- a/go/provision.go +++ b/go/provision.go @@ -6,7 +6,7 @@ import ( "time" ) -// A module's code reads its values at runtime (novox/hq ADR 0349 clause 14, ADR 0370, design 57 S1): its config +// A module's code reads its values at runtime (novox/hq ADR 0349 clause 14, ADR 0370, and the part of the feature 57 design that lets a running module read, rather than be handed in a file, its provisioned values): its config // values and its bindings' values that are not secret, for what its manifest declares, and watches them. The // runtime that launched this bundle asks its machine's provisioner client **as this bundle's module** — the module it // launched, never a name this code passes — and the client answers only what that module declares, only while the