geoprobe: reject forged, replayed and spoofed geolocation input

CHANGELOG.md

@@ -8,6 +8,12 @@ All notable changes to this project will be documented in this file.
 
 ### Changes
 
+- Geolocation
+  - geoprobe-target drops LocationOffsets whose signature chain does not verify instead of caching them and writing them to `location_offsets`, which the lake explorer publishes unfiltered. Verification itself now rejects the zero authority pubkey, since `ed25519.Verify` accepts the all-zero (pubkey, signature) pair for roughly one message in four — the zero key decodes to an order-4 point, not to nothing — so an unsigned datagram would otherwise verify. The same guard is applied to the signed TWAMP `ProbePacket.Verify` and `ReplyPacket.Verify`, the other two places a wire-supplied pubkey is used as the verification key.
+  - The geoprobe agent enforces RFC-16's replay mitigation: an inbound DZD offset is rejected unless its `MeasurementSlot` is within 15 minutes behind or 5 minutes ahead of the current ledger slot, which covers the 5-minute slot caches on both ends. An equal-RTT offset also no longer replaces the cache's best entry, because replacing it reset its expiry clock and let a replay pin the probe's reference point indefinitely.
+  - The signed TWAMP reflector verifies a probe's signature before touching per-sender pair state. `target_pk` is public onchain, so spoofed probes could previously consume a paying sender's pair budget, repoint its source IP and clear its challenge nonce. Unverified probes still get a reply, now off throwaway state and capped at one per window per pubkey — with a floor so the cap holds even where pair rate limiting is disabled. Ingestion also refuses a cached ledger slot older than two refresh periods, so an RPC outage cannot freeze the replay window around a stale slot.
+  - A completed target scan that matches nothing propagates instead of being mistaken for a skipped scan, so removing a user's last target or flipping them to Delinquent stops the probing.
+  - ICMP echo replies are matched on source address as well as ID and sequence.
 - CLI
   - `doublezero-solana shreds payments` asks the Solana node for version 1 and reads the JsonParsed instruction list, so a v1 fund no longer fails the listing with error -32015.
 - CI

controlplane/telemetry/cmd/geoprobe-agent/main.go

