From 2c29758164a2e634f9af0b54ce8ff623bf8aa139 Mon Sep 17 00:00:00 2001 From: Teodor Calin Date: Sat, 25 Jul 2026 17:03:22 +0300 Subject: [PATCH 1/2] beacon: verify punch grants against authoritative registry key + panic-recover hardening Punch-grant verification now resolves the target's pubkey from the registry's authoritative index (via SetAuthoritativeKeyLookup) instead of the Discover-populated cache, and the Discover key binding is first-write-wins, so an unauthenticated Discover can no longer rebind a node's key. Adds panic-recover backstops across the UDP read/dispatch and relay paths, a nil-socket guard, a WSS auth length guard, and fuzz + attack-replay regression coverage. Co-Authored-By: Claude Opus 5 --- recover.go | 53 ++++ server.go | 118 +++++++-- wss/server.go | 7 + wss/zz_fuzz_auth_test.go | 141 ++++++++++ zz_attack_replay_discover_hijack_test.go | 317 +++++++++++++++++++++++ zz_fuzz_dispatch_test.go | 308 ++++++++++++++++++++++ 6 files changed, 928 insertions(+), 16 deletions(-) create mode 100644 recover.go create mode 100644 wss/zz_fuzz_auth_test.go create mode 100644 zz_attack_replay_discover_hijack_test.go create mode 100644 zz_fuzz_dispatch_test.go diff --git a/recover.go b/recover.go new file mode 100644 index 0000000..f218a6b --- /dev/null +++ b/recover.go @@ -0,0 +1,53 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +package beacon + +import ( + "log/slog" + "runtime/debug" + "sync/atomic" +) + +// recoveredPanicCount is a process-global counter incremented every time +// recoverHandler swallows a panic. Exposed via RecoveredPanicCount so the +// dashboard / Prom scrape can alert on a non-zero rate — a panic is still a +// bug and must be visible; it just must not take the process down. The +// beacon runs embedded in the registry binary, so an unrecovered panic on a +// packet-handling goroutine takes the whole trust root with it. +var recoveredPanicCount atomic.Uint64 + +// RecoveredPanicCount returns the total number of panics swallowed by +// recoverHandler since process start. +func RecoveredPanicCount() uint64 { + return recoveredPanicCount.Load() +} + +// recoverHandler is the standard panic-recovery shim placed at the top of +// every packet-handling and background-loop goroutine in this package. +// Mirrors the shim in rendezvous/accept. Usage: +// +// defer recoverHandler("handlePacket") +// +// It must be the OUTERMOST defer in the function: defers run LIFO, so any +// mutex unlock / buffer return registered later still executes before the +// recover observes the panic. +func recoverHandler(label string) { + r := recover() + if r == nil { + return + } + count := recoveredPanicCount.Add(1) + slog.Error("beacon: panic recovered", + "where", label, + "panic", r, + "recovered_total", count, + "stack", string(debug.Stack()), + ) +} + +// safely runs fn under recoverHandler. Used by the periodic background +// loops so one bad tick cannot retire the ticker goroutine. +func safely(label string, fn func()) { + defer recoverHandler(label) + fn() +} diff --git a/server.go b/server.go index 6b5754e..c410147 100644 --- a/server.go +++ b/server.go @@ -109,6 +109,8 @@ type Server struct { // handlePacket the same way UDP datagrams do. wssServer *bwss.Server + authKeyLookup atomic.Pointer[authKeyLookupFn] + done chan struct{} // closed on shutdown // Close idempotency. The actual teardown runs at most once @@ -217,6 +219,16 @@ func NewWithPeers(beaconID uint32, peers []string) *Server { return s } +type authKeyLookupFn func(nodeID uint32) (ed25519.PublicKey, bool) + +func (s *Server) SetAuthoritativeKeyLookup(fn authKeyLookupFn) { + if fn == nil { + s.authKeyLookup.Store(nil) + return + } + s.authKeyLookup.Store(&fn) +} + // EnableCompatWSS attaches a WSS-bridge listener for compat-mode // daemons. After Start, the beacon's relay worker checks the WSS // peer map BEFORE the UDP tier-1/2 lookups: relay packets destined @@ -235,6 +247,11 @@ func (s *Server) EnableCompatWSS(bindAddr string, pubKeyLookup bwss.PubKeyLookup if s.wssServer != nil { return fmt.Errorf("beacon: compat WSS already enabled") } + if pubKeyLookup != nil { + s.SetAuthoritativeKeyLookup(func(id uint32) (ed25519.PublicKey, bool) { + return pubKeyLookup(id) + }) + } ws, err := bwss.New(bwss.Config{ BindAddr: bindAddr, PubKeyLookup: pubKeyLookup, @@ -454,18 +471,40 @@ const readBatchCap = 32 // allocation contention. On non-Linux platforms the x/net/ipv4 // fallback degrades to per-message ReadFrom, preserving correctness. func (s *Server) readLoop(conn *net.UDPConn) error { + // A panic on the receive path must not silently retire this socket's + // reader: recover, log, and resume the batch loop. Only a closed + // socket ends it. The per-packet boundary lives in handlePacket; this + // one covers the batch bookkeeping around it. + for { + err, done := s.readLoopBatches(conn) + if done { + return err + } + select { + case <-s.done: + return nil + default: + } + } +} + +// readLoopBatches runs the recvmmsg loop until the socket closes (done=true) +// or a panic unwinds it (done=false, caller resumes). +func (s *Server) readLoopBatches(conn *net.UDPConn) (err error, done bool) { + defer recoverHandler("readLoop") + pc := ipv4.NewPacketConn(conn) msgs := make([]ipv4.Message, readBatchCap) for i := range msgs { msgs[i].Buffers = [][]byte{make([]byte, 65535)} } for { - n, err := pc.ReadBatch(msgs, 0) - if err != nil { - if opErr, ok := err.(*net.OpError); ok && opErr.Err.Error() == "use of closed network connection" { - return nil + n, rerr := pc.ReadBatch(msgs, 0) + if rerr != nil { + if opErr, ok := rerr.(*net.OpError); ok && opErr.Err.Error() == "use of closed network connection" { + return nil, true } - slog.Debug("beacon read batch error", "err", err) + slog.Debug("beacon read batch error", "err", rerr) continue } for i := 0; i < n; i++ { @@ -530,6 +569,11 @@ func (s *Server) RelayDropped() uint64 { return s.relayDropped.Load() } func (s *Server) RelayNotFound() uint64 { return s.relayNotFound.Load() } func (s *Server) handlePacket(data []byte, remote *net.UDPAddr) { + // Outermost defer: a panic anywhere in the dispatch below drops this + // one datagram and leaves the read loop running. Every inbound packet + // on this path is unauthenticated remote input. + defer recoverHandler("handlePacket") + // Empty-frame guard. The UDP readLoop screens len(data)<1, but the // WSS OnFrame path only enforces an upper bound — a 0-byte binary // frame from any authenticated WSS peer used to crash the beacon @@ -579,7 +623,7 @@ func (s *Server) handleDiscover(data []byte, remote *net.UDPAddr, wantDest bool) if len(data) >= 4+ed25519.PublicKeySize { pub := make(ed25519.PublicKey, ed25519.PublicKeySize) copy(pub, data[4:4+ed25519.PublicKeySize]) - s.nodePubKeys.Store(nodeID, pub) + s.nodePubKeys.LoadOrStore(nodeID, pub) } // Per-nodeID endpoint update rate limit (PILOT-334) — prevents a single @@ -617,6 +661,12 @@ func (s *Server) handleDiscover(data []byte, remote *net.UDPAddr, wantDest bool) copy(reply[2:2+len(ip)], ip) binary.BigEndian.PutUint16(reply[2+len(ip):], uint16(remote.Port)) + // The compat WSS bridge starts accepting frames in EnableCompatWSS, + // which runs before ListenAndServe binds the UDP sockets — a frame + // arriving in that window reaches here with no socket to reply on. + if s.conn == nil { + return + } if _, err := s.conn.WriteToUDP(reply, remote); err != nil { slog.Debug("beacon discover reply failed", "node_id", nodeID, "err", err) } @@ -634,11 +684,7 @@ func (s *Server) verifyPunchGrant(trailer []byte, requesterID, targetID uint32) return false } sig := trailer[punchGrantExpirySize : punchGrantExpirySize+ed25519.SignatureSize] - v, ok := s.nodePubKeys.Load(targetID) - if !ok { - return false - } - pub, ok := v.(ed25519.PublicKey) + pub, ok := s.targetPubKey(targetID) if !ok || len(pub) != ed25519.PublicKeySize { return false } @@ -646,6 +692,21 @@ func (s *Server) verifyPunchGrant(trailer []byte, requesterID, targetID uint32) return ed25519.Verify(pub, []byte(preimage), sig) } +func (s *Server) targetPubKey(targetID uint32) (ed25519.PublicKey, bool) { + if p := s.authKeyLookup.Load(); p != nil { + return (*p)(targetID) + } + v, ok := s.nodePubKeys.Load(targetID) + if !ok { + return nil, false + } + pub, ok := v.(ed25519.PublicKey) + if !ok { + return nil, false + } + return pub, true +} + func (s *Server) handlePunchRequest(data []byte, remote *net.UDPAddr) { if len(data) < 8 { return @@ -897,6 +958,26 @@ const relayFlushAfter = 2 * time.Millisecond // With per-fd workers, sends parallelise across fds and the only // serialisation left is per-worker (its own batch state). func (s *Server) relayWorker(sendConn *ipv4.PacketConn, rawConn *net.UDPConn) { + // A panic while shaping one relay job must not permanently retire this + // worker — that would cut the beacon's drain capacity by 1/N for the + // life of the process. Resume with fresh batch state instead. + for { + if s.relayWorkerLoop(sendConn, rawConn) { + return + } + select { + case <-s.done: + return + default: + } + } +} + +// relayWorkerLoop is the relayWorker body. Returns true when the server is +// shutting down, false when a panic unwound it and the caller should resume. +func (s *Server) relayWorkerLoop(sendConn *ipv4.PacketConn, rawConn *net.UDPConn) (shutdown bool) { + defer recoverHandler("relayWorker") + msgs := make([]ipv4.Message, 0, relayBatchCap) payloadsToReturn := make([][]byte, 0, relayBatchCap) outBufsToReturn := make([]*[]byte, 0, relayBatchCap) @@ -983,7 +1064,7 @@ func (s *Server) relayWorker(sendConn *ipv4.PacketConn, rawConn *net.UDPConn) { select { case <-s.done: flush() - return + return true case <-timer.C: timerActive = false flush() @@ -1138,6 +1219,11 @@ func (s *Server) relayStatsLoop() { // SendPunchCommand tells a node to send UDP to a target endpoint. func (s *Server) SendPunchCommand(nodeID uint32, targetIP net.IP, targetPort uint16) error { + // See handleDiscover: reachable via the compat WSS bridge before + // ListenAndServe has bound the UDP sockets. + if s.conn == nil { + return fmt.Errorf("beacon: not listening") + } // Snapshot returns the address under RLock so we don't race with // a concurrent Upsert mutating beaconNode.addr — see // nodes_shard.go Get() caveat (race detector flag, 2026-05-19). @@ -1177,7 +1263,7 @@ func (s *Server) reapLoop() { for { select { case <-ticker.C: - s.reapStaleNodes() + safely("reapStaleNodes", s.reapStaleNodes) case <-s.done: return } @@ -1238,7 +1324,7 @@ func (s *Server) gossipLoop() { for { select { case <-ticker.C: - s.sendGossip() + safely("sendGossip", s.sendGossip) case <-s.done: return } @@ -1420,11 +1506,11 @@ func (s *Server) registryDiscoveryLoop() { defer ticker.Stop() // Run immediately, then on tick - s.registryDiscover() + safely("registryDiscover", s.registryDiscover) for { select { case <-ticker.C: - s.registryDiscover() + safely("registryDiscover", s.registryDiscover) case <-s.done: return } diff --git a/wss/server.go b/wss/server.go index 1ea402a..6464f75 100644 --- a/wss/server.go +++ b/wss/server.go @@ -419,6 +419,13 @@ func (s *Server) runAuth(ctx context.Context, conn *websocket.Conn) (uint32, err if !ok { return 0, fmt.Errorf("unknown node_id %d", reply.NodeID) } + // ed25519.Verify panics on a key that is not exactly PublicKeySize + // bytes. PubKeyLookup is supplied by the embedding process (in + // production, the registry's pubkey index) so its output is not + // guaranteed here; reject rather than pass it through. + if len(pubKey) != ed25519.PublicKeySize { + return 0, fmt.Errorf("node_id %d has malformed public key (%d bytes)", reply.NodeID, len(pubKey)) + } sig, err := base64.StdEncoding.DecodeString(reply.Sig) if err != nil { return 0, fmt.Errorf("bad sig encoding: %w", err) diff --git a/wss/zz_fuzz_auth_test.go b/wss/zz_fuzz_auth_test.go new file mode 100644 index 0000000..819ea46 --- /dev/null +++ b/wss/zz_fuzz_auth_test.go @@ -0,0 +1,141 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +package wss_test + +import ( + "context" + "crypto/ed25519" + "encoding/base64" + "encoding/json" + "testing" + "time" + + cw "github.com/coder/websocket" + + "github.com/pilot-protocol/beacon/wss" +) + +// The compat-mode WSS bridge terminates connections from anywhere the +// operator's reverse proxy accepts, and the auth reply is pure +// attacker-controlled JSON. The reply's node_id selects which key +// PubKeyLookup returns, and that key goes to ed25519.Verify — which panics +// on anything that is not exactly PublicKeySize bytes. PubKeyLookup is +// supplied by the embedding process (the registry, in production), so the +// bridge cannot assume its output is well-formed. +// +// Node IDs 1..5 below map to the malformed key shapes; node 9 maps to a +// real key so the well-formed path stays covered. +func fuzzAuthServer(tb testing.TB) (*wss.Server, string) { + tb.Helper() + + realPub, _, err := ed25519.GenerateKey(nil) + if err != nil { + tb.Fatalf("keygen: %v", err) + } + keys := map[uint32]ed25519.PublicKey{ + 1: nil, + 2: {}, + 3: make([]byte, 1), + 4: make([]byte, ed25519.PublicKeySize-1), + 5: make([]byte, ed25519.PublicKeySize+1), + 9: realPub, + } + + s, err := wss.New(wss.Config{ + BindAddr: "127.0.0.1:0", + AuthTimeout: 2 * time.Second, + IdleTimeout: 2 * time.Second, + PubKeyLookup: func(id uint32) (ed25519.PublicKey, bool) { k, ok := keys[id]; return k, ok }, + OnFrame: func(uint32, []byte) {}, + }) + if err != nil { + tb.Fatalf("wss.New: %v", err) + } + if err := s.Start(); err != nil { + tb.Fatalf("wss.Start: %v", err) + } + tb.Cleanup(func() { _ = s.Close() }) + return s, "ws://" + s.Addr() + "/v1/compat" +} + +// FuzzWSSAuthReply drives arbitrary bytes into the post-upgrade auth reply. +// A malformed reply must close the connection with an auth failure, never +// panic the handler. +func FuzzWSSAuthReply(f *testing.F) { + sig := base64.StdEncoding.EncodeToString(make([]byte, ed25519.SignatureSize)) + + add := func(v map[string]interface{}) { + if b, err := json.Marshal(v); err == nil { + f.Add(b) + } + } + // One reply per malformed-key node id, plus the well-formed one. + for _, id := range []int{1, 2, 3, 4, 5, 9, 0, 77} { + add(map[string]interface{}{ + "type": "auth_reply", "node_id": id, + "public_key": base64.StdEncoding.EncodeToString(make([]byte, ed25519.PublicKeySize)), + "sig": sig, + }) + } + // Wrong-length and non-base64 signatures against a malformed-key node. + for _, n := range []int{0, 1, 63, 65, 128} { + add(map[string]interface{}{ + "type": "auth_reply", "node_id": 1, + "sig": base64.StdEncoding.EncodeToString(make([]byte, n)), + }) + } + add(map[string]interface{}{"type": "auth_reply", "node_id": 1, "sig": "@@@not base64@@@"}) + add(map[string]interface{}{"type": "wrong_type", "node_id": 1, "sig": sig}) + add(map[string]interface{}{"node_id": 1}) + f.Add([]byte("")) + f.Add([]byte("{")) + f.Add([]byte("null")) + f.Add([]byte(`{"node_id":99999999999999999999}`)) + + srv, url := fuzzAuthServer(f) + + f.Fuzz(func(t *testing.T, reply []byte) { + if len(reply) > 32*1024 { + reply = reply[:32*1024] + } + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + + before := srv.Metrics() + + conn, _, err := cw.Dial(ctx, url, &cw.DialOptions{Subprotocols: []string{"pilot.v1"}}) + if err != nil { + // Server may be shedding connections; not a finding. + return + } + defer conn.Close(cw.StatusNormalClosure, "") + + // Drain the server's challenge, then send the fuzzed reply. + if _, _, err := conn.Read(ctx); err != nil { + return + } + if err := conn.Write(ctx, cw.MessageText, reply); err != nil { + return + } + // Read the outcome and let the server settle. + _, _, _ = conn.Read(ctx) + + // Oracle: runAuth is called from an http.Handler, so net/http + // recovers any panic it raises and the connection just dies — no + // crash, no test signal. But handleUpgrade only reaches + // authOK.Add / authFail.Add if runAuth *returned*. So a reply that + // moves neither counter is one that unwound the handler, which is + // exactly the panic we are hunting. + deadline := time.Now().Add(2 * time.Second) + for { + after := srv.Metrics() + if after.AuthOK != before.AuthOK || after.AuthFail != before.AuthFail { + return + } + if time.Now().After(deadline) { + t.Fatalf("auth neither succeeded nor failed for reply %q — handler unwound", reply) + } + time.Sleep(2 * time.Millisecond) + } + }) +} diff --git a/zz_attack_replay_discover_hijack_test.go b/zz_attack_replay_discover_hijack_test.go new file mode 100644 index 0000000..9ece42c --- /dev/null +++ b/zz_attack_replay_discover_hijack_test.go @@ -0,0 +1,317 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +package beacon + +import ( + "crypto/ed25519" + "crypto/rand" + "encoding/binary" + "net" + "testing" + "time" + + "github.com/pilot-protocol/common/protocol" +) + +// Adversarial replay for the forged-Discover endpoint-hijack finding. +// +// BeaconMsgDiscover is unauthenticated by design (it is the STUN-style +// reflexive-address probe), and handleDiscover does two things with the +// attacker-supplied node ID: it upserts nodeID -> observed remote addr, +// and — if the payload carries 32 trailing bytes — it stores nodeID -> +// public key. Neither write proves the sender owns the node ID. +// +// Everything below runs against a local in-process beacon bound to +// 127.0.0.1 on an ephemeral port. Nothing here touches production. + +// sendDiscover writes a raw Discover for an arbitrary node ID from the +// caller's socket. Unlike registerNode it does not assume the beacon +// will treat the sender as the owner of that ID — that is the point. +func sendDiscover(t *testing.T, conn *net.UDPConn, nodeID uint32, pub ed25519.PublicKey) { + t.Helper() + msg := make([]byte, 5+len(pub)) + msg[0] = protocol.BeaconMsgDiscover + binary.BigEndian.PutUint32(msg[1:5], nodeID) + copy(msg[5:], pub) + if _, err := conn.Write(msg); err != nil { + t.Fatalf("send discover: %v", err) + } + buf := make([]byte, 64) + conn.SetReadDeadline(time.Now().Add(2 * time.Second)) + if _, err := conn.Read(buf); err != nil { + t.Fatalf("read discover reply: %v", err) + } +} + +func startLocalBeacon(t *testing.T, requirePunchToken bool) (*Server, *net.UDPAddr) { + t.Helper() + s := New() + s.SetRequirePunchToken(requirePunchToken) + go s.ListenAndServe("127.0.0.1:0") + <-s.Ready() + t.Cleanup(func() { s.Close() }) + return s, beaconUDPAddr(t, s) +} + +func mappedAddr(t *testing.T, s *Server, nodeID uint32) *net.UDPAddr { + t.Helper() + addr, ok := s.nodes.Snapshot(nodeID) + if !ok { + return nil + } + return addr +} + +// TestAttackReplay_DiscoverPreemptiveNodeIDClaim proves the base +// primitive: any UDP source can claim any unregistered node ID and the +// beacon maps that ID to the attacker's endpoint. This is the forged +// Discover from the pentest, replayed against a local beacon. +// +// The flag argument matters: SetRequirePunchToken gates handlePunchRequest +// only, so the claim lands identically with the flag off and on. Both are +// asserted so the test documents exactly what WS3 does not cover. +func TestAttackReplay_DiscoverPreemptiveNodeIDClaim(t *testing.T) { + t.Parallel() + + for _, tc := range []struct { + name string + requirePunchToken bool + }{ + {"punch_token_off", false}, + {"punch_token_on", true}, + } { + tc := tc + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + s, beaconAddr := startLocalBeacon(t, tc.requirePunchToken) + + const victimID = uint32(770001) + + attacker, err := net.DialUDP("udp", nil, beaconAddr) + if err != nil { + t.Fatalf("dial beacon: %v", err) + } + defer attacker.Close() + attackerAddr := attacker.LocalAddr().(*net.UDPAddr) + + sendDiscover(t, attacker, victimID, nil) + + got := mappedAddr(t, s, victimID) + if got == nil { + t.Fatalf("victim node ID was never mapped — test setup is wrong") + } + if got.Port == attackerAddr.Port { + t.Logf("HIJACK LANDED (require_punch_token=%v): node %d maps to attacker %v", + tc.requirePunchToken, victimID, got) + } else { + t.Fatalf("unexpected mapping for node %d: %v (attacker was %v)", victimID, got, attackerAddr) + } + }) + } +} + +// TestAttackReplay_DiscoverRateLimitBlocksImmediateTakeover is the +// honest counterpart: once a node ID has an endpoint, the per-nodeID +// discover rate limit (discoverMinInterval, 30s) suppresses the Upsert, +// so an attacker cannot instantly steal a *live* mapping. The limiter is +// a throttle, not an authentication check — it delays the takeover by +// one interval, it does not prevent it — but the immediate steal is +// blocked and that is worth pinning. +func TestAttackReplay_DiscoverRateLimitBlocksImmediateTakeover(t *testing.T) { + t.Parallel() + s, beaconAddr := startLocalBeacon(t, false) + + const victimID = uint32(770002) + + victim := registerNode(t, beaconAddr, victimID) + defer victim.Close() + victimAddr := victim.LocalAddr().(*net.UDPAddr) + + attacker, err := net.DialUDP("udp", nil, beaconAddr) + if err != nil { + t.Fatalf("dial beacon: %v", err) + } + defer attacker.Close() + attackerAddr := attacker.LocalAddr().(*net.UDPAddr) + + sendDiscover(t, attacker, victimID, nil) + + got := mappedAddr(t, s, victimID) + if got == nil { + t.Fatal("victim mapping vanished") + } + if got.Port == attackerAddr.Port { + t.Fatalf("ATTACK SUCCEEDED: live endpoint for node %d stolen inside the rate-limit window (now %v)", victimID, got) + } + if got.Port != victimAddr.Port { + t.Fatalf("unexpected mapping for node %d: %v (victim was %v)", victimID, got, victimAddr) + } + t.Logf("immediate takeover throttled by discoverMinInterval=%v; node %d still maps to the victim %v", + discoverMinInterval, victimID, got) +} + +// TestAttackReplay_DiscoverPubKeyOverwrite pins the fix for the +// key-rebind primitive: the Discover key binding is first-write-wins, so +// a later forged Discover claiming an existing node ID with a different +// key is dropped and the beacon keeps the key it first saw. +func TestAttackReplay_DiscoverPubKeyOverwrite(t *testing.T) { + t.Parallel() + s, beaconAddr := startLocalBeacon(t, true) + + const victimID = uint32(770003) + + victimPub, _, err := ed25519.GenerateKey(rand.Reader) + if err != nil { + t.Fatalf("generate victim key: %v", err) + } + victim := registerNodeWithPubKey(t, beaconAddr, victimID, victimPub) + defer victim.Close() + + attackerPub, _, err := ed25519.GenerateKey(rand.Reader) + if err != nil { + t.Fatalf("generate attacker key: %v", err) + } + + attacker, err := net.DialUDP("udp", nil, beaconAddr) + if err != nil { + t.Fatalf("dial beacon: %v", err) + } + defer attacker.Close() + + sendDiscover(t, attacker, victimID, attackerPub) + + v, ok := s.nodePubKeys.Load(victimID) + if !ok { + t.Fatal("victim public key vanished") + } + stored := v.(ed25519.PublicKey) + if stored.Equal(attackerPub) { + t.Fatalf("ATTACK SUCCEEDED: node %d's beacon-held key was rebound to the attacker", victimID) + } + if !stored.Equal(victimPub) { + t.Fatalf("unexpected stored key for node %d", victimID) + } + t.Logf("rebind blocked: node %d's beacon-held key still belongs to the victim", victimID) +} + +// TestAttackReplay_PunchTokenBypassViaPubKeyRebind chains the two +// primitives end to end against a beacon with require_punch_token=true: +// +// 1. victim registers with its real ed25519 key +// 2. attacker attempts to rebind the beacon's copy of that key via a +// forged Discover (now dropped by the first-write-wins guard) +// 3. attacker mints a punch grant with their own private key +// 4. beacon verifies the grant against the victim's real key and rejects +// +// The beacon must emit no punch command: the WS3 punch-token control +// holds against a host that can send UDP to the beacon. +func TestAttackReplay_PunchTokenBypassViaPubKeyRebind(t *testing.T) { + t.Parallel() + _, beaconAddr := startLocalBeacon(t, true) + + const victimID = uint32(770004) + const attackerID = uint32(770005) + + victimPub, _, err := ed25519.GenerateKey(rand.Reader) + if err != nil { + t.Fatalf("generate victim key: %v", err) + } + victimConn := registerNodeWithPubKey(t, beaconAddr, victimID, victimPub) + defer victimConn.Close() + victimAddr := victimConn.LocalAddr().(*net.UDPAddr) + + attackerPub, attackerPriv, err := ed25519.GenerateKey(rand.Reader) + if err != nil { + t.Fatalf("generate attacker key: %v", err) + } + attackerConn := registerNode(t, beaconAddr, attackerID) + defer attackerConn.Close() + attackerAddr := attackerConn.LocalAddr().(*net.UDPAddr) + + // Control, run twice with the same pacing the real attempt uses, so + // a rejection here cannot be blamed on the punch rate limiter: + // without the rebind, the grant the attacker can produce is rejected. + for i := 0; i < 2; i++ { + if i > 0 { + time.Sleep(punchPerSourceInterval + 200*time.Millisecond) + } + expiry := time.Now().Add(60 * time.Second).Unix() + forged := mintPunchGrant(t, attackerPriv, attackerID, victimID, expiry) + sendPunchRequest(t, attackerConn, attackerID, victimID, forged) + expectNoPacket(t, attackerConn, 500*time.Millisecond) + expectNoPacket(t, victimConn, 500*time.Millisecond) + } + + // Step 2: forged Discover rebinding the victim's beacon-held key. + sendDiscover(t, attackerConn, victimID, attackerPub) + + // The per-source punch rate limiter allows one punch per second. + time.Sleep(punchPerSourceInterval + 200*time.Millisecond) + + // Steps 3+4: same forged grant, now verifiable against the rebound key. + expiry := time.Now().Add(60 * time.Second).Unix() + forged := mintPunchGrant(t, attackerPriv, attackerID, victimID, expiry) + sendPunchRequest(t, attackerConn, attackerID, victimID, forged) + + buf := make([]byte, 64) + attackerConn.SetReadDeadline(time.Now().Add(2 * time.Second)) + n, err := attackerConn.Read(buf) + if err != nil { + t.Logf("no punch command reached the attacker — the token gate held (victim %v, attacker %v)", victimAddr, attackerAddr) + return + } + if n >= 1 && buf[0] == protocol.BeaconMsgPunchCommand { + t.Fatalf("ATTACK SUCCEEDED: require_punch_token=true yet the beacon coordinated a punch for an attacker holding no victim key") + } + t.Fatalf("unexpected reply type 0x%02x", buf[0]) +} + +// TestAttackReplay_DiscoverCannotRedirectToThirdParty is the bound on +// the finding, and the reason this is an interception primitive rather +// than a UDP reflector: handleDiscover upserts the *observed* source +// address, never an address chosen inside the payload. An attacker +// therefore cannot point a node ID at an unrelated victim host. +func TestAttackReplay_DiscoverCannotRedirectToThirdParty(t *testing.T) { + t.Parallel() + s, beaconAddr := startLocalBeacon(t, false) + + const nodeID = uint32(770006) + + // A third-party socket the attacker would love the beacon to + // believe owns nodeID. + thirdParty, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.ParseIP("127.0.0.1")}) + if err != nil { + t.Fatalf("listen third party: %v", err) + } + defer thirdParty.Close() + thirdPartyAddr := thirdParty.LocalAddr().(*net.UDPAddr) + + attacker, err := net.DialUDP("udp", nil, beaconAddr) + if err != nil { + t.Fatalf("dial beacon: %v", err) + } + defer attacker.Close() + attackerAddr := attacker.LocalAddr().(*net.UDPAddr) + + // Append a would-be address hint after the node ID; the wire format + // has no such field, and the parser must ignore it entirely. + msg := make([]byte, 5+6) + msg[0] = protocol.BeaconMsgDiscover + binary.BigEndian.PutUint32(msg[1:5], nodeID) + copy(msg[5:9], thirdPartyAddr.IP.To4()) + binary.BigEndian.PutUint16(msg[9:11], uint16(thirdPartyAddr.Port)) + if _, err := attacker.Write(msg); err != nil { + t.Fatalf("send discover: %v", err) + } + + if !waitUntil(2*time.Second, func() bool { return mappedAddr(t, s, nodeID) != nil }) { + t.Fatal("node never registered") + } + got := mappedAddr(t, s, nodeID) + if got.Port == thirdPartyAddr.Port { + t.Fatalf("ATTACK SUCCEEDED: beacon accepted a payload-supplied endpoint %v — this is a reflector", got) + } + if got.Port != attackerAddr.Port { + t.Fatalf("unexpected mapping: %v", got) + } +} diff --git a/zz_fuzz_dispatch_test.go b/zz_fuzz_dispatch_test.go new file mode 100644 index 0000000..9cce866 --- /dev/null +++ b/zz_fuzz_dispatch_test.go @@ -0,0 +1,308 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +package beacon + +import ( + "crypto/ed25519" + "encoding/binary" + "net" + "testing" + "time" + + "github.com/pilot-protocol/common/protocol" +) + +// The sibling zz_fuzz_handle_packet_test.go target drives the beacon over a +// real UDP socket. That exercises the full receive path but caps throughput +// at UDP I/O speed and, more importantly, cannot attribute a panic to the +// input that caused it — the panic unwinds the server's own read goroutine, +// not the fuzz goroutine. The targets below call the dispatch functions +// directly on the fuzzer's goroutine so a crash is both fast to find and +// reproducible from the recorded corpus entry. +// +// They deliberately bypass Server.handlePacket's own recover shim where the +// sub-handler is called directly, so a missing bounds check surfaces as a +// test failure rather than being swallowed. + +// fuzzServer builds a Server with a real bound socket but no read loop, so +// the fuzzer owns dispatch scheduling. Binding matters: the discover and +// punch handlers write replies through s.conn, and a nil socket would take +// those code paths out of reach. +func fuzzServer(tb testing.TB) *Server { + tb.Helper() + s := NewWithPeers(1, nil) + c, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1), Port: 0}) + if err != nil { + tb.Fatalf("bind: %v", err) + } + s.conn = c + s.conns = []*net.UDPConn{c} + tb.Cleanup(func() { + close(s.done) + c.Close() + }) + return s +} + +// fuzzAddr is the synthetic remote used for every iteration. A stable +// address keeps the per-source rate limiters in a steady state rather than +// growing an entry per iteration. +var fuzzAddr = &net.UDPAddr{IP: net.IPv4(198, 51, 100, 7), Port: 41234} + +func seedBeaconCorpus(f *testing.F) { + f.Helper() + + // Well-formed message of each type. + discover := make([]byte, 5) + discover[0] = protocol.BeaconMsgDiscover + binary.BigEndian.PutUint32(discover[1:], 42) + f.Add(discover) + + discoverEx := make([]byte, 5+ed25519.PublicKeySize) + discoverEx[0] = protocol.BeaconMsgDiscoverEx + binary.BigEndian.PutUint32(discoverEx[1:5], 42) + f.Add(discoverEx) + + punch := make([]byte, 9) + punch[0] = protocol.BeaconMsgPunchRequest + binary.BigEndian.PutUint32(punch[1:5], 100) + binary.BigEndian.PutUint32(punch[5:9], 200) + f.Add(punch) + + relay := make([]byte, 9+16) + relay[0] = protocol.BeaconMsgRelay + binary.BigEndian.PutUint32(relay[1:5], 1) + binary.BigEndian.PutUint32(relay[5:9], 2) + f.Add(relay) + + sync := make([]byte, 1+4+2+8) + sync[0] = protocol.BeaconMsgSync + binary.BigEndian.PutUint32(sync[1:5], 99) + binary.BigEndian.PutUint16(sync[5:7], 2) + f.Add(sync) + + // Adversarial shapes: empty, single byte per type, one byte under every + // length guard, and counters that claim far more body than is present. + f.Add([]byte{}) + for _, t := range []byte{ + protocol.BeaconMsgDiscover, + protocol.BeaconMsgDiscoverEx, + protocol.BeaconMsgPunchRequest, + protocol.BeaconMsgRelay, + protocol.BeaconMsgSync, + 0x00, 0x0F, 0xFF, + } { + f.Add([]byte{t}) + f.Add([]byte{t, 0x00}) + f.Add([]byte{t, 0xFF, 0xFF, 0xFF}) + f.Add([]byte{t, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF}) + } + + // Sync claiming the maximum node count with an empty body — the + // length-validation branch. + truncSync := make([]byte, 1+4+2) + truncSync[0] = protocol.BeaconMsgSync + binary.BigEndian.PutUint16(truncSync[5:7], 0xFFFF) + f.Add(truncSync) + + // Sync claiming a count one node larger than the body carries. + offByOne := make([]byte, 1+4+2+4) + offByOne[0] = protocol.BeaconMsgSync + binary.BigEndian.PutUint16(offByOne[5:7], 2) + f.Add(offByOne) + + // Relay with a payload right at the maxRelayPayload boundary is too + // large to keep in the corpus; use a small one plus a header-only frame. + f.Add([]byte{protocol.BeaconMsgRelay, 0, 0, 0, 1, 0, 0, 0, 2}) + + // Punch request carrying a truncated grant trailer (expiry + signature). + shortGrant := make([]byte, 9+punchGrantTrailerSize-1) + shortGrant[0] = protocol.BeaconMsgPunchRequest + binary.BigEndian.PutUint32(shortGrant[1:5], 100) + binary.BigEndian.PutUint32(shortGrant[5:9], 200) + f.Add(shortGrant) +} + +// FuzzBeaconDispatch drives arbitrary bytes through the full type-dispatch +// switch — the exact bytes an unauthenticated UDP source (or a compat-mode +// WSS peer, via EnableCompatWSS's OnFrame) can put on the wire. +func FuzzBeaconDispatch(f *testing.F) { + seedBeaconCorpus(f) + s := fuzzServer(f) + + f.Fuzz(func(t *testing.T, data []byte) { + if len(data) > maxRelayPayload { + data = data[:maxRelayPayload] + } + // handlePacket recovers internally, so a panic would otherwise be + // swallowed. Compare the counter across the call to surface it. + before := RecoveredPanicCount() + s.handlePacket(data, fuzzAddr) + if after := RecoveredPanicCount(); after != before { + t.Fatalf("handlePacket recovered a panic on input %x", data) + } + }) +} + +// FuzzBeaconHandleDiscover targets the discover parser directly (no recover +// shim in the way). Covers the variable-length pubkey trailer, the +// per-nodeID endpoint rate limiter, and the reply encoder. +func FuzzBeaconHandleDiscover(f *testing.F) { + f.Add([]byte{}) + f.Add([]byte{0x00, 0x00, 0x00}) + f.Add(make([]byte, 4)) + f.Add(make([]byte, 4+ed25519.PublicKeySize)) + f.Add(make([]byte, 4+ed25519.PublicKeySize-1)) + f.Add(make([]byte, 4+ed25519.PublicKeySize+1)) + + s := fuzzServer(f) + + f.Fuzz(func(t *testing.T, data []byte) { + if len(data) > 4096 { + data = data[:4096] + } + // wantDest toggles the dstNodes bookkeeping branch. + s.handleDiscover(data, fuzzAddr, false) + s.handleDiscover(data, fuzzAddr, true) + // An IPv6 remote drives the 16-byte reply-encoding branch. + s.handleDiscover(data, &net.UDPAddr{IP: net.ParseIP("2001:db8::1"), Port: 5}, false) + }) +} + +// FuzzBeaconHandleSync targets the gossip parser. The claimed node count is +// a wire-controlled uint16 used to size an allocation and to index into the +// body — the classic length-prefix mismatch shape. +func FuzzBeaconHandleSync(f *testing.F) { + f.Add([]byte{}) + f.Add(make([]byte, 5)) + f.Add(make([]byte, 6)) + { + b := make([]byte, 6) + binary.BigEndian.PutUint16(b[4:6], 0xFFFF) + f.Add(b) + } + { + b := make([]byte, 6+4*3) + binary.BigEndian.PutUint16(b[4:6], 3) + f.Add(b) + } + { + // Count larger than body by exactly one node. + b := make([]byte, 6+4*2) + binary.BigEndian.PutUint16(b[4:6], 3) + f.Add(b) + } + + s := fuzzServer(f) + + f.Fuzz(func(t *testing.T, data []byte) { + if len(data) > 65535 { + data = data[:65535] + } + s.handleSync(data, fuzzAddr) + }) +} + +// FuzzBeaconDispatchRelay targets the relay header parser and the pooled +// payload copy. The destination is pre-registered so the fuzzer gets past +// the not-found pre-check and into the enqueue path. +func FuzzBeaconDispatchRelay(f *testing.F) { + f.Add([]byte{}) + f.Add(make([]byte, 7)) + f.Add(make([]byte, 8)) + { + b := make([]byte, 8+32) + binary.BigEndian.PutUint32(b[0:4], 1) + binary.BigEndian.PutUint32(b[4:8], 7777) + f.Add(b) + } + + s := fuzzServer(f) + // Register the destination so the tier-1 pre-check passes. + s.nodes.Upsert(7777, fuzzAddr, time.Now(), maxBeaconNodes) + + f.Fuzz(func(t *testing.T, data []byte) { + if len(data) > maxRelayPayload { + data = data[:maxRelayPayload] + } + s.dispatchRelay(data) + // Drain so the buffered channel cannot fill across iterations. + for { + select { + case job := <-s.relayCh: + s.returnPayload(job.payload) + continue + default: + } + break + } + }) +} + +// FuzzBeaconPunchGrant targets the punch-grant trailer parser with the +// token requirement enabled. The signed-grant path slices a fixed-size +// expiry + signature out of caller-supplied bytes and hands the signature +// to ed25519.Verify against a key pulled from the nodePubKeys map, so it +// combines a bounds check with a key-length-sensitive verify. +func FuzzBeaconPunchGrant(f *testing.F) { + f.Add([]byte{}) + f.Add(make([]byte, 8)) + f.Add(make([]byte, punchGrantTrailerSize-1)) + f.Add(make([]byte, punchGrantTrailerSize)) + f.Add(make([]byte, punchGrantTrailerSize+64)) + { + // Non-expired grant so the ed25519 verify is actually reached. + b := make([]byte, punchGrantTrailerSize) + binary.BigEndian.PutUint64(b[0:8], uint64(time.Now().Add(time.Hour).Unix())) + f.Add(b) + } + + s := fuzzServer(f) + s.SetRequirePunchToken(true) + + pub, _, err := ed25519.GenerateKey(nil) + if err != nil { + f.Fatalf("keygen: %v", err) + } + // A well-formed key for one target and a deliberately wrong-length key + // for another: the second entry is the shape that made crypto.Verify + // panic before it grew a length guard. + s.nodePubKeys.Store(uint32(200), pub) + s.nodePubKeys.Store(uint32(201), ed25519.PublicKey(make([]byte, 5))) + s.nodePubKeys.Store(uint32(202), ed25519.PublicKey(nil)) + + f.Fuzz(func(t *testing.T, trailer []byte) { + if len(trailer) > 4096 { + trailer = trailer[:4096] + } + for _, target := range []uint32{200, 201, 202, 203} { + _ = s.verifyPunchGrant(trailer, 100, target) + } + }) +} + +// FuzzBeaconHandlePunchRequest drives the full punch-request handler, +// including the rate-limit bypass label and the grant trailer. +func FuzzBeaconHandlePunchRequest(f *testing.F) { + f.Add([]byte{}) + f.Add(make([]byte, 7)) + f.Add(make([]byte, 8)) + f.Add(make([]byte, 8+punchGrantTrailerSize)) + + s := fuzzServer(f) + // Wildcard whitelist so the global 10/s cap does not short-circuit + // almost every iteration before the parser runs. + s.SetPunchWhitelist([]string{"*"}) + s.nodes.Upsert(100, fuzzAddr, time.Now(), maxBeaconNodes) + s.nodes.Upsert(200, fuzzAddr, time.Now(), maxBeaconNodes) + + f.Fuzz(func(t *testing.T, data []byte) { + if len(data) > 4096 { + data = data[:4096] + } + for _, requireToken := range []bool{false, true} { + s.SetRequirePunchToken(requireToken) + s.handlePunchRequest(data, fuzzAddr) + } + }) +} From 85c80d57d3ac20c471bc615383462b4831d21268 Mon Sep 17 00:00:00 2001 From: Teodor Calin Date: Sat, 25 Jul 2026 17:11:45 +0300 Subject: [PATCH 2/2] test: widen UDP/WSS read deadlines for loaded CI runners The beacon UDP round-trip tests used 2-3s read deadlines and the compat WSS bridge a 30s accept budget; both flake on loaded public CI runners (pre-existing, observed on main). Widen the UDP reads to 10s and the WSS accept wait to 60s. A genuinely dropped reply still fails, just later. Co-Authored-By: Claude Opus 5 --- server_test.go | 2 +- zz_attack_replay_discover_hijack_test.go | 4 ++-- zz_compat_wss_test.go | 2 +- zz_punch_token_test.go | 4 ++-- 4 files changed, 6 insertions(+), 6 deletions(-) diff --git a/server_test.go b/server_test.go index d3b93c5..2875cb2 100644 --- a/server_test.go +++ b/server_test.go @@ -30,7 +30,7 @@ func registerNode(t *testing.T, beaconAddr *net.UDPAddr, nodeID uint32) *net.UDP // Read discover reply buf := make([]byte, 64) - conn.SetReadDeadline(time.Now().Add(2 * time.Second)) + conn.SetReadDeadline(time.Now().Add(10 * time.Second)) n, err := conn.Read(buf) if err != nil { t.Fatalf("read discover reply: %v", err) diff --git a/zz_attack_replay_discover_hijack_test.go b/zz_attack_replay_discover_hijack_test.go index 9ece42c..78b794e 100644 --- a/zz_attack_replay_discover_hijack_test.go +++ b/zz_attack_replay_discover_hijack_test.go @@ -37,7 +37,7 @@ func sendDiscover(t *testing.T, conn *net.UDPConn, nodeID uint32, pub ed25519.Pu t.Fatalf("send discover: %v", err) } buf := make([]byte, 64) - conn.SetReadDeadline(time.Now().Add(2 * time.Second)) + conn.SetReadDeadline(time.Now().Add(10 * time.Second)) if _, err := conn.Read(buf); err != nil { t.Fatalf("read discover reply: %v", err) } @@ -254,7 +254,7 @@ func TestAttackReplay_PunchTokenBypassViaPubKeyRebind(t *testing.T) { sendPunchRequest(t, attackerConn, attackerID, victimID, forged) buf := make([]byte, 64) - attackerConn.SetReadDeadline(time.Now().Add(2 * time.Second)) + attackerConn.SetReadDeadline(time.Now().Add(10 * time.Second)) n, err := attackerConn.Read(buf) if err != nil { t.Logf("no punch command reached the attacker — the token gate held (victim %v, attacker %v)", victimAddr, attackerAddr) diff --git a/zz_compat_wss_test.go b/zz_compat_wss_test.go index 94e9f28..475e86e 100644 --- a/zz_compat_wss_test.go +++ b/zz_compat_wss_test.go @@ -161,7 +161,7 @@ func TestEnableCompatWSS_HappyPath(t *testing.T) { // CI runners have been observed taking 13+ seconds before the // http.Server.Serve goroutine reaches Accept. 30s gives margin // without making the happy path slow (still ~10ms in practice). - if !waitUntil(30*time.Second, func() bool { + if !waitUntil(60*time.Second, func() bool { _, err := net.DialTimeout("tcp", wsAddr, 1*time.Second) return err == nil }) { diff --git a/zz_punch_token_test.go b/zz_punch_token_test.go index 9836748..1b1b240 100644 --- a/zz_punch_token_test.go +++ b/zz_punch_token_test.go @@ -31,7 +31,7 @@ func registerNodeWithPubKey(t *testing.T, beaconAddr *net.UDPAddr, nodeID uint32 } buf := make([]byte, 64) - conn.SetReadDeadline(time.Now().Add(2 * time.Second)) + conn.SetReadDeadline(time.Now().Add(10 * time.Second)) n, err := conn.Read(buf) if err != nil { t.Fatalf("read discover reply: %v", err) @@ -68,7 +68,7 @@ func sendPunchRequest(t *testing.T, conn *net.UDPConn, requesterID, targetID uin func expectPunchCommand(t *testing.T, conn *net.UDPConn, wantAddr *net.UDPAddr) { t.Helper() buf := make([]byte, 64) - conn.SetReadDeadline(time.Now().Add(3 * time.Second)) + conn.SetReadDeadline(time.Now().Add(10 * time.Second)) n, err := conn.Read(buf) if err != nil { t.Fatalf("read punch command: %v", err)