fix(store-forward): harden offline delivery

This commit is contained in:
callebtc 2026-08-18 02:16:05 +02:00
parent 4ec89e0b79
commit 8e5a28e5b1
12 changed files with 1136 additions and 48 deletions

View File

@ -28,6 +28,7 @@ import com.bitchat.android.service.MeshServiceHolder
import com.bitchat.android.service.TransportBridgeService
import com.bitchat.android.services.AppStateStore
import com.bitchat.android.services.ConversationStorageCipher
import com.bitchat.android.services.MessageRouter
import com.bitchat.android.ui.DataManager
import com.bitchat.android.ui.PrivateMediaRecipientResolver
import com.bitchat.android.ui.debug.DebugSettingsManager
@ -75,6 +76,13 @@ object TestHookDriver {
"announce" -> announce(context)
"broadcast_msg" -> broadcastMsg(context, intent.requiredString("content"), intent.getStringExtra("channel"))
"dm_send" -> dmSend(context, intent.requiredString("peer"), intent.requiredString("content"), intent.getStringExtra("msg_id"))
"router_private_send" -> routerPrivateSend(
context,
intent.requiredString("peer"),
intent.requiredString("content"),
intent.getStringExtra("msg_id")
)
"router_resume" -> routerResume(context)
"dm_recv" -> dmRecv(context, intent)
"msg_recv" -> msgRecv(context, intent)
"favorite_set" -> favoriteSet(
@ -96,6 +104,7 @@ object TestHookDriver {
"ptt_recv" -> pttRecv(context, intent)
"raw_send" -> rawSend(context, intent)
"courier_contract" -> courierContract(context)
"cache_peer_identity" -> cachePeerIdentity(context, intent.requiredString("peer"))
"sync_request" -> syncRequest(intent.requiredString("peer"))
"ble" -> setBle(intent.getBooleanExtra("enabled", true))
"wifi_aware" -> setWifiAware(intent.getBooleanExtra("enabled", true))
@ -252,6 +261,27 @@ object TestHookDriver {
return ok("dm_send").put("peer", peerID).put("msg_id", id)
}
/**
* Drives the same durable outbox/router path used by private-chat sends instead of the
* lower-level direct mesh API that backs `dm_send`.
*/
private fun routerPrivateSend(context: Context, peerID: String, content: String, msgID: String?): JSONObject {
val mesh = mesh(context)
val nickname = mesh.getPeerNicknames()[peerID] ?: peerID
val id = msgID ?: "testhook-router-${System.currentTimeMillis()}"
val route = MessageRouter.getInstance(context, mesh).sendPrivate(content, peerID, nickname, id)
return ok("router_private_send")
.put("peer", peerID)
.put("msg_id", id)
.put("route", route.name)
}
/** Recreates the durable router after a process restart without adding another message. */
private fun routerResume(context: Context): JSONObject {
MessageRouter.getInstance(context, mesh(context))
return ok("router_resume")
}
private suspend fun dmRecv(context: Context, intent: Intent): JSONObject {
val timeoutMs = intent.getLongExtra("timeout_ms", DEFAULT_RECV_TIMEOUT_MS)
val fromPeer = intent.getStringExtra("peer")
@ -744,6 +774,25 @@ object TestHookDriver {
}
}
/** Persist the currently authenticated peer key exactly as the normal UI session observer does. */
private fun cachePeerIdentity(context: Context, peerID: String): JSONObject {
val info = mesh(context).getPeerInfo(peerID)
?: return err("cache_peer_identity", "peer is not known")
val noiseKey = info.noisePublicKey
?: return err("cache_peer_identity", "peer Noise key is unavailable")
val noiseKeyHex = noiseKey.toHex()
val identityManager = SecureIdentityStateManager(context)
identityManager.cachePeerNoiseKey(peerID, noiseKeyHex)
identityManager.cacheNoiseFingerprint(noiseKeyHex, com.bitchat.android.services.ContactIdentityResolver.fingerprintHex(noiseKey))
info.nickname.takeIf { it.isNotBlank() }?.let { nickname ->
identityManager.cacheFingerprintNickname(
com.bitchat.android.services.ContactIdentityResolver.fingerprintHex(noiseKey),
nickname
)
}
return ok("cache_peer_identity").put("peer", peerID)
}
private fun syncRequest(peerID: String): JSONObject {
val manager = MeshServiceHolder.sharedGossipSyncManager
?: return err("sync_request", "gossip sync manager is unavailable")

View File

@ -14,19 +14,104 @@ import kotlinx.coroutines.delay
import kotlinx.coroutines.isActive
import kotlinx.coroutines.launch
/**
* Minimal relay surface used by the bridge courier.
*
* Keeping this adapter boundary small lets the kind-1401 contract be exercised with a
* deterministic in-memory relay in unit tests, without opening a network connection.
*/
internal interface BridgeCourierRelay {
fun subscribe(
filter: NostrFilter,
id: String,
targetRelayUrls: List<String>,
handler: (NostrEvent) -> Unit
)
fun unsubscribe(id: String)
fun hasConnectedRelay(relayUrls: Collection<String>): Boolean
fun sendEvent(
event: NostrEvent,
relayUrls: List<String>,
onAccepted: () -> Unit
): Boolean
}
internal interface BridgeCourierCipher {
fun staticPublicKey(): ByteArray?
fun seal(payload: ByteArray, recipientNoiseKey: ByteArray): ByteArray
}
private class NostrBridgeCourierRelay(
private val relayManager: NostrRelayManager
) : BridgeCourierRelay {
override fun subscribe(
filter: NostrFilter,
id: String,
targetRelayUrls: List<String>,
handler: (NostrEvent) -> Unit
) {
relayManager.subscribe(
filter = filter,
id = id,
targetRelayUrls = targetRelayUrls,
handler = handler
)
}
override fun unsubscribe(id: String) {
relayManager.unsubscribe(id)
}
override fun hasConnectedRelay(relayUrls: Collection<String>): Boolean =
relayManager.hasConnectedRelay(relayUrls)
override fun sendEvent(
event: NostrEvent,
relayUrls: List<String>,
onAccepted: () -> Unit
): Boolean = relayManager.sendEvent(event, relayUrls, onAccepted = onAccepted)
}
private class EncryptionBridgeCourierCipher(
private val encryptionService: EncryptionService
) : BridgeCourierCipher {
override fun staticPublicKey(): ByteArray? = encryptionService.getStaticPublicKey()
override fun seal(payload: ByteArray, recipientNoiseKey: ByteArray): ByteArray =
encryptionService.sealCourierPayload(payload, recipientNoiseKey)
}
/** Parks opaque courier envelopes on default Nostr relays using the iOS kind-1401 contract. */
class BridgeCourierService(
context: Context,
private val encryptionService: EncryptionService,
private val onEnvelope: (CourierEnvelope) -> Unit
class BridgeCourierService internal constructor(
private val cipher: BridgeCourierCipher,
private val onEnvelope: (CourierEnvelope) -> Unit,
private val relayManager: BridgeCourierRelay,
private val relayUrls: List<String> = NostrRelayManager.defaultRelays(),
private val clock: () -> Long = System::currentTimeMillis,
private val identityFactory: () -> NostrIdentity = NostrIdentity::generate
) {
constructor(
context: Context,
encryptionService: EncryptionService,
onEnvelope: (CourierEnvelope) -> Unit
) : this(
cipher = EncryptionBridgeCourierCipher(encryptionService),
onEnvelope = onEnvelope,
relayManager = NostrBridgeCourierRelay(
NostrRelayManager.getInstance(context.applicationContext)
)
)
companion object {
private const val KIND = 1401
private const val MAX_ENCODED_BYTES = 20 * 1024
private const val TAG_REFRESH_INTERVAL_MS = 60 * 60 * 1000L
}
private val relayManager = NostrRelayManager.getInstance(context.applicationContext)
private val seenEvents = Collections.synchronizedMap(
object : LinkedHashMap<String, Unit>(512, 0.75f, true) {
override fun removeEldestEntry(eldest: MutableMap.MutableEntry<String, Unit>?) = size > 512
@ -40,9 +125,10 @@ class BridgeCourierService(
@Synchronized
fun start() {
val localKey = encryptionService.getStaticPublicKey() ?: ByteArray(0)
val localKey = cipher.staticPublicKey() ?: ByteArray(0)
if (localKey.size == 32) {
val day = CourierEnvelope.epochDay(System.currentTimeMillis())
val now = clock()
val day = CourierEnvelope.epochDay(now)
if (subscribedDay == day) return
if (subscribedDay != null) relayManager.unsubscribe(subscriptionID)
val tags = listOf(day - 1u, day, day + 1u).map {
@ -50,14 +136,14 @@ class BridgeCourierService(
}
val filter = NostrFilter.Builder()
.kinds(KIND)
.since(System.currentTimeMillis() - CourierEnvelope.MAX_LIFETIME_MS)
.since(now - CourierEnvelope.MAX_LIFETIME_MS)
.limit(100)
.tag("x", *tags.toTypedArray())
.build()
relayManager.subscribe(
filter = filter,
id = subscriptionID,
targetRelayUrls = NostrRelayManager.defaultRelays(),
targetRelayUrls = relayUrls,
handler = ::handleEvent
)
subscribedDay = day
@ -92,9 +178,9 @@ class BridgeCourierService(
onAccepted: () -> Unit = {}
): Boolean {
start()
if (!relayManager.hasConnectedRelay(NostrRelayManager.defaultRelays())) return false
val sealed = try { encryptionService.sealCourierPayload(typedPayload, recipientNoiseKey) } catch (_: Exception) { return false }
val now = System.currentTimeMillis()
if (!relayManager.hasConnectedRelay(relayUrls)) return false
val sealed = try { cipher.seal(typedPayload, recipientNoiseKey) } catch (_: Exception) { return false }
val now = clock()
val envelope = CourierEnvelope(
recipientTag = CourierEnvelope.recipientTag(recipientNoiseKey, CourierEnvelope.epochDay(now)),
expiry = (now + CourierEnvelope.MAX_LIFETIME_MS).toULong(),
@ -103,7 +189,7 @@ class BridgeCourierService(
)
val encoded = envelope.encode() ?: return false
if (encoded.size > MAX_ENCODED_BYTES) return false
val identity = try { NostrIdentity.generate() } catch (_: Exception) { return false }
val identity = try { identityFactory() } catch (_: Exception) { return false }
val event = identity.signEvent(
NostrEvent(
pubkey = identity.publicKeyHex,
@ -116,7 +202,7 @@ class BridgeCourierService(
content = Base64.encodeToString(encoded, Base64.NO_WRAP)
)
)
return relayManager.sendEvent(event, NostrRelayManager.defaultRelays(), onAccepted = onAccepted)
return relayManager.sendEvent(event, relayUrls, onAccepted)
}
@Synchronized
@ -134,12 +220,12 @@ class BridgeCourierService(
val encoded = try { Base64.decode(event.content, Base64.DEFAULT) } catch (_: Exception) { return }
if (encoded.size > MAX_ENCODED_BYTES) return
val envelope = CourierEnvelope.decode(encoded) ?: return
val now = System.currentTimeMillis()
val now = clock()
if (envelope.expiry.toLong() <= now || envelope.expiry.toLong() > now + CourierEnvelope.MAX_LIFETIME_MS + 60 * 60 * 1000L) return
val eventTag = event.tags.firstOrNull { it.size > 1 && it[0] == "x" }?.get(1) ?: return
val expiration = event.tags.firstOrNull { it.size > 1 && it[0] == "expiration" }?.get(1)?.toULongOrNull() ?: return
if (eventTag != envelope.recipientTag.toHex() || expiration != envelope.expiry / 1000u) return
val localKey = encryptionService.getStaticPublicKey() ?: return
val localKey = cipher.staticPublicKey() ?: return
if (!envelope.matchesRecipient(localKey, now)) return
onEnvelope(envelope)
}

View File

@ -72,6 +72,9 @@ object ContactDirectory {
favorite != null -> favorite.peerNoisePublicKey
ContactIdentityResolver.isNoiseKeyHex(peerOrConversationID) ->
ContactIdentityResolver.bytesFromHex(peerOrConversationID)
contactFingerprint != null -> cachedNoiseKeyForFingerprint(contactFingerprint)
ContactIdentityResolver.isMeshPeerId(peerOrConversationID) ->
noiseKeyForAlias(peerOrConversationID)
else -> null
}
val liveMeshPeerID = contactFingerprint?.let { findLiveMeshPeerForFingerprint(it) }
@ -156,6 +159,18 @@ object ContactDirectory {
}
}
/**
* A contact conversation ID contains the SHA-256 fingerprint of the Noise key, whose
* first 16 hex characters are the stable mesh peer ID. Recovering that cached key keeps
* offline routing viable after the outbox has canonicalized a peer ID to `contact_…`.
* Recompute the fingerprint before returning it so stale or mismatched cache entries cannot
* redirect a private message.
*/
private fun cachedNoiseKeyForFingerprint(fingerprint: String): ByteArray? =
cachedNoiseKey(fingerprint.take(16))?.takeIf {
ContactIdentityResolver.fingerprintHex(it).equals(fingerprint, ignoreCase = true)
}
private fun cachedFingerprintNickname(fingerprint: String): String? {
val context = appContext ?: return null
return try {

View File

@ -411,7 +411,17 @@ class GossipSyncManager(
StandardCopyOption.REPLACE_EXISTING
)
} catch (_: Exception) {
temporary.delete()
// Some Android filesystems do not support ATOMIC_MOVE. Preserve the durable
// public-history guarantee with a regular replacement move before giving up.
try {
java.nio.file.Files.move(
temporary.toPath(),
file.toPath(),
StandardCopyOption.REPLACE_EXISTING
)
} catch (_: Exception) {
temporary.delete()
}
}
} catch (_: Exception) { }
}

View File

@ -7,6 +7,7 @@ import com.bitchat.android.model.BitchatFilePacket
import com.bitchat.android.model.NoisePayload
import com.bitchat.android.model.NoisePayloadType
import com.bitchat.android.model.PeerCapabilities
import com.bitchat.android.model.PrivateMessagePacket
import com.bitchat.android.model.RoutedPacket
import com.bitchat.android.noise.NoisePeerIdentity
import com.bitchat.android.noise.AuthenticatedNoiseSession
@ -25,9 +26,11 @@ import org.junit.Test
import org.junit.runner.RunWith
import org.mockito.kotlin.any
import org.mockito.kotlin.anyOrNull
import org.mockito.kotlin.argThat
import org.mockito.kotlin.eq
import org.mockito.kotlin.mock
import org.mockito.kotlin.never
import org.mockito.kotlin.times
import org.mockito.kotlin.verify
import org.mockito.kotlin.whenever
import org.robolectric.RobolectricTestRunner
@ -435,6 +438,48 @@ class MessageHandlerTest {
assertTrue(handler.handleNoiseEncrypted(RoutedPacket(encryptedPacket(), peerID, "direct-link")))
}
@Test
fun `opened courier private message is admitted as its Noise sender`() = runBlocking {
whenever(delegate.getPeerNickname(peerID)).thenReturn(nickname)
whenever(delegate.getMyNickname()).thenReturn("me")
val payload = NoisePayload(
NoisePayloadType.PRIVATE_MESSAGE,
requireNotNull(PrivateMessagePacket("courier-message", "opaque courier content").encode())
).encode()
assertTrue(
handler.handleOpenedCourierPayload(
RoutedPacket(encryptedPacket().copy(payload = payload), peerID, "courier-ingress")
)
)
verify(delegate).onMessageReceived(argThat {
id == "courier-message" &&
content == "opaque courier content" &&
senderPeerID == peerID &&
sender == nickname &&
recipientNickname == "me"
})
}
@Test
fun `opened courier delivery receipt acknowledges the original sender and rejects blank IDs`() = runBlocking {
val receipt = NoisePayload(NoisePayloadType.DELIVERED, "courier-message".toByteArray()).encode()
assertTrue(
handler.handleOpenedCourierPayload(
RoutedPacket(encryptedPacket().copy(payload = receipt), peerID, "courier-ingress")
)
)
verify(delegate).onDeliveryAckReceived("courier-message", peerID)
val blankReceipt = NoisePayload(NoisePayloadType.DELIVERED, ByteArray(0)).encode()
assertFalse(
handler.handleOpenedCourierPayload(
RoutedPacket(encryptedPacket().copy(payload = blankReceipt), peerID, "courier-ingress")
)
)
verify(delegate, times(1)).onDeliveryAckReceived("courier-message", peerID)
}
private fun encryptedPacket(): BitchatPacket = BitchatPacket(
version = 1u,
type = MessageType.NOISE_ENCRYPTED.value,

View File

@ -0,0 +1,240 @@
package com.bitchat.android.nostr
import com.bitchat.android.model.CourierEnvelope
import com.bitchat.android.model.NoisePayload
import com.bitchat.android.model.NoisePayloadType
import com.bitchat.android.model.PrivateMessagePacket
import org.junit.Assert.assertEquals
import org.junit.Assert.assertFalse
import org.junit.Assert.assertNotNull
import org.junit.Assert.assertTrue
import org.junit.Test
import java.util.Base64
class BridgeCourierServiceTest {
private val now = 1_750_000_000_000L
private val relays = listOf("wss://bridge-test.invalid")
private val recipientKey = ByteArray(32) { 0x32 }
@Test
fun `deposit creates a signed iOS-compatible event and receiver admits it once`() {
val senderRelay = FakeRelay()
val sender = BridgeCourierService(
cipher = FakeCipher(ByteArray(32) { 0x11 }),
onEnvelope = {},
relayManager = senderRelay,
relayUrls = relays,
clock = { now },
identityFactory = { NostrIdentity.fromSeed("bridge-courier-sender") }
)
var accepted = false
val event = try {
assertTrue(sender.deposit("synthetic bridge payload", "bridge-message", recipientKey) {
accepted = true
})
assertTrue(accepted)
assertEquals(1, senderRelay.sentEvents.size)
senderRelay.sentEvents.single().also {
assertEquals(1401, it.kind)
assertTrue(it.isValidSignature())
assertEquals(relays, senderRelay.lastSendRelays)
}
} finally {
sender.stop()
}
val encoded = Base64.getDecoder().decode(event.content)
val envelope = requireNotNull(CourierEnvelope.decode(encoded))
assertTrue(envelope.matchesRecipient(recipientKey, now))
assertEquals(1u.toUByte(), envelope.copies)
assertEquals(
listOf("x", envelope.recipientTag.toHex()),
event.tags.first { it.firstOrNull() == "x" }
)
assertEquals(
listOf("expiration", (envelope.expiry / 1000u).toString()),
event.tags.first { it.firstOrNull() == "expiration" }
)
val typed = requireNotNull(NoisePayload.decode(envelope.ciphertext))
assertEquals(NoisePayloadType.PRIVATE_MESSAGE, typed.type)
val privateMessage = requireNotNull(PrivateMessagePacket.decode(typed.data))
assertEquals("bridge-message", privateMessage.messageID)
assertEquals("synthetic bridge payload", privateMessage.content)
val receiverRelay = FakeRelay()
val received = mutableListOf<CourierEnvelope>()
val receiver = BridgeCourierService(
cipher = FakeCipher(recipientKey),
onEnvelope = received::add,
relayManager = receiverRelay,
relayUrls = relays,
clock = { now },
identityFactory = { NostrIdentity.fromSeed("bridge-courier-receiver") }
)
try {
receiver.start()
assertNotNull(receiverRelay.subscription)
assertTrue(receiverRelay.subscription!!.filter.matches(event))
assertEquals(relays, receiverRelay.subscription!!.relayUrls)
receiverRelay.emit(event)
receiverRelay.emit(event)
assertEquals(listOf(envelope), received)
} finally {
receiver.stop()
}
assertEquals(1, receiverRelay.unsubscribedIDs.size)
}
@Test
fun `receiver rejects a validly signed event with mismatched metadata and deposit fails closed`() {
val relay = FakeRelay()
val received = mutableListOf<CourierEnvelope>()
val receiver = BridgeCourierService(
cipher = FakeCipher(recipientKey),
onEnvelope = received::add,
relayManager = relay,
relayUrls = relays,
clock = { now },
identityFactory = { NostrIdentity.fromSeed("bridge-courier-receiver") }
)
val envelope = CourierEnvelope(
recipientTag = CourierEnvelope.recipientTag(recipientKey, CourierEnvelope.epochDay(now)),
expiry = (now + CourierEnvelope.MAX_LIFETIME_MS).toULong(),
ciphertext = byteArrayOf(1, 2, 3),
copies = 1u
)
val identity = NostrIdentity.fromSeed("bridge-courier-invalid-event")
val mismatchedTag = identity.signEvent(
NostrEvent(
pubkey = identity.publicKeyHex,
createdAt = (now / 1000).toInt(),
kind = 1401,
tags = listOf(
listOf("x", "00".repeat(CourierEnvelope.TAG_LENGTH)),
listOf("expiration", (envelope.expiry / 1000u).toString())
),
content = Base64.getEncoder().encodeToString(requireNotNull(envelope.encode()))
)
)
try {
receiver.start()
relay.emit(mismatchedTag)
assertTrue(received.isEmpty())
val oversizedSender = BridgeCourierService(
cipher = FakeCipher(ByteArray(32) { 0x21 }, ByteArray(CourierEnvelope.MAX_CIPHERTEXT_BYTES + 1)),
onEnvelope = {},
relayManager = relay,
relayUrls = relays,
clock = { now },
identityFactory = { NostrIdentity.fromSeed("bridge-courier-oversized") }
)
try {
assertFalse(oversizedSender.depositPayload(byteArrayOf(7), recipientKey))
relay.connected = false
assertFalse(oversizedSender.depositPayload(byteArrayOf(7), recipientKey))
} finally {
oversizedSender.stop()
}
} finally {
receiver.stop()
}
}
@Test
fun `deposit completion waits for relay acceptance`() {
val relay = FakeRelay().apply { acceptImmediately = false }
val sender = BridgeCourierService(
cipher = FakeCipher(ByteArray(32) { 0x11 }),
onEnvelope = {},
relayManager = relay,
relayUrls = relays,
clock = { now },
identityFactory = { NostrIdentity.fromSeed("bridge-courier-pending") }
)
var accepted = 0
try {
assertTrue(sender.depositPayload(byteArrayOf(1, 2, 3), recipientKey) { accepted++ })
assertEquals(0, accepted)
relay.acceptPendingEvent()
assertEquals(1, accepted)
} finally {
sender.stop()
}
}
private class FakeCipher(
private val staticKey: ByteArray,
private val sealedOverride: ByteArray? = null
) : BridgeCourierCipher {
override fun staticPublicKey(): ByteArray = staticKey
override fun seal(payload: ByteArray, recipientNoiseKey: ByteArray): ByteArray =
sealedOverride ?: payload.copyOf()
}
private class FakeRelay : BridgeCourierRelay {
data class Subscription(
val filter: NostrFilter,
val id: String,
val relayUrls: List<String>,
val handler: (NostrEvent) -> Unit
)
var connected = true
var acceptImmediately = true
var subscription: Subscription? = null
val sentEvents = mutableListOf<NostrEvent>()
var lastSendRelays: List<String> = emptyList()
val unsubscribedIDs = mutableListOf<String>()
private var pendingAccepted: (() -> Unit)? = null
override fun subscribe(
filter: NostrFilter,
id: String,
targetRelayUrls: List<String>,
handler: (NostrEvent) -> Unit
) {
subscription = Subscription(filter, id, targetRelayUrls, handler)
}
override fun unsubscribe(id: String) {
unsubscribedIDs += id
if (subscription?.id == id) subscription = null
}
override fun hasConnectedRelay(relayUrls: Collection<String>): Boolean = connected
override fun sendEvent(
event: NostrEvent,
relayUrls: List<String>,
onAccepted: () -> Unit
): Boolean {
if (!connected) return false
sentEvents += event
lastSendRelays = relayUrls
if (acceptImmediately) onAccepted() else pendingAccepted = onAccepted
return true
}
fun acceptPendingEvent() {
val callback = requireNotNull(pendingAccepted)
pendingAccepted = null
callback()
}
fun emit(event: NostrEvent) {
subscription?.handler?.invoke(event)
}
}
private fun ByteArray.toHex(): String = joinToString("") { "%02x".format(it) }
}

View File

@ -6,6 +6,7 @@ import com.bitchat.android.identity.SecureIdentityStateManager
import org.junit.After
import org.junit.Assert.assertEquals
import org.junit.Assert.assertNull
import org.junit.Assert.assertTrue
import org.junit.Before
import org.junit.Test
import org.junit.runner.RunWith
@ -56,4 +57,16 @@ class ContactDirectoryTest {
assertNull(resolution.displayName)
}
@Test
fun `offline contact resolves a fingerprint-validated cached Noise key`() {
val noiseKey = ByteArray(32) { it.toByte() }
val fingerprint = ContactIdentityResolver.fingerprintHex(noiseKey)
val peerID = ContactIdentityResolver.peerIdForNoiseKey(noiseKey)
identityManager.cachePeerNoiseKey(peerID, ContactIdentityResolver.noiseKeyHex(noiseKey))
val resolution = ContactDirectory.resolve("contact_$fingerprint")
assertTrue(resolution.noisePublicKey!!.contentEquals(noiseKey))
}
}

View File

@ -14,6 +14,7 @@ import org.junit.Test
import org.junit.runner.RunWith
import org.mockito.kotlin.any
import org.mockito.kotlin.anyOrNull
import org.mockito.kotlin.argThat
import org.mockito.kotlin.clearInvocations
import org.mockito.kotlin.eq
import org.mockito.kotlin.mock
@ -36,6 +37,7 @@ class MessageRouterTest {
private lateinit var mesh: MeshService
private lateinit var router: MessageRouter
private lateinit var identityManager: SecureIdentityStateManager
private var fakeTime = 1_000_000L
private val expired = mutableListOf<String>()
@ -46,7 +48,7 @@ class MessageRouterTest {
"message-router-test-${UUID.randomUUID()}",
Context.MODE_PRIVATE
)
val identityManager = SecureIdentityStateManager(prefs, testOnly = true)
identityManager = SecureIdentityStateManager(prefs, testOnly = true)
ContactDirectory.identityManagerProvider = { identityManager }
mesh = mock()
@ -213,6 +215,117 @@ class MessageRouterTest {
verify(mesh).sendPrivateMessage("durable", peerID, "peer", "msg-durable")
}
@Test
fun `offline cached contact deposits once with a verified peer courier`() {
val recipientNoiseKey = ByteArray(32) { 0x2A }
val recipientPeerID = ContactIdentityResolver.peerIdForNoiseKey(recipientNoiseKey)
val courierPeerID = "9999aaaabbbbcccc"
val courierNoiseKey = ByteArray(32) { 0x3B }
identityManager.cachePeerNoiseKey(
recipientPeerID,
ContactIdentityResolver.noiseKeyHex(recipientNoiseKey)
)
whenever(mesh.getPeerInfo(recipientPeerID)).thenReturn(
PeerInfo(
id = recipientPeerID,
nickname = "offline contact",
isConnected = false,
isDirectConnection = false,
noisePublicKey = recipientNoiseKey,
signingPublicKey = ByteArray(32) { 0x0A },
isVerifiedNickname = false,
lastSeen = fakeTime
)
)
whenever(mesh.getPeerInfos()).thenReturn(
listOf(
PeerInfo(
id = courierPeerID,
nickname = "verified courier",
isConnected = true,
isDirectConnection = true,
noisePublicKey = courierNoiseKey,
signingPublicKey = ByteArray(32) { 0x0C },
isVerifiedNickname = false,
lastSeen = fakeTime,
hasVerifiedAnnouncement = true
)
)
)
whenever(mesh.sendCourierMessage(any(), any(), any(), any())).thenReturn(listOf(courierPeerID))
val result = router.sendPrivate("courier payload", recipientPeerID, "offline contact", "msg-courier")
assertEquals(MessageRouter.RouteResult.QUEUED, result)
verify(mesh).sendCourierMessage(
eq("courier payload"),
eq("msg-courier"),
argThat { contentEquals(recipientNoiseKey) },
eq(listOf(courierPeerID))
)
router.tickOutbox()
verify(mesh, times(1)).sendCourierMessage(
eq("courier payload"),
eq("msg-courier"),
argThat { contentEquals(recipientNoiseKey) },
eq(listOf(courierPeerID))
)
}
@Test
fun `courier acknowledgement clears an offline contact outbox entry before direct reconnect`() {
val recipientNoiseKey = ByteArray(32) { 0x4A }
val recipientPeerID = ContactIdentityResolver.peerIdForNoiseKey(recipientNoiseKey)
identityManager.cachePeerNoiseKey(
recipientPeerID,
ContactIdentityResolver.noiseKeyHex(recipientNoiseKey)
)
whenever(mesh.getPeerNicknames()).thenReturn(mapOf(recipientPeerID to "offline contact"))
whenever(mesh.getPeerInfo(recipientPeerID)).thenReturn(
PeerInfo(
id = recipientPeerID,
nickname = "offline contact",
isConnected = false,
isDirectConnection = false,
noisePublicKey = recipientNoiseKey,
signingPublicKey = ByteArray(32) { 0x0A },
isVerifiedNickname = false,
lastSeen = fakeTime
)
)
assertEquals(
MessageRouter.RouteResult.QUEUED,
router.sendPrivate("courier payload", recipientPeerID, "offline contact", "msg-courier-ack")
)
router.onMessageAcknowledged("msg-courier-ack", recipientPeerID)
whenever(mesh.getPeerInfo(recipientPeerID)).thenReturn(
PeerInfo(
id = recipientPeerID,
nickname = "offline contact",
isConnected = true,
isDirectConnection = true,
noisePublicKey = recipientNoiseKey,
signingPublicKey = ByteArray(32) { 0x0A },
isVerifiedNickname = false,
lastSeen = fakeTime
)
)
whenever(mesh.hasEstablishedSession(recipientPeerID)).thenReturn(true)
router.tickOutbox()
verify(mesh, never()).sendPrivateMessage(
eq("courier payload"),
eq(recipientPeerID),
any(),
eq("msg-courier-ack")
)
}
@Test
fun `outbox stops transport retries after eight attempts`() {
peerReady()

View File

@ -1,5 +1,6 @@
package com.bitchat.android.sync
import android.content.ContextWrapper
import android.os.Build
import com.bitchat.android.model.RequestSyncPacket
import com.bitchat.android.protocol.BitchatPacket
@ -9,7 +10,10 @@ import org.junit.Assert.assertEquals
import org.junit.Test
import org.junit.runner.RunWith
import org.robolectric.RobolectricTestRunner
import org.robolectric.RuntimeEnvironment
import org.robolectric.annotation.Config
import java.io.File
import java.nio.file.Files
@RunWith(RobolectricTestRunner::class)
@Config(sdk = [Build.VERSION_CODES.P], manifest = Config.NONE)
@ -147,6 +151,59 @@ class GossipSyncManagerTest {
assertEquals(listOf(1), sent.map { it.payload.single().toInt() })
}
@Test
fun `public history survives manager recreation for post-reconnect sync`() {
val filesDir = Files.createTempDirectory("gossip-sync-test-").toFile()
val context = object : ContextWrapper(RuntimeEnvironment.getApplication()) {
override fun getApplicationContext() = this
override fun getFilesDir(): File = filesDir
}
try {
val now = System.currentTimeMillis()
GossipSyncManager("1111222233334444", TestScope(), config, context)
.onPublicPacketSeen(packet(MessageType.MESSAGE, now, 42))
val sent = mutableListOf<BitchatPacket>()
val restored = GossipSyncManager("1111222233334444", TestScope(), config, context)
restored.delegate = recordingDelegate(sent)
restored.handleRequestSync(
"peer",
RequestSyncPacket(
p = 1,
m = 1,
data = byteArrayOf(),
types = SyncTypeFlags.PUBLIC_MESSAGES
)
)
assertEquals(listOf(42), sent.map { it.payload.single().toInt() })
} finally {
filesDir.deleteRecursively()
}
}
@Test
fun `announcement-only request returns the latest announcement independently of message history`() {
val sent = mutableListOf<BitchatPacket>()
val manager = GossipSyncManager("1111222233334444", TestScope(), config)
manager.delegate = recordingDelegate(sent)
val now = System.currentTimeMillis()
manager.onPublicPacketSeen(packet(MessageType.ANNOUNCE, now - 1_000, 6))
manager.onPublicPacketSeen(packet(MessageType.MESSAGE, now, 7))
manager.handleRequestSync(
"peer",
RequestSyncPacket(
p = 1,
m = 1,
data = byteArrayOf(),
types = SyncTypeFlags.ANNOUNCE
)
)
assertEquals(listOf(6), sent.map { it.payload.single().toInt() })
}
private fun recordingDelegate(sent: MutableList<BitchatPacket>) =
object : GossipSyncManager.Delegate {
override fun sendPacket(packet: BitchatPacket) = Unit

124
docs/PR-977-TEST.md Normal file
View File

@ -0,0 +1,124 @@
# PR-977 test notes
This note records the test coverage and findings added while validating the
expanded store-and-forward work. All test content and physical-device evidence
used synthetic data. Raw device evidence remains local and is not committed.
## Coverage matrix
| Capability | Automated coverage | Physical Mesh Lab coverage | Result |
|---|---|---|---|
| Durable private-message outbox | `MessageOutboxStoreTest`, `MessageRouterTest` | `durable_outbox`: recipient offline, sender process death, then delivery after reconnection | Pass |
| Direct and peer-courier routing | `CourierEnvelopeTest`, `CourierStoreTest`, `DirectCourierDepositPolicyTest`, `MessageRouterTest` | `courier_contract` | Pass for two-phone contracts; three-party delivery remains blocked below |
| Courier Noise X confidentiality and sender authentication | `NoiseCourierTest`, `BridgeCourierServiceTest` | N/A | Pass |
| Courier delivery acknowledgements | `MessageHandlerTest`, `MessageRouterTest` | Three-party courier delivery pending | Automated pass; physical blocked below |
| Courier fragmentation and custody commit | `FragmentingPacketSenderTest`, `CourierStoreTest` | `courier_contract` | Pass |
| Nostr kind-1401 bridge wire contract | `BridgeCourierServiceTest` with a deterministic in-memory relay | N/A | Pass |
| Relay acceptance before bridge completion | `BridgeCourierServiceTest` | N/A | Pass |
| Public-history retention and recovery | `GossipSyncManagerTest` | `sync_auto_recovery` and `sync_recovery` | Pass |
| Public file and fragment recovery | `GossipSyncManagerTest` | `sync_file_recovery` with a synthetic digest | Pass |
| Ordinary mesh regressions | Existing unit suite | `broadcast`, `dm`, `favorite_verification` | Pass |
The deterministic bridge test checks signed kind-1401 events, recipient and
expiry tags, receiver filtering/deduplication, size limits, rejection of
inconsistent metadata, and the rule that the outbox completion callback waits
for relay acceptance. `CourierEnvelopeTest` separately covers rotating recipient
tags. No test connects to public relays.
## Commands completed
- `./gradlew --no-daemon :app:testDebugUnitTest --max-workers=1`
- `./gradlew --no-daemon :wear:testDebugUnitTest --max-workers=1`
- `./gradlew --no-daemon lintDebug clientRewriteContractTest --max-workers=1`
- `./gradlew --no-daemon :app:assembleDebug --max-workers=1`
- `python3 -m py_compile tools/release_gate/mesh_lab.py`
- `git diff --check`
## Physical Mesh Lab results
The completed scenarios used two authorized phones in logical A/B roles and a
current debug build. They passed with synthetic message and file content:
- `broadcast`
- `dm`
- `favorite_verification`
- `courier_contract`
- `durable_outbox`
- `sync_recovery`
- `sync_auto_recovery`
- `sync_file_recovery`
`durable_outbox` exercises the production `MessageRouter` path, queues while
the recipient is actually offline, force-stops the sender, recreates the
router, and verifies receiver-side delivery only after reconnection.
`sync_auto_recovery` intentionally does not call the debug sync-request hook.
`sync_file_recovery` verifies receiver-side digest equality after replaying a
missed public file and its fragments.
## Findings
### Fixed: offline known contacts could skip peer-courier deposit
**Impact:** An offline known contact could resolve to its canonical contact ID
without recovering its cached Noise public key. The router then had no
recipient key and skipped peer-courier deposit entirely.
**Fix:** Contact resolution now recovers only a fingerprint-validated cached
Noise key for that canonical contact. The `ContactDirectoryTest` and
`MessageRouterTest` regressions cover resolution and the resulting single
courier deposit.
### Fixed: public-history archive could be lost on filesystems without atomic move
**Impact:** Persisting the public gossip archive silently discarded the
temporary archive when atomic rename was unsupported, defeating recovery after
restart on those filesystems.
**Fix:** Archive persistence now falls back to a normal replace-existing move
only when atomic move is unavailable. `GossipSyncManagerTest` verifies history
survives manager recreation.
### Fixed: Mesh Lab direct-link assertion produced a false failure
**Impact:** The runner required each endpoint to independently establish a
GATT client connection. One direct GATT link is bidirectional, so a healthy
link could be incorrectly reported as failed after one endpoint restarted.
**Fix:** `ensure_direct_link` now accepts a direct link observed from either
endpoint.
### Test-harness isolation improvement: debug transport toggles alone were insufficient
**Observation:** Disabling debug transports did not always make a counterpart
observe link loss promptly enough for an offline-routing test.
**Change:** Offline scenarios now force-stop the test app and temporarily turn
off the Bluetooth adapter before asserting the recipient is absent. This is a
test-isolation improvement; it is not classified as a production defect.
### Test infrastructure finding: concurrent test worker crashed before assertions completed
**Observation:** The initial combined Gradle invocation ended with a JVM-level
test-worker crash during the app unit-test task, before any test assertion
failure was reported.
**Resolution:** Re-running the app and Wear unit suites serially in fresh,
single-use Gradle processes passed. This is recorded as local test-runtime
instability, not as a product defect. The raw crash diagnostic remains local
and is not committed.
## Remaining coverage limits
- `courier_delivery` is implemented as a three-phone A→B→C scenario: C is
offline for the deposit, A then goes offline, and C returns to receive only
through B. It is **blocked (not run)** because only two authorized phones are
currently attached. A third phone is required; reusing C offline/online is
already part of the scenario.
- No public Nostr relay or desktop courier bridge was used. The bridge has a
deterministic in-memory relay contract test, but live desktop/relay
interoperability remains **not run**.
- Android/iOS live courier interoperability remains **not run**.
No unresolved product defects were found in the completed automated and
two-phone physical coverage.

View File

@ -299,6 +299,10 @@ python3 tools/release_gate/mesh_lab.py scenario all \
| `favorite_verification` | favorite signal, orange-outline/filled mutual state, and peer fingerprint verification |
| `broadcast` | public mesh message A→B |
| `sync_recovery` | message missed while B's BLE and Wi-Fi Aware transports are disabled is recovered by the gossip sync request after reconnection |
| `sync_auto_recovery` | message missed while B is offline is recovered after restart by the production gossip scheduler, without a test-hook sync request |
| `sync_file_recovery` | broadcast file missed while B is offline is recovered through sync; validates the synthetic file digest and fragment replay |
| `durable_outbox` | a router-queued DM survives sender process death while the recipient is offline, then reaches the recipient after reconnection |
| `courier_delivery` | three-phone sender→courier→recipient handoff: recipient is offline for deposit and sender remains offline for recipient pickup |
| `courier_contract` | on-device courier wire/store contract: prekey TLV retention, split reservations, cancellation, reverse commits, and persisted spray history |
| `ptt_dm` | Noise-encrypted 440 Hz PTT in both directions; asserts real-time capture, zero sequence gaps, decoded PCM duration/energy/continuity, and finalized-note absorption |
| `ptt_broadcast` | signed public 440 Hz PTT with the same bidirectional packet and decoded-audio quality assertions |
@ -309,7 +313,7 @@ python3 tools/release_gate/mesh_lab.py scenario all \
| `raw` | raw packet injection is accepted by the mesh |
| `session_recovery` | force-stop B mid-session: identity persists, re-handshake, DMs flow again |
| `identity_reset` | pm clear B mid-session: new identity, rediscovery, handshake, DMs |
| `all` | every scenario above in sequence |
| `all` | every supported two-phone scenario in sequence, plus `courier_delivery` when a third phone is supplied |
Each run writes `<scenario>-evidence.json` to `--out` (digests, timings,
session states, logcat excerpts on failure) and exits non-zero on failure.
@ -328,10 +332,11 @@ python3 tools/release_gate/mesh_lab.py cmd --serial <serial> state # full mesh
See `TestHookDriver.kt` for the full command set (`ping`, `start`, `stop`,
`whoami`, `set_nickname`, `scan`, `peers`, `connect`, `handshake`, `session`,
`announce`, `broadcast_msg`, `dm_send`, `dm_recv`, `msg_recv`, `favorite_set`,
`announce`, `broadcast_msg`, `dm_send`, `router_private_send`, `router_resume`,
`dm_recv`, `msg_recv`, `favorite_set`,
`favorite_status`, `verification_set`, `verification_status`, `file_send`,
`file_recv`, `file_cancel`, `ptt_send`, `ptt_recv`, `raw_send`, `courier_contract`,
`sync_request`, `ble`, `wifi_aware`, `state`, `clear_results`).
`cache_peer_identity`, `sync_request`, `ble`, `wifi_aware`, `state`, `clear_results`).
`msg_recv` accepts `include_existing=true` for deterministic backfill checks when
the packet's original timestamp predates the receive hook; use a unique synthetic

View File

@ -166,11 +166,14 @@ class Device:
def reset_bluetooth(self) -> None:
"""Cycle the BT adapter; clears zombie GATT connections from peer restarts."""
_shell(self.serial, "svc bluetooth disable")
self.disable_bluetooth()
time.sleep(2)
_shell(self.serial, "svc bluetooth enable")
time.sleep(3)
def disable_bluetooth(self) -> None:
_shell(self.serial, "svc bluetooth disable")
def enable_bluetooth(self) -> None:
subprocess.run(
[find_adb(), "-s", self.serial, "shell", "svc", "bluetooth", "enable"],
@ -365,6 +368,31 @@ def setup_pair(
wait_for_mutual_discovery(a, b)
def setup_triad(
a: Device,
b: Device,
c: Device,
apk: Path | None,
nickname_a: str,
nickname_b: str,
nickname_c: str,
) -> None:
"""Install and isolate three phones, then ensure every pair can discover each other."""
for device, nickname in ((a, nickname_a), (b, nickname_b), (c, nickname_c)):
device.reset_bluetooth()
device.enable_bluetooth()
if apk is not None:
device.install(apk)
device.clear_app_data()
device.grant_permissions()
device.wake()
device.launch()
device.cmd_ok("start")
device.cmd_ok("set_nickname", name=nickname)
for left, right in ((a, b), (a, c), (b, c)):
wait_for_mutual_discovery(left, right)
def whoami(device: Device) -> dict:
return device.cmd_ok("whoami")
@ -391,6 +419,47 @@ def wait_for_mutual_discovery(a: Device, b: Device) -> None:
fb.result()
def wait_for_peer_absent(device: Device, peer_id: str, timeout_s: int = 90) -> None:
"""Wait until the peer is absent or no longer connected in the visible mesh state."""
deadline = time.monotonic() + timeout_s
while time.monotonic() < deadline:
peers = device.cmd_ok("peers").get("peers", [])
peer = next((item for item in peers if item.get("id") == peer_id), None)
if peer is None or not peer.get("connected", False):
return
time.sleep(2)
raise MeshLabError(f"[{device.alias}] peer {peer_id} remained connected after transport shutdown")
def disable_transports(device: Device) -> bool:
"""Make a phone unreachable through either local mesh transport and return its Wi-Fi setting."""
device.cmd_ok("ble", enabled=False)
return bool(device.cmd_ok("wifi_aware", enabled=False)["previous_enabled"])
def resume_transports(device: Device, wifi_enabled: bool) -> None:
"""Bring a force-stopped phone back as the same identity with its prior Wi-Fi setting."""
device.wake()
device.launch()
device.cmd_ok("start")
device.cmd_ok("ble", enabled=True)
device.cmd_ok("wifi_aware", enabled=wifi_enabled)
def take_device_offline(device: Device) -> bool:
"""Stop the app and radio so the counterpart must observe a real link loss."""
wifi_enabled = disable_transports(device)
device.force_stop()
device.disable_bluetooth()
return wifi_enabled
def bring_device_online(device: Device, wifi_enabled: bool) -> None:
device.enable_bluetooth()
time.sleep(3)
resume_transports(device, wifi_enabled)
# MARK: - scenarios
def scenario_dm(a: Device, b: Device) -> dict:
@ -585,6 +654,233 @@ def scenario_sync_recovery(a: Device, b: Device) -> dict:
pass
def scenario_sync_auto_recovery(a: Device, b: Device) -> dict:
"""A restarted receiver recovers a missed public message without a test-hook sync request."""
id_a = whoami(a)["peer_id"]
id_b = whoami(b)["peer_id"]
token = f"sync-auto-{uuid.uuid4().hex[:8]}"
wifi_before = disable_transports(b)
try:
send_result = a.cmd_ok("broadcast_msg", 30_000, content=f"automatic sync recovery {token}")
time.sleep(2)
offline_result = b.cmd("msg_recv", 5_000, contains=token, include_existing=True)
if offline_result.get("status") == "ok":
raise MeshLabError(f"message was delivered while receiver transports were disabled: {offline_result}")
b.force_stop()
resume_transports(b, wifi_before)
with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool:
recv = pool.submit(
b.cmd_ok,
"msg_recv",
180_000,
contains=token,
include_existing=True,
)
time.sleep(2)
ensure_direct_link(a, b, id_a, id_b)
# No sync_request command here: this must arrive from the production initial/periodic
# gossip scheduler created by the restarted receiver.
recv_result = recv.result()
assert recv_result["from"] == id_a, recv_result
return {
"send": send_result,
"offline_delivery": {"status": offline_result.get("status", "missing")},
"synced_recv": recv_result,
}
finally:
try:
bring_device_online(b, wifi_before)
except MeshLabError:
pass
def scenario_sync_file_recovery(a: Device, b: Device) -> dict:
"""Sync an offline-missed broadcast transfer, exercising FILE_TRANSFER and FRAGMENT replay."""
id_a = whoami(a)["peer_id"]
id_b = whoami(b)["peer_id"]
fixture = make_fixtures(
Path(tempfile.mkdtemp(prefix="meshlab-sync-file-")),
names=["small_1k.bin"],
)["small_1k.bin"]
name = f"sync-file-{uuid.uuid4().hex[:8]}.bin"
remote = a.push_fixture(fixture["path"], name=name)
b.clear_incoming()
wifi_before = disable_transports(b)
try:
send_result = a.cmd_ok("file_send", 180_000, path=remote)
time.sleep(2)
offline_result = b.cmd("file_recv", 10_000, name_contains=name)
if offline_result.get("status") == "ok":
raise MeshLabError(f"file was delivered while receiver transports were disabled: {offline_result}")
b.force_stop()
resume_transports(b, wifi_before)
with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool:
recv = pool.submit(b.cmd_ok, "file_recv", 240_000, name_contains=name)
time.sleep(2)
ensure_direct_link(a, b, id_a, id_b)
b.cmd_ok("sync_request", peer=id_a)
recv_result = recv.result()
if recv_result["sha256"] != fixture["sha256"]:
raise MeshLabError("synced file digest did not match the synthetic fixture")
return {
"send": send_result,
"offline_delivery": {"status": offline_result.get("status", "missing")},
"synced_recv": {
"name": recv_result["name"],
"bytes": recv_result["bytes"],
"digest_match": True,
},
}
finally:
try:
b.cmd_ok("ble", enabled=True)
except MeshLabError:
pass
try:
b.cmd_ok("wifi_aware", enabled=wifi_before)
except MeshLabError:
pass
def scenario_durable_outbox(a: Device, b: Device) -> dict:
"""Queue a private message offline, kill the sender, and deliver it after both phones return."""
id_a = whoami(a)["peer_id"]
id_b = whoami(b)["peer_id"]
ensure_direct_link(a, b, id_a, id_b)
force_handshake(a, id_b)
force_handshake(b, id_a)
a.cmd_ok("cache_peer_identity", peer=id_b)
token = f"durable-outbox-{uuid.uuid4().hex[:8]}"
message_id = f"outbox-{uuid.uuid4().hex}"
wifi_before = take_device_offline(b)
try:
wait_for_peer_absent(a, id_b)
queued: dict | None = None
for _attempt in range(5):
candidate = a.cmd_ok(
"router_private_send",
peer=id_b,
content=f"durable outbox {token}",
msg_id=message_id,
)
if candidate.get("route") == "QUEUED":
queued = candidate
break
time.sleep(3)
if queued is None:
raise MeshLabError("private message did not enter the durable outbox while recipient was offline")
# The queued entry must survive a complete sender process death, not just an in-memory
# reconnect. B is restored before A so its receiver is ready for the resumed delivery.
a.force_stop()
bring_device_online(b, wifi_before)
a.wake()
a.launch()
a.cmd_ok("start")
a.cmd_ok("router_resume")
with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool:
recv = pool.submit(b.cmd_ok, "dm_recv", 180_000, peer=id_a, contains=token)
time.sleep(2)
ensure_direct_link(a, b, id_a, id_b)
force_handshake(a, id_b)
force_handshake(b, id_a)
recv_result = recv.result()
assert recv_result["from"] == id_a, recv_result
return {
"queued_route": queued["route"],
"sender_process_restarted": True,
"received_after_restart": {"msg_id": recv_result["msg_id"], "from_sender": True},
}
finally:
try:
bring_device_online(b, wifi_before)
except MeshLabError:
pass
def _make_mutual_favorites(a: Device, b: Device, id_a: str, id_b: str) -> None:
"""Establish a trusted direct A↔B relationship for a courier deposit."""
ensure_direct_link(a, b, id_a, id_b)
force_handshake(a, id_b)
force_handshake(b, id_a)
a.cmd_ok("favorite_set", peer=id_b, enabled=True)
b.cmd_ok("favorite_set", peer=id_a, enabled=True)
deadline = time.monotonic() + 45
while time.monotonic() < deadline:
a_status = a.cmd_ok("favorite_status", peer=id_b)
b_status = b.cmd_ok("favorite_status", peer=id_a)
if a_status.get("is_mutual") and b_status.get("is_mutual"):
return
time.sleep(1)
raise MeshLabError("courier depositor and courier did not establish a mutual favorite relationship")
def scenario_courier_delivery(a: Device, b: Device, c: Device) -> dict:
"""A deposits to B while C is offline; C returns only after A is unavailable."""
id_a = whoami(a)["peer_id"]
id_b = whoami(b)["peer_id"]
id_c = whoami(c)["peer_id"]
_make_mutual_favorites(a, b, id_a, id_b)
ensure_direct_link(a, c, id_a, id_c)
a.cmd_ok("cache_peer_identity", peer=id_c)
token = f"courier-delivery-{uuid.uuid4().hex[:8]}"
message_id = f"courier-{uuid.uuid4().hex}"
wifi_c = take_device_offline(c)
wifi_a: bool | None = None
try:
wait_for_peer_absent(a, id_c)
ensure_direct_link(a, b, id_a, id_b)
queued: dict | None = None
for _attempt in range(5):
candidate = a.cmd_ok(
"router_private_send",
peer=id_c,
content=f"courier delivery {token}",
msg_id=message_id,
)
if candidate.get("route") == "QUEUED":
queued = candidate
break
time.sleep(3)
if queued is None:
raise MeshLabError("offline recipient message did not enter the courier-capable outbox")
# A must be unavailable when C returns. This prevents the sender's direct outbox retry
# from masking whether B actually retained and handed over the courier envelope.
wifi_a = take_device_offline(a)
wait_for_peer_absent(b, id_a)
bring_device_online(c, wifi_c)
with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool:
recv = pool.submit(c.cmd_ok, "dm_recv", 240_000, peer=id_a, contains=token)
time.sleep(2)
ensure_direct_link(b, c, id_b, id_c)
c.cmd_ok("announce")
recv_result = recv.result()
assert recv_result["from"] == id_a, recv_result
return {
"queued_route": queued["route"],
"sender_offline_before_recipient_return": True,
"received_from_original_sender": True,
"received_message_id": recv_result["msg_id"],
}
finally:
try:
bring_device_online(c, wifi_c)
except MeshLabError:
pass
if wifi_a is not None:
try:
bring_device_online(a, wifi_a)
except MeshLabError:
pass
def scenario_courier_contract(a: Device, b: Device) -> dict:
"""Run the real courier wire/store contract on device A's debug build."""
result = a.cmd_ok("courier_contract", 60_000)
@ -801,26 +1097,26 @@ def ensure_direct_link(a: Device, b: Device, id_a: str, id_b: str) -> None:
wait_for_peer(a, id_b, timeout_s=120)
wait_for_peer(b, id_a, timeout_s=120)
for device, peer, announcer in ((a, id_b, b), (b, id_a, a)):
connected = False
last: dict = {}
for _attempt in range(4):
last = device.cmd("connect", timeout_ms=45_000, peer=peer)
if last.get("status") == "ok" and last.get("direct"):
connected = True
break
# Already acceptable if the mesh formed a direct link on its own.
peers = device.cmd_ok("peers").get("peers", [])
match = next((p for p in peers if p.get("id") == peer), None)
if match and match.get("direct"):
connected = True
break
# A GATT link carries traffic both ways. Requiring a second client connection from
# the other endpoint turns a healthy single direct link into a false test failure,
# especially after one phone has just restarted and has not rebuilt its address map.
return
# Already acceptable if either endpoint formed a direct link on its own.
for observer, observed_peer in ((device, peer), (announcer, whoami(device)["peer_id"])):
peers = observer.cmd_ok("peers").get("peers", [])
match = next((p for p in peers if p.get("id") == observed_peer), None)
if match and match.get("direct"):
return
try:
announcer.cmd_ok("announce")
except MeshLabError:
pass
time.sleep(4)
if not connected:
raise MeshLabError(f"[{device.alias}] no direct link to {peer}: connect={last}")
raise MeshLabError(f"no direct link formed between {a.alias} and {b.alias}: connect={last}")
def force_handshake(device: Device, peer_id: str, attempts: int = 5, per_attempt_s: int = 20) -> dict:
@ -943,6 +1239,9 @@ SCENARIOS = {
"favorite_verification": scenario_favorite_verification,
"broadcast": scenario_broadcast,
"sync_recovery": scenario_sync_recovery,
"sync_auto_recovery": scenario_sync_auto_recovery,
"sync_file_recovery": scenario_sync_file_recovery,
"durable_outbox": scenario_durable_outbox,
"courier_contract": scenario_courier_contract,
"ptt_dm": scenario_ptt_dm,
"ptt_broadcast": scenario_ptt_broadcast,
@ -968,6 +1267,11 @@ SCENARIOS = {
"identity_reset": scenario_identity_reset,
}
# End-to-end peer courier delivery needs distinct sender, courier, and recipient identities.
TRIAD_SCENARIOS = {
"courier_delivery": scenario_courier_delivery,
}
# Scenarios supported when device B is a watch (file scenarios are receive-only: phone sends,
# the watch must receive with matching digests).
WATCH_SCENARIOS = [
@ -984,9 +1288,16 @@ WATCH_SCENARIOS = [
]
def run_scenario(name: str, a: Device, b: Device, out: Path | None) -> dict:
def run_scenario(
name: str,
a: Device,
b: Device,
out: Path | None,
c: Device | None = None,
) -> dict:
started = time.time()
evidence: dict[str, object] = {"scenario": name, "devices": [a.alias, b.alias]}
devices = [a, b] + ([c] if c is not None else [])
evidence: dict[str, object] = {"scenario": name, "devices": [device.alias for device in devices]}
try:
supported = WATCH_SCENARIOS if isinstance(b, WatchDevice) else list(SCENARIOS)
if name == "all":
@ -997,9 +1308,20 @@ def run_scenario(name: str, a: Device, b: Device, out: Path | None) -> dict:
results[n] = sub.get("results", {"error": sub.get("error", "unknown")})
if sub["status"] != "pass":
failures.append(n)
if c is not None:
sub = run_scenario("courier_delivery", a, b, out, c)
results["courier_delivery"] = sub.get("results", {"error": sub.get("error", "unknown")})
if sub["status"] != "pass":
failures.append("courier_delivery")
evidence["results"] = results
if failures:
raise MeshLabError(f"sub-scenarios failed: {', '.join(failures)}")
elif name in TRIAD_SCENARIOS:
if c is None:
raise MeshLabError(f"scenario '{name}' requires --serial-c")
if isinstance(b, WatchDevice):
raise MeshLabError(f"scenario '{name}' requires three phones")
evidence["results"] = TRIAD_SCENARIOS[name](a, b, c)
elif name not in supported:
raise MeshLabError(f"scenario '{name}' is not supported on device '{b.alias}'")
else:
@ -1008,7 +1330,7 @@ def run_scenario(name: str, a: Device, b: Device, out: Path | None) -> dict:
except (MeshLabError, AssertionError) as error:
evidence["status"] = "fail"
evidence["error"] = str(error)
evidence["logcat"] = {d.alias: d.logcat_dump() for d in (a, b)}
evidence["logcat"] = {d.alias: d.logcat_dump() for d in devices}
evidence["duration_s"] = round(time.time() - started, 1)
if out is not None:
out.mkdir(parents=True, exist_ok=True)
@ -1025,16 +1347,19 @@ def build_parser() -> argparse.ArgumentParser:
setup = commands.add_parser("setup", help="install, grant, launch, nickname, discover")
setup.add_argument("--serial-a", required=True)
setup.add_argument("--serial-b")
setup.add_argument("--serial-c", help="third phone for a three-party scenario")
setup.add_argument("--serial-watch", help="watch serial; used as device B (overrides --serial-b)")
setup.add_argument("--apk", type=Path, default=None)
setup.add_argument("--watch-apk", type=Path, default=None)
setup.add_argument("--nickname-a", default="alice")
setup.add_argument("--nickname-b", default="bob")
setup.add_argument("--nickname-c", default="charlie")
scenario = commands.add_parser("scenario", help="run a test scenario on two devices")
scenario.add_argument("name", choices=[*SCENARIOS.keys(), "all"])
scenario = commands.add_parser("scenario", help="run a test scenario on two phones, or three for courier delivery")
scenario.add_argument("name", choices=[*SCENARIOS.keys(), *TRIAD_SCENARIOS.keys(), "all"])
scenario.add_argument("--serial-a", required=True)
scenario.add_argument("--serial-b")
scenario.add_argument("--serial-c", help="third phone; required for courier_delivery")
scenario.add_argument("--serial-watch", help="watch serial; used as device B (overrides --serial-b)")
scenario.add_argument("--out", type=Path, default=None, help="evidence output directory")
@ -1047,30 +1372,36 @@ def build_parser() -> argparse.ArgumentParser:
return parser
def _resolve_devices(args: argparse.Namespace) -> tuple[Device, Device]:
def _resolve_devices(args: argparse.Namespace) -> tuple[Device, Device, Device | None]:
"""Device A is always the phone; device B is a watch when --serial-watch is given."""
a = Device(args.serial_a, "alpha")
if getattr(args, "serial_watch", None):
return a, WatchDevice(args.serial_watch)
if getattr(args, "serial_c", None):
raise MeshLabError("--serial-c cannot be combined with --serial-watch")
return a, WatchDevice(args.serial_watch), None
if not getattr(args, "serial_b", None):
raise MeshLabError("either --serial-b or --serial-watch is required")
return a, Device(args.serial_b, "beta")
c = Device(args.serial_c, "charlie") if getattr(args, "serial_c", None) else None
return a, Device(args.serial_b, "beta"), c
def main(argv: list[str] | None = None) -> int:
args = build_parser().parse_args(argv)
try:
if args.command == "setup":
a, b = _resolve_devices(args)
a, b, c = _resolve_devices(args)
nickname_b = "watch" if isinstance(b, WatchDevice) and args.nickname_b == "bob" else args.nickname_b
setup_pair(
a, b, args.apk, args.nickname_a, nickname_b,
apk_b=args.watch_apk if isinstance(b, WatchDevice) else None,
)
if c is not None:
setup_triad(a, b, c, args.apk, args.nickname_a, nickname_b, args.nickname_c)
else:
setup_pair(
a, b, args.apk, args.nickname_a, nickname_b,
apk_b=args.watch_apk if isinstance(b, WatchDevice) else None,
)
print(json.dumps({"status": "ok", "step": "setup"}))
elif args.command == "scenario":
a, b = _resolve_devices(args)
evidence = run_scenario(args.name, a, b, args.out)
a, b, c = _resolve_devices(args)
evidence = run_scenario(args.name, a, b, args.out, c)
print(json.dumps(evidence, indent=2, default=str))
return 0 if evidence["status"] == "pass" else 1
elif args.command == "cmd":