Merge 2d80548ddbc5739bbed6ac90a97181e7a3865ac3 into 4f828b5643ff87a5064c8d384d0a4cb64001d2b2

app/src/main/java/com/bitchat/android/mesh/BluetoothMeshService.kt

@@ -654,7 +654,7 @@ class BluetoothMeshService(private val context: Context) : TransportBridgeServic
                 // Decode request and respond with missing packets
                 val fromPeer = routed.peerID ?: return
                 val req = RequestSyncPacket.decode(routed.packet.payload) ?: return
-                gossipSyncManager.handleRequestSync(fromPeer, req)
+                gossipSyncManager.handleRequestSync(fromPeer, req, routed.ingressLinkID)
             }
         }
         

app/src/main/java/com/bitchat/android/mesh/MeshCore.kt

@@ -525,7 +525,7 @@ class MeshCore(
             override fun handleRequestSync(routed: RoutedPacket) {
                 val fromPeer = routed.peerID ?: return
                 val req = RequestSyncPacket.decode(routed.packet.payload) ?: return
-                gossipSyncManager.handleRequestSync(fromPeer, req)
+                gossipSyncManager.handleRequestSync(fromPeer, req, routed.ingressLinkID)
             }
         }
     }

app/src/main/java/com/bitchat/android/sync/GossipSyncManager.kt

@@ -6,8 +6,10 @@ import com.bitchat.android.model.RequestSyncPacket
 import com.bitchat.android.protocol.BitchatPacket
 import com.bitchat.android.protocol.MessageType
 import com.bitchat.android.protocol.SpecialRecipients
+import com.bitchat.android.util.AppConstants
 import kotlinx.coroutines.*
 import java.util.concurrent.ConcurrentHashMap
+import kotlin.math.floor
 
 /**
  * Gossip-based synchronization manager using on-demand GCS filters.
@@ -17,7 +19,8 @@ import java.util.concurrent.ConcurrentHashMap
 class GossipSyncManager(
     private val myPeerID: String,
     private val scope: CoroutineScope,
-    private val configProvider: ConfigProvider
+    private val configProvider: ConfigProvider,
+    private val nowMs: () -> Long = { System.currentTimeMillis() }
 ) {
     interface Delegate {
         fun sendPacket(packet: BitchatPacket)
@@ -47,6 +50,25 @@ class GossipSyncManager(
     // - announcements: only keep latest per sender peerID
     private val latestAnnouncementByPeer = ConcurrentHashMap<String, Pair<String, BitchatPacket>>()
 
+    /**
+     * Per-requester token buckets. `requestTokens` bounds how often we are willing to decode a
+     * peer's filter and hash our stored packets against it; `responseTokens` bounds how many
+     * packets that peer can pull out of us. Both refill with elapsed time.
+     */
+    private class RequesterBudget(
+        var requestTokens: Double,
+        var responseTokens: Double,
+        var lastRefillMs: Long
+    )
+
+    // Access-ordered so the eldest entry is the least recently seen requester.
+    private val requesterBudgets =
+        object : LinkedHashMap<String, RequesterBudget>(16, 0.75f, true) {
+            override fun removeEldestEntry(
+                eldest: MutableMap.MutableEntry<String, RequesterBudget>
+            ): Boolean = size > AppConstants.Sync.MAX_TRACKED_REQUESTERS
+        }
+
     private var periodicJob: Job? = null
     private var cleanupJob: Job? = null
     fun start() {
@@ -85,6 +107,9 @@ class GossipSyncManager(
             messages.clear()
         }
         latestAnnouncementByPeer.clear()
+        synchronized(requesterBudgets) {
+            requesterBudgets.clear()
+        }
         Log.d(TAG, "Cleared all gossip sync messages and announcements")
     }
 
@@ -174,7 +199,80 @@ class GossipSyncManager(
         delegate?.sendPacketToPeer(peerID, signed)
     }
 
