offline-protocol 0.30.0

//! [`SealedFileMlsStorage`]: MLS material in files, sealed under a store key.

use std::collections::{BTreeSet, HashSet};
use std::path::{Path, PathBuf};
use std::sync::Mutex;

use chacha20poly1305::aead::{Aead, KeyInit, Payload};
use chacha20poly1305::{ChaCha20Poly1305, Key, Nonce};
use hkdf::Hkdf;
use hmac::{Hmac, Mac};
use offline_protocol_mls::storage::{MlsStorage, StorageError, StorageResult};
use rand_core::{OsRng, RngCore};
use sha2::Sha256;
use zeroize::Zeroizing;

use super::key::{StoreKeyError, StoreKeyProvider, STORE_KEY_BYTES};
use super::pair::{StoredPairing, PAIRING_ID_BYTES};
use super::records::{self, BoundedRead, StoreLock, SweepMemo, MAX_FRAME_BYTES};
use super::{account_directory, lock_directory, FileStoreError};

/// Magic of a sealed record file. Distinct from the plain frame's `OPS1`, so
/// neither store mistakes the other's file for its own.
const SEALED_MAGIC: &[u8; 4] = b"OMS1";

/// ChaCha20-Poly1305 nonce length.
const NONCE_BYTES: usize = 12;

/// Poly1305 tag length.
const TAG_BYTES: usize = 16;

/// Bytes a seal adds to its plaintext.
const SEAL_OVERHEAD: usize = SEALED_MAGIC.len() + NONCE_BYTES + TAG_BYTES;

/// Largest sealed file a read accepts: the largest frame, sealed.
const MAX_SEALED_BYTES: usize = MAX_FRAME_BYTES + SEAL_OVERHEAD;

/// HKDF salt: names this construction, so a store key reused elsewhere (it
/// should not be) derives unrelated keys here.
const HKDF_SALT: &[u8] = b"offline-protocol sealed file mls store v1";

/// Prefix of a type directory: `c_` and the keyed digest of the key type.
const TYPE_PREFIX: &str = "c_";

/// Prefix of a record file: `e_` and the keyed digest of the type and id.
const ENTRY_PREFIX: &str = "e_";

/// The file that proves the store key before any record is read.
const KEY_CHECK_FILE: &str = "store-key-check";

/// What the key check seals.
const KEY_CHECK_PLAINTEXT: &[u8] = b"offline-protocol sealed file mls store key check v1";

/// Records the key check tries when the check file is missing or damaged.
const KEY_CHECK_PROBE_LIMIT: usize = 64;

/// The lock file this store holds in its account directory. Distinct from
/// the protocol-state store's, so the two can share a root.
const LOCK_FILE: &str = "mls-store.lock";

/// The pairing id this store shares with the protocol-state store beside it,
/// sealed like the key check. In the account directory, outside every type
/// directory, so [`SealedFileMlsStorage::holds_records`] never takes it for an
/// identity: it is written before the engine has minted one.
pub(super) const PAIRING_FILE: &str = "mls-store.pair";

/// MLS storage in files, every record sealed under a key derived from the
/// operator's store key.
///
/// This is the headless host's stand-in for the platform keystore the mobile
/// bindings use. What it protects against is someone who can read the
/// directory but does not have the store key: a copied disk, a backup, another
/// local user. Against that reader:
///
/// - **Values are sealed.** ChaCha20-Poly1305 under a key derived with HKDF
///   from the store key and the account namespace, so a record copied into
///   another account's directory does not open.
/// - **Keys are sealed too.** The record's `(key_type, key_id)` travels inside
///   the ciphertext, and file names are an HMAC under a second derived key. A
///   listing reveals how many records of each kind exist, never which peers or
///   groups they belong to. (The protocol-state store's names are an unsalted
///   digest, a format shared with the bindings.)
/// - **A record is bound to its slot.** The associated data is the file name,
///   and the decrypted header must name the key asked for, so a record moved
///   onto another's name does not open.
///
/// What it cannot protect against is the store key itself. It arrives from a
/// [`StoreKeyProvider`], and on a host without a keystore that means in the
/// clear from the operator's configuration (residual risk R18).
///
/// A record that does not authenticate is reported as
/// [`StorageError::CorruptedData`], naming its file, and **left in place**.
/// The store never deletes on a read, because the common cause is a wrong key
/// and deleting on that would destroy the identity. [`Self::open`] checks the
/// key against a sealed check file first and refuses a wrong one before any
/// record is read, so a record that fails afterwards is damaged: the error
/// names the file, and what deleting it costs depends on what it held (the
/// engine regenerates a damaged protocol-state record key by itself).
pub struct SealedFileMlsStorage {
    directory: PathBuf,
    cipher: ChaCha20Poly1305,
    name_key: Zeroizing<[u8; 32]>,
    lock: StoreLock,
    swept: SweepMemo,
    /// Records the listing has already warned about. The engine lists key
    /// packages and groups often, and a damaged record stays damaged, so one
    /// warning per file per store says everything without flooding the log.
    listing_warned: Mutex<HashSet<PathBuf>>,
}

impl std::fmt::Debug for SealedFileMlsStorage {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("SealedFileMlsStorage")
            .field("directory", &self.directory)
            .finish_non_exhaustive()
    }
}

/// What the key check found when its file was missing or did not open.
enum Probe {
    NoRecords,
    Opened,
    NoneOpened,
}

/// What one record file turned out to be.
///
/// A file that could not be read is not a file that did not open. Folding
/// the two together sent a permission or disk error to the operator as "does
/// not open under this store key", and counted it towards refusing the key.
enum Opened {
    Record(String, String),
    /// Removed between the listing and the read.
    Absent,
    /// Read, and does not authenticate under this key and name.
    DoesNotOpen,
}

