mirror of
https://github.com/permissionlesstech/bitchat-android.git
synced 2026-10-18 07:33:01 +00:00
Bound the work a REQUEST_SYNC peer can trigger
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.
This commit is contained in:
parent
5156f7de89
commit
e2affe10da
@ -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,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 {
|
||||
|
||||
@ -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 {
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user