Merge pull request #2169 from vitorpamplona/claude/review-marmot-mls-C5whm

Add thread safety and robustness improvements to Marmot MLS implementation
This commit is contained in:
Vitor Pamplona
2026-04-07 19:27:23 -04:00
committed by GitHub
24 changed files with 985 additions and 337 deletions
@@ -54,7 +54,7 @@ class AndroidMlsGroupStateStore(
) = withContext(Dispatchers.IO) { ) = withContext(Dispatchers.IO) {
val file = stateFile(nostrGroupId) val file = stateFile(nostrGroupId)
file.parentFile?.mkdirs() file.parentFile?.mkdirs()
file.writeBytes(encryption.encrypt(state)) atomicWrite(file, encryption.encrypt(state))
} }
override suspend fun load(nostrGroupId: String): ByteArray? = override suspend fun load(nostrGroupId: String): ByteArray? =
@@ -112,7 +112,7 @@ class AndroidMlsGroupStateStore(
offset += len offset += len
} }
file.writeBytes(encryption.encrypt(buffer)) atomicWrite(file, encryption.encrypt(buffer))
} }
override suspend fun loadRetainedEpochs(nostrGroupId: String): List<ByteArray> = override suspend fun loadRetainedEpochs(nostrGroupId: String): List<ByteArray> =
@@ -144,4 +144,21 @@ class AndroidMlsGroupStateStore(
} }
result result
} }
/**
* Write data atomically: write to a temp file first, then rename.
* This avoids corrupted state if the app crashes mid-write.
*/
private fun atomicWrite(
target: File,
data: ByteArray,
) {
val tempFile = File(target.parentFile, "${target.name}.tmp")
tempFile.writeBytes(data)
if (!tempFile.renameTo(target)) {
// Fallback: if rename fails (e.g., cross-filesystem), copy and delete
tempFile.copyTo(target, overwrite = true)
tempFile.delete()
}
}
} }
@@ -51,23 +51,45 @@ class MarmotGroupEventsEoseManager(
val manager = key.account.marmotManager ?: return emptyList() val manager = key.account.marmotManager ?: return emptyList()
if (!key.account.isWriteable()) return emptyList() if (!key.account.isWriteable()) return emptyList()
// Build Marmot filters (kind:445 per group + kind:30443 own key packages) val result = mutableListOf<RelayBasedFilter>()
val marmotFilters = manager.subscriptionManager.activeGroupFilters() val fallbackRelays = key.account.homeRelays.flow.value
// Per-group kind:445 filters — route each to the group's own relays
val groupStates = manager.subscriptionManager.activeGroupIds()
for (groupId in groupStates) {
val filter =
manager.subscriptionManager.let { sub ->
// Build the filter for this specific group
val groupFilters = sub.activeGroupFilters()
// activeGroupFilters() returns one filter per group; match by tag content
groupFilters.find { f ->
f.tags?.any { it.value?.contains(groupId) == true } == true
}
} ?: continue
// Use group-specific relays from MLS metadata; fall back to home relays
val metadata = manager.groupMetadata(groupId)
val groupRelays =
metadata
?.relays
?.mapNotNull {
com.vitorpamplona.quartz.nip01Core.relay.normalizer.RelayUrlNormalizer.normalizeOrNull(it)
}?.toSet()
val relaysForGroup = if (!groupRelays.isNullOrEmpty()) groupRelays else fallbackRelays
for (relay in relaysForGroup) {
result.add(RelayBasedFilter(relay = relay, filter = filter))
}
}
// Own KeyPackage filter (kind:30443) — use home relays
if (fallbackRelays.isNotEmpty()) {
val ownKeyPackageFilter = manager.subscriptionManager.ownKeyPackageFilter() val ownKeyPackageFilter = manager.subscriptionManager.ownKeyPackageFilter()
val allFilters = marmotFilters + ownKeyPackageFilter for (relay in fallbackRelays) {
result.add(RelayBasedFilter(relay = relay, filter = ownKeyPackageFilter))
// Send to the home relays (where group events are relayed)
val relays = key.account.homeRelays.flow.value
if (relays.isEmpty()) return emptyList()
return relays.flatMap { relay ->
allFilters.map { filter ->
RelayBasedFilter(
relay = relay,
filter = filter,
)
} }
} }
return result
} }
val userJobMap = mutableMapOf<User, List<Job>>() val userJobMap = mutableMapOf<User, List<Job>>()
@@ -1452,7 +1452,7 @@ class AccountViewModel(
description = text, description = text,
) )
val innerEvent = account.signer.sign<com.vitorpamplona.quartz.nip01Core.core.Event>(template) val innerEvent = account.signer.sign<com.vitorpamplona.quartz.nip01Core.core.Event>(template)
val relays = account.outboxRelays.flow.value val relays = marmotGroupRelays(nostrGroupId)
account.sendMarmotGroupMessage(nostrGroupId, innerEvent, relays) account.sendMarmotGroupMessage(nostrGroupId, innerEvent, relays)
} }
@@ -1467,10 +1467,26 @@ class AccountViewModel(
fun hasPublishedKeyPackage(): Boolean = account.hasPublishedKeyPackage() fun hasPublishedKeyPackage(): Boolean = account.hasPublishedKeyPackage()
suspend fun leaveMarmotGroup(nostrGroupId: String) { suspend fun leaveMarmotGroup(nostrGroupId: String) {
val relays = account.outboxRelays.flow.value val relays = marmotGroupRelays(nostrGroupId)
account.leaveMarmotGroup(nostrGroupId, relays) account.leaveMarmotGroup(nostrGroupId, relays)
} }
/**
* Get the relay set for a Marmot group from MLS GroupContext metadata.
* Falls back to outbox relays if the group has no configured relays.
*/
private fun marmotGroupRelays(nostrGroupId: String): Set<NormalizedRelayUrl> {
val metadata = account.marmotManager?.groupMetadata(nostrGroupId)
val groupRelays =
metadata
?.relays
?.mapNotNull {
com.vitorpamplona.quartz.nip01Core.relay.normalizer.RelayUrlNormalizer
.normalizeOrNull(it)
}?.toSet()
return if (!groupRelays.isNullOrEmpty()) groupRelays else account.outboxRelays.flow.value
}
fun marmotGroupMembers(nostrGroupId: String): List<com.vitorpamplona.amethyst.commons.marmot.GroupMemberInfo> = account.marmotManager?.memberPubkeys(nostrGroupId) ?: emptyList() fun marmotGroupMembers(nostrGroupId: String): List<com.vitorpamplona.amethyst.commons.marmot.GroupMemberInfo> = account.marmotManager?.memberPubkeys(nostrGroupId) ?: emptyList()
suspend fun addMarmotGroupMember( suspend fun addMarmotGroupMember(
@@ -20,6 +20,7 @@
*/ */
package com.vitorpamplona.amethyst.ui.screen.loggedIn.chats.marmotGroup package com.vitorpamplona.amethyst.ui.screen.loggedIn.chats.marmotGroup
import android.widget.Toast
import androidx.compose.foundation.layout.Column import androidx.compose.foundation.layout.Column
import androidx.compose.foundation.layout.consumeWindowInsets import androidx.compose.foundation.layout.consumeWindowInsets
import androidx.compose.foundation.layout.fillMaxWidth import androidx.compose.foundation.layout.fillMaxWidth
@@ -36,6 +37,7 @@ import androidx.compose.runtime.remember
import androidx.compose.runtime.rememberCoroutineScope import androidx.compose.runtime.rememberCoroutineScope
import androidx.compose.runtime.setValue import androidx.compose.runtime.setValue
import androidx.compose.ui.Modifier import androidx.compose.ui.Modifier
import androidx.compose.ui.platform.LocalContext
import androidx.compose.ui.unit.dp import androidx.compose.ui.unit.dp
import com.vitorpamplona.amethyst.ui.navigation.navs.INav import com.vitorpamplona.amethyst.ui.navigation.navs.INav
import com.vitorpamplona.amethyst.ui.navigation.routes.Route import com.vitorpamplona.amethyst.ui.navigation.routes.Route
@@ -44,7 +46,7 @@ import com.vitorpamplona.amethyst.ui.screen.loggedIn.AccountViewModel
import com.vitorpamplona.quartz.nip01Core.core.toHexKey import com.vitorpamplona.quartz.nip01Core.core.toHexKey
import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.launch import kotlinx.coroutines.launch
import kotlin.random.Random import java.security.SecureRandom
@Composable @Composable
fun CreateGroupScreen( fun CreateGroupScreen(
@@ -54,6 +56,7 @@ fun CreateGroupScreen(
var groupName by remember { mutableStateOf("") } var groupName by remember { mutableStateOf("") }
var isCreating by remember { mutableStateOf(false) } var isCreating by remember { mutableStateOf(false) }
val scope = rememberCoroutineScope() val scope = rememberCoroutineScope()
val context = LocalContext.current
Scaffold( Scaffold(
topBar = { topBar = {
@@ -62,7 +65,8 @@ fun CreateGroupScreen(
onPost = { onPost = {
isCreating = true isCreating = true
scope.launch(Dispatchers.IO) { scope.launch(Dispatchers.IO) {
val nostrGroupId = Random.nextBytes(32).toHexKey() try {
val nostrGroupId = ByteArray(32).also { SecureRandom().nextBytes(it) }.toHexKey()
accountViewModel.createMarmotGroup(nostrGroupId) accountViewModel.createMarmotGroup(nostrGroupId)
if (groupName.isNotBlank()) { if (groupName.isNotBlank()) {
accountViewModel.account.marmotGroupList accountViewModel.account.marmotGroupList
@@ -70,6 +74,17 @@ fun CreateGroupScreen(
.displayName.value = groupName .displayName.value = groupName
} }
nav.nav(Route.MarmotGroupChat(nostrGroupId)) nav.nav(Route.MarmotGroupChat(nostrGroupId))
} catch (e: Exception) {
isCreating = false
launch(Dispatchers.Main) {
Toast
.makeText(
context,
"Failed to create group: ${e.message}",
Toast.LENGTH_LONG,
).show()
}
}
} }
}, },
isActive = { !isCreating }, isActive = { !isCreating },
@@ -20,6 +20,7 @@
*/ */
package com.vitorpamplona.amethyst.ui.screen.loggedIn.chats.marmotGroup package com.vitorpamplona.amethyst.ui.screen.loggedIn.chats.marmotGroup
import android.widget.Toast
import androidx.compose.foundation.layout.Column import androidx.compose.foundation.layout.Column
import androidx.compose.foundation.layout.Spacer import androidx.compose.foundation.layout.Spacer
import androidx.compose.foundation.layout.fillMaxHeight import androidx.compose.foundation.layout.fillMaxHeight
@@ -30,12 +31,14 @@ import androidx.compose.material3.MaterialTheme
import androidx.compose.material3.Text import androidx.compose.material3.Text
import androidx.compose.material3.TextFieldDefaults import androidx.compose.material3.TextFieldDefaults
import androidx.compose.runtime.Composable import androidx.compose.runtime.Composable
import androidx.compose.runtime.DisposableEffect
import androidx.compose.runtime.derivedStateOf import androidx.compose.runtime.derivedStateOf
import androidx.compose.runtime.getValue import androidx.compose.runtime.getValue
import androidx.compose.runtime.remember import androidx.compose.runtime.remember
import androidx.compose.runtime.rememberCoroutineScope import androidx.compose.runtime.rememberCoroutineScope
import androidx.compose.ui.Modifier import androidx.compose.ui.Modifier
import androidx.compose.ui.graphics.Color import androidx.compose.ui.graphics.Color
import androidx.compose.ui.platform.LocalContext
import androidx.lifecycle.viewmodel.compose.viewModel import androidx.lifecycle.viewmodel.compose.viewModel
import com.vitorpamplona.amethyst.R import com.vitorpamplona.amethyst.R
import com.vitorpamplona.amethyst.ui.components.ThinPaddingTextField import com.vitorpamplona.amethyst.ui.components.ThinPaddingTextField
@@ -72,6 +75,16 @@ fun MarmotGroupChatView(
WatchLifecycleAndUpdateModel(feedViewModel) WatchLifecycleAndUpdateModel(feedViewModel)
val chatroom =
remember(nostrGroupId) {
accountViewModel.account.marmotGroupList.getOrCreateGroup(nostrGroupId)
}
DisposableEffect(nostrGroupId) {
chatroom.markAsRead()
onDispose { }
}
Column(Modifier.fillMaxHeight()) { Column(Modifier.fillMaxHeight()) {
Column( Column(
modifier = modifier =
@@ -110,6 +123,7 @@ fun MarmotGroupMessageComposer(
val scope = rememberCoroutineScope() val scope = rememberCoroutineScope()
val messageState = remember { TextFieldState() } val messageState = remember { TextFieldState() }
val canPost by remember { derivedStateOf { messageState.text.isNotBlank() } } val canPost by remember { derivedStateOf { messageState.text.isNotBlank() } }
val context = LocalContext.current
Column(modifier = EditFieldModifier) { Column(modifier = EditFieldModifier) {
ThinPaddingTextField( ThinPaddingTextField(
@@ -130,9 +144,20 @@ fun MarmotGroupMessageComposer(
val text = messageState.text.toString().trim() val text = messageState.text.toString().trim()
if (text.isNotEmpty()) { if (text.isNotEmpty()) {
scope.launch(Dispatchers.IO) { scope.launch(Dispatchers.IO) {
try {
accountViewModel.sendMarmotGroupMessage(nostrGroupId, text) accountViewModel.sendMarmotGroupMessage(nostrGroupId, text)
messageState.clearText() messageState.clearText()
onMessageSent() onMessageSent()
} catch (e: Exception) {
launch(Dispatchers.Main) {
Toast
.makeText(
context,
"Failed to send message: ${e.message}",
Toast.LENGTH_SHORT,
).show()
}
}
} }
} }
} }
@@ -20,6 +20,7 @@
*/ */
package com.vitorpamplona.amethyst.ui.screen.loggedIn.chats.marmotGroup package com.vitorpamplona.amethyst.ui.screen.loggedIn.chats.marmotGroup
import android.widget.Toast
import androidx.compose.foundation.clickable import androidx.compose.foundation.clickable
import androidx.compose.foundation.layout.Arrangement import androidx.compose.foundation.layout.Arrangement
import androidx.compose.foundation.layout.Column import androidx.compose.foundation.layout.Column
@@ -52,6 +53,7 @@ import androidx.compose.runtime.rememberCoroutineScope
import androidx.compose.runtime.setValue import androidx.compose.runtime.setValue
import androidx.compose.ui.Alignment import androidx.compose.ui.Alignment
import androidx.compose.ui.Modifier import androidx.compose.ui.Modifier
import androidx.compose.ui.platform.LocalContext
import androidx.compose.ui.text.font.FontWeight import androidx.compose.ui.text.font.FontWeight
import androidx.compose.ui.text.style.TextOverflow import androidx.compose.ui.text.style.TextOverflow
import androidx.compose.ui.unit.dp import androidx.compose.ui.unit.dp
@@ -83,8 +85,10 @@ fun MarmotGroupInfoScreen(
val groupRelays by chatroom.relays.collectAsStateWithLifecycle() val groupRelays by chatroom.relays.collectAsStateWithLifecycle()
var members by remember { mutableStateOf(emptyList<GroupMemberInfo>()) } var members by remember { mutableStateOf(emptyList<GroupMemberInfo>()) }
var showLeaveDialog by remember { mutableStateOf(false) } var showLeaveDialog by remember { mutableStateOf(false) }
var isLeaving by remember { mutableStateOf(false) }
val scope = rememberCoroutineScope() val scope = rememberCoroutineScope()
val myPubkey = accountViewModel.account.signer.pubKey val myPubkey = accountViewModel.account.signer.pubKey
val context = LocalContext.current
LaunchedEffect(nostrGroupId) { LaunchedEffect(nostrGroupId) {
members = accountViewModel.marmotGroupMembers(nostrGroupId) members = accountViewModel.marmotGroupMembers(nostrGroupId)
@@ -222,10 +226,24 @@ fun MarmotGroupInfoScreen(
groupName = displayName ?: "this group", groupName = displayName ?: "this group",
onConfirm = { onConfirm = {
showLeaveDialog = false showLeaveDialog = false
isLeaving = true
scope.launch(Dispatchers.IO) { scope.launch(Dispatchers.IO) {
try {
accountViewModel.leaveMarmotGroup(nostrGroupId) accountViewModel.leaveMarmotGroup(nostrGroupId)
} accountViewModel.account.marmotGroupList.removeGroup(nostrGroupId)
nav.nav(Route.MarmotGroupList) nav.nav(Route.MarmotGroupList)
} catch (e: Exception) {
isLeaving = false
launch(Dispatchers.Main) {
Toast
.makeText(
context,
"Failed to leave group: ${e.message}",
Toast.LENGTH_LONG,
).show()
}
}
}
}, },
onDismiss = { showLeaveDialog = false }, onDismiss = { showLeaveDialog = false },
) )
@@ -20,6 +20,7 @@
*/ */
package com.vitorpamplona.amethyst.ui.screen.loggedIn.chats.marmotGroup package com.vitorpamplona.amethyst.ui.screen.loggedIn.chats.marmotGroup
import androidx.compose.foundation.background
import androidx.compose.foundation.clickable import androidx.compose.foundation.clickable
import androidx.compose.foundation.layout.Arrangement import androidx.compose.foundation.layout.Arrangement
import androidx.compose.foundation.layout.Box import androidx.compose.foundation.layout.Box
@@ -57,10 +58,13 @@ import androidx.compose.runtime.rememberCoroutineScope
import androidx.compose.runtime.setValue import androidx.compose.runtime.setValue
import androidx.compose.ui.Alignment import androidx.compose.ui.Alignment
import androidx.compose.ui.Modifier import androidx.compose.ui.Modifier
import androidx.compose.ui.draw.clip
import androidx.compose.ui.graphics.StrokeCap import androidx.compose.ui.graphics.StrokeCap
import androidx.compose.ui.text.font.FontWeight import androidx.compose.ui.text.font.FontWeight
import androidx.compose.ui.text.style.TextAlign
import androidx.compose.ui.text.style.TextOverflow import androidx.compose.ui.text.style.TextOverflow
import androidx.compose.ui.unit.dp import androidx.compose.ui.unit.dp
import androidx.compose.ui.unit.sp
import androidx.lifecycle.compose.collectAsStateWithLifecycle import androidx.lifecycle.compose.collectAsStateWithLifecycle
import com.vitorpamplona.amethyst.commons.model.marmotGroups.MarmotGroupChatroom import com.vitorpamplona.amethyst.commons.model.marmotGroups.MarmotGroupChatroom
import com.vitorpamplona.amethyst.ui.navigation.navs.INav import com.vitorpamplona.amethyst.ui.navigation.navs.INav
@@ -220,6 +224,7 @@ fun MarmotGroupListItem(
onClick: () -> Unit, onClick: () -> Unit,
) { ) {
val displayName by chatroom.displayName.collectAsStateWithLifecycle() val displayName by chatroom.displayName.collectAsStateWithLifecycle()
val unread by chatroom.unreadCount.collectAsStateWithLifecycle()
val newestMessage = chatroom.newestMessage val newestMessage = chatroom.newestMessage
Row( Row(
@@ -235,7 +240,7 @@ fun MarmotGroupListItem(
Text( Text(
text = displayName ?: "Group ${groupId.take(8)}...", text = displayName ?: "Group ${groupId.take(8)}...",
style = MaterialTheme.typography.titleSmall, style = MaterialTheme.typography.titleSmall,
fontWeight = FontWeight.Bold, fontWeight = if (unread > 0) FontWeight.Bold else FontWeight.Normal,
maxLines = 1, maxLines = 1,
overflow = TextOverflow.Ellipsis, overflow = TextOverflow.Ellipsis,
) )
@@ -258,6 +263,24 @@ fun MarmotGroupListItem(
} }
} }
Column(horizontalAlignment = Alignment.End) { Column(horizontalAlignment = Alignment.End) {
if (unread > 0) {
Box(
modifier =
Modifier
.size(22.dp)
.clip(CircleShape)
.background(MaterialTheme.colorScheme.primary),
contentAlignment = Alignment.Center,
) {
Text(
text = if (unread > 99) "99+" else unread.toString(),
color = MaterialTheme.colorScheme.onPrimary,
fontSize = 11.sp,
fontWeight = FontWeight.Bold,
textAlign = TextAlign.Center,
)
}
} else {
Text( Text(
text = "${chatroom.messages.size} msgs", text = "${chatroom.messages.size} msgs",
style = MaterialTheme.typography.labelSmall, style = MaterialTheme.typography.labelSmall,
@@ -266,3 +289,4 @@ fun MarmotGroupListItem(
} }
} }
} }
}
@@ -105,6 +105,7 @@ class MarmotManager(
} }
is GroupEventResult.CommitPending, is GroupEventResult.CommitPending,
is GroupEventResult.Duplicate,
is GroupEventResult.Error, is GroupEventResult.Error,
-> {} -> {}
} }
@@ -120,6 +121,14 @@ class MarmotManager(
welcomeEvent: WelcomeEvent, welcomeEvent: WelcomeEvent,
nostrGroupId: HexKey, nostrGroupId: HexKey,
): WelcomeResult { ): WelcomeResult {
// Validate that the provided nostrGroupId matches the WelcomeEvent's h-tag if present
val eventGroupId = welcomeEvent.nostrGroupId()
if (eventGroupId != null && eventGroupId != nostrGroupId) {
return WelcomeResult.Error(
"nostrGroupId mismatch: expected $nostrGroupId but WelcomeEvent has $eventGroupId",
)
}
val result = inboundProcessor.processWelcome(welcomeEvent, nostrGroupId) val result = inboundProcessor.processWelcome(welcomeEvent, nostrGroupId)
if (result is WelcomeResult.Joined) { if (result is WelcomeResult.Joined) {
@@ -152,6 +161,20 @@ class MarmotManager(
keyPackageEventId: HexKey, keyPackageEventId: HexKey,
relays: List<NormalizedRelayUrl>, relays: List<NormalizedRelayUrl>,
): Pair<OutboundGroupEvent, WelcomeDelivery?> { ): Pair<OutboundGroupEvent, WelcomeDelivery?> {
// Verify that the KeyPackage credential matches the expected member pubkey
val kp =
com.vitorpamplona.quartz.marmot.mls.messages.MlsKeyPackage.decodeTls(
com.vitorpamplona.quartz.marmot.mls.codec
.TlsReader(keyPackageBytes),
)
val credential = kp.leafNode.credential
require(credential is Credential.Basic) {
"KeyPackage must use BasicCredential"
}
require(credential.identity.toHexKey() == memberPubKey) {
"KeyPackage credential identity does not match memberPubKey"
}
val commitResult = groupManager.addMember(nostrGroupId, keyPackageBytes) val commitResult = groupManager.addMember(nostrGroupId, keyPackageBytes)
val commitEvent = outboundProcessor.buildCommitEvent(nostrGroupId, commitResult.commitBytes) val commitEvent = outboundProcessor.buildCommitEvent(nostrGroupId, commitResult.commitBytes)
@@ -182,9 +205,41 @@ class MarmotManager(
* Returns proposal bytes to publish (as a GroupEvent). * Returns proposal bytes to publish (as a GroupEvent).
*/ */
suspend fun leaveGroup(nostrGroupId: HexKey): OutboundGroupEvent { suspend fun leaveGroup(nostrGroupId: HexKey): OutboundGroupEvent {
val proposalBytes = groupManager.leaveGroup(nostrGroupId) // Build the outbound event BEFORE deleting group state (needs exporter secret)
val group =
groupManager.getGroup(nostrGroupId)
?: throw IllegalStateException("Not a member of group $nostrGroupId")
val proposalBytes = group.selfRemove()
val outboundEvent = outboundProcessor.buildCommitEvent(nostrGroupId, proposalBytes)
// Now clean up group state
groupManager.removeGroupState(nostrGroupId)
subscriptionManager.unsubscribeGroup(nostrGroupId) subscriptionManager.unsubscribeGroup(nostrGroupId)
return outboundProcessor.buildCommitEvent(nostrGroupId, proposalBytes) return outboundEvent
}
/**
* Remove a member from a group.
* Returns the commit GroupEvent to publish.
*/
suspend fun removeMember(
nostrGroupId: HexKey,
targetLeafIndex: Int,
): OutboundGroupEvent {
val commitResult = groupManager.removeMember(nostrGroupId, targetLeafIndex)
return outboundProcessor.buildCommitEvent(nostrGroupId, commitResult.commitBytes)
}
/**
* Update group metadata (name, description, etc.) via a GroupContextExtensions proposal.
* Creates a GCE proposal, commits it, and returns the commit event to publish.
*/
suspend fun updateGroupMetadata(
nostrGroupId: HexKey,
metadata: MarmotGroupData,
): OutboundGroupEvent {
val commitResult = groupManager.updateGroupExtensions(nostrGroupId, listOf(metadata.toExtension()))
return outboundProcessor.buildCommitEvent(nostrGroupId, commitResult.commitBytes)
} }
// --- KeyPackage Management --- // --- KeyPackage Management ---
@@ -47,6 +47,7 @@ class MarmotGroupChatroom(
var relays = MutableStateFlow<List<String>>(emptyList()) var relays = MutableStateFlow<List<String>>(emptyList())
var memberCount = MutableStateFlow(0) var memberCount = MutableStateFlow(0)
var newestMessage: Note? = null var newestMessage: Note? = null
val unreadCount = MutableStateFlow(0)
private var changesFlow: WeakReference<MutableSharedFlow<ListChange<Note>>> = WeakReference(null) private var changesFlow: WeakReference<MutableSharedFlow<ListChange<Note>>> = WeakReference(null)
@@ -73,6 +74,7 @@ class MarmotGroupChatroom(
newestMessage = msg newestMessage = msg
} }
unreadCount.value += 1
changesFlow.get()?.tryEmit(ListChange.Addition(msg)) changesFlow.get()?.tryEmit(ListChange.Addition(msg))
return true return true
} }
@@ -95,6 +97,10 @@ class MarmotGroupChatroom(
return false return false
} }
fun markAsRead() {
unreadCount.value = 0
}
fun pruneMessagesToTheLatestOnly(): Set<Note> { fun pruneMessagesToTheLatestOnly(): Set<Note> {
val sorted = messages.sortedWith(DefaultFeedOrder) val sorted = messages.sortedWith(DefaultFeedOrder)
val toKeep = val toKeep =
@@ -59,6 +59,11 @@ class MarmotGroupList {
} }
} }
fun removeGroup(nostrGroupId: HexKey) {
rooms.remove(nostrGroupId)
_groupListChanges.tryEmit(nostrGroupId)
}
fun allGroupIds(): List<HexKey> { fun allGroupIds(): List<HexKey> {
val result = mutableListOf<HexKey>() val result = mutableListOf<HexKey>()
rooms.forEach { key, _ -> result.add(key) } rooms.forEach { key, _ -> result.add(key) }
@@ -71,6 +71,13 @@ sealed class GroupEventResult {
val epoch: Long, val epoch: Long,
) : GroupEventResult() ) : GroupEventResult()
/**
* The event was already processed (duplicate).
*/
data class Duplicate(
val groupId: HexKey,
) : GroupEventResult()
/** /**
* The event could not be processed. * The event could not be processed.
*/ */
@@ -119,6 +126,16 @@ class MarmotInboundProcessor(
private val keyPackageRotationManager: KeyPackageRotationManager, private val keyPackageRotationManager: KeyPackageRotationManager,
) { ) {
private val commitTracker = CommitOrdering.EpochCommitTracker() 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). * Process an inbound GroupEvent (kind:445).
@@ -136,6 +153,13 @@ class MarmotInboundProcessor(
* @return the processing result * @return the processing result
*/ */
suspend fun processGroupEvent(groupEvent: GroupEvent): GroupEventResult { 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 = val groupId =
groupEvent.groupId() groupEvent.groupId()
?: return GroupEventResult.Error(null, "GroupEvent missing h tag (group ID)") ?: return GroupEventResult.Error(null, "GroupEvent missing h tag (group ID)")
@@ -144,10 +168,10 @@ class MarmotInboundProcessor(
return GroupEventResult.Error(groupId, "Not a member of group $groupId") return GroupEventResult.Error(groupId, "Not a member of group $groupId")
} }
return try { val result =
try {
// Step 1: Outer ChaCha20-Poly1305 decryption // Step 1: Outer ChaCha20-Poly1305 decryption
val exporterKey = groupManager.exporterSecret(groupId) val mlsBytes = decryptOuterLayer(groupId, groupEvent.encryptedContent())
val mlsBytes = GroupEventEncryption.decrypt(groupEvent.encryptedContent(), exporterKey)
// Step 2: Parse the MLS message // Step 2: Parse the MLS message
val mlsMessage = MlsMessage.decodeTls(TlsReader(mlsBytes)) val mlsMessage = MlsMessage.decodeTls(TlsReader(mlsBytes))
@@ -160,6 +184,22 @@ class MarmotInboundProcessor(
} catch (e: Exception) { } catch (e: Exception) {
GroupEventResult.Error(groupId, "Failed to process GroupEvent: ${e.message}", e) 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()
}
}
}
return result
} }
/** /**
@@ -223,18 +263,18 @@ class MarmotInboundProcessor(
epoch: Long, epoch: Long,
): GroupEventResult? { ): GroupEventResult? {
val winner = val winner =
commitTracker.resolve(epoch) commitTracker.resolve(groupId, epoch)
?: return null ?: return null
val result = applyCommit(groupId, winner) val result = applyCommit(groupId, winner)
commitTracker.clearEpoch(epoch) commitTracker.clearEpoch(groupId, epoch)
return result return result
} }
/** /**
* Get all epochs that have pending unresolved commits. * Get all (group, epoch) keys that have pending unresolved commits.
*/ */
fun pendingCommitEpochs(): Set<Long> = commitTracker.pendingEpochs() fun pendingCommitGroupEpochs(): Set<CommitOrdering.GroupEpochKey> = commitTracker.pendingGroupEpochs()
/** /**
* Clear all pending commit state. * Clear all pending commit state.
@@ -303,13 +343,13 @@ class MarmotInboundProcessor(
groupManager.getGroup(groupId) groupManager.getGroup(groupId)
?: return GroupEventResult.Error(groupId, "Group not found") ?: return GroupEventResult.Error(groupId, "Group not found")
val currentEpoch = group.epoch val currentEpoch = group.epoch
commitTracker.addCommit(currentEpoch, groupEvent) commitTracker.addCommit(groupId, currentEpoch, groupEvent)
// If this is the only commit for this epoch, apply immediately // If this is the only commit for this epoch, apply immediately
val pending = commitTracker.pendingForEpoch(currentEpoch) val pending = commitTracker.pendingForEpoch(groupId, currentEpoch)
return if (pending.size == 1) { return if (pending.size == 1) {
val result = applyCommit(groupId, groupEvent) val result = applyCommit(groupId, groupEvent)
commitTracker.clearEpoch(currentEpoch) commitTracker.clearEpoch(groupId, currentEpoch)
result result
} else { } else {
GroupEventResult.CommitPending(groupId, currentEpoch) GroupEventResult.CommitPending(groupId, currentEpoch)
@@ -321,8 +361,7 @@ class MarmotInboundProcessor(
commitEvent: GroupEvent, commitEvent: GroupEvent,
): GroupEventResult = ): GroupEventResult =
try { try {
val exporterKey = groupManager.exporterSecret(groupId) val mlsBytes = decryptOuterLayer(groupId, commitEvent.encryptedContent())
val mlsBytes = GroupEventEncryption.decrypt(commitEvent.encryptedContent(), exporterKey)
val mlsMessage = MlsMessage.decodeTls(TlsReader(mlsBytes)) val mlsMessage = MlsMessage.decodeTls(TlsReader(mlsBytes))
when (mlsMessage.wireFormat) { when (mlsMessage.wireFormat) {
@@ -360,15 +399,43 @@ class MarmotInboundProcessor(
GroupEventResult.Error(groupId, "Failed to apply commit: ${e.message}", e) GroupEventResult.Error(groupId, "Failed to apply commit: ${e.message}", e)
} }
/**
* Decrypt the outer ChaCha20-Poly1305 layer, trying the current epoch's
* exporter key first and falling back to retained epoch exporter keys.
*
* After a commit advances the epoch, late-arriving messages encrypted
* with the previous epoch's exporter key would fail without this fallback.
*/
private fun decryptOuterLayer(
groupId: HexKey,
encryptedContent: String,
): ByteArray {
// Try current epoch key first
try {
val exporterKey = groupManager.exporterSecret(groupId)
return GroupEventEncryption.decrypt(encryptedContent, exporterKey)
} catch (_: Exception) {
// Current epoch key failed — try retained epoch keys
}
// Try retained epoch exporter keys (most recent first)
val retainedKeys = groupManager.retainedExporterSecrets(groupId)
for (retainedKey in retainedKeys) {
try {
return GroupEventEncryption.decrypt(encryptedContent, retainedKey)
} catch (_: Exception) {
// This retained key didn't work — try the next one
}
}
// All keys exhausted — throw to let callers produce an error result
throw IllegalStateException(
"Outer decryption failed with current and ${retainedKeys.size} retained epoch key(s)",
)
}
private fun hexToBytes(hex: HexKey?): ByteArray { private fun hexToBytes(hex: HexKey?): ByteArray {
if (hex == null) return ByteArray(0) if (hex == null) return ByteArray(0)
return hex.hexToByteArray() return hex.hexToByteArray()
} }
companion object {
/**
* Check if an unwrapped event is a Marmot WelcomeEvent.
*/
fun isWelcomeEvent(event: Event): Boolean = event.kind == WelcomeEvent.KIND
}
} }
@@ -22,6 +22,8 @@ package com.vitorpamplona.quartz.marmot
import com.vitorpamplona.quartz.nip01Core.core.HexKey import com.vitorpamplona.quartz.nip01Core.core.HexKey
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
import kotlinx.coroutines.sync.Mutex
import kotlinx.coroutines.sync.withLock
/** /**
* Subscription state for a single Marmot group. * Subscription state for a single Marmot group.
@@ -52,6 +54,7 @@ data class GroupSubscriptionState(
class MarmotSubscriptionManager( class MarmotSubscriptionManager(
private val userPubKey: HexKey, private val userPubKey: HexKey,
) { ) {
private val mutex = Mutex()
private val groupSubscriptions = mutableMapOf<HexKey, GroupSubscriptionState>() private val groupSubscriptions = mutableMapOf<HexKey, GroupSubscriptionState>()
private var giftWrapSince: Long? = null private var giftWrapSince: Long? = null
@@ -62,10 +65,10 @@ class MarmotSubscriptionManager(
* @param nostrGroupId hex-encoded Nostr group ID * @param nostrGroupId hex-encoded Nostr group ID
* @param since optional timestamp to resume from (e.g., last seen event) * @param since optional timestamp to resume from (e.g., last seen event)
*/ */
fun subscribeGroup( suspend fun subscribeGroup(
nostrGroupId: HexKey, nostrGroupId: HexKey,
since: Long? = null, since: Long? = null,
) { ) = mutex.withLock {
groupSubscriptions[nostrGroupId] = groupSubscriptions[nostrGroupId] =
GroupSubscriptionState( GroupSubscriptionState(
nostrGroupId = nostrGroupId, nostrGroupId = nostrGroupId,
@@ -78,7 +81,8 @@ class MarmotSubscriptionManager(
* Unsubscribe from a group's events. * Unsubscribe from a group's events.
* Call this when leaving a group. * Call this when leaving a group.
*/ */
fun unsubscribeGroup(nostrGroupId: HexKey) { suspend fun unsubscribeGroup(nostrGroupId: HexKey) =
mutex.withLock {
groupSubscriptions.remove(nostrGroupId) groupSubscriptions.remove(nostrGroupId)
} }
@@ -86,17 +90,18 @@ class MarmotSubscriptionManager(
* Update the `since` timestamp for a group after processing events. * Update the `since` timestamp for a group after processing events.
* This ensures reconnections only fetch newer events. * This ensures reconnections only fetch newer events.
*/ */
fun updateGroupSince( suspend fun updateGroupSince(
nostrGroupId: HexKey, nostrGroupId: HexKey,
since: Long, since: Long,
) { ) = mutex.withLock {
groupSubscriptions[nostrGroupId]?.since = since groupSubscriptions[nostrGroupId]?.since = since
} }
/** /**
* Update the `since` timestamp for gift wrap subscriptions. * Update the `since` timestamp for gift wrap subscriptions.
*/ */
fun updateGiftWrapSince(since: Long) { suspend fun updateGiftWrapSince(since: Long) =
mutex.withLock {
giftWrapSince = since giftWrapSince = since
} }
@@ -185,25 +190,31 @@ class MarmotSubscriptionManager(
* *
* @param activeGroupIds the set of group IDs from [MlsGroupManager.activeGroupIds] * @param activeGroupIds the set of group IDs from [MlsGroupManager.activeGroupIds]
*/ */
fun syncWithGroupManager(activeGroupIds: Set<HexKey>) { suspend fun syncWithGroupManager(activeGroupIds: Set<HexKey>) =
mutex.withLock {
// Add new groups // Add new groups
for (groupId in activeGroupIds) { for (groupId in activeGroupIds) {
if (!groupSubscriptions.containsKey(groupId)) { if (!groupSubscriptions.containsKey(groupId)) {
subscribeGroup(groupId) groupSubscriptions[groupId] =
GroupSubscriptionState(
nostrGroupId = groupId,
active = true,
)
} }
} }
// Remove stale groups // Remove stale groups
val staleGroups = groupSubscriptions.keys - activeGroupIds val staleGroups = groupSubscriptions.keys - activeGroupIds
for (groupId in staleGroups) { for (groupId in staleGroups) {
unsubscribeGroup(groupId) groupSubscriptions.remove(groupId)
} }
} }
/** /**
* Clear all subscription state. * Clear all subscription state.
*/ */
fun clear() { suspend fun clear() =
mutex.withLock {
groupSubscriptions.clear() groupSubscriptions.clear()
giftWrapSince = null giftWrapSince = null
} }
@@ -31,6 +31,8 @@ import com.vitorpamplona.quartz.marmot.mls.tree.LeafNode
import com.vitorpamplona.quartz.marmot.mls.tree.LeafNodeSource import com.vitorpamplona.quartz.marmot.mls.tree.LeafNodeSource
import com.vitorpamplona.quartz.marmot.mls.tree.Lifetime import com.vitorpamplona.quartz.marmot.mls.tree.Lifetime
import com.vitorpamplona.quartz.utils.TimeUtils import com.vitorpamplona.quartz.utils.TimeUtils
import kotlinx.coroutines.sync.Mutex
import kotlinx.coroutines.sync.withLock
/** /**
* Manages KeyPackage creation and rotation lifecycle (MIP-00). * Manages KeyPackage creation and rotation lifecycle (MIP-00).
@@ -50,6 +52,7 @@ import com.vitorpamplona.quartz.utils.TimeUtils
* KeyPackage slots, rotating consumed ones promptly. * KeyPackage slots, rotating consumed ones promptly.
*/ */
class KeyPackageRotationManager { class KeyPackageRotationManager {
private val mutex = Mutex()
private val activeBundles = mutableMapOf<String, KeyPackageBundle>() private val activeBundles = mutableMapOf<String, KeyPackageBundle>()
private val pendingRotations = mutableSetOf<String>() private val pendingRotations = mutableSetOf<String>()
@@ -60,7 +63,7 @@ class KeyPackageRotationManager {
* @param dTagSlot the d-tag slot for addressable replacement * @param dTagSlot the d-tag slot for addressable replacement
* @return a [KeyPackageBundle] containing the KeyPackage and all private keys * @return a [KeyPackageBundle] containing the KeyPackage and all private keys
*/ */
fun generateKeyPackage( suspend fun generateKeyPackage(
identity: ByteArray, identity: ByteArray,
dTagSlot: String = KeyPackageUtils.PRIMARY_SLOT, dTagSlot: String = KeyPackageUtils.PRIMARY_SLOT,
): KeyPackageBundle { ): KeyPackageBundle {
@@ -93,7 +96,9 @@ class KeyPackageRotationManager {
) )
val bundle = KeyPackageBundle(keyPackage, initKp.privateKey, encKp.privateKey, sigKp.privateKey) val bundle = KeyPackageBundle(keyPackage, initKp.privateKey, encKp.privateKey, sigKp.privateKey)
mutex.withLock {
activeBundles[dTagSlot] = bundle activeBundles[dTagSlot] = bundle
}
return bundle return bundle
} }
@@ -117,7 +122,8 @@ class KeyPackageRotationManager {
* The slot will be included in [pendingRotationSlots] and should be * The slot will be included in [pendingRotationSlots] and should be
* rotated by the caller. * rotated by the caller.
*/ */
fun markConsumed(dTagSlot: String) { suspend fun markConsumed(dTagSlot: String) =
mutex.withLock {
activeBundles.remove(dTagSlot) activeBundles.remove(dTagSlot)
pendingRotations.add(dTagSlot) pendingRotations.add(dTagSlot)
} }
@@ -125,7 +131,8 @@ class KeyPackageRotationManager {
/** /**
* Mark a slot as consumed by looking up the KeyPackage reference. * Mark a slot as consumed by looking up the KeyPackage reference.
*/ */
fun markConsumedByRef(keyPackageRef: ByteArray) { suspend fun markConsumedByRef(keyPackageRef: ByteArray) =
mutex.withLock {
val entry = val entry =
activeBundles.entries.find { (_, bundle) -> activeBundles.entries.find { (_, bundle) ->
bundle.keyPackage.reference().contentEquals(keyPackageRef) bundle.keyPackage.reference().contentEquals(keyPackageRef)
@@ -145,7 +152,8 @@ class KeyPackageRotationManager {
* Clear a slot from the pending rotation set after a new KeyPackage * Clear a slot from the pending rotation set after a new KeyPackage
* has been published. * has been published.
*/ */
fun clearPendingRotation(dTagSlot: String) { suspend fun clearPendingRotation(dTagSlot: String) =
mutex.withLock {
pendingRotations.remove(dTagSlot) pendingRotations.remove(dTagSlot)
} }
@@ -167,12 +175,14 @@ class KeyPackageRotationManager {
* @param dTagSlot the slot to rotate * @param dTagSlot the slot to rotate
* @return the new [KeyPackageBundle] ready for publishing * @return the new [KeyPackageBundle] ready for publishing
*/ */
fun rotateSlot( suspend fun rotateSlot(
identity: ByteArray, identity: ByteArray,
dTagSlot: String, dTagSlot: String,
): KeyPackageBundle { ): KeyPackageBundle {
val bundle = generateKeyPackage(identity, dTagSlot) val bundle = generateKeyPackage(identity, dTagSlot)
mutex.withLock {
pendingRotations.remove(dTagSlot) pendingRotations.remove(dTagSlot)
}
return bundle return bundle
} }
@@ -22,8 +22,10 @@ package com.vitorpamplona.quartz.marmot.mip01Groups
import androidx.compose.runtime.Immutable import androidx.compose.runtime.Immutable
import com.vitorpamplona.quartz.marmot.mls.codec.TlsReader import com.vitorpamplona.quartz.marmot.mls.codec.TlsReader
import com.vitorpamplona.quartz.marmot.mls.codec.TlsWriter
import com.vitorpamplona.quartz.marmot.mls.tree.Extension import com.vitorpamplona.quartz.marmot.mls.tree.Extension
import com.vitorpamplona.quartz.nip01Core.core.HexKey import com.vitorpamplona.quartz.nip01Core.core.HexKey
import com.vitorpamplona.quartz.nip01Core.core.hexToByteArray
import com.vitorpamplona.quartz.nip01Core.core.toHexKey import com.vitorpamplona.quartz.nip01Core.core.toHexKey
/** /**
@@ -89,6 +91,45 @@ data class MarmotGroupData(
/** Whether this group has an encrypted image set */ /** Whether this group has an encrypted image set */
fun hasImage(): Boolean = imageHash != null && imageKey != null && imageNonce != null fun hasImage(): Boolean = imageHash != null && imageKey != null && imageNonce != null
/**
* Encode this MarmotGroupData to TLS wire format bytes.
* Mirrors the [decodeTls] format.
*/
fun encodeTls(): ByteArray {
val writer = TlsWriter()
writer.putUint16(version)
writer.putBytes(nostrGroupId.hexToByteArray())
writer.putOpaque2(name.encodeToByteArray())
writer.putOpaque2(description.encodeToByteArray())
// Admin pubkeys: concatenated 32-byte keys within a length-prefixed block
val adminBytes = ByteArray(adminPubkeys.size * 32)
adminPubkeys.forEachIndexed { index, key ->
key.hexToByteArray().copyInto(adminBytes, index * 32)
}
writer.putOpaque2(adminBytes)
// Relays: length-prefixed block of length-prefixed UTF-8 strings
val relayWriter = TlsWriter()
for (relay in relays) {
relayWriter.putOpaque2(relay.encodeToByteArray())
}
writer.putOpaque2(relayWriter.toByteArray())
// Optional image fields
writer.putOpaque2(imageHash?.hexToByteArray() ?: ByteArray(0))
writer.putOpaque2(imageKey ?: ByteArray(0))
writer.putOpaque2(imageNonce ?: ByteArray(0))
writer.putOpaque2(imageUploadKey ?: ByteArray(0))
return writer.toByteArray()
}
/**
* Convert this MarmotGroupData to an MLS Extension for use in GroupContextExtensions proposals.
*/
fun toExtension(): Extension = Extension(EXTENSION_ID_INT, encodeTls())
companion object { companion object {
const val CURRENT_VERSION = 2 const val CURRENT_VERSION = 2
@@ -67,57 +67,95 @@ object CommitOrdering {
} }
/** /**
* Tracks pending commits per epoch and resolves conflicts. * Key for tracking pending commits: must be unique per (group, epoch) pair
* * since epoch numbers are not globally unique across different groups.
* Accumulate commits as they arrive from relays, then call [resolve]
* to determine which commit wins for each epoch.
*/ */
class EpochCommitTracker { data class GroupEpochKey(
private val pendingByEpoch = mutableMapOf<Long, MutableList<GroupEvent>>() val groupId: String,
val epoch: Long,
)
/** /**
* Adds a commit for a given epoch. * Tracks pending commits per (group, epoch) and resolves conflicts.
* *
* Accumulate commits as they arrive from relays, then call [resolve]
* to determine which commit wins for each (group, epoch).
*/
class EpochCommitTracker {
private val pendingByGroupEpoch = mutableMapOf<GroupEpochKey, MutableList<GroupEvent>>()
companion object {
/** Maximum number of (group, epoch) entries to track before evicting oldest. */
const val MAX_TRACKED_EPOCHS = 1000
}
/**
* Adds a commit for a given group and epoch.
*
* @param groupId the Nostr group ID
* @param epoch the MLS epoch number this commit targets * @param epoch the MLS epoch number this commit targets
* @param commit the GroupEvent containing the commit * @param commit the GroupEvent containing the commit
*/ */
fun addCommit( fun addCommit(
groupId: String,
epoch: Long, epoch: Long,
commit: GroupEvent, commit: GroupEvent,
) { ) {
pendingByEpoch.getOrPut(epoch) { mutableListOf() }.add(commit) val key = GroupEpochKey(groupId, epoch)
pendingByGroupEpoch.getOrPut(key) { mutableListOf() }.add(commit)
// Evict oldest entries if the tracker grows too large
if (pendingByGroupEpoch.size > MAX_TRACKED_EPOCHS) {
val oldestKeys =
pendingByGroupEpoch.keys
.sortedBy { it.epoch }
.take(pendingByGroupEpoch.size - MAX_TRACKED_EPOCHS)
for (oldKey in oldestKeys) {
pendingByGroupEpoch.remove(oldKey)
}
}
} }
/** /**
* Returns pending commits for a specific epoch. * Returns pending commits for a specific group and epoch.
*/ */
fun pendingForEpoch(epoch: Long): List<GroupEvent> = pendingByEpoch[epoch] ?: emptyList() fun pendingForEpoch(
groupId: String,
epoch: Long,
): List<GroupEvent> = pendingByGroupEpoch[GroupEpochKey(groupId, epoch)] ?: emptyList()
/** /**
* Resolves the winning commit for a specific epoch. * Resolves the winning commit for a specific group and epoch.
* *
* @param groupId the Nostr group ID
* @param epoch the MLS epoch to resolve * @param epoch the MLS epoch to resolve
* @return the winning commit, or null if no commits exist for this epoch * @return the winning commit, or null if no commits exist for this (group, epoch)
*/ */
fun resolve(epoch: Long): GroupEvent? = selectWinner(pendingByEpoch[epoch] ?: emptyList()) fun resolve(
groupId: String,
epoch: Long,
): GroupEvent? = selectWinner(pendingByGroupEpoch[GroupEpochKey(groupId, epoch)] ?: emptyList())
/** /**
* Clears pending commits for an epoch after it has been resolved. * Clears pending commits for a (group, epoch) after it has been resolved.
*/ */
fun clearEpoch(epoch: Long) { fun clearEpoch(
pendingByEpoch.remove(epoch) groupId: String,
epoch: Long,
) {
pendingByGroupEpoch.remove(GroupEpochKey(groupId, epoch))
} }
/** /**
* Returns all epochs that have pending commits. * Returns all (group, epoch) keys that have pending commits.
*/ */
fun pendingEpochs(): Set<Long> = pendingByEpoch.keys.toSet() fun pendingGroupEpochs(): Set<GroupEpochKey> = pendingByGroupEpoch.keys.toSet()
/** /**
* Clears all pending state. * Clears all pending state.
*/ */
fun clear() { fun clear() {
pendingByEpoch.clear() pendingByGroupEpoch.clear()
} }
} }
} }
@@ -31,6 +31,11 @@ class TlsReader(
private var position: Int = 0, private var position: Int = 0,
private val limit: Int = data.size, private val limit: Int = data.size,
) { ) {
companion object {
/** Maximum allowed size for a single opaque field (1 MB) */
const val MAX_OPAQUE_SIZE = 1_048_576
}
val remaining: Int get() = limit - position val remaining: Int get() = limit - position
val hasRemaining: Boolean get() = position < limit val hasRemaining: Boolean get() = position < limit
@@ -87,12 +92,18 @@ class TlsReader(
/** Read a variable-length opaque with 2-byte length prefix */ /** Read a variable-length opaque with 2-byte length prefix */
fun readOpaque2(): ByteArray { fun readOpaque2(): ByteArray {
val length = readUint16() val length = readUint16()
require(length <= MAX_OPAQUE_SIZE) {
"Opaque2 length $length exceeds maximum allowed size $MAX_OPAQUE_SIZE"
}
return readBytes(length) return readBytes(length)
} }
/** Read a variable-length opaque with 4-byte length prefix */ /** Read a variable-length opaque with 4-byte length prefix */
fun readOpaque4(): ByteArray { fun readOpaque4(): ByteArray {
val length = readUint32().toInt() val length = readUint32().toInt()
require(length <= MAX_OPAQUE_SIZE) {
"Opaque4 length $length exceeds maximum allowed size $MAX_OPAQUE_SIZE"
}
return readBytes(length) return readBytes(length)
} }
@@ -130,6 +141,9 @@ class TlsReader(
/** Read a variable-length opaque with QUIC-style VarInt length prefix */ /** Read a variable-length opaque with QUIC-style VarInt length prefix */
fun readOpaqueVarInt(): ByteArray { fun readOpaqueVarInt(): ByteArray {
val length = readVarInt() val length = readVarInt()
require(length <= MAX_OPAQUE_SIZE) {
"OpaqueVarInt length $length exceeds maximum allowed size $MAX_OPAQUE_SIZE"
}
return readBytes(length) return readBytes(length)
} }
@@ -105,6 +105,9 @@ class MlsGroup private constructor(
private val pskStore: MutableMap<String, ByteArray> = mutableMapOf(), private val pskStore: MutableMap<String, ByteArray> = mutableMapOf(),
private val pendingProposals: MutableList<PendingProposal> = mutableListOf(), private val pendingProposals: MutableList<PendingProposal> = mutableListOf(),
private val sentKeys: MutableMap<Int, com.vitorpamplona.quartz.marmot.mls.schedule.KeyNonceGeneration> = mutableMapOf(), private val sentKeys: MutableMap<Int, com.vitorpamplona.quartz.marmot.mls.schedule.KeyNonceGeneration> = mutableMapOf(),
/** Staged keys from proposeSigningKeyRotation — only promoted on successful commit */
private var pendingSigningKey: ByteArray? = null,
private var pendingEncryptionKey: ByteArray? = null,
) { ) {
val groupId: ByteArray get() = groupContext.groupId val groupId: ByteArray get() = groupContext.groupId
val epoch: Long get() = groupContext.epoch val epoch: Long get() = groupContext.epoch
@@ -150,6 +153,7 @@ class MlsGroup private constructor(
senderDataSecret = epochSecrets.senderDataSecret, senderDataSecret = epochSecrets.senderDataSecret,
encryptionSecret = epochSecrets.encryptionSecret, encryptionSecret = epochSecrets.encryptionSecret,
leafCount = tree.leafCount, leafCount = tree.leafCount,
exporterSecret = epochSecrets.exporterSecret,
) )
val memberCount: Int val memberCount: Int
@@ -293,13 +297,23 @@ class MlsGroup private constructor(
val proposal = Proposal.Update(newLeafNode) val proposal = Proposal.Update(newLeafNode)
pendingProposals.add(PendingProposal(proposal, myLeafIndex)) pendingProposals.add(PendingProposal(proposal, myLeafIndex))
// Stage the new keys — they take effect when commit() is called // Stage the new keys — they take effect only when commit() succeeds
signingPrivateKey = newSigKp.privateKey pendingSigningKey = newSigKp.privateKey
encryptionPrivateKey = newEncKp.privateKey pendingEncryptionKey = newEncKp.privateKey
return proposal return proposal
} }
/**
* Create a GroupContextExtensions proposal to update the group's extensions.
* Used for changing group metadata (name, description, etc.) via MIP-01.
*/
fun proposeGroupContextExtensions(extensions: List<Extension>): Proposal.GroupContextExtensions {
val proposal = Proposal.GroupContextExtensions(extensions)
pendingProposals.add(PendingProposal(proposal, myLeafIndex))
return proposal
}
/** /**
* Create a PSK proposal to include a pre-shared key in the next epoch. * Create a PSK proposal to include a pre-shared key in the next epoch.
* The PSK must be registered via registerPsk() before committing. * The PSK must be registered via registerPsk() before committing.
@@ -347,12 +361,15 @@ class MlsGroup private constructor(
val addedMembers = mutableListOf<Pair<Int, MlsKeyPackage>>() val addedMembers = mutableListOf<Pair<Int, MlsKeyPackage>>()
for (pending in proposals) { for (pending in proposals) {
val p = pending.proposal val p = pending.proposal
// Track added members for Welcome generation (before apply changes leaf count)
if (p is Proposal.Add) { if (p is Proposal.Add) {
addedMembers.add(tree.leafCount to p.keyPackage) // applyProposal calls tree.addLeaf() which returns the actual leaf index
} // (may reuse a blank slot instead of appending)
val leafIndex = applyProposalAdd(p)
addedMembers.add(leafIndex to p.keyPackage)
} else {
applyProposal(p, pending.senderLeafIndex) applyProposal(p, pending.senderLeafIndex)
} }
}
// Generate new path secrets on the updated tree // Generate new path secrets on the updated tree
val leafSecret = MlsCryptoProvider.randomBytes(MlsCryptoProvider.HASH_OUTPUT_LENGTH) val leafSecret = MlsCryptoProvider.randomBytes(MlsCryptoProvider.HASH_OUTPUT_LENGTH)
@@ -465,6 +482,12 @@ class MlsGroup private constructor(
null null
} }
// Promote staged keys from proposeSigningKeyRotation if present
pendingSigningKey?.let { signingPrivateKey = it }
pendingEncryptionKey?.let { encryptionPrivateKey = it }
pendingSigningKey = null
pendingEncryptionKey = null
pendingProposals.clear() pendingProposals.clear()
sentKeys.clear() sentKeys.clear()
@@ -478,6 +501,15 @@ class MlsGroup private constructor(
* Encrypt an application message as a PrivateMessage. * Encrypt an application message as a PrivateMessage.
*/ */
fun encrypt(plaintext: ByteArray): ByteArray { fun encrypt(plaintext: ByteArray): ByteArray {
// Trim sentKeys if it grows too large
if (sentKeys.size > MAX_SENT_KEYS) {
val sortedKeys = sentKeys.keys.sorted()
val toRemove = sortedKeys.take(sentKeys.size - MAX_SENT_KEYS)
for (key in toRemove) {
sentKeys.remove(key)
}
}
val kng = secretTree.nextApplicationKeyNonce(myLeafIndex) val kng = secretTree.nextApplicationKeyNonce(myLeafIndex)
sentKeys[kng.generation] = kng sentKeys[kng.generation] = kng
val ciphertext = MlsCryptoProvider.aeadEncrypt(kng.key, kng.nonce, ByteArray(0), plaintext) val ciphertext = MlsCryptoProvider.aeadEncrypt(kng.key, kng.nonce, ByteArray(0), plaintext)
@@ -605,6 +637,13 @@ class MlsGroup private constructor(
senderLeafIndex: Int, senderLeafIndex: Int,
confirmationTag: ByteArray? = null, confirmationTag: ByteArray? = null,
) { ) {
require(senderLeafIndex >= 0 && senderLeafIndex < tree.leafCount) {
"Invalid sender leaf index: $senderLeafIndex"
}
require(tree.getLeaf(senderLeafIndex) != null) {
"Sender leaf is blank at index $senderLeafIndex"
}
val commit = Commit.decodeTls(TlsReader(commitBytes)) val commit = Commit.decodeTls(TlsReader(commitBytes))
// Apply proposals (resolve references from pending pool) // Apply proposals (resolve references from pending pool)
@@ -641,6 +680,11 @@ class MlsGroup private constructor(
tree.setLeaf(senderLeafIndex, updatePath.leafNode) tree.setLeaf(senderLeafIndex, updatePath.leafNode)
tree.applyUpdatePath(senderLeafIndex, updatePath.nodes) tree.applyUpdatePath(senderLeafIndex, updatePath.nodes)
// Verify parent hash chain (RFC 9420 Section 7.9.2)
require(verifyParentHash(senderLeafIndex, updatePath)) {
"Parent hash verification failed for UpdatePath"
}
// Decrypt path secret from our copath node // Decrypt path secret from our copath node
val directPath = BinaryTree.directPath(senderLeafIndex, tree.leafCount) val directPath = BinaryTree.directPath(senderLeafIndex, tree.leafCount)
val myPath = BinaryTree.directPath(myLeafIndex, tree.leafCount) val myPath = BinaryTree.directPath(myLeafIndex, tree.leafCount)
@@ -726,7 +770,7 @@ class MlsGroup private constructor(
// Verify confirmation tag (RFC 9420 Section 6.1) // Verify confirmation tag (RFC 9420 Section 6.1)
if (confirmationTag != null) { if (confirmationTag != null) {
val expectedTag = computeConfirmationTag(epochSecrets.confirmationKey, newConfirmedTranscriptHash) val expectedTag = computeConfirmationTag(epochSecrets.confirmationKey, newConfirmedTranscriptHash)
require(confirmationTag.contentEquals(expectedTag)) { require(constantTimeEquals(confirmationTag, expectedTag)) {
"Confirmation tag verification failed" "Confirmation tag verification failed"
} }
} }
@@ -846,7 +890,7 @@ class MlsGroup private constructor(
for (pskProposal in pskProposals) { for (pskProposal in pskProposals) {
val pskValue = val pskValue =
pskStore[pskProposal.pskId.toHexKey()] pskStore[pskProposal.pskId.toHexKey()]
?: ByteArray(MlsCryptoProvider.HASH_OUTPUT_LENGTH) // Unknown PSK = zeros ?: throw IllegalStateException("PSK not found in store: ${pskProposal.pskId.toHexKey()}")
pskSecret = MlsCryptoProvider.hkdfExtract(pskSecret, pskValue) pskSecret = MlsCryptoProvider.hkdfExtract(pskSecret, pskValue)
} }
return pskSecret return pskSecret
@@ -978,7 +1022,37 @@ class MlsGroup private constructor(
.MacInstance("HmacSHA256", epochSecrets.membershipKey) .MacInstance("HmacSHA256", epochSecrets.membershipKey)
mac.update(authenticatedContentBytes) mac.update(authenticatedContentBytes)
val expectedTag = mac.doFinal() val expectedTag = mac.doFinal()
return expectedTag.contentEquals(membershipTag) return constantTimeEquals(expectedTag, membershipTag)
}
/**
* Constant-time byte array comparison to prevent timing side-channels.
* Returns true only if both arrays have the same length and contents.
*/
private fun constantTimeEquals(
a: ByteArray,
b: ByteArray,
): Boolean {
if (a.size != b.size) return false
var result = 0
for (i in a.indices) {
result = result or (a[i].toInt() xor b[i].toInt())
}
return result == 0
}
/**
* Apply an Add proposal and return the assigned leaf index.
*/
private fun applyProposalAdd(proposal: Proposal.Add): Int {
val lifetime = proposal.keyPackage.leafNode.lifetime
if (lifetime != null) {
val now = TimeUtils.now()
require(now >= lifetime.notBefore && now <= lifetime.notAfter) {
"KeyPackage lifetime expired or not yet valid"
}
}
return tree.addLeaf(proposal.keyPackage.leafNode)
} }
private fun applyProposal( private fun applyProposal(
@@ -987,15 +1061,7 @@ class MlsGroup private constructor(
) { ) {
when (proposal) { when (proposal) {
is Proposal.Add -> { is Proposal.Add -> {
// Validate KeyPackage lifetime (RFC 9420 Section 10.1) applyProposalAdd(proposal)
val lifetime = proposal.keyPackage.leafNode.lifetime
if (lifetime != null) {
val now = TimeUtils.now()
require(now >= lifetime.notBefore && now <= lifetime.notAfter) {
"KeyPackage lifetime expired or not yet valid"
}
}
tree.addLeaf(proposal.keyPackage.leafNode)
} }
is Proposal.Remove -> { is Proposal.Remove -> {
@@ -1143,6 +1209,7 @@ class MlsGroup private constructor(
} }
companion object { companion object {
private const val MAX_SENT_KEYS = 10_000
private const val RATCHET_TREE_EXTENSION_TYPE = 0x0001 private const val RATCHET_TREE_EXTENSION_TYPE = 0x0001
private const val REQUIRED_CAPABILITIES_EXTENSION_TYPE = 0x0002 private const val REQUIRED_CAPABILITIES_EXTENSION_TYPE = 0x0002
private const val EXTERNAL_PUB_EXTENSION_TYPE = 0x0003 private const val EXTERNAL_PUB_EXTENSION_TYPE = 0x0003
@@ -1300,7 +1367,7 @@ class MlsGroup private constructor(
// Find our leaf index by matching our signature key // Find our leaf index by matching our signature key
val mySignatureKey = Ed25519.publicFromPrivate(bundle.signaturePrivateKey) val mySignatureKey = Ed25519.publicFromPrivate(bundle.signaturePrivateKey)
var myLeafIndex = 0 var myLeafIndex = -1
for (i in 0 until tree.leafCount) { for (i in 0 until tree.leafCount) {
val leaf = tree.getLeaf(i) val leaf = tree.getLeaf(i)
if (leaf != null && leaf.signatureKey.contentEquals(mySignatureKey)) { if (leaf != null && leaf.signatureKey.contentEquals(mySignatureKey)) {
@@ -1308,14 +1375,16 @@ class MlsGroup private constructor(
break break
} }
} }
require(myLeafIndex >= 0) { "Joiner's signature key not found in ratchet tree" }
// Verify GroupInfo signature using signer's key from the tree // Verify GroupInfo signature using signer's key from the tree
val signerLeaf = tree.getLeaf(groupInfo.signer) val signerLeaf = tree.getLeaf(groupInfo.signer)
if (signerLeaf != null) { requireNotNull(signerLeaf) {
"Signer leaf is null at index ${groupInfo.signer} — cannot verify GroupInfo signature"
}
require(groupInfo.verifySignature(signerLeaf.signatureKey)) { require(groupInfo.verifySignature(signerLeaf.signatureKey)) {
"Invalid GroupInfo signature" "Invalid GroupInfo signature"
} }
}
// Derive epoch secrets directly from memberSecret (RFC 9420 Section 8.3) // Derive epoch secrets directly from memberSecret (RFC 9420 Section 8.3)
// For Welcome, epoch_secret = ExpandWithLabel(member_secret, "epoch", GroupContext, Nh) // For Welcome, epoch_secret = ExpandWithLabel(member_secret, "epoch", GroupContext, Nh)
@@ -1351,6 +1420,15 @@ class MlsGroup private constructor(
val secretTree = SecretTree(epochSecrets.encryptionSecret, tree.leafCount) val secretTree = SecretTree(epochSecrets.encryptionSecret, tree.leafCount)
// Compute interim_transcript_hash from confirmed_transcript_hash + confirmation_tag
val confirmMac = MacInstance("HmacSHA256", epochSecrets.confirmationKey)
confirmMac.update(groupContext.confirmedTranscriptHash)
val confirmationTag = confirmMac.doFinal()
val interimInput = TlsWriter()
interimInput.putBytes(groupContext.confirmedTranscriptHash)
interimInput.putOpaqueVarInt(confirmationTag)
val interimTranscriptHash = MlsCryptoProvider.hash(interimInput.toByteArray())
return MlsGroup( return MlsGroup(
groupContext = groupContext, groupContext = groupContext,
tree = tree, tree = tree,
@@ -1360,7 +1438,7 @@ class MlsGroup private constructor(
initSecret = epochSecrets.initSecret, initSecret = epochSecrets.initSecret,
signingPrivateKey = bundle.signaturePrivateKey, signingPrivateKey = bundle.signaturePrivateKey,
encryptionPrivateKey = bundle.encryptionPrivateKey, encryptionPrivateKey = bundle.encryptionPrivateKey,
interimTranscriptHash = ByteArray(0), interimTranscriptHash = interimTranscriptHash,
) )
} }
@@ -25,8 +25,13 @@ import com.vitorpamplona.quartz.marmot.mls.codec.TlsWriter
import com.vitorpamplona.quartz.marmot.mls.crypto.MlsCryptoProvider import com.vitorpamplona.quartz.marmot.mls.crypto.MlsCryptoProvider
import com.vitorpamplona.quartz.marmot.mls.messages.CommitResult import com.vitorpamplona.quartz.marmot.mls.messages.CommitResult
import com.vitorpamplona.quartz.marmot.mls.messages.KeyPackageBundle import com.vitorpamplona.quartz.marmot.mls.messages.KeyPackageBundle
import com.vitorpamplona.quartz.marmot.mls.schedule.KeySchedule
import com.vitorpamplona.quartz.marmot.mls.schedule.SecretTree import com.vitorpamplona.quartz.marmot.mls.schedule.SecretTree
import com.vitorpamplona.quartz.marmot.mls.tree.Extension
import com.vitorpamplona.quartz.nip01Core.core.HexKey import com.vitorpamplona.quartz.nip01Core.core.HexKey
import com.vitorpamplona.quartz.utils.Log
import kotlinx.coroutines.sync.Mutex
import kotlinx.coroutines.sync.withLock
/** /**
* High-level coordinator for MLS group lifecycle and state management. * High-level coordinator for MLS group lifecycle and state management.
@@ -77,8 +82,11 @@ import com.vitorpamplona.quartz.nip01Core.core.HexKey
* The epoch secret retention window is [EPOCH_RETENTION_WINDOW] = 2, meaning secrets * The epoch secret retention window is [EPOCH_RETENTION_WINDOW] = 2, meaning secrets
* for the current and previous epoch are kept for late-message decryption. * for the current and previous epoch are kept for late-message decryption.
* *
* Thread safety: All public methods are suspending and should be called * Thread safety: All suspending mutation methods are guarded by a [Mutex]
* from a single coroutine context (e.g., the Account's scope). * 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 MlsGroup The low-level MLS state machine
* @see MlsGroupStateStore Storage abstraction for group state persistence * @see MlsGroupStateStore Storage abstraction for group state persistence
@@ -86,6 +94,7 @@ import com.vitorpamplona.quartz.nip01Core.core.HexKey
class MlsGroupManager( class MlsGroupManager(
private val store: MlsGroupStateStore, private val store: MlsGroupStateStore,
) { ) {
private val mutex = Mutex()
private val groups = mutableMapOf<HexKey, MlsGroup>() private val groups = mutableMapOf<HexKey, MlsGroup>()
private val retainedEpochs = mutableMapOf<HexKey, MutableList<RetainedEpochSecrets>>() private val retainedEpochs = mutableMapOf<HexKey, MutableList<RetainedEpochSecrets>>()
@@ -93,7 +102,8 @@ class MlsGroupManager(
* Restore all groups from persistent storage on startup. * Restore all groups from persistent storage on startup.
* Call this once during Account initialization. * Call this once during Account initialization.
*/ */
suspend fun restoreAll() { suspend fun restoreAll() =
mutex.withLock {
val groupIds = store.listGroups() val groupIds = store.listGroups()
for (nostrGroupId in groupIds) { for (nostrGroupId in groupIds) {
try { try {
@@ -109,8 +119,13 @@ class MlsGroupManager(
.map { RetainedEpochSecrets.decodeTls(TlsReader(it)) } .map { RetainedEpochSecrets.decodeTls(TlsReader(it)) }
.toMutableList() .toMutableList()
} }
} catch (_: Exception) { } catch (e: Exception) {
// Corrupted state — remove it so it doesn't block future joins // Corrupted state — log and remove it so it doesn't block future joins
Log.e(
"MlsGroupManager",
"Corrupted state for group $nostrGroupId, deleting: ${e.message}",
e,
)
store.delete(nostrGroupId) store.delete(nostrGroupId)
} }
} }
@@ -145,11 +160,12 @@ class MlsGroupManager(
nostrGroupId: HexKey, nostrGroupId: HexKey,
identity: ByteArray, identity: ByteArray,
signingKey: ByteArray? = null, signingKey: ByteArray? = null,
): MlsGroup { ): MlsGroup =
mutex.withLock {
val group = MlsGroup.create(identity, signingKey) val group = MlsGroup.create(identity, signingKey)
groups[nostrGroupId] = group groups[nostrGroupId] = group
persistGroup(nostrGroupId) persistGroup(nostrGroupId)
return group group
} }
// --- Joining --- // --- Joining ---
@@ -172,7 +188,8 @@ class MlsGroupManager(
nostrGroupId: HexKey, nostrGroupId: HexKey,
welcomeBytes: ByteArray, welcomeBytes: ByteArray,
bundle: KeyPackageBundle, bundle: KeyPackageBundle,
): MlsGroup { ): MlsGroup =
mutex.withLock {
val group = MlsGroup.processWelcome(welcomeBytes, bundle) val group = MlsGroup.processWelcome(welcomeBytes, bundle)
groups[nostrGroupId] = group groups[nostrGroupId] = group
@@ -180,7 +197,7 @@ class MlsGroupManager(
// reused. Caller must discard the bundle and rotate KeyPackages. // reused. Caller must discard the bundle and rotate KeyPackages.
persistGroup(nostrGroupId) persistGroup(nostrGroupId)
return group group
} }
/** /**
@@ -197,11 +214,12 @@ class MlsGroupManager(
groupInfoBytes: ByteArray, groupInfoBytes: ByteArray,
identity: ByteArray, identity: ByteArray,
signingKey: ByteArray? = null, signingKey: ByteArray? = null,
): Pair<MlsGroup, ByteArray> { ): Pair<MlsGroup, ByteArray> =
mutex.withLock {
val (group, commitBytes) = MlsGroup.externalJoin(groupInfoBytes, identity, signingKey) val (group, commitBytes) = MlsGroup.externalJoin(groupInfoBytes, identity, signingKey)
groups[nostrGroupId] = group groups[nostrGroupId] = group
persistGroup(nostrGroupId) persistGroup(nostrGroupId)
return Pair(group, commitBytes) Pair(group, commitBytes)
} }
// --- Epoch Transitions --- // --- Epoch Transitions ---
@@ -215,7 +233,8 @@ class MlsGroupManager(
* @param nostrGroupId hex-encoded Nostr group ID * @param nostrGroupId hex-encoded Nostr group ID
* @return the [CommitResult] to publish * @return the [CommitResult] to publish
*/ */
suspend fun commit(nostrGroupId: HexKey): CommitResult { suspend fun commit(nostrGroupId: HexKey): CommitResult =
mutex.withLock {
val group = requireGroup(nostrGroupId) val group = requireGroup(nostrGroupId)
// Retain current epoch secrets before transition // Retain current epoch secrets before transition
@@ -223,7 +242,7 @@ class MlsGroupManager(
val result = group.commit() val result = group.commit()
persistGroup(nostrGroupId) persistGroup(nostrGroupId)
return result result
} }
/** /**
@@ -239,7 +258,7 @@ class MlsGroupManager(
commitBytes: ByteArray, commitBytes: ByteArray,
senderLeafIndex: Int, senderLeafIndex: Int,
confirmationTag: ByteArray? = null, confirmationTag: ByteArray? = null,
) { ) = mutex.withLock {
val group = requireGroup(nostrGroupId) val group = requireGroup(nostrGroupId)
// Retain current epoch secrets before transition // Retain current epoch secrets before transition
@@ -307,12 +326,13 @@ class MlsGroupManager(
suspend fun addMember( suspend fun addMember(
nostrGroupId: HexKey, nostrGroupId: HexKey,
keyPackageBytes: ByteArray, keyPackageBytes: ByteArray,
): CommitResult { ): CommitResult =
mutex.withLock {
val group = requireGroup(nostrGroupId) val group = requireGroup(nostrGroupId)
retainEpochSecrets(nostrGroupId, group) retainEpochSecrets(nostrGroupId, group)
val result = group.addMember(keyPackageBytes) val result = group.addMember(keyPackageBytes)
persistGroup(nostrGroupId) persistGroup(nostrGroupId)
return result result
} }
/** /**
@@ -321,12 +341,13 @@ class MlsGroupManager(
suspend fun removeMember( suspend fun removeMember(
nostrGroupId: HexKey, nostrGroupId: HexKey,
targetLeafIndex: Int, targetLeafIndex: Int,
): CommitResult { ): CommitResult =
mutex.withLock {
val group = requireGroup(nostrGroupId) val group = requireGroup(nostrGroupId)
retainEpochSecrets(nostrGroupId, group) retainEpochSecrets(nostrGroupId, group)
val result = group.removeMember(targetLeafIndex) val result = group.removeMember(targetLeafIndex)
persistGroup(nostrGroupId) persistGroup(nostrGroupId)
return result result
} }
/** /**
@@ -338,13 +359,31 @@ class MlsGroupManager(
* @param nostrGroupId hex-encoded Nostr group ID * @param nostrGroupId hex-encoded Nostr group ID
* @return the [CommitResult] to publish * @return the [CommitResult] to publish
*/ */
suspend fun rotateSigningKey(nostrGroupId: HexKey): CommitResult { suspend fun rotateSigningKey(nostrGroupId: HexKey): CommitResult =
mutex.withLock {
val group = requireGroup(nostrGroupId) val group = requireGroup(nostrGroupId)
retainEpochSecrets(nostrGroupId, group) retainEpochSecrets(nostrGroupId, group)
group.proposeSigningKeyRotation() group.proposeSigningKeyRotation()
val result = group.commit() val result = group.commit()
persistGroup(nostrGroupId) persistGroup(nostrGroupId)
return result result
}
/**
* Update group extensions (e.g., MIP-01 metadata) via a GroupContextExtensions proposal.
* Creates the proposal, commits it, and persists the new state.
*/
suspend fun updateGroupExtensions(
nostrGroupId: HexKey,
extensions: List<Extension>,
): CommitResult =
mutex.withLock {
val group = requireGroup(nostrGroupId)
retainEpochSecrets(nostrGroupId, group)
group.proposeGroupContextExtensions(extensions)
val result = group.commit()
persistGroup(nostrGroupId)
result
} }
/** /**
@@ -352,13 +391,27 @@ class MlsGroupManager(
* Returns the SelfRemove proposal bytes to publish, then removes * Returns the SelfRemove proposal bytes to publish, then removes
* local state. * local state.
*/ */
suspend fun leaveGroup(nostrGroupId: HexKey): ByteArray { suspend fun leaveGroup(nostrGroupId: HexKey): ByteArray =
mutex.withLock {
val group = requireGroup(nostrGroupId) val group = requireGroup(nostrGroupId)
val proposalBytes = group.selfRemove() val proposalBytes = group.selfRemove()
removeGroupStateUnlocked(nostrGroupId)
proposalBytes
}
/**
* Remove all local state for a group without sending any proposals.
* Used after the leave event has already been built.
*/
suspend fun removeGroupState(nostrGroupId: HexKey) =
mutex.withLock {
removeGroupStateUnlocked(nostrGroupId)
}
private suspend fun removeGroupStateUnlocked(nostrGroupId: HexKey) {
groups.remove(nostrGroupId) groups.remove(nostrGroupId)
retainedEpochs.remove(nostrGroupId) retainedEpochs.remove(nostrGroupId)
store.delete(nostrGroupId) store.delete(nostrGroupId)
return proposalBytes
} }
// --- Key Export --- // --- Key Export ---
@@ -374,6 +427,33 @@ class MlsGroupManager(
32, 32,
) )
/**
* Return exporter secrets from retained epochs for a group.
*
* Used by the inbound processor to attempt outer decryption with
* previous epoch keys when the current epoch's key fails (e.g.,
* after a commit has advanced the epoch but late-arriving messages
* still use the old exporter key).
*
* @param nostrGroupId hex-encoded Nostr group ID
* @return list of retained exporter secrets (most recent first), each
* derived via MLS-Exporter("marmot", "group-event", 32)
*/
fun retainedExporterSecrets(nostrGroupId: HexKey): List<ByteArray> {
val retained = retainedEpochs[nostrGroupId] ?: return emptyList()
return retained
.filter { it.exporterSecret.isNotEmpty() }
.sortedByDescending { it.epoch }
.map { epochSecrets ->
KeySchedule.mlsExporter(
epochSecrets.exporterSecret,
"marmot",
"group-event".encodeToByteArray(),
32,
)
}
}
// --- Private Helpers --- // --- Private Helpers ---
private fun requireGroup(nostrGroupId: HexKey): MlsGroup = private fun requireGroup(nostrGroupId: HexKey): MlsGroup =
@@ -171,6 +171,7 @@ data class RetainedEpochSecrets(
val senderDataSecret: ByteArray, val senderDataSecret: ByteArray,
val encryptionSecret: ByteArray, val encryptionSecret: ByteArray,
val leafCount: Int, val leafCount: Int,
val exporterSecret: ByteArray = ByteArray(0),
) { ) {
override fun equals(other: Any?): Boolean { override fun equals(other: Any?): Boolean {
if (this === other) return true if (this === other) return true
@@ -185,15 +186,29 @@ data class RetainedEpochSecrets(
writer.putOpaqueVarInt(senderDataSecret) writer.putOpaqueVarInt(senderDataSecret)
writer.putOpaqueVarInt(encryptionSecret) writer.putOpaqueVarInt(encryptionSecret)
writer.putUint32(leafCount.toLong()) writer.putUint32(leafCount.toLong())
writer.putOpaqueVarInt(exporterSecret)
} }
companion object { companion object {
fun decodeTls(reader: TlsReader): RetainedEpochSecrets = fun decodeTls(reader: TlsReader): RetainedEpochSecrets {
RetainedEpochSecrets( val epoch = reader.readUint64()
epoch = reader.readUint64(), val senderDataSecret = reader.readOpaqueVarInt()
senderDataSecret = reader.readOpaqueVarInt(), val encryptionSecret = reader.readOpaqueVarInt()
encryptionSecret = reader.readOpaqueVarInt(), val leafCount = reader.readUint32().toInt()
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,
) )
} }
} }
}
@@ -118,10 +118,14 @@ data class MlsKeyPackage(
} }
companion object { companion object {
fun decodeTls(reader: TlsReader): MlsKeyPackage = fun decodeTls(reader: TlsReader): MlsKeyPackage {
MlsKeyPackage( val version = reader.readUint16()
version = reader.readUint16(), require(version == 1) { "Unsupported MLS version: $version" }
cipherSuite = reader.readUint16(), val cipherSuite = reader.readUint16()
require(cipherSuite == 1) { "Unsupported ciphersuite: $cipherSuite" }
return MlsKeyPackage(
version = version,
cipherSuite = cipherSuite,
initKey = reader.readOpaqueVarInt(), initKey = reader.readOpaqueVarInt(),
leafNode = LeafNode.decodeTls(reader), leafNode = LeafNode.decodeTls(reader),
extensions = reader.readVectorVarInt { Extension.decodeTls(it) }, extensions = reader.readVectorVarInt { Extension.decodeTls(it) },
@@ -129,6 +133,7 @@ data class MlsKeyPackage(
) )
} }
} }
}
/** /**
* A KeyPackage bundled with its private keys (for the owner). * A KeyPackage bundled with its private keys (for the owner).
@@ -56,6 +56,22 @@ class SecretTree(
/** Consumed (sender, generation) pairs for replay detection (RFC 9420 Section 9.1) */ /** Consumed (sender, generation) pairs for replay detection (RFC 9420 Section 9.1) */
private val consumedGenerations = mutableMapOf<Int, MutableSet<Int>>() private val consumedGenerations = mutableMapOf<Int, MutableSet<Int>>()
/**
* Cache of key/nonce pairs for skipped generations.
* Key: (leafIndex, generation) -> derived KeyNonceGeneration.
* When fast-forwarding a ratchet, intermediate generations are saved here
* so that out-of-order messages arriving later can still be decrypted.
*/
private val skippedKeys = mutableMapOf<Pair<Int, Int>, KeyNonceGeneration>()
private companion object {
/** Maximum number of skipped key entries to retain (prevents unbounded memory growth). */
const val MAX_SKIPPED_KEYS = 1000
/** Maximum consumed generation entries to track per sender before pruning. */
const val MAX_CONSUMED_GENERATIONS_PER_SENDER = 1000
}
init { init {
// Seed the root // Seed the root
treeSecrets[BinaryTree.root(leafCount)] = encryptionSecret treeSecrets[BinaryTree.root(leafCount)] = encryptionSecret
@@ -111,11 +127,27 @@ class SecretTree(
/** /**
* Get the (key, nonce) for a specific generation, consuming secrets up to that point. * Get the (key, nonce) for a specific generation, consuming secrets up to that point.
* Used when decrypting an out-of-order message. * Used when decrypting an out-of-order message.
*
* When fast-forwarding past intermediate generations, their key/nonce pairs
* are cached in [skippedKeys] so that out-of-order messages arriving later
* can still be decrypted.
*/ */
fun applicationKeyNonceForGeneration( fun applicationKeyNonceForGeneration(
leafIndex: Int, leafIndex: Int,
generation: Int, generation: Int,
): KeyNonceGeneration { ): KeyNonceGeneration {
// Check skipped keys cache first (out-of-order message for a previously skipped generation)
val cachedKey = skippedKeys.remove(Pair(leafIndex, generation))
if (cachedKey != null) {
// Still mark as consumed for replay detection
val senderConsumed = consumedGenerations.getOrPut(leafIndex) { mutableSetOf() }
require(generation !in senderConsumed) {
"Replay detected: generation $generation from sender $leafIndex already consumed"
}
senderConsumed.add(generation)
return cachedKey
}
val state = getOrInitSender(leafIndex) val state = getOrInitSender(leafIndex)
require(generation >= state.applicationGeneration) { require(generation >= state.applicationGeneration) {
@@ -129,10 +161,22 @@ class SecretTree(
} }
senderConsumed.add(generation) senderConsumed.add(generation)
// Fast-forward the ratchet // Prune consumed generations below the current minimum for this sender
if (senderConsumed.size > MAX_CONSUMED_GENERATIONS_PER_SENDER) {
val minGeneration = state.applicationGeneration
senderConsumed.removeAll { it < minGeneration }
}
// Fast-forward the ratchet, caching intermediate key/nonce pairs
var secret = state.applicationSecret var secret = state.applicationSecret
var gen = state.applicationGeneration var gen = state.applicationGeneration
while (gen < generation) { while (gen < generation) {
// Save the intermediate generation's key/nonce for later out-of-order retrieval
val intermediateKng = deriveKeyNonce(secret, gen)
val cacheKey = Pair(leafIndex, gen)
if (skippedKeys.size < MAX_SKIPPED_KEYS) {
skippedKeys[cacheKey] = intermediateKng
}
secret = MlsCryptoProvider.expandWithLabel(secret, "secret", generationContext(gen), MlsCryptoProvider.HASH_OUTPUT_LENGTH) secret = MlsCryptoProvider.expandWithLabel(secret, "secret", generationContext(gen), MlsCryptoProvider.HASH_OUTPUT_LENGTH)
gen++ gen++
} }
@@ -109,6 +109,13 @@ class RatchetTree(
for (i in 0 until _leafCount) { for (i in 0 until _leafCount) {
if (getLeaf(i) == null) { if (getLeaf(i) == null) {
setLeaf(i, leafNode) setLeaf(i, leafNode)
// Blank the direct path (RFC 9420 Section 7.7)
val directPath = BinaryTree.directPath(i, _leafCount)
for (nodeIdx in directPath) {
if (nodeIdx < nodes.size) {
nodes[nodeIdx] = null
}
}
return i return i
} }
} }
@@ -121,6 +128,13 @@ class RatchetTree(
nodes.add(null) nodes.add(null)
} }
setLeaf(newLeafIndex, leafNode) setLeaf(newLeafIndex, leafNode)
// Blank the direct path for the new leaf
val directPath = BinaryTree.directPath(newLeafIndex, _leafCount)
for (nodeIdx in directPath) {
if (nodeIdx < nodes.size) {
nodes[nodeIdx] = null
}
}
return newLeafIndex return newLeafIndex
} }
@@ -247,18 +261,18 @@ class RatchetTree(
var currentSecret = leafSecret var currentSecret = leafSecret
for (nodeIdx in directPath) { for (nodeIdx in directPath) {
// path_secret[n] = DeriveSecret(path_secret[n-1], "path") // RFC 9420 Section 7.4: path_secret[0] = leafSecret,
val pathSecret = MlsCryptoProvider.deriveSecret(currentSecret, "path") // node_secret[n] = DeriveSecret(path_secret[n], "node")
val nodeSecret = MlsCryptoProvider.deriveSecret(currentSecret, "node")
// node_secret = DeriveSecret(path_secret, "node")
val nodeSecret = MlsCryptoProvider.deriveSecret(pathSecret, "node")
// Derive HPKE key pair from node_secret // Derive HPKE key pair from node_secret
val privateKey = MlsCryptoProvider.expandWithLabel(nodeSecret, "hpke", ByteArray(0), 32) val privateKey = MlsCryptoProvider.expandWithLabel(nodeSecret, "hpke", ByteArray(0), 32)
val publicKey = X25519.publicFromPrivate(privateKey) val publicKey = X25519.publicFromPrivate(privateKey)
results.add(PathSecretAndKey(pathSecret, privateKey, publicKey)) results.add(PathSecretAndKey(currentSecret, privateKey, publicKey))
currentSecret = pathSecret
// path_secret[n+1] = DeriveSecret(path_secret[n], "path")
currentSecret = MlsCryptoProvider.deriveSecret(currentSecret, "path")
} }
return results return results
@@ -344,9 +358,20 @@ class RatchetTree(
val tree = RatchetTree() val tree = RatchetTree()
tree.nodes.addAll(nodesList) tree.nodes.addAll(nodesList)
// Leaf count is derived from the total serialized node count. // Leaf count must account for tree trimming: the serialized tree
// nodeCount = 2 * leafCount - 1 // may have trailing blank nodes removed. Count actual leaves by
tree._leafCount = (nodesList.size + 1) / 2 // scanning for the highest occupied leaf position.
var maxLeafIndex = -1
for (i in nodesList.indices) {
if (BinaryTree.isLeaf(i) && nodesList[i] != null) {
maxLeafIndex = BinaryTree.nodeToLeaf(i)
}
}
// Leaf count is at least maxLeafIndex + 1, but also must be at
// least (nodesList.size + 1) / 2 since parent nodes at high indices
// imply leaves beyond the serialized range.
val fromNodes = (nodesList.size + 1) / 2
tree._leafCount = maxOf(fromNodes, maxLeafIndex + 1)
return tree return tree
} }
} }
@@ -22,6 +22,7 @@ package com.vitorpamplona.quartz.marmot
import com.vitorpamplona.quartz.marmot.mip03GroupMessages.GroupEvent import com.vitorpamplona.quartz.marmot.mip03GroupMessages.GroupEvent
import com.vitorpamplona.quartz.nip59Giftwrap.wraps.GiftWrapEvent import com.vitorpamplona.quartz.nip59Giftwrap.wraps.GiftWrapEvent
import kotlinx.coroutines.test.runTest
import kotlin.test.Test import kotlin.test.Test
import kotlin.test.assertEquals import kotlin.test.assertEquals
import kotlin.test.assertFalse import kotlin.test.assertFalse
@@ -38,7 +39,8 @@ class MarmotSubscriptionManagerTest {
private val groupId2 = "c".repeat(64) private val groupId2 = "c".repeat(64)
@Test @Test
fun testSubscribeGroup() { fun testSubscribeGroup() =
runTest {
val manager = MarmotSubscriptionManager(userPubKey) val manager = MarmotSubscriptionManager(userPubKey)
manager.subscribeGroup(groupId1) manager.subscribeGroup(groupId1)
@@ -48,7 +50,8 @@ class MarmotSubscriptionManagerTest {
} }
@Test @Test
fun testSubscribeGroupWithSince() { fun testSubscribeGroupWithSince() =
runTest {
val manager = MarmotSubscriptionManager(userPubKey) val manager = MarmotSubscriptionManager(userPubKey)
val since = 1700000000L val since = 1700000000L
@@ -62,7 +65,8 @@ class MarmotSubscriptionManagerTest {
} }
@Test @Test
fun testUnsubscribeGroup() { fun testUnsubscribeGroup() =
runTest {
val manager = MarmotSubscriptionManager(userPubKey) val manager = MarmotSubscriptionManager(userPubKey)
manager.subscribeGroup(groupId1) manager.subscribeGroup(groupId1)
@@ -73,7 +77,8 @@ class MarmotSubscriptionManagerTest {
} }
@Test @Test
fun testMultipleGroups() { fun testMultipleGroups() =
runTest {
val manager = MarmotSubscriptionManager(userPubKey) val manager = MarmotSubscriptionManager(userPubKey)
manager.subscribeGroup(groupId1) manager.subscribeGroup(groupId1)
@@ -86,7 +91,8 @@ class MarmotSubscriptionManagerTest {
} }
@Test @Test
fun testUpdateGroupSince() { fun testUpdateGroupSince() =
runTest {
val manager = MarmotSubscriptionManager(userPubKey) val manager = MarmotSubscriptionManager(userPubKey)
val newSince = 1700000000L val newSince = 1700000000L
@@ -110,7 +116,8 @@ class MarmotSubscriptionManagerTest {
} }
@Test @Test
fun testGiftWrapFilterWithSince() { fun testGiftWrapFilterWithSince() =
runTest {
val manager = MarmotSubscriptionManager(userPubKey) val manager = MarmotSubscriptionManager(userPubKey)
val since = 1700000000L val since = 1700000000L
@@ -121,7 +128,8 @@ class MarmotSubscriptionManagerTest {
} }
@Test @Test
fun testActiveGroupFiltersContainCorrectKind() { fun testActiveGroupFiltersContainCorrectKind() =
runTest {
val manager = MarmotSubscriptionManager(userPubKey) val manager = MarmotSubscriptionManager(userPubKey)
manager.subscribeGroup(groupId1) manager.subscribeGroup(groupId1)
@@ -134,7 +142,8 @@ class MarmotSubscriptionManagerTest {
} }
@Test @Test
fun testBuildFiltersIncludesBothTypes() { fun testBuildFiltersIncludesBothTypes() =
runTest {
val manager = MarmotSubscriptionManager(userPubKey) val manager = MarmotSubscriptionManager(userPubKey)
manager.subscribeGroup(groupId1) manager.subscribeGroup(groupId1)
@@ -164,7 +173,8 @@ class MarmotSubscriptionManagerTest {
} }
@Test @Test
fun testSyncWithGroupManager() { fun testSyncWithGroupManager() =
runTest {
val manager = MarmotSubscriptionManager(userPubKey) val manager = MarmotSubscriptionManager(userPubKey)
// Start with one group // Start with one group
@@ -179,7 +189,8 @@ class MarmotSubscriptionManagerTest {
} }
@Test @Test
fun testClear() { fun testClear() =
runTest {
val manager = MarmotSubscriptionManager(userPubKey) val manager = MarmotSubscriptionManager(userPubKey)
manager.subscribeGroup(groupId1) manager.subscribeGroup(groupId1)
@@ -73,6 +73,11 @@ actual object X25519 {
val secret = ka.generateSecret() val secret = ka.generateSecret()
// Check for small-subgroup attack (RFC 9180 Section 4.1)
require(!secret.all { it == 0.toByte() }) {
"DH produced all-zero shared secret (possible small-subgroup attack)"
}
// XDH returns big-endian, X25519 shared secret is 32 bytes // XDH returns big-endian, X25519 shared secret is 32 bytes
// Pad or trim to KEY_LENGTH // Pad or trim to KEY_LENGTH
return if (secret.size == KEY_LENGTH) { return if (secret.size == KEY_LENGTH) {
@@ -119,18 +124,20 @@ actual object X25519 {
/** /**
* Extract raw scalar bytes from JCA XECPrivateKey. * Extract raw scalar bytes from JCA XECPrivateKey.
* Scalar is in little-endian byte order per JCA spec.
*/ */
private fun extractPrivateKeyBytes(privKey: java.security.interfaces.XECPrivateKey): ByteArray { private fun extractPrivateKeyBytes(privKey: java.security.interfaces.XECPrivateKey): ByteArray {
val scalar = privKey.scalar.orElseThrow { IllegalStateException("No scalar in private key") } val scalar = privKey.scalar.orElseThrow { IllegalStateException("No scalar in private key") }
return if (scalar.size == KEY_LENGTH) { return if (scalar.size == KEY_LENGTH) {
scalar scalar
} else if (scalar.size < KEY_LENGTH) { } else if (scalar.size < KEY_LENGTH) {
// Pad with leading zeros // Pad with trailing zeros (little-endian: high-order bytes at end)
val result = ByteArray(KEY_LENGTH) val result = ByteArray(KEY_LENGTH)
scalar.copyInto(result, KEY_LENGTH - scalar.size) scalar.copyInto(result, 0)
result result
} else { } else {
scalar.copyOfRange(scalar.size - KEY_LENGTH, scalar.size) // Take only the first KEY_LENGTH bytes (little-endian low-order bytes)
scalar.copyOfRange(0, KEY_LENGTH)
} }
} }
@@ -153,12 +160,11 @@ actual object X25519 {
private fun bigIntegerToBytes(bi: java.math.BigInteger): ByteArray { private fun bigIntegerToBytes(bi: java.math.BigInteger): ByteArray {
val beBytes = bi.toByteArray() val beBytes = bi.toByteArray()
val result = ByteArray(KEY_LENGTH) val result = ByteArray(KEY_LENGTH)
// BigInteger is big-endian, possibly with leading zero byte // BigInteger is big-endian, possibly with a leading zero sign byte.
val start = if (beBytes.size > KEY_LENGTH) beBytes.size - KEY_LENGTH else 0 // Reverse into little-endian, taking at most KEY_LENGTH bytes from the low end.
val length = minOf(beBytes.size, KEY_LENGTH) val bytesToCopy = minOf(beBytes.size, KEY_LENGTH)
// Reverse into little-endian result for (i in 0 until bytesToCopy) {
for (i in 0 until length) { result[i] = beBytes[beBytes.size - 1 - i]
result[i] = beBytes[beBytes.size - 1 - i + start]
} }
return result return result
} }