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

@@ -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: `<node>.<module>.<provision>`, and `.<local>` 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
+// its entry, and each entry of this provider no bind asks for is removed. An assignment that cannot be read, or that
-// the pass before anything is removed. Run passes by itself; Pass is for a provider that passes when it chooses.
+// 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; "+
+			o := Outcome{Key: key, Status: Failed, Why: fmt.Sprintf("the assignment %s is not the controller's, so "+
-				"nothing is answered or removed", key, p.opts.Module, p.opts.Node)
+				"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 `<node>.<module>` 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 {
+	// The plan's form wins; the offer's is the default.
-	case Shared:
+	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 = redact(why, credential)
-				why = strings.ReplaceAll(why, string(credential), "<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, "<credential>")
+	}
+	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 {

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
+// An assignment that cannot be read is skipped and said, by its key and why; the rest of the pass goes on (issue 825,
-// consumer still reads.
+// finding 3). Its consumer's own entry is left alone, since a bind missed would read as an unbind.
-func TestAnUnreadableAssignmentRemovesNothing(t *testing.T) {
+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"}}
+		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("{not json"))
+		_ = bus.Put(context.Background(), AssignmentsBucket, "ace.letta", []byte(bad))
-	if _, err := p.Pass(context.Background()); err == nil {
+		assign(t, bus, "g14", "umami", bind("postgres-database", "", "novox", "postgres"))
-		t.Fatal("a pass over an unreadable assignment went on")
+		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.Fatal("an entry was removed on an unreadable assignment")
+			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, "<credential>") {
+			t.Fatalf("%s form: %q", name, out[0].Why)
+		}
+	}
+}

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