openrtc 2.5.0

//! Private transport-agnostic RFC 9420 group encryption for managed rooms.
//!
//! This module owns only MLS membership and application envelopes. The avenue
//! owner decides the authorized room membership and the Rust peer-session actor
//! still owns every physical route. A managed router sees these serialized MLS
//! messages but never receives the group state or plaintext.

use anyhow::{anyhow, bail, Result};
use chacha20poly1305::{
    aead::{Aead, KeyInit, Payload},
    XChaCha20Poly1305, XNonce,
};
use openmls::prelude::{tls_codec::*, *};
use openmls_basic_credential::SignatureKeyPair;
use openmls_rust_crypto::OpenMlsRustCrypto;
use zeroize::Zeroize;

const CIPHERSUITE: Ciphersuite = Ciphersuite::MLS_128_DHKEMX25519_CHACHA20POLY1305_SHA256_Ed25519;
const MAX_IDENTITY_BYTES: usize = 512;
const MAX_GROUP_ID_BYTES: usize = 512;
const MAX_PERSISTED_STATE_BYTES: usize = 16 * 1024 * 1024;
const MAX_STORAGE_ENTRIES: usize = 100_000;
const STATE_MAGIC: &[u8; 8] = b"ORTCMLS1";
const SEALED_STATE_MAGIC: &[u8; 8] = b"ORTCMLSE";
const SEALED_STATE_NONCE_BYTES: usize = 24;

/// One serialized membership change. `welcome` is present only when a member
/// is added. `epoch` is the committed MLS epoch after the local merge.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ManagedGroupCommit {
    pub commit: Vec<u8>,
    pub welcome: Option<Vec<u8>>,
    pub epoch: u64,
}

/// Private MLS state for one admitted OpenRTC room member.
///
/// The durable device identity is carried as the MLS basic credential. Callers
/// must admit that identity before accepting its KeyPackage; MLS does not
/// replace OpenRTC's avenue authorization.
pub struct ManagedGroupMember {
    provider: OpenMlsRustCrypto,
    signer: SignatureKeyPair,
    credential: CredentialWithKey,
    group: Option<MlsGroup>,
}

impl ManagedGroupMember {
    /// Create an MLS identity bound to the already-authenticated durable device
    /// identifier. No key material is exposed through the public SDK.
    pub fn new(device_id: impl AsRef<[u8]>) -> Result<Self> {
        let identity = device_id.as_ref();
        if identity.is_empty() || identity.len() > MAX_IDENTITY_BYTES {
            bail!("managed group identity must contain 1..={MAX_IDENTITY_BYTES} bytes");
        }

        let provider = OpenMlsRustCrypto::default();
        let signer = SignatureKeyPair::new(CIPHERSUITE.signature_algorithm())
            .map_err(|error| anyhow!("managed group signature key generation failed: {error:?}"))?;
        signer.store(provider.storage()).map_err(|error| {
            anyhow!("managed group signature key persistence failed: {error:?}")
        })?;
        let credential = CredentialWithKey {
            credential: BasicCredential::new(identity.to_vec()).into(),
            signature_key: signer.to_public_vec().into(),
        };

        Ok(Self {
            provider,
            signer,
            credential,
            group: None,
        })
    }

    /// Generate a fresh one-use KeyPackage. Its corresponding private material
    /// remains in this member's provider until a Welcome is processed.
    pub fn key_package(&self) -> Result<Vec<u8>> {
        KeyPackage::builder()
            .build(
                CIPHERSUITE,
                &self.provider,
                &self.signer,
                self.credential.clone(),
            )
            .map_err(|error| anyhow!("managed group key package generation failed: {error:?}"))?
            .key_package()
            .tls_serialize_detached()
            .map_err(|error| anyhow!("managed group key package serialization failed: {error:?}"))
    }