-    fun handleRequestSync(fromPeerID: String, request: RequestSyncPacket) {
+    /**
+     * Charges [budgetKey] for one request and returns how many packets may be sent in response,
+     * or 0 when the request must be dropped. Called before any decoding or hashing so a flood
+     * costs us the admission check and nothing else.
+     */
+    private fun admitRequest(budgetKey: String): Int {
+        val capacity = try {
+            configProvider.seenCapacity()
+        } catch (_: Exception) {
+            0
+        }.coerceAtLeast(1).toDouble()
+        val now = nowMs()
+
+        synchronized(requesterBudgets) {
+            val budget = requesterBudgets.getOrPut(budgetKey) {
+                RequesterBudget(
+                    requestTokens = AppConstants.Sync.REQUEST_BURST.toDouble(),
+                    responseTokens = capacity,
+                    lastRefillMs = now
+                )
+            }
+
+            // A clock that jumped backwards must not hand out free tokens.
+            val elapsedMs = (now - budget.lastRefillMs).coerceAtLeast(0L).toDouble()
+            budget.lastRefillMs = now
+            budget.requestTokens = (
+                budget.requestTokens + elapsedMs / AppConstants.Sync.REQUEST_INTERVAL_MS
+                ).coerceAtMost(AppConstants.Sync.REQUEST_BURST.toDouble())
+            budget.responseTokens = (
+                budget.responseTokens +
+                    capacity * elapsedMs / AppConstants.Sync.RESPONSE_REFILL_WINDOW_MS
+                ).coerceAtMost(capacity)
+
+            if (budget.requestTokens < 1.0) return 0
+            val allowance = floor(budget.responseTokens).toInt()
+            if (allowance <= 0) return 0
+
+            budget.requestTokens -= 1.0
+            return allowance
+        }
+    }
+
+    internal fun trackedRequesterCount(): Int = synchronized(requesterBudgets) {
+        requesterBudgets.size
+    }
+
+    private fun chargeResponses(budgetKey: String, sent: Int) {
+        if (sent <= 0) return
+        synchronized(requesterBudgets) {
+            val budget = requesterBudgets[budgetKey] ?: return
+            budget.responseTokens = (budget.responseTokens - sent).coerceAtLeast(0.0)
+        }
+    }
+
+    /**
+     * @param ingressLinkID locally-assigned identity of the link the request arrived on. REQUEST_SYNC
+     *   is not in the set of types `SecurityManager.verifyPacketSignature` authenticates, so the
+     *   sender ID on the wire is unauthenticated and free to rotate — budgeting by it would let a
+     *   peer mint a fresh allowance per request. This value is assigned by the transport and never
+     *   serialized onto the mesh, so a flooder is bounded by the links it can actually hold open.
+     */
+    fun handleRequestSync(
+        fromPeerID: String,
+        request: RequestSyncPacket,
+        ingressLinkID: String? = null
+    ) {
+        val budgetKey = ingressLinkID ?: fromPeerID
+        var remaining = admitRequest(budgetKey)
+        if (remaining <= 0) {
+            Log.d(TAG, "Dropping REQUEST_SYNC from $fromPeerID: over budget")
+            return
+        }
+        val allowance = remaining
+
         // Decode GCS into sorted set for membership checks
         val sorted = GCSFilter.decodeToSortedSet(request.p, request.m, request.data)
         fun mightContain(id: ByteArray): Boolean {
@@ -185,12 +283,14 @@ class GossipSyncManager(
 
         // 1) Announcements: send latest per peerID if remote doesn't have them
         for ((_, pair) in latestAnnouncementByPeer.entries) {
+            if (remaining <= 0) break
             val (id, pkt) = pair
             val idBytes = hexToBytes(id)
             if (!mightContain(idBytes)) {
                 // Send original packet unchanged to requester only (keep local TTL)
                 val toSend = pkt.copy(ttl = com.bitchat.android.util.AppConstants.SYNC_TTL_HOPS)
                 delegate?.sendPacketToPeer(fromPeerID, toSend)
+                remaining--
                 Log.d(TAG, "Sent sync announce: Type ${toSend.type} from ${toSend.senderID.toHexString()} to $fromPeerID packet id ${idBytes.toHexString()}")
             }
         }
@@ -198,13 +298,17 @@ class GossipSyncManager(
         // 2) Broadcast messages: send all they lack
         val toSendMsgs = synchronized(messages) { messages.values.toList() }
         for (pkt in toSendMsgs) {
+            if (remaining <= 0) break
             val idBytes = PacketIdUtil.computeIdBytes(pkt)
             if (!mightContain(idBytes)) {
                 val toSend = pkt.copy(ttl = com.bitchat.android.util.AppConstants.SYNC_TTL_HOPS)
                 delegate?.sendPacketToPeer(fromPeerID, toSend)
+                remaining--
                 Log.d(TAG, "Sent sync message: Type ${toSend.type} to $fromPeerID packet id ${idBytes.toHexString()}")
             }
         }
+
+        chargeResponses(budgetKey, allowance - remaining)
     }
 
     private fun hexStringToByteArray(hexString: String): ByteArray {

app/src/main/java/com/bitchat/android/util/AppConstants.kt

@@ -32,6 +32,25 @@ object AppConstants {
 
     object Sync {
         const val CLEANUP_INTERVAL_MS: Long = 60_000L
+
+        // REQUEST_SYNC is a public, unauthenticated packet, so a neighbor decides how often we
+        // decode a filter, hash every stored packet, and transmit whatever it claims to lack.
+        // The budgets below bound that work without touching what an honest peer needs.
+
+        // Shortest spacing between two serviced requests from the same peer. Honest peers ask on
+        // a 30s timer plus one scheduled first request, so only floods are shed.
+        const val REQUEST_INTERVAL_MS: Long = 5_000L
+
+        // Lets the scheduled first sync and a periodic one land back to back.
+        const val REQUEST_BURST: Int = 2
+
+        // Window over which a peer's response allowance refills to seenCapacity() packets — the
+        // most a peer can legitimately be missing in one round.
+        const val RESPONSE_REFILL_WINDOW_MS: Long = 30_000L
+
+        // Requester IDs come off the wire, so the bookkeeping itself has to be bounded or the
+        // mitigation becomes its own memory exhaustion vector.
+        const val MAX_TRACKED_REQUESTERS: Int = 256
     }
 
     object Fragmentation {

app/src/test/kotlin/com/bitchat/android/sync/GossipSyncRequestBudgetTest.kt

@@ -0,0 +1,206 @@
+package com.bitchat.android.sync
+
+import com.bitchat.android.model.RequestSyncPacket
+import com.bitchat.android.protocol.BitchatPacket
+import com.bitchat.android.protocol.MessageType
+import com.bitchat.android.protocol.SpecialRecipients
+import com.bitchat.android.util.AppConstants
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import org.junit.Assert.assertEquals
+import org.junit.Assert.assertTrue
+import org.junit.Before
+import org.junit.Test
+
+/**
+ * REQUEST_SYNC is a public packet handled straight off the wire, so an unauthenticated neighbor
+ * decides how often we decode a filter, hash every stored packet, and transmit what it claims to
+ * lack. These cover the budget that bounds that without changing what an honest peer receives.
+ */
+class GossipSyncRequestBudgetTest {
+
+    private companion object {
+        const val CAPACITY = 5
+        const val MY_PEER = "aabbccddeeff0011"
+        const val REQUESTER = "1122334455667788"
+    }
+
+    /** A filter that claims to hold nothing, so every stored packet counts as missing. */
+    private val emptyFilter = RequestSyncPacket(p = 1, m = 1L, data = ByteArray(0))
+
+    private var clock = 1_000_000L
+    private val sentTo = mutableListOf<Pair<String, BitchatPacket>>()
+
+    private lateinit var manager: GossipSyncManager
+
+    private val delegate = object : GossipSyncManager.Delegate {
+        override fun sendPacket(packet: BitchatPacket) = Unit
+        override fun sendPacketToPeer(peerID: String, packet: BitchatPacket) {
+            sentTo.add(peerID to packet)
+        }
+
+        override fun signPacketForBroadcast(packet: BitchatPacket): BitchatPacket = packet
+    }
+
+    private val config = object : GossipSyncManager.ConfigProvider {
+        override fun seenCapacity(): Int = CAPACITY
+        override fun gcsMaxBytes(): Int = 400
+        override fun gcsTargetFpr(): Double = 0.01
+    }
+
+    @Before
+    fun setUp() {
+        sentTo.clear()
+        clock = 1_000_000L
+        manager = GossipSyncManager(
+            myPeerID = MY_PEER,
+            scope = CoroutineScope(Dispatchers.Unconfined),
+            configProvider = config,
+            nowMs = { clock }
+        )
+        manager.delegate = delegate
+    }
+
+    private fun broadcastMessage(body: String): BitchatPacket = BitchatPacket(
+        type = MessageType.MESSAGE.value,
+        senderID = ByteArray(8) { 0x11 },
+        recipientID = SpecialRecipients.BROADCAST,
+        timestamp = clock.toULong(),
+        payload = body.toByteArray(),
+        ttl = 3u
+    )
+
+    private fun storeMessages(count: Int) {
+        repeat(count) { manager.onPublicPacketSeen(broadcastMessage("m$it")) }
+    }
+
+    @Test
+    fun `an honest peer still receives every packet it is missing`() {
+        storeMessages(CAPACITY)
+
+        manager.handleRequestSync(REQUESTER, emptyFilter)
+
+        assertEquals(CAPACITY, sentTo.size)
+        assertTrue(sentTo.all { it.first == REQUESTER })
+    }
+
+    @Test
+    fun `back to back requests stop being serviced after the burst allowance`() {
+        storeMessages(CAPACITY)
+
+        // No time passes between these, so nothing refills.
+        repeat(100) { manager.handleRequestSync(REQUESTER, emptyFilter) }
+
+        // Before the fix every one of the 100 requests replayed all CAPACITY packets.
+        assertEquals(CAPACITY, sentTo.size)
+    }
+
+    @Test
+    fun `a flood cannot pull more packets than the refill window allows`() {
+        storeMessages(CAPACITY)
+
+        // Hammer for a simulated minute, far faster than any honest peer syncs.
+        repeat(600) {
+            clock += 100L
+            manager.handleRequestSync(REQUESTER, emptyFilter)
+        }
+
+        // 60s of hammering earns two windows' worth of packets, not 600 replays.
+        val windows = 60_000L / AppConstants.Sync.RESPONSE_REFILL_WINDOW_MS
+        assertTrue(
+            "sent ${sentTo.size} packets, expected at most ${CAPACITY * (windows + 1)}",
+            sentTo.size <= CAPACITY * (windows + 1)
+        )
+        assertTrue("honest sync must still make progress", sentTo.isNotEmpty())
+    }
+
+    @Test
+    fun `the budget refills so a peer can keep syncing`() {
+        storeMessages(CAPACITY)
+
+        manager.handleRequestSync(REQUESTER, emptyFilter)
+        assertEquals(CAPACITY, sentTo.size)
+
+        sentTo.clear()
+        clock += AppConstants.Sync.RESPONSE_REFILL_WINDOW_MS
+        manager.handleRequestSync(REQUESTER, emptyFilter)
+
+        assertEquals(CAPACITY, sentTo.size)
+    }
+
+    @Test
+    fun `one flooding peer does not consume another peer's budget`() {
+        storeMessages(CAPACITY)
+        repeat(50) { manager.handleRequestSync(REQUESTER, emptyFilter) }
+        sentTo.clear()
+
+        manager.handleRequestSync("99aabbccddeeff00", emptyFilter)
+
+        assertEquals(CAPACITY, sentTo.size)
+    }
+
+    @Test
+    fun `requester bookkeeping stays bounded when peer ids are spoofed`() {
+        storeMessages(1)
+
+        repeat(AppConstants.Sync.MAX_TRACKED_REQUESTERS * 4) { i ->
+            manager.handleRequestSync("%016x".format(i), emptyFilter)
+        }
+
+        assertEquals(
+            AppConstants.Sync.MAX_TRACKED_REQUESTERS,
+            manager.trackedRequesterCount()
+        )
+    }
+
+    @Test
+    fun `rotating the sender id does not mint a fresh budget on one link`() {
+        storeMessages(CAPACITY)
+
+        // REQUEST_SYNC is not in the set SecurityManager.verifyPacketSignature authenticates, so
+        // the sender ID is free for a flooder to rotate. Cycling more IDs than the LRU tracks
+        // would hand out a full allowance per request if the budget were keyed on it.
+        repeat(AppConstants.Sync.MAX_TRACKED_REQUESTERS * 2) { i ->
+            manager.handleRequestSync("%016x".format(i), emptyFilter, ingressLinkID = "link-1")
+        }
+
+        assertEquals(CAPACITY, sentTo.size)
+        assertEquals(1, manager.trackedRequesterCount())
+    }
+
+    @Test
+    fun `separate links keep separate budgets`() {
+        storeMessages(CAPACITY)
+
+        manager.handleRequestSync(REQUESTER, emptyFilter, ingressLinkID = "link-1")
+        repeat(50) { manager.handleRequestSync(REQUESTER, emptyFilter, ingressLinkID = "link-1") }
+        sentTo.clear()
+
+        // A different physical link is a different neighbour and must not inherit the throttle.
+        manager.handleRequestSync(REQUESTER, emptyFilter, ingressLinkID = "link-2")
+
+        assertEquals(CAPACITY, sentTo.size)
+    }
+
+    @Test
+    fun `a backwards clock jump does not hand out free budget`() {
+        storeMessages(CAPACITY)
+
+        repeat(100) {
+            clock -= 5_000L
+            manager.handleRequestSync(REQUESTER, emptyFilter)
+        }
+
+        assertEquals(CAPACITY, sentTo.size)
+    }
+
+    @Test
+    fun `clearing sync state also clears the budgets`() {
+        storeMessages(CAPACITY)
+        manager.handleRequestSync(REQUESTER, emptyFilter)
+
+        manager.clear()
+
+        assertEquals(0, manager.trackedRequesterCount())
+    }
+}

docs/sync.md

@@ -71,6 +71,8 @@ Sender behavior:
 
 Receiver behavior:
 - Decode the REQUEST_SYNC payload and reconstruct the sorted set of mapped values using the provided P, M, and bitstream.
+- Admit the request against the per-ingress budget described under “Validation and limits” before
+  decoding anything; drop it otherwise.
 - For each locally stored public packet ID:
   - Compute h64(ID) % M and check if it is in the reconstructed set; if NOT present, send the original packet back with `ttl=0` to the requester only.
   - For announcements, send only the latest announcement per (sender peerID).
@@ -143,6 +145,22 @@ Validation and limits (recommended):
 
 - Reject malformed REQUEST_SYNC payloads (e.g., P < 1, M <= 0, or data length too large for local limits).
 - Practical bounds: data length in [0, 1024]; P in [1, 24]; M up to 2^32‑1.
+- Rate-limit how often a REQUEST_SYNC is serviced, and bound how many packets one requester can
+  pull. A filter that claims to hold nothing is a few bytes, and answering it costs a filter
+  decode, a packet-ID hash over every stored candidate, and up to “max packets per sync”
+  transmissions — so an unbounded responder amplifies a trivial request into the whole retention
+  set, repeatedly. The admission check MUST run before decoding or hashing, so a shed request
+  costs only the check.
+- Key that budget on a **locally-assigned ingress identity** (the link or connection the request
+  arrived on), never on the sender ID in the packet. REQUEST_SYNC carries no signature that
+  receivers verify, so the sender ID is free to rotate: budgeting by it lets one neighbor mint a
+  fresh allowance per request, and lets it spend another peer's. Bookkeeping keyed on anything
+  taken from the wire MUST itself be bounded, or the limiter becomes a memory-exhaustion vector.
+- Suggested shape (Android): one serviced request per 5s with a burst of 2, so the scheduled
+  first sync and a periodic one both land, plus a response allowance refilling to “max packets
+  per sync” over the 30s periodic interval — the most a peer can legitimately be missing in one
+  round. These are local policy; they need no cross-implementation agreement, but every
+  implementation needs some equivalent.
 
 Versioning: