car-sync 0.37.0

//! Org-key ROTATION on member removal (slice 10) — the granter-side primitive
//! that bumps the epoch, plus the authenticated freshness marker that decides
//! which epoch is allowed to author the org's *present*.
//!
//! ## Inertness — scoped precisely (grep-provable)
//!
//! As of slice 11 exactly ONE function here has a live call site:
//! [`resolve_all_org_roots`], wired into `open_sync_subsystem` behind the
//! DEFAULT-OFF `PARSLEE_SYNC_ORG_SCOPE` gate (unset → byte-identical personal-only
//! path; still audit-gated for real tenants). It only READS the epochs a member
//! can unwrap — provisioning, no rotation authority.
//!
//! Everything ELSE remains a leaf primitive with NO live call site — like slice
//! 9's `MultiEpochOrgCipher`, shipping the MECHANISM and leaving the wiring for
//! later (audit-gated): [`rotate_org_key`], [`sign_rotation_floor`],
//! [`verify_rotation_floor`], and in particular [`is_authoritative`], a PURE
//! predicate deliberately NOT wired into the reducer (`SyncState`/`fold_onto`) —
//! live-fold enforcement of freshness is deferred to the cryptographer audit. No
//! floor is consumed anywhere yet, so the anti-rollback caller obligation (see
//! [`verify_rotation_floor`]) does not attach to any live path.
//!
//! ## What rotation is (and is NOT)
//!
//! Rotation does not — cannot — take a key away. History was authored under the
//! old `K_org`, and remaining members must keep reading it ([`crate::org_key_provider::OrgAwareKeyProvider`]
//! holds every epoch it can unwrap). Rotation is about which key is allowed to
//! author NEW ops: mint a fresh `K_org@(N+1)` ([`crate::crypto::generate_org_key`],
//! independent of `N`), wrap it for the REMAINING members only, publish at epoch
//! N+1, and raise a signed [`RotationFloor`] so remaining members treat post-cut
//! epoch-N writes as non-authoritative.
//!
//! ## The freshness boundary — what it does and does NOT prove (audit note)
//!
//! [`is_authoritative`] gates on `kid >= floor_epoch || hlc <= rotation_hlc`. The
//! `kid` half is UNFORGEABLE: a removed member cannot produce an op at epoch N+1
//! (they lack `K_org@(N+1)` and cannot derive it). The `hlc` half is only a
//! best-effort partition of HONEST history — the HLC is client-stamped
//! ([`crate::oplog::Hlc`]), so a removed member holding `K_org@N` can BACKDATE an
//! op to `hlc <= rotation_hlc` and slip past the historical side. That is
//! tolerable ONLY because such a forged op is confined to epoch N (already
//! readable to them, constraint (c)) and can only ever LOSE last-writer-wins to a
//! genuine epoch-(N+1) op. Fully refusing a removed member's writes needs per-op
//! author signatures bound to a membership epoch, or backend admission authz —
//! both DEFERRED to the audit / later slices. Do not read this module as closing
//! that residual.

use zeroize::Zeroizing;

use crate::crypto::{
    from_hex, parse_x25519_pub, to_hex, unwrap_org_key, wrap_org_key, CryptoError, WrappedOrgKey,
};
use crate::oplog::Hlc;
use crate::org_key_directory::{MemberPublicKey, OrgKeyDirectory, OrgKeyDirectoryError};
use ed25519_dalek::{Signature, Signer, SigningKey, VerifyingKey};
use x25519_dalek::{PublicKey, StaticSecret};

/// Domain tag for the rotation-floor signature transcript. Distinct from the
/// wrap tag so a floor signature can never be replayed as a wrap signature.
const ALG_ROTATION_FLOOR: &str = "org-rotation-floor/v1";

/// Failures rotating an org key or handling a rotation floor.
#[derive(Debug)]
pub enum RotationError {
    /// A wrap failed to construct (e.g. a low-order recipient key).
    Crypto(CryptoError),
    /// Publishing to the directory failed (I/O or backend authz refusal).
    Directory(OrgKeyDirectoryError),
    /// The floor signature did not verify against any trusted granter key.
    Untrusted,
    /// The floor is not internally well-formed: `floor_epoch <= prev_floor_epoch`.
    /// NOTE: this is a self-consistency check, NOT anti-rollback — see
    /// [`verify_rotation_floor`] for why the caller still owns replay defense.
    NonMonotonic,
    /// The `car_floor` algorithm tag was not [`ALG_ROTATION_FLOOR`].
    WrongAlgorithm(String),
    /// The stored signature hex was malformed.
    BadSignature(String),
}