    pub fn key_package_identity(key_package: &[u8]) -> Result<Vec<u8>> {
        let provider = OpenMlsRustCrypto::default();
        let key_package = KeyPackageIn::tls_deserialize_exact(key_package)
            .map_err(|error| anyhow!("managed group key package parse failed: {error:?}"))?
            .validate(provider.crypto(), ProtocolVersion::Mls10)
            .map_err(|error| anyhow!("managed group key package validation failed: {error:?}"))?;
        Ok(key_package
            .leaf_node()
            .credential()
            .serialized_content()
            .to_vec())
    }

    /// Create the room's initial MLS epoch. This is separate from the room
    /// architecture epoch and advances only for cryptographic membership.
    pub fn create_group(&mut self, room_id: impl AsRef<[u8]>) -> Result<u64> {
        if self.group.is_some() {
            bail!("managed group member is already attached to a group");
        }
        let room_id = room_id.as_ref();
        if room_id.is_empty() || room_id.len() > MAX_GROUP_ID_BYTES {
            bail!("managed group id must contain 1..={MAX_GROUP_ID_BYTES} bytes");
        }

        let config = MlsGroupCreateConfig::builder()
            .ciphersuite(CIPHERSUITE)
            .use_ratchet_tree_extension(true)
            .build();
        let group = MlsGroup::new_with_group_id(
            &self.provider,
            &self.signer,
            &config,
            GroupId::from_slice(room_id),
            self.credential.clone(),
        )
        .map_err(|error| anyhow!("managed group creation failed: {error:?}"))?;
        let epoch = group.epoch().as_u64();
        self.group = Some(group);
        Ok(epoch)
    }

    /// Add one already-admitted member and advance the encryption epoch.
    pub fn add_member(&mut self, key_package: &[u8]) -> Result<ManagedGroupCommit> {
        self.add_members(&[key_package.to_vec()])
    }

    /// Batch already-admitted members into one tree update. Large rooms must
    /// not create one hosted coordination transaction and one MLS epoch per
    /// participant when a single authorized commit can carry the same change.
    pub fn add_members(&mut self, key_packages: &[Vec<u8>]) -> Result<ManagedGroupCommit> {
        if key_packages.is_empty() {
            bail!("managed group member add requires at least one KeyPackage");
        }
        let key_packages = key_packages
            .iter()
            .map(|key_package| {
                KeyPackageIn::tls_deserialize_exact(key_package)
                    .map_err(|error| anyhow!("managed group key package parse failed: {error:?}"))?
                    .validate(self.provider.crypto(), ProtocolVersion::Mls10)
                    .map_err(|error| {
                        anyhow!("managed group key package validation failed: {error:?}")
                    })
            })
            .collect::<Result<Vec<_>>>()?;
        let Self {
            provider,
            signer,
            group,
            ..
        } = self;
        let group = group
            .as_mut()
            .ok_or_else(|| anyhow!("managed group member is not attached to a group"))?;
        let (commit, welcome, _) = group
            .add_members(provider, signer, &key_packages)
            .map_err(|error| anyhow!("managed group member add failed: {error:?}"))?;
        let commit = serialize_message(commit, "membership commit")?;
        let welcome = serialize_message(welcome, "membership welcome")?;
        group
            .merge_pending_commit(provider)
            .map_err(|error| anyhow!("managed group local add commit failed: {error:?}"))?;
        Ok(ManagedGroupCommit {
            commit,
            welcome: Some(welcome),
            epoch: group.epoch().as_u64(),
        })
    }

    /// Reconcile the current tree to the avenue-authorized KeyPackage set in
    /// one MLS epoch. The caller supplies only authenticated room members.
    /// Existing members keep their current leaf; newly admitted members use
    /// their one-use KeyPackage and removed identities are revoked by the same
    /// commit, so room churn never exposes an intermediate encryption epoch.
    pub fn rekey_members(&mut self, key_packages: &[Vec<u8>]) -> Result<ManagedGroupCommit> {
        let Self {
            provider,
            signer,
            group,
            ..
        } = self;
        let group = group
            .as_mut()
            .ok_or_else(|| anyhow!("managed group member is not attached to a group"))?;
        let validated = key_packages
            .iter()
            .map(|key_package| {
                KeyPackageIn::tls_deserialize_exact(key_package)
                    .map_err(|error| anyhow!("managed group key package parse failed: {error:?}"))?
                    .validate(provider.crypto(), ProtocolVersion::Mls10)
                    .map_err(|error| {
                        anyhow!("managed group key package validation failed: {error:?}")
                    })
            })
            .collect::<Result<Vec<_>>>()?;
        let target_identities = validated
            .iter()
            .map(|key_package| {
                key_package
                    .leaf_node()
                    .credential()
                    .serialized_content()
                    .to_vec()
            })
            .collect::<std::collections::BTreeSet<_>>();
        if target_identities.len() != validated.len() {
            bail!("managed group reconciliation contains duplicate identities");
        }
        let current = group
            .members()
            .map(|member| {
                (
                    member.index,
                    member.credential.serialized_content().to_vec(),
                )
            })
            .collect::<Vec<_>>();
        let current_identities = current
            .iter()
            .map(|(_, identity)| identity.clone())
            .collect::<std::collections::BTreeSet<_>>();

        let removals = current
            .iter()
            .filter(|(_, identity)| !target_identities.contains(identity))
            .map(|(index, _)| *index)
            .collect::<Vec<_>>();
        let additions = validated
            .iter()
            .filter(|key_package| {
                !current_identities
                    .contains(key_package.leaf_node().credential().serialized_content())
            })
            .collect::<Vec<_>>();
        let membership_unchanged = removals.is_empty() && additions.is_empty();
        let bundle = group
            .commit_builder()
            .consume_proposal_store(false)
            .force_self_update(membership_unchanged)
            .propose_removals(removals)
            .propose_adds(additions.into_iter().cloned())
            .load_psks(provider.storage())
            .map_err(|error| anyhow!("managed group reconciliation PSK load failed: {error:?}"))?
            .build(provider.rand(), provider.crypto(), signer, |_| true)
            .map_err(|error| anyhow!("managed group reconciliation commit failed: {error:?}"))?
            .stage_commit(provider)
            .map_err(|error| anyhow!("managed group reconciliation staging failed: {error:?}"))?;
        let commit =
            serialize_message(bundle.commit().clone(), "membership reconciliation commit")?;
        let welcome = bundle
            .to_welcome_msg()
            .map(|message| serialize_message(message, "membership reconciliation welcome"))
            .transpose()?;
        group
            .merge_pending_commit(provider)
            .map_err(|error| anyhow!("managed group reconciliation merge failed: {error:?}"))?;
        Ok(ManagedGroupCommit {
            commit,
            welcome,
            epoch: group.epoch().as_u64(),
        })
    }

    /// Join from a Welcome produced by the authorized group committer.
    pub fn join_group(&mut self, welcome: &[u8]) -> Result<u64> {
        if self.group.is_some() {
            bail!("managed group member is already attached to a group");
        }
        let message = MlsMessageIn::tls_deserialize_exact(welcome)
            .map_err(|error| anyhow!("managed group welcome parse failed: {error:?}"))?;
        let welcome = match message.extract() {
            MlsMessageBodyIn::Welcome(welcome) => welcome,
            _ => bail!("managed group join requires an MLS Welcome"),
        };
        let staged = StagedWelcome::new_from_welcome(
            &self.provider,
            &MlsGroupJoinConfig::default(),
            welcome,
            None,
        )
        .map_err(|error| anyhow!("managed group welcome validation failed: {error:?}"))?;
        let group = staged
            .into_group(&self.provider)
            .map_err(|error| anyhow!("managed group join failed: {error:?}"))?;
        let epoch = group.epoch().as_u64();
        self.group = Some(group);
        Ok(epoch)
    }

