diff --git a/app/src/main/java/com/bitchat/android/mesh/BluetoothConnectionManager.kt b/app/src/main/java/com/bitchat/android/mesh/BluetoothConnectionManager.kt index a8649bd5..81e80b68 100644 --- a/app/src/main/java/com/bitchat/android/mesh/BluetoothConnectionManager.kt +++ b/app/src/main/java/com/bitchat/android/mesh/BluetoothConnectionManager.kt @@ -377,6 +377,16 @@ class BluetoothConnectionManager( ) } + suspend fun sendToPeerAndAwaitCompletion(peerID: String, routed: RoutedPacket): Boolean { + if (!isActive || !isBleTransportEnabled()) return false + return packetBroadcaster.sendPacketToPeerAndAwaitCompletion( + routed, + peerID, + serverManager.getGattServer(), + serverManager.getCharacteristic() + ) + } + fun cancelTransfer(transferId: String): Boolean { return packetBroadcaster.cancelTransfer(transferId) } 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 99a2c886..71e26a13 100644 --- a/app/src/main/java/com/bitchat/android/mesh/BluetoothMeshService.kt +++ b/app/src/main/java/com/bitchat/android/mesh/BluetoothMeshService.kt @@ -1620,7 +1620,7 @@ class BluetoothMeshService(private val context: Context) : TransportBridgeServic ) if (direct) { serviceScope.launch { - if (connectionManager.sendToPeerAndAwaitAcceptance( + if (connectionManager.sendToPeerAndAwaitCompletion( peerID, RoutedPacket(signPacketBeforeBroadcast(packet)) ) @@ -1641,7 +1641,7 @@ class BluetoothMeshService(private val context: Context) : TransportBridgeServic ttl = MAX_TTL ) serviceScope.launch { - if (connectionManager.sendToPeerAndAwaitAcceptance( + if (connectionManager.sendToPeerAndAwaitCompletion( peerID, RoutedPacket(signPacketBeforeBroadcast(packet)) ) diff --git a/app/src/main/java/com/bitchat/android/mesh/BluetoothPacketBroadcaster.kt b/app/src/main/java/com/bitchat/android/mesh/BluetoothPacketBroadcaster.kt index 5a9babca..a8fb804c 100644 --- a/app/src/main/java/com/bitchat/android/mesh/BluetoothPacketBroadcaster.kt +++ b/app/src/main/java/com/bitchat/android/mesh/BluetoothPacketBroadcaster.kt @@ -22,6 +22,7 @@ import kotlinx.coroutines.cancel import kotlinx.coroutines.delay import kotlinx.coroutines.isActive import kotlinx.coroutines.launch +import kotlinx.coroutines.withTimeoutOrNull import kotlinx.coroutines.channels.actor import java.util.ArrayDeque @@ -55,6 +56,7 @@ class BluetoothPacketBroadcaster( private const val MAX_PENDING_BYTES_PER_LINK = 1_048_576 private const val SEND_RETRY_DELAY_MS = 15L private const val MAX_CALLBACK_RETRIES = 3 + private const val SEND_COMPLETION_TIMEOUT_MS = 30_000L } // Optional nickname resolver injected by higher layer (peerID -> nickname?) @@ -140,6 +142,7 @@ class BluetoothPacketBroadcaster( val gatt: BluetoothGatt? = null, val gattServer: BluetoothGattServer? = null, val characteristic: BluetoothGattCharacteristic, + val completion: CompletableDeferred? = null, var callbackFailures: Int = 0 ) @@ -216,6 +219,18 @@ class BluetoothPacketBroadcaster( } } + suspend fun sendPacketToPeerAndAwaitCompletion( + routed: RoutedPacket, + targetPeerID: String, + gattServer: BluetoothGattServer?, + characteristic: BluetoothGattCharacteristic? + ): Boolean { + if (!hasPeerConnection(targetPeerID)) return false + return fragmentingSender.sendAndAwaitAcceptance(routed, "BLE peer ${targetPeerID.take(8)}") { packet -> + sendSinglePacketToPeerAndAwaitCompletion(packet, targetPeerID, gattServer, characteristic) + } + } + fun sendPacketToLink( routed: RoutedPacket, deviceAddress: String, @@ -279,6 +294,35 @@ class BluetoothPacketBroadcaster( return false } + private suspend fun sendSinglePacketToPeerAndAwaitCompletion( + routed: RoutedPacket, + targetPeerID: String, + gattServer: BluetoothGattServer?, + characteristic: BluetoothGattCharacteristic? + ): Boolean { + val packet = routed.packet + val data = packet.toBinaryData( + padding = BLEPacketPaddingPolicy.shouldPadForBLE(packet.type) + ) ?: return false + val completion = CompletableDeferred() + + val serverTarget = connectionTracker.getSubscribedDevices() + .firstOrNull { connectionTracker.addressPeerMap[it.address] == targetPeerID } + val queuedOnServer = serverTarget != null && + notifyDevice(serverTarget, data, gattServer, characteristic, completion) + val queued = if (queuedOnServer) { + true + } else { + val clientTarget = connectionTracker.getConnectedDevices().values + .firstOrNull { connectionTracker.addressPeerMap[it.device.address] == targetPeerID } + ?: return false + writeToDeviceConn(clientTarget, data, completion) + } + if (!queued) return false + + return withTimeoutOrNull(SEND_COMPLETION_TIMEOUT_MS) { completion.await() } ?: false + } + /** * Public entry point for broadcasting - submits request to actor for serialization @@ -486,7 +530,8 @@ class BluetoothPacketBroadcaster( device: BluetoothDevice, data: ByteArray, gattServer: BluetoothGattServer?, - characteristic: BluetoothGattCharacteristic? + characteristic: BluetoothGattCharacteristic?, + completion: CompletableDeferred? = null ): Boolean { val server = gattServer ?: return false val char = characteristic ?: return false @@ -496,7 +541,13 @@ class BluetoothPacketBroadcaster( ?: return false return enqueueSend( SendKey(device.address, linkID, SendDirection.SERVER_NOTIFICATION), - PendingSend(data.copyOf(), device, gattServer = server, characteristic = char) + PendingSend( + data.copyOf(), + device, + gattServer = server, + characteristic = char, + completion = completion + ) ) } @@ -505,13 +556,20 @@ class BluetoothPacketBroadcaster( */ private fun writeToDeviceConn( deviceConn: BluetoothConnectionTracker.DeviceConnection, - data: ByteArray + data: ByteArray, + completion: CompletableDeferred? = null ): Boolean { val gatt = deviceConn.gatt ?: return false val char = deviceConn.characteristic ?: return false return enqueueSend( SendKey(deviceConn.device.address, deviceConn.linkID, SendDirection.CLIENT_WRITE), - PendingSend(data.copyOf(), deviceConn.device, gatt = gatt, characteristic = char) + PendingSend( + data.copyOf(), + deviceConn.device, + gatt = gatt, + characteristic = char, + completion = completion + ) ) } @@ -643,6 +701,7 @@ class BluetoothPacketBroadcaster( } state.pending.removeFirst() state.pendingBytes -= head.data.size + head.completion?.complete(status == BluetoothGatt.GATT_SUCCESS) if (state.pending.isEmpty()) { sendStates.remove(key) false @@ -657,8 +716,13 @@ class BluetoothPacketBroadcaster( fun onLinkDisconnected(deviceAddress: String, linkID: String?) { synchronized(sendLock) { - sendStates.keys.removeAll { key -> - key.deviceAddress == deviceAddress && (linkID == null || key.linkID == linkID) + val iterator = sendStates.entries.iterator() + while (iterator.hasNext()) { + val (key, state) = iterator.next() + if (key.deviceAddress == deviceAddress && (linkID == null || key.linkID == linkID)) { + state.pending.forEach { it.completion?.complete(false) } + iterator.remove() + } } } } @@ -679,7 +743,12 @@ class BluetoothPacketBroadcaster( * Shutdown the broadcaster actor gracefully */ fun shutdown() { - synchronized(sendLock) { sendStates.clear() } + synchronized(sendLock) { + sendStates.values.forEach { state -> + state.pending.forEach { it.completion?.complete(false) } + } + sendStates.clear() + } // Close the actor gracefully broadcasterActor.close() diff --git a/app/src/main/java/com/bitchat/android/mesh/FragmentingPacketSender.kt b/app/src/main/java/com/bitchat/android/mesh/FragmentingPacketSender.kt index 8fc0bf5c..adac55c2 100644 --- a/app/src/main/java/com/bitchat/android/mesh/FragmentingPacketSender.kt +++ b/app/src/main/java/com/bitchat/android/mesh/FragmentingPacketSender.kt @@ -115,7 +115,7 @@ class FragmentingPacketSender( suspend fun sendAndAwaitAcceptance( routed: RoutedPacket, description: String, - sendSingle: (RoutedPacket) -> Boolean + sendSingle: suspend (RoutedPacket) -> Boolean ): Boolean { val packets = packetsForTransport(routed) ?: return false Log.d(logTag, "Sending ${packets.size} packet(s) for $description") diff --git a/app/src/main/java/com/bitchat/android/mesh/MessageHandler.kt b/app/src/main/java/com/bitchat/android/mesh/MessageHandler.kt index 97d07b88..ae890316 100644 --- a/app/src/main/java/com/bitchat/android/mesh/MessageHandler.kt +++ b/app/src/main/java/com/bitchat/android/mesh/MessageHandler.kt @@ -481,7 +481,10 @@ class MessageHandler(private val myPeerID: String, private val appContext: andro val peerID = routed.peerID ?: "unknown" if (packet.timestamp > Long.MAX_VALUE.toULong()) return val ageMs = System.currentTimeMillis() - packet.timestamp.toLong() - if (ageMs !in 0..com.bitchat.android.sync.GossipSyncManager.PUBLIC_MESSAGE_MAX_AGE_MS) return + if (ageMs !in + -com.bitchat.android.sync.GossipSyncManager.PUBLIC_PACKET_FUTURE_SKEW_MS.. + com.bitchat.android.sync.GossipSyncManager.PUBLIC_MESSAGE_MAX_AGE_MS + ) return // Enforce: only accept public messages from verified peers we know val peerInfo = delegate?.getPeerInfo(peerID) diff --git a/app/src/main/java/com/bitchat/android/mesh/PacketProcessor.kt b/app/src/main/java/com/bitchat/android/mesh/PacketProcessor.kt index b111433c..039c08e1 100644 --- a/app/src/main/java/com/bitchat/android/mesh/PacketProcessor.kt +++ b/app/src/main/java/com/bitchat/android/mesh/PacketProcessor.kt @@ -132,7 +132,9 @@ class PacketProcessor(private val myPeerID: String) { com.bitchat.android.sync.GossipSyncManager.FRAGMENT_MAX_AGE_MS else -> null } - if (maxAgeMs != null && ageMs !in 0..maxAgeMs) return + if (maxAgeMs != null && ageMs !in + -com.bitchat.android.sync.GossipSyncManager.PUBLIC_PACKET_FUTURE_SKEW_MS..maxAgeMs + ) return } else if (isBroadcast && packet.timestamp > Long.MAX_VALUE.toULong()) { return } 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 92fb25c9..68d6d79c 100644 --- a/app/src/main/java/com/bitchat/android/sync/GossipSyncManager.kt +++ b/app/src/main/java/com/bitchat/android/sync/GossipSyncManager.kt @@ -39,6 +39,7 @@ class GossipSyncManager( private const val TAG = "GossipSyncManager" const val PUBLIC_MESSAGE_MAX_AGE_MS = 6 * 60 * 60 * 1000L const val FRAGMENT_MAX_AGE_MS = 15 * 60 * 1000L + const val PUBLIC_PACKET_FUTURE_SKEW_MS = 10 * 60 * 1000L private const val ARCHIVE_FILE = "gossip-public-history.bin" } @@ -132,7 +133,7 @@ class GossipSyncManager( if (isBroadcastMessage) { val now = System.currentTimeMillis() val age = now - packet.timestamp.toLong() - if (age !in 0..PUBLIC_MESSAGE_MAX_AGE_MS) return + if (age !in -PUBLIC_PACKET_FUTURE_SKEW_MS..PUBLIC_MESSAGE_MAX_AGE_MS) return synchronized(messages) { messages[id] = packet // Enforce capacity (remove oldest when exceeded) @@ -145,7 +146,7 @@ class GossipSyncManager( if (!restoringArchive) persistArchive() } else if (isFragment) { val age = System.currentTimeMillis() - packet.timestamp.toLong() - if (age !in 0..FRAGMENT_MAX_AGE_MS) return + if (age !in -PUBLIC_PACKET_FUTURE_SKEW_MS..FRAGMENT_MAX_AGE_MS) return synchronized(fragments) { fragments[id] = packet while (fragments.size > configProvider.seenCapacity().coerceAtLeast(1)) { diff --git a/app/src/test/kotlin/com/bitchat/android/mesh/BluetoothPacketBroadcasterCompletionTest.kt b/app/src/test/kotlin/com/bitchat/android/mesh/BluetoothPacketBroadcasterCompletionTest.kt new file mode 100644 index 00000000..121c0f91 --- /dev/null +++ b/app/src/test/kotlin/com/bitchat/android/mesh/BluetoothPacketBroadcasterCompletionTest.kt @@ -0,0 +1,110 @@ +package com.bitchat.android.mesh + +import android.bluetooth.BluetoothDevice +import android.bluetooth.BluetoothGatt +import android.bluetooth.BluetoothGattCharacteristic +import android.os.Build +import com.bitchat.android.model.RoutedPacket +import com.bitchat.android.protocol.BitchatPacket +import com.bitchat.android.protocol.MessageType +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.SupervisorJob +import kotlinx.coroutines.async +import kotlinx.coroutines.cancel +import kotlinx.coroutines.runBlocking +import org.junit.After +import org.junit.Assert.assertFalse +import org.junit.Assert.assertTrue +import org.junit.Test +import org.junit.runner.RunWith +import org.mockito.kotlin.any +import org.mockito.kotlin.mock +import org.mockito.kotlin.timeout +import org.mockito.kotlin.verify +import org.mockito.kotlin.whenever +import org.robolectric.RobolectricTestRunner +import org.robolectric.annotation.Config + +@RunWith(RobolectricTestRunner::class) +@Config(sdk = [Build.VERSION_CODES.P], manifest = Config.NONE) +class BluetoothPacketBroadcasterCompletionTest { + private val scope = CoroutineScope(Dispatchers.Default + SupervisorJob()) + private val tracker = BluetoothConnectionTracker(scope, mock()) + private val broadcaster = BluetoothPacketBroadcaster(scope, tracker, null, MY_PEER_ID) + + @After + fun tearDown() { + broadcaster.shutdown() + scope.cancel() + } + + @Test + fun `queue admission waits for successful GATT completion`() = runBlocking { + val connection = connectedPeer() + + val result = async(Dispatchers.Default) { + broadcaster.sendPacketToPeerAndAwaitCompletion(packet(), PEER_ID, null, null) + } + + verify(connection.gatt!!, timeout(1_000)).writeCharacteristic(connection.characteristic!!) + assertFalse(result.isCompleted) + + broadcaster.onGattClientWriteComplete(DEVICE_ADDRESS, LINK_ID, BluetoothGatt.GATT_SUCCESS) + assertTrue(result.await()) + } + + @Test + fun `disconnect fails an admitted send before custody can be released`() = runBlocking { + val connection = connectedPeer() + + val result = async(Dispatchers.Default) { + broadcaster.sendPacketToPeerAndAwaitCompletion(packet(), PEER_ID, null, null) + } + + verify(connection.gatt!!, timeout(1_000)).writeCharacteristic(connection.characteristic!!) + assertFalse(result.isCompleted) + + broadcaster.onLinkDisconnected(DEVICE_ADDRESS, LINK_ID) + assertFalse(result.await()) + } + + private fun connectedPeer(): BluetoothConnectionTracker.DeviceConnection { + val device = mock() + val gatt = mock() + val characteristic = mock() + whenever(device.address).thenReturn(DEVICE_ADDRESS) + whenever(gatt.writeCharacteristic(any())).thenReturn(true) + val connection = BluetoothConnectionTracker.DeviceConnection( + device = device, + gatt = gatt, + characteristic = characteristic, + isClient = true, + linkID = LINK_ID + ) + tracker.addDeviceConnection(DEVICE_ADDRESS, connection) + tracker.observePeerIfCurrent(DEVICE_ADDRESS, LINK_ID, PEER_ID) + return connection + } + + private fun packet() = RoutedPacket( + BitchatPacket( + type = MessageType.COURIER_ENVELOPE.value, + senderID = MY_PEER_ID.hexToBytes(), + recipientID = PEER_ID.hexToBytes(), + timestamp = System.currentTimeMillis().toULong(), + payload = byteArrayOf(1, 2, 3), + ttl = 7u + ) + ) + + private fun String.hexToBytes(): ByteArray = + chunked(2).map { it.toInt(16).toByte() }.toByteArray() + + private companion object { + const val MY_PEER_ID = "1111222233334444" + const val PEER_ID = "aaaabbbbccccdddd" + const val DEVICE_ADDRESS = "00:11:22:33:44:55" + const val LINK_ID = "synthetic-link" + } +} diff --git a/app/src/test/kotlin/com/bitchat/android/mesh/PacketProcessorAnnounceSideEffectTest.kt b/app/src/test/kotlin/com/bitchat/android/mesh/PacketProcessorAnnounceSideEffectTest.kt index 3a676382..2564273b 100644 --- a/app/src/test/kotlin/com/bitchat/android/mesh/PacketProcessorAnnounceSideEffectTest.kt +++ b/app/src/test/kotlin/com/bitchat/android/mesh/PacketProcessorAnnounceSideEffectTest.kt @@ -64,6 +64,26 @@ class PacketProcessorAnnounceSideEffectTest { assertEquals(PEER_ID, withTimeout(1_000) { delegate.lastSeen.await() }) } + @Test + fun `broadcast message within future skew is handled`() = runBlocking { + val delegate = RecordingDelegate(acceptAnnounce = true) + val processor = processor(delegate) + + processor.processPacket(message(timestampOffsetMs = 60_000)) + + withTimeout(1_000) { delegate.messageHandled.await() } + } + + @Test + fun `broadcast message beyond future skew is rejected`() = runBlocking { + val delegate = RecordingDelegate(acceptAnnounce = true) + val processor = processor(delegate) + + processor.processPacket(message(timestampOffsetMs = 11 * 60_000L)) + + assertNull(withTimeoutOrNull(250) { delegate.messageHandled.await() }) + } + private fun processor(delegate: RecordingDelegate): PacketProcessor = PacketProcessor(MY_PEER_ID).also { it.delegate = delegate @@ -96,12 +116,26 @@ class PacketProcessorAnnounceSideEffectTest { return RoutedPacket(packet, PEER_ID, "direct-link") } + private fun message(timestampOffsetMs: Long): RoutedPacket { + val packet = BitchatPacket( + version = 1u, + type = MessageType.MESSAGE.value, + senderID = PEER_ID.hexToBytes(), + recipientID = SpecialRecipients.BROADCAST, + timestamp = (System.currentTimeMillis() + timestampOffsetMs).toULong(), + payload = byteArrayOf(0x01), + ttl = 7u + ) + return RoutedPacket(packet, PEER_ID, "direct-link") + } + private class RecordingDelegate( private val acceptAnnounce: Boolean, private val acceptHandshake: Boolean = false ) : PacketProcessorDelegate { val handled = CompletableDeferred() val handshakeHandled = CompletableDeferred() + val messageHandled = CompletableDeferred() val lastSeen = CompletableDeferred() @Volatile var relayCount = 0 @@ -121,7 +155,9 @@ class PacketProcessorAnnounceSideEffectTest { handled.complete(Unit) return acceptAnnounce } - override fun handleMessage(routed: RoutedPacket) = Unit + override fun handleMessage(routed: RoutedPacket) { + messageHandled.complete(Unit) + } override fun handleLeave(routed: RoutedPacket) = Unit override fun handleFragment(packet: BitchatPacket): BitchatPacket? = null override fun handleRequestSync(routed: RoutedPacket) = Unit diff --git a/app/src/test/kotlin/com/bitchat/android/sync/GossipSyncManagerTest.kt b/app/src/test/kotlin/com/bitchat/android/sync/GossipSyncManagerTest.kt index 1b0f2cad..c2d1c2c4 100644 --- a/app/src/test/kotlin/com/bitchat/android/sync/GossipSyncManagerTest.kt +++ b/app/src/test/kotlin/com/bitchat/android/sync/GossipSyncManagerTest.kt @@ -41,6 +41,27 @@ class GossipSyncManagerTest { assertEquals(listOf(1, 3, 5), sent.map { it.payload.single().toInt() }) } + @Test + fun `public history tolerates bounded future clock skew`() { + val sent = mutableListOf() + val manager = GossipSyncManager("1111222233334444", TestScope(), config) + manager.delegate = object : GossipSyncManager.Delegate { + override fun sendPacket(packet: BitchatPacket) = Unit + override fun sendPacketToPeer(peerID: String, packet: BitchatPacket) { sent += packet } + override fun signPacketForBroadcast(packet: BitchatPacket) = packet + } + val now = System.currentTimeMillis() + manager.onPublicPacketSeen(packet(MessageType.MESSAGE, now + 60_000L, 1)) + manager.onPublicPacketSeen(packet(MessageType.FRAGMENT, now + 60_000L, 2)) + manager.onPublicPacketSeen(packet(MessageType.FILE_TRANSFER, now + 60_000L, 3)) + manager.onPublicPacketSeen(packet(MessageType.MESSAGE, now + 11 * 60_000L, 4)) + manager.onPublicPacketSeen(packet(MessageType.FRAGMENT, now + 11 * 60_000L, 5)) + + manager.handleRequestSync("peer", RequestSyncPacket(p = 1, m = 1, data = byteArrayOf())) + + assertEquals(listOf(1, 2, 3), sent.map { it.payload.single().toInt() }) + } + private fun packet(type: MessageType, timestamp: Long, marker: Int) = BitchatPacket( type = type.value, senderID = ByteArray(8) { 1 },