impl std::fmt::Display for RotationError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            RotationError::Crypto(e) => write!(f, "rotation crypto error: {e}"),
            RotationError::Directory(e) => write!(f, "rotation directory error: {e}"),
            RotationError::Untrusted => {
                write!(f, "rotation floor not signed by a trusted granter")
            }
            RotationError::NonMonotonic => {
                write!(f, "rotation floor epoch not strictly greater than previous")
            }
            RotationError::WrongAlgorithm(t) => {
                write!(f, "rotation floor has unexpected algorithm tag: {t:?}")
            }
            RotationError::BadSignature(m) => write!(f, "rotation floor signature malformed: {m}"),
        }
    }
}

impl std::error::Error for RotationError {}

impl From<CryptoError> for RotationError {
    fn from(e: CryptoError) -> Self {
        RotationError::Crypto(e)
    }
}
impl From<OrgKeyDirectoryError> for RotationError {
    fn from(e: OrgKeyDirectoryError) -> Self {
        RotationError::Directory(e)
    }
}

/// A signed, monotonic marker that org `org` rotated to `floor_epoch` at logical
/// time `rotation_hlc`. Remaining members use it via [`is_authoritative`] to
/// stop treating a removed member's post-cut epoch-`<floor_epoch>` writes as the
/// org's current truth.
///
/// The signature (by a TRUSTED granter, the same authority that wraps `K_org`) is
/// LOAD-BEARING: an unsigned/unverified floor is a censorship primitive — anyone
/// could publish `floor_epoch = u64::MAX` and mark every legitimate op stale.
/// Always [`verify_rotation_floor`] before honoring one.
#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct RotationFloor {
    /// Algorithm tag ([`ALG_ROTATION_FLOOR`]).
    pub car_floor: String,
    pub org: String,
    /// The epoch at/after which authorship is required for new ops.
    pub floor_epoch: u64,
    /// The logical time the cut happened. Ops with `hlc <= rotation_hlc` are
    /// treated as pre-cut history (see the module's audit note on forgeability).
    pub rotation_hlc: Hlc,
    /// The floor this one supersedes. `verify_rotation_floor` checks
    /// `floor_epoch > prev_floor_epoch` for internal well-formedness ONLY — that
    /// does NOT stop an attacker replaying an OLDER validly-signed floor (whose
    /// own `floor_epoch > prev_floor_epoch` still holds). Anti-rollback is the
    /// caller's job: persist the max accepted `floor_epoch` and reject anything
    /// `<=` it. See [`verify_rotation_floor`].
    pub prev_floor_epoch: u64,
    /// Ed25519 signature (hex) by a trusted granter over
    /// [`rotation_floor_transcript`]. Verified with `verify_strict`.
    pub signature: String,
}

/// The bytes a [`RotationFloor`] signs — domain-tagged, length-prefixed, all
/// integers little-endian, matching `wrap_sign_transcript`'s discipline so the
/// encoding is injective (no field-boundary ambiguity) and one endianness holds
/// crate-wide. EVERY security-relevant field is bound here; under-binding would
/// pass functional tests while silently letting a field be tampered.
fn rotation_floor_transcript(
    org: &str,
    floor_epoch: u64,
    rotation_hlc: &Hlc,
    prev_floor_epoch: u64,
) -> Vec<u8> {
    fn lp(buf: &mut Vec<u8>, field: &[u8]) {
        buf.extend_from_slice(&(field.len() as u64).to_le_bytes());
        buf.extend_from_slice(field);
    }
    let mut t = Vec::new();
    lp(&mut t, ALG_ROTATION_FLOOR.as_bytes());
    lp(&mut t, org.as_bytes());
    t.extend_from_slice(&floor_epoch.to_le_bytes());
    // Bind the full HLC triple (wall_ms, counter, device_id).
    t.extend_from_slice(&rotation_hlc.wall_ms.to_le_bytes());
    t.extend_from_slice(&rotation_hlc.counter.to_le_bytes());
    lp(&mut t, rotation_hlc.device_id.as_bytes());
    t.extend_from_slice(&prev_floor_epoch.to_le_bytes());
    t
}

/// Sign a rotation floor as a trusted granter. Callers must ensure
/// `floor_epoch > prev_floor_epoch` (also re-checked on verify).
pub fn sign_rotation_floor(
    org: &str,
    floor_epoch: u64,
    rotation_hlc: Hlc,
    prev_floor_epoch: u64,
    signer: &SigningKey,
) -> RotationFloor {
    let sig = signer.sign(&rotation_floor_transcript(
        org,
        floor_epoch,
        &rotation_hlc,
        prev_floor_epoch,
    ));
    RotationFloor {
        car_floor: ALG_ROTATION_FLOOR.to_string(),
        org: org.to_string(),
        floor_epoch,
        rotation_hlc,
        prev_floor_epoch,
        signature: to_hex(&sig.to_bytes()),
    }
}

/// Verify a rotation floor: its `car_floor` tag must be [`ALG_ROTATION_FLOOR`],
/// its signature must `verify_strict` against one of the caller's `trusted`
/// granter keys, AND it must be internally well-formed
/// (`floor_epoch > prev_floor_epoch`). Returns the verified `floor_epoch` on
/// success. Fail-closed: any tag/parse/verify/well-formedness failure is an
/// `Err`, never a silently-accepted floor.
///
/// ANTI-ROLLBACK IS NOT DONE HERE. This function is stateless, so it CANNOT stop
/// an attacker replaying an OLDER, legitimately-signed floor: `(floor_epoch=2,
/// prev=1)` is well-formed and verifies even after the org has advanced to
/// `(5,4)`, downgrading the boundary from 5 back to 2. The CALLER MUST persist
/// the maximum `floor_epoch` it has ever accepted for the org and reject any
/// verified floor whose `floor_epoch <= that maximum`. The
/// `floor_epoch > prev_floor_epoch` check is only self-consistency, not replay
/// defense — do not mistake it for one.
pub fn verify_rotation_floor(
    floor: &RotationFloor,
    trusted: &[VerifyingKey],
) -> Result<u64, RotationError> {
    // Algorithm tag — reject foreign/garbage floors before touching the signature
    // (mirrors unwrap_org_key's car_wrap check). The signature domain is bound to
    // the CONSTANT tag in the transcript, so this also keeps car_floor honest.
    if floor.car_floor != ALG_ROTATION_FLOOR {
        return Err(RotationError::WrongAlgorithm(floor.car_floor.clone()));
    }
    // Well-formedness only — NOT anti-rollback (see the doc above; the caller
    // tracks the max accepted floor_epoch).
    if floor.floor_epoch <= floor.prev_floor_epoch {
        return Err(RotationError::NonMonotonic);
    }
    let sig_bytes: [u8; 64] = from_hex(&floor.signature)
        .map_err(RotationError::BadSignature)?
        .try_into()
        .map_err(|_| RotationError::BadSignature("signature must be 64 bytes".into()))?;
    let signature = Signature::from_bytes(&sig_bytes);
    let transcript = rotation_floor_transcript(
        &floor.org,
        floor.floor_epoch,
        &floor.rotation_hlc,
        floor.prev_floor_epoch,
    );
    let ok = trusted
        .iter()
        .any(|vk| vk.verify_strict(&transcript, &signature).is_ok());
    if ok {
        Ok(floor.floor_epoch)
    } else {
        Err(RotationError::Untrusted)
    }
}

/// Is an op — identified by its org-cipher epoch `kid` and its `hlc` — still the
/// org's CURRENT authoritative truth under `floor`?
///
/// - `kid >= floor.floor_epoch` → authored at/after the cut with a key only a
///   remaining member holds. UNFORGEABLE authority.
/// - `hlc <= floor.rotation_hlc` → pre-cut history, kept authoritative so reads
///   stay non-lossy (constraint (c)). Best-effort for the honest case only — see
///   the module audit note; a backdated forgery lands here but can only lose LWW.
/// - otherwise → a post-cut write under a stale epoch → NOT authoritative.
///
/// `kid == None` (a personal-shaped envelope) is never org-authoritative on the
/// org path — org ciphertext always carries a `kid` (slice 9). Such an op only
/// stays authoritative via the pre-cut `hlc` allowance.
///
/// WIRING CONTRACT (constraint (a) — tolerate-the-window): a `false` here means
/// "does NOT override a competing op", i.e. an **LWW tiebreak input** — it must
/// NOT be used as a filter that DROPS the op. An honest lagging member who hasn't
/// yet received the new-epoch grant keeps writing the old `kid` with post-cut
/// timestamps and returns `false` here, indistinguishable from a malicious
/// stale write by `(kid, hlc)` alone. Dropping it would lose that member's work;
/// instead let it survive UNLESS a genuine higher-epoch op competes, so
/// convergence heals once the member catches up and re-emits under the new epoch.
/// Wiring this as a discard violates (a).
pub fn is_authoritative(kid: Option<u64>, hlc: &Hlc, floor: &RotationFloor) -> bool {
    if let Some(k) = kid {
        if k >= floor.floor_epoch {
            return true;
        }
    }
    *hlc <= floor.rotation_hlc
}

