diff --git a/CHANGELOG.md b/CHANGELOG.md index 0eb230e1cb..331074fee7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,9 @@ All notable changes to this project will be documented in this file. ### Changes - Geolocation + - geoprobe-target bounds offset replay with a per-sender `MeasurementSlot` floor (#4286). A signature stays valid forever, so an offset captured off the wire could be replayed into `location_offsets` — which the lake explorer publishes — indefinitely. `MeasurementSlot` is inside the signed payload, so the highest slot a signing key has proven is a lower bound on real time that a replay can repeat but cannot advance without the private key; offsets more than 2 minutes below that floor are dropped, as are repeats once the floor has stood still for 30 minutes. Floors are keyed by `AuthorityPubkey`, the key the signature is verified against, not by the signed-but-unauthenticated `SenderPubkey` — otherwise one minted keypair stamping a real geoprobe's `SenderPubkey` with a huge slot would lock that geoprobe out of the table. The floor is derived from the offset stream rather than a ledger clock because geoprobe-target holds no RPC connection by design. In steady state a sender's floor advances every ~5 minutes, so the replay window is about 7 minutes; the 30-minute stall bound only governs the degraded case where the sending probe is riding out its own RPC outage on a frozen cached slot. Rejections log at Warn with a machine-readable `reason`, the authority and sender pubkeys, the offered slot, the floor and the floor's age — geoprobe-target exposes no metrics, so these lines are the only signal that a sender stopped being ingested. + - `signature_valid` no longer claims a verification that did not run. With `-verify-signatures=false` (not the deployed configuration) rows were written asserting `signature_valid = true` despite nothing being checked; they now carry `false` with `signature_error = "signature verification disabled"`. + - `MinCache.Update` no longer replaces `best` on an equal RTT, matching the geoprobe agent's own cache. Replacing also reset `best`'s expiry clock, which let a repeated measurement hold `best` past its TTL. - 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. diff --git a/CLAUDE.md b/CLAUDE.md index e8a7dd7a93..4a09aecb0d 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -29,6 +29,7 @@ DoubleZero is a protocol for building and operating high-performance, permission - **telemetry-agent** runs on DZDs (Arista devices). It measures geoprobes via TWAMP and sends them location offsets. DZDs are the roots of trust with known coordinates. - **geoprobe-agent** runs on geoprobe servers (Ubuntu bare metal or VPS), NOT on DZDs. It receives TWAMP probes from DZDs, caches their location offsets, and measures outbound targets. It currently requires no special privileges (all UDP, unprivileged ports). - These are two separate binaries with opposite roles. Never conflate them: DZDs measure geoprobes; geoprobes measure targets. +- **geoprobe-target holds no ledger connection, and this is an invariant, not just the current state.** RFC-16 makes it example target software: the signature chain is what lets an arbitrary host verify a measurement with no trust relationship and no chain access, so requiring every target host to reach a DZ ledger RPC endpoint would invert the design. Anything that needs a time reference there — the offset replay bound, for instance — must derive it from the signed offset stream, not from a slot lookup. ## Build Commands diff --git a/controlplane/telemetry/cmd/geoprobe-target/main.go b/controlplane/telemetry/cmd/geoprobe-target/main.go index 9acee2fb28..34362cb58a 100644 --- a/controlplane/telemetry/cmd/geoprobe-target/main.go +++ b/controlplane/telemetry/cmd/geoprobe-target/main.go @@ -30,8 +30,52 @@ const ( nanosecondsPerMs = 1000000.0 rateLimitCleanupInterval = 5 * time.Minute rateLimitEntryTTL = 10 * time.Minute + + // dzSlotDuration is the nominal DoubleZero Ledger slot time. + dzSlotDuration = 400 * time.Millisecond + + // RFC-16's replay bound, derived from the offset stream instead of a ledger + // clock: geoprobe-target holds no RPC connection by design, so the only + // time reference it has is the highest MeasurementSlot a signing key has + // proven. A replay can repeat that floor but cannot push it forward without + // the corresponding private key. + // + // maxSlotRegression is how far below the floor an offer may sit and still be + // accepted. A probe re-reads the slot from a load-balanced RPC pool, so a + // lagging finalized replica can hand it a slot slightly behind the last one; + // without slack that would silently stop the sender's ingestion. + maxSlotRegression = uint64(2 * time.Minute / dzSlotDuration) + + // maxFloorStall is how long the floor may stand still before repeats stop + // counting as live. A healthy sender stamps from geoprobe.SlotCacheTTL, so + // the floor advances every ~5m and the steady-state replay window is ~7m. + // The stall bound only governs the degraded case where the sending probe + // rides out its own RPC outage on a frozen cached slot: six refresh periods + // keeps ingesting its genuine measurements, because a dropped measurement + // is lost permanently rather than deferred. + maxFloorStall = 6 * geoprobe.SlotCacheTTL + + // floorEntryTTL is how long a silent sender's floor is remembered. Only + // total silence expires it — any offer, accepted or rejected, keeps it + // alive — so this bounds the map, not the replay window. Kept off + // -max-offset-age, which tunes the display cache and would otherwise let a + // cache setting shorten a security bound. + floorEntryTTL = 2 * maxFloorStall +) + +// Machine-readable rejection reasons, logged as the "reason" field so an +// operator can alert on a sender's measurements going missing. geoprobe-target +// exposes no prometheus metrics, so these log lines are the only signal. +const ( + rejectSlotRegressed = "slot_regressed" + rejectFloorStalled = "floor_stalled" ) +// signatureUnverifiedMarker records in signature_error that no verification ran +// at all, distinguishing a -verify-signatures=false row from a historical row +// whose verification genuinely failed. +const signatureUnverifiedMarker = "signature verification disabled" + var ( twampPort = flag.Uint("twamp-port", defaultTWAMPPort, "Port to listen for TWAMP probes") udpPort = flag.Uint("udp-port", defaultUDPPort, "Port to listen for LocationOffset UDP datagrams") @@ -75,6 +119,9 @@ func main() { "rate_limit", *rateLimit, "max_reference_depth", maxReferenceDepth, "max_offset_age", *maxOffsetAge, + "max_slot_regression", maxSlotRegression, + "max_floor_stall", maxFloorStall, + "floor_entry_ttl", floorEntryTTL, ) // Keyed by SenderPubkey (geoprobe identity). Each geoprobe is an independent @@ -101,10 +148,11 @@ func main() { if *rateLimit > 0 { go limiter.cleanup(ctx) } - go sweepCaches(ctx, caches) + floor := newSlotFloor(floorEntryTTL) + go sweepCaches(ctx, caches, floor) go runTWAMPReflector(ctx, log, *twampPort, errCh) - go runUDPListener(ctx, log, *udpPort, *verifySignature, limiter, chWriter, caches, errCh) + go runUDPListener(ctx, log, *udpPort, *verifySignature, limiter, chWriter, caches, floor, errCh) select { case err := <-errCh: @@ -214,7 +262,92 @@ func (rl *rateLimiter) cleanup(ctx context.Context) { } } -func sweepCaches(ctx context.Context, caches *geoprobe.MinCacheMap[[32]byte, geoprobe.LocationOffset]) { +type floorEntry struct { + slot uint64 + advancedAt time.Time + lastSeen time.Time +} + +// slotFloor bounds offset replay without a ledger clock. MeasurementSlot is +// inside the signed payload, so the highest slot a key has proven is a lower +// bound on real time that a replay can repeat but cannot advance. +// +// Floors are keyed by AuthorityPubkey, the key VerifyOffsetChain actually +// checks the signature against — not by SenderPubkey, which is signed but +// unauthenticated. Anyone can mint a keypair and stamp a real geoprobe's +// SenderPubkey, so a SenderPubkey-keyed floor would let one datagram carrying +// a huge slot lock that geoprobe out of the table permanently. In production a +// probe signs its own offsets, so the two keys move together for real senders. +type slotFloor struct { + mu sync.Mutex + entries map[[32]byte]*floorEntry + ttl time.Duration + nowFunc func() time.Time // for testing; defaults to time.Now +} + +func newSlotFloor(ttl time.Duration) *slotFloor { + return &slotFloor{ + entries: make(map[[32]byte]*floorEntry), + ttl: ttl, + nowFunc: time.Now, + } +} + +// accept reports whether an offer may be ingested, advancing the signing key's +// floor when it does. On rejection it returns a reason token plus the floor +// state, for the log line. +// +// Callers verify the signature chain first, so in the deployed configuration +// only a proven slot moves a floor. With -verify-signatures=false nothing is +// checked and the floor is fed unverified slots along with everything else. +// +// A never-seen key seeds its floor from its own first offer, so one stale +// capture is accepted per key per process restart; the live stream raises the +// floor past it within minutes. A rejected offer still refreshes lastSeen, so a +// sustained replay cannot outlive the entry and reseed from itself. +func (f *slotFloor) accept(authority [32]byte, slot uint64) (ok bool, reason string, floorSlot uint64, floorAge time.Duration) { + f.mu.Lock() + defer f.mu.Unlock() + + now := f.nowFunc() + entry, exists := f.entries[authority] + if !exists { + f.entries[authority] = &floorEntry{slot: slot, advancedAt: now, lastSeen: now} + return true, "", slot, 0 + } + entry.lastSeen = now + + age := now.Sub(entry.advancedAt) + switch { + case slot > entry.slot: + entry.slot = slot + entry.advancedAt = now + age = 0 + case entry.slot-slot > maxSlotRegression: + return false, rejectSlotRegressed, entry.slot, age + case age > maxFloorStall: + // The floor has not moved in maxFloorStall, so repeats no longer + // evidence a live sender. Only a strictly higher slot does. + return false, rejectFloorStalled, entry.slot, age + } + + return true, "", entry.slot, age +} + +// sweep drops keys silent for ttl, bounding the map the same way the offset +// caches are bounded. +func (f *slotFloor) sweep() { + f.mu.Lock() + defer f.mu.Unlock() + now := f.nowFunc() + for k, entry := range f.entries { + if now.Sub(entry.lastSeen) > f.ttl { + delete(f.entries, k) + } + } +} + +func sweepCaches(ctx context.Context, caches *geoprobe.MinCacheMap[[32]byte, geoprobe.LocationOffset], floor *slotFloor) { ticker := time.NewTicker(rateLimitCleanupInterval) defer ticker.Stop() for { @@ -223,6 +356,7 @@ func sweepCaches(ctx context.Context, caches *geoprobe.MinCacheMap[[32]byte, geo return case <-ticker.C: caches.Sweep() + floor.sweep() } } } @@ -245,7 +379,7 @@ func runTWAMPReflector(ctx context.Context, log *slog.Logger, port uint, errCh c } } -func runUDPListener(ctx context.Context, log *slog.Logger, port uint, verifySignatures bool, limiter *rateLimiter, chWriter *geoprobe.ClickhouseWriter, caches *geoprobe.MinCacheMap[[32]byte, geoprobe.LocationOffset], errCh chan<- error) { +func runUDPListener(ctx context.Context, log *slog.Logger, port uint, verifySignatures bool, limiter *rateLimiter, chWriter *geoprobe.ClickhouseWriter, caches *geoprobe.MinCacheMap[[32]byte, geoprobe.LocationOffset], floor *slotFloor, errCh chan<- error) { conn, err := geoprobe.NewUDPListener(int(port)) if err != nil { errCh <- fmt.Errorf("failed to create UDP listener: %w", err) @@ -307,7 +441,7 @@ func runUDPListener(ctx context.Context, log *slog.Logger, port uint, verifySign continue } - handleOffset(log, offset, addr, verifySignatures, chWriter, caches) + handleOffset(log, offset, addr, verifySignatures, chWriter, caches, floor) } } @@ -325,26 +459,45 @@ func countReferenceDepth(offset *geoprobe.LocationOffset) int { return maxDepth + 1 } -func handleOffset(log *slog.Logger, offset *geoprobe.LocationOffset, addr *net.UDPAddr, verifySignatures bool, chWriter *geoprobe.ClickhouseWriter, caches *geoprobe.MinCacheMap[[32]byte, geoprobe.LocationOffset]) { - signatureValid := true - var verifyError error +func handleOffset(log *slog.Logger, offset *geoprobe.LocationOffset, addr *net.UDPAddr, verifySignatures bool, chWriter *geoprobe.ClickhouseWriter, caches *geoprobe.MinCacheMap[[32]byte, geoprobe.LocationOffset], floor *slotFloor) { + // signature_valid is an assertion the row carries into the public + // location_offsets table, so it may only be true when a check actually ran. + // With -verify-signatures=false nothing is checked: record false and say + // why, rather than asserting a verification that did not happen. + signatureValid := false + signatureError := "" if verifySignatures { - verifyError = geoprobe.VerifyOffsetChain(offset) - signatureValid = verifyError == nil - 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 err := geoprobe.VerifyOffsetChain(offset); err != nil { + 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", err) + return + } + signatureValid = true + log.Debug("signature verification complete", "authority_pubkey", solana.PublicKeyFromBytes(offset.AuthorityPubkey[:]).String(), "valid", true) + } else { + signatureError = signatureUnverifiedMarker } - // 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", + // A valid signature never expires, so a captured offset stays verifiable + // forever. The slot floor is what stops it being replayed into the table. + if ok, reason, floorSlot, floorAge := floor.accept(offset.AuthorityPubkey, offset.MeasurementSlot); !ok { + log.Warn("dropping offset outside slot floor", + "reason", reason, "from", addr, "authority_pubkey", solana.PublicKeyFromBytes(offset.AuthorityPubkey[:]).String(), "sender_pubkey", solana.PublicKeyFromBytes(offset.SenderPubkey[:]).String(), - "error", verifyError) + "offset_slot", offset.MeasurementSlot, + "floor_slot", floorSlot, + "floor_age_seconds", floorAge.Seconds()) return } @@ -353,11 +506,7 @@ func handleOffset(log *slog.Logger, offset *geoprobe.LocationOffset, addr *net.U if err != nil { log.Error("failed to marshal offset for clickhouse", "error", err) } else { - sigErrStr := "" - if verifyError != nil { - sigErrStr = verifyError.Error() - } - row := geoprobe.OffsetRowFromLocationOffset(offset, addr.String(), signatureValid, sigErrStr, rawBytes) + row := geoprobe.OffsetRowFromLocationOffset(offset, addr.String(), signatureValid, signatureError, rawBytes) chWriter.Record(row) } } @@ -365,7 +514,7 @@ func handleOffset(log *slog.Logger, offset *geoprobe.LocationOffset, addr *net.U cache := caches.Get(offset.SenderPubkey) info := cache.Update(*offset) - output := formatLocationOffset(offset, addr, signatureValid, verifyError) + output := formatLocationOffset(offset, addr, signatureValid, signatureError) if *verbose || info.Changed() { if *logFormat == "json" { @@ -446,7 +595,7 @@ type ReferenceOutput struct { MeasuredRttMs float64 `json:"measured_rtt_ms"` } -func formatLocationOffset(offset *geoprobe.LocationOffset, addr *net.UDPAddr, signatureValid bool, verifyError error) OffsetOutput { +func formatLocationOffset(offset *geoprobe.LocationOffset, addr *net.UDPAddr, signatureValid bool, signatureError string) OffsetOutput { rttMs := float64(offset.RttNs) / nanosecondsPerMs measuredRttMs := float64(offset.MeasuredRttNs) / nanosecondsPerMs maxDistanceMiles := calculateMaxDistance(offset.RttNs) @@ -465,10 +614,7 @@ func formatLocationOffset(offset *geoprobe.LocationOffset, addr *net.UDPAddr, si MaxDistanceKm: maxDistanceKm, MeasurementSlot: offset.MeasurementSlot, SignatureValid: signatureValid, - } - - if verifyError != nil { - output.SignatureError = verifyError.Error() + SignatureError: signatureError, } for _, ref := range offset.References { diff --git a/controlplane/telemetry/cmd/geoprobe-target/main_test.go b/controlplane/telemetry/cmd/geoprobe-target/main_test.go index f767179565..54eae53d74 100644 --- a/controlplane/telemetry/cmd/geoprobe-target/main_test.go +++ b/controlplane/telemetry/cmd/geoprobe-target/main_test.go @@ -47,17 +47,28 @@ func TestHandleOffset_DropsUnsignedOffset(t *testing.T) { copy(spoofed.AuthorityPubkey[:], impersonator.PublicKey().Bytes()) copy(spoofed.SenderPubkey[:], impersonator.PublicKey().Bytes()) spoofed.Signature[0] = 0xff + // Stamped far in the future: if the slot floor were consulted before the + // signature, a forgery could raise the impersonated key's floor and lock + // the real holder out. + spoofed.MeasurementSlot = 1 << 40 + floor := newSlotFloor(floorEntryTTL) for _, forged := range []*geoprobe.LocationOffset{unsigned, spoofed} { - handleOffset(log, forged, addr, true, writer, caches) + handleOffset(log, forged, addr, true, writer, caches, floor) - if got := writer.BufferedRows(); got != 0 { + if got := len(writer.PendingRows()); got != 0 { t.Errorf("expected forged offset to be dropped, got %d buffered clickhouse rows", got) } if _, ok := caches.Get(forged.SenderPubkey).Best(); ok { t.Error("expected forged offset to be dropped, but it entered the cache") } } + + genuine := signedOffsetAt(t, mustSigner(t, impersonator.PrivateKey, impersonator.PublicKey()), 1_000_000, 1_000_000) + handleOffset(log, genuine, addr, true, writer, caches, floor) + if got := len(writer.PendingRows()); got != 1 { + t.Errorf("a forged offset moved the impersonated key's floor: got %d rows for the genuine offset", got) + } } func TestHandleOffset_AcceptsSignedOffset(t *testing.T) { @@ -78,9 +89,9 @@ func TestHandleOffset_AcceptsSignedOffset(t *testing.T) { t.Fatalf("failed to sign offset: %v", err) } - handleOffset(log, offset, addr, true, writer, caches) + handleOffset(log, offset, addr, true, writer, caches, newSlotFloor(time.Hour)) - if got := writer.BufferedRows(); got != 1 { + if got := len(writer.PendingRows()); got != 1 { t.Errorf("expected 1 buffered clickhouse row for a signed offset, got %d", got) } best, ok := caches.Get(offset.SenderPubkey).Best() @@ -91,3 +102,225 @@ func TestHandleOffset_AcceptsSignedOffset(t *testing.T) { t.Errorf("expected cached RttNs=%d, got %d", offset.RttNs, best.RttNs) } } + +// signedOffsetAt returns an offset stamped with slot and rttNs, signed by +// signer so it passes the chain check. +func signedOffsetAt(t *testing.T, signer *geoprobe.OffsetSigner, slot, rttNs uint64) *geoprobe.LocationOffset { + t.Helper() + offset := newTestOffset() + offset.MeasurementSlot = slot + offset.RttNs = rttNs + if err := signer.SignOffset(offset); err != nil { + t.Fatalf("failed to sign offset: %v", err) + } + return offset +} + +func mustSigner(t *testing.T, key solana.PrivateKey, sender solana.PublicKey) *geoprobe.OffsetSigner { + t.Helper() + signer, err := geoprobe.NewOffsetSigner(key, sender) + if err != nil { + t.Fatalf("failed to create signer: %v", err) + } + return signer +} + +func newTestSigner(t *testing.T) *geoprobe.OffsetSigner { + t.Helper() + return mustSigner(t, solana.NewWallet().PrivateKey, solana.NewWallet().PublicKey()) +} + +// A signature stays valid forever, so an offset captured off the wire replays +// cleanly through the signature gate. The slot floor is what stops it reaching +// location_offsets. +func TestHandleOffset_RejectsReplayedOffset(t *testing.T) { + log := slog.New(slog.NewTextHandler(io.Discard, nil)) + caches := newTestCaches() + writer := geoprobe.NewClickhouseWriter(geoprobe.ClickhouseConfig{Addr: "unused"}, log) + addr := &net.UDPAddr{IP: net.IPv4(203, 0, 113, 7), Port: 41234} + floor := newSlotFloor(time.Hour) + signer := newTestSigner(t) + + const currentSlot = 1_000_000 + live := signedOffsetAt(t, signer, currentSlot, 5_000_000) + handleOffset(log, live, addr, true, writer, caches, floor) + if got := len(writer.PendingRows()); got != 1 { + t.Fatalf("expected the live offset to be recorded, got %d rows", got) + } + + // A capture from well before the floor, with a lower RTT so it would win + // the cache if it were accepted. + replay := signedOffsetAt(t, signer, currentSlot-maxSlotRegression-1, 1_000_000) + handleOffset(log, replay, addr, true, writer, caches, floor) + + if got := len(writer.PendingRows()); got != 1 { + t.Errorf("expected the replay to be dropped, got %d buffered rows", got) + } + best, ok := caches.Get(replay.SenderPubkey).Best() + if !ok { + t.Fatal("expected the live offset to remain cached") + } + if best.RttNs != live.RttNs { + t.Errorf("replay entered the cache: best RttNs=%d, want %d", best.RttNs, live.RttNs) + } + + // Positive control: a fresh slot over the same path is still accepted. + handleOffset(log, signedOffsetAt(t, signer, currentSlot+1, 4_000_000), addr, true, writer, caches, floor) + if got := len(writer.PendingRows()); got != 2 { + t.Errorf("expected a fresh-slot offset to be recorded, got %d rows", got) + } +} + +// Offsets are sent every 30s but stamped from a 5m slot cache, so most carry a +// slot the sender has already used. Rejecting repeats would drop nine of every +// ten legitimate offsets. +func TestSlotFloor_AcceptsRepeatedSlotWhileFresh(t *testing.T) { + floor := newSlotFloor(time.Hour) + now := time.Now() + floor.nowFunc = func() time.Time { return now } + sender := [32]byte{1} + + for i := 0; i < 10; i++ { + if ok, reason, _, _ := floor.accept(sender, 1_000_000); !ok { + t.Fatalf("repeat %d rejected while floor is fresh: %s", i, reason) + } + now = now.Add(30 * time.Second) + } + + // A slot inside the regression allowance is also accepted. + if ok, reason, _, _ := floor.accept(sender, 1_000_000-maxSlotRegression); !ok { + t.Fatalf("slot inside the regression allowance rejected: %s", reason) + } +} + +// Once a sender stops advancing its slot, repeats no longer evidence a live +// sender: without this bound a capture taken while the probe was alive stays +// acceptable forever after it goes quiet. +func TestSlotFloor_RejectsStalledFloor(t *testing.T) { + floor := newSlotFloor(time.Hour) + now := time.Now() + floor.nowFunc = func() time.Time { return now } + sender := [32]byte{1} + + floor.accept(sender, 1_000_000) + now = now.Add(maxFloorStall + time.Minute) + + ok, reason, floorSlot, floorAge := floor.accept(sender, 1_000_000) + if ok { + t.Fatal("expected a repeat at the frozen slot to be rejected once the floor stalled") + } + if reason != rejectFloorStalled { + t.Errorf("reason = %q, want %q", reason, rejectFloorStalled) + } + if floorSlot != 1_000_000 || floorAge < maxFloorStall { + t.Errorf("floor state = (slot %d, age %s), want slot 1000000 and age > %s", floorSlot, floorAge, maxFloorStall) + } + + // Positive control: a strictly higher slot proves liveness and unfreezes it. + if ok, reason, _, _ := floor.accept(sender, 1_000_001); !ok { + t.Fatalf("expected a higher slot to be accepted after a stall, got %s", reason) + } +} + +func TestSlotFloor_IsPerKey(t *testing.T) { + floor := newSlotFloor(floorEntryTTL) + fast := [32]byte{1} + slow := [32]byte{2} + + floor.accept(fast, 5_000_000) + + if ok, reason, _, _ := floor.accept(slow, 1_000); !ok { + t.Fatalf("a second key was judged against the first key's floor: %s", reason) + } +} + +// A rejected offer still refreshes the entry: otherwise a sustained replay +// outlives its own floor, the sweep drops the entry, and the next replay +// reseeds from itself — handing the attacker a fresh acceptance window every +// TTL instead of one per process restart. +func TestSlotFloor_RejectionKeepsFloorAlive(t *testing.T) { + floor := newSlotFloor(floorEntryTTL) + now := time.Now() + floor.nowFunc = func() time.Time { return now } + key := [32]byte{1} + + floor.accept(key, 1_000_000) + now = now.Add(maxFloorStall + time.Minute) + + // Keep replaying the frozen slot for well past floorEntryTTL, sweeping as + // the daemon does. + for elapsed := time.Duration(0); elapsed < 2*floorEntryTTL; elapsed += 5 * time.Minute { + if ok, _, _, _ := floor.accept(key, 1_000_000); ok { + t.Fatalf("replay accepted again %s after the floor stalled", elapsed) + } + floor.sweep() + now = now.Add(5 * time.Minute) + } + + // Positive control: a key that goes genuinely silent is eventually + // forgotten, so the map stays bounded. + silent := [32]byte{2} + floor.accept(silent, 1_000_000) + now = now.Add(floorEntryTTL + time.Minute) + floor.sweep() + if _, ok := floor.entries[silent]; ok { + t.Error("expected a silent key's floor to be swept") + } +} + +// SenderPubkey is signed but unauthenticated — anyone can mint a keypair and +// stamp a real geoprobe's SenderPubkey. Keying floors on it would let one +// datagram carrying a huge slot lock that geoprobe out of location_offsets. +func TestHandleOffset_ForgedSenderCannotPoisonAnotherFloor(t *testing.T) { + log := slog.New(slog.NewTextHandler(io.Discard, nil)) + caches := newTestCaches() + writer := geoprobe.NewClickhouseWriter(geoprobe.ClickhouseConfig{Addr: "unused"}, log) + addr := &net.UDPAddr{IP: net.IPv4(203, 0, 113, 7), Port: 41234} + floor := newSlotFloor(floorEntryTTL) + + victim := solana.NewWallet().PublicKey() + victimSigner := mustSigner(t, solana.NewWallet().PrivateKey, victim) + // An attacker's own keypair, claiming the victim geoprobe as sender. + attackerSigner := mustSigner(t, solana.NewWallet().PrivateKey, victim) + + handleOffset(log, signedOffsetAt(t, attackerSigner, 1<<40, 1_000_000), addr, true, writer, caches, floor) + + genuine := signedOffsetAt(t, victimSigner, 1_000_000, 5_000_000) + before := len(writer.PendingRows()) + handleOffset(log, genuine, addr, true, writer, caches, floor) + if len(writer.PendingRows()) != before+1 { + t.Error("a forged sender claim locked the real geoprobe out of location_offsets") + } +} + +// With verification disabled nothing is checked, so a row asserting +// signature_valid=true would be a mislabel in the table the public explorer +// reads. +func TestHandleOffset_UnverifiedRowIsNotLabelledValid(t *testing.T) { + log := slog.New(slog.NewTextHandler(io.Discard, nil)) + caches := newTestCaches() + writer := geoprobe.NewClickhouseWriter(geoprobe.ClickhouseConfig{Addr: "unused"}, log) + addr := &net.UDPAddr{IP: net.IPv4(203, 0, 113, 7), Port: 41234} + + handleOffset(log, newTestOffset(), addr, false, writer, caches, newSlotFloor(time.Hour)) + + rows := writer.PendingRows() + if len(rows) != 1 { + t.Fatalf("expected 1 buffered row with verification disabled, got %d", len(rows)) + } + if rows[0].SignatureValid { + t.Error("row asserts signature_valid=true though no verification ran") + } + if rows[0].SignatureError != signatureUnverifiedMarker { + t.Errorf("signature_error = %q, want %q", rows[0].SignatureError, signatureUnverifiedMarker) + } + + // Positive control: a verified offset is still labelled valid. + caches2 := newTestCaches() + writer2 := geoprobe.NewClickhouseWriter(geoprobe.ClickhouseConfig{Addr: "unused"}, log) + handleOffset(log, signedOffsetAt(t, newTestSigner(t), 12345, 1_000_000), addr, true, writer2, caches2, newSlotFloor(time.Hour)) + rows2 := writer2.PendingRows() + if len(rows2) != 1 || !rows2[0].SignatureValid || rows2[0].SignatureError != "" { + t.Errorf("expected a verified offset to be labelled valid with no error, got %+v", rows2) + } +} diff --git a/controlplane/telemetry/internal/geoprobe/clickhouse.go b/controlplane/telemetry/internal/geoprobe/clickhouse.go index d0be346e44..4de8f8cc24 100644 --- a/controlplane/telemetry/internal/geoprobe/clickhouse.go +++ b/controlplane/telemetry/internal/geoprobe/clickhouse.go @@ -152,11 +152,11 @@ func (w *ClickhouseWriter) Record(row OffsetRow) { w.mu.Unlock() } -// BufferedRows returns the number of rows waiting to be flushed. -func (w *ClickhouseWriter) BufferedRows() int { +// PendingRows returns a copy of the rows waiting to be flushed. +func (w *ClickhouseWriter) PendingRows() []OffsetRow { w.mu.Lock() defer w.mu.Unlock() - return len(w.buf) + return append([]OffsetRow(nil), w.buf...) } func (w *ClickhouseWriter) connect(ctx context.Context) error { diff --git a/controlplane/telemetry/internal/geoprobe/mincache.go b/controlplane/telemetry/internal/geoprobe/mincache.go index 5fc650a932..45a4c063e9 100644 --- a/controlplane/telemetry/internal/geoprobe/mincache.go +++ b/controlplane/telemetry/internal/geoprobe/mincache.go @@ -117,15 +117,18 @@ func (c *MinCache[T]) Update(value T) UpdateInfo { info.Result = UpdateBest return info } - if rttNs <= c.best.rttNs { - // New record low: reset best's clock, clear backup. + if rttNs < c.best.rttNs { + // New record low: reset best's clock, clear backup. An equal-RTT value + // must not replace best, because replacing also restarts best's + // receivedAt clock — a repeated measurement would otherwise keep + // pushing best's expiry out. c.best = entry c.backup = nil info.Result = UpdateBest return info } - // rttNs > best. Only collect a backup while best is in its final guard. + // rttNs >= best. Only collect a backup while best is in its final guard. if c.maxAge-now.Sub(c.best.receivedAt) > guard { c.backup = nil return info diff --git a/controlplane/telemetry/internal/geoprobe/mincache_test.go b/controlplane/telemetry/internal/geoprobe/mincache_test.go index ab0cf7d566..3c9195bb19 100644 --- a/controlplane/telemetry/internal/geoprobe/mincache_test.go +++ b/controlplane/telemetry/internal/geoprobe/mincache_test.go @@ -56,16 +56,33 @@ func TestMinCache_LowerRTTReplacesBest(t *testing.T) { } } -func TestMinCache_EqualRTTReplacesBest(t *testing.T) { - c, _ := newTestCache(time.Hour) +// Replacing best on an equal RTT also resets best's receivedAt clock, so a +// replayed measurement could pin best past its TTL indefinitely. Only a +// strictly lower RTT may restart the clock. +func TestMinCache_EqualRTTDoesNotReplaceBest(t *testing.T) { + c, now := newTestCache(time.Hour) c.Update(testMeasurement{rttNs: 1000, label: "first"}) - info := c.Update(testMeasurement{rttNs: 1000, label: "second"}) - if info.Result != UpdateBest { - t.Fatalf("expected UpdateBest for equal RTT, got %v", info.Result) + firstReceivedAt := c.best.receivedAt + + *now = now.Add(5 * time.Minute) + if info := c.Update(testMeasurement{rttNs: 1000, label: "replay"}); info.Result == UpdateBest { + t.Fatalf("equal-RTT sample became the new best, got %v", info.Result) } - got, _ := c.Best() - if got.label != "second" { - t.Fatalf("expected second, got %s", got.label) + if !c.best.receivedAt.Equal(firstReceivedAt) { + t.Fatalf("equal-RTT sample refreshed best's clock (%v -> %v); a replay could hold best forever", + firstReceivedAt, c.best.receivedAt) + } + if got, ok := c.Best(); !ok || got.label != "first" { + t.Fatalf("expected best to remain 'first', got %v (ok=%v)", got, ok) + } + + // Positive control: a strictly lower RTT does replace best and restart its clock. + *now = now.Add(5 * time.Minute) + if info := c.Update(testMeasurement{rttNs: 999, label: "lower"}); info.Result != UpdateBest { + t.Fatalf("expected UpdateBest for a lower RTT, got %v", info.Result) + } + if c.best.receivedAt.Equal(firstReceivedAt) { + t.Fatal("expected a lower-RTT sample to restart best's clock") } }