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]