/// Rotate `org` to `new_epoch`: wrap `k_org_new` for the REMAINING members ONLY
/// and publish each wrap at `new_epoch`. The removed member is simply absent from
/// `remaining`, so no wrap is ever authored for them — that omission, not any
/// revocation, is the whole mechanism.
///
/// `k_org_new` MUST be a fresh [`crate::crypto::generate_org_key`] draw
/// independent of the previous epoch (see that fn's contract). This function does
/// not generate it, so callers can rotate deterministically in tests, but the
/// production caller passes a CSPRNG key. The freshness invariant rests on caller
/// discipline + the deliberate ABSENCE of any derive-from-old API — it is NOT
/// checked here.
///
/// CALLER OBLIGATIONS (unchecked here — a trusted granter is assumed):
/// - **`new_epoch` monotonicity.** `new_epoch` MUST strictly exceed every epoch
///   already published for the org. It is NOT validated; publishing at a
///   colliding epoch LWW-overwrites live wraps with a different key and breaks
///   existing members.
/// - **Non-atomic + retry MUST reuse the SAME `k_org_new`.** This loop publishes
///   per member with no rollback; a mid-loop `publish_wrapped` failure leaves
///   some members granted and some not. On retry the caller MUST pass the
///   IDENTICAL `k_org_new` — calling `generate_org_key` again would grant a
///   DIFFERENT key at the same epoch, so `resolve_all_org_roots` would hand
///   incompatible bytes for one epoch to different members (split brain: ops one
///   authors are undecryptable to another). Reuse the key; only bump the epoch to
///   start a genuinely new generation.
///
/// This publishes wraps but does NOT produce or distribute the [`RotationFloor`];
/// the caller signs one with [`sign_rotation_floor`] and distributes it (its
/// directory/transport home lands with real provisioning in a later slice).
pub fn rotate_org_key(
    directory: &mut dyn OrgKeyDirectory,
    org: &str,
    new_epoch: u64,
    remaining: &[(&str, &PublicKey)],
    k_org_new: &[u8; 32],
    granter_user_id: &str,
    granter_signer: &SigningKey,
) -> Result<(), RotationError> {
    for (recipient_id, recipient_pub) in remaining {
        let wrapped = wrap_org_key(
            k_org_new,
            org,
            new_epoch,
            recipient_id,
            recipient_pub,
            granter_user_id,
            granter_signer,
        )?;
        directory.publish_wrapped(&wrapped)?;
    }
    Ok(())
}

/// The per-member outcome of [`provision_org_members`]. Best-effort: `granted`
/// are the account ids that received a wrap this round; `skipped` pairs every
/// other account id with WHY it got none (malformed pubkey, a low-order /
/// non-contributory key that `wrap_org_key` rejects, or a publish failure). A
/// non-empty `skipped` is NOT a batch abort — the caller reports both.
#[derive(Debug, Default, Clone)]
pub struct ProvisionReport {
    pub granted: Vec<String>,
    pub skipped: Vec<(String, String)>,
}

/// Wrap `k_org` at `epoch` for every member in `members` and publish each wrap —
/// the granter's provisioning entry point over the DIRECTORY-NATIVE
/// [`MemberPublicKey`] shape ([`OrgKeyDirectory::fetch_pubkeys`] output), so a
/// granter tool never has to touch X25519 types.
///
/// BEST-EFFORT (returns a [`ProvisionReport`], never aborts the batch): a member
/// whose pubkey is malformed hex, whose key is low-order (`wrap_org_key`'s
/// contributory check rejects it — an unauthenticated directory lets anyone
/// publish such a point, so a batch-abort here would be a cheap DoS), or whose
/// wrap fails to publish is recorded in `skipped` with a reason; the rest proceed.
///
/// CALLER OBLIGATIONS — this carries the SAME non-atomic re-key hazard as
/// [`rotate_org_key`]. Publishing is per-member with no rollback, and
/// `publish_wrapped` is LWW keyed on `(epoch, recipient)`. So:
/// - If `epoch` ALREADY has grants, calling this with a FRESH `k_org` RE-KEYS
///   every member at that epoch (LWW overwrite), orphaning any data written under
///   the previous key. There is no re-wrap-the-existing-key path here — to add a
///   member at a populated epoch you must deliberately re-key everyone. A caller
///   that must not silently re-key should check for existing grants at `epoch`
///   first (see `examples/org_granter.rs`).
/// - A retry after a partial publish MUST reuse the SAME `k_org`; a fresh one
///   splits the epoch into incompatible keys.
/// - For a clean rotation, mint a fresh key AND bump `epoch` monotonically.
pub fn provision_org_members(
    directory: &mut dyn OrgKeyDirectory,
    org: &str,
    epoch: u64,
    k_org: &[u8; 32],
    members: &[MemberPublicKey],
    granter_user_id: &str,
    granter_signer: &SigningKey,
) -> ProvisionReport {
    let mut report = ProvisionReport::default();
    for m in members {
        // Every failure mode is a per-member SKIP-with-reason, never a batch abort
        // — one bad (or maliciously poisoned) entry must not block the others.
        let recipient_pub = match parse_x25519_pub(&m.public_hex) {
            Ok(p) => p,
            Err(e) => {
                report
                    .skipped
                    .push((m.account_id.clone(), format!("malformed pubkey: {e}")));
                continue;
            }
        };
        let wrapped = match wrap_org_key(
            k_org,
            org,
            epoch,
            &m.account_id,
            &recipient_pub,
            granter_user_id,
            granter_signer,
        ) {
            Ok(w) => w,
            Err(e) => {
                report
                    .skipped
                    .push((m.account_id.clone(), format!("wrap failed: {e}")));
                continue;
            }
        };
        match directory.publish_wrapped(&wrapped) {
            Ok(()) => report.granted.push(m.account_id.clone()),
            Err(e) => report
                .skipped
                .push((m.account_id.clone(), format!("publish failed: {e}"))),
        }
    }
    report
}

