diff --git a/handshake.go b/handshake.go index 2dcc311..6fcc64a 100644 --- a/handshake.go +++ b/handshake.go @@ -98,6 +98,17 @@ const ( // operators ample time to decide without letting the queue grow // unboundedly under sustained inbound handshake pressure. pendingHandshakeTTL = 30 * 24 * time.Hour + // outgoingRequestTTL is how long a request this node sent stays + // answerable: the peer keeps it pending for pendingHandshakeTTL, and an + // acceptance is honoured only while the matching outgoing entry exists. + // It used to be 2 x handshakeMaxAge (10 minutes), so a person who + // accepted later than that, as people do, was trusted on their side and + // silently dropped as "unsolicited" on the requester's, which then showed + // the request as waiting forever. + outgoingRequestTTL = pendingHandshakeTTL + // maxOutgoingRequests bounds the outgoing set; past it the oldest + // request is forgotten first. + maxOutgoingRequests = 4096 ) // replayEntry is one recorded handshake message hash: when it was first @@ -301,7 +312,7 @@ func (hm *Manager) goRPCLocked(fn func()) { func (hm *Manager) markTrustedLocked(nodeID uint32, rec *TrustRecord) { hm.trusted[nodeID] = rec hm.deletePendingLocked(nodeID) - delete(hm.outgoing, nodeID) + hm.dropOutgoingLocked(nodeID) hm.trustWaitersMu.Lock() waiters := hm.trustWaiters[nodeID] delete(hm.trustWaiters, nodeID) @@ -379,6 +390,14 @@ type trustSnapshot struct { Trusted []trustSnapshotEntry `json:"trusted"` Pending []pendingSnapshotEntry `json:"pending,omitempty"` Revoked []revokedSnapshotEntry `json:"revoked,omitempty"` + // Outgoing are the requests this node sent and is still waiting on, so + // an acceptance that arrives after a restart is still honoured. + Outgoing []outgoingSnapshotEntry `json:"outgoing,omitempty"` +} + +type outgoingSnapshotEntry struct { + NodeID uint32 `json:"node_id"` + SentAt string `json:"sent_at"` } type trustSnapshotEntry struct { @@ -463,6 +482,11 @@ func (hm *Manager) saveTrustLocked() { Source: p.Source, }) } + for nodeID, sentAt := range hm.outgoing { + if time.Since(sentAt) < outgoingRequestTTL { + snap.Outgoing = append(snap.Outgoing, outgoingSnapshotEntry{NodeID: nodeID, SentAt: sentAt.Format(time.RFC3339)}) + } + } for nodeID, until := range hm.revoked { // Only persist entries whose cooldown hasn't expired yet. if time.Now().Before(until) { @@ -585,7 +609,18 @@ func (hm *Manager) loadTrust() { hm.revoked[e.NodeID] = until } } - slog.Info("loaded trust state", "peers", len(hm.trusted), "pending", len(hm.pending), "revoked", len(hm.revoked)) + for _, e := range snap.Outgoing { + sent, err := time.Parse(time.RFC3339, e.SentAt) + if err != nil { + slog.Warn("outgoing entry has malformed SentAt — skipping", + "node_id", e.NodeID, "sent_at", e.SentAt, "err", err) + continue + } + if _, trusted := hm.trusted[e.NodeID]; !trusted && time.Since(sent) < outgoingRequestTTL { + hm.outgoing[e.NodeID] = sent + } + } + slog.Info("loaded trust state", "peers", len(hm.trusted), "pending", len(hm.pending), "revoked", len(hm.revoked), "outgoing", len(hm.outgoing)) } // Start binds port 444 and begins handling handshake connections. @@ -917,8 +952,8 @@ func (hm *Manager) reapReplayLocked(now time.Time) { // // outgoing: peers we sent a handshake request to but never heard back from. // A peer that never responds (wrong address, dropped, ignore-listed) would -// otherwise hold a map entry forever. We evict after 2× handshakeMaxAge — -// long enough for any in-flight response to arrive. +// otherwise hold a map entry forever. We evict after outgoingRequestTTL, +// as long as the peer keeps the request pending for a person to answer. // // revoked: peers in post-revocation cooldown. The lazy-delete in handleRequest // / handleRelayRequest only fires on the next *contact from that peer*, which @@ -927,10 +962,9 @@ func (hm *Manager) reapOutgoingAndRevoked() { hm.mu.Lock() defer hm.mu.Unlock() now := time.Now() - outgoingTTL := 2 * handshakeMaxAge for nodeID, sentAt := range hm.outgoing { - if now.Sub(sentAt) > outgoingTTL { - delete(hm.outgoing, nodeID) + if now.Sub(sentAt) > outgoingRequestTTL { + hm.dropOutgoingLocked(nodeID) } } for nodeID, until := range hm.revoked { @@ -940,6 +974,33 @@ func (hm *Manager) reapOutgoingAndRevoked() { } } +// dropOutgoingLocked forgets a request this node sent, and marks the state +// for saving, since outgoing requests are persisted. Caller MUST hold +// hm.mu (write-locked). +func (hm *Manager) dropOutgoingLocked(nodeID uint32) { + if _, ok := hm.outgoing[nodeID]; ok { + delete(hm.outgoing, nodeID) + hm.markDirty() + } +} + +// addOutgoingLocked records a request this node sent, forgetting the oldest +// when the set is full. Caller MUST hold hm.mu (write-locked). +func (hm *Manager) addOutgoingLocked(nodeID uint32, at time.Time) { + if _, ok := hm.outgoing[nodeID]; !ok && len(hm.outgoing) >= maxOutgoingRequests { + var oldest uint32 + var oldestAt time.Time + for id, t := range hm.outgoing { + if oldestAt.IsZero() || t.Before(oldestAt) { + oldest, oldestAt = id, t + } + } + delete(hm.outgoing, oldest) + } + hm.outgoing[nodeID] = at + hm.markDirty() +} + func (hm *Manager) deletePendingLocked(nodeID uint32) { p, ok := hm.pending[nodeID] if !ok { @@ -1027,7 +1088,7 @@ func (hm *Manager) handleRequest(stream coreapi.Stream, msg *HandshakeMsg, regis // cannot claim any NodeID with their own keypair and slip into trust. if _, ok := hm.outgoing[peerNodeID]; ok && registryBound { // Mutual! Auto-approve (registry confirmed pubkey binding) - delete(hm.outgoing, peerNodeID) + hm.dropOutgoingLocked(peerNodeID) hm.markTrustedLocked(peerNodeID, &TrustRecord{ NodeID: peerNodeID, PublicKey: msg.PublicKey, @@ -1179,7 +1240,7 @@ func (hm *Manager) handleAccept(msg *HandshakeMsg) { if until, ok := hm.revoked[peerNodeID]; ok { if time.Now().Before(until) { slog.Info("ignoring handshake acceptance from recently-revoked peer", "peer_node_id", peerNodeID) - delete(hm.outgoing, peerNodeID) + hm.dropOutgoingLocked(peerNodeID) return } delete(hm.revoked, peerNodeID) @@ -1199,7 +1260,7 @@ func (hm *Manager) handleAccept(msg *HandshakeMsg) { if existing, ok := hm.trusted[peerNodeID]; ok { bound := existing.PublicKey if !hm.reconcileTrustBindingLocked(peerNodeID, msg.PublicKey) { - delete(hm.outgoing, peerNodeID) + hm.dropOutgoingLocked(peerNodeID) return } // Preserve an already-bound key when the accept omits one. @@ -1208,7 +1269,7 @@ func (hm *Manager) handleAccept(msg *HandshakeMsg) { } } - delete(hm.outgoing, peerNodeID) + hm.dropOutgoingLocked(peerNodeID) hm.markTrustedLocked(peerNodeID, &TrustRecord{ NodeID: peerNodeID, PublicKey: msg.PublicKey, @@ -1229,7 +1290,7 @@ func (hm *Manager) handleRejectMsg(msg *HandshakeMsg) { slog.Warn("handshake rejected by peer", "peer_node_id", msg.NodeID, "reason", msg.Reason) hm.mu.Lock() - delete(hm.outgoing, msg.NodeID) + hm.dropOutgoingLocked(msg.NodeID) hm.mu.Unlock() } @@ -1257,7 +1318,7 @@ func (hm *Manager) SendRequest(peerNodeID uint32, justification string) error { // themselves locked out of a peer that was auto-pruned by policy // (trust-decay, etc.) with no clean recovery path. delete(hm.revoked, peerNodeID) - hm.outgoing[peerNodeID] = time.Now() + hm.addOutgoingLocked(peerNodeID, time.Now()) hm.mu.Unlock() pubKeyStr := "" @@ -1324,7 +1385,7 @@ func (hm *Manager) processRelayedRequest(fromNodeID uint32, justification string // Check if we have an outgoing request to this peer (mutual handshake) if _, ok := hm.outgoing[fromNodeID]; ok { - delete(hm.outgoing, fromNodeID) + hm.dropOutgoingLocked(fromNodeID) hm.markTrustedLocked(fromNodeID, &TrustRecord{ NodeID: fromNodeID, ApprovedAt: time.Now(), @@ -1468,7 +1529,7 @@ func (hm *Manager) processRelayedApproval(fromNodeID uint32) { if until, ok := hm.revoked[fromNodeID]; ok { if time.Now().Before(until) { slog.Info("ignoring relayed approval for recently-revoked peer", "peer_node_id", fromNodeID) - delete(hm.outgoing, fromNodeID) + hm.dropOutgoingLocked(fromNodeID) return } delete(hm.revoked, fromNodeID) @@ -1479,7 +1540,7 @@ func (hm *Manager) processRelayedApproval(fromNodeID uint32) { return } - delete(hm.outgoing, fromNodeID) + hm.dropOutgoingLocked(fromNodeID) hm.markTrustedLocked(fromNodeID, &TrustRecord{ NodeID: fromNodeID, ApprovedAt: time.Now(), @@ -1555,7 +1616,7 @@ func (hm *Manager) ProcessRelayedRejection(fromNodeID uint32) { // processRelayedRejection handles a handshake rejection received via registry relay. func (hm *Manager) processRelayedRejection(fromNodeID uint32) { hm.mu.Lock() - delete(hm.outgoing, fromNodeID) + hm.dropOutgoingLocked(fromNodeID) hm.mu.Unlock() slog.Info("handshake rejected via relay", "peer_node_id", fromNodeID) } @@ -1654,7 +1715,7 @@ func (hm *Manager) RevokeTrust(peerNodeID uint32) error { _, wasPending := hm.pending[peerNodeID] delete(hm.trusted, peerNodeID) hm.deletePendingLocked(peerNodeID) - delete(hm.outgoing, peerNodeID) + hm.dropOutgoingLocked(peerNodeID) // Block stale relayed approvals still sitting in the registry inbox from // re-establishing trust right after a local revoke. 5-minute cooldown covers // the normal poll cycle many times over. @@ -1719,7 +1780,7 @@ func (hm *Manager) handleRevokeMsg(msg *HandshakeMsg) { _, wasPending := hm.pending[peerNodeID] delete(hm.trusted, peerNodeID) hm.deletePendingLocked(peerNodeID) - delete(hm.outgoing, peerNodeID) + hm.dropOutgoingLocked(peerNodeID) if wasTrusted || wasPending { hm.markDirty() } diff --git a/zz_outgoing_lifetime_test.go b/zz_outgoing_lifetime_test.go new file mode 100644 index 0000000..976e61b --- /dev/null +++ b/zz_outgoing_lifetime_test.go @@ -0,0 +1,126 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +package handshake + +import ( + "path/filepath" + "testing" + "time" +) + +// A person accepts a friend request when they see it, often hours later. +// The acceptance must still complete trust on the requester's side; it used +// to be dropped as unsolicited once the outgoing entry passed 10 minutes. +func TestOutgoing_AcceptanceHoursLaterIsHonoured(t *testing.T) { + t.Parallel() + hm := newTestHM(t, "") + t.Cleanup(hm.Stop) + hm.mu.Lock() + hm.addOutgoingLocked(77, time.Now().Add(-3*time.Hour)) + hm.mu.Unlock() + hm.reapOutgoingAndRevoked() // the reaper must not drop it + + hm.handleAccept(&HandshakeMsg{Type: HandshakeAccept, NodeID: 77, PublicKey: "k", Timestamp: time.Now().Unix()}) + + hm.mu.RLock() + defer hm.mu.RUnlock() + if _, ok := hm.trusted[77]; !ok { + t.Fatal("an acceptance three hours after the request was dropped") + } +} + +// The same, relayed through the registry, which is how most private nodes +// learn of an acceptance. +func TestOutgoing_RelayedApprovalHoursLaterIsHonoured(t *testing.T) { + t.Parallel() + hm := newTestHM(t, "") + t.Cleanup(hm.Stop) + hm.mu.Lock() + hm.addOutgoingLocked(78, time.Now().Add(-5*time.Hour)) + hm.mu.Unlock() + hm.reapOutgoingAndRevoked() + + hm.ProcessRelayedApproval(78) + + hm.mu.RLock() + defer hm.mu.RUnlock() + if _, ok := hm.trusted[78]; !ok { + t.Fatal("a relayed approval five hours after the request was dropped") + } +} + +// A request survives a restart: it is saved with the trust state, and an +// acceptance that arrives after the restart completes trust. +func TestOutgoing_SurvivesRestart(t *testing.T) { + t.Parallel() + path := filepath.Join(t.TempDir(), "trust.json") + before := newTestHM(t, path) + before.mu.Lock() + before.addOutgoingLocked(90, time.Now().Add(-time.Hour)) + before.addOutgoingLocked(91, time.Now().Add(-outgoingRequestTTL-time.Hour)) // too old to keep + before.mu.Unlock() + before.saveTrust() + + after := newTestHM(t, path) + t.Cleanup(after.Stop) + after.loadTrust() + after.mu.RLock() + _, kept := after.outgoing[90] + _, stale := after.outgoing[91] + after.mu.RUnlock() + if !kept || stale { + t.Fatalf("after a restart: request 90 kept=%v (want true), expired 91 kept=%v (want false)", kept, stale) + } + + after.ProcessRelayedApproval(90) + after.mu.RLock() + defer after.mu.RUnlock() + if _, ok := after.trusted[90]; !ok { + t.Fatal("an acceptance after a restart was dropped") + } +} + +// Cancelling a request (revoking) forgets it for good, also across a restart, +// so a late acceptance of a cancelled request is not honoured. +func TestOutgoing_CancelledRequestIsForgottenAcrossRestart(t *testing.T) { + t.Parallel() + path := filepath.Join(t.TempDir(), "trust.json") + before := newTestHM(t, path) + before.mu.Lock() + before.addOutgoingLocked(95, time.Now()) + before.mu.Unlock() + before.saveTrust() + before.RevokeTrust(95) + before.saveTrust() + + after := newTestHM(t, path) + t.Cleanup(after.Stop) + after.loadTrust() + after.mu.RLock() + defer after.mu.RUnlock() + if _, ok := after.outgoing[95]; ok { + t.Fatal("a cancelled request came back after a restart") + } +} + +// The outgoing set is bounded: past maxOutgoingRequests the oldest goes. +func TestOutgoing_Bounded(t *testing.T) { + t.Parallel() + hm := newTestHM(t, "") + t.Cleanup(hm.Stop) + start := time.Now().Add(-time.Hour) + hm.mu.Lock() + defer hm.mu.Unlock() + for i := 0; i <= maxOutgoingRequests; i++ { + hm.addOutgoingLocked(uint32(1000+i), start.Add(time.Duration(i)*time.Millisecond)) + } + if len(hm.outgoing) != maxOutgoingRequests { + t.Fatalf("outgoing holds %d, want %d", len(hm.outgoing), maxOutgoingRequests) + } + if _, ok := hm.outgoing[1000]; ok { + t.Fatal("the oldest request should have been forgotten first") + } + if _, ok := hm.outgoing[uint32(1000+maxOutgoingRequests)]; !ok { + t.Fatal("the newest request is missing") + } +} diff --git a/zz_relayed_more_test.go b/zz_relayed_more_test.go index 96d8d8f..3145647 100644 --- a/zz_relayed_more_test.go +++ b/zz_relayed_more_test.go @@ -173,10 +173,12 @@ func TestReapOutgoingAndRevoked_PrunesExpired(t *testing.T) { t.Parallel() hm := newTestHM(t, "") hm.mu.Lock() - // Old outgoing — should be reaped. - hm.outgoing[10] = time.Now().Add(-2 * time.Hour) + // Past outgoingRequestTTL — should be reaped. + hm.outgoing[10] = time.Now().Add(-outgoingRequestTTL - time.Hour) // Fresh outgoing — should remain. hm.outgoing[20] = time.Now() + // Hours old: a person may still accept it — should remain. + hm.outgoing[30] = time.Now().Add(-2 * time.Hour) hm.mu.Unlock() hm.reapOutgoingAndRevoked() @@ -189,4 +191,7 @@ func TestReapOutgoingAndRevoked_PrunesExpired(t *testing.T) { if _, ok := hm.outgoing[20]; !ok { t.Error("fresh outgoing should remain") } + if _, ok := hm.outgoing[30]; !ok { + t.Error("a request sent two hours ago should still be answerable") + } }