car-sync 0.50.0

//! `OrgAwareKeyProvider` — the [`SyncKeyProvider`] that encrypts org-scoped ops
//! under the SHARED org key `K_org` (mutually readable across an org's members)
//! while personal ops keep the per-user key. This is the activation core of the
//! org shared brain: swapping it in is what makes `Scope::Shared { org }` ops
//! readable by every member instead of only their author.
//!
//! ## How it works (and why the wire form doesn't change)
//!
//! For `Scope::Personal` it delegates to an inner per-user provider (e.g.
//! [`crate::crypto::DerivedKeyProvider`]). For `Scope::Shared { org }` it takes
//! the org's `K_org` root and derives the audience key through the SAME HKDF the
//! per-user path uses — [`derive_key`]`(K_org, "org:<org>")` → [`LocalKeyCipher`].
//! The only thing that differs from the per-user provider is the *master*: a
//! shared `K_org` instead of a per-user secret. Same audience string, same
//! [`crate::crypto::Envelope`] shape — so the ciphertext becomes mutually
//! readable without any wire change. That is exactly the bug being fixed: today
//! two members derive DIFFERENT org keys from DIFFERENT per-user masters and
//! cannot read each other; a shared `K_org` master makes them converge.
//!
//! ## No directory, no identity secret on the hot path (option C)
//!
//! [`SyncKeyProvider::cipher_for`] is called per-op on both the write path
//! (`SyncSession::append`) and the read path (the fold's `decrypted_tail`). So
//! this provider holds ONLY pre-resolved `K_org` roots — no
//! [`crate::org_key_directory::OrgKeyDirectory`] and no X25519 identity secret.
//! Fetching wraps, unwrapping with the identity secret, and picking the newest
//! epoch all happen OUT OF BAND (a builder that runs at open time, in the
//! post-audit activation slice), never inside `cipher_for`. HKDF for a given org
//! runs once and is cached, so the per-op cost is a map lookup.
//!
//! ## Fail-closed
//!
//! A member with no `K_org` for an org (not yet granted the key, or org-scope not
//! activated) gets a [`DenyCipher`] for that scope — encrypt AND decrypt error.
//! Never a personal cipher, never a different org's cipher, never a freshly
//! minted key. On write that rolls the op back (a non-member cannot author an org
//! op); on read the ciphertext stays an opaque blob (fail-closed, never
//! plaintext, never wrong-key plaintext).
//!
//! ## Single-newest epoch (structural limit)
//!
//! An [`crate::oplog::OpRecord`]'s envelope carries no key-id/epoch, and
//! `cipher_for` sees only `&Scope`, so per-op multi-epoch key selection is
//! UNEXPRESSIBLE through this trait today. This slice therefore resolves ONE
//! `K_org` per org (the newest the member can unwrap): correct for writes (always
//! the current key) and inert for reads (nothing rotates yet). Multi-epoch
//! decryption is a future ADDITIVE change (an authenticated key-id on the
//! envelope) — do NOT trial-decrypt across a keyring as a stand-in (guessing keys
//! until an AEAD opens is a wrong-key-acceptance surface). The internal map is
//! keyed by org so it can grow to `(org, epoch)` without a public-API change.
//!
//! ## Migration seam (for the activation slice)
//!
//! Because the audience string is identical in both providers, the envelope
//! carries NO signal of which master produced it. Org ops written under the
//! OLD per-user-derived org key will not open under `K_org` after activation.
//! That is a one-time migration concern for whoever flips the switch — named
//! here, not silently ignored.
//!
//! ## INERT — no wiring, no flag
//!
//! Org-scope E2E must not be enabled in prod before a cryptographer audit (the
//! `login_secret → derive_x25519_identity` entropy dependency). So this type is
//! BUILT, exported, and unit-tested, but referenced from ZERO call sites in the
//! sync-subsystem construction. Inertness is grep-provable: search the type name
//! and only tests answer. A default-OFF flag is deliberately NOT used — a flag is
//! still an activation path (env drift, a copied config), which is precisely what
//! the audit exists to gate. Absence is the proof a flag cannot give.

use std::collections::{BTreeMap, HashMap};
use std::sync::{Arc, Mutex};

use serde_json::Value;
use zeroize::Zeroizing;

use crate::crypto::{
    derive_key, encryption_audience, CryptoError, MultiEpochOrgCipher, PayloadCipher,
    SyncKeyProvider,
};
use crate::oplog::Scope;

/// A [`PayloadCipher`] that always fails — the fail-closed cipher for a scope
/// this provider has no key for. Both directions error, so it can never yield
/// plaintext or accept a wrong-key ciphertext.
struct DenyCipher {
    reason: String,
}

impl PayloadCipher for DenyCipher {
    fn encrypt(&self, _plaintext: &Value) -> Result<Value, CryptoError> {
        Err(CryptoError::Key(self.reason.clone()))
    }
    fn decrypt(&self, _envelope: &Value) -> Result<Value, CryptoError> {
        Err(CryptoError::Key(self.reason.clone()))
    }
}

/// A [`SyncKeyProvider`] that keys `Scope::Shared { org }` ops on a shared
/// `K_org` and delegates `Scope::Personal` to an inner per-user provider. See the
/// module docs for the hot-path, fail-closed, epoch, and inertness contracts.
pub struct OrgAwareKeyProvider {
    personal: Arc<dyn SyncKeyProvider>,
    /// `org → { epoch → K_org@epoch }` — every generation of the org master the
    /// member can currently unwrap, not just the newest. Holding the full set is
    /// what makes rotation NON-LOSSY: after a member-removal rotation bumps the
    /// epoch, remaining members still hold the older epochs and can decrypt ops
    /// authored before the bump (selected per-op by the envelope's `kid`). Roots
    /// arrive in [`Zeroizing`] and are wiped on drop.
    org_roots: HashMap<String, BTreeMap<u64, Zeroizing<[u8; 32]>>>,
    /// `audience → derived cipher` — HKDF for an org runs once, not per op.
    ciphers: Mutex<HashMap<String, Arc<dyn PayloadCipher>>>,
}

impl std::fmt::Debug for OrgAwareKeyProvider {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        // Never print key material; show only which orgs are resolved.
        f.debug_struct("OrgAwareKeyProvider")
            .field("orgs", &self.org_roots.keys().collect::<Vec<_>>())
            .finish_non_exhaustive()
    }
}

impl OrgAwareKeyProvider {
    /// Build over an inner per-user `personal` provider and a set of
    /// pre-resolved `org → K_org` roots (each the newest epoch the member can
    /// unwrap). Resolution — fetch wraps, `unwrap_org_key`, pick newest — happens
    /// OUT OF BAND before construction; this type never touches the directory or
    /// the identity secret.
    ///
    /// Roots arrive already in [`Zeroizing`] so the root secret is never a bare
    /// `Copy` array in a heap map on the way in: a `HashMap<_, [u8;32]>` would be
    /// freed WITHOUT wiping (`[u8;32]: Copy` → wrapping copies and drops the
    /// original un-zeroized). The out-of-band builder must wrap each
    /// `unwrap_org_key` result at the source.
    pub fn new(
        personal: Arc<dyn SyncKeyProvider>,
        org_roots: HashMap<String, BTreeMap<u64, Zeroizing<[u8; 32]>>>,
    ) -> Self {
        Self {
            personal,
            org_roots,
            ciphers: Mutex::new(HashMap::new()),
        }
    }

    fn org_cipher(&self, org: &str, audience: &str) -> Arc<dyn PayloadCipher> {
        let mut cache = self.ciphers.lock().expect("org cipher cache poisoned");
        if let Some(c) = cache.get(audience) {
            return c.clone();
        }
        let cipher: Arc<dyn PayloadCipher> = match self.org_roots.get(org) {
            // Derive the org-audience AEAD key for EACH held epoch through the SAME
            // HKDF the per-user path uses (only the master differs), and hand the
            // per-epoch keyring to a cipher that selects one by the envelope's kid.
            // An empty map (org known but no epoch keys) is treated as no key.
            Some(epochs) if !epochs.is_empty() => {
                let keys: BTreeMap<u64, Zeroizing<[u8; 32]>> = epochs
                    .iter()
                    .map(|(&epoch, k_org)| (epoch, Zeroizing::new(derive_key(&k_org[..], audience))))
                    .collect();
                Arc::new(MultiEpochOrgCipher::new(audience, keys))
            }
            // fail closed: no key for this org → a cipher that errors both ways.
            _ => Arc::new(DenyCipher {
                reason: format!(
                    "no org key for {audience}: member not granted K_org, or org-scope not activated"
                ),
            }),
        };
        cache.insert(audience.to_string(), cipher.clone());
        cipher
    }
}

impl SyncKeyProvider for OrgAwareKeyProvider {
    fn cipher_for(&self, scope: &Scope) -> Arc<dyn PayloadCipher> {
        match scope {
            Scope::Personal => self.personal.cipher_for(scope),
            Scope::Shared { org } => self.org_cipher(org, &encryption_audience(scope)),
        }
    }
}

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

    fn personal(user: &str) -> Arc<dyn SyncKeyProvider> {
        // from_login_secret (issued path, no Argon2id) keeps these provider tests
        // fast — the passphrase stretch is covered by a dedicated crypto test.
        Arc::new(DerivedKeyProvider::from_login_secret(b"pass", user))
    }

    // Single-epoch (epoch 1) convenience — most tests don't care about rotation.
    fn provider(user: &str, roots: &[(&str, [u8; 32])]) -> OrgAwareKeyProvider {
        OrgAwareKeyProvider::new(
            personal(user),
            roots
                .iter()
                .map(|(o, k)| {
                    let m = BTreeMap::from([(1u64, Zeroizing::new(*k))]);
                    (o.to_string(), m)
                })
                .collect(),
        )
    }

    // Multi-epoch: each org carries a set of `(epoch, K_org)` roots.
    fn provider_multi(user: &str, roots: &[(&str, &[(u64, [u8; 32])])]) -> OrgAwareKeyProvider {
        OrgAwareKeyProvider::new(
            personal(user),
            roots
                .iter()
                .map(|(o, epochs)| {
                    let m: BTreeMap<u64, Zeroizing<[u8; 32]>> = epochs
                        .iter()
                        .map(|(e, k)| (*e, Zeroizing::new(*k)))
                        .collect();
                    (o.to_string(), m)
                })
                .collect(),
        )
    }

    #[test]
    fn personal_scope_round_trips_via_the_inner_provider() {
        let p = provider("alice", &[]);
        let msg = serde_json::json!({"note": "personal"});
        let ct = p.cipher_for(&Scope::Personal).encrypt(&msg).unwrap();
        let pt = p.cipher_for(&Scope::Personal).decrypt(&ct).unwrap();
        assert_eq!(pt, msg);
    }

    #[test]
    fn shared_org_ops_are_mutually_readable_across_members() {
        // THE point of the feature: two DIFFERENT members (different personal
        // masters) holding the SAME K_org read each other's org-scoped ops.
        let k_org = [9u8; 32];
        let alice = provider("alice", &[("acme", k_org)]);
        let bob = provider("bob", &[("acme", k_org)]);
        let scope = Scope::Shared { org: "acme".into() };

        let msg = serde_json::json!({"shared": "brain"});
        let ct = alice.cipher_for(&scope).encrypt(&msg).unwrap();
        // Bob, a different member, recovers Alice's org op.
        assert_eq!(bob.cipher_for(&scope).decrypt(&ct).unwrap(), msg);
    }

    #[test]
    fn shared_org_without_key_fails_closed_both_ways() {
        // A member not granted K_org for the org: encrypt AND decrypt error —
        // never plaintext, never a wrong key.
        let p = provider("alice", &[]);
        let scope = Scope::Shared { org: "acme".into() };
        assert!(matches!(
            p.cipher_for(&scope).encrypt(&serde_json::json!({"x": 1})),
            Err(CryptoError::Key(_))
        ));
        assert!(matches!(
            p.cipher_for(&scope)
                .decrypt(&serde_json::json!({"car_enc": "x"})),
            Err(CryptoError::Key(_))
        ));
    }

    #[test]
    fn distinct_orgs_derive_independent_keys() {
        let acme = provider("alice", &[("acme", [1u8; 32])]);
        let globex = provider("alice", &[("globex", [2u8; 32])]);
        let msg = serde_json::json!({"secret": "acme-only"});
        let ct = acme
            .cipher_for(&Scope::Shared { org: "acme".into() })
            .encrypt(&msg)
            .unwrap();
        // A different org's key cannot open it (wrong key → Decrypt error).
        assert!(matches!(
            globex
                .cipher_for(&Scope::Shared {
                    org: "globex".into()
                })
                .decrypt(&ct),
            Err(CryptoError::Decrypt)
        ));
    }

    #[test]
    fn personal_and_org_audiences_are_independent() {
        let p = provider("alice", &[("acme", [7u8; 32])]);
        let msg = serde_json::json!({"m": 1});
        let org_ct = p
            .cipher_for(&Scope::Shared { org: "acme".into() })
            .encrypt(&msg)
            .unwrap();
        // The personal cipher must not open an org ciphertext.
        assert!(matches!(
            p.cipher_for(&Scope::Personal).decrypt(&org_ct),
            Err(CryptoError::Decrypt)
        ));
    }

    #[test]
    fn org_cipher_is_cached_per_audience() {
        let p = provider("alice", &[("acme", [3u8; 32])]);
        let scope = Scope::Shared { org: "acme".into() };
        let a = p.cipher_for(&scope);
        let b = p.cipher_for(&scope);
        assert!(
            Arc::ptr_eq(&a, &b),
            "HKDF should run once per org, not per op"
        );
    }

    #[test]
    fn debug_never_prints_key_material() {
        let p = provider("alice", &[("acme", [0xabu8; 32])]);
        let dbg = format!("{p:?}");
        assert!(dbg.contains("acme"), "shows which orgs are resolved");
        assert!(!dbg.contains("abab"), "must not leak key bytes");
    }

    #[test]
    fn encrypt_uses_newest_epoch_and_older_ops_still_decrypt() {
        // Rotation is non-lossy: a member holding {1, 2} encrypts under the NEWEST
        // epoch (2), yet can still decrypt an op authored under the OLD epoch (1) —
        // the kid on each envelope selects the right key.
        let scope = Scope::Shared { org: "acme".into() };
        let old_only = provider_multi("alice", &[("acme", &[(1, [1u8; 32])])]);
        let both = provider_multi("alice", &[("acme", &[(1, [1u8; 32]), (2, [2u8; 32])])]);

        // Op authored while only epoch 1 existed.
        let old_msg = serde_json::json!({"gen": 1});
        let old_ct = old_only.cipher_for(&scope).encrypt(&old_msg).unwrap();

        // A post-rotation member still holds epoch 1 → reads the old op...
        assert_eq!(both.cipher_for(&scope).decrypt(&old_ct).unwrap(), old_msg);

        // ...and NEW ops are stamped with epoch 2.
        let new_ct = both
            .cipher_for(&scope)
            .encrypt(&serde_json::json!({"gen": 2}))
            .unwrap();
        assert_eq!(new_ct.get("kid").and_then(|v| v.as_u64()), Some(2));
    }

    #[test]
    fn dropped_epoch_can_no_longer_decrypt_its_ops() {
        // The removal case: an op authored under epoch 1, then the member's keyring
        // rotates to hold ONLY epoch 2 (epoch-1 key withdrawn). The old op is now
        // opaque — kid=1 is not held → fail closed, never a wrong-key open.
        let scope = Scope::Shared { org: "acme".into() };
        let e1 = provider_multi("alice", &[("acme", &[(1, [1u8; 32])])]);
        let old_ct = e1
            .cipher_for(&scope)
            .encrypt(&serde_json::json!({"gen": 1}))
            .unwrap();

        let e2_only = provider_multi("alice", &[("acme", &[(2, [2u8; 32])])]);
        assert!(matches!(
            e2_only.cipher_for(&scope).decrypt(&old_ct),
            Err(CryptoError::Key(_))
        ));
    }

    #[test]
    fn org_envelope_missing_kid_fails_closed() {
        // A personal-shaped envelope (no kid) must NOT be accepted on the org path —
        // the org cipher requires an explicit epoch, never a default.
        let scope = Scope::Shared { org: "acme".into() };
        let p = provider("alice", &[("acme", [4u8; 32])]);
        let no_kid = serde_json::json!({
            "car_enc": "chacha20poly1305",
            "nonce": "000000000000000000000000",
            "ct": "00",
        });
        assert!(matches!(
            p.cipher_for(&scope).decrypt(&no_kid),
            Err(CryptoError::BadEnvelope(_))
        ));
    }

    #[test]
    fn org_envelopes_carry_a_kid_personal_ones_do_not() {
        // The wire distinction: org ciphertext stamps kid; personal stays kid-less
        // (byte-identical to pre-org-scope envelopes — see Envelope::kid docs).
        let p = provider("alice", &[("acme", [5u8; 32])]);
        let org_ct = p
            .cipher_for(&Scope::Shared { org: "acme".into() })
            .encrypt(&serde_json::json!({"x": 1}))
            .unwrap();
        assert!(org_ct.get("kid").is_some(), "org envelope stamps epoch");

        let personal_ct = p
            .cipher_for(&Scope::Personal)
            .encrypt(&serde_json::json!({"x": 1}))
            .unwrap();
        assert!(
            personal_ct.get("kid").is_none(),
            "personal envelope omits kid (wire-compatible)"
        );
    }
}