@@ -43,6 +43,28 @@ const (
 	discoveryInterval                 = 60 * time.Second
 	defaultDeliveryDNSRefreshInterval = 5 * time.Minute
 	defaultDeliveryDNSTTL             = defaultDeliveryDNSRefreshInterval * 5 / 2
+
+	// dzSlotDuration is the nominal DoubleZero Ledger slot time.
+	dzSlotDuration = 400 * time.Millisecond
+
+	// Acceptance window for an inbound offset's MeasurementSlot, RFC-16's replay
+	// mitigation. A DZD stamps offsets with a slot it caches for
+	// geoprobe.SlotCacheTTL (5m) and this agent compares against its own slot
+	// cache with the same TTL, so a legitimate offset can sit ~5m either side of
+	// our view before any RPC or finalization lag. maxOffsetSlotLag allows 15m of
+	// that skew in the past; maxOffsetSlotLead allows 5m in the future, for when
+	// our own cached slot is the stale one.
+	maxOffsetSlotLag  = uint64(15 * time.Minute / dzSlotDuration)
+	maxOffsetSlotLead = uint64(5 * time.Minute / dzSlotDuration)
+
+	// maxSlotReferenceAge bounds how stale the cached ledger slot may be before
+	// it stops counting as "now" for the replay check. getCurrentSlot falls back
+	// to its cache indefinitely when RPC fails, so without this bound an outage
+	// freezes the acceptance window around an old slot: replays near that slot
+	// stay acceptable for as long as the outage lasts, and genuinely fresh
+	// offsets eventually fall outside maxOffsetSlotLead. Two refresh periods, so
+	// a single missed refresh does not stop ingestion.
+	maxSlotReferenceAge = 2 * geoprobe.SlotCacheTTL
 )
 
 var (
@@ -148,8 +170,10 @@ func (c *offsetCache) Put(offset *geoprobe.LocationOffset) {
 		return
 	}
 
-	if offset.RttNs <= sender.best.offset.RttNs {
-		// New offset is better than or equal to best: replace best.
+	if offset.RttNs < sender.best.offset.RttNs {
+		// Strictly better than best: replace it. An equal-RTT offset must not
+		// replace best, because replacing also resets best's receivedAt clock —
+		// a replayed offset would otherwise hold best forever.
 		sender.best = entry
 	} else {
 		// New offset has higher RTT than best, consider it for second-best.
@@ -217,6 +241,24 @@ func (c *offsetCache) Evict() int {
 	return evicted
 }
 
+// slotReference returns slot only while the value cached at cachedAt is recent
+// enough to serve as the "now" the replay check measures against.
+func slotReference(slot uint64, cachedAt, now time.Time) (uint64, error) {
+	if age := now.Sub(cachedAt); age > maxSlotReferenceAge {
+		return 0, fmt.Errorf("cached slot is %s old, exceeds %s", age.Round(time.Second), maxSlotReferenceAge)
+	}
+	return slot, nil
+}
+
+// offsetSlotFresh reports whether an offset's MeasurementSlot falls inside the
+// acceptance window around the current ledger slot.
+func offsetSlotFresh(measurementSlot, currentSlot uint64) bool {
+	if measurementSlot > currentSlot {
+		return measurementSlot-currentSlot <= maxOffsetSlotLead
+	}
+	return currentSlot-measurementSlot <= maxOffsetSlotLag
+}
+
 func marshalBestOffset(cache *offsetCache) [][]byte {
 	best := cache.GetBest()
 	if best == nil {
@@ -470,6 +512,21 @@ func main() {
 		return slot, nil
 	}
 
+	// Offset ingestion needs a slot it can still treat as "now", which the
+	// stale-cache fallback above does not guarantee. Composite offsets keep
+	// using getCurrentSlot: stamping a slightly stale slot is better than the
+	// probe emitting nothing during an RPC blip.
+	getSlotReference := func(ctx context.Context) (uint64, error) {
+		slot, err := getCurrentSlot(ctx)
+		if err != nil {
+			return 0, err
+		}
+		slotMu.RLock()
+		cachedAt := slotCachedAt
+		slotMu.RUnlock()
+		return slotReference(slot, cachedAt, time.Now())
+	}
+
 	// Set up UDP sender for composite offsets.
 	senderConn, err := geoprobe.NewUDPConn()
 	if err != nil {
@@ -496,7 +553,7 @@ func main() {
 
 	// Run UDP offset listener.
 	go func() {
-		runOffsetListener(ctx, log, offsetListener, cache, pState, signedReflector, m)
+		runOffsetListener(ctx, log, offsetListener, cache, pState, signedReflector, m, getSlotReference)
 	}()
 
 	// Run eviction goroutine.
@@ -658,6 +715,7 @@ func runOffsetListener(
 	parents *parentState,
 	signedReflector signed.Reflector,
 	m *geoprobe.Metrics,
+	getCurrentSlot func(ctx context.Context) (uint64, error),
 ) {
 	log.Info("Starting offset listener", "addr", conn.LocalAddr().String())
 
@@ -718,6 +776,30 @@ func runOffsetListener(
 
 		log.Debug("signature verification successful", "authority_pubkey", authorityPK)
 
+		// RFC-16 replay mitigation. A signature stays valid forever, and the
+		// only other freshness bound is receipt wall-clock, which a replay
+		// refreshes by definition — so MeasurementSlot has to be checked against
+		// the current ledger slot or a captured offset can be replayed
+		// indefinitely to pin this probe's attested reference point.
+		currentSlot, err := getCurrentSlot(ctx)
+		if err != nil {
+			log.Warn("Rejecting offset, current slot unavailable",
+				"sender_pubkey", senderPK, "addr", addr, "error", err)
+			m.OffsetsRejected.WithLabelValues(geoprobe.RejectSlotUnavailable).Inc()
+			continue
+		}
+		if !offsetSlotFresh(offset.MeasurementSlot, currentSlot) {
+			log.Warn("Rejecting offset with measurement slot outside acceptance window",
+				"sender_pubkey", senderPK,
+				"addr", addr,
+				"measurement_slot", offset.MeasurementSlot,
+				"current_slot", currentSlot,
+				"max_lag_slots", maxOffsetSlotLag,
+				"max_lead_slots", maxOffsetSlotLead)
+			m.OffsetsRejected.WithLabelValues(geoprobe.RejectSlotOutOfWindow).Inc()
+			continue
+		}
+
 		cache.Put(offset)
 		signedReflector.SetOffsets(marshalBestOffset(cache))
 		m.OffsetsReceived.Inc()

controlplane/telemetry/cmd/geoprobe-agent/main_test.go

@@ -1,11 +1,17 @@
 package main
 
 import (
+	"context"
+	"io"
+	"log/slog"
+	"net"
 	"sync"
 	"testing"
 	"time"
 
+	"github.com/gagliardetto/solana-go"
 	"github.com/malbeclabs/doublezero/controlplane/telemetry/internal/geoprobe"
+	"github.com/prometheus/client_golang/prometheus"
 )
 
 func makeTestOffset(senderPubkey [32]byte, rttNs uint64) *geoprobe.LocationOffset {
@@ -352,3 +358,173 @@ func TestOffsetCache_ConcurrentAccess(t *testing.T) {
 	}
 	wg.Wait()
 }
+
+func TestOffsetSlotFresh(t *testing.T) {
+	const current = uint64(1_000_000)
+
+	tests := []struct {
+		name            string
+		measurementSlot uint64
+		want            bool
+	}{
+		{"same slot", current, true},
+		{"lagging within window", current - maxOffsetSlotLag/2, true},
+		{"at lag boundary", current - maxOffsetSlotLag, true},
+		{"replay past window", current - maxOffsetSlotLag - 1, false},
+		{"leading within window", current + maxOffsetSlotLead, true},
+		{"leading past window", current + maxOffsetSlotLead + 1, false},
+		{"zero slot", 0, false},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := offsetSlotFresh(tt.measurementSlot, current); got != tt.want {
+				t.Errorf("offsetSlotFresh(%d, %d) = %v, want %v", tt.measurementSlot, current, got, tt.want)
+			}
+		})
+	}
+}
+
+// A replayed offset carries the RTT it was captured with. Letting an equal-RTT
+// offset replace best also resets best's expiry clock, which is how a replay
+// pins the probe's attested reference point forever.
+func TestOffsetCache_PutEqualRTTDoesNotRefreshBest(t *testing.T) {
+	cache := newOffsetCache(1 * time.Hour)
+	pubkey := [32]byte{1}
+
+	cache.Put(makeTestOffset(pubkey, 1000))
+	firstReceivedAt := cache.entries[pubkey].best.receivedAt
+
+	time.Sleep(2 * time.Millisecond)
+	cache.Put(makeTestOffset(pubkey, 1000)) // byte-identical replay
+
+	if got := cache.entries[pubkey].best.receivedAt; !got.Equal(firstReceivedAt) {
+		t.Errorf("equal-RTT offset refreshed best's clock (%v -> %v); a replay could hold best forever",
+			firstReceivedAt, got)
+	}
+}
+
+// stubSignedReflector satisfies signed.Reflector for listener tests.
+type stubSignedReflector struct{}
+
+func (s *stubSignedReflector) Run(context.Context) error     { return nil }
+func (s *stubSignedReflector) Close() error                  { return nil }
+func (s *stubSignedReflector) Port() uint16                  { return 0 }
+func (s *stubSignedReflector) SetAuthorizedKeys([][32]byte)  {}
+func (s *stubSignedReflector) SetOffsets([][]byte)           {}
+func (s *stubSignedReflector) SetLogger(logger *slog.Logger) {}
+
+func TestRunOffsetListener_RejectsOffsetOutsideSlotWindow(t *testing.T) {
+	const currentSlot = uint64(1_000_000)
+
+	listener, err := geoprobe.NewUDPListener(0)
+	if err != nil {
+		t.Fatalf("failed to create listener: %v", err)
+	}
+	defer listener.Close()
+	listenAddr := listener.LocalAddr().(*net.UDPAddr)
+
+	device := solana.NewWallet()
+	authority := solana.NewWallet()
+	signer, err := geoprobe.NewOffsetSigner(authority.PrivateKey, device.PublicKey())
+	if err != nil {
+		t.Fatalf("failed to create signer: %v", err)
+	}
+
+	var devicePK, authorityPK [32]byte
+	copy(devicePK[:], device.PublicKey().Bytes())
+	copy(authorityPK[:], authority.PublicKey().Bytes())
+
+	cache := newOffsetCache(1 * time.Hour)
+	parents := &parentState{authorities: map[[32]byte][32]byte{devicePK: authorityPK}}
+	metrics := geoprobe.NewMetrics("test", device.PublicKey().String(), prometheus.NewRegistry())
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		runOffsetListener(ctx, slog.New(slog.NewTextHandler(io.Discard, nil)), listener, cache, parents,
+			&stubSignedReflector{}, metrics, func(context.Context) (uint64, error) { return currentSlot, nil })
+	}()
+	defer func() {
+		cancel()
+		<-done
+	}()
+
+	sender, err := net.ListenUDP("udp4", &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1)})
+	if err != nil {
+		t.Fatalf("failed to create sender socket: %v", err)
+	}
+	defer sender.Close()
+
+	sendOffset := func(slot uint64) {
+		t.Helper()
+		offset := makeTestOffset(devicePK, 1000)
+		offset.Version = geoprobe.LocationOffsetVersion
+		offset.MeasurementSlot = slot
+		if err := signer.SignOffset(offset); err != nil {
+			t.Fatalf("failed to sign offset: %v", err)
+		}
+		if err := geoprobe.SendOffset(sender, &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1), Port: listenAddr.Port}, offset); err != nil {
+			t.Fatalf("failed to send offset: %v", err)
+		}
+	}
+
+	// A signed offset stamped with a slot outside the acceptance window is a
+	// replay: it must never reach the cache.
+	sendOffset(currentSlot - maxOffsetSlotLag - 1)
+	time.Sleep(500 * time.Millisecond)
+	if got := cache.Get(devicePK); got != nil {
+		t.Fatalf("expected replayed offset to be rejected, got %+v", got)
+	}
+
+	// A fresh offset over the same path must still be accepted, proving the
+	// rejection above came from the slot check and not from plumbing.
+	sendOffset(currentSlot)
+	deadline := time.Now().Add(2 * time.Second)
+	for cache.Get(devicePK) == nil {
+		if time.Now().After(deadline) {
+			t.Fatal("expected fresh offset to be cached")
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+}
+
+// getCurrentSlot serves its cache indefinitely when ledger RPC fails. A frozen
+// slot would freeze the replay window with it, so ingestion refuses a reference
+// that is too old to mean "now".
+func TestSlotReference(t *testing.T) {
+	now := time.Now()
+
+	tests := []struct {
+		name     string
+		cachedAt time.Time
+		wantErr  bool
+	}{
+		{"just fetched", now, false},
+		{"one refresh period old", now.Add(-geoprobe.SlotCacheTTL), false},
+		{"at the age bound", now.Add(-maxSlotReferenceAge), false},
+		{"past the age bound", now.Add(-maxSlotReferenceAge - time.Second), true},
+		{"never fetched", time.Time{}, true},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			slot, err := slotReference(1_000_000, tt.cachedAt, now)
+			if tt.wantErr {
+				if err == nil {
+					t.Fatalf("expected a stale-reference error, got slot %d", slot)
+				}
+				return
+			}
+			if err != nil {
+				t.Fatalf("unexpected error: %v", err)
+			}
+			if slot != 1_000_000 {
+				t.Errorf("expected slot 1000000, got %d", slot)
+			}
+		})
+	}
+}

controlplane/telemetry/cmd/geoprobe-target/main.go

@@ -335,6 +335,19 @@ func handleOffset(log *slog.Logger, offset *geoprobe.LocationOffset, addr *net.U
 		log.Debug("signature verification complete", "authority_pubkey", solana.PublicKeyFromBytes(offset.AuthorityPubkey[:]).String(), "valid", signatureValid)
 	}
 
+	// Until the chain verifies, a LocationOffset is just an attacker-chosen UDP
+	// datagram. Drop it instead of recording it: every row written to
+	// location_offsets is aggregated by the public lake explorer without
+	// filtering on signature_valid, so persisting forgeries would publish them.
+	if !signatureValid {
+		log.Warn("dropping offset with invalid signature chain",
+			"from", addr,
+			"authority_pubkey", solana.PublicKeyFromBytes(offset.AuthorityPubkey[:]).String(),
+			"sender_pubkey", solana.PublicKeyFromBytes(offset.SenderPubkey[:]).String(),
+			"error", verifyError)
+		return
+	}
+
 	if chWriter != nil {
 		rawBytes, err := offset.Marshal()
 		if err != nil {