diff --git a/app/src/main/java/com/bitchat/android/mesh/BluetoothMeshService.kt b/app/src/main/java/com/bitchat/android/mesh/BluetoothMeshService.kt index 22f178678..b71b4a157 100644 --- a/app/src/main/java/com/bitchat/android/mesh/BluetoothMeshService.kt +++ b/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) } } diff --git a/app/src/main/java/com/bitchat/android/mesh/MeshCore.kt b/app/src/main/java/com/bitchat/android/mesh/MeshCore.kt index 28cd3d021..4e3993f2b 100644 --- a/app/src/main/java/com/bitchat/android/mesh/MeshCore.kt +++ b/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) } } } diff --git a/app/src/main/java/com/bitchat/android/sync/GossipSyncManager.kt b/app/src/main/java/com/bitchat/android/sync/GossipSyncManager.kt index 42a10ceb8..71fd19c22 100644 --- a/app/src/main/java/com/bitchat/android/sync/GossipSyncManager.kt +++ b/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>() + /** + * 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(16, 0.75f, true) { + override fun removeEldestEntry( + eldest: MutableMap.MutableEntry + ): 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 { diff --git a/app/src/main/java/com/bitchat/android/util/AppConstants.kt b/app/src/main/java/com/bitchat/android/util/AppConstants.kt index b18228063..d13acbc92 100644 --- a/app/src/main/java/com/bitchat/android/util/AppConstants.kt +++ b/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 { diff --git a/app/src/test/kotlin/com/bitchat/android/sync/GossipSyncRequestBudgetTest.kt b/app/src/test/kotlin/com/bitchat/android/sync/GossipSyncRequestBudgetTest.kt new file mode 100644 index 000000000..f131589e8 --- /dev/null +++ b/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>() + + 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()) + } +} diff --git a/docs/sync.md b/docs/sync.md index 0bbd0735b..7cc2f6b46 100644 --- a/docs/sync.md +++ b/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: