diff --git a/core/data/src/commonMain/kotlin/org/meshtastic/core/data/manager/MeshDataHandlerImpl.kt b/core/data/src/commonMain/kotlin/org/meshtastic/core/data/manager/MeshDataHandlerImpl.kt index f97ff1fdca..fa19243fac 100644 --- a/core/data/src/commonMain/kotlin/org/meshtastic/core/data/manager/MeshDataHandlerImpl.kt +++ b/core/data/src/commonMain/kotlin/org/meshtastic/core/data/manager/MeshDataHandlerImpl.kt @@ -21,6 +21,7 @@ import co.touchlab.kermit.Severity import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Job import kotlinx.coroutines.flow.first +import kotlinx.coroutines.flow.firstOrNull import okio.ByteString import org.koin.core.annotation.Named import org.koin.core.annotation.Single @@ -33,6 +34,7 @@ import org.meshtastic.core.model.Node import org.meshtastic.core.model.NodeAddress import org.meshtastic.core.model.Reaction import org.meshtastic.core.model.destination +import org.meshtastic.core.model.geofence.activeWaypointPackets import org.meshtastic.core.model.isBroadcast import org.meshtastic.core.model.isFromLocal import org.meshtastic.core.model.isModifiableBy @@ -291,8 +293,33 @@ class MeshDataHandlerImpl( val u = Waypoint.ADAPTER.decode(payload) // A locked waypoint may only be created/updated by its owner; drop it if the sender isn't allowed to modify it. if (!u.isModifiableBy(packet.from)) return - val currentSecond = nowSeconds.toInt() - rememberDataPacket(dataPacket, myNodeNum, updateNotification = u.expire > currentSecond, session = session) + val updateNotification = u.expire > nowSeconds.toInt() + radioInterfaceService.launchSessionWork(scope, session) { + // Persisted-owner enforcement: a stored, locked waypoint may only be modified by the node it is locked to. + // The inbound check above only validates the incoming payload, so without this a non-owner could hijack a + // stored locked waypoint by replaying its id with locked_to = 0 (unlock) or their own num (takeover). The + // read and the write share this one lease so the check sees a committed snapshot (see [persistDataPacket]). + if (!storedWaypointModifiableBy(u.id, packet.from)) return@launchSessionWork + persistDataPacket(dataPacket, myNodeNum, updateNotification) + } + } + + /** + * Whether an inbound waypoint update from [from] may modify the currently-stored waypoint with [waypointId]. A new + * waypoint (nothing stored) or an unlocked stored waypoint is always modifiable; a locked stored waypoint is + * modifiable only by the node it is locked to — this deliberately rejects inbound unlock (`locked_to = 0`) attempts + * from anyone else. + * + * [PacketRepository.getWaypoints] is a row-per-transmission firehose, so it is collapsed via + * [activeWaypointPackets] (newest-per-id, expired dropped) — the same normalisation the map UI and geofence engine + * use, so the three cannot drift. Waypoint packets are infrequent, so this one-shot read per waypoint is not hot. + */ + private suspend fun storedWaypointModifiableBy(waypointId: Int, from: Int): Boolean { + // firstOrNull().orEmpty(): getWaypoints() is a hot repository flow that always emits (an empty list when there + // are none), but tolerate a flow that completes without emitting rather than throwing on the inbound path. + val active = packetRepository.value.getWaypoints().firstOrNull().orEmpty().activeWaypointPackets(nowSeconds) + val stored = active[waypointId]?.waypoint + return stored == null || stored.isModifiableBy(from) } private fun handleTextMessage( @@ -405,6 +432,18 @@ class MeshDataHandlerImpl( session: RadioSessionContext?, ) { if (dataPacket.dataType !in rememberDataType) return + radioInterfaceService.launchSessionWork(scope, session) { + persistDataPacket(dataPacket, myNodeNum, updateNotification) + } + } + + /** + * Deduplicates, filters, persists, and (when appropriate) notifies for a single [dataPacket]. Runs inside a session + * lease — callers must launch it via [RadioInterfaceService.launchSessionWork] and must have already confirmed the + * packet's [DataPacket.dataType] is one of [rememberDataType]. Split out from [rememberDataPacket] so the waypoint + * path can gate persistence on a repository read within the same lease (see [handleWaypoint]). + */ + private suspend fun persistDataPacket(dataPacket: DataPacket, myNodeNum: Int, updateNotification: Boolean) { val fromLocal = dataPacket.isFromLocal(myNodeNum) val toBroadcast = dataPacket.isBroadcast val contactId = if (fromLocal || toBroadcast) dataPacket.to else dataPacket.from @@ -412,33 +451,24 @@ class MeshDataHandlerImpl( // contactKey: unique contact key filter (channel)+(nodeId) val contactKey = "${dataPacket.channel}$contactId" - radioInterfaceService.launchSessionWork(scope, session) { - packetRepository.value.apply { - // Check for duplicates before inserting - val existingPackets = findPacketsWithId(dataPacket.id) - if (existingPackets.isNotEmpty()) { - Logger.d { - "Skipping duplicate packet: packetId=${dataPacket.id} from=${dataPacket.from} " + - "to=${dataPacket.to} contactKey=$contactKey" + - " (already have ${existingPackets.size} packet(s))" - } - return@launchSessionWork + packetRepository.value.apply { + // Check for duplicates before inserting + val existingPackets = findPacketsWithId(dataPacket.id) + if (existingPackets.isNotEmpty()) { + Logger.d { + "Skipping duplicate packet: packetId=${dataPacket.id} from=${dataPacket.from} " + + "to=${dataPacket.to} contactKey=$contactKey" + + " (already have ${existingPackets.size} packet(s))" } + return + } - // Check if message should be filtered - val isFiltered = shouldFilterMessage(dataPacket, contactKey) + // Check if message should be filtered + val isFiltered = shouldFilterMessage(dataPacket, contactKey) - insert( - dataPacket, - myNodeNum, - contactKey, - nowMillis, - read = fromLocal || isFiltered, - filtered = isFiltered, - ) - if (!isFiltered) { - handlePacketNotification(dataPacket, contactKey, updateNotification) - } + insert(dataPacket, myNodeNum, contactKey, nowMillis, read = fromLocal || isFiltered, filtered = isFiltered) + if (!isFiltered) { + handlePacketNotification(dataPacket, contactKey, updateNotification) } } } diff --git a/core/data/src/commonTest/kotlin/org/meshtastic/core/data/manager/MeshDataHandlerTest.kt b/core/data/src/commonTest/kotlin/org/meshtastic/core/data/manager/MeshDataHandlerTest.kt index 80c181a832..e7e6ac5614 100644 --- a/core/data/src/commonTest/kotlin/org/meshtastic/core/data/manager/MeshDataHandlerTest.kt +++ b/core/data/src/commonTest/kotlin/org/meshtastic/core/data/manager/MeshDataHandlerTest.kt @@ -32,6 +32,7 @@ import kotlinx.coroutines.cancel import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.StateFlow import kotlinx.coroutines.flow.emptyFlow +import kotlinx.coroutines.flow.flowOf import kotlinx.coroutines.test.StandardTestDispatcher import kotlinx.coroutines.test.TestScope import kotlinx.coroutines.test.advanceUntilIdle @@ -74,6 +75,7 @@ import org.meshtastic.proto.Position import org.meshtastic.proto.Routing import org.meshtastic.proto.Telemetry import org.meshtastic.proto.User +import org.meshtastic.proto.Waypoint import kotlin.test.AfterTest import kotlin.test.BeforeTest import kotlin.test.Test @@ -943,4 +945,129 @@ class MeshDataHandlerTest { serviceNotifications.updateMessageNotification(any(), any(), any(), any(), any(), isSilent = false) } } + + // --- Waypoint persisted-owner enforcement --- + // + // A locked waypoint (locked_to != 0) may only be modified by the node it is locked to. The inbound-payload check + // alone (locked_to == from) cannot enforce this: a non-owner can still replay an existing id with locked_to = 0 + // (unlock) or their own num (takeover). handleWaypoint additionally consults the currently-stored owner. + + private fun waypointPacket(txId: Int, from: Int, waypoint: Waypoint): MeshPacket { + val payload = waypoint.encode().toByteString() + val packet = + MeshPacket(id = txId, from = from, decoded = Data(portnum = PortNum.WAYPOINT_APP, payload = payload)) + val dataPacket = + DataPacket( + id = txId, + from = NodeAddress.numToDefaultId(from), + to = NodeAddress.ID_BROADCAST, + bytes = payload, + dataType = PortNum.WAYPOINT_APP.value, + ) + every { dataMapper.toDataPacket(packet) } returns dataPacket + return packet + } + + /** Persist a single stored waypoint (via the getWaypoints firehose) so handleWaypoint can read its owner. */ + private fun storeWaypoint(id: Int, lockedTo: Int) { + val stored = + DataPacket(to = NodeAddress.ID_BROADCAST, channel = 0, waypoint = Waypoint(id = id, locked_to = lockedTo)) + every { packetRepository.getWaypoints() } returns flowOf(listOf(stored)) + } + + private fun stubWaypointPersistDependencies(txId: Int) { + everySuspend { packetRepository.findPacketsWithId(txId) } returns emptyList() + everySuspend { packetRepository.getContactSettings(any()) } returns ContactSettings(contactKey = "test") + every { messageFilter.shouldFilter(any(), any()) } returns false + } + + @Test + fun `non-owner unlock replay of a locked waypoint is dropped`() = testScope.runTest { + storeWaypoint(id = 42, lockedTo = 111) + // Mallory (999) replays waypoint 42 with locked_to = 0 (unlock). The inbound check passes, so only the + // stored-owner check can drop it. + val packet = waypointPacket(txId = 500, from = 999, waypoint = Waypoint(id = 42, locked_to = 0)) + stubWaypointPersistDependencies(500) + + handler.handleReceivedData(packet, 123) + advanceUntilIdle() + + verifySuspend(exactly(0)) { packetRepository.insert(any(), any(), any(), any(), any(), any()) } + } + + @Test + fun `non-owner takeover of a locked waypoint is dropped`() = testScope.runTest { + storeWaypoint(id = 42, lockedTo = 111) + // Mallory (999) locks waypoint 42 to herself. The inbound check passes (locked_to == from), so only the + // stored-owner check can catch this. + val packet = waypointPacket(txId = 501, from = 999, waypoint = Waypoint(id = 42, locked_to = 999)) + stubWaypointPersistDependencies(501) + + handler.handleReceivedData(packet, 123) + advanceUntilIdle() + + verifySuspend(exactly(0)) { packetRepository.insert(any(), any(), any(), any(), any(), any()) } + } + + @Test + fun `owner unlock of their own locked waypoint is accepted`() = testScope.runTest { + storeWaypoint(id = 42, lockedTo = 111) + val packet = waypointPacket(txId = 502, from = 111, waypoint = Waypoint(id = 42, locked_to = 0)) + stubWaypointPersistDependencies(502) + + handler.handleReceivedData(packet, 123) + advanceUntilIdle() + + verifySuspend { packetRepository.insert(any(), 123, any(), any(), any(), any()) } + } + + @Test + fun `owner edit of their own locked waypoint is accepted`() = testScope.runTest { + storeWaypoint(id = 42, lockedTo = 111) + val packet = waypointPacket(txId = 503, from = 111, waypoint = Waypoint(id = 42, locked_to = 111)) + stubWaypointPersistDependencies(503) + + handler.handleReceivedData(packet, 123) + advanceUntilIdle() + + verifySuspend { packetRepository.insert(any(), 123, any(), any(), any(), any()) } + } + + @Test + fun `new waypoint from a non-owner is accepted when none is stored`() = testScope.runTest { + // Nothing persisted yet: getWaypoints() emits an empty list (as Room does). A creation, not a hijack. + every { packetRepository.getWaypoints() } returns flowOf(emptyList()) + val packet = waypointPacket(txId = 504, from = 999, waypoint = Waypoint(id = 42, locked_to = 0)) + stubWaypointPersistDependencies(504) + + handler.handleReceivedData(packet, 123) + advanceUntilIdle() + + verifySuspend { packetRepository.insert(any(), 123, any(), any(), any(), any()) } + } + + @Test + fun `non-owner update to an unlocked waypoint is accepted`() = testScope.runTest { + storeWaypoint(id = 42, lockedTo = 0) + val packet = waypointPacket(txId = 505, from = 999, waypoint = Waypoint(id = 42, locked_to = 0)) + stubWaypointPersistDependencies(505) + + handler.handleReceivedData(packet, 123) + advanceUntilIdle() + + verifySuspend { packetRepository.insert(any(), 123, any(), any(), any(), any()) } + } + + @Test + fun `waypoint locked to someone other than the sender is dropped`() = testScope.runTest { + // Pre-existing inbound-payload rule: a node can only lock a waypoint to itself. Nothing stored here — the + // payload itself is invalid, so it is rejected before any repository read. + val packet = waypointPacket(txId = 506, from = 999, waypoint = Waypoint(id = 42, locked_to = 111)) + stubWaypointPersistDependencies(506) + + handler.handleReceivedData(packet, 123) + advanceUntilIdle() + + verifySuspend(exactly(0)) { packetRepository.insert(any(), any(), any(), any(), any(), any()) } + } }