cmd/mesh-host/nodeagent.go

@@ -556,9 +556,15 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, sess
 		// its own key, `<node>.node`, under candidates.
 		shadowHooks := hooks
 		shadowHooks.refused = nil
-		shadow := selfcompose.NewShadow(composerEnv(session, node, say, shadowHooks),
+		shadowEnv := composerEnv(session, node, say, shadowHooks)
+		// And, after each answer, what runs of an own secret the answer says the vault holds another value of, handed to
+		// the vault while the node applies the controller's compositions (issue 1126, cause C4; ownresync).
+		composes := &composesNow{}
+		resync := ownResync(session, node, say, shadowEnv, composes)
+		shadow := selfcompose.NewShadow(shadowEnv,
 			func(ctx context.Context, answers map[string]delivery.CandidateAnswer) {
-				session.NoteCandidates(ctx, answers)
+				session.NoteCandidates(ctx, delivery.WithValuesDiffer(answers))
+				go resync.take(ctx, answers)
 			})
 		go shadow.Keep(run, link.Backoff) // this run's alone: a switch begins a run, and its shadow, again
 		composer, why := selfComposer(linked, session, node, say, hooks)
@@ -569,12 +575,15 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, sess
 			continue
 		}
 		if composer != nil {
+			composes.set(delivery.ComposesOwn)
 			env.Composed, env.SelfComposed, env.Trusted, env.Policy = composer, true, nil, composer.Policy
 			session.NoteComposes(linked, delivery.ComposesOwn, why)
 			// And once the catalogue no longer places it, the agent begins again on the controller's compositions: the
 			// placement key deleted ends this run, the way back (issue 1126, the seed's revert).
 			go endWhenUnplaced(run, session, node, stopRun, say)
 		} else {
+			composes.set(delivery.ComposesController)
+			go resync.again(run)
 			session.NoteComposes(linked, delivery.ComposesController, why)
 			// And once the catalogue places it, the agent begins again composing its own: the placement key appearing ends
 			// this run, so a node does not wait for a restart to follow the seed of its placement (ADR 0549 §4).

cmd/mesh-host/ownresync.go

@@ -0,0 +1,61 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"sync"
+
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+
+	"github.com/novox/mesh-host/internal/identity"
+	"github.com/novox/mesh-host/internal/link"
+	"github.com/novox/mesh-host/internal/ownresync"
+	"github.com/novox/mesh-host/internal/selfcompose"
+)
+
+// composesNow is whose compositions this run of the agent applies, once decided ("" before).
+type composesNow struct {
+	mu  sync.Mutex
+	now string
+}
+
+func (c *composesNow) set(s string) { c.mu.Lock(); c.now = s; c.mu.Unlock() }
+func (c *composesNow) get() string  { c.mu.Lock(); defer c.mu.Unlock(); return c.now }
+
+// resyncStep is the step that hands the vault what runs of an own secret a shadow answer says another value of (issue
+// 1126, cause C4; internal/ownresync): one pass at a time, and the last answers taken again once the agent has decided
+// whose compositions it applies.
+type resyncStep struct {
+	r    *ownresync.Resync
+	one  sync.Mutex
+	mu   sync.Mutex
+	last map[string]delivery.CandidateAnswer
+}
+
+func ownResync(session *link.Session, node string, say func(string), env selfcompose.Env, composes *composesNow) *resyncStep {
+	return &resyncStep{r: ownresync.New(ownresync.Env{Node: node, Composes: composes.get, Running: env.Running,
+		Get: env.Get, Unseal: env.Unseal, Seal: identity.Seal, Say: say,
+		Ask: func(ctx context.Context, verb string, args map[string]any) (json.RawMessage, error) {
+			return session.AskVault(ctx, verb, args)
+		}})}
+}
+
+// take keeps answers as the last and takes them.
+func (s *resyncStep) take(ctx context.Context, answers map[string]delivery.CandidateAnswer) {
+	s.mu.Lock()
+	s.last = answers
+	s.mu.Unlock()
+	s.again(ctx)
+}
+
+// again takes the last answers.
+func (s *resyncStep) again(ctx context.Context) {
+	s.one.Lock()
+	defer s.one.Unlock()
+	s.mu.Lock()
+	last := s.last
+	s.mu.Unlock()
+	if last != nil {
+		s.r.Take(ctx, last)
+	}
+}

go.mod

@@ -3,7 +3,7 @@ module github.com/novox/mesh-host
 go 1.26.0
 
 require (
-	git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028003907-362a3f1378aa
+	git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028084903-1062b33d32c5
 	github.com/nats-io/nats-server/v2 v2.11.17
 	github.com/nats-io/nats.go v1.54.0
 	golang.org/x/crypto v0.57.0

internal/ownresync/ownresync.go

@@ -0,0 +1,292 @@
+// Package ownresync hands the vault the value of this machine's own secret that runs, where the vault holds another
+// (novox/hq issue 1126, cause C4).
+//
+// The vault's `move-in` never replaces a value it holds, so its first import of an own secret stays fixed; where the
+// controller later made that secret anew (a rotation at start, a re-issue), the new value runs and the vault keeps the
+// old one. The node's shadow composer sees it: composing from the vault's value, it opens both seals with this
+// machine's key and its answer's shown body says delivery.MaskedAnotherValue. After each shadow answer this step asks the
+// vault's `move-in` with replace_running: true and what runs, for each own secret so masked, once per running value, and
+// says what it asked and what the vault answered. It is move-in's, the verb this node may ask already, so no seat and no
+// bus grant changes for it.
+//
+// **Only while this node applies the controller's compositions**, and only for a module whose composition that runs is
+// the one its want names: then the controller's value is what runs, and what runs is the truth the vault lacks. Once the
+// node composes its own, the vault's value is the source and nothing is handed. **The switch** is the vault's
+// (`own_take_switch`, a person's act): while it is off the vault refuses every ask, and this step says so once per value.
+// The vault also refuses any value it changed in the last 15 minutes, so a re-issue the node has not applied yet is
+// never put back.
+package ownresync
+
+import (
+	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	"encoding/json"
+	"fmt"
+	"sort"
+	"sync"
+
+	"git.novox.be/novox/mesh-sdk/go/catalogue"
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+)
+
+// Verb is the vault's verb this step asks, with ReplaceRunning true in both steps.
+const Verb = "move-in"
+
+// ReplaceRunning is move-in's argument that makes it take the running value over the one the vault holds.
+const ReplaceRunning = "replace_running"
+
+// Env is what the step reads and asks.
+type Env struct {
+	Node string
+	// Composes is whose compositions the node applies now: delivery.ComposesController, delivery.ComposesOwn, or ""
+	// while its agent has not decided. The step hands nothing but for ComposesController.
+	Composes func() string
+	// Running is what the node runs now (agent.Everything).
+	Running func() []delivery.Composed
+	// Get reads one key's latest value; nil and no error when there is none.
+	Get func(ctx context.Context, bucket, key string) ([]byte, error)
+	// Unseal opens a value sealed to this machine; Seal seals a value to a raw X25519 key in base64 (identity.Seal).
+	Unseal func(sealed string) ([]byte, error)
+	Seal   func(to string, value []byte) (string, error)
+	// Ask asks one of the vault's verbs in this engine's own name (link.Session.AskVault).
+	Ask func(ctx context.Context, verb string, args map[string]any) (json.RawMessage, error)
+	Say func(string)
+}
+
+// Outcome is what the step did for one own secret: asked or not, and what was said.
+type Outcome struct {
+	Module string
+	Name   string
+	Asked  bool
+	Said   string
+}
+
+// Resync is the step of one node agent's run.
+type Resync struct {
+	env Env
+
+	mu sync.Mutex
+	// done is, by module/name, the running value's fingerprint the vault was asked with or why it was not: each is
+	// asked, and each reason said, once.
+	done map[string]string
+}
+
+// New is the step for one node.
+func New(env Env) *Resync {
+	if env.Say == nil {
+		env.Say = func(string) {}
+	}
+	return &Resync{env: env, done: map[string]string{}}
+}
+
+// Take reads the node's shadow answers and hands the vault what runs of each own secret they say another value of.
+func (r *Resync) Take(ctx context.Context, answers map[string]delivery.CandidateAnswer) []Outcome {
+	seen := map[delivery.OwnValueDiffers]bool{}
+	var drift []delivery.OwnValueDiffers
+	groups := make([]string, 0, len(answers))
+	for g := range answers {
+		groups = append(groups, g)
+	}
+	sort.Strings(groups)
+	for _, g := range groups {
+		for _, o := range delivery.OwnValuesDiffer(answers[g]) {
+			if !seen[o] {
+				seen[o] = true
+				drift = append(drift, o)
+			}
+		}
+	}
+	if len(drift) == 0 {
+		return nil
+	}
+	composes := ""
+	if r.env.Composes != nil {
+		composes = r.env.Composes()
+	}
+	var out []Outcome
+	for _, o := range drift {
+		res := r.one(ctx, composes, o)
+		if res.Said != "" {
+			out = append(out, res)
+		}
+	}
+	return out
+}
+
+// one is the step for one own secret: Said is empty when it was done before for the same value and reason.
+func (r *Resync) one(ctx context.Context, composes string, o delivery.OwnValueDiffers) Outcome {
+	res := Outcome{Module: o.Module, Name: o.Name}
+	key := o.Module + "/" + o.Name
+	if composes == "" {
+		return res // the agent has not decided whose compositions it applies: taken again once it has
+	}
+	if composes != delivery.ComposesController {
+		return r.said(key, "not:composes", res, fmt.Sprintf("%s holds another value in the vault than runs here, and "+
+			"nothing is handed: this node composes its own (%q), so the vault's value is the source", key, composes))
+	}
+	sealed, why := r.running(ctx, o)
+	if why != "" {
+		return r.said(key, "not:"+why, res, fmt.Sprintf("%s holds another value in the vault than runs here, and "+
+			"nothing is handed: %s", key, why))
+	}
+	sum := sha256.Sum256([]byte(sealed))
+	fp := hex.EncodeToString(sum[:])
+	r.mu.Lock()
+	asked := r.done[key] == fp
+	r.mu.Unlock()
+	if asked {
+		return res
+	}
+	value, err := r.env.Unseal(sealed)
+	if err != nil {
+		return r.said(key, "not:unseal:"+fp, res, fmt.Sprintf("%s: what runs does not open with this machine's key, so "+
+			"nothing is handed: %v", key, err))
+	}
+	defer wipe(value)
+	said, err := r.hand(ctx, o, value)
+	if err != nil {
+		if ctx.Err() != nil {
+			return res // the run ended: asked again in the next
+		}
+		said = fmt.Sprintf("%s: the vault's %s with replace_running refused or did not answer: %v", key, Verb, err)
+	}
+	// Once per running value, whatever the vault answered: a refusal is the vault's word, said once, and a new value
+	// that runs asks again.
+	r.mu.Lock()
+	r.done[key] = fp
+	r.mu.Unlock()
+	res.Asked, res.Said = true, said
+	r.env.Say(said)
+	return res
+}
+
+// said says why once per reason, and keeps the reason as done.
+func (r *Resync) said(key, reason string, res Outcome, why string) Outcome {
+	r.mu.Lock()
+	again := r.done[key] == reason
+	r.done[key] = reason
+	r.mu.Unlock()
+	if again {
+		return res
+	}
+	res.Said = why
+	r.env.Say(why)
+	return res
+}
+
+// running is the sealed value of o in the composition that runs, when that composition is the one the module's want
+// names; or why it is not handed.
+func (r *Resync) running(ctx context.Context, o delivery.OwnValueDiffers) (string, string) {
+	var comp *delivery.Composed
+	if r.env.Running != nil {
+		for _, c := range r.env.Running() {
+			if c.Module == o.Module {
+				c := c
+				comp = &c
+				break
+			}
+		}
+	}
+	if comp == nil {
+		return "", o.Module + " runs no composition here"
+	}
+	wantKey, err := delivery.WantKey(r.env.Node, o.Module)
+	if err != nil {
+		return "", err.Error()
+	}
+	raw, err := r.env.Get(ctx, delivery.AssignerWantsBucket, wantKey)
+	if err != nil || raw == nil {
+		return "", fmt.Sprintf("its want %s could not be read (%v)", wantKey, err)
+	}
+	want, err := delivery.DecodeWant(raw)
+	if err != nil {
+		return "", fmt.Sprintf("its want %s is not one: %v", wantKey, err)
+	}
+	if want.Self || want.Removed || want.Composed == "" {
+		return "", fmt.Sprintf("its want %s names no composition of the controller's", wantKey)
+	}
+	// What runs is the composition the want names, as the module's have says: applied and running.
+	haveKey, err := delivery.HaveKey(r.env.Node, o.Module)
+	if err != nil {
+		return "", err.Error()
+	}
+	raw, err = r.env.Get(ctx, delivery.HaveBucket, haveKey)
+	if err != nil || raw == nil {
+		return "", fmt.Sprintf("its have %s could not be read (%v)", haveKey, err)
+	}
+	have, err := delivery.DecodeHave(raw)
+	if err != nil {
+		return "", fmt.Sprintf("its have %s is not one: %v", haveKey, err)
+	}
+	if have.Composed != want.Composed || have.Phase != delivery.PhaseRunning {
+		return "", fmt.Sprintf("what runs (%s, %s) is not yet the composition its want names (%s)", have.Composed,
+			have.Phase, want.Composed)
+	}
+	id := catalogue.NeedID(o.Name)
+	for _, res := range comp.Resources {
+		var body struct {
+			ID     string `json:"id"`
+			Sealed any    `json:"sealed"`
+		}
+		if json.Unmarshal(res, &body) != nil || body.ID != id {
+			continue
+		}
+		if s, ok := body.Sealed.(string); ok && s != "" {
+			return s, ""
+		}
+		return "", fmt.Sprintf("its resource %s carries no value sealed to this machine", id)
+	}
+	return "", fmt.Sprintf("what runs holds no resource %s", id)
+}
+
+// hand asks the vault's two steps with value, and checks what the vault holds after by its fingerprint.
+func (r *Resync) hand(ctx context.Context, o delivery.OwnValueDiffers, value []byte) (string, error) {
+	raw, err := r.env.Ask(ctx, Verb, map[string]any{ReplaceRunning: true, "module": o.Module, "name": o.Name,
+		"why": "this node's shadow answer says the vault holds another value than runs here"})
+	if err != nil {
+		return "", err
+	}
+	var opened struct {
+		Call   string `json:"call"`
+		SealTo string `json:"seal_to"`
+		Secret string `json:"secret"`
+	}
+	if err := json.Unmarshal(raw, &opened); err != nil || opened.Call == "" || opened.SealTo == "" {
+		return "", fmt.Errorf("the vault answered no call and no key to seal to")
+	}
+	sealed, err := r.env.Seal(opened.SealTo, value)
+	if err != nil {
+		return "", fmt.Errorf("the vault's key for the call is not a sealing key: %w", err)
+	}
+	raw, err = r.env.Ask(ctx, Verb, map[string]any{ReplaceRunning: true, "call": opened.Call, "sealed": sealed})
+	if err != nil {
+		return "", err
+	}
+	var kept struct {
+		Secret      string `json:"secret"`
+		From        int    `json:"from_generation"`
+		To          int    `json:"to_generation"`
+		Unchanged   bool   `json:"unchanged"`
+		Fingerprint string `json:"fingerprint"`
+		Said        string `json:"said"`
+	}
+	if err := json.Unmarshal(raw, &kept); err != nil || kept.Secret == "" {
+		return "", fmt.Errorf("the vault answered %s with no secret", opened.Secret)
+	}
+	sum := sha256.Sum256(value)
+	if kept.Fingerprint != "sha256:"+hex.EncodeToString(sum[:]) {
+		return "", fmt.Errorf("the vault says it holds %s at generation %d, but not the value handed in", kept.Secret, kept.To)
+	}
+	if kept.Unchanged {
+		return fmt.Sprintf("handed the vault what runs of %s: it held that value already (generation %d)", kept.Secret, kept.To), nil
+	}
+	return fmt.Sprintf("handed the vault what runs of %s: generation %d replaced by %d, the value that runs", kept.Secret,
+		kept.From, kept.To), nil
+}
+
+func wipe(b []byte) {
+	for i := range b {
+		b[i] = 0
+	}
+}

internal/ownresync/ownresync_test.go

@@ -0,0 +1,102 @@
+package ownresync
+
+import (
+	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"strings"
+	"testing"
+
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+)
+
+// Issue 1126, cause C4: after a shadow answer saying the vault holds another value of an own secret than runs, the node
+// hands the vault what runs, in the verb's two steps, once per running value; and only while it applies the
+// controller's compositions and runs the one its want names.
+func TestTheNodeHandsTheVaultWhatRunsOfADriftedOwnSecret(t *testing.T) {
+	const running = "the-value-that-runs"
+	digest := "sha256:" + strings.Repeat("a", 64)
+	fp := "src2:" + strings.Repeat("b", 64)
+	sum := sha256.Sum256([]byte(running))
+	answers := map[string]delivery.CandidateAnswer{"g": {Head: "h", Verdict: delivery.CandidateDiffers,
+		Differs: []delivery.ModuleDiffers{{Module: "onlyoffice", Why: "the resource differs", Bodies: []delivery.ResourceBodies{
+			{ID: "needs-jwt", Composed: `{"sealed":"` + delivery.MaskedAnotherValue + `"}`, Running: `{"sealed":"` + delivery.MaskedAnotherValue + `"}`},
+			{ID: "needs-salt", Composed: `{"sealed":"<masked: another seal of the same value>"}`}}}}}}
+	for name, row := range map[string]struct {
+		composes string
+		runs     string // the have's composition digest
+		refuse   string // the vault's refusal
+		asks     int    // the vault's verb asked, both passes
+		says     string
+	}{
+		"applies the controller's, runs its want": {delivery.ComposesController, digest, "", 2, "generation 1 replaced by 2"},
+		"composes its own":                        {delivery.ComposesOwn, digest, "", 0, "composes its own"},
+		"not decided yet":                         {"", digest, "", 0, ""},
+		"runs not yet what its want names":        {delivery.ComposesController, "sha256:" + strings.Repeat("c", 64), "", 0, "not yet the composition"},
+		"the vault refuses":                       {delivery.ComposesController, digest, "switched off", 1, "switched off"},
+	} {
+		var asked []map[string]any
+		var said []string
+		want, _ := delivery.EncodeWant(delivery.Want{FP: fp, Composed: digest})
+		have, _ := delivery.EncodeHave(delivery.Have{FP: fp, Composed: row.runs, Phase: delivery.PhaseRunning, Health: delivery.HealthOK})
+		r := New(Env{Node: "novox", Composes: func() string { return row.composes },
+			Running: func() []delivery.Composed {
+				return []delivery.Composed{{Node: "novox", Module: "onlyoffice", FP: fp, Resources: []json.RawMessage{
+					json.RawMessage(`{"id":"needs-jwt","type":"file","path":"/etc/jwt","sealed":"box-of-the-running-value"}`)}}}
+			},
+			Get: func(_ context.Context, bucket, key string) ([]byte, error) {
+				switch {
+				case bucket == delivery.AssignerWantsBucket && key == "want.novox.onlyoffice":
+					return want, nil
+				case bucket == delivery.HaveBucket && key == "novox.onlyoffice":
+					return have, nil
+				}
+				return nil, nil
+			},
+			Unseal: func(sealed string) ([]byte, error) {
+				if sealed != "box-of-the-running-value" {
+					return nil, errors.New("not this machine's")
+				}
+				return []byte(running), nil
+			},
+			Seal: func(to string, value []byte) (string, error) { return to + "(" + string(value) + ")", nil },
+			Ask: func(_ context.Context, verb string, args map[string]any) (json.RawMessage, error) {
+				asked = append(asked, args)
+				if verb != "move-in" || args["replace_running"] != true {
+					t.Errorf("%s: asked %s with %v, not move-in with replace_running", name, verb, args)
+				}
+				if row.refuse != "" {
+					return nil, errors.New(row.refuse)
+				}
+				if args["call"] == nil {
+					if args["module"] != "onlyoffice" || args["name"] != "jwt" || args["node"] != nil {
+						t.Errorf("%s: the first step names %v", name, args)
+					}
+					return json.RawMessage(`{"call":"c1","seal_to":"key","secret":"own/novox/onlyoffice/jwt"}`), nil
+				}
+				if args["sealed"] != "key("+running+")" {
+					t.Errorf("%s: handed %v, not what runs sealed to the call's key", name, args["sealed"])
+				}
+				return json.RawMessage(`{"secret":"own/novox/onlyoffice/jwt","from_generation":1,"to_generation":2,` +
+					`"fingerprint":"sha256:` + hex.EncodeToString(sum[:]) + `"}`), nil
+			},
+			Say: func(s string) { said = append(said, s) }})
+		first := r.Take(context.Background(), answers)
+		again := r.Take(context.Background(), answers)
+		if len(asked) != row.asks {
+			t.Errorf("%s: the vault asked %d times, want %d: %v", name, len(asked), row.asks, asked)
+		}
+		if len(again) != 0 {
+			t.Errorf("%s: the same answer did or said something again: %+v", name, again)
+		}
+		all := strings.Join(said, "\n")
+		if row.says == "" && (len(first) != 0 || all != "") || row.says != "" && (len(first) != 1 || !strings.Contains(all, row.says)) {
+			t.Errorf("%s: did %+v and said %q, want %q", name, first, all, row.says)
+		}
+		if strings.Contains(all, running) {
+			t.Errorf("%s: the value is in what the node said", name)
+		}
+	}
+}