From 0cbb44e02f1fe3aee5a7b0cb18d6a05ac251ff30 Mon Sep 17 00:00:00 2001 From: Teodor Calin Date: Thu, 23 Jul 2026 23:32:13 +0300 Subject: [PATCH 1/2] =?UTF-8?q?handshake:=20WS5=20hardening=20=E2=80=94=20?= =?UTF-8?q?reject=20empty-pubkey=20requests,=20per-source=20pending=20cap,?= =?UTF-8?q?=20sanitize=20justification?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Rejects HandshakeRequest messages with no pubkey/signature before they enter the pending queue or auto-trust path. Bounds the pending map per source (maxPendingPerSource=16) on top of the global cap to limit flood surface. Sanitizes attacker-controlled justification before display. Adds WS5 flood test. No enforcement flag is toggled; empty-pubkey reject is unconditional. Co-Authored-By: Claude Opus 4.8 --- go.mod | 2 +- go.sum | 4 +- handshake.go | 180 ++++++++++++++++++++++++--------- zz_ws5_flood_test.go | 235 +++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 370 insertions(+), 51 deletions(-) create mode 100644 zz_ws5_flood_test.go diff --git a/go.mod b/go.mod index bc095e2..4b8f895 100644 --- a/go.mod +++ b/go.mod @@ -11,7 +11,7 @@ require ( github.com/coder/websocket v1.8.15 // indirect github.com/pilot-protocol/rendezvous v0.2.5 // indirect github.com/pilot-protocol/trustedagents v0.2.4 // indirect - golang.org/x/sys v0.46.0 // indirect + golang.org/x/sys v0.47.0 // indirect ) replace github.com/pilot-protocol/pilotprotocol => ../web4 diff --git a/go.sum b/go.sum index bba5283..b8a1a3e 100644 --- a/go.sum +++ b/go.sum @@ -4,5 +4,5 @@ github.com/pilot-protocol/rendezvous v0.2.5 h1:PvApwKHU2DnvK8pA6K9RAohUomZNu6vX6 github.com/pilot-protocol/rendezvous v0.2.5/go.mod h1:nUPZaM1R1x0iEKbTp1TuYqro2wgBArwT0E4wauR8I2E= github.com/pilot-protocol/trustedagents v0.2.4 h1:NqYjU3eoxBzyzzlhTQY2vV4q0l5+4eaADAhrOD1OkQU= github.com/pilot-protocol/trustedagents v0.2.4/go.mod h1:Y3Eq/IOZqAUIbtzVBcxW4epkciaFAFAYkEyu/E7DXe4= -golang.org/x/sys v0.46.0 h1:noSf2Fq6F8DBgS+LysIkx7rIExoNHJsxOAtPp4rthXw= -golang.org/x/sys v0.46.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= +golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs= +golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= diff --git a/handshake.go b/handshake.go index e423f61..469b434 100644 --- a/handshake.go +++ b/handshake.go @@ -3,21 +3,23 @@ package handshake import ( - "crypto/subtle" "crypto/sha256" + "crypto/subtle" "encoding/base64" "encoding/json" "fmt" "log/slog" "os" "path/filepath" + "strings" "sync" "time" + "unicode" "github.com/pilot-protocol/common/coreapi" - "github.com/pilot-protocol/common/protocol" "github.com/pilot-protocol/common/crypto" "github.com/pilot-protocol/common/fsutil" + "github.com/pilot-protocol/common/protocol" ) // Handshake message types. @@ -54,6 +56,7 @@ type PendingHandshake struct { PublicKey string Justification string ReceivedAt time.Time + Source uint32 } // Handshake timing constants. @@ -65,6 +68,8 @@ const ( handshakeCloseDelay = 500 * time.Millisecond // delay before closing after send to let data flush maxReplaySetEntries = 8192 // cap replay set to prevent unbounded growth between reaps maxPendingHandshakes = 256 // cap pending (unapproved) handshake requests + maxPendingPerSource = 16 + maxJustificationLen = 1024 // pendingHandshakeTTL caps how long a pending (unapproved-by- // operator) handshake request can sit in hm.pending before the // reaper drops it. Without this, a request from a peer that the @@ -77,20 +82,21 @@ const ( // Manager handles the trust handshake protocol on port 444. type Manager struct { - mu sync.RWMutex - rt Runtime - ln coreapi.Listener // bound on PortHandshake; closed by Stop - trusted map[uint32]*TrustRecord // approved peers - pending map[uint32]*PendingHandshake // incoming unapproved requests - outgoing map[uint32]time.Time // nodes we've sent requests to → sent-at (for TTL reap) - revoked map[uint32]time.Time // peer → cooldown-until (blocks stale relayed approvals) - storePath string // path to persist trust state (empty = no persistence) - stopping bool // set under mu.Lock() before wg.Wait() in Stop - wg sync.WaitGroup // tracks background RPCs for clean shutdown - reapStop chan struct{} // signals replay reaper to stop - stopOnce sync.Once // ensures reapStop is closed only once - dirty chan struct{} // buffered 1: non-blocking signal to drain goroutine - done chan struct{} // closed in Stop to signal drain goroutine exit + mu sync.RWMutex + rt Runtime + ln coreapi.Listener // bound on PortHandshake; closed by Stop + trusted map[uint32]*TrustRecord // approved peers + pending map[uint32]*PendingHandshake // incoming unapproved requests + pendingBySource map[uint32]int + outgoing map[uint32]time.Time // nodes we've sent requests to → sent-at (for TTL reap) + revoked map[uint32]time.Time // peer → cooldown-until (blocks stale relayed approvals) + storePath string // path to persist trust state (empty = no persistence) + stopping bool // set under mu.Lock() before wg.Wait() in Stop + wg sync.WaitGroup // tracks background RPCs for clean shutdown + reapStop chan struct{} // signals replay reaper to stop + stopOnce sync.Once // ensures reapStop is closed only once + dirty chan struct{} // buffered 1: non-blocking signal to drain goroutine + done chan struct{} // closed in Stop to signal drain goroutine exit // Replay protection replayMu sync.Mutex @@ -108,15 +114,16 @@ type Manager struct { // path; otherwise the manager is in-memory only. func NewManager(rt Runtime) *Manager { hm := &Manager{ - rt: rt, - trusted: make(map[uint32]*TrustRecord), - pending: make(map[uint32]*PendingHandshake), - outgoing: make(map[uint32]time.Time), - revoked: make(map[uint32]time.Time), - replaySet: make(map[[32]byte]time.Time), - trustWaiters: make(map[uint32][]chan struct{}), - dirty: make(chan struct{}, 1), - done: make(chan struct{}), + rt: rt, + trusted: make(map[uint32]*TrustRecord), + pending: make(map[uint32]*PendingHandshake), + pendingBySource: make(map[uint32]int), + outgoing: make(map[uint32]time.Time), + revoked: make(map[uint32]time.Time), + replaySet: make(map[[32]byte]time.Time), + trustWaiters: make(map[uint32][]chan struct{}), + dirty: make(chan struct{}, 1), + done: make(chan struct{}), } if path := rt.IdentityPath(); path != "" { @@ -201,7 +208,7 @@ func (hm *Manager) goRPCLocked(fn func()) { // ~724, etc.) remain correct. func (hm *Manager) markTrustedLocked(nodeID uint32, rec *TrustRecord) { hm.trusted[nodeID] = rec - delete(hm.pending, nodeID) + hm.deletePendingLocked(nodeID) delete(hm.outgoing, nodeID) hm.trustWaitersMu.Lock() waiters := hm.trustWaiters[nodeID] @@ -277,9 +284,9 @@ func (hm *Manager) removeWaiter(nodeID uint32, target chan struct{}) { // --- Trust persistence --- type trustSnapshot struct { - Trusted []trustSnapshotEntry `json:"trusted"` - Pending []pendingSnapshotEntry `json:"pending,omitempty"` - Revoked []revokedSnapshotEntry `json:"revoked,omitempty"` + Trusted []trustSnapshotEntry `json:"trusted"` + Pending []pendingSnapshotEntry `json:"pending,omitempty"` + Revoked []revokedSnapshotEntry `json:"revoked,omitempty"` } type trustSnapshotEntry struct { @@ -295,11 +302,12 @@ type pendingSnapshotEntry struct { PublicKey string `json:"public_key,omitempty"` Justification string `json:"justification,omitempty"` ReceivedAt string `json:"received_at"` + Source uint32 `json:"source,omitempty"` } type revokedSnapshotEntry struct { - NodeID uint32 `json:"node_id"` - Until string `json:"until"` // RFC3339 timestamp of cooldown expiry + NodeID uint32 `json:"node_id"` + Until string `json:"until"` // RFC3339 timestamp of cooldown expiry } func (hm *Manager) saveTrust() { @@ -326,6 +334,7 @@ func (hm *Manager) saveTrust() { PublicKey: p.PublicKey, Justification: p.Justification, ReceivedAt: p.ReceivedAt.Format(time.RFC3339), + Source: p.Source, }) } for nodeID, until := range hm.revoked { @@ -422,6 +431,13 @@ func (hm *Manager) loadTrust() { PublicKey: e.PublicKey, Justification: e.Justification, ReceivedAt: received, + Source: e.Source, + } + if e.Source != 0 { + if hm.pendingBySource == nil { + hm.pendingBySource = make(map[uint32]int) + } + hm.pendingBySource[e.Source]++ } } for _, e := range snap.Revoked { @@ -556,6 +572,10 @@ func (hm *Manager) processMessage(stream coreapi.Stream, msg *HandshakeMsg) { // M12 fix: verify P2P signature if the sender provides a public key registryBound := false + if msg.PublicKey == "" && msg.Type == HandshakeRequest { + slog.Warn("handshake: request missing public key, rejecting", "peer_node_id", msg.NodeID) + return + } if msg.PublicKey != "" { if msg.Signature == "" { slog.Warn("handshake: missing signature from authenticated node", "peer_node_id", msg.NodeID) @@ -647,6 +667,48 @@ func (hm *Manager) reapOutgoingAndRevoked() { } } +func (hm *Manager) deletePendingLocked(nodeID uint32) { + p, ok := hm.pending[nodeID] + if !ok { + return + } + delete(hm.pending, nodeID) + if p.Source == 0 || hm.pendingBySource == nil { + return + } + if hm.pendingBySource[p.Source] <= 1 { + delete(hm.pendingBySource, p.Source) + } else { + hm.pendingBySource[p.Source]-- + } +} + +func (hm *Manager) setPendingLocked(nodeID uint32, entry *PendingHandshake) { + hm.deletePendingLocked(nodeID) + hm.pending[nodeID] = entry + if entry.Source == 0 { + return + } + if hm.pendingBySource == nil { + hm.pendingBySource = make(map[uint32]int) + } + hm.pendingBySource[entry.Source]++ +} + +func sanitizeJustification(s string) string { + var b strings.Builder + for _, r := range s { + if unicode.IsPrint(r) { + b.WriteRune(r) + } + } + out := b.String() + if len(out) > maxJustificationLen { + out = out[:maxJustificationLen] + } + return out +} + // handleRequest processes an incoming handshake request. // // registryBound is true iff processMessage confirmed the claimed @@ -655,11 +717,15 @@ func (hm *Manager) reapOutgoingAndRevoked() { // agents auto-accept path; signature-only authentication is not // enough to safely key on node_id. func (hm *Manager) handleRequest(stream coreapi.Stream, msg *HandshakeMsg, registryBound bool) { - _ = stream // reserved for future response-on-same-stream support peerNodeID := msg.NodeID - slog.Info("handshake request received", "peer_node_id", peerNodeID, "justification", msg.Justification) + var source uint32 + if stream != nil { + source = stream.RemoteAddr().Node + } + justification := sanitizeJustification(msg.Justification) + slog.Info("handshake request received", "peer_node_id", peerNodeID, "justification", justification) hm.rt.PublishEvent("handshake.received", map[string]interface{}{ - "peer_node_id": peerNodeID, "justification": msg.Justification, + "peer_node_id": peerNodeID, "justification": justification, }) hm.mu.Lock() @@ -785,20 +851,27 @@ func (hm *Manager) handleRequest(stream coreapi.Stream, msg *HandshakeMsg, regis } // Store as pending (cap to prevent unbounded growth from spam) - if _, exists := hm.pending[peerNodeID]; !exists && len(hm.pending) >= maxPendingHandshakes { - slog.Warn("pending handshake queue full, rejecting", "peer_node_id", peerNodeID) - return + if _, exists := hm.pending[peerNodeID]; !exists { + if len(hm.pending) >= maxPendingHandshakes { + slog.Warn("pending handshake queue full, rejecting", "peer_node_id", peerNodeID) + return + } + if source != 0 && hm.pendingBySource[source] >= maxPendingPerSource { + slog.Warn("pending handshake queue full for source, rejecting", "peer_node_id", peerNodeID, "source", source) + return + } } - hm.pending[peerNodeID] = &PendingHandshake{ + hm.setPendingLocked(peerNodeID, &PendingHandshake{ NodeID: peerNodeID, PublicKey: msg.PublicKey, - Justification: msg.Justification, + Justification: justification, ReceivedAt: time.Now(), - } + Source: source, + }) hm.markDirty() slog.Info("handshake request pending approval", "peer_node_id", peerNodeID) hm.rt.PublishEvent("handshake.pending", map[string]interface{}{ - "peer_node_id": peerNodeID, "justification": msg.Justification, + "peer_node_id": peerNodeID, "justification": justification, }) } @@ -906,6 +979,7 @@ func (hm *Manager) ProcessRelayedRequest(fromNodeID uint32, justification string // processRelayedRequest handles a handshake request received via registry relay. func (hm *Manager) processRelayedRequest(fromNodeID uint32, justification string) { + justification = sanitizeJustification(justification) hm.mu.Lock() defer hm.mu.Unlock() @@ -1021,11 +1095,22 @@ func (hm *Manager) processRelayedRequest(fromNodeID uint32, justification string } // Store as pending (for manual approval via pilotctl approve) - hm.pending[fromNodeID] = &PendingHandshake{ + if _, exists := hm.pending[fromNodeID]; !exists { + if len(hm.pending) >= maxPendingHandshakes { + slog.Warn("pending handshake queue full, rejecting relayed request", "peer_node_id", fromNodeID) + return + } + if hm.pendingBySource[fromNodeID] >= maxPendingPerSource { + slog.Warn("pending handshake queue full for source, rejecting relayed request", "peer_node_id", fromNodeID) + return + } + } + hm.setPendingLocked(fromNodeID, &PendingHandshake{ NodeID: fromNodeID, Justification: justification, ReceivedAt: time.Now(), - } + Source: fromNodeID, + }) hm.markDirty() slog.Info("relayed handshake request pending approval", "from_node_id", fromNodeID, "justification", justification) } @@ -1125,7 +1210,6 @@ func (hm *Manager) ApproveHandshake(peerNodeID uint32) error { hm.mu.Unlock() return nil } - delete(hm.pending, peerNodeID) hm.markTrustedLocked(peerNodeID, &TrustRecord{ NodeID: peerNodeID, PublicKey: req.PublicKey, @@ -1161,7 +1245,7 @@ func (hm *Manager) ApproveHandshake(peerNodeID uint32) error { // RejectHandshake rejects a pending handshake request. func (hm *Manager) RejectHandshake(peerNodeID uint32, reason string) error { hm.mu.Lock() - delete(hm.pending, peerNodeID) + hm.deletePendingLocked(peerNodeID) hm.markDirty() hm.mu.Unlock() @@ -1204,7 +1288,7 @@ func (hm *Manager) RevokeTrust(peerNodeID uint32) error { _, wasTrusted := hm.trusted[peerNodeID] _, wasPending := hm.pending[peerNodeID] delete(hm.trusted, peerNodeID) - delete(hm.pending, peerNodeID) + hm.deletePendingLocked(peerNodeID) delete(hm.outgoing, peerNodeID) // Block stale relayed approvals still sitting in the registry inbox from // re-establishing trust right after a local revoke. 5-minute cooldown covers @@ -1269,7 +1353,7 @@ func (hm *Manager) handleRevokeMsg(msg *HandshakeMsg) { _, wasTrusted := hm.trusted[peerNodeID] _, wasPending := hm.pending[peerNodeID] delete(hm.trusted, peerNodeID) - delete(hm.pending, peerNodeID) + hm.deletePendingLocked(peerNodeID) delete(hm.outgoing, peerNodeID) if wasTrusted || wasPending { hm.markDirty() @@ -1449,7 +1533,7 @@ func (hm *Manager) reapStalePending() { return } for _, id := range stale { - delete(hm.pending, id) + hm.deletePendingLocked(id) } hm.markDirty() slog.Info("reaped stale pending handshakes", diff --git a/zz_ws5_flood_test.go b/zz_ws5_flood_test.go new file mode 100644 index 0000000..15a1b1b --- /dev/null +++ b/zz_ws5_flood_test.go @@ -0,0 +1,235 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +package handshake + +import ( + "encoding/base64" + "fmt" + "testing" + "time" + + "github.com/pilot-protocol/common/coreapi" + "github.com/pilot-protocol/common/crypto" +) + +type addrStream struct { + addr coreapi.Addr +} + +func (s *addrStream) Read(p []byte) (int, error) { return 0, errStub("addrStream: not implemented") } +func (s *addrStream) Write(p []byte) (int, error) { return len(p), nil } +func (s *addrStream) Close() error { return nil } +func (s *addrStream) LocalAddr() coreapi.Addr { return coreapi.Addr{} } +func (s *addrStream) LocalPort() uint16 { return 0 } +func (s *addrStream) RemoteAddr() coreapi.Addr { return s.addr } +func (s *addrStream) RemotePort() uint16 { return 0 } +func (s *addrStream) SetDeadline(time.Time) error { return nil } +func (s *addrStream) SetReadDeadline(time.Time) error { return nil } +func (s *addrStream) SetWriteDeadline(time.Time) error { return nil } + +func signedHandshakeRequest(t *testing.T, id *crypto.Identity, claimedNodeID, selfNodeID uint32, justification string) *HandshakeMsg { + t.Helper() + challenge := fmt.Sprintf("handshake:%d:%d", claimedNodeID, selfNodeID) + sig := id.Sign([]byte(challenge)) + return &HandshakeMsg{ + Type: HandshakeRequest, + NodeID: claimedNodeID, + PublicKey: base64.StdEncoding.EncodeToString(id.PublicKey), + Signature: base64.StdEncoding.EncodeToString(sig), + Justification: justification, + Timestamp: time.Now().Unix(), + } +} + +func TestProcessMessageEmptyPublicKeyRequestRejected(t *testing.T) { + t.Parallel() + rt := newTestRuntime() + rt.nodeID = 1 + rt.trustAutoApprove = true + hm := NewManager(rt) + t.Cleanup(hm.Stop) + + msg := &HandshakeMsg{ + Type: HandshakeRequest, + NodeID: 424242, + Timestamp: time.Now().Unix(), + } + hm.processMessage(nil, msg) + + hm.mu.RLock() + _, pending := hm.pending[424242] + _, trusted := hm.trusted[424242] + hm.mu.RUnlock() + if pending { + t.Fatal("empty-pubkey handshake_request must never be enqueued in pending") + } + if trusted { + t.Fatal("empty-pubkey handshake_request must never be auto-approved, even with trust-auto-approve on") + } +} + +func TestProcessMessageEmptyPublicKeyWithSignaturePresentStillRejected(t *testing.T) { + t.Parallel() + rt := newTestRuntime() + rt.nodeID = 1 + rt.trustAutoApprove = true + hm := NewManager(rt) + t.Cleanup(hm.Stop) + + msg := &HandshakeMsg{ + Type: HandshakeRequest, + NodeID: 555555, + Signature: "not-empty-but-meaningless-without-a-pubkey", + Timestamp: time.Now().Unix(), + } + hm.processMessage(nil, msg) + + hm.mu.RLock() + _, pending := hm.pending[555555] + _, trusted := hm.trusted[555555] + hm.mu.RUnlock() + if pending || trusted { + t.Fatal("handshake_request with empty PublicKey must be rejected regardless of Signature content") + } +} + +func TestProcessMessageEmptySignatureWithPublicKeyRejected(t *testing.T) { + t.Parallel() + rt := newTestRuntime() + rt.nodeID = 1 + rt.trustAutoApprove = true + hm := NewManager(rt) + t.Cleanup(hm.Stop) + + id, err := crypto.GenerateIdentity() + if err != nil { + t.Fatalf("generate identity: %v", err) + } + + msg := &HandshakeMsg{ + Type: HandshakeRequest, + NodeID: 666666, + PublicKey: base64.StdEncoding.EncodeToString(id.PublicKey), + Timestamp: time.Now().Unix(), + } + hm.processMessage(nil, msg) + + hm.mu.RLock() + _, pending := hm.pending[666666] + _, trusted := hm.trusted[666666] + hm.mu.RUnlock() + if pending || trusted { + t.Fatal("handshake_request with non-empty PublicKey but empty Signature must be rejected") + } +} + +func TestPendingQueuePerSourceCapPreventsSingleSourceExhaustion(t *testing.T) { + t.Parallel() + rt := newTestRuntime() + rt.nodeID = 1 + hm := NewManager(rt) + t.Cleanup(hm.Stop) + + attacker, err := crypto.GenerateIdentity() + if err != nil { + t.Fatalf("generate identity: %v", err) + } + streamA := &addrStream{addr: coreapi.Addr{Network: 0, Node: 111}} + + for i := 0; i < maxPendingPerSource; i++ { + nodeID := uint32(5000 + i) + msg := signedHandshakeRequest(t, attacker, nodeID, rt.nodeID, "spam") + hm.processMessage(streamA, msg) + hm.mu.RLock() + _, ok := hm.pending[nodeID] + hm.mu.RUnlock() + if !ok { + t.Fatalf("request %d from source A should have been accepted (under per-source cap)", i) + } + } + + hm.mu.RLock() + countAfterFill := len(hm.pending) + hm.mu.RUnlock() + if countAfterFill != maxPendingPerSource { + t.Fatalf("pending count = %d, want %d", countAfterFill, maxPendingPerSource) + } + + overCapMsg := signedHandshakeRequest(t, attacker, 6000, rt.nodeID, "spam-overflow") + hm.processMessage(streamA, overCapMsg) + hm.mu.RLock() + _, overCapAccepted := hm.pending[6000] + hm.mu.RUnlock() + if overCapAccepted { + t.Fatal("source A should not be able to exceed its per-source pending cap") + } + + victim, err := crypto.GenerateIdentity() + if err != nil { + t.Fatalf("generate identity: %v", err) + } + streamB := &addrStream{addr: coreapi.Addr{Network: 0, Node: 222}} + legitMsg := signedHandshakeRequest(t, victim, 7000, rt.nodeID, "legitimate request") + hm.processMessage(streamB, legitMsg) + + hm.mu.RLock() + _, legitAccepted := hm.pending[7000] + hm.mu.RUnlock() + if !legitAccepted { + t.Fatal("a request from a different source must still get a pending slot after source A is capped") + } +} + +func TestJustificationSanitizedStripsControlCharsAndNewlines(t *testing.T) { + t.Parallel() + rt := newTestRuntime() + rt.nodeID = 1 + hm := NewManager(rt) + t.Cleanup(hm.Stop) + + attacker, err := crypto.GenerateIdentity() + if err != nil { + t.Fatalf("generate identity: %v", err) + } + + dirty := "line-one\nline-two\r\ttab\x00null\x1bescape" + msg := signedHandshakeRequest(t, attacker, 8000, rt.nodeID, dirty) + hm.processMessage(nil, msg) + + hm.mu.RLock() + p, ok := hm.pending[8000] + hm.mu.RUnlock() + if !ok { + t.Fatal("pending[8000] missing") + } + if p.Justification == dirty { + t.Fatalf("justification was not sanitized: %q", p.Justification) + } + for _, r := range p.Justification { + if r == '\n' || r == '\r' || r == '\t' || r == 0x00 || r == 0x1b { + t.Fatalf("sanitized justification still contains control char %q: %q", r, p.Justification) + } + } + if p.Justification != "line-oneline-twotabnullescape" { + t.Fatalf("unexpected sanitized justification: %q", p.Justification) + } +} + +func TestJustificationSanitizationAppliedOnRelayedPath(t *testing.T) { + t.Parallel() + hm := newTestHM(t, "") + t.Cleanup(hm.Stop) + + dirty := "hello\nworld\x00" + hm.processRelayedRequest(9000, dirty) + + hm.mu.RLock() + p, ok := hm.pending[9000] + hm.mu.RUnlock() + if !ok { + t.Fatal("pending[9000] missing") + } + if p.Justification != "helloworld" { + t.Fatalf("relayed justification not sanitized: %q", p.Justification) + } +} From 7b2668e87af31833b0130dde23c9bc51b94202b2 Mon Sep 17 00:00:00 2001 From: Teodor Calin Date: Fri, 24 Jul 2026 20:31:01 +0300 Subject: [PATCH 2/2] security: bind accept/revoke to authenticated identity, gate relayed approval, timestamp-independent replay key Addresses reported vulnerabilities PPA-001/002/008 (replay-set), verified against source. - PPA-001: HandshakeAccept/HandshakeRevoke now require the claimed msg.NodeID to equal the AEAD-authenticated stream identity (stream.RemoteAddr().Node). A forged unsigned accept/revoke for a victim node arrives on the attacker's own authenticated stream and is rejected; honest accept/revoke (msg.NodeID == own id == authenticated peer) still complete. Closes the trust-injection / forced-revoke bypass that reached the accept/revoke handlers by choosing a message Type other than Request. - PPA-002: processRelayedApproval requires a matching outgoing request before granting Mutual trust, so an unsolicited relayed approval for a peer we never contacted no longer establishes trust. - PPA-008 (replay-set): the replay/dedup key is derived from signature-covered fields plus message Type instead of the full JSON (which included the mutable, unsigned Timestamp), so a captured signed message re-stamped with a fresh timestamp is caught. Including Type keeps a legitimate revoke following a request from being falsely deduped. The wire-breaking signed-freshness change (nonce inside the signed preimage) is deferred. Co-Authored-By: Claude Opus 4.8 --- handshake.go | 22 ++- zz_handshake_accept_request_test.go | 3 +- zz_handshake_connection_parse_test.go | 2 +- zz_handshake_relayed_trust_test.go | 16 +- zz_ppa001_accept_revoke_identity_test.go | 152 ++++++++++++++++++ ...a002_relayed_approval_precondition_test.go | 49 ++++++ zz_ppa008_replay_key_timestamp_test.go | 93 +++++++++++ 7 files changed, 321 insertions(+), 16 deletions(-) create mode 100644 zz_ppa001_accept_revoke_identity_test.go create mode 100644 zz_ppa002_relayed_approval_precondition_test.go create mode 100644 zz_ppa008_replay_key_timestamp_test.go diff --git a/handshake.go b/handshake.go index 469b434..ffee263 100644 --- a/handshake.go +++ b/handshake.go @@ -553,9 +553,8 @@ func (hm *Manager) processMessage(stream coreapi.Stream, msg *HandshakeMsg) { return } - // Replay detection: hash the message and check set - msgBytes, _ := json.Marshal(msg) - msgHash := sha256.Sum256(msgBytes) + 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() @@ -616,6 +615,18 @@ func (hm *Manager) processMessage(stream coreapi.Stream, msg *HandshakeMsg) { } } + if msg.Type == HandshakeAccept || msg.Type == HandshakeRevoke { + 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", + "claimed_node_id", msg.NodeID, "authenticated_node_id", authenticated, "type", msg.Type) + return + } + } + switch msg.Type { case HandshakeRequest: hm.handleRequest(stream, msg, registryBound) @@ -1144,6 +1155,11 @@ func (hm *Manager) processRelayedApproval(fromNodeID uint32) { delete(hm.revoked, fromNodeID) } + if _, ok := hm.outgoing[fromNodeID]; !ok { + slog.Warn("dropping relayed approval with no matching outgoing request", "peer_node_id", fromNodeID) + return + } + delete(hm.outgoing, fromNodeID) hm.markTrustedLocked(fromNodeID, &TrustRecord{ NodeID: fromNodeID, diff --git a/zz_handshake_accept_request_test.go b/zz_handshake_accept_request_test.go index 951eadf..5094c6b 100644 --- a/zz_handshake_accept_request_test.go +++ b/zz_handshake_accept_request_test.go @@ -6,6 +6,7 @@ import ( "testing" "time" + "github.com/pilot-protocol/common/coreapi" "github.com/pilot-protocol/common/crypto" ) @@ -307,7 +308,7 @@ func TestProcessMessageDispatchesHandshakeAccept(t *testing.T) { NodeID: 44, Timestamp: time.Now().Unix(), } - hm.processMessage(nil, msg) + hm.processMessage(&addrStream{addr: coreapi.Addr{Node: 44}}, msg) hm.mu.RLock() _, ok := hm.trusted[44] diff --git a/zz_handshake_connection_parse_test.go b/zz_handshake_connection_parse_test.go index 0237c41..777a753 100644 --- a/zz_handshake_connection_parse_test.go +++ b/zz_handshake_connection_parse_test.go @@ -327,7 +327,7 @@ func TestProcessMessageRevokeDispatchRemovesTrusted(t *testing.T) { NodeID: 99, Timestamp: time.Now().Unix(), } - hm.processMessage(nil, msg) + hm.processMessage(&addrStream{addr: coreapi.Addr{Node: 99}}, msg) hm.mu.RLock() _, stillTrusted := hm.trusted[99] diff --git a/zz_handshake_relayed_trust_test.go b/zz_handshake_relayed_trust_test.go index 681e2fa..819f408 100644 --- a/zz_handshake_relayed_trust_test.go +++ b/zz_handshake_relayed_trust_test.go @@ -232,25 +232,19 @@ func TestProcessRelayedApprovalEstablishesTrustAndClearsOutgoing(t *testing.T) { } } -// --- processRelayedApproval: no outgoing entry still establishes trust --- +// --- processRelayedApproval: no outgoing entry is dropped, not trusted --- -func TestProcessRelayedApprovalNoOutgoingStillEstablishes(t *testing.T) { +func TestProcessRelayedApprovalNoOutgoingDropped(t *testing.T) { t.Parallel() - // The outgoing map is not a precondition — if somehow the registry relays an - // approval for a peer we didn't track as outgoing, the code still promotes - // them to trusted (delete of missing key is a safe no-op in Go). hm := newTestHM(t, "") t.Cleanup(hm.Stop) hm.processRelayedApproval(300) hm.mu.RLock() - rec, ok := hm.trusted[300] + _, ok := hm.trusted[300] hm.mu.RUnlock() - if !ok { - t.Fatal("trusted[300] missing — relayed approval without outgoing should still establish") - } - if !rec.Mutual { - t.Fatal("rec.Mutual should be true") + if ok { + t.Fatal("trusted[300] set — relayed approval without an outgoing request must be dropped") } } diff --git a/zz_ppa001_accept_revoke_identity_test.go b/zz_ppa001_accept_revoke_identity_test.go new file mode 100644 index 0000000..68ea34a --- /dev/null +++ b/zz_ppa001_accept_revoke_identity_test.go @@ -0,0 +1,152 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +package handshake + +import ( + "testing" + "time" + + "github.com/pilot-protocol/common/coreapi" +) + +func newBoundHM(t *testing.T, selfNodeID uint32) (*Manager, *testRuntime) { + t.Helper() + rt := newTestRuntime() + rt.nodeID = selfNodeID + hm := NewManager(rt) + t.Cleanup(hm.Stop) + return hm, rt +} + +func TestPPA001_ForgedUnsignedAcceptForVictimRejected(t *testing.T) { + t.Parallel() + hm, _ := newBoundHM(t, 1) + + const victim = uint32(700) + const attacker = uint32(666) + + msg := &HandshakeMsg{ + Type: HandshakeAccept, + NodeID: victim, + Timestamp: time.Now().Unix(), + } + hm.processMessage(&addrStream{addr: coreapi.Addr{Node: attacker}}, msg) + + hm.mu.RLock() + _, trusted := hm.trusted[victim] + hm.mu.RUnlock() + if trusted { + t.Fatal("forged unsigned accept from a non-matching authenticated node must not inject trust") + } +} + +func TestPPA001_UnsignedAcceptWithNoStreamRejected(t *testing.T) { + t.Parallel() + hm, _ := newBoundHM(t, 1) + + msg := &HandshakeMsg{ + Type: HandshakeAccept, + NodeID: 44, + Timestamp: time.Now().Unix(), + } + hm.processMessage(nil, msg) + + hm.mu.RLock() + _, trusted := hm.trusted[44] + hm.mu.RUnlock() + if trusted { + t.Fatal("accept with no authenticated transport must not inject trust") + } +} + +func TestPPA001_HonestAcceptOverMatchingStreamTrusts(t *testing.T) { + t.Parallel() + hm, _ := newBoundHM(t, 1) + + const peer = uint32(500) + msg := &HandshakeMsg{ + Type: HandshakeAccept, + NodeID: peer, + Timestamp: time.Now().Unix(), + } + hm.processMessage(&addrStream{addr: coreapi.Addr{Node: peer}}, msg) + + hm.mu.RLock() + _, trusted := hm.trusted[peer] + hm.mu.RUnlock() + if !trusted { + t.Fatal("honest accept over a matching authenticated stream must still establish trust") + } +} + +func TestPPA001_ForgedUnsignedRevokeForVictimRejected(t *testing.T) { + t.Parallel() + hm, rt := newBoundHM(t, 1) + + const victim = uint32(700) + const attacker = uint32(666) + + hm.mu.Lock() + hm.trusted[victim] = &TrustRecord{NodeID: victim, ApprovedAt: time.Now()} + hm.mu.Unlock() + + msg := &HandshakeMsg{ + Type: HandshakeRevoke, + NodeID: victim, + Timestamp: time.Now().Unix(), + } + hm.processMessage(&addrStream{addr: coreapi.Addr{Node: attacker}}, msg) + + hm.mu.RLock() + _, stillTrusted := hm.trusted[victim] + hm.mu.RUnlock() + if !stillTrusted { + t.Fatal("forged unsigned revoke from a non-matching authenticated node must not revoke trust") + } + + rt.removedPeersMu.Lock() + removed := append([]uint32(nil), rt.removedPeers...) + rt.removedPeersMu.Unlock() + for _, id := range removed { + if id == victim { + t.Fatal("forged revoke must not tear down the victim's tunnel") + } + } +} + +func TestPPA001_HonestRevokeOverMatchingStreamRevokes(t *testing.T) { + t.Parallel() + hm, rt := newBoundHM(t, 1) + + const peer = uint32(800) + hm.mu.Lock() + hm.trusted[peer] = &TrustRecord{NodeID: peer, ApprovedAt: time.Now()} + hm.mu.Unlock() + + msg := &HandshakeMsg{ + Type: HandshakeRevoke, + NodeID: peer, + Timestamp: time.Now().Unix(), + } + hm.processMessage(&addrStream{addr: coreapi.Addr{Node: peer}}, msg) + + hm.mu.RLock() + _, stillTrusted := hm.trusted[peer] + hm.mu.RUnlock() + if stillTrusted { + t.Fatal("honest revoke over a matching authenticated stream must revoke trust") + } + + rt.removedPeersMu.Lock() + removed := append([]uint32(nil), rt.removedPeers...) + rt.removedPeersMu.Unlock() + var tornDown bool + for _, id := range removed { + if id == peer { + tornDown = true + } + } + if !tornDown { + t.Fatal("honest revoke must tear down the peer's tunnel") + } +} diff --git a/zz_ppa002_relayed_approval_precondition_test.go b/zz_ppa002_relayed_approval_precondition_test.go new file mode 100644 index 0000000..9d8bdf5 --- /dev/null +++ b/zz_ppa002_relayed_approval_precondition_test.go @@ -0,0 +1,49 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +package handshake + +import ( + "testing" + "time" +) + +func TestPPA002_RelayedApprovalWithoutOutgoingDropped(t *testing.T) { + t.Parallel() + hm := newTestHM(t, "") + t.Cleanup(hm.Stop) + + hm.ProcessRelayedApproval(4242) + + hm.mu.RLock() + _, trusted := hm.trusted[4242] + hm.mu.RUnlock() + if trusted { + t.Fatal("relayed approval for a peer we never sent a request to must be dropped") + } +} + +func TestPPA002_RelayedApprovalWithOutgoingTrusts(t *testing.T) { + t.Parallel() + hm := newTestHM(t, "") + t.Cleanup(hm.Stop) + + hm.mu.Lock() + hm.outgoing[4242] = time.Now() + hm.mu.Unlock() + + hm.ProcessRelayedApproval(4242) + + hm.mu.RLock() + rec, trusted := hm.trusted[4242] + _, stillOutgoing := hm.outgoing[4242] + hm.mu.RUnlock() + if !trusted { + t.Fatal("relayed approval matching an outgoing request must establish trust") + } + if !rec.Mutual { + t.Fatal("relayed approval must record mutual trust") + } + if stillOutgoing { + t.Fatal("outgoing entry must be cleared once trust is established") + } +} diff --git a/zz_ppa008_replay_key_timestamp_test.go b/zz_ppa008_replay_key_timestamp_test.go new file mode 100644 index 0000000..b3292fd --- /dev/null +++ b/zz_ppa008_replay_key_timestamp_test.go @@ -0,0 +1,93 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +package handshake + +import ( + "encoding/base64" + "fmt" + "testing" + "time" + + "github.com/pilot-protocol/common/crypto" +) + +func TestPPA008_SignedMessageReStampedIsDeduped(t *testing.T) { + t.Parallel() + rt := newTestRuntime() + rt.nodeID = 1 + hm := NewManager(rt) + t.Cleanup(hm.Stop) + + id, err := crypto.GenerateIdentity() + if err != nil { + t.Fatalf("generate identity: %v", err) + } + + const peer = uint32(909) + challenge := fmt.Sprintf("handshake:%d:%d", peer, rt.nodeID) + sig := base64.StdEncoding.EncodeToString(id.Sign([]byte(challenge))) + pub := base64.StdEncoding.EncodeToString(id.PublicKey) + + first := &HandshakeMsg{ + Type: HandshakeReject, + NodeID: peer, + PublicKey: pub, + Signature: sig, + Timestamp: time.Now().Unix(), + } + + hm.mu.Lock() + hm.outgoing[peer] = time.Now() + hm.mu.Unlock() + hm.processMessage(nil, first) + + hm.mu.RLock() + _, clearedByFirst := hm.outgoing[peer] + hm.mu.RUnlock() + if clearedByFirst { + t.Fatal("first signed reject should have dispatched and cleared outgoing") + } + + hm.mu.Lock() + hm.outgoing[peer] = time.Now() + hm.mu.Unlock() + + reStamped := &HandshakeMsg{ + Type: HandshakeReject, + NodeID: peer, + PublicKey: pub, + Signature: sig, + Timestamp: first.Timestamp + 7, + } + hm.processMessage(nil, reStamped) + + hm.mu.RLock() + _, stillOutgoing := hm.outgoing[peer] + hm.mu.RUnlock() + if !stillOutgoing { + t.Fatal("re-stamped copy of a captured signed message must hit the replay set (timestamp must not be part of the key)") + } +} + +func TestPPA008_DistinctSendersGetDistinctReplayKeys(t *testing.T) { + t.Parallel() + hm := newTestHM(t, "") + t.Cleanup(hm.Stop) + + hm.mu.Lock() + hm.outgoing[11] = time.Now() + hm.outgoing[22] = time.Now() + 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.mu.RLock() + _, out11 := hm.outgoing[11] + _, out22 := hm.outgoing[22] + hm.mu.RUnlock() + if out11 || out22 { + t.Fatal("distinct senders must get distinct replay keys and both dispatch — no false dedup") + } +}