From e2affe10da2fae9eeda831324f631fa859d5c7fa Mon Sep 17 00:00:00 2001 From: Taksh Date: Mon, 5 Oct 2026 16:16:07 +0530 Subject: [PATCH] Bound the work a REQUEST_SYNC peer can trigger MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit REQUEST_SYNC is a public packet, handled in PacketProcessor's unauthenticated branch and passed straight to GossipSyncManager. Nothing between the radio and the response limits how often a neighbor may send one, so a single peer decided how often we decode a GCS filter, run SHA-256 over every stored packet, and transmit whatever its filter claims to be missing. A filter that claims to hold nothing is a few bytes and makes us replay up to seenCapacity() packets — 500 by default. Repeating it costs the sender nothing and saturates a link measured in kbps. Give each requester a token bucket: one request per 5s with a burst of 2, so the scheduled first sync and a periodic one both land, and a response allowance that refills to seenCapacity() over the 30s periodic interval — the most a peer can legitimately be missing in one round. An honest peer never reaches either limit, so what it receives is unchanged. The bookkeeping is keyed by a peer ID off the wire, so it is an access-ordered LinkedHashMap capped at 256 entries; otherwise the mitigation would be its own memory exhaustion vector. --- .../bitchat/android/sync/GossipSyncManager.kt | 94 ++++++++++++++++++- .../com/bitchat/android/util/AppConstants.kt | 19 ++++ 2 files changed, 112 insertions(+), 1 deletion(-) 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 42a10ceb..bc2465d9 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,68 @@ class GossipSyncManager( delegate?.sendPacketToPeer(peerID, signed) } + /** + * Charges [fromPeerID] for one request and returns how many packets it 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(fromPeerID: String): Int { + val capacity = try { + configProvider.seenCapacity() + } catch (_: Exception) { + 0 + }.coerceAtLeast(1).toDouble() + val now = nowMs() + + synchronized(requesterBudgets) { + val budget = requesterBudgets.getOrPut(fromPeerID) { + 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(fromPeerID: String, sent: Int) { + if (sent <= 0) return + synchronized(requesterBudgets) { + val budget = requesterBudgets[fromPeerID] ?: return + budget.responseTokens = (budget.responseTokens - sent).coerceAtLeast(0.0) + } + } + fun handleRequestSync(fromPeerID: String, request: RequestSyncPacket) { + var remaining = admitRequest(fromPeerID) + 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 +271,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 +286,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(fromPeerID, 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 b1822806..d13acbc9 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 {