/// Resolve EVERY epoch of `org`'s `K_org` a member can currently unwrap — the
/// plural of [`crate::resolve_org_root`]. This is what populates the multi-epoch
/// map [`crate::org_key_provider::OrgAwareKeyProvider`] already accepts: after a
/// rotation a remaining member holds `{N, N+1}` and reads both generations.
///
/// A removed member, having no wrap at the new epoch, resolves only the OLD
/// epoch(s) — exactly the fail-closed outcome rotation intends. Every candidate
/// is verify-then-decrypt against `trusted` (via [`unwrap_org_key`]); untrusted
/// or wrong-org wraps are skipped, never adopted.
///
/// Fail-closed return contract mirrors [`crate::resolve_org_root`]: `Err` means
/// the directory was UNREACHABLE (caller fails closed to `DenyCipher`); `Ok` with
/// an EMPTY map means reachable-but-ungranted (also `DenyCipher`, benign).
pub fn resolve_all_org_roots(
    directory: &dyn OrgKeyDirectory,
    org: &str,
    my_secret: &StaticSecret,
    my_user_id: &str,
    trusted: &[VerifyingKey],
) -> Result<std::collections::BTreeMap<u64, Zeroizing<[u8; 32]>>, OrgKeyDirectoryError> {
    let mut roots = std::collections::BTreeMap::new();
    // Err propagates → caller fails closed. Do NOT map to an empty map.
    let candidates: Vec<WrappedOrgKey> = directory.fetch_wrapped_for(my_user_id)?;
    for w in candidates {
        // Defense-in-depth: a trusted granter's wrap for a DIFFERENT org would
        // verify+decrypt cleanly to that org's key; refuse to adopt it here.
        if w.org != org {
            continue;
        }
        // Forward-looking defense: the current directory keys wraps by
        // (recipient, epoch) so it returns at most ONE wrap per epoch — this guard
        // is inert today. If a future transport can return multiple wraps at one
        // epoch, gating on SUCCESSFUL insertion (not iteration) is the safe order:
        // a leading UNTRUSTED wrap fails unwrap, leaves the epoch unheld, so a
        // following TRUSTED wrap at the same epoch is still tried and wins (an
        // untrusted wrap can never shadow the real one). Adopting a genuinely wrong
        // key would need two DIFFERENT trusted-signed keys at one epoch — a granter
        // fault, the named audit assumption.
        if roots.contains_key(&w.epoch) {
            continue;
        }
        if let Ok(k) = unwrap_org_key(&w, my_secret, my_user_id, trusted) {
            roots.insert(w.epoch, Zeroizing::new(k));
        }
    }
    Ok(roots)
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::crypto::{ed25519_verifying, generate_org_key, x25519_public, StretchedMaster};
    use crate::org_key_directory::InMemoryOrgKeyDirectory;

    // Fast test identity helpers (issued-high-entropy skips Argon2id).
    fn x25519_id(secret: &[u8], user: &str) -> StaticSecret {
        crate::crypto::derive_x25519_identity(
            &StretchedMaster::from_issued_high_entropy(secret, user),
            user,
        )
    }
    fn ed25519_id(secret: &[u8], user: &str) -> SigningKey {
        crate::crypto::derive_ed25519_identity(
            &StretchedMaster::from_issued_high_entropy(secret, user),
            user,
        )
    }
    fn granter() -> SigningKey {
        ed25519_id(b"granter-login", "acc_granter")
    }

    fn hlc(wall_ms: u64, counter: u32) -> Hlc {
        Hlc {
            wall_ms,
            counter,
            device_id: "dev-a".into(),
        }
    }

    fn hex_of(pk: &x25519_dalek::PublicKey) -> String {
        pk.as_bytes().iter().map(|b| format!("{b:02x}")).collect()
    }

    // ---- rotation orchestration + resolve_all_org_roots ----

    #[test]
    fn rotation_grants_remaining_members_and_omits_the_removed() {
        // Alice + Bob at epoch 1; Carol is removed on the rotation to epoch 2.
        let mut dir = InMemoryOrgKeyDirectory::new();
        let g = granter();
        let alice = x25519_id(b"alice", "acc_alice");
        let bob = x25519_id(b"bob", "acc_bob");
        let carol = x25519_id(b"carol", "acc_carol");
        let trusted = [ed25519_verifying(&g)];

        // Initial grant at epoch 1 for all three.
        let k1 = [1u8; 32];
        for (id, sk) in [
            ("acc_alice", &alice),
            ("acc_bob", &bob),
            ("acc_carol", &carol),
        ] {
            let w =
                wrap_org_key(&k1, "acme", 1, id, &x25519_public(sk), "acc_granter", &g).unwrap();
            dir.publish_wrapped(&w).unwrap();
        }

        // Rotate to epoch 2 for the REMAINING members only (Carol excluded).
        let k2 = generate_org_key();
        assert_ne!(*k2, k1, "new epoch key is independent of the old");
        rotate_org_key(
            &mut dir,
            "acme",
            2,
            &[
                ("acc_alice", &x25519_public(&alice)),
                ("acc_bob", &x25519_public(&bob)),
            ],
            &k2,
            "acc_granter",
            &g,
        )
        .unwrap();

        // Remaining members hold BOTH epochs...
        let a_roots = resolve_all_org_roots(&dir, "acme", &alice, "acc_alice", &trusted).unwrap();
        assert_eq!(a_roots.keys().copied().collect::<Vec<_>>(), vec![1, 2]);
        assert_eq!(*a_roots[&1], k1);
        assert_eq!(*a_roots[&2], *k2);

        // ...the removed member holds ONLY the old epoch (fail-closed on new ops).
        let c_roots = resolve_all_org_roots(&dir, "acme", &carol, "acc_carol", &trusted).unwrap();
        assert_eq!(c_roots.keys().copied().collect::<Vec<_>>(), vec![1]);
        assert!(
            !c_roots.contains_key(&2),
            "removed member gets no new epoch"
        );
    }

    #[test]
    fn provision_over_member_pubkeys_grants_and_skips_malformed() {
        // The granter-tool entry point: wrap for the directory-native MemberPublicKey
        // set. Well-formed members become resolvable; a malformed pubkey is skipped
        // (reported via the return list), never a batch abort.
        let mut dir = InMemoryOrgKeyDirectory::new();
        let g = granter();
        let trusted = [ed25519_verifying(&g)];
        let alice = x25519_id(b"alice", "acc_alice");
        let bob = x25519_id(b"bob", "acc_bob");
        let members = vec![
            MemberPublicKey {
                account_id: "acc_alice".into(),
                public_hex: hex_of(&x25519_public(&alice)),
            },
            MemberPublicKey {
                account_id: "acc_bob".into(),
                public_hex: hex_of(&x25519_public(&bob)),
            },
            MemberPublicKey {
                account_id: "acc_broken".into(),
                public_hex: "not-hex".into(),
            },
            MemberPublicKey {
                // Hex-valid but LOW-ORDER (all zeros) — parses, then wrap_org_key's
                // contributory check rejects it. Must SKIP-with-reason, not abort
                // the batch (else a poisoned pubkey bricks every grant round).
                account_id: "acc_loworder".into(),
                public_hex: "0".repeat(64),
            },
        ];

        let k = [3u8; 32];
        let report = provision_org_members(&mut dir, "acme", 1, &k, &members, "acc_granter", &g);
        assert_eq!(
            report.granted,
            vec!["acc_alice", "acc_bob"],
            "well-formed granted"
        );
        // Both the malformed-hex AND the low-order member skip-with-reason — neither
        // aborts the batch, so the two good members are still granted.
        let skipped_ids: Vec<&str> = report.skipped.iter().map(|(id, _)| id.as_str()).collect();
        assert_eq!(skipped_ids, vec!["acc_broken", "acc_loworder"]);
        assert!(report.skipped.iter().all(|(_, why)| !why.is_empty()));

        // Both well-formed members resolve the granted key; the broken one got none.
        let a = resolve_all_org_roots(&dir, "acme", &alice, "acc_alice", &trusted).unwrap();
        assert_eq!(*a[&1], k);
        let broken = x25519_id(b"broken", "acc_broken");
        assert!(
            resolve_all_org_roots(&dir, "acme", &broken, "acc_broken", &trusted)
                .unwrap()
                .is_empty()
        );
    }

    #[test]
    fn generate_org_key_is_independent_across_draws() {
        // Two draws must differ (CSPRNG). This is the crude stand-in for
        // "N+1 not derivable from N" — the API simply offers no derive-from-old path.
        assert_ne!(*generate_org_key(), *generate_org_key());
    }

    #[test]
    fn resolve_skips_untrusted_higher_epoch_but_keeps_real_one() {
        // A bogus higher-epoch wrap from an UNTRUSTED signer must not be adopted;
        // the genuine epoch still resolves (substitution → no; DoS → possible/named).
        let mut dir = InMemoryOrgKeyDirectory::new();
        let g = granter();
        let mallory = ed25519_id(b"mallory", "acc_mallory");
        let alice = x25519_id(b"alice", "acc_alice");
        let trusted = [ed25519_verifying(&g)];

        let real = wrap_org_key(
            &[7u8; 32],
            "acme",
            1,
            "acc_alice",
            &x25519_public(&alice),
            "acc_granter",
            &g,
        )
        .unwrap();
        dir.publish_wrapped(&real).unwrap();
        // Mallory publishes a "newer" epoch-9 wrap of a bogus key for Alice.
        let bogus = wrap_org_key(
            &[9u8; 32],
            "acme",
            9,
            "acc_alice",
            &x25519_public(&alice),
            "acc_mallory",
            &mallory,
        )
        .unwrap();
        dir.publish_wrapped(&bogus).unwrap();

        let roots = resolve_all_org_roots(&dir, "acme", &alice, "acc_alice", &trusted).unwrap();
        assert_eq!(roots.keys().copied().collect::<Vec<_>>(), vec![1]);
        assert_eq!(*roots[&1], [7u8; 32]);
        assert!(
            !roots.contains_key(&9),
            "untrusted higher epoch is skipped, not adopted"
        );
    }

    #[test]
    fn resolve_all_wrong_org_wrap_is_skipped() {
        let mut dir = InMemoryOrgKeyDirectory::new();
        let g = granter();
        let alice = x25519_id(b"alice", "acc_alice");
        let trusted = [ed25519_verifying(&g)];
        // A trusted granter's wrap, but for "other" — must not be adopted as acme's.
        let w = wrap_org_key(
            &[5u8; 32],
            "other",
            1,
            "acc_alice",
            &x25519_public(&alice),
            "acc_granter",
            &g,
        )
        .unwrap();
        dir.publish_wrapped(&w).unwrap();
        let roots = resolve_all_org_roots(&dir, "acme", &alice, "acc_alice", &trusted).unwrap();
        assert!(roots.is_empty(), "wrong-org wrap skipped");
    }

    // ---- rotation floor sign / verify ----

    #[test]
    fn floor_signed_by_trusted_granter_verifies() {
        let g = granter();
        let floor = sign_rotation_floor("acme", 2, hlc(100, 0), 1, &g);
        assert_eq!(
            verify_rotation_floor(&floor, &[ed25519_verifying(&g)]).unwrap(),
            2
        );
    }

    #[test]
    fn floor_from_untrusted_signer_is_rejected() {
        let mallory = ed25519_id(b"mallory", "acc_mallory");
        let floor = sign_rotation_floor("acme", 2, hlc(100, 0), 1, &mallory);
        // Verified against the granter's key, not Mallory's → Untrusted.
        assert!(matches!(
            verify_rotation_floor(&floor, &[ed25519_verifying(&granter())]),
            Err(RotationError::Untrusted)
        ));
    }

    #[test]
    fn non_monotonic_floor_is_rejected_even_if_signed() {
        let g = granter();
        // floor_epoch == prev_floor_epoch → not strictly newer → rejected.
        let floor = sign_rotation_floor("acme", 2, hlc(100, 0), 2, &g);
        assert!(matches!(
            verify_rotation_floor(&floor, &[ed25519_verifying(&g)]),
            Err(RotationError::NonMonotonic)
        ));
    }

    #[test]
    fn tampered_floor_field_fails_verification() {
        let g = granter();
        let mut floor = sign_rotation_floor("acme", 5, hlc(100, 0), 1, &g);
        // Flip the epoch after signing → transcript mismatch → Untrusted.
        floor.floor_epoch = 6;
        assert!(matches!(
            verify_rotation_floor(&floor, &[ed25519_verifying(&g)]),
            Err(RotationError::Untrusted)
        ));
    }

    #[test]
    fn floor_with_wrong_algorithm_tag_is_rejected() {
        let g = granter();
        let mut floor = sign_rotation_floor("acme", 2, hlc(100, 0), 1, &g);
        // A foreign/garbage tag is rejected before the signature is even checked
        // (mirrors unwrap_org_key's car_wrap gate).
        floor.car_floor = "org-key-wrap/v2".into();
        assert!(matches!(
            verify_rotation_floor(&floor, &[ed25519_verifying(&g)]),
            Err(RotationError::WrongAlgorithm(_))
        ));
    }

    #[test]
    fn older_valid_floor_still_verifies_anti_rollback_is_the_callers_job() {
        // THREAT-MODEL CASE (documents C1): verify_rotation_floor is stateless, so
        // an OLDER but legitimately-signed floor (2,1) still verifies Ok(2) even
        // after the org advanced to (5,4). This function does NOT prevent that — the
        // caller must persist the max accepted floor_epoch and reject <= it. If this
        // ever starts erroring, verify gained rollback state — revisit the doc.
        let g = granter();
        let trusted = [ed25519_verifying(&g)];
        let advanced = sign_rotation_floor("acme", 5, hlc(500, 0), 4, &g);
        let stale = sign_rotation_floor("acme", 2, hlc(200, 0), 1, &g);
        assert_eq!(verify_rotation_floor(&advanced, &trusted).unwrap(), 5);
        assert_eq!(
            verify_rotation_floor(&stale, &trusted).unwrap(),
            2,
            "stale-but-valid floor verifies; rollback defense is external"
        );
    }

    // ---- is_authoritative freshness predicate ----

    #[test]
    fn authoritative_accepts_new_epoch_regardless_of_time() {
        let g = granter();
        let floor = sign_rotation_floor("acme", 2, hlc(100, 0), 1, &g);
        // Post-cut epoch-2 op, even far in the future → authoritative (unforgeable).
        assert!(is_authoritative(Some(2), &hlc(9_999, 0), &floor));
        assert!(is_authoritative(Some(3), &hlc(9_999, 0), &floor));
    }

    #[test]
    fn authoritative_keeps_precut_history_readable() {
        let g = granter();
        let floor = sign_rotation_floor("acme", 2, hlc(100, 0), 1, &g);
        // Old epoch-1 op authored BEFORE the cut → still authoritative (non-lossy).
        assert!(is_authoritative(Some(1), &hlc(50, 0), &floor));
        // Exactly at the cut boundary counts as pre-cut.
        assert!(is_authoritative(Some(1), &hlc(100, 0), &floor));
    }

    #[test]
    fn authoritative_rejects_postcut_stale_epoch_write() {
        let g = granter();
        let floor = sign_rotation_floor("acme", 2, hlc(100, 0), 1, &g);
        // A removed member writing epoch-1 AFTER the cut → NOT authoritative.
        assert!(!is_authoritative(Some(1), &hlc(101, 0), &floor));
        // A personal-shaped (no-kid) op after the cut is likewise non-authoritative.
        assert!(!is_authoritative(None, &hlc(101, 0), &floor));
    }

    #[test]
    fn malicious_backdate_is_a_named_limitation_not_a_fix() {
        // THREAT-MODEL CASE (documents the gap, does not assert it closed): a
        // removed member holding K_org@1 can BACKDATE an op to hlc <= rotation_hlc
        // and is_authoritative() will accept it. This is tolerated ONLY because such
        // an op is confined to epoch 1 (already readable to them) and LOSES LWW to
        // any genuine epoch-2 op. Fully refusing it needs per-op author signatures /
        // admission authz (audit-deferred). If this assertion ever flips, the gate
        // was strengthened — revisit the module audit note.
        let g = granter();
        let floor = sign_rotation_floor("acme", 2, hlc(100, 0), 1, &g);
        let backdated = hlc(50, 0); // attacker chooses a pre-cut timestamp
        assert!(
            is_authoritative(Some(1), &backdated, &floor),
            "backdated forgery slips the HLC half — known, audit-deferred limitation"
        );
    }
}