    /// Remove one member by its authenticated credential bytes and advance the
    /// encryption epoch. The removed member cannot decrypt later epochs.
    pub fn remove_member(&mut self, device_id: &[u8]) -> Result<ManagedGroupCommit> {
        let Self {
            provider,
            signer,
            group,
            ..
        } = self;
        let group = group
            .as_mut()
            .ok_or_else(|| anyhow!("managed group member is not attached to a group"))?;
        let member = group
            .members()
            .find(|member| member.credential.serialized_content() == device_id)
            .ok_or_else(|| anyhow!("managed group member is not present"))?;
        let (commit, welcome, _) = group
            .remove_members(provider, signer, &[member.index])
            .map_err(|error| anyhow!("managed group member removal failed: {error:?}"))?;
        if welcome.is_some() {
            bail!("managed group removal unexpectedly produced a Welcome");
        }
        let commit = serialize_message(commit, "removal commit")?;
        group
            .merge_pending_commit(provider)
            .map_err(|error| anyhow!("managed group local removal commit failed: {error:?}"))?;
        Ok(ManagedGroupCommit {
            commit,
            welcome: None,
            epoch: group.epoch().as_u64(),
        })
    }

    /// Apply an authenticated membership commit from another admitted member.
    pub fn apply_commit(&mut self, commit: &[u8]) -> Result<u64> {
        let message = protocol_message(commit, "membership commit")?;
        let Self {
            provider, group, ..
        } = self;
        let group = group
            .as_mut()
            .ok_or_else(|| anyhow!("managed group member is not attached to a group"))?;
        let processed = group
            .process_message(provider, message)
            .map_err(|error| anyhow!("managed group commit verification failed: {error:?}"))?;
        let staged = match processed.into_content() {
            ProcessedMessageContent::StagedCommitMessage(staged) => staged,
            _ => bail!("managed group expected an MLS membership commit"),
        };
        group
            .merge_staged_commit(provider, *staged)
            .map_err(|error| anyhow!("managed group commit merge failed: {error:?}"))?;
        Ok(group.epoch().as_u64())
    }

    /// Encrypt one application envelope for content-blind duplication.
    pub fn seal(&mut self, payload: &[u8], associated_data: &[u8]) -> Result<Vec<u8>> {
        let Self {
            provider,
            signer,
            group,
            ..
        } = self;
        let group = group
            .as_mut()
            .ok_or_else(|| anyhow!("managed group member is not attached to a group"))?;
        group.set_aad(associated_data.to_vec());
        let message = group
            .create_message(provider, signer, payload)
            .map_err(|error| anyhow!("managed group encryption failed: {error:?}"))?;
        serialize_message(message, "application message")
    }

    /// Verify and decrypt one application envelope from the current MLS epoch.
    pub fn open(&mut self, message: &[u8], associated_data: &[u8]) -> Result<Vec<u8>> {
        let message = protocol_message(message, "application message")?;
        let Self {
            provider, group, ..
        } = self;
        let group = group
            .as_mut()
            .ok_or_else(|| anyhow!("managed group member is not attached to a group"))?;
        let processed = group
            .process_message(provider, message)
            .map_err(|error| anyhow!("managed group decryption failed: {error:?}"))?;
        if processed.aad() != associated_data {
            bail!("managed group associated data mismatch");
        }
        match processed.into_content() {
            ProcessedMessageContent::ApplicationMessage(application) => {
                Ok(application.into_bytes())
            }
            _ => bail!("managed group expected an MLS application message"),
        }
    }

    pub fn encryption_epoch(&self) -> Option<u64> {
        self.group.as_ref().map(|group| group.epoch().as_u64())
    }

    pub fn member_count(&self) -> usize {
        self.group
            .as_ref()
            .map(|group| group.members().count())
            .unwrap_or_default()
    }

    pub fn member_identities(&self) -> Vec<Vec<u8>> {
        self.group
            .as_ref()
            .map(|group| {
                group
                    .members()
                    .map(|member| member.credential.serialized_content().to_vec())
                    .collect()
            })
            .unwrap_or_default()
    }

    /// Export one encrypted, opaque restart snapshot. The wrapping key is
    /// supplied by the existing platform-private device store and never leaves
    /// the Rust/WASM owner. The returned bytes are safe for chunked host
    /// persistence; they contain no plaintext MLS state.
    pub fn seal_persisted_state(&self, wrapping_key: &[u8; 32]) -> Result<Vec<u8>> {
        let mut plaintext = self.serialize_persisted_state()?;
        let mut nonce = [0_u8; SEALED_STATE_NONCE_BYTES];
        getrandom::getrandom(&mut nonce)
            .map_err(|error| anyhow!("managed group state nonce generation failed: {error}"))?;
        let cipher = XChaCha20Poly1305::new(wrapping_key.into());
        let associated_data = persisted_state_aad(self.credential.credential.serialized_content());
        let encrypted = cipher
            .encrypt(
                XNonce::from_slice(&nonce),
                Payload {
                    msg: &plaintext,
                    aad: &associated_data,
                },
            )
            .map_err(|_| anyhow!("managed group state encryption failed"));
        plaintext.zeroize();
        let encrypted = encrypted?;
        let mut sealed = Vec::with_capacity(
            SEALED_STATE_MAGIC.len() + SEALED_STATE_NONCE_BYTES + encrypted.len(),
        );
        sealed.extend_from_slice(SEALED_STATE_MAGIC);
        sealed.extend_from_slice(&nonce);
        sealed.extend_from_slice(&encrypted);
        Ok(sealed)
    }

    /// Restore a previously sealed snapshot for exactly the authenticated
    /// durable device identity. Tampering, key mismatch, identity mismatch,
    /// oversized state, and incomplete OpenMLS storage all fail closed.
    pub fn restore_persisted_state(
        device_id: impl AsRef<[u8]>,
        wrapping_key: &[u8; 32],
        sealed: &[u8],
    ) -> Result<Self> {
        let identity = device_id.as_ref();
        if identity.is_empty() || identity.len() > MAX_IDENTITY_BYTES {
            bail!("managed group identity must contain 1..={MAX_IDENTITY_BYTES} bytes");
        }
        if sealed.len() <= SEALED_STATE_MAGIC.len() + SEALED_STATE_NONCE_BYTES
            || sealed.len() > MAX_PERSISTED_STATE_BYTES
            || &sealed[..SEALED_STATE_MAGIC.len()] != SEALED_STATE_MAGIC
        {
            bail!("managed group sealed state is invalid");
        }
        let nonce_start = SEALED_STATE_MAGIC.len();
        let ciphertext_start = nonce_start + SEALED_STATE_NONCE_BYTES;
        let cipher = XChaCha20Poly1305::new(wrapping_key.into());
        let associated_data = persisted_state_aad(identity);
        let mut plaintext = cipher
            .decrypt(
                XNonce::from_slice(&sealed[nonce_start..ciphertext_start]),
                Payload {
                    msg: &sealed[ciphertext_start..],
                    aad: &associated_data,
                },
            )
            .map_err(|_| anyhow!("managed group sealed state authentication failed"))?;
        let restored = Self::deserialize_persisted_state(identity, &plaintext);
        plaintext.zeroize();
        restored
    }

    fn serialize_persisted_state(&self) -> Result<Vec<u8>> {
        let identity = self.credential.credential.serialized_content();
        let public_key = self.signer.public();
        let group_id = self
            .group
            .as_ref()
            .map(|group| group.group_id().as_slice())
            .unwrap_or_default();
        let mut entries = self
            .provider
            .storage()
            .values
            .read()
            .map_err(|_| anyhow!("managed group storage is unavailable"))?
            .iter()
            .map(|(key, value)| (key.clone(), value.clone()))
            .collect::<Vec<_>>();
        entries.sort_by(|left, right| left.0.cmp(&right.0));
        if entries.len() > MAX_STORAGE_ENTRIES {
            bail!("managed group storage entry capacity exceeded");
        }
        let mut output = Vec::new();
        output.extend_from_slice(STATE_MAGIC);
        write_field(&mut output, identity)?;
        write_field(&mut output, public_key)?;
        write_field(&mut output, group_id)?;
        write_u32(&mut output, entries.len())?;
        for (key, value) in entries {
            write_field(&mut output, &key)?;
            write_field(&mut output, &value)?;
            if output.len() > MAX_PERSISTED_STATE_BYTES {
                bail!("managed group persisted state capacity exceeded");
            }
        }
        Ok(output)
    }

    fn deserialize_persisted_state(expected_identity: &[u8], bytes: &[u8]) -> Result<Self> {
        if bytes.len() > MAX_PERSISTED_STATE_BYTES || !bytes.starts_with(STATE_MAGIC) {
            bail!("managed group persisted state is invalid");
        }
        let mut cursor = STATE_MAGIC.len();
        let identity = read_field(bytes, &mut cursor, MAX_IDENTITY_BYTES)?;
        if identity != expected_identity {
            bail!("managed group persisted identity mismatch");
        }
        let public_key = read_field(bytes, &mut cursor, 512)?;
        let group_id = read_field(bytes, &mut cursor, MAX_GROUP_ID_BYTES)?;
        let entry_count = read_u32(bytes, &mut cursor)?;
        if entry_count > MAX_STORAGE_ENTRIES {
            bail!("managed group persisted storage entry capacity exceeded");
        }

        let provider = OpenMlsRustCrypto::default();
        {
            let mut values = provider
                .storage()
                .values
                .write()
                .map_err(|_| anyhow!("managed group storage is unavailable"))?;
            for _ in 0..entry_count {
                let key = read_field(bytes, &mut cursor, MAX_PERSISTED_STATE_BYTES)?;
                let value = read_field(bytes, &mut cursor, MAX_PERSISTED_STATE_BYTES)?;
                if values.insert(key, value).is_some() {
                    bail!("managed group persisted storage contains duplicate entries");
                }
            }
        }
        if cursor != bytes.len() {
            bail!("managed group persisted state has trailing bytes");
        }
        let signer = SignatureKeyPair::read(
            provider.storage(),
            &public_key,
            CIPHERSUITE.signature_algorithm(),
        )
        .ok_or_else(|| anyhow!("managed group persisted signer is unavailable"))?;
        let credential = CredentialWithKey {
            credential: BasicCredential::new(identity).into(),
            signature_key: public_key.into(),
        };
        let group = if group_id.is_empty() {
            None
        } else {
            MlsGroup::load(provider.storage(), &GroupId::from_slice(&group_id))
                .map_err(|error| anyhow!("managed group persisted state load failed: {error:?}"))?
                .ok_or_else(|| anyhow!("managed group persisted state is incomplete"))?
                .into()
        };
        Ok(Self {
            provider,
            signer,
            credential,
            group,
        })
    }
}

fn persisted_state_aad(identity: &[u8]) -> Vec<u8> {
    let mut aad = b"openrtc-managed-group-state-v1:".to_vec();
    aad.extend_from_slice(identity);
    aad
}

fn write_u32(output: &mut Vec<u8>, value: usize) -> Result<()> {
    let value = u32::try_from(value)
        .map_err(|_| anyhow!("managed group persisted state length exceeds u32"))?;
    output.extend_from_slice(&value.to_be_bytes());
    Ok(())
}

fn write_field(output: &mut Vec<u8>, value: &[u8]) -> Result<()> {
    write_u32(output, value.len())?;
    output.extend_from_slice(value);
    Ok(())
}

fn read_u32(bytes: &[u8], cursor: &mut usize) -> Result<usize> {
    let end = cursor
        .checked_add(4)
        .filter(|end| *end <= bytes.len())
        .ok_or_else(|| anyhow!("managed group persisted state is truncated"))?;
    let value = u32::from_be_bytes(
        bytes[*cursor..end]
            .try_into()
            .map_err(|_| anyhow!("managed group persisted length is invalid"))?,
    ) as usize;
    *cursor = end;
    Ok(value)
}

fn read_field(bytes: &[u8], cursor: &mut usize, max_len: usize) -> Result<Vec<u8>> {
    let len = read_u32(bytes, cursor)?;
    if len > max_len {
        bail!("managed group persisted field capacity exceeded");
    }
    let end = cursor
        .checked_add(len)
        .filter(|end| *end <= bytes.len())
        .ok_or_else(|| anyhow!("managed group persisted state is truncated"))?;
    let value = bytes[*cursor..end].to_vec();
    *cursor = end;
    Ok(value)
}