impl SealedFileMlsStorage {
    /// Opens (creating if needed) the store at `<root>/<namespace>/`, sealed
    /// under the key `provider` supplies.
    ///
    /// The provider is asked once. The first open writes a sealed check file;
    /// every later open must open it, and a key that does not is refused with
    /// [`FileStoreError::WrongStoreKey`] before any record is read or changed.
    /// A check file that is missing or damaged is rebuilt only when an
    /// existing record opens under the key, which proves it. A check that
    /// does not open over a store with no records is refused with
    /// [`FileStoreError::KeyCheckUnverifiable`], which names the check file.
    ///
    /// A second store over the same directory is refused with
    /// [`FileStoreError::InUse`] until this one is closed or dropped.
    ///
    /// `root` should be a directory that survives an upgrade: this store holds
    /// the device's identity, and losing it gives the device a new address.
    pub fn open(
        root: impl AsRef<Path>,
        namespace: &str,
        provider: &dyn StoreKeyProvider,
    ) -> Result<Self, FileStoreError> {
        let directory = account_directory(root.as_ref(), namespace)?;
        let directory_lock = lock_directory(&directory, LOCK_FILE)?;
        let store_key = provider.store_key()?;
        let hkdf = Hkdf::<Sha256>::new(Some(HKDF_SALT), store_key.as_slice());
        let seal_key = derive(&hkdf, b"seal", namespace)?;
        let name_key = derive(&hkdf, b"names", namespace)?;
        let store = Self {
            directory,
            cipher: ChaCha20Poly1305::new(Key::from_slice(seal_key.as_slice())),
            name_key,
            lock: StoreLock::new(directory_lock),
            swept: SweepMemo::default(),
            listing_warned: Mutex::default(),
        };
        store.verify_store_key()?;
        Ok(store)
    }

    /// The directory this store owns.
    pub fn directory(&self) -> &Path {
        &self.directory
    }

    /// Releases the directory and refuses every later operation.
    ///
    /// The lock goes at once, whoever still holds the store: an engine that
    /// was handed it, a host object its runtime has not freed yet. An
    /// operation in flight finishes first. Afterwards every operation is an
    /// error naming the directory, and a new store may open over it. Closing
    /// twice is closing once.
    pub fn close(&self) {
        self.lock.close();
    }

    /// Whether `<root>/<namespace>/` holds MLS records, without opening,
    /// creating or locking anything.
    ///
    /// A record file, not the directory and not a type directory. An open
    /// refused after the account directory was made (a wrong key, a
    /// protocol-state root that could not be created) leaves the directory,
    /// its lock file and its key check, and no identity. A first write that
    /// failed (a full disk, a crash) leaves a type directory with nothing in
    /// it but a temporary. A directory that cannot be read, at either level,
    /// is an [`FileStoreError::Io`], never "no records".
    pub fn holds_records(root: impl AsRef<Path>, namespace: &str) -> Result<bool, FileStoreError> {
        super::account_holds(root.as_ref(), namespace, |path, name| {
            if !name.starts_with(TYPE_PREFIX) {
                return Ok(false);
            }
            // A file that merely carries the prefix holds no record. Asked
            // of the entry, not left to the error a directory read gives for
            // a file, which differs by platform. `metadata`, not `is_dir`:
            // an account directory that can be listed but not searched
            // fails here, and `is_dir` would answer "not a directory", which
            // reads as "no records".
            if !std::fs::metadata(path)?.is_dir() {
                return Ok(false);
            }
            records::holds_entry(path, ENTRY_PREFIX)
        })
    }

    /// The pairing id this store holds, read under the store key.
    ///
    /// `Damaged` is a file that is there and does not open: it names no
    /// state root, and the pair falls back to asking the records.
    pub(super) fn pairing(&self) -> Result<StoredPairing, FileStoreError> {
        let _guard = self
            .lock
            .open()
            .ok_or_else(|| FileStoreError::Closed(self.directory.clone()))?;
        let path = self.directory.join(PAIRING_FILE);
        let raw = match records::read_bounded(&path, PAIRING_ID_BYTES + SEAL_OVERHEAD).map_err(
            |source| FileStoreError::Io {
                path: path.clone(),
                source,
            },
        )? {
            BoundedRead::Bytes(raw) => raw,
            BoundedRead::Absent => return Ok(StoredPairing::Absent),
            BoundedRead::Oversized => return Ok(StoredPairing::Damaged),
        };
        Ok(match self.open_sealed(PAIRING_FILE, &raw) {
            Some(plain) if plain.len() == PAIRING_ID_BYTES => {
                let mut id = [0u8; PAIRING_ID_BYTES];
                id.copy_from_slice(&plain);
                StoredPairing::Present(id)
            }
            _ => StoredPairing::Damaged,
        })
    }

    /// Records `id` as this store's pairing id, durably.
    pub(super) fn set_pairing(&self, id: &[u8; PAIRING_ID_BYTES]) -> Result<(), FileStoreError> {
        let _guard = self
            .lock
            .open()
            .ok_or_else(|| FileStoreError::Closed(self.directory.clone()))?;
        let sealed = self.seal(PAIRING_FILE, id).ok_or_else(|| {
            FileStoreError::StoreKey(StoreKeyError::Unavailable(
                "sealing the pairing id failed".to_string(),
            ))
        })?;
        let path = self.directory.join(PAIRING_FILE);
        records::write_atomic(&self.directory, &path, &sealed)
            .map_err(|source| FileStoreError::Io { path, source })
    }

    /// What a closed store answers.
    fn closed(&self) -> String {
        format!("the store at {} is closed", self.directory.display())
    }

    fn keyed_name(&self, prefix: &str, parts: &[&[u8]]) -> String {
        // Every part but the last is length-prefixed, so no two different
        // (type, id) pairs feed the MAC the same bytes.
        let mut mac = <Hmac<Sha256> as Mac>::new_from_slice(self.name_key.as_slice())
            .unwrap_or_else(|_| unreachable!("HMAC accepts a key of any length"));
        if let Some((last, rest)) = parts.split_last() {
            for part in rest {
                mac.update(&(part.len() as u64).to_be_bytes());
                mac.update(part);
            }
            mac.update(last);
        }
        format!("{prefix}{}", hex::encode(mac.finalize().into_bytes()))
    }

    fn type_directory(&self, key_type: &str) -> PathBuf {
        self.directory
            .join(self.keyed_name(TYPE_PREFIX, &[b"type", key_type.as_bytes()]))
    }

    fn entry_name(&self, key_type: &str, key_id: &str) -> String {
        self.keyed_name(
            ENTRY_PREFIX,
            &[b"entry", key_type.as_bytes(), key_id.as_bytes()],
        )
    }

    fn entry_path(&self, key_type: &str, key_id: &str) -> PathBuf {
        self.type_directory(key_type)
            .join(self.entry_name(key_type, key_id))
    }

    /// The file a record lives in, for tests outside this module that damage
    /// one on purpose.
    #[cfg(test)]
    pub(crate) fn record_path(&self, key_type: &str, key_id: &str) -> PathBuf {
        self.entry_path(key_type, key_id)
    }

    /// Reports a record that is present but cannot be read, and leaves it.
    fn corrupt(path: &Path, reason: &str) -> StorageError {
        tracing::warn!(path = %path.display(), reason, "sealed MLS record is unreadable; left in place");
        StorageError::CorruptedData(format!("{reason}: {}; left in place", path.display()))
    }

    fn associated_data(name: &str) -> Vec<u8> {
        let mut aad = Vec::with_capacity(SEALED_MAGIC.len() + name.len());
        aad.extend_from_slice(SEALED_MAGIC);
        aad.extend_from_slice(name.as_bytes());
        aad
    }

    fn seal(&self, name: &str, plaintext: &[u8]) -> Option<Vec<u8>> {
        let mut nonce = [0u8; NONCE_BYTES];
        OsRng.fill_bytes(&mut nonce);
        let ciphertext = self
            .cipher
            .encrypt(
                Nonce::from_slice(&nonce),
                Payload {
                    msg: plaintext,
                    aad: &Self::associated_data(name),
                },
            )
            .ok()?;
        let mut sealed = Vec::with_capacity(SEAL_OVERHEAD + plaintext.len());
        sealed.extend_from_slice(SEALED_MAGIC);
        sealed.extend_from_slice(&nonce);
        sealed.extend_from_slice(&ciphertext);
        Some(sealed)
    }

    fn open_sealed(&self, name: &str, sealed: &[u8]) -> Option<Zeroizing<Vec<u8>>> {
        let header = SEALED_MAGIC.len() + NONCE_BYTES;
        if sealed.len() < header + TAG_BYTES || &sealed[..SEALED_MAGIC.len()] != SEALED_MAGIC {
            return None;
        }
        self.cipher
            .decrypt(
                Nonce::from_slice(&sealed[SEALED_MAGIC.len()..header]),
                Payload {
                    msg: &sealed[header..],
                    aad: &Self::associated_data(name),
                },
            )
            .ok()
            .map(Zeroizing::new)
    }

    /// Opens one record file into its `(key_type, key_id)`. An error is a
    /// read that failed, never a record that does not open. The value is not
    /// copied: the only plaintext is the zeroizing buffer, dropped here.
    fn open_record(&self, path: &Path, name: &str) -> std::io::Result<Opened> {
        let raw = match records::read_bounded(path, MAX_SEALED_BYTES)? {
            BoundedRead::Bytes(raw) => raw,
            BoundedRead::Absent => return Ok(Opened::Absent),
            BoundedRead::Oversized => return Ok(Opened::DoesNotOpen),
        };
        let opened = self
            .open_sealed(name, &raw)
            .and_then(|plain| records::parse_header(&plain))
            .map_or(Opened::DoesNotOpen, |header| {
                Opened::Record(header.key_type, header.key_id)
            });
        Ok(opened)
    }

    /// Whether this is the first listing to warn about `path` since the file
    /// there was last written or removed.
    fn first_listing_warning(&self, path: &Path) -> bool {
        self.listing_warned
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .insert(path.to_path_buf())
    }

    /// Forgets that a listing warned about `path`, once the file there has
    /// been written or removed: a record damaged again later is news.
    fn forget_listing_warning(&self, path: &Path) {
        self.listing_warned
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .remove(path);
    }

    /// Proves the store key against the check file, minting the file on a
    /// fresh store and rebuilding it when a record proves the key.
    fn verify_store_key(&self) -> Result<(), FileStoreError> {
        let _guard = self
            .lock
            .open()
            .ok_or_else(|| FileStoreError::Closed(self.directory.clone()))?;
        // The check is the one file written into the account directory, so
        // a crash mid-write orphans its temporary here, where no store write
        // would ever sweep.
        self.swept.sweep_once(&self.directory);
        let path = self.directory.join(KEY_CHECK_FILE);
        let io = |source| FileStoreError::Io {
            path: path.clone(),
            source,
        };
        let check_present =
            match records::read_bounded(&path, KEY_CHECK_PLAINTEXT.len() + SEAL_OVERHEAD)
                .map_err(io)?
            {
                BoundedRead::Bytes(raw) => match self.open_sealed(KEY_CHECK_FILE, &raw) {
                    Some(plain) if plain.as_slice() == KEY_CHECK_PLAINTEXT => return Ok(()),
                    _ => true,
                },
                BoundedRead::Oversized => true,
                BoundedRead::Absent => false,
            };
        // The check is missing, or present and not opening. Records that
        // open under this key prove it, whatever happened to the check: a
        // crash before its first write finished, an operator's cleanup, or
        // damage to the check alone. Records that exist and none of which
        // opens mean the key is wrong, and minting a check for it would lock
        // the right key out.
        //
        // Rebuilding a damaged check on a record's proof costs nothing: anyone
        // who can damage the check can delete it, which already reaches this
        // path. A check that does not open with no records to prove anything
        // stays a refusal; there is nothing sealed to lose, and the operator
        // can remove the check file to start the store afresh.
        match self.probe_existing_records()? {
            Probe::Opened => {
                if check_present {
                    tracing::warn!(
                        path = %path.display(),
                        "store key check does not open but a record does; rebuilding the check"
                    );
                }
            }
            Probe::NoneOpened => return Err(FileStoreError::WrongStoreKey(self.directory.clone())),
            Probe::NoRecords if check_present => {
                return Err(FileStoreError::KeyCheckUnverifiable(path.clone()))
            }
            Probe::NoRecords => {}
        }
        let sealed = self
            .seal(KEY_CHECK_FILE, KEY_CHECK_PLAINTEXT)
            .ok_or_else(|| {
                FileStoreError::StoreKey(StoreKeyError::Unavailable(
                    "sealing the key check failed".to_string(),
                ))
            })?;
        records::write_atomic(&self.directory, &path, &sealed).map_err(io)
    }

    /// Tries up to [`KEY_CHECK_PROBE_LIMIT`] records against the store key.
    ///
    /// A record that cannot be read says nothing about the key, so it is
    /// skipped rather than counted against it, and a readable record later in
    /// the walk can still prove the key. Only when none opens is a read
    /// failure the answer, naming the record: refusing the key as wrong over
    /// a permission error would send the operator after the wrong fault.
    fn probe_existing_records(&self) -> Result<Probe, FileStoreError> {
        let directory_error = |source| FileStoreError::Io {
            path: self.directory.clone(),
            source,
        };
        let mut probe = Probe::NoRecords;
        let mut tried = 0usize;
        let mut unreadable: Option<FileStoreError> = None;
        for type_entry in std::fs::read_dir(&self.directory).map_err(directory_error)? {
            let type_entry = type_entry.map_err(directory_error)?;
            let is_type_dir = type_entry
                .file_name()
                .to_str()
                .is_some_and(|n| n.starts_with(TYPE_PREFIX))
                && type_entry.file_type().map(|t| t.is_dir()).unwrap_or(false);
            if !is_type_dir {
                continue;
            }
            let type_directory = type_entry.path();
            records::for_each_entry(
                &type_directory,
                ENTRY_PREFIX,
                KEY_CHECK_PROBE_LIMIT,
                |path, name| {
                    if matches!(probe, Probe::Opened) || tried >= KEY_CHECK_PROBE_LIMIT {
                        return;
                    }
                    match self.open_record(path, name) {
                        Ok(Opened::Record(..)) => {
                            tried += 1;
                            probe = Probe::Opened;
                        }
                        Ok(Opened::DoesNotOpen) => {
                            tried += 1;
                            probe = Probe::NoneOpened;
                        }
                        // Removed meanwhile: not a record.
                        Ok(Opened::Absent) => {}
                        Err(source) => {
                            unreadable.get_or_insert(FileStoreError::Io {
                                path: path.to_path_buf(),
                                source,
                            });
                        }
                    }
                },
            )
            .map_err(|source| FileStoreError::Io {
                path: type_directory.clone(),
                source,
            })?;
            if matches!(probe, Probe::Opened) || tried >= KEY_CHECK_PROBE_LIMIT {
                break;
            }
        }
        match (probe, unreadable) {
            (Probe::Opened, _) => Ok(Probe::Opened),
            (_, Some(err)) => Err(err),
            (probe, None) => Ok(probe),
        }
    }
}

fn derive(
    hkdf: &Hkdf<Sha256>,
    label: &[u8],
    namespace: &str,
) -> Result<Zeroizing<[u8; STORE_KEY_BYTES]>, FileStoreError> {
    let mut info = Vec::with_capacity(label.len() + 1 + namespace.len());
    info.extend_from_slice(label);
    info.push(0);
    info.extend_from_slice(namespace.as_bytes());
    let mut out = Zeroizing::new([0u8; STORE_KEY_BYTES]);
    hkdf.expand(&info, out.as_mut_slice()).map_err(|_| {
        FileStoreError::StoreKey(StoreKeyError::Unavailable(
            "key derivation failed".to_string(),
        ))
    })?;
    Ok(out)
}

impl MlsStorage for SealedFileMlsStorage {
    fn store(&self, key_type: &str, key_id: &str, data: &[u8]) -> StorageResult<()> {
        let framed = Zeroizing::new(
            records::frame(key_type, key_id, data)
                .map_err(|err| StorageError::StoreFailed(err.to_string()))?,
        );
        let name = self.entry_name(key_type, key_id);
        let sealed = self
            .seal(&name, &framed)
            .ok_or_else(|| StorageError::StoreFailed("sealing the record failed".to_string()))?;
        let _guard = self
            .lock
            .open()
            .ok_or_else(|| StorageError::StoreFailed(self.closed()))?;
        let directory = self.type_directory(key_type);
        records::private_mkdir(&directory)
            .map_err(|err| StorageError::StoreFailed(err.to_string()))?;
        self.swept.sweep_once(&directory);
        let path = directory.join(&name);
        records::write_atomic(&directory, &path, &sealed)
            .map_err(|err| StorageError::StoreFailed(err.to_string()))?;
        self.forget_listing_warning(&path);
        Ok(())
    }

    fn load(&self, key_type: &str, key_id: &str) -> StorageResult<Option<Vec<u8>>> {
        let _guard = self
            .lock
            .open()
            .ok_or_else(|| StorageError::LoadFailed(self.closed()))?;
        let name = self.entry_name(key_type, key_id);
        let path = self.type_directory(key_type).join(&name);
        let raw = match records::read_bounded(&path, MAX_SEALED_BYTES)
            .map_err(|err| StorageError::LoadFailed(err.to_string()))?
        {
            BoundedRead::Absent => return Ok(None),
            BoundedRead::Oversized => {
                return Err(Self::corrupt(&path, "sealed record is over the ceiling"))
            }
            BoundedRead::Bytes(raw) => raw,
        };
        let plain = self.open_sealed(&name, &raw).ok_or_else(|| {
            Self::corrupt(
                &path,
                "sealed record does not authenticate under this store key",
            )
        })?;
        // The file name is in the associated data, so a record moved onto
        // another's name has already failed to open. This check is the second
        // lock: it holds even if two keys' names ever collided.
        match records::parse_header(&plain) {
            Some(header) if header.key_type == key_type && header.key_id == key_id => {
                Ok(Some(plain[header.value_offset..].to_vec()))
            }
            _ => Err(Self::corrupt(&path, "sealed record names another key")),
        }
    }

    fn delete(&self, key_type: &str, key_id: &str) -> StorageResult<()> {
        let _guard = self
            .lock
            .open()
            .ok_or_else(|| StorageError::DeleteFailed(self.closed()))?;
        let path = self.entry_path(key_type, key_id);
        records::remove(&path).map_err(|err| StorageError::DeleteFailed(err.to_string()))?;
        self.forget_listing_warning(&path);
        Ok(())
    }

    fn list_keys(&self, key_type: &str) -> StorageResult<Vec<String>> {
        let _guard = self
            .lock
            .open()
            .ok_or_else(|| StorageError::LoadFailed(self.closed()))?;
        let mut keys = BTreeSet::new();
        records::for_each_entry(
            &self.type_directory(key_type),
            ENTRY_PREFIX,
            records::MAX_LISTED_ENTRIES,
            |path, name| match self.open_record(path, name) {
                Ok(Opened::Record(record_type, record_id)) if record_type == key_type => {
                    keys.insert(record_id);
                }
                // Deleted since the directory was read: not a record.
                Ok(Opened::Absent) => {}
                // Left out of the listing, and named with the real cause. A
                // listing that failed outright on one unreadable file would
                // stop an engine from finding any of its groups.
                Err(err) => {
                    if self.first_listing_warning(path) {
                        tracing::warn!(
                            path = %path.display(),
                            error = %err,
                            "sealed MLS record could not be read; left out of the listing"
                        );
                    }
                }
                // The listing is how the engine finds its groups, so a record
                // skipped here is one the engine never learns it had. Same
                // level and same path as `load` reports: names are keyed
                // digests, and the path is the only way to find the file.
                // Once per file per store: listings are frequent.
                Ok(Opened::Record(..) | Opened::DoesNotOpen) => {
                    if self.first_listing_warning(path) {
                        tracing::warn!(
                            path = %path.display(),
                            "sealed MLS record does not open under this store key; \
                             left in place and left out of the listing"
                        );
                    }
                }
            },
        )
        .map_err(|err| StorageError::LoadFailed(err.to_string()))?;
        Ok(keys.into_iter().collect())
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::file_store::test_support::{files_under, TempRoot};
    use crate::file_store::{account_storage_namespace, StaticStoreKey};

    fn namespace() -> String {
        account_storage_namespace("com.example.mls", "alice")
    }

    fn key(byte: u8) -> StaticStoreKey {
        StaticStoreKey::new([byte; 32])
    }

    fn open(root: &TempRoot, byte: u8) -> SealedFileMlsStorage {
        SealedFileMlsStorage::open(root.path(), &namespace(), &key(byte)).expect("open")
    }

    fn snapshot(dir: &Path) -> Vec<(PathBuf, Vec<u8>)> {
        files_under(dir)
            .into_iter()
            .map(|p| {
                let bytes = std::fs::read(&p).expect("read");
                (p, bytes)
            })
            .collect()
    }

    /// An open leaves a lock file and a key check and no record, so the
    /// directory existing says nothing about an identity. Only a stored
    /// record counts, and the probe itself creates nothing.
    #[test]
    fn holds_records_counts_records_not_the_directory() {
        let root = TempRoot::new("mls-holds");
        let probe = || SealedFileMlsStorage::holds_records(root.path(), &namespace());
        assert!(!probe().expect("absent"));
        assert!(
            files_under(root.path()).is_empty(),
            "the probe creates nothing"
        );
        let store = open(&root, 1);
        assert!(!probe().expect("lock and key check only"));

        // What a first write that failed leaves: the type directory, and a
        // temporary in it. Neither is an identity. Nor is a file that merely
        // carries a type directory's prefix.
        let type_directory = store.type_directory("identity");
        std::fs::create_dir(&type_directory).expect("type directory");
        assert!(!probe().expect("an empty type directory"));
        std::fs::write(
            type_directory.join(format!("{}crashed", records::TEMP_PREFIX)),
            b"half a record",
        )
        .expect("temporary");
        std::fs::write(store.directory().join("c_not-a-directory"), b"").expect("stray file");
        assert!(!probe().expect("a temporary and a stray file"));

        store.store("identity", "self", b"key").expect("store");
        assert!(probe().expect("a record"));
        assert!(matches!(
            SealedFileMlsStorage::holds_records(root.path(), "../escape"),
            Err(FileStoreError::InvalidNamespace)
        ));
    }

    /// Closing releases the directory while the store is still held, which
    /// is the point: a host whose runtime has not freed the object yet must
    /// not keep the next store out. What the closed store then answers is a
    /// failed operation, never a lost record, because the engine settles a
    /// lost record as gone for good and regenerates a lost record key.
    #[test]
    fn a_closed_store_releases_the_directory_and_refuses_every_operation() {
        let root = TempRoot::new("mls-close");
        let store = open(&root, 1);
        store.store("identity", "self", b"key").expect("store");
        assert!(matches!(
            SealedFileMlsStorage::open(root.path(), &namespace(), &key(1)),
            Err(FileStoreError::InUse(_))
        ));

        store.close();
        store.close();
        let next = open(&root, 1);
        assert_eq!(
            next.load("identity", "self").expect("load"),
            Some(b"key".to_vec()),
            "the record written before the close is the next store's to read"
        );

        assert!(matches!(
            store.store("identity", "self", b"other"),
            Err(StorageError::StoreFailed(m)) if m.contains("closed")
        ));
        assert!(matches!(
            store.load("identity", "self"),
            Err(StorageError::LoadFailed(m)) if m.contains("closed")
        ));
        assert!(matches!(
            store.delete("identity", "self"),
            Err(StorageError::DeleteFailed(m)) if m.contains("closed")
        ));
        assert!(matches!(
            store.list_keys("identity"),
            Err(StorageError::LoadFailed(m)) if m.contains("closed")
        ));
        assert_eq!(
            next.load("identity", "self").expect("load"),
            Some(b"key".to_vec()),
            "a closed store changes nothing in a directory it no longer holds"
        );
    }

    #[test]
    fn the_store_passes_the_conformance_suite() {
        let root = TempRoot::new("mls-conformance");
        let report = offline_protocol_mls::storage_conformance::run(&open(&root, 1));
        assert!(report.is_green(), "{report:#?}");
    }

    #[test]
    fn records_survive_a_reopen_under_the_same_key() {
        let root = TempRoot::new("mls-reopen");
        let large = vec![0x5au8; offline_protocol_mls::storage_conformance::LARGE_RECORD_BYTES];
        {
            let store = open(&root, 1);
            store
                .store("identity", "key_pair", b"identity")
                .expect("store");
            store.store("group_state", "team", &large).expect("store");
        }
        let store = open(&root, 1);
        assert_eq!(
            store.load("identity", "key_pair").expect("load"),
            Some(b"identity".to_vec())
        );
        assert_eq!(
            store.load("group_state", "team").expect("load"),
            Some(large)
        );
        assert_eq!(store.list_keys("group_state").expect("list"), vec!["team"]);
    }

    #[test]
    fn a_wrong_key_is_refused_at_open_and_nothing_changes() {
        let root = TempRoot::new("mls-wrong-key");
        {
            let store = open(&root, 1);
            store
                .store("identity", "key_pair", b"identity")
                .expect("store");
        }
        let before = snapshot(root.path());

        let result = SealedFileMlsStorage::open(root.path(), &namespace(), &key(2));
        assert!(matches!(result, Err(FileStoreError::WrongStoreKey(_))));
        assert_eq!(snapshot(root.path()), before);
    }

    #[test]
    fn a_lost_key_check_is_rebuilt_only_by_the_right_key() {
        let root = TempRoot::new("mls-lost-check");
        let check = {
            let store = open(&root, 1);
            store
                .store("identity", "key_pair", b"identity")
                .expect("store");
            store.directory().join(KEY_CHECK_FILE)
        };
        std::fs::remove_file(&check).expect("remove check");
        let before = snapshot(root.path());

        // A wrong key must not mint a check for itself over the records.
        assert!(matches!(
            SealedFileMlsStorage::open(root.path(), &namespace(), &key(2)),
            Err(FileStoreError::WrongStoreKey(_))
        ));
        assert_eq!(snapshot(root.path()), before);

        let store = open(&root, 1);
        assert!(check.exists(), "the right key rebuilds the check");
        assert_eq!(
            store.load("identity", "key_pair").expect("load"),
            Some(b"identity".to_vec())
        );
    }

    /// Damage to the check alone must not lock the right key out of its own
    /// identity: a record that opens proves the key, and the check is
    /// rebuilt. A wrong key still proves nothing and is still refused.
    #[test]
    fn a_damaged_key_check_is_rebuilt_only_by_the_right_key() {
        let root = TempRoot::new("mls-damaged-check");
        let check = {
            let store = open(&root, 1);
            store
                .store("identity", "key_pair", b"identity")
                .expect("store");
            store.directory().join(KEY_CHECK_FILE)
        };
        let mut bytes = std::fs::read(&check).expect("read");
        let last = bytes.len() - 1;
        bytes[last] ^= 1;
        std::fs::write(&check, &bytes).expect("write");
        let before = snapshot(root.path());

        assert!(matches!(
            SealedFileMlsStorage::open(root.path(), &namespace(), &key(2)),
            Err(FileStoreError::WrongStoreKey(_))
        ));
        assert_eq!(snapshot(root.path()), before, "a wrong key changed nothing");

        let store = open(&root, 1);
        assert_ne!(std::fs::read(&check).expect("read"), bytes, "check rebuilt");
        assert_eq!(
            store.load("identity", "key_pair").expect("load"),
            Some(b"identity".to_vec())
        );
        drop(store);
        // And the rebuilt check is a real one: the next open takes the fast path.
        open(&root, 1);
    }

    /// With no record to prove anything, a check that does not open stays a
    /// refusal, whichever key is offered. The error is not `WrongStoreKey`,
    /// because the key may be right, and it names the check file, because
    /// removing that file is the remedy when it is.
    #[test]
    fn a_damaged_key_check_over_an_empty_store_is_refused() {
        let root = TempRoot::new("mls-damaged-empty");
        let check = open(&root, 1).directory().join(KEY_CHECK_FILE);
        let mut bytes = std::fs::read(&check).expect("read");
        let last = bytes.len() - 1;
        bytes[last] ^= 1;
        std::fs::write(&check, &bytes).expect("write");
        for byte in [1, 2] {
            match SealedFileMlsStorage::open(root.path(), &namespace(), &key(byte)) {
                Err(FileStoreError::KeyCheckUnverifiable(path)) => assert_eq!(path, check),
                other => panic!("expected KeyCheckUnverifiable, got {other:?}"),
            }
        }
        assert_eq!(
            std::fs::read(&check).expect("read"),
            bytes,
            "nothing changed"
        );

        // The remedy the error names: remove the check and the store opens.
        std::fs::remove_file(&check).expect("remove check");
        open(&root, 1);
    }

    /// The probe keeps going past a record that does not open: one damaged
    /// record must not hide the good one that proves the key.
    #[test]
    fn a_lost_key_check_is_rebuilt_past_an_unreadable_record() {
        let _callsite = listing_warning_callsite();
        let root = TempRoot::new("mls-lost-check-damaged");
        let (check, damaged) = {
            let store = open(&root, 1);
            store.store("group_state", "a", b"one").expect("store");
            store.store("group_state", "b", b"two").expect("store");
            (
                store.directory().join(KEY_CHECK_FILE),
                store.entry_path("group_state", "a"),
            )
        };
        std::fs::remove_file(&check).expect("remove check");
        let mut bytes = std::fs::read(&damaged).expect("read");
        bytes[SEALED_MAGIC.len() + NONCE_BYTES] ^= 1;
        std::fs::write(&damaged, bytes).expect("write");

        let store = open(&root, 1);
        assert!(check.exists(), "the good record proved the key");
        assert_eq!(store.list_keys("group_state").expect("list"), vec!["b"]);
    }

    /// Held by every test that lists a store holding a record that does not
    /// open, which is every test that reaches the listing's warning.
    ///
    /// `tracing` caches whether a callsite is wanted process-wide, and the
    /// first thread to reach a callsite fills that cache. A test thread with
    /// no subscriber can compute "never" before the capturing test installs
    /// its subscriber and store it after, which hides the warning from the
    /// capture. A test holding this guard has finished any first
    /// registration before it releases it, so the capturing test's rebuild
    /// always comes after. Any new test that lists an unopenable record
    /// takes it too.
    static LISTING_WARNING_CALLSITE: std::sync::Mutex<()> = std::sync::Mutex::new(());

    fn listing_warning_callsite() -> std::sync::MutexGuard<'static, ()> {
        LISTING_WARNING_CALLSITE
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
    }

    /// Captures every warning raised on this thread, with its fields.
    #[derive(Clone, Default)]
    struct Warnings(std::sync::Arc<std::sync::Mutex<Vec<String>>>);

    impl tracing::Subscriber for Warnings {
        fn enabled(&self, metadata: &tracing::Metadata<'_>) -> bool {
            *metadata.level() <= tracing::Level::WARN
        }
        fn new_span(&self, _: &tracing::span::Attributes<'_>) -> tracing::span::Id {
            tracing::span::Id::from_u64(1)
        }
        fn record(&self, _: &tracing::span::Id, _: &tracing::span::Record<'_>) {}
        fn record_follows_from(&self, _: &tracing::span::Id, _: &tracing::span::Id) {}
        fn event(&self, event: &tracing::Event<'_>) {
            struct Fields(String);
            impl tracing::field::Visit for Fields {
                fn record_debug(
                    &mut self,
                    field: &tracing::field::Field,
                    value: &dyn std::fmt::Debug,
                ) {
                    self.0.push_str(&format!("{}={value:?} ", field.name()));
                }
            }
            let mut fields = Fields(String::new());
            event.record(&mut fields);
            self.0.lock().expect("warnings").push(fields.0);
        }
        fn enter(&self, _: &tracing::span::Id) {}
        fn exit(&self, _: &tracing::span::Id) {}
    }

    /// The listing is how the engine enumerates its groups, so a record it
    /// skips is one the engine never learns about. It must be reported as
    /// loudly as `load` reports it, with the path, since names are digests.
    #[test]
    fn a_record_the_listing_skips_is_warned_about_with_its_path() {
        let root = TempRoot::new("mls-list-warn");
        let store = open(&root, 1);
        store.store("group_state", "a", b"one").expect("store");
        store.store("group_state", "b", b"two").expect("store");
        let damaged = store.entry_path("group_state", "a");
        let mut bytes = std::fs::read(&damaged).expect("read");
        bytes[SEALED_MAGIC.len() + NONCE_BYTES] ^= 1;
        std::fs::write(&damaged, bytes).expect("write");

        // One subscriber for the whole test, installed while no other test
        // can be registering the callsite, and the interest cache rebuilt
        // once it is in place (see `LISTING_WARNING_CALLSITE`).
        let _callsite = listing_warning_callsite();
        let warnings = Warnings::default();
        let _default = tracing::dispatcher::set_default(&tracing::Dispatch::new(warnings.clone()));
        tracing::callsite::rebuild_interest_cache();

        let path = damaged.display().to_string();
        let named = || {
            warnings
                .0
                .lock()
                .expect("warnings")
                .iter()
                .filter(|w| w.contains(&path))
                .count()
        };

        assert_eq!(store.list_keys("group_state").expect("list"), vec!["b"]);
        assert_eq!(named(), 1, "one warning names {path}");

        // Listings are frequent; a record that stays damaged is warned about
        // once per store, not on every listing.
        assert_eq!(store.list_keys("group_state").expect("list"), vec!["b"]);
        assert_eq!(named(), 1, "the second listing warned again");
        assert!(store
            .listing_warned
            .lock()
            .expect("memo")
            .contains(&damaged));
    }

    /// A record written or removed after a listing warned about it is a new
    /// file: damaged again later, it is warned about again.
    #[test]
    fn a_record_rewritten_after_a_warning_is_warned_about_again() {
        let root = TempRoot::new("mls-list-rewarn");
        let store = open(&root, 1);
        store.store("group_state", "a", b"one").expect("store");
        let damaged = store.entry_path("group_state", "a");
        let damage = || {
            let mut bytes = std::fs::read(&damaged).expect("read");
            bytes[SEALED_MAGIC.len() + NONCE_BYTES] ^= 1;
            std::fs::write(&damaged, bytes).expect("write");
        };
        damage();

        let _callsite = listing_warning_callsite();
        let warnings = Warnings::default();
        let _default = tracing::dispatcher::set_default(&tracing::Dispatch::new(warnings.clone()));
        tracing::callsite::rebuild_interest_cache();
        let path = damaged.display().to_string();
        let named = || {
            warnings
                .0
                .lock()
                .expect("warnings")
                .iter()
                .filter(|w| w.contains(&path))
                .count()
        };

        assert!(store.list_keys("group_state").expect("list").is_empty());
        assert_eq!(named(), 1);

        store.store("group_state", "a", b"again").expect("store");
        damage();
        assert!(store.list_keys("group_state").expect("list").is_empty());
        assert_eq!(
            named(),
            2,
            "a rewritten record damaged again was not reported"
        );

        store.delete("group_state", "a").expect("delete");
        assert!(
            !store
                .listing_warned
                .lock()
                .expect("memo")
                .contains(&damaged),
            "a removed record's warning is still remembered"
        );
    }

    /// Makes `path` unreadable, or answers false where the test runs with
    /// the power to read it anyway.
    #[cfg(unix)]
    fn make_unreadable(path: &Path) -> bool {
        use std::os::unix::fs::PermissionsExt;
        std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o000)).expect("chmod");
        std::fs::read(path).is_err()
    }

    #[cfg(unix)]
    fn make_readable(path: &Path) {
        use std::os::unix::fs::PermissionsExt;
        std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o600)).expect("chmod");
    }

    /// A record that cannot be read is not evidence against the store key.
    /// Counted as one that does not open, a permission error on the only
    /// record refused the right key as a wrong one.
    #[cfg(unix)]
    #[test]
    fn an_unreadable_record_is_an_io_error_not_a_wrong_key() {
        let root = TempRoot::new("mls-unreadable-probe");
        let (check, record) = {
            let store = open(&root, 1);
            store.store("group_state", "a", b"one").expect("store");
            (
                store.directory().join(KEY_CHECK_FILE),
                store.entry_path("group_state", "a"),
            )
        };
        std::fs::remove_file(&check).expect("remove check");
        if !make_unreadable(&record) {
            return;
        }
        let outcome = SealedFileMlsStorage::open(root.path(), &namespace(), &key(1));
        make_readable(&record);
        match outcome {
            Err(FileStoreError::Io { path, .. }) => {
                assert_eq!(path, record, "the error does not name the record");
            }
            Err(other) => panic!("expected an I/O error naming the record, got {other}"),
            Ok(_) => panic!("an unreadable record proved the key"),
        }
        assert!(
            !check.exists(),
            "no key check was minted on an unread record"
        );
        open(&root, 1);
        assert!(check.exists(), "once readable, the record proves the key");
    }

    /// An unreadable record does not stand in the way of a readable one that
    /// proves the key, whichever the directory lists first.
    ///
    /// File names are keyed digests, so the order the probe meets records in
    /// is not the test's to choose. Four of five are unreadable, so a probe
    /// that stopped at the first unreadable record fails here on most runs.
    #[cfg(unix)]
    #[test]
    fn an_unreadable_record_does_not_hide_one_that_proves_the_key() {
        let root = TempRoot::new("mls-unreadable-beside-good");
        let ids = ["a", "b", "c", "d", "e"];
        let (check, unreadable) = {
            let store = open(&root, 1);
            for id in ids {
                store.store("group_state", id, b"value").expect("store");
            }
            (
                store.directory().join(KEY_CHECK_FILE),
                ids[1..]
                    .iter()
                    .map(|id| store.entry_path("group_state", id))
                    .collect::<Vec<_>>(),
            )
        };
        std::fs::remove_file(&check).expect("remove check");
        if !unreadable.iter().all(|path| make_unreadable(path)) {
            unreadable.iter().for_each(|path| make_readable(path));
            return;
        }
        let outcome = SealedFileMlsStorage::open(root.path(), &namespace(), &key(1));
        unreadable.iter().for_each(|path| make_readable(path));
        outcome.expect("a readable record proves the key");
        assert!(check.exists(), "the key check is rebuilt");
    }

    /// The listing names a read failure as one, not as a record sealed under
    /// another key, and still returns the records it could read.
    #[cfg(unix)]
    #[test]
    fn an_unreadable_record_is_listed_as_unreadable() {
        let root = TempRoot::new("mls-unreadable-list");
        let store = open(&root, 1);
        store.store("group_state", "a", b"one").expect("store");
        store.store("group_state", "b", b"two").expect("store");
        let record = store.entry_path("group_state", "a");
        if !make_unreadable(&record) {
            return;
        }

        let _callsite = listing_warning_callsite();
        let warnings = Warnings::default();
        let _default = tracing::dispatcher::set_default(&tracing::Dispatch::new(warnings.clone()));
        tracing::callsite::rebuild_interest_cache();

        let listed = store.list_keys("group_state");
        make_readable(&record);
        assert_eq!(listed.expect("list"), vec!["b"]);
        let path = record.display().to_string();
        let said: Vec<String> = warnings
            .0
            .lock()
            .expect("warnings")
            .iter()
            .filter(|w| w.contains(&path))
            .cloned()
            .collect();
        assert_eq!(said.len(), 1, "{said:?}");
        assert!(
            said[0].contains("could not be read") && !said[0].contains("store key"),
            "the warning blames the key for a read failure: {}",
            said[0]
        );
    }

    /// Nothing to write down in a record's name, so the error must carry the
    /// path: file names are keyed digests, and an operator cannot find the
    /// damaged file any other way.
    #[test]
    fn a_corrupt_record_error_names_its_file() {
        let root = TempRoot::new("mls-corrupt-path");
        let store = open(&root, 1);
        store.store("identity", "key_pair", b"id").expect("store");
        let path = store.entry_path("identity", "key_pair");
        let mut bytes = std::fs::read(&path).expect("read");
        let last = bytes.len() - 1;
        bytes[last] ^= 1;
        std::fs::write(&path, bytes).expect("write");

        match store.load("identity", "key_pair") {
            Err(StorageError::CorruptedData(message)) => assert!(
                message.contains(&path.display().to_string()),
                "{message} does not name {}",
                path.display()
            ),
            other => panic!("expected CorruptedData, got {other:?}"),
        }
    }

    #[test]
    fn a_second_store_over_one_directory_is_refused() {
        let root = TempRoot::new("mls-in-use");
        let first = open(&root, 1);
        assert!(matches!(
            SealedFileMlsStorage::open(root.path(), &namespace(), &key(1)),
            Err(FileStoreError::InUse(_))
        ));
        drop(first);
        open(&root, 1);
    }

    /// The point of sealing names as well as values: a copied directory must
    /// not say which peers or groups the device knows.
    #[test]
    fn neither_values_nor_keys_reach_the_disk_in_the_clear() {
        let root = TempRoot::new("mls-opaque");
        let store = open(&root, 1);
        let secret_value = b"the-secret-value-marker";
        let secret_id = "group-named-secret-team";
        let secret_type = "contact_key_package";
        store
            .store(secret_type, secret_id, secret_value)
            .expect("store");

        for (path, bytes) in snapshot(root.path()) {
            let name = path.to_string_lossy();
            for needle in [
                &secret_value[..],
                secret_id.as_bytes(),
                secret_type.as_bytes(),
            ] {
                assert!(
                    !bytes.windows(needle.len()).any(|w| w == needle),
                    "{} holds {:?} in the clear",
                    name,
                    String::from_utf8_lossy(needle)
                );
                assert!(!name.contains(std::str::from_utf8(needle).expect("utf8")));
            }
        }
    }

    #[test]
    fn a_record_moved_onto_another_name_does_not_open_and_is_kept() {
        let _callsite = listing_warning_callsite();
        let root = TempRoot::new("mls-moved");
        let store = open(&root, 1);
        store.store("group_state", "a", b"one").expect("store");
        store.store("group_state", "b", b"two").expect("store");
        let a = store.entry_path("group_state", "a");
        std::fs::copy(store.entry_path("group_state", "b"), &a).expect("copy");

        assert!(matches!(
            store.load("group_state", "a"),
            Err(StorageError::CorruptedData(_))
        ));
        assert!(a.exists(), "the sealed store never deletes on a read");
        assert_eq!(store.list_keys("group_state").expect("list"), vec!["b"]);
        assert_eq!(
            store.load("group_state", "b").expect("load"),
            Some(b"two".to_vec())
        );
    }

    #[test]
    fn a_tampered_or_oversized_record_is_corrupt_and_kept() {
        let _callsite = listing_warning_callsite();
        let root = TempRoot::new("mls-tampered");
        let store = open(&root, 1);
        store
            .store("identity", "key_pair", b"identity")
            .expect("store");
        store.store("identity", "other", b"other").expect("store");
        let path = store.entry_path("identity", "key_pair");
        let mut bytes = std::fs::read(&path).expect("read");
        bytes[SEALED_MAGIC.len() + NONCE_BYTES] ^= 1;
        std::fs::write(&path, &bytes).expect("write");

        assert!(matches!(
            store.load("identity", "key_pair"),
            Err(StorageError::CorruptedData(_))
        ));
        assert!(path.exists());
        assert_eq!(store.list_keys("identity").expect("list"), vec!["other"]);

        let handle = std::fs::OpenOptions::new()
            .write(true)
            .open(&path)
            .expect("open");
        handle
            .set_len((MAX_SEALED_BYTES + 1) as u64)
            .expect("sparse extend");
        drop(handle);
        assert!(matches!(
            store.load("identity", "key_pair"),
            Err(StorageError::CorruptedData(_))
        ));
        assert!(path.exists());
    }

    /// The seal and name keys are derived per namespace, so one store key
    /// serving two accounts does not let one account's record open in the
    /// other's directory.
    #[test]
    fn a_record_copied_into_another_account_does_not_open() {
        let root = TempRoot::new("mls-accounts");
        let alice = open(&root, 1);
        let bob = SealedFileMlsStorage::open(
            root.path(),
            &account_storage_namespace("com.example.mls", "bob"),
            &key(1),
        )
        .expect("open");
        alice
            .store("identity", "key_pair", b"alice")
            .expect("store");
        bob.store("identity", "key_pair", b"bob").expect("store");

        std::fs::copy(
            alice.entry_path("identity", "key_pair"),
            bob.entry_path("identity", "key_pair"),
        )
        .expect("copy");
        assert!(matches!(
            bob.load("identity", "key_pair"),
            Err(StorageError::CorruptedData(_))
        ));
        assert_ne!(
            alice.entry_name("identity", "key_pair"),
            bob.entry_name("identity", "key_pair")
        );
    }

    #[test]
    fn ids_differing_only_in_case_are_distinct_records() {
        let root = TempRoot::new("mls-case");
        let store = open(&root, 1);
        store.store("group_state", "Team", b"upper").expect("store");
        store.store("group_state", "team", b"lower").expect("store");
        assert_eq!(
            store.load("group_state", "Team").expect("load"),
            Some(b"upper".to_vec())
        );
        assert_eq!(
            store.load("group_state", "team").expect("load"),
            Some(b"lower".to_vec())
        );
    }

    /// Length-prefixing keeps `("ab", "c")` and `("a", "bc")` apart.
    #[test]
    fn a_type_and_id_split_differently_name_different_files() {
        let root = TempRoot::new("mls-split");
        let store = open(&root, 1);
        assert_ne!(store.entry_name("ab", "c"), store.entry_name("a", "bc"));
    }
}