From e5f848de28336b0d75a7da862418fe86c5d0ae54 Mon Sep 17 00:00:00 2001 From: Teodor Calin Date: Sun, 26 Jul 2026 15:05:41 +0300 Subject: [PATCH] handshake: trust-transition preconditions, replay accounting, key-bound allowlist gate Five hardening changes to the port-444 handshake manager. No wire-format change: HandshakeMsg, its fields, and the signed challenge strings are untouched, so deployed daemons stay interoperable. handleAccept now requires and consumes a matching hm.outgoing entry before granting trust, mirroring the precondition processRelayedApproval already enforces on the relay path. handshake_reject joins handshake_accept / handshake_revoke in the transport-binding check: all three act on existing local state, so the authenticated sender node ID must match the claimed one. Only handshake_request, which creates fresh state, stays exempt. The replay set is now written after a message clears authentication rather than before, and capacity is enforced by eviction instead of refusal. Entries are attributed to a peer, a peer over maxReplayPerPeer displaces its own oldest entry, and a full set drops expired entries first and the globally oldest one after that. The trusted-agents auto-accept gate routes through the new optional KeyedTrustChecker interface when the Runtime provides it, passing the key the peer authenticated with (direct path) or the empty key (relay path, which carries none). Runtimes that only implement IsTrusted keep the node-ID-only answer, so existing adapters compile unchanged. Manager.IsTrustedWithKey is added: approved peers whose trust record carries a public key must present a constant-time match; records with no key recorded (relay-established, not yet backfilled) answer on node ID alone. Tests: new regression coverage per change in zz_findings_hardening_test.go (all six behavioral cases fail against the previous logic), plus existing tests updated to supply an authenticated transport and an in-flight outgoing request where the new preconditions require them. Co-Authored-By: Claude Opus 5 --- handshake.go | 174 ++++++++-- runtime.go | 15 + zz_ceiling_branches_test.go | 30 +- zz_findings_hardening_test.go | 425 +++++++++++++++++++++++ zz_handshake_accept_request_test.go | 7 +- zz_handshake_connection_parse_test.go | 19 +- zz_logic_test.go | 5 +- zz_pending_cleanup_test.go | 12 +- zz_ppa001_accept_revoke_identity_test.go | 4 + zz_ppa008_replay_key_timestamp_test.go | 9 +- 10 files changed, 654 insertions(+), 46 deletions(-) create mode 100644 zz_findings_hardening_test.go diff --git a/handshake.go b/handshake.go index 014427b..831646b 100644 --- a/handshake.go +++ b/handshake.go @@ -79,6 +79,7 @@ const ( handshakeRecvTimeout = 10 * time.Second // time to wait for handshake message handshakeCloseDelay = 500 * time.Millisecond // delay before closing after send to let data flush maxReplaySetEntries = 8192 // cap replay set to prevent unbounded growth between reaps + maxReplayPerPeer = 256 // cap replay entries attributable to any single peer maxPendingHandshakes = 256 // cap pending (unapproved) handshake requests maxPendingPerSource = 16 maxJustificationLen = 1024 @@ -92,6 +93,14 @@ const ( pendingHandshakeTTL = 30 * 24 * time.Hour ) +// replayEntry is one recorded handshake message hash: when it was first +// accepted, and which peer node ID it was attributed to. The peer field +// backs the per-peer accounting in recordReplay. +type replayEntry struct { + seen time.Time + peer uint32 +} + // Manager handles the trust handshake protocol on port 444. type Manager struct { mu sync.RWMutex @@ -111,8 +120,9 @@ type Manager struct { done chan struct{} // closed in Stop to signal drain goroutine exit // Replay protection - replayMu sync.Mutex - replaySet map[[32]byte]time.Time // message hash → first seen + replayMu sync.Mutex + replaySet map[[32]byte]replayEntry // message hash → first seen + attributed peer + replayPeer map[uint32]int // peer → number of entries it holds in replaySet // WaitForTrust notification: per-node list of channels closed when trust // transitions to granted. Separate mu from hm.mu (always taken AFTER hm.mu) @@ -132,7 +142,8 @@ func NewManager(rt Runtime) *Manager { pendingBySource: make(map[uint32]int), outgoing: make(map[uint32]time.Time), revoked: make(map[uint32]time.Time), - replaySet: make(map[[32]byte]time.Time), + replaySet: make(map[[32]byte]replayEntry), + replayPeer: make(map[uint32]int), trustWaiters: make(map[uint32][]chan struct{}), dirty: make(chan struct{}, 1), done: make(chan struct{}), @@ -570,21 +581,15 @@ func (hm *Manager) processMessage(stream coreapi.Stream, msg *HandshakeMsg) { return } + // The replay set is consulted here but only written after the message + // has cleared authentication below, so an unauthenticated sender + // cannot consume replay-set capacity on behalf of a peer. replayChallenge := fmt.Sprintf("handshake:%d:%d", msg.NodeID, hm.rt.NodeID()) msgHash := sha256.Sum256([]byte(msg.Type + "\x00" + replayChallenge + "\x00" + msg.Signature)) - hm.replayMu.Lock() - if _, seen := hm.replaySet[msgHash]; seen { - hm.replayMu.Unlock() + if hm.replaySeen(msgHash) { slog.Warn("handshake replay detected", "peer_node_id", msg.NodeID) return } - if len(hm.replaySet) >= maxReplaySetEntries { - hm.replayMu.Unlock() - slog.Warn("handshake replay set full, rejecting", "peer_node_id", msg.NodeID) - return - } - hm.replaySet[msgHash] = now - hm.replayMu.Unlock() // M12 fix: verify P2P signature if the sender provides a public key registryBound := false @@ -632,18 +637,29 @@ func (hm *Manager) processMessage(stream coreapi.Stream, msg *HandshakeMsg) { } } - if msg.Type == HandshakeAccept || msg.Type == HandshakeRevoke { + // accept / reject / revoke all act on existing local state (they grant + // trust, cancel an outgoing request, or drop trust), so each must + // arrive on a transport whose authenticated node ID matches the + // claimed sender. Only request creates fresh state and is exempt. + if msg.Type == HandshakeAccept || msg.Type == HandshakeRevoke || msg.Type == HandshakeReject { if stream == nil || stream.RemoteAddr().Node != msg.NodeID { var authenticated uint32 if stream != nil { authenticated = stream.RemoteAddr().Node } - slog.Warn("handshake: accept/revoke node_id does not match authenticated sender, rejecting", + slog.Warn("handshake: accept/reject/revoke node_id does not match authenticated sender, rejecting", "claimed_node_id", msg.NodeID, "authenticated_node_id", authenticated, "type", msg.Type) return } } + // Message is authenticated: record it so an identical copy replayed + // later is dropped by the replaySeen check above. + if !hm.recordReplay(msg.NodeID, msgHash, now) { + slog.Warn("handshake replay detected", "peer_node_id", msg.NodeID) + return + } + switch msg.Type { case HandshakeRequest: hm.handleRequest(stream, msg, registryBound) @@ -656,14 +672,101 @@ func (hm *Manager) processMessage(stream coreapi.Stream, msg *HandshakeMsg) { } } +// replaySeen reports whether msgHash is already recorded. +func (hm *Manager) replaySeen(msgHash [32]byte) bool { + hm.replayMu.Lock() + defer hm.replayMu.Unlock() + _, seen := hm.replaySet[msgHash] + return seen +} + +// recordReplay stores msgHash against peerNodeID. Returns false when the +// hash was already present (the message is a duplicate). +// +// Capacity is enforced by eviction, not refusal. A peer that already +// holds maxReplayPerPeer entries displaces its own oldest entry, so one +// talkative peer cannot crowd out the rest. When the whole set is at +// maxReplaySetEntries, expired entries are dropped first and the +// globally oldest entry after that. Recording therefore always succeeds +// for a message that is not itself a duplicate. +func (hm *Manager) recordReplay(peerNodeID uint32, msgHash [32]byte, now time.Time) bool { + hm.replayMu.Lock() + defer hm.replayMu.Unlock() + + if _, seen := hm.replaySet[msgHash]; seen { + return false + } + for hm.replayPeer[peerNodeID] >= maxReplayPerPeer { + if !hm.evictOldestReplayLocked(peerNodeID, true) { + break + } + } + if len(hm.replaySet) >= maxReplaySetEntries { + hm.reapReplayLocked(now) + } + for len(hm.replaySet) >= maxReplaySetEntries { + if !hm.evictOldestReplayLocked(0, false) { + break + } + } + + hm.replaySet[msgHash] = replayEntry{seen: now, peer: peerNodeID} + hm.replayPeer[peerNodeID]++ + return true +} + +// evictOldestReplayLocked drops the oldest replay entry, restricted to +// entries attributed to peerNodeID when scoped is true. Reports whether +// an entry was removed. Caller MUST hold replayMu. +func (hm *Manager) evictOldestReplayLocked(peerNodeID uint32, scoped bool) bool { + var ( + oldestHash [32]byte + oldestSeen time.Time + found bool + ) + for hash, e := range hm.replaySet { + if scoped && e.peer != peerNodeID { + continue + } + if !found || e.seen.Before(oldestSeen) { + oldestHash, oldestSeen, found = hash, e.seen, true + } + } + if !found { + return false + } + hm.deleteReplayLocked(oldestHash) + return true +} + +// deleteReplayLocked removes one entry and keeps the per-peer counter in +// step. Caller MUST hold replayMu. +func (hm *Manager) deleteReplayLocked(hash [32]byte) { + e, ok := hm.replaySet[hash] + if !ok { + return + } + delete(hm.replaySet, hash) + if hm.replayPeer[e.peer] <= 1 { + delete(hm.replayPeer, e.peer) + return + } + hm.replayPeer[e.peer]-- +} + // reapReplay removes expired entries from the replay set. func (hm *Manager) reapReplay() { hm.replayMu.Lock() defer hm.replayMu.Unlock() - threshold := time.Now().Add(-2 * handshakeMaxAge) - for hash, seen := range hm.replaySet { - if seen.Before(threshold) { - delete(hm.replaySet, hash) + hm.reapReplayLocked(time.Now()) +} + +// reapReplayLocked is reapReplay's body. Caller MUST hold replayMu. +func (hm *Manager) reapReplayLocked(now time.Time) { + threshold := now.Add(-2 * handshakeMaxAge) + for hash, e := range hm.replaySet { + if e.seen.Before(threshold) { + hm.deleteReplayLocked(hash) } } } @@ -833,7 +936,10 @@ func (hm *Manager) handleRequest(stream coreapi.Stream, msg *HandshakeMsg, regis // into auto-accept. if registryBound { // Trust gate via L10 TrustChecker. Nil → trust nothing. - name, ok := hm.rt.IsTrusted(peerNodeID) + // msg.PublicKey is the key the peer authenticated with and that + // the registry lookup bound to peerNodeID, so it is the key the + // allowlist pin is checked against. + name, ok := hm.trustedAgentName(peerNodeID, msg.PublicKey) if ok { hm.markTrustedLocked(peerNodeID, &TrustRecord{ NodeID: peerNodeID, @@ -908,6 +1014,11 @@ func (hm *Manager) handleRequest(stream coreapi.Stream, msg *HandshakeMsg, regis } // handleAccept processes a handshake acceptance from a peer. +// +// An acceptance only completes a handshake this node started: it is +// matched against, and consumes, the hm.outgoing entry created by +// SendRequest. Accepts with no matching entry are dropped, mirroring +// processRelayedApproval's precondition on the relay path. func (hm *Manager) handleAccept(msg *HandshakeMsg) { peerNodeID := msg.NodeID slog.Info("handshake accepted by peer", "peer_node_id", peerNodeID) @@ -925,6 +1036,13 @@ func (hm *Manager) handleAccept(msg *HandshakeMsg) { delete(hm.revoked, peerNodeID) } + // Trust follows a request we actually made: an accept with no + // matching outgoing entry is unsolicited and is dropped. + if _, ok := hm.outgoing[peerNodeID]; !ok { + slog.Warn("dropping handshake acceptance with no matching outgoing request", "peer_node_id", peerNodeID) + return + } + // An accept that carries a different key than the record we hold // comes from a different key holder behind the same node ID. Drop // the record instead of rebinding it — re-trusting is an explicit @@ -1089,7 +1207,10 @@ func (hm *Manager) processRelayedRequest(fromNodeID uint32, justification string // Auto-approve when the sender is in the embedded trusted-agents list. // The registry relay vouches for the sender's node_id (only authenticated // nodes can relay), so no separate registryBound check is needed here. - if name, ok := hm.rt.IsTrusted(fromNodeID); ok { + // + // The relay carries no peer public key, so the empty key is passed: + // allowlist entries that carry a pin do not match on this path. + if name, ok := hm.trustedAgentName(fromNodeID, ""); ok { hm.markTrustedLocked(fromNodeID, &TrustRecord{ NodeID: fromNodeID, ApprovedAt: time.Now(), @@ -1448,6 +1569,17 @@ func (hm *Manager) handleRevokeMsg(msg *HandshakeMsg) { } } +// trustedAgentName consults the allowlist gate for peerNodeID, passing +// the authenticated peer key when the Runtime implements the key-bound +// form. Runtimes that only implement IsTrusted keep the node-ID-only +// answer. +func (hm *Manager) trustedAgentName(peerNodeID uint32, pubKeyB64 string) (string, bool) { + if kc, ok := hm.rt.(KeyedTrustChecker); ok { + return kc.IsTrustedWithKey(peerNodeID, pubKeyB64) + } + return hm.rt.IsTrusted(peerNodeID) +} + // IsTrusted returns whether a peer has been approved. // // This answers the node-ID-only question, for call sites that have no diff --git a/runtime.go b/runtime.go index 3e3fa27..9a07709 100644 --- a/runtime.go +++ b/runtime.go @@ -89,6 +89,21 @@ type Runtime interface { Registry() RegistryClient } +// KeyedTrustChecker is the key-bound form of Runtime.IsTrusted. It is +// an optional extension: the handshake manager type-asserts its Runtime +// for it and falls back to IsTrusted when the adapter does not provide +// it, so existing Runtime implementations keep compiling unchanged. +// +// pubKeyB64 is the base64 Ed25519 key the peer proved possession of on +// this handshake, or "" when no authenticated key is in scope (the +// registry-relayed path). Implementations that carry a pinned key for +// the node MUST require an exact match and MUST treat "" as a miss; +// entries with no pin answer the node-ID-only question, matching +// IsTrusted. +type KeyedTrustChecker interface { + IsTrustedWithKey(nodeID uint32, pubKeyB64 string) (name string, ok bool) +} + // RegistryClient is the subset of pkg/registry/client.Client the // handshake manager uses. Defined here as an interface so the manager // stays free of import cycles and tests can supply a fake. diff --git a/zz_ceiling_branches_test.go b/zz_ceiling_branches_test.go index 29d74e7..e0881d3 100644 --- a/zz_ceiling_branches_test.go +++ b/zz_ceiling_branches_test.go @@ -11,6 +11,8 @@ import ( "path/filepath" "testing" "time" + + "github.com/pilot-protocol/common/coreapi" ) // This file targets every coverage hole identified by the iter-baseline @@ -522,16 +524,17 @@ func TestReapStalePending_PersistsViaSaveTrust(t *testing.T) { } // ----------------------------------------------------------------------- -// processMessage: replay set FULL — second message past the cap is dropped +// processMessage: a full replay set evicts rather than dropping the +// message, and stays at the cap // ----------------------------------------------------------------------- -func TestProcessMessage_ReplaySetFullDropsNewMessages(t *testing.T) { +func TestProcessMessage_ReplaySetFullStillDispatches(t *testing.T) { t.Parallel() hm := newTestHM(t, "") t.Cleanup(hm.Stop) - // Fill the replay set to exactly the cap so the next message hits the - // "set full" branch. + // Fill the replay set to exactly the cap, spread across many peers so + // no single peer is over its own limit. hm.replayMu.Lock() for i := 0; i < maxReplaySetEntries; i++ { var h [32]byte @@ -540,7 +543,9 @@ func TestProcessMessage_ReplaySetFullDropsNewMessages(t *testing.T) { h[1] = byte(i >> 8) h[2] = byte(i >> 16) h[3] = byte(i >> 24) - hm.replaySet[h] = time.Now() + peer := uint32(1000 + i) + hm.replaySet[h] = replayEntry{seen: time.Now(), peer: peer} + hm.replayPeer[peer]++ } hm.replayMu.Unlock() @@ -550,15 +555,22 @@ func TestProcessMessage_ReplaySetFullDropsNewMessages(t *testing.T) { Reason: "test", Timestamp: time.Now().Unix(), } - // Seed an outgoing entry so a dispatch would be observable. + // Seed an outgoing entry so the dispatch is observable. hm.outgoing[77] = time.Now() - hm.processMessage(nil, msg) + hm.processMessage(&addrStream{addr: coreapi.Addr{Node: 77}}, msg) hm.mu.RLock() _, stillOut := hm.outgoing[77] hm.mu.RUnlock() - if !stillOut { - t.Fatal("replay-set-full path must early-return — outgoing[77] should be preserved") + if stillOut { + t.Fatal("a full replay set must not block an authenticated message — outgoing[77] should have been cleared") + } + + hm.replayMu.Lock() + size := len(hm.replaySet) + hm.replayMu.Unlock() + if size > maxReplaySetEntries { + t.Fatalf("replay set grew past the cap: %d > %d", size, maxReplaySetEntries) } } diff --git a/zz_findings_hardening_test.go b/zz_findings_hardening_test.go new file mode 100644 index 0000000..38d6d81 --- /dev/null +++ b/zz_findings_hardening_test.go @@ -0,0 +1,425 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +package handshake + +import ( + "encoding/base64" + "testing" + "time" + + "github.com/pilot-protocol/common/coreapi" +) + +// ----------------------------------------------------------------------- +// handleAccept consumes a matching outgoing request +// ----------------------------------------------------------------------- + +func TestHandleAccept_WithoutOutgoingRequestDropped(t *testing.T) { + t.Parallel() + hm := newTestHM(t, "") + t.Cleanup(hm.Stop) + + // No hm.outgoing entry: this node never asked to handshake with 4242. + hm.handleAccept(&HandshakeMsg{ + Type: HandshakeAccept, + NodeID: 4242, + PublicKey: "pk4242", + Timestamp: time.Now().Unix(), + }) + + hm.mu.RLock() + _, trusted := hm.trusted[4242] + hm.mu.RUnlock() + if trusted { + t.Fatal("an acceptance with no matching outgoing request must not establish trust") + } +} + +func TestProcessMessage_UnsolicitedAcceptOverAuthenticatedStreamDropped(t *testing.T) { + t.Parallel() + hm := newTestHM(t, "") + t.Cleanup(hm.Stop) + + const peer = uint32(4343) + // The stream is authenticated as the claimed sender — the only + // remaining precondition is a pending outgoing request, and there is + // none. + hm.processMessage(&addrStream{addr: coreapi.Addr{Node: peer}}, &HandshakeMsg{ + Type: HandshakeAccept, + NodeID: peer, + Timestamp: time.Now().Unix(), + }) + + hm.mu.RLock() + _, trusted := hm.trusted[peer] + hm.mu.RUnlock() + if trusted { + t.Fatal("unsolicited accept must not establish trust even over an authenticated stream") + } +} + +func TestHandleAccept_ConsumesOutgoingEntryOnce(t *testing.T) { + t.Parallel() + hm := newTestHM(t, "") + t.Cleanup(hm.Stop) + + const peer = uint32(4444) + hm.mu.Lock() + hm.outgoing[peer] = time.Now() + hm.mu.Unlock() + + msg := &HandshakeMsg{ + Type: HandshakeAccept, + NodeID: peer, + PublicKey: "pk4444", + Timestamp: time.Now().Unix(), + } + hm.handleAccept(msg) + + hm.mu.RLock() + _, trusted := hm.trusted[peer] + _, stillOutgoing := hm.outgoing[peer] + hm.mu.RUnlock() + if !trusted { + t.Fatal("a solicited accept must still establish trust") + } + if stillOutgoing { + t.Fatal("the outgoing entry must be consumed by the accept") + } + + // Second delivery of the same acceptance has nothing left to consume. + hm.mu.Lock() + delete(hm.trusted, peer) + hm.mu.Unlock() + hm.handleAccept(msg) + + hm.mu.RLock() + _, reTrusted := hm.trusted[peer] + hm.mu.RUnlock() + if reTrusted { + t.Fatal("a replayed accept must not re-establish trust after the outgoing entry is consumed") + } +} + +// ----------------------------------------------------------------------- +// reject is bound to the authenticated sender +// ----------------------------------------------------------------------- + +func TestProcessMessage_RejectFromMismatchedStreamIgnored(t *testing.T) { + t.Parallel() + hm := newTestHM(t, "") + t.Cleanup(hm.Stop) + + const victim = uint32(700) + const other = uint32(666) + hm.mu.Lock() + hm.outgoing[victim] = time.Now() + hm.mu.Unlock() + + hm.processMessage(&addrStream{addr: coreapi.Addr{Node: other}}, &HandshakeMsg{ + Type: HandshakeReject, + NodeID: victim, + Reason: "not from the victim", + Timestamp: time.Now().Unix(), + }) + + hm.mu.RLock() + _, stillOutgoing := hm.outgoing[victim] + hm.mu.RUnlock() + if !stillOutgoing { + t.Fatal("a reject claiming another node's ID must not cancel that node's outgoing request") + } +} + +func TestProcessMessage_RejectWithNoTransportIgnored(t *testing.T) { + t.Parallel() + hm := newTestHM(t, "") + t.Cleanup(hm.Stop) + + const peer = uint32(701) + hm.mu.Lock() + hm.outgoing[peer] = time.Now() + hm.mu.Unlock() + + hm.processMessage(nil, &HandshakeMsg{ + Type: HandshakeReject, + NodeID: peer, + Timestamp: time.Now().Unix(), + }) + + hm.mu.RLock() + _, stillOutgoing := hm.outgoing[peer] + hm.mu.RUnlock() + if !stillOutgoing { + t.Fatal("a reject with no authenticated transport must not cancel an outgoing request") + } +} + +func TestProcessMessage_RejectOverMatchingStreamCancels(t *testing.T) { + t.Parallel() + hm := newTestHM(t, "") + t.Cleanup(hm.Stop) + + const peer = uint32(702) + hm.mu.Lock() + hm.outgoing[peer] = time.Now() + hm.mu.Unlock() + + hm.processMessage(&addrStream{addr: coreapi.Addr{Node: peer}}, &HandshakeMsg{ + Type: HandshakeReject, + NodeID: peer, + Timestamp: time.Now().Unix(), + }) + + hm.mu.RLock() + _, stillOutgoing := hm.outgoing[peer] + hm.mu.RUnlock() + if stillOutgoing { + t.Fatal("an honest reject over a matching authenticated stream must cancel the outgoing request") + } +} + +// ----------------------------------------------------------------------- +// replay-set entries are only recorded for authenticated messages +// ----------------------------------------------------------------------- + +func TestProcessMessage_UnverifiedMessageDoesNotConsumeReplayCapacity(t *testing.T) { + t.Parallel() + hm := newTestHM(t, "") + t.Cleanup(hm.Stop) + + // Well-formed encodings, but the signature does not verify against + // the claimed key. + for i := 0; i < 32; i++ { + hm.processMessage(&addrStream{addr: coreapi.Addr{Node: 900}}, &HandshakeMsg{ + Type: HandshakeReject, + NodeID: 900, + Reason: string(rune('a' + i)), + PublicKey: base64.StdEncoding.EncodeToString(make([]byte, 32)), + Signature: base64.StdEncoding.EncodeToString(append(make([]byte, 63), byte(i))), + Timestamp: time.Now().Unix(), + }) + } + + hm.replayMu.Lock() + size := len(hm.replaySet) + hm.replayMu.Unlock() + if size != 0 { + t.Fatalf("messages that fail signature verification must not be recorded; replaySet has %d entries", size) + } + + // An authenticated message is still recorded. + hm.mu.Lock() + hm.outgoing[901] = time.Now() + hm.mu.Unlock() + hm.processMessage(&addrStream{addr: coreapi.Addr{Node: 901}}, &HandshakeMsg{ + Type: HandshakeReject, + NodeID: 901, + Timestamp: time.Now().Unix(), + }) + + hm.replayMu.Lock() + size = len(hm.replaySet) + hm.replayMu.Unlock() + if size != 1 { + t.Fatalf("an authenticated message must be recorded exactly once; replaySet has %d entries", size) + } +} + +func TestRecordReplay_PerPeerCapEvictsOnlyThatPeersEntries(t *testing.T) { + t.Parallel() + hm := newTestHM(t, "") + t.Cleanup(hm.Stop) + + base := time.Now().Add(-time.Minute) + + // A quiet peer's entries must survive a loud peer's flood. + var quiet [][32]byte + for i := 0; i < 4; i++ { + var h [32]byte + h[0] = 0xAA + h[1] = byte(i) + quiet = append(quiet, h) + if !hm.recordReplay(1, h, base.Add(time.Duration(i)*time.Millisecond)) { + t.Fatalf("recordReplay(quiet, %d) returned false", i) + } + } + + for i := 0; i < maxReplayPerPeer+64; i++ { + var h [32]byte + h[0] = 0xBB + h[1] = byte(i) + h[2] = byte(i >> 8) + if !hm.recordReplay(2, h, base.Add(time.Duration(i)*time.Millisecond)) { + t.Fatalf("recordReplay(loud, %d) returned false", i) + } + } + + hm.replayMu.Lock() + defer hm.replayMu.Unlock() + if got := hm.replayPeer[2]; got > maxReplayPerPeer { + t.Fatalf("loud peer holds %d entries, cap is %d", got, maxReplayPerPeer) + } + if got := hm.replayPeer[1]; got != len(quiet) { + t.Fatalf("quiet peer holds %d entries, want %d", got, len(quiet)) + } + for i, h := range quiet { + if _, ok := hm.replaySet[h]; !ok { + t.Fatalf("quiet peer entry %d was evicted by another peer's traffic", i) + } + } +} + +func TestRecordReplay_RejectsDuplicateHash(t *testing.T) { + t.Parallel() + hm := newTestHM(t, "") + t.Cleanup(hm.Stop) + + h := [32]byte{9, 9, 9} + now := time.Now() + if !hm.recordReplay(5, h, now) { + t.Fatal("first record should succeed") + } + if hm.recordReplay(5, h, now) { + t.Fatal("a duplicate hash must be reported as already seen") + } + hm.replayMu.Lock() + defer hm.replayMu.Unlock() + if hm.replayPeer[5] != 1 { + t.Fatalf("duplicate must not double-count; replayPeer[5] = %d", hm.replayPeer[5]) + } +} + +// ----------------------------------------------------------------------- +// key-bound allowlist gate +// ----------------------------------------------------------------------- + +// keyedTestRuntime adds the optional KeyedTrustChecker behavior to +// testRuntime: nodeID must be in pinned AND present the pinned key. +type keyedTestRuntime struct { + *testRuntime + pinned map[uint32]string +} + +func (r *keyedTestRuntime) IsTrustedWithKey(nodeID uint32, pubKeyB64 string) (string, bool) { + want, ok := r.pinned[nodeID] + if !ok { + return "", false + } + if want != pubKeyB64 { + return "", false + } + return "pinned-agent", true +} + +func newKeyedHM(t *testing.T, pinned map[uint32]string) *Manager { + t.Helper() + rt := &keyedTestRuntime{testRuntime: newTestRuntime(), pinned: pinned} + hm := NewManager(rt) + t.Cleanup(hm.Stop) + return hm +} + +func TestHandleRequest_KeyedGateRejectsMismatchedKey(t *testing.T) { + t.Parallel() + const peer = uint32(3131) + hm := newKeyedHM(t, map[uint32]string{peer: "PINNED-KEY"}) + + hm.handleRequest(nil, &HandshakeMsg{ + Type: HandshakeRequest, + NodeID: peer, + PublicKey: "OTHER-KEY", + Timestamp: time.Now().Unix(), + }, true) + + hm.mu.RLock() + _, trusted := hm.trusted[peer] + _, pending := hm.pending[peer] + hm.mu.RUnlock() + if trusted { + t.Fatal("auto-accept must not fire when the presented key differs from the pinned one") + } + if !pending { + t.Fatal("the request should fall through to pending approval") + } +} + +func TestHandleRequest_KeyedGateAcceptsMatchingKey(t *testing.T) { + t.Parallel() + const peer = uint32(3232) + hm := newKeyedHM(t, map[uint32]string{peer: "PINNED-KEY"}) + + hm.handleRequest(nil, &HandshakeMsg{ + Type: HandshakeRequest, + NodeID: peer, + PublicKey: "PINNED-KEY", + Timestamp: time.Now().Unix(), + }, true) + + hm.mu.RLock() + _, trusted := hm.trusted[peer] + hm.mu.RUnlock() + if !trusted { + t.Fatal("auto-accept must fire when the presented key matches the pinned one") + } +} + +func TestProcessRelayedRequest_KeyedGateHasNoPeerKey(t *testing.T) { + t.Parallel() + const peer = uint32(3333) + hm := newKeyedHM(t, map[uint32]string{peer: "PINNED-KEY"}) + + hm.processRelayedRequest(peer, "relayed") + + hm.mu.RLock() + _, trusted := hm.trusted[peer] + _, pending := hm.pending[peer] + hm.mu.RUnlock() + if trusted { + t.Fatal("the relay carries no peer key, so a pinned entry must not auto-accept") + } + if !pending { + t.Fatal("the relayed request should fall through to pending approval") + } +} + +// ----------------------------------------------------------------------- +// Manager.IsTrustedWithKey +// ----------------------------------------------------------------------- + +// keyBoundTrustService mirrors the method set the daemon type-asserts +// for when it wants the key-bound answer from the handshake service. +// The guard below keeps the Manager's signatures aligned with it. +type keyBoundTrustService interface { + IsTrusted(nodeID uint32) bool + IsTrustedWithKey(nodeID uint32, pubKeyB64 string) bool +} + +var _ keyBoundTrustService = (*Manager)(nil) + +func TestManagerIsTrustedWithKey(t *testing.T) { + t.Parallel() + hm := newTestHM(t, "") + t.Cleanup(hm.Stop) + + hm.mu.Lock() + hm.trusted[1] = &TrustRecord{NodeID: 1, PublicKey: "KEY-ONE", ApprovedAt: time.Now()} + hm.trusted[2] = &TrustRecord{NodeID: 2, ApprovedAt: time.Now()} // relay-established, key not yet backfilled + hm.mu.Unlock() + + if hm.IsTrustedWithKey(99, "KEY-ONE") { + t.Fatal("an unknown node must not be trusted") + } + if !hm.IsTrustedWithKey(1, "KEY-ONE") { + t.Fatal("a matching key must be trusted") + } + if hm.IsTrustedWithKey(1, "KEY-TWO") { + t.Fatal("a differing key must not be trusted") + } + if hm.IsTrustedWithKey(1, "") { + t.Fatal("an empty key must not match a recorded one") + } + if !hm.IsTrustedWithKey(2, "ANY-KEY") { + t.Fatal("a record with no key recorded answers on node ID alone") + } +} diff --git a/zz_handshake_accept_request_test.go b/zz_handshake_accept_request_test.go index 15f0ee5..f7937ed 100644 --- a/zz_handshake_accept_request_test.go +++ b/zz_handshake_accept_request_test.go @@ -372,7 +372,12 @@ func TestProcessMessageDispatchesHandshakeAccept(t *testing.T) { t.Cleanup(hm.Stop) // Omit PublicKey so processMessage skips the signature-required check - // and dispatches through the Type switch to handleAccept. + // and dispatches through the Type switch to handleAccept. The accept + // completes an outgoing request, so one has to be in flight. + hm.mu.Lock() + hm.outgoing[44] = time.Now() + hm.mu.Unlock() + msg := &HandshakeMsg{ Type: HandshakeAccept, NodeID: 44, diff --git a/zz_handshake_connection_parse_test.go b/zz_handshake_connection_parse_test.go index 777a753..94e53cf 100644 --- a/zz_handshake_connection_parse_test.go +++ b/zz_handshake_connection_parse_test.go @@ -21,12 +21,19 @@ import ( type mockStream struct { data []byte pos int - err error // returned immediately from Read when set + err error // returned immediately from Read when set + addr coreapi.Addr // reported by RemoteAddr } func newMockStreamData(data []byte) *mockStream { return &mockStream{data: data} } func newMockStreamErr(err error) *mockStream { return &mockStream{err: err} } +// newMockStreamFrom is newMockStreamData with an authenticated sender +// node ID, as the transport would supply for a real peer connection. +func newMockStreamFrom(node uint32, data []byte) *mockStream { + return &mockStream{data: data, addr: coreapi.Addr{Node: node}} +} + func (s *mockStream) Read(p []byte) (int, error) { if s.err != nil { return 0, s.err @@ -42,7 +49,7 @@ func (s *mockStream) Write(p []byte) (int, error) { return len(p), nil } func (s *mockStream) Close() error { return nil } func (s *mockStream) LocalAddr() coreapi.Addr { return coreapi.Addr{} } func (s *mockStream) LocalPort() uint16 { return 0 } -func (s *mockStream) RemoteAddr() coreapi.Addr { return coreapi.Addr{} } +func (s *mockStream) RemoteAddr() coreapi.Addr { return s.addr } func (s *mockStream) RemotePort() uint16 { return 0 } func (s *mockStream) SetDeadline(time.Time) error { return nil } func (s *mockStream) SetReadDeadline(time.Time) error { return nil } @@ -108,7 +115,7 @@ func TestHandleConnectionValidMsgDispatchesToProcessMessage(t *testing.T) { } raw, _ := json.Marshal(&msg) - stream := newMockStreamData(raw) + stream := newMockStreamFrom(321, raw) done := make(chan struct{}) go func() { @@ -190,8 +197,10 @@ func TestProcessMessageReplayCaughtOnSecondDispatch(t *testing.T) { Timestamp: time.Now().Unix(), } + stream := &addrStream{addr: coreapi.Addr{Node: 33}} + // First call: clean dispatch → outgoing[33] deleted + hash inserted in replaySet - hm.processMessage(nil, msg) + hm.processMessage(stream, msg) hm.mu.RLock() _, firstStillOut := hm.outgoing[33] hm.mu.RUnlock() @@ -204,7 +213,7 @@ func TestProcessMessageReplayCaughtOnSecondDispatch(t *testing.T) { hm.mu.Unlock() // Second call: replay detected → early return, outgoing[33] NOT mutated - hm.processMessage(nil, msg) + hm.processMessage(stream, msg) hm.mu.RLock() _, stillOut := hm.outgoing[33] hm.mu.RUnlock() diff --git a/zz_logic_test.go b/zz_logic_test.go index 22ed928..a28c618 100644 --- a/zz_logic_test.go +++ b/zz_logic_test.go @@ -155,8 +155,9 @@ func TestReapReplayRemovesStaleEntriesKeepsFresh(t *testing.T) { stale := [32]byte{1} fresh := [32]byte{2} hm.replayMu.Lock() - hm.replaySet[stale] = time.Now().Add(-24 * time.Hour) - hm.replaySet[fresh] = time.Now() + hm.replaySet[stale] = replayEntry{seen: time.Now().Add(-24 * time.Hour), peer: 1} + hm.replaySet[fresh] = replayEntry{seen: time.Now(), peer: 1} + hm.replayPeer[1] = 2 hm.replayMu.Unlock() hm.reapReplay() diff --git a/zz_pending_cleanup_test.go b/zz_pending_cleanup_test.go index 40f102f..07a4c2f 100644 --- a/zz_pending_cleanup_test.go +++ b/zz_pending_cleanup_test.go @@ -27,7 +27,8 @@ func TestMarkTrustedClearsPending(t *testing.T) { pending: make(map[uint32]*PendingHandshake), outgoing: make(map[uint32]time.Time), revoked: make(map[uint32]time.Time), - replaySet: make(map[[32]byte]time.Time), + replaySet: make(map[[32]byte]replayEntry), + replayPeer: make(map[uint32]int), trustWaiters: make(map[uint32][]chan struct{}), } @@ -55,7 +56,8 @@ func TestMarkTrustedClearsOutgoing(t *testing.T) { pending: make(map[uint32]*PendingHandshake), outgoing: make(map[uint32]time.Time), revoked: make(map[uint32]time.Time), - replaySet: make(map[[32]byte]time.Time), + replaySet: make(map[[32]byte]replayEntry), + replayPeer: make(map[uint32]int), trustWaiters: make(map[uint32][]chan struct{}), } @@ -80,7 +82,8 @@ func TestMarkTrustedClearsBothMaps(t *testing.T) { pending: make(map[uint32]*PendingHandshake), outgoing: make(map[uint32]time.Time), revoked: make(map[uint32]time.Time), - replaySet: make(map[[32]byte]time.Time), + replaySet: make(map[[32]byte]replayEntry), + replayPeer: make(map[uint32]int), trustWaiters: make(map[uint32][]chan struct{}), } @@ -113,7 +116,8 @@ func TestMarkTrustedNoEntryIsIdempotent(t *testing.T) { pending: make(map[uint32]*PendingHandshake), outgoing: make(map[uint32]time.Time), revoked: make(map[uint32]time.Time), - replaySet: make(map[[32]byte]time.Time), + replaySet: make(map[[32]byte]replayEntry), + replayPeer: make(map[uint32]int), trustWaiters: make(map[uint32][]chan struct{}), } diff --git a/zz_ppa001_accept_revoke_identity_test.go b/zz_ppa001_accept_revoke_identity_test.go index 68ea34a..805face 100644 --- a/zz_ppa001_accept_revoke_identity_test.go +++ b/zz_ppa001_accept_revoke_identity_test.go @@ -64,6 +64,10 @@ func TestPPA001_HonestAcceptOverMatchingStreamTrusts(t *testing.T) { hm, _ := newBoundHM(t, 1) const peer = uint32(500) + hm.mu.Lock() + hm.outgoing[peer] = time.Now() + hm.mu.Unlock() + msg := &HandshakeMsg{ Type: HandshakeAccept, NodeID: peer, diff --git a/zz_ppa008_replay_key_timestamp_test.go b/zz_ppa008_replay_key_timestamp_test.go index b3292fd..ea69d24 100644 --- a/zz_ppa008_replay_key_timestamp_test.go +++ b/zz_ppa008_replay_key_timestamp_test.go @@ -8,6 +8,7 @@ import ( "testing" "time" + "github.com/pilot-protocol/common/coreapi" "github.com/pilot-protocol/common/crypto" ) @@ -39,7 +40,7 @@ func TestPPA008_SignedMessageReStampedIsDeduped(t *testing.T) { hm.mu.Lock() hm.outgoing[peer] = time.Now() hm.mu.Unlock() - hm.processMessage(nil, first) + hm.processMessage(&addrStream{addr: coreapi.Addr{Node: peer}}, first) hm.mu.RLock() _, clearedByFirst := hm.outgoing[peer] @@ -59,7 +60,7 @@ func TestPPA008_SignedMessageReStampedIsDeduped(t *testing.T) { Signature: sig, Timestamp: first.Timestamp + 7, } - hm.processMessage(nil, reStamped) + hm.processMessage(&addrStream{addr: coreapi.Addr{Node: peer}}, reStamped) hm.mu.RLock() _, stillOutgoing := hm.outgoing[peer] @@ -80,8 +81,8 @@ func TestPPA008_DistinctSendersGetDistinctReplayKeys(t *testing.T) { hm.mu.Unlock() ts := time.Now().Unix() - hm.processMessage(nil, &HandshakeMsg{Type: HandshakeReject, NodeID: 11, Timestamp: ts}) - hm.processMessage(nil, &HandshakeMsg{Type: HandshakeReject, NodeID: 22, Timestamp: ts}) + hm.processMessage(&addrStream{addr: coreapi.Addr{Node: 11}}, &HandshakeMsg{Type: HandshakeReject, NodeID: 11, Timestamp: ts}) + hm.processMessage(&addrStream{addr: coreapi.Addr{Node: 22}}, &HandshakeMsg{Type: HandshakeReject, NodeID: 22, Timestamp: ts}) hm.mu.RLock() _, out11 := hm.outgoing[11]