fn serialize_message(message: MlsMessageOut, label: &str) -> Result<Vec<u8>> {
    message
        .tls_serialize_detached()
        .map_err(|error| anyhow!("managed group {label} serialization failed: {error:?}"))
}

fn protocol_message(bytes: &[u8], label: &str) -> Result<ProtocolMessage> {
    MlsMessageIn::tls_deserialize_exact(bytes)
        .map_err(|error| anyhow!("managed group {label} parse failed: {error:?}"))?
        .try_into_protocol_message()
        .map_err(|error| anyhow!("managed group {label} is not a protocol message: {error:?}"))
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn add_encrypt_revoke_advances_only_the_encryption_epoch() {
        let mut host = ManagedGroupMember::new("device-host").unwrap();
        let mut viewer = ManagedGroupMember::new("device-viewer").unwrap();
        assert_eq!(host.create_group("room-1").unwrap(), 0);

        let key_package = viewer.key_package().unwrap();
        let add = host.add_member(&key_package).unwrap();
        assert_eq!(add.epoch, 1);
        assert_eq!(
            viewer.join_group(add.welcome.as_deref().unwrap()).unwrap(),
            1
        );
        assert_eq!(host.member_count(), 2);
        assert_eq!(viewer.member_count(), 2);

        let encrypted = host.seal(b"room-state", b"room-1:message-1").unwrap();
        assert_ne!(encrypted, b"room-state");
        assert_eq!(
            viewer.open(&encrypted, b"room-1:message-1").unwrap(),
            b"room-state"
        );

        let removal = host.remove_member(b"device-viewer").unwrap();
        assert_eq!(removal.epoch, 2);
        assert_eq!(host.member_count(), 1);
        assert_eq!(viewer.apply_commit(&removal.commit).unwrap(), 2);
        assert_eq!(viewer.member_count(), 1);

        let after_revocation = host.seal(b"host-only", b"room-1:message-2").unwrap();
        assert!(viewer.open(&after_revocation, b"room-1:message-2").is_err());
    }

    #[test]
    fn rejects_wrong_aad_and_unattached_use() {
        let mut host = ManagedGroupMember::new("device-host").unwrap();
        assert!(host.seal(b"payload", b"aad").is_err());
        host.create_group("room-1").unwrap();
        let encrypted = host.seal(b"payload", b"expected").unwrap();

        let mut other = ManagedGroupMember::new("device-other").unwrap();
        assert!(other.join_group(&encrypted).is_err());
        assert!(host.open(&encrypted, b"wrong").is_err());
    }

    #[test]
    fn sealed_restart_state_preserves_group_and_rejects_wrong_owner_or_key() {
        let wrapping_key = [7_u8; 32];
        let mut host = ManagedGroupMember::new("device-host").unwrap();
        let mut viewer = ManagedGroupMember::new("device-viewer").unwrap();
        host.create_group("room-restart").unwrap();
        let add = host.add_member(&viewer.key_package().unwrap()).unwrap();
        viewer.join_group(add.welcome.as_deref().unwrap()).unwrap();

        let host_state = host.seal_persisted_state(&wrapping_key).unwrap();
        let viewer_state = viewer.seal_persisted_state(&wrapping_key).unwrap();
        assert!(!host_state
            .windows(b"device-host".len())
            .any(|window| window == b"device-host"));
        assert!(ManagedGroupMember::restore_persisted_state(
            "device-other",
            &wrapping_key,
            &host_state,
        )
        .is_err());
        assert!(ManagedGroupMember::restore_persisted_state(
            "device-host",
            &[8_u8; 32],
            &host_state,
        )
        .is_err());

        let mut host =
            ManagedGroupMember::restore_persisted_state("device-host", &wrapping_key, &host_state)
                .unwrap();
        let mut viewer = ManagedGroupMember::restore_persisted_state(
            "device-viewer",
            &wrapping_key,
            &viewer_state,
        )
        .unwrap();
        assert_eq!(host.member_count(), 2);
        assert_eq!(viewer.encryption_epoch(), Some(1));
        let encrypted = host.seal(b"after-restart", b"room-restart:1").unwrap();
        assert_eq!(
            viewer.open(&encrypted, b"room-restart:1").unwrap(),
            b"after-restart"
        );

        let removal = host.remove_member(b"device-viewer").unwrap();
        viewer.apply_commit(&removal.commit).unwrap();
        let encrypted = host.seal(b"revoked", b"room-restart:2").unwrap();
        assert!(viewer.open(&encrypted, b"room-restart:2").is_err());
    }

    #[test]
    fn reconcile_members_adds_and_revokes_in_one_epoch() {
        let mut host = ManagedGroupMember::new("device-host").unwrap();
        let mut departing = ManagedGroupMember::new("device-departing").unwrap();
        let mut joining = ManagedGroupMember::new("device-joining").unwrap();
        host.create_group("room-churn").unwrap();
        let initial = host.add_member(&departing.key_package().unwrap()).unwrap();
        departing
            .join_group(initial.welcome.as_deref().unwrap())
            .unwrap();

        let epoch_before = host.encryption_epoch().unwrap();
        let target = vec![host.key_package().unwrap(), joining.key_package().unwrap()];
        let update = host.rekey_members(&target).unwrap();

        assert_eq!(update.epoch, epoch_before + 1);
        assert_eq!(
            host.member_identities()
                .into_iter()
                .collect::<std::collections::BTreeSet<_>>(),
            [b"device-host".to_vec(), b"device-joining".to_vec()]
                .into_iter()
                .collect(),
        );
        assert_eq!(
            departing.apply_commit(&update.commit).unwrap(),
            update.epoch
        );
        assert_eq!(
            joining
                .join_group(update.welcome.as_deref().unwrap())
                .unwrap(),
            update.epoch,
        );
        let encrypted = host.seal(b"current-members-only", b"room-churn:2").unwrap();
        assert_eq!(
            joining.open(&encrypted, b"room-churn:2").unwrap(),
            b"current-members-only",
        );
        assert!(departing.open(&encrypted, b"room-churn:2").is_err());
    }

    #[test]
    fn unchanged_members_still_rotate_one_epoch_for_committer_failover() {
        let mut host = ManagedGroupMember::new("device-host").unwrap();
        host.create_group("room-rekey").unwrap();
        let epoch_before = host.encryption_epoch().unwrap();

        let update = host.rekey_members(&[host.key_package().unwrap()]).unwrap();

        assert_eq!(update.epoch, epoch_before + 1);
        assert!(update.welcome.is_none());
        assert_eq!(host.member_identities(), vec![b"device-host".to_vec()]);
    }

    #[test]
    #[ignore = "explicit large-room crypto scale lane"]
    fn one_thousand_members_share_one_bounded_membership_commit() {
        let started = std::time::Instant::now();
        let mut host = ManagedGroupMember::new("device-0000").unwrap();
        host.create_group("room-1000").unwrap();
        let key_packages = (1..1_000)
            .map(|index| {
                ManagedGroupMember::new(format!("device-{index:04}"))
                    .unwrap()
                    .key_package()
                    .unwrap()
            })
            .collect::<Vec<_>>();

        let commit = host.add_members(&key_packages).unwrap();
        assert_eq!(commit.epoch, 1);
        assert_eq!(host.member_count(), 1_000);
        assert!(commit.welcome.is_some());
        eprintln!(
            "1,000-member MLS commit bytes={} welcome bytes={}",
            commit.commit.len(),
            commit.welcome.as_ref().map(Vec::len).unwrap_or_default(),
        );
        let sealed_state = host.seal_persisted_state(&[9_u8; 32]).unwrap();
        eprintln!("sealed 1,000-member MLS state bytes={}", sealed_state.len());
        assert!(sealed_state.len() <= MAX_PERSISTED_STATE_BYTES);
        assert!(
            started.elapsed() < std::time::Duration::from_secs(30),
            "1,000-member MLS batch exceeded the 30-second local crypto bound"
        );
    }
}