diff --git a/services/nvpair-cluster-manager/spec.md b/services/nvpair-cluster-manager/spec.md index ea65943e..fc633ffe 100644 --- a/services/nvpair-cluster-manager/spec.md +++ b/services/nvpair-cluster-manager/spec.md @@ -164,6 +164,7 @@ All traffic is JSON-RPC 2.0. The local interface uses `stdin`/`stdout` (or a nam | `cluster:invite-declined` | CM → Broker | notification | Inviter-side: the joiner declined; carries the `Invite` with `state:"declined"` so the UI can clear the PIN / abandon a throwaway solo cluster. | | `cluster:invite-failed` | CM → Broker | notification | Inviter-side: the Completion Exchange failed (e.g. a wrong PIN); carries the `Invite` with `state:"failed"` (and `reason:"incorrect-pin"` on a wrong PIN) so the UI can show an "Incorrect PIN" error / abandon a throwaway solo cluster. | | `cluster:identity-changed` | CM → Broker | notification | The local `clusterId` changed (e.g. the joiner adopted the inviter's on pairing); payload `{clusterId, clusterFriendlyName}` so the Broker can persist it to `nvpair-node-settings`. | +| `cluster:trust-changed` | CM → Broker | notification | The trusted-peer store changed, or live authorization was revoked; empty payload. The Broker refreshes consumers that cache trust-derived state. | | `nodes:changed` _(proposed, §4)_ | CM → Broker | notification | Full membership snapshot pushed on every change. | - **`cluster:get-node-id` (Broker → CM)**: pure read of *this* node's own identity. Params: none. Result `{nodeUuid, nodeId, name, certFingerprint, clusterId}`. The `nodeUuid` is the self-generated, persisted identity (§7.4); `nodeId`/`name` default to the OS hostname; `clusterId` is `""` until set, created, or adopted — a `""` here is the unclustered signal a UI can surface (e.g. render "not in a cluster" or offer an explicit `cluster:create`). The caller no longer needs to create before inviting: `cluster:invite-node` auto-founds when unclustered (§7.0/§7.2), so this read is informational rather than a required gate on the invite path. Always available (the identity is minted at startup), so the UI can render "this is me" immediately. @@ -183,6 +184,7 @@ All traffic is JSON-RPC 2.0. The local interface uses `stdin`/`stdout` (or a nam - **`cluster:invite-received` (CM → Broker)**: emitted once the inbound Initial Exchange completes. Params: the `Invite` (with `state: "pending"`, no `pin`). The Broker prompts the user to enter the PIN shown on the inviting node, then drives `cluster:respond-to-invite`. - **`cluster:invite-canceled` (CM → Broker)**: emitted on the **joiner** when the inviter cancels a still-pending inbound invite (a `phase:"cancel"` arrives on the pairing channel, §7.2), or when a newer invite from the same authenticated sender supersedes it. Params: the `Invite` (now `state: "canceled"`, no `pin`). The joiner drops the old pending-inbound invite/session; on supersession it retains the sender's single pending member row for the replacement invite. The Broker uses this event to dismiss the old PIN prompt before handling the replacement `cluster:invite-received`. Idempotent — a duplicate or late cancel for an already-resolved invite is a no-op and emits nothing. - **`cluster:identity-changed` (CM → Broker)**: emitted when the local `clusterId` originates **here** rather than from a `cluster:set-identity` call — i.e. in the two cases where this service creates or adopts a cluster identity: (a) `cluster:create` founding a new cluster, and (b) the joiner adopting the inviter's cluster on a successful pair (§4 working model). Params `{clusterId, clusterFriendlyName}`; the Broker persists it to `nvpair-node-settings`. Not emitted for changes the Broker itself drove via `cluster:set-identity` (those already came from node-settings). +- **`cluster:trust-changed` (CM → Broker)**: emitted after a trusted peer's pin is added, updated, or removed, or its live authorization is forgotten. An endorsement merge triggers it only after a new endorsement is persisted. Duplicate endorsements, same-cert and same-admission re-pins with no new endorsement, empty merges, missing endorsement targets, and failed endorsement writes remain silent. The payload is `{}`: the Broker re-derives trust-dependent state rather than applying an event diff. The store releases its mutation lock before emitting the notification, so a recipient can read the updated state. - **`nodes:changed` (CM → Broker, proposed)**: full `{nodes: ClusterNode[]}` snapshot pushed whenever membership changes (accept, decline, removal, peer-initiated removal). Lets the Broker stay live without re-polling `nodes:get-initial`. Marked proposed in §4. Example `cluster:invite-node` request and result (`stdin` → `stdout`): @@ -421,7 +423,8 @@ Operations on the store map directly to single-file filesystem actions, so an ed - **Pin** (on pairing success): write `trusted/.json.tmp` then atomically rename to `trusted/.json` (`0600`). - **Remove** (`nodes:remove`): `os.Remove("trusted/.json")` — a single atomic delete; a missing file is treated as already-removed (idempotent success). - **Load** (startup): read `trusted/*.json` into the in-memory `uuid → pin` map, **skipping** `*.tmp` and any file that doesn't parse. On load, verify the file's inner `nodeUuid` (and the `certPem` subject `CN`) matches the filename ``; reject/skip-and-log a mismatch so a renamed or tampered file cannot masquerade as another UUID. -- **Re-pin guard** (§7.2): re-pinning the *same* cert for an existing `.json` is a no-op; a *different* cert for an already-present `.json` is rejected, not overwritten (explicit re-pin/re-invite required — §12 key rotation). +- **Re-pin guard** (§7.2): re-pinning the *same* cert and admission epoch for an existing `.json` merges newly received endorsements. An older admission epoch is ignored; a newer one for the same cert is persisted as a new incarnation. A *different* cert for an already-present `.json` is rejected, not overwritten (explicit re-pin/re-invite required — §12 key rotation). +- **Endorsement merge**: both an identical re-pin and a direct addition to an existing pin deduplicate endorsements by signer and signature. A merge with no new endorsement does not rewrite the file or emit `cluster:trust-changed`; a failed endorsement write leaves the live and stored pin unchanged and emits nothing. Adding endorsements for a missing peer is a no-op and does not create a pin. Writes are serialized (membership changes are rare and human-driven), so readers tolerate seeing a just-added or just-removed file without a global lock; there is no single-file consistent snapshot, which is acceptable for this low-churn store. diff --git a/services/nvpair-cluster-manager/truststore.go b/services/nvpair-cluster-manager/truststore.go index 592897b9..aa425a11 100644 --- a/services/nvpair-cluster-manager/truststore.go +++ b/services/nvpair-cluster-manager/truststore.go @@ -207,8 +207,8 @@ func (ts *TrustStore) Pin(pin *TrustedPin) error { } // Identical re-pin: fold in any newly-seen endorsements so the // gossiped trust web thickens (idempotent on the cert itself). - err := ts.mergeEndorsementsLocked(pin.NodeUUID, pin.Endorsements) - changed = err == nil + merged, err := ts.mergeEndorsementsLocked(pin.NodeUUID, pin.Endorsements) + changed = merged return err } return fmt.Errorf("uuid %s already pinned to a different certificate; remove it first to re-pin", pin.NodeUUID) @@ -239,10 +239,10 @@ func (ts *TrustStore) writePinLocked(pin *TrustedPin) error { // mergeEndorsementsLocked unions incoming endorsements into an existing pin // (dedup by signer+signature) and persists if anything new was added. Caller // holds ts.mu. A missing pin is a silent no-op. -func (ts *TrustStore) mergeEndorsementsLocked(uuid string, incoming []Endorsement) error { +func (ts *TrustStore) mergeEndorsementsLocked(uuid string, incoming []Endorsement) (bool, error) { pin, ok := ts.pins[uuid] if !ok || len(incoming) == 0 { - return nil + return false, nil } seen := make(map[string]struct{}, len(pin.Endorsements)) for _, e := range pin.Endorsements { @@ -260,9 +260,12 @@ func (ts *TrustStore) mergeEndorsementsLocked(uuid string, incoming []Endorsemen added = true } if !added { - return nil + return false, nil } - return ts.writePinLocked(updated) + if err := ts.writePinLocked(updated); err != nil { + return false, err + } + return true, nil } func endorsementKey(e Endorsement) string { @@ -279,8 +282,8 @@ func (ts *TrustStore) AddEndorsements(uuid string, endorsements []Endorsement) e defer ts.announce(&changed) ts.mu.Lock() defer ts.mu.Unlock() - err := ts.mergeEndorsementsLocked(uuid, endorsements) - changed = err == nil + merged, err := ts.mergeEndorsementsLocked(uuid, endorsements) + changed = merged return err } diff --git a/services/nvpair-cluster-manager/truststore_announce_test.go b/services/nvpair-cluster-manager/truststore_announce_test.go index 1fb7c387..054e9521 100644 --- a/services/nvpair-cluster-manager/truststore_announce_test.go +++ b/services/nvpair-cluster-manager/truststore_announce_test.go @@ -4,6 +4,13 @@ package main import ( + "bytes" + "errors" + "os" + "path/filepath" + "reflect" + "sync" + "sync/atomic" "testing" "time" ) @@ -44,8 +51,8 @@ func testPin(t *testing.T, uuid string) *TrustedPin { // announcement exists to prevent — so the hook lives on the store rather than at // the ~19 call sites that pin and unpin peers. // -// Pinning, removing, forgetting, and a display-name update all mutate what is on -// disk and must each announce exactly once. +// Pinning, removing, and a display-name update mutate disk state; forgetting +// intentionally changes only live authorization. Each must announce once. func TestTrustStoreAnnouncesEveryMutation(t *testing.T) { ts, count := newAnnouncingStore(t) const uuid = "principal-peer" @@ -80,6 +87,215 @@ func TestTrustStoreAnnouncesEveryMutation(t *testing.T) { } } +// assertStoredEndorsements checks the live snapshot and a separately loaded +// store. It is also safe to call from an onChange callback in a joined worker. +func assertStoredEndorsements(t *testing.T, ts *TrustStore, uuid string, want []Endorsement) { + t.Helper() + pin, ok := ts.Get(uuid) + if !ok || !reflect.DeepEqual(pin.Endorsements, want) { + t.Errorf("live endorsements = %+v, want %+v", pin, want) + } + reloaded, err := newTrustStore(filepath.Dir(ts.dir)) + if err != nil { + t.Errorf("reload trust store: %v", err) + return + } + pin, ok = reloaded.Get(uuid) + if !ok || !reflect.DeepEqual(pin.Endorsements, want) { + t.Errorf("reloaded endorsements = %+v, want %+v", pin, want) + } +} + +type endorsementMerge func(*TrustStore, *TrustedPin, []Endorsement) error + +func addEndorsements(ts *TrustStore, pin *TrustedPin, batch []Endorsement) error { + return ts.AddEndorsements(pin.NodeUUID, batch) +} + +func pinWithEndorsements(ts *TrustStore, pin *TrustedPin, batch []Endorsement) error { + updated := *pin + updated.Endorsements = batch + return ts.Pin(&updated) +} + +func TestTrustStoreAnnouncesNewEndorsementsAfterPersistence(t *testing.T) { + test := func(name string, merge endorsementMerge) { + t.Run(name, func(t *testing.T) { + ts, _ := newAnnouncingStore(t) + pin := testPin(t, "principal-peer") + first := Endorsement{By: "trusted-peer", Sig: "signature-1"} + second := Endorsement{By: "trusted-peer", SigV2: "signature-2"} + pin.Endorsements = []Endorsement{first} + if err := ts.Pin(pin); err != nil { + t.Fatalf("pin: %v", err) + } + want := []Endorsement{first, second} + calls := 0 + ts.SetOnChange(func() { + calls++ + // Acquiring Get's read lock here also witnesses that the + // mutation lock was released before announcing the change. + assertStoredEndorsements(t, ts, pin.NodeUUID, want) + }) + // Mix an existing endorsement, a new endorsement, and an + // in-batch duplicate. One operation causes one announcement. + batch := []Endorsement{first, second, second} + if err := merge(ts, pin, batch); err != nil { + t.Fatalf("merge: %v", err) + } + if calls != 1 { + t.Fatalf("announcements after merge = %d, want 1", calls) + } + assertStoredEndorsements(t, ts, pin.NodeUUID, want) + before, err := os.ReadFile(ts.pinPath(pin.NodeUUID)) + if err != nil { + t.Fatal(err) + } + if err := merge(ts, pin, batch); err != nil { + t.Fatalf("duplicate merge: %v", err) + } + if calls != 1 { + t.Fatalf("announcements after duplicate merge = %d, want 1", calls) + } + if err := merge(ts, pin, nil); err != nil { + t.Fatalf("empty merge: %v", err) + } + if calls != 1 { + t.Fatalf("announcements after empty merge = %d, want 1", calls) + } + after, err := os.ReadFile(ts.pinPath(pin.NodeUUID)) + if err != nil { + t.Fatalf("read pin after no-op merges: %v", err) + } + if !bytes.Equal(before, after) { + t.Fatal("no-op merges changed disk contents") + } + assertStoredEndorsements(t, ts, pin.NodeUUID, want) + }) + } + test("AddEndorsements", addEndorsements) + test("IdenticalPin", pinWithEndorsements) +} + +func TestTrustStoreStaysSilentWhenEndorsementWriteFails(t *testing.T) { + test := func(name string, merge endorsementMerge) { + t.Run(name, func(t *testing.T) { + ts, count := newAnnouncingStore(t) + pin := testPin(t, "principal-peer") + first := Endorsement{By: "trusted-peer", Sig: "signature-1"} + second := Endorsement{By: "trusted-peer", SigV2: "signature-2"} + pin.Endorsements = []Endorsement{first} + if err := ts.Pin(pin); err != nil { + t.Fatalf("pin: %v", err) + } + before, err := os.ReadFile(ts.pinPath(pin.NodeUUID)) + if err != nil { + t.Fatal(err) + } + beforeCount := count() + // Fail the final replace, after the temporary file was written. + // The existing pin must remain intact on disk and in memory. + originalRename := renameFile + writeErr := errors.New("injected endorsement replace failure") + renameFile = func(_, _ string) error { return writeErr } + t.Cleanup(func() { renameFile = originalRename }) + if err := merge(ts, pin, []Endorsement{second}); !errors.Is(err, writeErr) { + t.Fatalf("merge error = %v, want injected replace failure", err) + } + if count() != beforeCount { + t.Fatalf("announcements after failed write = %d, want %d", count(), beforeCount) + } + after, err := os.ReadFile(ts.pinPath(pin.NodeUUID)) + if err != nil { + t.Fatalf("read pin after failed write: %v", err) + } + if !bytes.Equal(before, after) { + t.Fatal("failed write changed disk contents") + } + assertStoredEndorsements(t, ts, pin.NodeUUID, []Endorsement{first}) + entries, err := os.ReadDir(ts.dir) + if err != nil { + t.Fatalf("list trusted directory after failed write: %v", err) + } + if len(entries) != 1 || entries[0].Name() != pin.NodeUUID+".json" { + t.Fatalf("failed write left temporary residue: entries=%v", entries) + } + renameFile = originalRename + if err := merge(ts, pin, []Endorsement{second}); err != nil { + t.Fatalf("retry after storage recovery: %v", err) + } + if count() != beforeCount+1 { + t.Fatalf("announcements after retry = %d, want %d", count(), beforeCount+1) + } + assertStoredEndorsements(t, ts, pin.NodeUUID, []Endorsement{first, second}) + }) + } + test("AddEndorsements", addEndorsements) + test("IdenticalPin", pinWithEndorsements) +} + +func TestTrustStoreConcurrentDuplicateEndorsementsAnnounceOnce(t *testing.T) { + ts, err := newTrustStore(t.TempDir()) + if err != nil { + t.Fatal(err) + } + pin := testPin(t, "principal-peer") + if err := ts.Pin(pin); err != nil { + t.Fatal(err) + } + endorsement := Endorsement{By: "trusted-peer", SigV2: "signature-1"} + want := []Endorsement{endorsement} + var calls atomic.Int32 + ts.SetOnChange(func() { + calls.Add(1) + assertStoredEndorsements(t, ts, pin.NodeUUID, want) + }) + const workers = 16 + start := make(chan struct{}) + errs := make(chan error, workers) + var wg sync.WaitGroup + for i := range workers { + wg.Add(1) + go func() { + defer wg.Done() + <-start + if i%2 == 0 { + updated := *pin + updated.Endorsements = want + errs <- ts.Pin(&updated) + } else { + errs <- ts.AddEndorsements(pin.NodeUUID, want) + } + }() + } + close(start) + wg.Wait() + close(errs) + for err := range errs { + if err != nil { + t.Errorf("concurrent merge: %v", err) + } + } + if got := calls.Load(); got != 1 { + t.Errorf("announcements for concurrent identical submissions = %d, want 1", got) + } + assertStoredEndorsements(t, ts, pin.NodeUUID, want) +} + +func TestTrustStoreMissingEndorsementTargetStaysSilent(t *testing.T) { + ts, count := newAnnouncingStore(t) + if err := ts.AddEndorsements("principal-stranger", []Endorsement{{By: "trusted-peer", SigV2: "signature-1"}}); err != nil { + t.Fatal(err) + } + if count() != 0 || len(ts.List()) != 0 { + t.Fatalf("missing-target merge changed live state: announcements=%d pins=%v", count(), ts.List()) + } + entries, err := os.ReadDir(ts.dir) + if err != nil || len(entries) != 0 { + t.Fatalf("missing-target merge changed disk state: entries=%v err=%v", entries, err) + } +} + // TestTrustStoreStaysSilentWhenNothingChanged keeps the announcement meaningful. // The scanner answers it by walking its whole directory, and the broker relays // it, so a store that announced on every call — including the idempotent re-pin @@ -96,11 +312,14 @@ func TestTrustStoreStaysSilentWhenNothingChanged(t *testing.T) { before := count() // An identical re-pin folds in no new endorsements and rewrites nothing. - if err := ts.Pin(testPin(t, uuid)); err == nil { - // A fresh leaf for the same uuid is a DIFFERENT certificate, which the - // store refuses rather than silently re-pinning; either way it must not - // announce a change it did not make. - t.Log("re-pin with a new certificate was accepted") + // Reuse the exact pin: generating another fixture would mint a different + // certificate and exercise the key-rotation rejection path instead. + if err := ts.Pin(pin); err != nil { + t.Fatalf("identical re-pin: %v", err) + } + // An empty endorsement merge is also a no-op and must stay silent. + if err := ts.AddEndorsements(uuid, nil); err != nil { + t.Fatalf("empty endorsement merge: %v", err) } // A rename to the values already stored changes nothing. if ok, err := ts.UpdateIdentity(uuid, uuid, uuid); err != nil || ok {