From 974e5aca142bc7f6a2ef9121397c50baedf3305c Mon Sep 17 00:00:00 2001 From: Teodor Calin Date: Sun, 26 Jul 2026 00:50:21 +0300 Subject: [PATCH 1/2] registry: shard the trust-pair and handshake-relay locks Live mutex/block profiles from the production registry under a fleet reconverge showed two single global locks dominating: handshakeMu 50.6% of all block time (HandlePollHandshakes) trust st.mu 32.1% of all mutex time (HandleReportTrust) Every agent polls its handshake inbox on a timer, so at fleet scale all polls serialised on one mutex despite each touching only its own node id; report_trust did the same for trust pairs. Both are now sharded 256 ways by node id / pair key. Every mutating critical section already touched exactly one key, so no path needs two shards and there is no lock order to observe. Co-Authored-By: Claude Opus 5 --- trust/handshake_shard.go | 183 +++++++++++++++++++++++++++++++++++++++ trust/pairset.go | 93 ++++++++++++++++++++ trust/trust.go | 164 ++++++++--------------------------- 3 files changed, 311 insertions(+), 129 deletions(-) create mode 100644 trust/handshake_shard.go create mode 100644 trust/pairset.go diff --git a/trust/handshake_shard.go b/trust/handshake_shard.go new file mode 100644 index 0000000..dbfd8fd --- /dev/null +++ b/trust/handshake_shard.go @@ -0,0 +1,183 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +package trust + +import ( + "errors" + "sync" + "time" +) + +// handshakeShards is the shard count for handshake relay state. A power of +// two so the index is a mask. Every agent polls its handshake inbox on a +// timer, so at fleet scale a single global mutex serialised every poll even +// though each one touches only its own node id — that lock dominated the +// registry's blocking profile. Sharding by node id lets polls for different +// nodes proceed in parallel. +const handshakeShards = 256 + +func handshakeShardIdx(nodeID uint32) uint32 { + return (nodeID * 2654435761) & (handshakeShards - 1) +} + +var ( + // errInboxFull reports that a node's handshake inbox is at capacity. + errInboxFull = errors.New("handshake inbox full") + // errAlreadyPending reports a duplicate request from the same origin. + errAlreadyPending = errors.New("handshake request already pending") +) + +// handshakeState is the sharded store for handshake relay inboxes, +// responses and pending-request tracking. Every mutating operation touches +// exactly one node id, so no code path needs two shards at once and there +// is no lock ordering to observe. +type handshakeState struct { + shards [handshakeShards]struct { + mu sync.Mutex + inbox map[uint32][]*HandshakeRelayMsg + responses map[uint32][]*HandshakeResponseMsg + pending map[uint32]map[uint32]struct{} + } +} + +func newHandshakeState() *handshakeState { + hs := &handshakeState{} + for i := range hs.shards { + hs.shards[i].inbox = make(map[uint32][]*HandshakeRelayMsg) + hs.shards[i].responses = make(map[uint32][]*HandshakeResponseMsg) + hs.shards[i].pending = make(map[uint32]map[uint32]struct{}) + } + return hs +} + +// relay appends a handshake request to toNodeID's inbox and records it as +// pending. Returns errInboxFull or errAlreadyPending when rejected. +func (hs *handshakeState) relay(toNodeID, fromNodeID uint32, justification string, maxInbox int) error { + sh := &hs.shards[handshakeShardIdx(toNodeID)] + sh.mu.Lock() + defer sh.mu.Unlock() + + if len(sh.inbox[toNodeID]) >= maxInbox { + return errInboxFull + } + for _, existing := range sh.inbox[toNodeID] { + if existing.FromNodeID == fromNodeID { + return errAlreadyPending + } + } + sh.inbox[toNodeID] = append(sh.inbox[toNodeID], &HandshakeRelayMsg{ + FromNodeID: fromNodeID, + Justification: justification, + Timestamp: time.Now(), + }) + if sh.pending[toNodeID] == nil { + sh.pending[toNodeID] = make(map[uint32]struct{}) + } + sh.pending[toNodeID][fromNodeID] = struct{}{} + return nil +} + +// pop returns and clears nodeID's request and response inboxes. +func (hs *handshakeState) pop(nodeID uint32) ([]*HandshakeRelayMsg, []*HandshakeResponseMsg) { + sh := &hs.shards[handshakeShardIdx(nodeID)] + sh.mu.Lock() + defer sh.mu.Unlock() + inbox := sh.inbox[nodeID] + delete(sh.inbox, nodeID) + resp := sh.responses[nodeID] + delete(sh.responses, nodeID) + return inbox, resp +} + +// takePending consumes a pending request from peerID to nodeID, reporting +// whether one existed. +func (hs *handshakeState) takePending(nodeID, peerID uint32) bool { + sh := &hs.shards[handshakeShardIdx(nodeID)] + sh.mu.Lock() + defer sh.mu.Unlock() + pending := sh.pending[nodeID] + if _, found := pending[peerID]; !found { + return false + } + delete(pending, peerID) + if len(pending) == 0 { + delete(sh.pending, nodeID) + } + return true +} + +// appendResponse queues a handshake response for peerID to collect. +func (hs *handshakeState) appendResponse(peerID uint32, msg *HandshakeResponseMsg) { + sh := &hs.shards[handshakeShardIdx(peerID)] + sh.mu.Lock() + sh.responses[peerID] = append(sh.responses[peerID], msg) + sh.mu.Unlock() +} + +// snapshot copies every shard's inboxes for serialisation. Returns nil maps +// when empty, matching the pre-shard behaviour. +func (hs *handshakeState) snapshot() ( + inbox map[uint32][]*HandshakeRelayMsg, + responses map[uint32][]*HandshakeResponseMsg, +) { + for i := range hs.shards { + sh := &hs.shards[i] + sh.mu.Lock() + for id, msgs := range sh.inbox { + if inbox == nil { + inbox = make(map[uint32][]*HandshakeRelayMsg) + } + inbox[id] = msgs + } + for id, msgs := range sh.responses { + if responses == nil { + responses = make(map[uint32][]*HandshakeResponseMsg) + } + responses[id] = msgs + } + sh.mu.Unlock() + } + return inbox, responses +} + +// restore loads snapshotted inboxes, rebuilding pending state from the +// restored requests. Call before serving. +func (hs *handshakeState) restore( + inbox map[uint32][]*HandshakeRelayMsg, + responses map[uint32][]*HandshakeResponseMsg, +) { + for id, msgs := range inbox { + sh := &hs.shards[handshakeShardIdx(id)] + sh.mu.Lock() + sh.inbox[id] = msgs + for _, msg := range msgs { + if sh.pending[id] == nil { + sh.pending[id] = make(map[uint32]struct{}) + } + sh.pending[id][msg.FromNodeID] = struct{}{} + } + sh.mu.Unlock() + } + for id, msgs := range responses { + sh := &hs.shards[handshakeShardIdx(id)] + sh.mu.Lock() + sh.responses[id] = msgs + sh.mu.Unlock() + } +} + +// size totals pending requests and responses across shards, for metrics. +func (hs *handshakeState) size() (requests, responses int) { + for i := range hs.shards { + sh := &hs.shards[i] + sh.mu.Lock() + for _, msgs := range sh.inbox { + requests += len(msgs) + } + for _, msgs := range sh.responses { + responses += len(msgs) + } + sh.mu.Unlock() + } + return requests, responses +} diff --git a/trust/pairset.go b/trust/pairset.go new file mode 100644 index 0000000..66c3b0d --- /dev/null +++ b/trust/pairset.go @@ -0,0 +1,93 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +package trust + +import "sync" + +// trustPairShards is the shard count for the trust-pair set. A power of two +// so the index is a mask. Sized well above the registry's request +// concurrency so report_trust writes for different pairs proceed in +// parallel instead of serialising on one lock — the report_trust flood +// during a fleet reconverge was starving IsTrusted / check_trust readers. +const trustPairShards = 256 + +func trustShardIdx(key string) uint32 { + var h uint32 = 2166136261 + for i := 0; i < len(key); i++ { + h ^= uint32(key[i]) + h *= 16777619 + } + return h & (trustPairShards - 1) +} + +// trustPairSet is a sharded set of canonical "min:max" trust-pair keys. +// Each shard carries its own RWMutex. +type trustPairSet struct { + shards [trustPairShards]struct { + mu sync.RWMutex + m map[string]bool + } +} + +func newTrustPairSet() *trustPairSet { + s := &trustPairSet{} + for i := range s.shards { + s.shards[i].m = make(map[string]bool) + } + return s +} + +func (s *trustPairSet) has(key string) bool { + sh := &s.shards[trustShardIdx(key)] + sh.mu.RLock() + defer sh.mu.RUnlock() + return sh.m[key] +} + +func (s *trustPairSet) add(key string) { + sh := &s.shards[trustShardIdx(key)] + sh.mu.Lock() + sh.m[key] = true + sh.mu.Unlock() +} + +// remove deletes key, returning true if it was present. +func (s *trustPairSet) remove(key string) bool { + sh := &s.shards[trustShardIdx(key)] + sh.mu.Lock() + defer sh.mu.Unlock() + if !sh.m[key] { + return false + } + delete(sh.m, key) + return true +} + +func (s *trustPairSet) count() int { + n := 0 + for i := range s.shards { + sh := &s.shards[i] + sh.mu.RLock() + n += len(sh.m) + sh.mu.RUnlock() + } + return n +} + +func (s *trustPairSet) keys() []string { + out := make([]string, 0) + for i := range s.shards { + sh := &s.shards[i] + sh.mu.RLock() + for k := range sh.m { + out = append(out, k) + } + sh.mu.RUnlock() + } + return out +} + +// addUnlocked inserts without locking — for startup restore before serving. +func (s *trustPairSet) addUnlocked(key string) { + s.shards[trustShardIdx(key)].m[key] = true +} diff --git a/trust/trust.go b/trust/trust.go index d4aa348..ff17f64 100644 --- a/trust/trust.go +++ b/trust/trust.go @@ -8,9 +8,9 @@ package trust import ( "encoding/base64" + "errors" "fmt" "log/slog" - "sync" "time" "github.com/pilot-protocol/common/protocol" @@ -69,27 +69,23 @@ type Callbacks struct { // // LOCK ORDERING // -// mu (RWMutex) — protects trustPairs. +// trustPairs — sharded internally, one RWMutex per shard. // handshakeMu (Mutex) — protects handshakeInbox, handshakeResponses, // and pendingHandshakes. // -// These two locks are independent; neither may be acquired while holding +// These locks are independent; neither may be acquired while holding // the other. No code path needs both simultaneously. type Store struct { nodes NodeView cb Callbacks - mu sync.RWMutex - trustPairs map[string]bool - - handshakeMu sync.Mutex - handshakeInbox map[uint32][]*HandshakeRelayMsg - handshakeResponses map[uint32][]*HandshakeResponseMsg - // pendingHandshakes tracks requests that have been sent but not yet - // responded to. Unlike handshakeInbox it is NOT cleared by poll, so - // respond_handshake can verify a prior request existed even after the - // recipient has polled the inbox. Keyed as [responderID][requesterID]. - pendingHandshakes map[uint32]map[uint32]struct{} + trustPairs *trustPairSet + + // handshakes holds the relay inboxes, responses and pending-request + // tracking, sharded by node id. Pending requests are NOT cleared by + // poll, so respond_handshake can verify a prior request existed even + // after the recipient has polled its inbox. + handshakes *handshakeState } // NewStore creates an empty, ready-to-use Store. @@ -97,10 +93,8 @@ func NewStore(nodes NodeView, cb Callbacks) *Store { return &Store{ nodes: nodes, cb: cb, - trustPairs: make(map[string]bool), - handshakeInbox: make(map[uint32][]*HandshakeRelayMsg), - handshakeResponses: make(map[uint32][]*HandshakeResponseMsg), - pendingHandshakes: make(map[uint32]map[uint32]struct{}), + trustPairs: newTrustPairSet(), + handshakes: newHandshakeState(), } } @@ -108,17 +102,13 @@ func NewStore(nodes NodeView, cb Callbacks) *Store { // Count returns the total number of trust pairs currently stored. func (st *Store) Count() int { - st.mu.RLock() - defer st.mu.RUnlock() - return len(st.trustPairs) + return st.trustPairs.count() } // IsTrusted reports whether nodes a and b have an established trust pair. // The relation is symmetric: IsTrusted(a, b) == IsTrusted(b, a). func (st *Store) IsTrusted(a, b uint32) bool { - st.mu.RLock() - defer st.mu.RUnlock() - return st.trustPairs[pairKey(a, b)] + return st.trustPairs.has(pairKey(a, b)) } // --- Snapshot / restore --- @@ -126,20 +116,14 @@ func (st *Store) IsTrusted(a, b uint32) bool { // Pairs returns a snapshot of all trust-pair keys for serialisation. // Each key is of the form "min:max" where min <= max. func (st *Store) Pairs() []string { - st.mu.RLock() - defer st.mu.RUnlock() - out := make([]string, 0, len(st.trustPairs)) - for k := range st.trustPairs { - out = append(out, k) - } - return out + return st.trustPairs.keys() } // RestorePairs loads trust pairs from a snapshot during startup. It is // NOT safe for concurrent use — call it before serving requests. func (st *Store) RestorePairs(keys []string) { for _, key := range keys { - st.trustPairs[key] = true + st.trustPairs.addUnlocked(key) } } @@ -149,22 +133,7 @@ func (st *Store) InboxSnapshot() ( inbox map[uint32][]*HandshakeRelayMsg, responses map[uint32][]*HandshakeResponseMsg, ) { - st.handshakeMu.Lock() - defer st.handshakeMu.Unlock() - - if len(st.handshakeInbox) > 0 { - inbox = make(map[uint32][]*HandshakeRelayMsg, len(st.handshakeInbox)) - for id, msgs := range st.handshakeInbox { - inbox[id] = msgs - } - } - if len(st.handshakeResponses) > 0 { - responses = make(map[uint32][]*HandshakeResponseMsg, len(st.handshakeResponses)) - for id, msgs := range st.handshakeResponses { - responses[id] = msgs - } - } - return + return st.handshakes.snapshot() } // RestoreInbox loads handshake inboxes from a snapshot during startup. @@ -173,36 +142,15 @@ func (st *Store) RestoreInbox( inbox map[uint32][]*HandshakeRelayMsg, responses map[uint32][]*HandshakeResponseMsg, ) { - st.handshakeMu.Lock() - defer st.handshakeMu.Unlock() - for id, msgs := range inbox { - st.handshakeInbox[id] = msgs - // Rebuild pendingHandshakes from the restored inbox so that - // respond_handshake validation works correctly after a restart. - for _, msg := range msgs { - if st.pendingHandshakes[id] == nil { - st.pendingHandshakes[id] = make(map[uint32]struct{}) - } - st.pendingHandshakes[id][msg.FromNodeID] = struct{}{} - } - } - for id, msgs := range responses { - st.handshakeResponses[id] = msgs - } + // Pending state is rebuilt from the restored requests so that + // respond_handshake validation works correctly after a restart. + st.handshakes.restore(inbox, responses) } // InboxSize returns the total number of pending handshake requests and // responses (for metrics gauges). func (st *Store) InboxSize() (requests, responses int) { - st.handshakeMu.Lock() - defer st.handshakeMu.Unlock() - for _, msgs := range st.handshakeInbox { - requests += len(msgs) - } - for _, msgs := range st.handshakeResponses { - responses += len(msgs) - } - return + return st.handshakes.size() } // --- Handlers --- @@ -240,9 +188,7 @@ func (st *Store) HandleReportTrust(req map[string]interface{}) (map[string]inter } key := pairKey(nodeA, nodeB) - st.mu.Lock() - st.trustPairs[key] = true - st.mu.Unlock() + st.trustPairs.add(key) st.cb.Save() st.cb.IncTrustReports() @@ -278,13 +224,9 @@ func (st *Store) HandleRevokeTrust(req map[string]interface{}) (map[string]inter } key := pairKey(nodeA, nodeB) - st.mu.Lock() - if !st.trustPairs[key] { - st.mu.Unlock() + if !st.trustPairs.remove(key) { return nil, fmt.Errorf("no trust pair between %d and %d", nodeA, nodeB) } - delete(st.trustPairs, key) - st.mu.Unlock() st.cb.Save() st.cb.IncTrustRevocations() @@ -318,9 +260,7 @@ func (st *Store) HandleCheckTrust(req map[string]interface{}) (map[string]interf } } - st.mu.RLock() - trusted := st.trustPairs[pairKey(nodeA, nodeB)] - st.mu.RUnlock() + trusted := st.trustPairs.has(pairKey(nodeA, nodeB)) if !trusted { _, netsA, okA := st.nodes.LookupNode(nodeA) @@ -390,29 +330,14 @@ func (st *Store) HandleRequestHandshake(req map[string]interface{}) (map[string] return nil, fmt.Errorf("node %d: %w", toNodeID, protocol.ErrNodeNotFound) } - // Phase 3b: inbox mutation under handshakeMu - st.handshakeMu.Lock() - defer st.handshakeMu.Unlock() - - if len(st.handshakeInbox[toNodeID]) >= maxHandshakeInbox { + // Phase 3b: inbox mutation, sharded by recipient node id + switch err := st.handshakes.relay(toNodeID, fromNodeID, justification, maxHandshakeInbox); { + case errors.Is(err, errInboxFull): return nil, fmt.Errorf("handshake inbox full for node %d", toNodeID) - } - for _, existing := range st.handshakeInbox[toNodeID] { - if existing.FromNodeID == fromNodeID { - return nil, fmt.Errorf("handshake request already pending from node %d", fromNodeID) - } + case errors.Is(err, errAlreadyPending): + return nil, fmt.Errorf("handshake request already pending from node %d", fromNodeID) } - st.handshakeInbox[toNodeID] = append(st.handshakeInbox[toNodeID], &HandshakeRelayMsg{ - FromNodeID: fromNodeID, - Justification: justification, - Timestamp: time.Now(), - }) - if st.pendingHandshakes[toNodeID] == nil { - st.pendingHandshakes[toNodeID] = make(map[uint32]struct{}) - } - st.pendingHandshakes[toNodeID][fromNodeID] = struct{}{} - st.cb.IncHandshakeRequests() slog.Info("handshake request relayed", "from", fromNodeID, "to", toNodeID) @@ -443,13 +368,8 @@ func (st *Store) HandlePollHandshakes(req map[string]interface{}) (map[string]in return nil, err } - // Phase 3: handshakeMu protects only handshake state - st.handshakeMu.Lock() - inbox := st.handshakeInbox[nodeID] - delete(st.handshakeInbox, nodeID) - respInbox := st.handshakeResponses[nodeID] - delete(st.handshakeResponses, nodeID) - st.handshakeMu.Unlock() + // Phase 3: drain this node's shard only + inbox, respInbox := st.handshakes.pop(nodeID) requests := make([]map[string]interface{}, len(inbox)) for i, r := range inbox { @@ -524,24 +444,12 @@ func (st *Store) HandleRespondHandshake(req map[string]interface{}) (map[string] // pendingHandshakes is populated by HandleRequestHandshake and survives // poll drains, so this check holds even after the recipient has polled. if accept { - st.handshakeMu.Lock() - pending := st.pendingHandshakes[nodeID] - _, found := pending[peerID] - if found { - delete(pending, peerID) - if len(pending) == 0 { - delete(st.pendingHandshakes, nodeID) - } - } - st.handshakeMu.Unlock() - if !found { + if !st.handshakes.takePending(nodeID, peerID) { return nil, fmt.Errorf("no pending handshake request from node %d", peerID) } key := pairKey(nodeID, peerID) - st.mu.Lock() - st.trustPairs[key] = true - st.mu.Unlock() + st.trustPairs.add(key) st.cb.Save() slog.Info("handshake approved via relay, trust pair created", "node", nodeID, "peer", peerID) } else { @@ -549,14 +457,12 @@ func (st *Store) HandleRespondHandshake(req map[string]interface{}) (map[string] } st.cb.Audit("handshake.responded", "node_id", nodeID, "peer_id", peerID, "accept", accept) - // Phase 3c: response inbox append under handshakeMu - st.handshakeMu.Lock() - st.handshakeResponses[peerID] = append(st.handshakeResponses[peerID], &HandshakeResponseMsg{ + // Phase 3c: response inbox append, sharded by recipient node id + st.handshakes.appendResponse(peerID, &HandshakeResponseMsg{ FromNodeID: nodeID, Accept: accept, Timestamp: time.Now(), }) - st.handshakeMu.Unlock() return map[string]interface{}{ "type": "respond_handshake_ok", From 0ef4bd489257c15019f5074a276e78f87ebf3733 Mon Sep 17 00:00:00 2001 From: Teodor Calin Date: Sun, 26 Jul 2026 15:14:56 +0300 Subject: [PATCH 2/2] registry: constant-time admin-token compare, accurate WAL durability doc, opt-in heartbeat freshness + set_key_expiry binding MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit L3 — trust.checkAdminToken compared the supplied admin token with the string operator, so a rejection returned as soon as a byte differed and the time it took leaked how much of the token was right. Switched to subtle.ConstantTimeCompare, matching every sibling admin-token check in the repo (authz, membership, dashboard, accept). This was the only divergence. L4 — wal.Append's doc claimed the entry is fsync'd. The write(2) is synchronous, the fsync is batched (walSyncInterval / walSyncBatch), so host power loss can lose a bounded tail. Corrected the comment on Append, the package doc, and the constants rather than fsyncing every append, which would change throughput on the write path. L2 and M12 both change what a client has to sign, so both are behind default-off flags mirroring -strict-registration-auth. Default behaviour is unchanged. L2 — the heartbeat challenge covered only the node id, so a captured heartbeat stayed valid indefinitely and could hold a node that is gone in the "seen recently" state. -strict-heartbeat-freshness / RENDEZVOUS_STRICT_HEARTBEAT_FRESHNESS=1 requires a "ts" field within heartbeatMaxSkew and binds it into the challenge. Enabling it also bypasses the 120s signature-verification cache, which would otherwise wave through anything sent after one accepted heartbeat, and refuses the binary heartbeat encoding: its payload is a fixed node id + signature with no field for a timestamp, so leaving it open would make the gate decorative. Carrying a timestamp there needs a registry wire-format revision and is not attempted here. M12 — the set_key_expiry challenge covered only the node id, so one signature authorized any expiry for that node, including substituting a distant one. -strict-expiry-binding / RENDEZVOUS_STRICT_EXPIRY_BINDING=1 appends the requested expires_at to the challenge, the way HandleRotateKey's challenge appends the new public key. Co-Authored-By: Claude Opus 5 --- cmd/rendezvous/main.go | 10 + directory/directory.go | 89 ++++++- directory/zz_heartbeat_freshness_test.go | 312 +++++++++++++++++++++++ identity/identity.go | 29 ++- server.go | 7 + server_lifecycle.go | 42 ++- trust/trust.go | 8 +- trust/zz_admin_token_compare_test.go | 126 +++++++++ wal/wal.go | 21 +- wal/zz_append_durability_test.go | 97 +++++++ zz_set_key_expiry_binding_test.go | 101 ++++++++ 11 files changed, 831 insertions(+), 11 deletions(-) create mode 100644 directory/zz_heartbeat_freshness_test.go create mode 100644 trust/zz_admin_token_compare_test.go create mode 100644 wal/zz_append_durability_test.go create mode 100644 zz_set_key_expiry_binding_test.go diff --git a/cmd/rendezvous/main.go b/cmd/rendezvous/main.go index 3d97bf6..cebdf96 100644 --- a/cmd/rendezvous/main.go +++ b/cmd/rendezvous/main.go @@ -68,6 +68,8 @@ func main() { enableTLS := flag.Bool("tls", false, "enable TLS for registry connections") strictDirectoryAuth := flag.Bool("strict-directory-auth", false, "WS2: require trust/shared-network authorization on directory RPCs (lookup/resolve/punch/list_*/check_trust). Default false (not enforcing, wire-compatible with old agents). Env: RENDEZVOUS_STRICT_DIRECTORY_AUTH=1.") strictRegistrationAuth := flag.Bool("strict-registration-auth", false, "require a valid proof-of-possession signature on registrations that submit a public_key. Default false (signatures verified when present, but unsigned registrations from old agents are still accepted). Env: RENDEZVOUS_STRICT_REGISTRATION_AUTH=1.") + strictHeartbeatFreshness := flag.Bool("strict-heartbeat-freshness", false, "require heartbeats to carry a recent ts field and bind it into the signed challenge, and refuse the binary heartbeat encoding (which has no timestamp field). Default false (challenge covers only the node id, so a captured heartbeat stays replayable). Enable only once every client signs the bound form. Env: RENDEZVOUS_STRICT_HEARTBEAT_FRESHNESS=1.") + strictExpiryBinding := flag.Bool("strict-expiry-binding", false, "bind the requested expires_at into the set_key_expiry signed challenge, so a signature authorizes exactly the expiry it was produced for. Default false (challenge covers only the node id). Enable only once every client signs the bound form. Env: RENDEZVOUS_STRICT_EXPIRY_BINDING=1.") standbyPrimary := flag.String("standby", "", "run as hot standby replicating from the given primary address (e.g. primary:9000)") httpAddr := flag.String("http", "", "HTTP dashboard listen address (e.g. :3000)") logLevel := flag.String("log-level", "info", "log level (debug, info, warn, error)") @@ -140,6 +142,14 @@ func main() { r.SetStrictRegistrationAuth(true) slog.Info("strict registration authorization enabled") } + if *strictHeartbeatFreshness || os.Getenv("RENDEZVOUS_STRICT_HEARTBEAT_FRESHNESS") == "1" { + r.SetStrictHeartbeatFreshness(true) + slog.Info("heartbeat freshness enforcement enabled", "binary_heartbeat", "refused") + } + if *strictExpiryBinding || os.Getenv("RENDEZVOUS_STRICT_EXPIRY_BINDING") == "1" { + r.SetStrictExpiryBinding(true) + slog.Info("set_key_expiry challenge binding enabled") + } // Plumb the breaker manager into the in-process beacon so // beacon.punch / beacon.relay / beacon.discover can be flipped from // the same breakers.json file as the registry-side switches. diff --git a/directory/directory.go b/directory/directory.go index fd8ecbc..ed8308a 100644 --- a/directory/directory.go +++ b/directory/directory.go @@ -245,6 +245,12 @@ type Callbacks struct { ScanNetworkMemberships func(nodeID uint32) []uint16 StrictDirectoryAuth func() bool StrictRegistrationAuth func() bool + + // StrictHeartbeatFreshness reports whether heartbeats must carry a + // recent timestamp bound into the signed challenge. Nil or false + // keeps the original challenge, which covers only the node id — see + // heartbeatChallenge. + StrictHeartbeatFreshness func() bool } // -------------------------------------------------------------------------- @@ -360,6 +366,17 @@ func jsonUint32(msg map[string]interface{}, key string) uint32 { return 0 } +// jsonInt64 reads a JSON number as an int64. ok is false when the key is +// absent or not a number, which the caller distinguishes from a present +// zero. +func jsonInt64(msg map[string]interface{}, key string) (int64, bool) { + v, ok := msg[key].(float64) + if !ok { + return 0, false + } + return int64(v), true +} + func jsonUint16(msg map[string]interface{}, key string) uint16 { if v, ok := msg[key].(float64); ok { if v < 0 || v > float64(^uint16(0)) { @@ -1962,6 +1979,52 @@ func (st *Store) HandleDeregister(msg map[string]interface{}) (map[string]interf // Handler: heartbeat // -------------------------------------------------------------------------- +// heartbeatMaxSkew is how far a heartbeat's timestamp may sit from the +// registry's clock, in either direction, when freshness is enforced. It +// is wide enough to absorb ordinary clock drift and request latency and +// narrow enough that a captured heartbeat stops being usable quickly. +const heartbeatMaxSkew = 60 * time.Second + +// heartbeatVerifyCacheTTL is how long a successful signature check is +// reused for subsequent heartbeats from the same node. +const heartbeatVerifyCacheTTL = 120 * time.Second + +// strictHeartbeatFreshness reports whether the freshness gate is active. +func (st *Store) strictHeartbeatFreshness() bool { + return st.cb.StrictHeartbeatFreshness != nil && st.cb.StrictHeartbeatFreshness() +} + +// heartbeatChallenge returns the string a heartbeat must be signed over, +// and the timestamp it is bound to. +// +// The default form covers only the node id, so one signature is valid +// for that node forever: replaying a captured heartbeat keeps refreshing +// LastSeen, and the node keeps reading as online after it is gone. The +// bound form appends a caller-supplied unix timestamp, which the caller +// must also have signed, so a captured heartbeat stops being accepted +// once it falls outside heartbeatMaxSkew. +// +// The bound form changes what a signer must produce, so it is gated on +// Callbacks.StrictHeartbeatFreshness and off by default; enable it once +// every client in the deployment sends and signs a "ts" field. +func (st *Store) heartbeatChallenge(msg map[string]interface{}, nodeID uint32, now time.Time) (string, error) { + if !st.strictHeartbeatFreshness() { + return fmt.Sprintf("heartbeat:%d", nodeID), nil + } + ts, ok := jsonInt64(msg, "ts") + if !ok { + return "", fmt.Errorf("heartbeat requires a ts field") + } + skew := now.Unix() - ts + if skew < 0 { + skew = -skew + } + if skew > int64(heartbeatMaxSkew/time.Second) { + return "", fmt.Errorf("heartbeat ts is %ds from registry time (max %ds)", skew, int64(heartbeatMaxSkew/time.Second)) + } + return fmt.Sprintf("heartbeat:%d:%d", nodeID, ts), nil +} + // HandleHeartbeat handles a JSON heartbeat message. func (st *Store) HandleHeartbeat(msg map[string]interface{}) (map[string]interface{}, error) { nodeID := jsonUint32(msg, "node_id") @@ -1979,11 +2042,20 @@ func (st *Store) HandleHeartbeat(msg map[string]interface{}) (map[string]interfa now := st.cb.Now() + // The verify cache reuses one successful signature check for the + // following heartbeatVerifyCacheTTL, which would let a heartbeat + // through on the strength of an earlier one. When freshness is being + // enforced, every heartbeat is checked on its own merits. lastVerified := node.LastVerifiedNano.Load() - skipVerify := lastVerified > 0 && (now.UnixNano()-lastVerified) < int64(120*time.Second) + skipVerify := !st.strictHeartbeatFreshness() && + lastVerified > 0 && (now.UnixNano()-lastVerified) < int64(heartbeatVerifyCacheTTL) if !skipVerify { - if err := st.cb.VerifyNodeSignature(pubKey, adminToken, msg, fmt.Sprintf("heartbeat:%d", nodeID)); err != nil { + challenge, err := st.heartbeatChallenge(msg, nodeID, now) + if err != nil { + return nil, err + } + if err := st.cb.VerifyNodeSignature(pubKey, adminToken, msg, challenge); err != nil { return nil, err } node.LastVerifiedNano.Store(now.UnixNano()) @@ -2020,6 +2092,19 @@ func (st *Store) HandleBinaryHeartbeat(conn net.Conn, payload []byte) { st.cb.ObserveRequestDuration("heartbeat", time.Since(start).Seconds()) }() + // The binary heartbeat payload is a fixed node id + signature with no + // field for a timestamp, so there is nowhere to carry the freshness + // value the challenge would bind. Leaving this path open while the + // JSON path enforces freshness would make the gate decorative — a + // caller would simply use this encoding instead — so when freshness + // is being enforced this encoding is refused outright. Carrying a + // timestamp here needs a registry wire-format revision. + if st.strictHeartbeatFreshness() { + st.cb.IncErrorsTotal("heartbeat") + wire.WriteFrame(conn, wire.MsgError, wire.EncodeError("binary heartbeat encoding carries no timestamp; use the JSON heartbeat while freshness is enforced")) + return + } + st.mu.RLock() node, ok := st.nodes[req.NodeID] if !ok { diff --git a/directory/zz_heartbeat_freshness_test.go b/directory/zz_heartbeat_freshness_test.go new file mode 100644 index 0000000..c46fa97 --- /dev/null +++ b/directory/zz_heartbeat_freshness_test.go @@ -0,0 +1,312 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +package directory + +import ( + "errors" + "fmt" + "net" + "sync" + "testing" + "time" + + "github.com/pilot-protocol/common/crypto" + "github.com/pilot-protocol/common/registry/wire" +) + +// heartbeatProbe wires a test Store so the challenge string handed to +// signature verification is observable, the clock is controllable, and +// the freshness gate can be toggled. +type heartbeatProbe struct { + st *Store + + mu sync.Mutex + challenges []string + now time.Time + strict bool + verifyErr error +} + +func newHeartbeatProbe(t *testing.T, nodeID uint32) *heartbeatProbe { + t.Helper() + p := &heartbeatProbe{now: time.Unix(1_800_000_000, 0)} + p.st = newTestStore(t) + + p.st.cb.Now = func() time.Time { + p.mu.Lock() + defer p.mu.Unlock() + return p.now + } + p.st.cb.VerifyNodeSignature = func(_ []byte, _ string, _ map[string]interface{}, challenge string) error { + p.mu.Lock() + defer p.mu.Unlock() + p.challenges = append(p.challenges, challenge) + return p.verifyErr + } + p.st.cb.StrictHeartbeatFreshness = func() bool { + p.mu.Lock() + defer p.mu.Unlock() + return p.strict + } + + p.st.mu.Lock() + p.st.nodes[nodeID] = &NodeInfo{ID: nodeID, PublicKey: []byte("pubkey")} + p.st.mu.Unlock() + return p +} + +func (p *heartbeatProbe) setStrict(v bool) { + p.mu.Lock() + p.strict = v + p.mu.Unlock() +} + +func (p *heartbeatProbe) advance(d time.Duration) { + p.mu.Lock() + p.now = p.now.Add(d) + p.mu.Unlock() +} + +func (p *heartbeatProbe) unixNow() int64 { + p.mu.Lock() + defer p.mu.Unlock() + return p.now.Unix() +} + +func (p *heartbeatProbe) seen() []string { + p.mu.Lock() + defer p.mu.Unlock() + return append([]string(nil), p.challenges...) +} + +func (p *heartbeatProbe) reset() { + p.mu.Lock() + p.challenges = nil + p.mu.Unlock() +} + +// TestHeartbeatChallengeDefaultsUnbound pins the wire-compatible default: +// with the gate off the challenge covers only the node id, no ts field is +// required, and the verification cache still applies. +func TestHeartbeatChallengeDefaultsUnbound(t *testing.T) { + t.Parallel() + const nodeID = 4321 + p := newHeartbeatProbe(t, nodeID) + + if _, err := p.st.HandleHeartbeat(map[string]interface{}{"node_id": float64(nodeID)}); err != nil { + t.Fatalf("heartbeat without ts rejected while the gate is off: %v", err) + } + got := p.seen() + if len(got) != 1 || got[0] != fmt.Sprintf("heartbeat:%d", nodeID) { + t.Fatalf("challenge = %v; want [heartbeat:%d]", got, nodeID) + } + + // Second heartbeat inside the cache window: verification is reused. + p.reset() + if _, err := p.st.HandleHeartbeat(map[string]interface{}{"node_id": float64(nodeID)}); err != nil { + t.Fatalf("second heartbeat: %v", err) + } + if n := len(p.seen()); n != 0 { + t.Errorf("verification ran %d times inside the cache window; want 0 (cache reused)", n) + } +} + +// TestHeartbeatFreshnessBindsTimestamp pins the gated behaviour: the +// timestamp the caller supplies is bound into the challenge, so a +// signature is only good for the moment it was produced for. +func TestHeartbeatFreshnessBindsTimestamp(t *testing.T) { + t.Parallel() + const nodeID = 4322 + p := newHeartbeatProbe(t, nodeID) + p.setStrict(true) + + ts := p.unixNow() + if _, err := p.st.HandleHeartbeat(map[string]interface{}{ + "node_id": float64(nodeID), + "ts": float64(ts), + }); err != nil { + t.Fatalf("fresh heartbeat rejected: %v", err) + } + got := p.seen() + want := fmt.Sprintf("heartbeat:%d:%d", nodeID, ts) + if len(got) != 1 || got[0] != want { + t.Fatalf("challenge = %v; want [%s]", got, want) + } +} + +// TestHeartbeatFreshnessRejectsReplay is the property the gate exists +// for: a heartbeat captured off the wire stops being accepted once its +// timestamp falls outside the skew window, so it can no longer hold a +// node that is gone in the "seen recently" state. +func TestHeartbeatFreshnessRejectsReplay(t *testing.T) { + t.Parallel() + const nodeID = 4323 + p := newHeartbeatProbe(t, nodeID) + p.setStrict(true) + + captured := map[string]interface{}{ + "node_id": float64(nodeID), + "ts": float64(p.unixNow()), + } + if _, err := p.st.HandleHeartbeat(captured); err != nil { + t.Fatalf("original heartbeat rejected: %v", err) + } + + // Replayed well after the fact. + p.advance(heartbeatMaxSkew + time.Minute) + if _, err := p.st.HandleHeartbeat(captured); err == nil { + t.Fatal("replayed heartbeat accepted after the skew window; it should be refused") + } + + // A heartbeat minted now still works, so the gate rejects on age and + // not by refusing everything. + if _, err := p.st.HandleHeartbeat(map[string]interface{}{ + "node_id": float64(nodeID), + "ts": float64(p.unixNow()), + }); err != nil { + t.Fatalf("current heartbeat rejected: %v", err) + } +} + +// TestHeartbeatFreshnessRequiresTimestamp covers the missing-field and +// future-timestamp cases. +func TestHeartbeatFreshnessRequiresTimestamp(t *testing.T) { + t.Parallel() + const nodeID = 4324 + p := newHeartbeatProbe(t, nodeID) + p.setStrict(true) + + if _, err := p.st.HandleHeartbeat(map[string]interface{}{"node_id": float64(nodeID)}); err == nil { + t.Error("heartbeat with no ts accepted while freshness is enforced") + } + if _, err := p.st.HandleHeartbeat(map[string]interface{}{ + "node_id": float64(nodeID), + "ts": float64(p.unixNow() + int64((heartbeatMaxSkew+time.Minute)/time.Second)), + }); err == nil { + t.Error("heartbeat timestamped in the future accepted; the skew window is two-sided") + } +} + +// TestHeartbeatFreshnessBypassesVerifyCache pins that the gate is not +// undone by the signature-verification cache: without this, one accepted +// heartbeat would wave through anything sent in the following two +// minutes, signature and timestamp unchecked. +func TestHeartbeatFreshnessBypassesVerifyCache(t *testing.T) { + t.Parallel() + const nodeID = 4325 + p := newHeartbeatProbe(t, nodeID) + p.setStrict(true) + + beat := func() error { + _, err := p.st.HandleHeartbeat(map[string]interface{}{ + "node_id": float64(nodeID), + "ts": float64(p.unixNow()), + }) + return err + } + if err := beat(); err != nil { + t.Fatalf("first heartbeat: %v", err) + } + p.reset() + + // Well inside the cache window. + p.advance(time.Second) + p.mu.Lock() + p.verifyErr = errors.New("bad signature") + p.mu.Unlock() + + if err := beat(); err == nil { + t.Fatal("heartbeat with an invalid signature accepted inside the cache window") + } + if n := len(p.seen()); n != 1 { + t.Errorf("verification ran %d times; want 1 (the cache must not be consulted)", n) + } +} + +// TestBinaryHeartbeatRefusedWhenFreshnessEnforced pins that the gate +// cannot be sidestepped by switching encodings. The binary heartbeat +// payload is a fixed node id + signature with no field for a timestamp, +// so while freshness is enforced it is refused rather than silently +// accepted on the original unbound challenge. +func TestBinaryHeartbeatRefusedWhenFreshnessEnforced(t *testing.T) { + t.Parallel() + const nodeID = 4326 + p := newHeartbeatProbe(t, nodeID) + + // A genuinely valid binary heartbeat: real key, real signature over + // the unbound challenge. It is accepted on this encoding today, which + // is exactly why leaving the encoding open would make the gate + // decorative. + id, err := crypto.GenerateIdentity() + if err != nil { + t.Fatal(err) + } + p.st.mu.Lock() + p.st.nodes[nodeID].PublicKey = id.PublicKey + p.st.mu.Unlock() + payload := wire.EncodeHeartbeatReq(nodeID, id.Sign([]byte(fmt.Sprintf("heartbeat:%d", nodeID)))) + + send := func() (byte, error) { + srv, cli := net.Pipe() + defer srv.Close() + defer cli.Close() + + replies := make(chan wireReply, 1) + go func() { + typ, body, err := wire.ReadFrame(cli) + replies <- wireReply{typ: typ, payload: body, err: err} + }() + + done := make(chan struct{}) + go func() { + p.st.HandleBinaryHeartbeat(srv, payload) + close(done) + }() + + var got wireReply + select { + case got = <-replies: + case <-time.After(2 * time.Second): + t.Fatal("no reply to the binary heartbeat") + } + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("HandleBinaryHeartbeat blocked") + } + return got.typ, got.err + } + + // Baseline: the request is well-formed and accepted with the gate off. + if typ, err := send(); err != nil { + t.Fatalf("read reply: %v", err) + } else if typ != wire.MsgHeartbeatOK { + t.Fatalf("setup: binary heartbeat answered with frame type %#x; want MsgHeartbeatOK — the request must be valid for this test to mean anything", typ) + } + + p.st.mu.RLock() + p.st.nodes[nodeID].LastSeenNano.Store(0) + p.st.mu.RUnlock() + + // Same request, gate on: refused, because this encoding cannot carry + // the timestamp the challenge would bind. + p.setStrict(true) + if typ, err := send(); err != nil { + t.Fatalf("read reply: %v", err) + } else if typ != wire.MsgError { + t.Fatalf("binary heartbeat answered with frame type %#x while freshness is enforced; want an error frame", typ) + } + + p.st.mu.RLock() + node := p.st.nodes[nodeID] + p.st.mu.RUnlock() + if node.LastSeenNano.Load() != 0 { + t.Error("a refused binary heartbeat still refreshed the node's last-seen time") + } +} + +type wireReply struct { + typ byte + payload []byte + err error +} diff --git a/identity/identity.go b/identity/identity.go index 3583d27..3304817 100644 --- a/identity/identity.go +++ b/identity/identity.go @@ -194,6 +194,12 @@ type Callbacks struct { // Bus is the event bus used to publish "key.rotated" events. Bus events.Bus + + // StrictExpiryBinding reports whether set_key_expiry challenges must + // bind the requested expires_at value. Nil or false keeps the + // original challenge, which covers only the node id — see + // HandleSetKeyExpiry. + StrictExpiryBinding func() bool } // Store holds the mutable identity and key-lifecycle state. @@ -524,6 +530,27 @@ func (st *Store) HandleRotateKey(msg map[string]interface{}) (map[string]interfa }, nil } +// setKeyExpiryChallenge returns the string a set_key_expiry request must +// be signed over. +// +// The default form covers only the node id, so one signature authorizes +// any expiry value for that node — including a later request that +// substitutes a different expires_at. The bound form appends the +// requested value, the way HandleRotateKey's challenge appends the new +// public key, so a signature authorizes exactly the value it was +// produced for. +// +// The bound form is what a signer must produce, so switching it changes +// what an existing client's signature verifies against. It is therefore +// gated on Callbacks.StrictExpiryBinding and off by default; enable it +// once every client in the deployment signs the bound form. +func (st *Store) setKeyExpiryChallenge(nodeID uint32, expiresAtStr string) string { + if st.cb.StrictExpiryBinding != nil && st.cb.StrictExpiryBinding() { + return fmt.Sprintf("set_key_expiry:%d:%s", nodeID, expiresAtStr) + } + return fmt.Sprintf("set_key_expiry:%d", nodeID) +} + // HandleSetKeyExpiry implements the "set_key_expiry" protocol command. func (st *Store) HandleSetKeyExpiry(msg map[string]interface{}) (map[string]interface{}, error) { nodeID := jsonUint32(msg, "node_id") @@ -554,7 +581,7 @@ func (st *Store) HandleSetKeyExpiry(msg map[string]interface{}) (map[string]inte adminToken := st.nodes.AdminToken() // Phase 2 — verify signature outside the lock. - sigErr := st.nodes.VerifyHeartbeatSignature(currentPubKey, adminToken, msg, fmt.Sprintf("set_key_expiry:%d", nodeID)) + sigErr := st.nodes.VerifyHeartbeatSignature(currentPubKey, adminToken, msg, st.setKeyExpiryChallenge(nodeID, expiresAtStr)) if sigErr != nil { if err := st.nodes.CheckAdminToken(msg); err != nil { return nil, sigErr diff --git a/server.go b/server.go index 969f5d3..7be9bcc 100644 --- a/server.go +++ b/server.go @@ -286,6 +286,13 @@ type Server struct { strictDirectoryAuth atomic.Bool strictRegistrationAuth atomic.Bool + + // strictHeartbeatFreshness requires heartbeats to bind a recent + // timestamp into the signed challenge. strictExpiryBinding requires + // set_key_expiry challenges to bind the requested expires_at. Both + // change what a client must sign, so both default off. + strictHeartbeatFreshness atomic.Bool + strictExpiryBinding atomic.Bool } // listNodesCacheState is defined in the directory sub-package (R4.2). diff --git a/server_lifecycle.go b/server_lifecycle.go index 4fde4d2..0d67eb9 100644 --- a/server_lifecycle.go +++ b/server_lifecycle.go @@ -133,6 +133,40 @@ func (s *Server) StrictRegistrationAuth() bool { return s.strictRegistrationAuth.Load() } +// SetStrictHeartbeatFreshness enables the heartbeat freshness gate: a +// heartbeat must carry a "ts" field within heartbeatMaxSkew of registry +// time, that value is bound into the signed challenge, and the +// signature-verification cache is bypassed so every heartbeat stands on +// its own. It also refuses the binary heartbeat encoding, which has no +// field for a timestamp. +// +// This changes what clients must sign, so it defaults off. Enable it +// only once every client in the deployment sends and signs "ts". +func (s *Server) SetStrictHeartbeatFreshness(enabled bool) { + s.strictHeartbeatFreshness.Store(enabled) +} + +// StrictHeartbeatFreshness reports whether the heartbeat freshness gate +// is enforced. +func (s *Server) StrictHeartbeatFreshness() bool { + return s.strictHeartbeatFreshness.Load() +} + +// SetStrictExpiryBinding enables binding the requested expires_at value +// into the set_key_expiry challenge, so a signature authorizes exactly +// the expiry it was produced for rather than any expiry for that node. +// +// This changes what clients must sign, so it defaults off. +func (s *Server) SetStrictExpiryBinding(enabled bool) { + s.strictExpiryBinding.Store(enabled) +} + +// StrictExpiryBinding reports whether set_key_expiry challenges bind the +// requested expires_at value. +func (s *Server) StrictExpiryBinding() bool { + return s.strictExpiryBinding.Load() +} + // SetDashboardToken gates per-network stats on the dashboard. // Empty string restricts the dashboard to global aggregates only. func (s *Server) SetDashboardToken(token string) { @@ -669,7 +703,8 @@ func NewWithStore(beaconAddr, storePath string) *Server { s.pubKeyIdx[newPubKeyB64] = nodeID s.mu.Unlock() }, - Bus: s.bus, + Bus: s.bus, + StrictExpiryBinding: s.StrictExpiryBinding, }) s.policy = policypkg.NewStore( // R2.4: policy sub-package func(netID uint16) (policypkg.NetworkState, error) { @@ -949,8 +984,9 @@ func NewWithStore(beaconAddr, storePath string) *Server { } return nets }, - StrictDirectoryAuth: s.StrictDirectoryAuth, - StrictRegistrationAuth: s.StrictRegistrationAuth, + StrictDirectoryAuth: s.StrictDirectoryAuth, + StrictRegistrationAuth: s.StrictRegistrationAuth, + StrictHeartbeatFreshness: s.StrictHeartbeatFreshness, }, ) diff --git a/trust/trust.go b/trust/trust.go index ff17f64..77e8518 100644 --- a/trust/trust.go +++ b/trust/trust.go @@ -7,6 +7,7 @@ package trust import ( + "crypto/subtle" "encoding/base64" "errors" "fmt" @@ -516,13 +517,16 @@ func verifyHeartbeatSignature(pubKey []byte, adminToken string, msg map[string]i return nil } -// checkAdminToken validates the "admin_token" field against the expected value. +// checkAdminToken validates the "admin_token" field against the expected +// value. The comparison runs in time independent of how far the supplied +// token matches, matching every other admin-token check in the repo +// (authz, membership, dashboard, accept). func checkAdminToken(msg map[string]interface{}, adminToken string) error { if adminToken == "" { return fmt.Errorf("no admin token configured") } token, _ := msg["admin_token"].(string) - if token != adminToken { + if subtle.ConstantTimeCompare([]byte(token), []byte(adminToken)) != 1 { return fmt.Errorf("invalid admin token") } return nil diff --git a/trust/zz_admin_token_compare_test.go b/trust/zz_admin_token_compare_test.go new file mode 100644 index 0000000..e22da4f --- /dev/null +++ b/trust/zz_admin_token_compare_test.go @@ -0,0 +1,126 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +package trust + +import ( + "go/ast" + "go/parser" + "go/token" + "strings" + "testing" +) + +// TestCheckAdminTokenComparesInConstantTime pins that the admin-token +// check does not short-circuit on the first differing byte, which would +// let the time a rejection takes reveal how much of the token was +// correct and make the secret recoverable byte by byte. +// +// Constant-time behaviour cannot be asserted by measuring: the +// difference is nanoseconds and any timing threshold is a flake. So this +// inspects the comparison the function actually performs — it must go +// through crypto/subtle, and must not compare the supplied token against +// the expected one with a language-level string operator, which is what +// every sibling admin-token check in this repo (authz, membership, +// dashboard, accept) already does. +func TestCheckAdminTokenComparesInConstantTime(t *testing.T) { + t.Parallel() + + fset := token.NewFileSet() + file, err := parser.ParseFile(fset, "trust.go", nil, 0) + if err != nil { + t.Fatalf("parse trust.go: %v", err) + } + + fn := findFunc(file, "checkAdminToken") + if fn == nil { + t.Fatal("checkAdminToken not found in trust.go") + } + + var usesSubtle bool + var directCompares []string + ast.Inspect(fn, func(n ast.Node) bool { + switch v := n.(type) { + case *ast.CallExpr: + if sel, ok := v.Fun.(*ast.SelectorExpr); ok { + if pkg, ok := sel.X.(*ast.Ident); ok && + pkg.Name == "subtle" && sel.Sel.Name == "ConstantTimeCompare" { + usesSubtle = true + } + } + case *ast.BinaryExpr: + if v.Op != token.EQL && v.Op != token.NEQ { + return true + } + l, r := exprName(v.X), exprName(v.Y) + if (l == "token" && r == "adminToken") || (l == "adminToken" && r == "token") { + directCompares = append(directCompares, l+" "+v.Op.String()+" "+r) + } + } + return true + }) + + if !usesSubtle { + t.Error("checkAdminToken does not call subtle.ConstantTimeCompare") + } + if len(directCompares) > 0 { + t.Errorf("checkAdminToken compares the token directly (%s); the comparison must not short-circuit", + strings.Join(directCompares, ", ")) + } +} + +func findFunc(file *ast.File, name string) *ast.FuncDecl { + for _, decl := range file.Decls { + if fn, ok := decl.(*ast.FuncDecl); ok && fn.Name.Name == name && fn.Recv == nil { + return fn + } + } + return nil +} + +func exprName(e ast.Expr) string { + if id, ok := e.(*ast.Ident); ok { + return id.Name + } + return "" +} + +// TestCheckAdminTokenOutcomesUnchanged pins that swapping the comparison +// did not change which tokens are accepted, including the cases where a +// length-based shortcut would be tempting. +func TestCheckAdminTokenOutcomesUnchanged(t *testing.T) { + t.Parallel() + + cases := []struct { + name string + supplied interface{} + expected string + wantErr bool + }{ + {"exact match", "s3cret", "s3cret", false}, + {"no token configured", "anything", "", true}, + {"empty supplied", "", "s3cret", true}, + {"field absent", nil, "s3cret", true}, + {"wrong type", 42.0, "s3cret", true}, + {"correct prefix", "s3c", "s3cret", true}, + {"correct prefix plus suffix", "s3cretX", "s3cret", true}, + {"same length, last byte differs", "s3crey", "s3cret", true}, + {"case differs", "S3CRET", "s3cret", true}, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + msg := map[string]interface{}{} + if tc.supplied != nil { + msg["admin_token"] = tc.supplied + } + err := checkAdminToken(msg, tc.expected) + if tc.wantErr && err == nil { + t.Fatalf("checkAdminToken(%v, %q) = nil; want an error", tc.supplied, tc.expected) + } + if !tc.wantErr && err != nil { + t.Fatalf("checkAdminToken(%v, %q) = %v; want nil", tc.supplied, tc.expected, err) + } + }) + } +} diff --git a/wal/wal.go b/wal/wal.go index d456904..492bc3a 100644 --- a/wal/wal.go +++ b/wal/wal.go @@ -5,9 +5,13 @@ // // The WAL closes the data-loss window between snapshot saves. flushSave runs // on saveLoopInterval (5s); a crash between saves would otherwise drop every -// mutation in that window. Each low-frequency mutation records a delta to the +// mutation in that window. Each low-frequency mutation writes a delta to the // WAL synchronously. On startup the snapshot is loaded, then any post-snapshot // WAL entries are replayed on top. +// +// The fsync behind those writes is batched, not per-entry — see the +// walSyncInterval / walSyncBatch constants. So the log survives a process +// crash in full, and survives host power loss up to a bounded tail. package wal import ( @@ -83,6 +87,10 @@ type WAL struct { syncDone chan struct{} } +// Appends are fsync'd in batches rather than one at a time: whichever of +// these two thresholds is reached first triggers the flush, so the +// power-loss window is bounded by walSyncInterval of wall time or +// walSyncBatch of un-fsync'd entries. Close flushes whatever is pending. const ( walSyncInterval = 200 * time.Millisecond walSyncBatch = 200 @@ -142,8 +150,15 @@ func (w *WAL) syncLoop() { } } -// Append writes a delta entry to the WAL. The entry is fsync'd to ensure -// durability. Returns an error if the write fails. +// Append writes a delta entry to the WAL. Returns an error if the write +// fails. +// +// The write(2) is synchronous, so the entry is visible to Replay and to +// any other reader of the file as soon as Append returns. The fsync is +// not: it is batched, so a host that loses power (as opposed to a +// process that crashes) can lose entries appended within the last +// walSyncInterval or fewer than walSyncBatch back. Close flushes the +// pending tail, so an orderly shutdown loses nothing. func (w *WAL) Append(entry DeltaEntry) error { if w == nil { return nil diff --git a/wal/zz_append_durability_test.go b/wal/zz_append_durability_test.go new file mode 100644 index 0000000..d5dcc53 --- /dev/null +++ b/wal/zz_append_durability_test.go @@ -0,0 +1,97 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +package wal_test + +import ( + "os" + "path/filepath" + "testing" + + "github.com/pilot-protocol/rendezvous/wal" +) + +// TestAppendIsVisibleImmediatelyAfterReturn pins the half of the +// durability contract Append does guarantee: the write(2) is synchronous, +// so the bytes are in the file and readable by anyone who opens it as +// soon as Append returns — no flush, no Close, no waiting for the +// batching timer. +// +// The other half is the part the doc comment used to overstate: the +// fsync behind that write is batched, so this says nothing about +// surviving host power loss. That distinction is what the comment on +// Append and the walSyncInterval / walSyncBatch constants now spell out. +func TestAppendIsVisibleImmediatelyAfterReturn(t *testing.T) { + t.Parallel() + path := filepath.Join(t.TempDir(), "visible.wal") + + w, err := wal.NewWAL(path) + if err != nil { + t.Fatalf("NewWAL: %v", err) + } + defer w.Close() + + // A single append: far below walSyncBatch and returning long before + // walSyncInterval elapses, so no fsync has run yet. + if err := w.Append(wal.DeltaEntry{SeqNo: 1, Type: wal.DeltaRegister, NodeID: 7}); err != nil { + t.Fatalf("Append: %v", err) + } + + info, err := os.Stat(path) + if err != nil { + t.Fatalf("stat: %v", err) + } + if info.Size() == 0 { + t.Fatal("the file is still empty after Append returned; the write is supposed to be synchronous") + } + if got := w.Size(); got != info.Size() { + t.Errorf("WAL.Size() = %d; the file on disk is %d", got, info.Size()) + } +} + +// TestCloseFlushesPendingAppends pins the shutdown guarantee the Append +// comment now states: an orderly Close flushes whatever the batching +// thresholds have not yet synced, so a clean restart replays everything. +func TestCloseFlushesPendingAppends(t *testing.T) { + t.Parallel() + path := filepath.Join(t.TempDir(), "close.wal") + + w, err := wal.NewWAL(path) + if err != nil { + t.Fatalf("NewWAL: %v", err) + } + + // Fewer than walSyncBatch entries, appended and closed well inside + // walSyncInterval, so the flush can only come from Close. + const n = 5 + for i := 0; i < n; i++ { + if err := w.Append(wal.DeltaEntry{SeqNo: uint64(i + 1), Type: wal.DeltaHeartbeat, NodeID: uint32(i + 1)}); err != nil { + t.Fatalf("Append %d: %v", i, err) + } + } + if err := w.Close(); err != nil { + t.Fatalf("Close: %v", err) + } + + reopened, err := wal.NewWAL(path) + if err != nil { + t.Fatalf("reopen: %v", err) + } + defer reopened.Close() + + var replayed []wal.DeltaEntry + count, err := reopened.Replay(func(e wal.DeltaEntry) error { + replayed = append(replayed, e) + return nil + }) + if err != nil { + t.Fatalf("Replay: %v", err) + } + if count != n { + t.Fatalf("replayed %d entries after Close; want %d", count, n) + } + for i, e := range replayed { + if e.SeqNo != uint64(i+1) || e.NodeID != uint32(i+1) { + t.Errorf("entry %d = %+v; want SeqNo/NodeID %d", i, e, i+1) + } + } +} diff --git a/zz_set_key_expiry_binding_test.go b/zz_set_key_expiry_binding_test.go new file mode 100644 index 0000000..8da11ac --- /dev/null +++ b/zz_set_key_expiry_binding_test.go @@ -0,0 +1,101 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +package server + +import ( + "encoding/base64" + "fmt" + "testing" + "time" + + "github.com/pilot-protocol/common/crypto" +) + +// setKeyExpiryReq builds a set_key_expiry message for the given node and +// expiry, signed over challenge. +func setKeyExpiryReq(id *crypto.Identity, nodeID uint32, expiresAt, challenge string) map[string]interface{} { + return map[string]interface{}{ + "node_id": float64(nodeID), + "expires_at": expiresAt, + "signature": base64.StdEncoding.EncodeToString(id.Sign([]byte(challenge))), + } +} + +// TestSetKeyExpiryBindingRejectsSubstitutedValue pins the gated +// behaviour: with binding on, a signature is good only for the expiry it +// was produced for. Without it, the challenge covers the node id alone, +// so one captured signature authorizes any expiry for that node — +// including pushing it far enough out that the key never expires. +func TestSetKeyExpiryBindingRejectsSubstitutedValue(t *testing.T) { + t.Parallel() + const nodeID = uint32(1180) + s := newTestServer(t, "") + id, _ := seedNodeWithIdentity(t, s, nodeID, "alice") + seedEnterpriseNetwork(t, s, 60, nodeID) + + soon := time.Now().Add(24 * time.Hour).UTC().Truncate(time.Second).Format(time.RFC3339) + distant := time.Now().Add(9 * 365 * 24 * time.Hour).UTC().Truncate(time.Second).Format(time.RFC3339) + + s.SetStrictExpiryBinding(true) + if !s.StrictExpiryBinding() { + t.Fatal("SetStrictExpiryBinding(true) did not take effect") + } + + bound := fmt.Sprintf("set_key_expiry:%d:%s", nodeID, soon) + + // The signature matches the value it was made for. + if _, err := s.identity.HandleSetKeyExpiry(setKeyExpiryReq(id, nodeID, soon, bound)); err != nil { + t.Fatalf("request signed for its own expiry rejected: %v", err) + } + + // Same signature, different expiry: the challenge no longer matches. + substituted := setKeyExpiryReq(id, nodeID, soon, bound) + substituted["expires_at"] = distant + if _, err := s.identity.HandleSetKeyExpiry(substituted); err == nil { + t.Fatal("a signature produced for one expiry authorized a different one") + } +} + +// TestSetKeyExpiryBindingDefaultsOff pins that the challenge is +// unchanged by default, so clients signing the original form keep +// working until an operator turns binding on. +func TestSetKeyExpiryBindingDefaultsOff(t *testing.T) { + t.Parallel() + const nodeID = uint32(1181) + s := newTestServer(t, "") + id, _ := seedNodeWithIdentity(t, s, nodeID, "bob") + seedEnterpriseNetwork(t, s, 61, nodeID) + + if s.StrictExpiryBinding() { + t.Fatal("expiry binding is on by default; it must be opt-in") + } + + expires := time.Now().Add(48 * time.Hour).UTC().Truncate(time.Second).Format(time.RFC3339) + unbound := fmt.Sprintf("set_key_expiry:%d", nodeID) + resp, err := s.identity.HandleSetKeyExpiry(setKeyExpiryReq(id, nodeID, expires, unbound)) + if err != nil { + t.Fatalf("request signed with the original challenge rejected: %v", err) + } + if resp["type"] != "set_key_expiry_ok" { + t.Fatalf("resp = %v; want set_key_expiry_ok", resp) + } +} + +// TestSetKeyExpiryBindingRejectsOldChallengeWhenOn pins that turning the +// gate on actually changes what must be signed, so a rollout that flips +// it before clients are updated fails loudly rather than silently +// accepting the old form. +func TestSetKeyExpiryBindingRejectsOldChallengeWhenOn(t *testing.T) { + t.Parallel() + const nodeID = uint32(1182) + s := newTestServer(t, "") + id, _ := seedNodeWithIdentity(t, s, nodeID, "carol") + seedEnterpriseNetwork(t, s, 62, nodeID) + s.SetStrictExpiryBinding(true) + + expires := time.Now().Add(48 * time.Hour).UTC().Truncate(time.Second).Format(time.RFC3339) + unbound := fmt.Sprintf("set_key_expiry:%d", nodeID) + if _, err := s.identity.HandleSetKeyExpiry(setKeyExpiryReq(id, nodeID, expires, unbound)); err == nil { + t.Fatal("a request signed with the unbound challenge was accepted while binding is enforced") + } +}