nick/amethyst
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

fix: HIGH/MEDIUM Marmot bugs - thread safety, dedup, UI error handling

Browse Source 
- H8: Add Mutex to MarmotSubscriptionManager for thread safety
- H10: Add retained exporter secret to MlsGroupState/RetainedEpochSecrets
  for outer decryption of out-of-order messages
- H14: Add error handling to CreateGroupScreen and MarmotGroupChatView
- H15: Add removeGroup() to MarmotGroupList, clean up after leave
- M3: Use SecureRandom for nostrGroupId generation
- M5: Add event deduplication to MarmotInboundProcessor
- M6: Add KeyPackage credential validation in MarmotManager.addMember()
- M7: Validate nostrGroupId matches WelcomeEvent h-tag

https://claude.ai/code/session_018gVkmmYgMFtBH7G31pCk9N
This commit is contained in:
Claude
2026-04-07 23:03:27 +00:00
parent 29d1610d1a
commit 8c8ab4bb2c
9 changed files with 160 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files
quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/marmot/MarmotInboundProcessor.kt
+53 -19
View File
@@ -71,6 +71,13 @@ sealed class GroupEventResult {
val epoch: Long,
) : GroupEventResult()
/**
* The event was already processed (duplicate).
*/
data class Duplicate(
val groupId: HexKey,
) : GroupEventResult()
/**
* The event could not be processed.
*/
@@ -119,6 +126,16 @@ class MarmotInboundProcessor(
private val keyPackageRotationManager: KeyPackageRotationManager,
) {
private val commitTracker = CommitOrdering.EpochCommitTracker()
private val processedEventIds = LinkedHashSet<String>()
companion object {
private const val MAX_PROCESSED_IDS = 10_000
/**
* Check if an unwrapped event is a Marmot WelcomeEvent.
*/
fun isWelcomeEvent(event: Event): Boolean = event.kind == WelcomeEvent.KIND
}
/**
* Process an inbound GroupEvent (kind:445).
@@ -136,6 +153,13 @@ class MarmotInboundProcessor(
* @return the processing result
*/
suspend fun processGroupEvent(groupEvent: GroupEvent): GroupEventResult {
// Deduplicate already-processed events
val eventId = groupEvent.id
if (eventId in processedEventIds) {
val gId = groupEvent.groupId()
return GroupEventResult.Duplicate(gId ?: "")
}
val groupId =
groupEvent.groupId()
?: return GroupEventResult.Error(null, "GroupEvent missing h tag (group ID)")
@@ -144,22 +168,39 @@ class MarmotInboundProcessor(
return GroupEventResult.Error(groupId, "Not a member of group $groupId")
}
return try {
// Step 1: Outer ChaCha20-Poly1305 decryption
val exporterKey = groupManager.exporterSecret(groupId)
val mlsBytes = GroupEventEncryption.decrypt(groupEvent.encryptedContent(), exporterKey)
val result =
try {
// Step 1: Outer ChaCha20-Poly1305 decryption
val exporterKey = groupManager.exporterSecret(groupId)
val mlsBytes = GroupEventEncryption.decrypt(groupEvent.encryptedContent(), exporterKey)
// Step 2: Parse the MLS message
val mlsMessage = MlsMessage.decodeTls(TlsReader(mlsBytes))
// Step 2: Parse the MLS message
val mlsMessage = MlsMessage.decodeTls(TlsReader(mlsBytes))
when (mlsMessage.wireFormat) {
WireFormat.PRIVATE_MESSAGE -> processPrivateMessage(groupId, mlsMessage, groupEvent)
WireFormat.PUBLIC_MESSAGE -> processPublicMessage(groupId, mlsMessage, groupEvent)
else -> GroupEventResult.Error(groupId, "Unexpected wire format: ${mlsMessage.wireFormat}")
when (mlsMessage.wireFormat) {
WireFormat.PRIVATE_MESSAGE -> processPrivateMessage(groupId, mlsMessage, groupEvent)
WireFormat.PUBLIC_MESSAGE -> processPublicMessage(groupId, mlsMessage, groupEvent)
else -> GroupEventResult.Error(groupId, "Unexpected wire format: ${mlsMessage.wireFormat}")
}
} catch (e: Exception) {
GroupEventResult.Error(groupId, "Failed to process GroupEvent: ${e.message}", e)
}
// Track successfully processed events for deduplication
if (result !is GroupEventResult.Error) {
processedEventIds.add(eventId)
// Trim the set if it exceeds the max size
if (processedEventIds.size > MAX_PROCESSED_IDS) {
val iterator = processedEventIds.iterator()
val toRemove = processedEventIds.size - MAX_PROCESSED_IDS
repeat(toRemove) {
iterator.next()
iterator.remove()
}
}
} catch (e: Exception) {
GroupEventResult.Error(groupId, "Failed to process GroupEvent: ${e.message}", e)
}
return result
}
/**
@@ -364,11 +405,4 @@ class MarmotInboundProcessor(
if (hex == null) return ByteArray(0)
return hex.hexToByteArray()
}
companion object {
/**
* Check if an unwrapped event is a Marmot WelcomeEvent.
*/
fun isWelcomeEvent(event: Event): Boolean = event.kind == WelcomeEvent.KIND
}
}
quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/marmot/MarmotSubscriptionManager.kt
+35 -26
View File
@@ -52,6 +52,7 @@ data class GroupSubscriptionState(
class MarmotSubscriptionManager(
private val userPubKey: HexKey,
) {
private val mutex = Mutex()
private val groupSubscriptions = mutableMapOf<HexKey, GroupSubscriptionState>()
private var giftWrapSince: Long? = null
@@ -62,10 +63,10 @@ class MarmotSubscriptionManager(
* @param nostrGroupId hex-encoded Nostr group ID
* @param since optional timestamp to resume from (e.g., last seen event)
*/
fun subscribeGroup(
suspend fun subscribeGroup(
nostrGroupId: HexKey,
since: Long? = null,
) {
) = mutex.withLock {
groupSubscriptions[nostrGroupId] =
GroupSubscriptionState(
nostrGroupId = nostrGroupId,
@@ -78,27 +79,29 @@ class MarmotSubscriptionManager(
* Unsubscribe from a group's events.
* Call this when leaving a group.
*/
fun unsubscribeGroup(nostrGroupId: HexKey) {
groupSubscriptions.remove(nostrGroupId)
}
suspend fun unsubscribeGroup(nostrGroupId: HexKey) =
mutex.withLock {
groupSubscriptions.remove(nostrGroupId)
}
/**
* Update the `since` timestamp for a group after processing events.
* This ensures reconnections only fetch newer events.
*/
fun updateGroupSince(
suspend fun updateGroupSince(
nostrGroupId: HexKey,
since: Long,
) {
) = mutex.withLock {
groupSubscriptions[nostrGroupId]?.since = since
}
/**
* Update the `since` timestamp for gift wrap subscriptions.
*/
fun updateGiftWrapSince(since: Long) {
giftWrapSince = since
}
suspend fun updateGiftWrapSince(since: Long) =
mutex.withLock {
giftWrapSince = since
}
/**
* Returns all active group IDs being tracked.
@@ -185,26 +188,32 @@ class MarmotSubscriptionManager(
*
* @param activeGroupIds the set of group IDs from [MlsGroupManager.activeGroupIds]
*/
fun syncWithGroupManager(activeGroupIds: Set<HexKey>) {
// Add new groups
for (groupId in activeGroupIds) {
if (!groupSubscriptions.containsKey(groupId)) {
subscribeGroup(groupId)
suspend fun syncWithGroupManager(activeGroupIds: Set<HexKey>) =
mutex.withLock {
// Add new groups
for (groupId in activeGroupIds) {
if (!groupSubscriptions.containsKey(groupId)) {
groupSubscriptions[groupId] =
GroupSubscriptionState(
nostrGroupId = groupId,
active = true,
)
}
}
// Remove stale groups
val staleGroups = groupSubscriptions.keys - activeGroupIds
for (groupId in staleGroups) {
groupSubscriptions.remove(groupId)
}
}
// Remove stale groups
val staleGroups = groupSubscriptions.keys - activeGroupIds
for (groupId in staleGroups) {
unsubscribeGroup(groupId)
}
}
/**
* Clear all subscription state.
*/
fun clear() {
groupSubscriptions.clear()
giftWrapSince = null
}
suspend fun clear() =
mutex.withLock {
groupSubscriptions.clear()
giftWrapSince = null
}
}
quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/marmot/mls/group/MlsGroupManager.kt
+5 -2
View File
@@ -79,8 +79,11 @@ import kotlinx.coroutines.sync.withLock
* The epoch secret retention window is [EPOCH_RETENTION_WINDOW] = 2, meaning secrets
* for the current and previous epoch are kept for late-message decryption.
*
* Thread safety: All public methods are suspending and should be called
* from a single coroutine context (e.g., the Account's scope).
* Thread safety: All suspending mutation methods are guarded by a [Mutex]
* to prevent concurrent state corruption. Non-suspending read methods
* ([getGroup], [isMember], [activeGroupIds], [encrypt], [decrypt],
* [decryptOrNull], [exporterSecret]) must be called from the same
* coroutine context that owns this manager (e.g., the Account's scope).
*
* @see MlsGroup The low-level MLS state machine
* @see MlsGroupStateStore Storage abstraction for group state persistence
quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/marmot/mls/group/MlsGroupState.kt
+21 -6
View File
@@ -171,6 +171,7 @@ data class RetainedEpochSecrets(
val senderDataSecret: ByteArray,
val encryptionSecret: ByteArray,
val leafCount: Int,
val exporterSecret: ByteArray = ByteArray(0),
) {
override fun equals(other: Any?): Boolean {
if (this === other) return true
@@ -185,15 +186,29 @@ data class RetainedEpochSecrets(
writer.putOpaqueVarInt(senderDataSecret)
writer.putOpaqueVarInt(encryptionSecret)
writer.putUint32(leafCount.toLong())
writer.putOpaqueVarInt(exporterSecret)
}
companion object {
fun decodeTls(reader: TlsReader): RetainedEpochSecrets =
RetainedEpochSecrets(
epoch = reader.readUint64(),
senderDataSecret = reader.readOpaqueVarInt(),
encryptionSecret = reader.readOpaqueVarInt(),
leafCount = reader.readUint32().toInt(),
fun decodeTls(reader: TlsReader): RetainedEpochSecrets {
val epoch = reader.readUint64()
val senderDataSecret = reader.readOpaqueVarInt()
val encryptionSecret = reader.readOpaqueVarInt()
val leafCount = reader.readUint32().toInt()
// exporterSecret was added later; tolerate its absence in older serialized data
val exporterSecret =
if (reader.hasRemaining) {
reader.readOpaqueVarInt()
} else {
ByteArray(0)
}
return RetainedEpochSecrets(
epoch = epoch,
senderDataSecret = senderDataSecret,
encryptionSecret = encryptionSecret,
leafCount = leafCount,
exporterSecret = exporterSecret,
)
}
}
}