cachekit/

encryption.rs

//! Zero-knowledge encryption layer using AES-256-GCM with AAD v0x03 format.
//!
//! Wraps `cachekit_core::ZeroKnowledgeEncryptor` with tenant key derivation
//! and cache-key-bound Additional Authenticated Data (AAD). The AAD binding
//! prevents ciphertext substitution attacks within the same tenant (CVSS 8.5).
//!
//! # AAD v0x03 Format
//!
//! ```text
//! [version(0x03)][len(4)][tenant_id][len(4)][cache_key][len(4)][format][len(4)][compressed]
//! ```
//!
//! Each component is length-prefixed with a 4-byte big-endian u32 to prevent
//! collision attacks from boundary confusion.

use std::sync::atomic::{AtomicU64, Ordering};

use zeroize::Zeroizing;

use cachekit_core::{Keyring, ZeroKnowledgeEncryptor};

use crate::error::CachekitError;

/// AAD protocol version byte.
const AAD_VERSION: u8 = 0x03;

/// Required length of a raw-bytes master key (`spec/intent-presets.md`
/// § Master Key Input rule 4).
const RAW_KEY_LEN: usize = 32;

/// Zero-knowledge encryption layer with per-tenant key derivation.
///
/// Holds a derived encryption key (zeroized on drop), a [`Keyring`] of master
/// keys for multi-key decrypt during rotation, and the
/// `ZeroKnowledgeEncryptor` from cachekit-core for AES-256-GCM operations.
///
/// L1 stores **ciphertext**, not plaintext — the zero-knowledge property
/// is preserved across all cache layers.
///
/// # Key rotation
///
/// Writes always encrypt under the current master key. Reads decrypt via the
/// keyring: sequential attempts, current key first, then each decrypt-only
/// previous key in order, rebuilding the identical AAD per attempt
/// (cachekit-rs entries carry no per-entry key identity — sequential
/// attempts are the spec-assigned branch, `protocol/spec/encryption.md` →
/// "Key Rotation (Keyring)"). Reads served by a previous key are counted —
/// see [`Self::previous_key_hits`] for the rotation drain workflow.
pub struct EncryptionLayer {
    encryptor: ZeroKnowledgeEncryptor,
    derived_key: Zeroizing<[u8; 32]>,
    keyring: Keyring,
    tenant_id: String,
    /// `hits[i]` ↔ `previous_keys[i]`; see [`Self::previous_key_hits`].
    previous_key_hits: Vec<AtomicU64>,
}

impl EncryptionLayer {
    /// Create a new encryption layer with HKDF-derived tenant keys.
    ///
    /// Equivalent to [`Self::with_previous_keys`] with no previous keys.
    ///
    /// # Arguments
    /// * `master_key_bytes` — Raw master key, exactly 32 bytes
    /// * `tenant_id` — Tenant identifier for cryptographic isolation
    ///
    /// # Errors
    /// - Master key not exactly 32 bytes ([`CachekitError::Config`])
    /// - HKDF derivation failure
    /// - Encryptor initialization failure
    pub fn new(master_key_bytes: &[u8], tenant_id: &str) -> Result<Self, CachekitError> {
        Self::with_previous_keys(master_key_bytes, &[], tenant_id)
    }

    /// Create an encryption layer with decrypt-only previous master keys.
    ///
    /// `previous_keys` are retained during a key-rotation grace window, in
    /// attempt order. Writes always use `master_key_bytes`; reads attempt it
    /// first, then each previous key sequentially.
    ///
    /// Every key, current and previous, must be **exactly 32 raw bytes** —
    /// the only length every CacheKit SDK accepts. Pass the decoded key, never
    /// the ASCII bytes of a hex string (64 bytes, which would derive a key no
    /// other SDK derives); hex keys go through
    /// [`CacheKitBuilder::encryption_with_previous`](crate::client::CacheKitBuilder::encryption_with_previous),
    /// the hex counterpart for rotation.
    ///
    /// # Errors
    /// - Any key not exactly 32 bytes ([`CachekitError::Config`])
    /// - More than 3 previous keys ([`CachekitError::Config`] — rejected,
    ///   never truncated)
    /// - The current master key also present in `previous_keys`
    ///   ([`CachekitError::Config`] — rotation is forward-only)
    /// - HKDF derivation failure
    /// - Encryptor initialization failure
    pub fn with_previous_keys(
        master_key_bytes: &[u8],
        previous_keys: &[&[u8]],
        tenant_id: &str,
    ) -> Result<Self, CachekitError> {
        // Raw bytes carry no encoding to check, so length is the only guard
        // against a hex string passed as bytes. Hex paths are floored at
        // `>= 32` by `decode_master_key_hex`; `from_hex_decoded_keys`
        // re-checks that floor at the crypto layer for its internal callers.
        for (i, key) in previous_keys.iter().enumerate() {
            if key.len() != RAW_KEY_LEN {
                return Err(CachekitError::Config(format!(
                    "previous master key {i} must be exactly {RAW_KEY_LEN} bytes; got {}",
                    key.len()
                )));
            }
        }
        if master_key_bytes.len() != RAW_KEY_LEN {
            return Err(CachekitError::Config(format!(
                "master key must be exactly {RAW_KEY_LEN} bytes; got {} \
                 (pass hex-encoded keys to `.encryption()`, not their ASCII bytes)",
                master_key_bytes.len()
            )));
        }
        Self::from_hex_decoded_keys(master_key_bytes, previous_keys, tenant_id)
    }

    /// Build from hex-decoded keys, which may be longer than 32 bytes
    /// (`spec/intent-presets.md` § Master Key Input rule 3). Crate-private:
    /// only the hex paths reach it, after `decode_master_key_hex`.
    pub(crate) fn from_hex_decoded_keys(
        master_key_bytes: &[u8],
        previous_keys: &[&[u8]],
        tenant_id: &str,
    ) -> Result<Self, CachekitError> {
        for (i, key) in previous_keys.iter().enumerate() {
            if key.len() < 32 {
                return Err(CachekitError::Config(format!(
                    "previous master key {i} must be at least 32 bytes; got {}",
                    key.len()
                )));
            }
        }
        // Key-length and tenant checks are configuration errors, kept in the
        // same class as the previous-key and Keyring construction checks
        // below (config errors must not fold into crypto failures).
        if master_key_bytes.len() < 32 {
            return Err(CachekitError::Config(format!(
                "master key must be at least 32 bytes; got {}",
                master_key_bytes.len()
            )));
        }
        if tenant_id.is_empty() {
            return Err(CachekitError::Config(
                "tenant_id must not be empty".to_owned(),
            ));
        }
        if tenant_id.len() > 255 {
            return Err(CachekitError::Config(format!(
                "tenant_id must be at most 255 bytes; got {}",
                tenant_id.len()
            )));
        }

        let tenant_keys = cachekit_core::encryption::key_derivation::derive_tenant_keys(
            master_key_bytes,
            tenant_id,
        )
        .map_err(|e| CachekitError::Encryption(format!("key derivation failed: {e}")))?;

        let encryptor = ZeroKnowledgeEncryptor::new()
            .map_err(|e| CachekitError::Encryption(format!("encryptor init failed: {e}")))?;

        // The shared multi-key decrypt helper (cachekit-core) owns
        // the keyring invariants: cap of 3, forward-only self-collision,
        // sequential current-first attempts. Its construction failures are
        // configuration errors, not crypto failures.
        let keyring = Keyring::new(master_key_bytes, previous_keys)
            .map_err(|e| CachekitError::Config(format!("keyring: {e}")))?;

        Ok(Self {
            encryptor,
            derived_key: Zeroizing::new(tenant_keys.encryption_key),
            keyring,
            tenant_id: tenant_id.to_owned(),
            previous_key_hits: previous_keys.iter().map(|_| AtomicU64::new(0)).collect(),
        })
    }

    /// Encrypt plaintext with AAD bound to the cache key.
    ///
    /// Output format: `[nonce(12)][ciphertext + auth_tag(16)]`
    pub fn encrypt(&self, plaintext: &[u8], cache_key: &str) -> Result<Vec<u8>, CachekitError> {
        // compressed=false is normative, not a stub — see build_aad's invariant note.
        let aad = self.build_aad(cache_key, false);
        self.encryptor
            .encrypt_aes_gcm(plaintext, &*self.derived_key, &aad)
            .map_err(|e| CachekitError::Encryption(format!("encrypt failed: {e}")))
    }

    /// Decrypt ciphertext with AAD bound to the cache key.
    ///
    /// Returns the original plaintext. Fails if the cache key does not match
    /// the one used during encryption (ciphertext substitution protection).
    ///
    /// Decryption goes through the keyring: the current master key is
    /// attempted first, then each decrypt-only previous key in order, with
    /// the identical AAD per attempt. Entries written before a key rotation
    /// stay readable as long as their key remains in the previous list.
    ///
    /// A read served by a previous key is counted in
    /// [`Self::previous_key_hits`] (the rotation drain signal).
    pub fn decrypt(&self, ciphertext: &[u8], cache_key: &str) -> Result<Vec<u8>, CachekitError> {
        // compressed=false is normative, not a stub — see build_aad's invariant note.
        let aad = self.build_aad(cache_key, false);
        let (plaintext, index) = self
            .keyring
            .decrypt_indexed(&self.encryptor, ciphertext, &self.tenant_id, &aad)
            .map_err(|e| match e {
                // Config-class errors stay config-class:
                // a derivation failure or caller bug must never masquerade as
                // a decrypt failure that fail-open callers read as a miss.
                cachekit_core::EncryptionError::KeyDerivation(_)
                | cachekit_core::EncryptionError::KeyringIndexOutOfRange { .. } => {
                    CachekitError::Config(format!("keyring decrypt misconfiguration: {e}"))
                }
                _ => CachekitError::Encryption(format!("decrypt failed: {e}")),
            })?;
        // index 0 = current key (no signal); i >= 1 = previous_keys[i - 1].
        // The keyring was built from the same slice, so the slot always
        // exists; `get` keeps a core bug from panicking the read path.
        debug_assert!(
            index <= self.previous_key_hits.len(),
            "keyring index {index} beyond previous-key slots"
        );
        if let Some(hits) = index
            .checked_sub(1)
            .and_then(|i| self.previous_key_hits.get(i))
        {
            hits.fetch_add(1, Ordering::Relaxed);
        }
        Ok(plaintext)
    }

    /// Reads decrypted by each previous key, by position in the previous-key
    /// list (`hits[i]` ↔ `previous_keys[i]`); empty when there are none.
    /// Reads served by the current key are not counted.
    ///
    /// This is the rotation **drain signal**. It confirms that a grace window
    /// has drained; it does not shorten one. Follow the protocol's
    /// [scheduled-rotation runbook](https://github.com/cachekit-io/protocol/blob/main/decisions/key-rotation.md#runbooks-normative-for-docs):
    /// audit for non-expiring entries first, add the incoming key as
    /// decrypt-only fleet-wide, then promote it. The window clock starts only
    /// when that promotion deploy has completed on every instance — until
    /// then a lagging instance still writes fresh ciphertext under the
    /// retiring key and reads it silently as *its* current key (index 0).
    /// From that point, wait at least the longest TTL in use (per-entry TTLs
    /// passed to `set_with_ttl` count, not just the default), aggregating
    /// counts across every instance — they are per process and reset on
    /// restart. Only when the retiring key's count has stayed flat over that
    /// whole window has every live entry aged out or been re-encrypted on
    /// write, and the key can be dropped from the previous list without a
    /// hard cut-over. Positions are comparable across instances only once
    /// they all run the same keyring configuration. The signal carries no
    /// key material — positions and counts only.
    ///
    /// ```
    /// use cachekit::EncryptionLayer;
    ///
    /// let k1 = [0x11u8; 32]; // retiring master key
    /// let k2 = [0x22u8; 32]; // current master key after rotation
    ///
    /// // Encrypted under k1, before the rotation...
    /// let ciphertext = EncryptionLayer::new(&k1, "tenant-123")?.encrypt(b"cached value", "user:1")?;
    ///
    /// // ...a layer [current=k2, previous=[k1]] serves it from previous[0]:
    /// // the retiring key is still draining, not yet safe to drop.
    /// let layer = EncryptionLayer::with_previous_keys(&k2, &[&k1], "tenant-123")?;
    /// assert_eq!(layer.previous_key_hits(), vec![0]);
    /// assert_eq!(layer.decrypt(&ciphertext, "user:1")?, b"cached value");
    /// assert_eq!(layer.previous_key_hits(), vec![1]);
    /// # Ok::<(), cachekit::CachekitError>(())
    /// ```
    pub fn previous_key_hits(&self) -> Vec<u64> {
        self.previous_key_hits
            .iter()
            .map(|h| h.load(Ordering::Relaxed))
            .collect()
    }

    /// Whether cachekit-core detected AES hardware support on this host.
    ///
    /// Forwards `ZeroKnowledgeEncryptor::hardware_acceleration_enabled()`.
    /// Informational only — `ring`/`aes-gcm` dispatch independently of it; the
    /// per-architecture semantics are core's ([README → Zero-Knowledge
    /// Encryption](https://github.com/cachekit-io/cachekit-rs#zero-knowledge-encryption)).
    pub fn hardware_acceleration_enabled(&self) -> bool {
        self.encryptor.hardware_acceleration_enabled()
    }

    /// Return the tenant ID used for key derivation.
    pub fn tenant_id(&self) -> &str {
        &self.tenant_id
    }

    /// Build AAD v0x03 for a given cache key and compression flag.
    ///
    /// Format: `[0x03][len][tenant_id][len][cache_key][len]["msgpack"][len]["True"/"False"]`
    ///
    /// All lengths are 4-byte big-endian u32 to prevent boundary-confusion attacks.
    ///
    /// # Invariant: `compressed` is always `false` in production
    ///
    /// `encrypt` and `decrypt` pass the literal `false`, and that is normative, not a
    /// gap: this SDK's only cross-SDK encrypted surface is interop mode, and
    /// `protocol/spec/interop-mode.md` ("Encryption in Interop Mode") mandates the AAD
    /// components `format = "msgpack"`, `compressed = "False"` — there is no
    /// compression in interop mode.
    ///
    /// Why a reader cannot recover from a wrong flag here: in the general flow a reader
    /// rebuilds the AAD from the **stored cleartext metadata** written alongside the
    /// ciphertext (`protocol/spec/encryption.md`, encrypt step 6 / decrypt step 3).
    /// cachekit-rs stores no such per-entry metadata — it writes plain
    /// `nonce ‖ ciphertext ‖ tag` with no header — so its readers have nothing to read
    /// the flag from and must reconstruct the AAD from the interop-pinned constants
    /// alone. Combined with the no-retry rule (a reader MUST NOT retry decryption with
    /// any alternative AAD input, `protocol/spec/encryption.md`), emitting
    /// `compressed = "True"` here would produce ciphertext that no conformant peer can
    /// authenticate, and none may probe for.
    ///
    /// So do not flip this literal to thread a live compression flag through
    /// `encrypt`/`decrypt`. If this SDK ever does compress, the flag must be *stored*
    /// per entry and threaded from that stored value — `protocol/spec/encryption.md`
    /// requires `compressed` to describe the actual plaintext, and claiming `"False"`
    /// over compressed bytes is the cachekit-py#166 conformance bug (round-trips
    /// in-process, fails authentication for every correct second reader). A compressed
    /// cross-SDK profile is a versioned protocol change (interop/v2),
    /// not an SDK flag.
    ///
    /// The `"True"`/`"False"` tokens are frozen byte-level protocol constants —
    /// normative byte table in `protocol/spec/encryption.md`, section
    /// "`compressed` tokens"; decided in
    /// [protocol#12](https://github.com/cachekit-io/protocol/issues/12) (resolved
    /// 2026-07-19: a spec correction, not a wire change). The `true` branch stays,
    /// exercised by conformance tests, so those bytes remain pinned.
    pub fn build_aad(&self, cache_key: &str, compressed: bool) -> Vec<u8> {
        let format_str = b"msgpack";
        let compressed_str = if compressed {
            b"True" as &[u8]
        } else {
            b"False"
        };

        let tenant_bytes = self.tenant_id.as_bytes();
        let key_bytes = cache_key.as_bytes();

        // Pre-allocate: version(1) + 4 length fields(16) + data
        let capacity =
            1 + 16 + tenant_bytes.len() + key_bytes.len() + format_str.len() + compressed_str.len();
        let mut aad = Vec::with_capacity(capacity);

        aad.push(AAD_VERSION);

        // All components are bounded: tenant_id <= 255 (validated in new()),
        // cache_key <= 1024 (validated by client), format/compressed are constants.
        // Safe to use len_u32 helper which saturates on overflow.

        // tenant_id
        aad.extend_from_slice(&len_u32(tenant_bytes.len()).to_be_bytes());
        aad.extend_from_slice(tenant_bytes);

        // cache_key
        aad.extend_from_slice(&len_u32(key_bytes.len()).to_be_bytes());
        aad.extend_from_slice(key_bytes);

        // format
        aad.extend_from_slice(&len_u32(format_str.len()).to_be_bytes());
        aad.extend_from_slice(format_str);

        // compressed flag
        aad.extend_from_slice(&len_u32(compressed_str.len()).to_be_bytes());
        aad.extend_from_slice(compressed_str);

        aad
    }
}

/// Convert a usize length to u32 for AAD encoding, saturating on overflow.
/// In practice all inputs are validated to fit (tenant_id <= 255, cache_key <= 1024).
#[allow(clippy::cast_possible_truncation)]
fn len_u32(len: usize) -> u32 {
    u32::try_from(len).unwrap_or(u32::MAX)
}

impl std::fmt::Debug for EncryptionLayer {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("EncryptionLayer")
            .field("tenant_id", &self.tenant_id)
            .field("derived_key", &"[REDACTED]")
            .finish()
    }
}

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

    const TEST_MASTER_KEY: &[u8] = b"test_master_key_32_bytes_long!!!";
    const TEST_TENANT: &str = "test-tenant";

    #[test]
    fn roundtrip_encrypt_decrypt() {
        let layer = EncryptionLayer::new(TEST_MASTER_KEY, TEST_TENANT).unwrap();
        let plaintext = b"hello, zero-knowledge world";

        let ciphertext = layer.encrypt(plaintext, "my:key").unwrap();
        let decrypted = layer.decrypt(&ciphertext, "my:key").unwrap();

        assert_eq!(decrypted, plaintext);
    }

    #[test]
    fn wrong_cache_key_fails_decryption() {
        let layer = EncryptionLayer::new(TEST_MASTER_KEY, TEST_TENANT).unwrap();
        let ciphertext = layer.encrypt(b"secret", "key:a").unwrap();

        let result = layer.decrypt(&ciphertext, "key:b");
        assert!(result.is_err(), "decryption with wrong cache key must fail");
    }

    #[test]
    fn different_tenants_produce_different_ciphertext() {
        let layer_a = EncryptionLayer::new(TEST_MASTER_KEY, "tenant-a").unwrap();
        let layer_b = EncryptionLayer::new(TEST_MASTER_KEY, "tenant-b").unwrap();

        let ct_a = layer_a.encrypt(b"same data", "same:key").unwrap();
        let ct_b = layer_b.encrypt(b"same data", "same:key").unwrap();

        // Nonces differ, so ciphertext differs, but also keys differ
        assert_ne!(ct_a, ct_b);

        // Cross-tenant decryption must fail
        assert!(layer_b.decrypt(&ct_a, "same:key").is_err());
    }

    #[test]
    fn master_key_too_short() {
        let result = EncryptionLayer::new(b"short", "tenant");
        assert!(result.is_err());
        let msg = result.unwrap_err().to_string();
        assert!(msg.contains("exactly 32 bytes"), "got: {msg}");
    }

    fn assert_config_err(result: &Result<EncryptionLayer, CachekitError>, what: &str) {
        assert!(
            matches!(result, Err(CachekitError::Config(_))),
            "{what}: expected Config error, got {result:?}"
        );
    }

    #[test]
    fn new_requires_exactly_32_bytes() {
        assert_config_err(&EncryptionLayer::new(&[7; 31], TEST_TENANT), "31 bytes");
        assert_config_err(&EncryptionLayer::new(&[7; 33], TEST_TENANT), "33 bytes");
        assert!(EncryptionLayer::new(&[7; 32], TEST_TENANT).is_ok());
    }

    #[test]
    fn with_previous_keys_requires_exactly_32_byte_current_key() {
        for len in [31, 33, 48] {
            let current = vec![7u8; len];
            assert_config_err(
                &EncryptionLayer::with_previous_keys(&current, &[K1], TEST_TENANT),
                &format!("{len}-byte current key"),
            );
        }
        assert!(EncryptionLayer::with_previous_keys(K2, &[K1], TEST_TENANT).is_ok());
    }

    #[test]
    fn with_previous_keys_requires_exactly_32_byte_previous_keys() {
        for len in [31, 33] {
            let previous = vec![7u8; len];
            assert_config_err(
                &EncryptionLayer::with_previous_keys(K2, &[K1, &previous], TEST_TENANT),
                &format!("{len}-byte previous key"),
            );
        }
    }

    #[test]
    fn hex_decoded_keys_accept_longer_than_32_bytes() {
        // Rule 3: the hex path accepts >= 32 decoded bytes.
        let long = [7u8; 48];
        let layer =
            EncryptionLayer::from_hex_decoded_keys(&long, &[&long[..33]], TEST_TENANT).unwrap();
        let ct = layer.encrypt(b"v", "k").unwrap();
        assert_eq!(layer.decrypt(&ct, "k").unwrap(), b"v");
        assert_config_err(
            &EncryptionLayer::from_hex_decoded_keys(&long[..31], &[], TEST_TENANT),
            "hex path keeps its 32-byte floor",
        );
    }

    #[test]
    fn aad_v03_format() {
        let layer = EncryptionLayer::new(TEST_MASTER_KEY, TEST_TENANT).unwrap();
        let aad = layer.build_aad("user:42", false);

        // Version byte
        assert_eq!(aad[0], 0x03);

        // tenant_id length (4 bytes BE) + tenant_id
        let tenant_len = u32::from_be_bytes(aad[1..5].try_into().unwrap()) as usize;
        assert_eq!(tenant_len, TEST_TENANT.len());
        assert_eq!(&aad[5..5 + tenant_len], TEST_TENANT.as_bytes());

        // cache_key length + cache_key
        let offset = 5 + tenant_len;
        let key_len = u32::from_be_bytes(aad[offset..offset + 4].try_into().unwrap()) as usize;
        assert_eq!(key_len, 7); // "user:42"
        assert_eq!(&aad[offset + 4..offset + 4 + key_len], b"user:42");

        // format length + format
        let offset = offset + 4 + key_len;
        let fmt_len = u32::from_be_bytes(aad[offset..offset + 4].try_into().unwrap()) as usize;
        assert_eq!(&aad[offset + 4..offset + 4 + fmt_len], b"msgpack");

        // compressed length + compressed
        let offset = offset + 4 + fmt_len;
        let comp_len = u32::from_be_bytes(aad[offset..offset + 4].try_into().unwrap()) as usize;
        assert_eq!(&aad[offset + 4..offset + 4 + comp_len], b"False");
    }

    #[test]
    fn aad_compressed_flag() {
        let layer = EncryptionLayer::new(TEST_MASTER_KEY, TEST_TENANT).unwrap();
        let aad_false = layer.build_aad("k", false);
        let aad_true = layer.build_aad("k", true);

        assert_ne!(aad_false, aad_true);
        // "True" is at the end
        assert!(aad_true.ends_with(b"True"));
        assert!(aad_false.ends_with(b"False"));
    }

    // ── Keyring rotation ─────────────────────────────────────────────────────

    const K1: &[u8] = &[0x11; 32];
    const K2: &[u8] = &[0x22; 32];

    #[test]
    fn previous_key_decrypts_after_rotation() {
        // Written under k1 before the rotation...
        let old_layer = EncryptionLayer::new(K1, TEST_TENANT).unwrap();
        let ciphertext = old_layer.encrypt(b"pre-rotation value", "user:1").unwrap();

        // ...still readable with current=k2, previous=[k1] (identical AAD)...
        let rotated = EncryptionLayer::with_previous_keys(K2, &[K1], TEST_TENANT).unwrap();
        let plaintext = rotated.decrypt(&ciphertext, "user:1").unwrap();
        assert_eq!(plaintext, b"pre-rotation value");

        // ...and unreadable after a hard cut-over (previous=[]).
        let cut_over = EncryptionLayer::with_previous_keys(K2, &[], TEST_TENANT).unwrap();
        assert!(cut_over.decrypt(&ciphertext, "user:1").is_err());
    }

    #[test]
    fn rotated_layer_still_writes_under_current_key() {
        let rotated = EncryptionLayer::with_previous_keys(K2, &[K1], TEST_TENANT).unwrap();
        let ciphertext = rotated.encrypt(b"fresh write", "user:2").unwrap();

        // A current-key-only layer reads it: writes never use previous keys.
        let current_only = EncryptionLayer::new(K2, TEST_TENANT).unwrap();
        assert_eq!(
            current_only.decrypt(&ciphertext, "user:2").unwrap(),
            b"fresh write"
        );
    }

    #[test]
    fn exactly_three_previous_keys_is_accepted() {
        // Boundary success: a `>=` cap check instead of `>` would reject a
        // legitimate three-key rotation window and still pass every
        // rejecting-side test.
        let keys: Vec<[u8; 32]> = (1..=3).map(|i| [i; 32]).collect();
        let refs: Vec<&[u8]> = keys.iter().map(|k| k.as_slice()).collect();

        let layer = EncryptionLayer::with_previous_keys(K2, &refs, TEST_TENANT);
        assert!(
            layer.is_ok(),
            "cap is 3: exactly three previous keys must build"
        );
    }

    #[test]
    fn more_than_three_previous_keys_is_config_error() {
        let keys: Vec<[u8; 32]> = (1..=4).map(|i| [i; 32]).collect();
        let refs: Vec<&[u8]> = keys.iter().map(|k| k.as_slice()).collect();

        let result = EncryptionLayer::with_previous_keys(K2, &refs, TEST_TENANT);
        assert!(
            matches!(result, Err(CachekitError::Config(_))),
            "cap of 3 must reject, never truncate"
        );
    }

    #[test]
    fn current_key_in_previous_list_is_config_error() {
        let result = EncryptionLayer::with_previous_keys(K2, &[K1, K2], TEST_TENANT);
        assert!(
            matches!(result, Err(CachekitError::Config(_))),
            "forward-only: current key must not be decrypt-only"
        );
    }

    #[test]
    fn short_previous_key_is_config_error() {
        let short = [0x01u8; 16]; // core would accept 16; raw keys must be exactly 32
        let result = EncryptionLayer::with_previous_keys(K2, &[&short], TEST_TENANT);
        assert!(matches!(result, Err(CachekitError::Config(_))));
    }

    // ── Rotation drain signal ────────────────────────────────────────────────

    #[test]
    fn current_key_hit_is_not_counted() {
        // No previous keys: the drain signal has nothing to report.
        let single = EncryptionLayer::new(K2, TEST_TENANT).unwrap();
        let ct = single.encrypt(b"v", "user:3").unwrap();
        single.decrypt(&ct, "user:3").unwrap();
        assert!(single.previous_key_hits().is_empty());

        // With a previous key, a current-key read (index 0) is a zero reading.
        let rotated = EncryptionLayer::with_previous_keys(K2, &[K1], TEST_TENANT).unwrap();
        let ct = rotated.encrypt(b"fresh write", "user:3").unwrap();
        rotated.decrypt(&ct, "user:3").unwrap();
        assert_eq!(rotated.previous_key_hits(), vec![0]);
    }

    #[test]
    fn previous_key_hit_is_counted_at_its_position() {
        const K3: &[u8] = &[0x33; 32];
        let k1_ct = EncryptionLayer::new(K1, TEST_TENANT)
            .unwrap()
            .encrypt(b"k1 era", "user:4")
            .unwrap();
        let k2_ct = EncryptionLayer::new(K2, TEST_TENANT)
            .unwrap()
            .encrypt(b"k2 era", "user:5")
            .unwrap();

        // current=k3, previous=[k2, k1]: counts index by position in the previous list.
        let rotated = EncryptionLayer::with_previous_keys(K3, &[K2, K1], TEST_TENANT).unwrap();
        assert_eq!(rotated.previous_key_hits(), vec![0, 0]);

        rotated.decrypt(&k1_ct, "user:4").unwrap();
        assert_eq!(rotated.previous_key_hits(), vec![0, 1], "k1 is previous[1]");

        rotated.decrypt(&k2_ct, "user:5").unwrap();
        rotated.decrypt(&k2_ct, "user:5").unwrap();
        assert_eq!(rotated.previous_key_hits(), vec![2, 1], "k2 is previous[0]");
    }

    #[test]
    fn failed_decrypt_is_not_a_previous_key_hit() {
        let k1_ct = EncryptionLayer::new(K1, TEST_TENANT)
            .unwrap()
            .encrypt(b"v", "key:a")
            .unwrap();
        let rotated = EncryptionLayer::with_previous_keys(K2, &[K1], TEST_TENANT).unwrap();

        // Wrong cache key: every attempt fails authentication — no key won.
        assert!(rotated.decrypt(&k1_ct, "key:b").is_err());
        assert_eq!(rotated.previous_key_hits(), vec![0]);
    }

    #[test]
    fn hardware_acceleration_reports_core_detection() {
        let layer = EncryptionLayer::new(TEST_MASTER_KEY, TEST_TENANT).unwrap();
        // Delegation, not a constant: agrees with a fresh core encryptor...
        let core = ZeroKnowledgeEncryptor::new().unwrap();
        assert_eq!(
            layer.hardware_acceleration_enabled(),
            core.hardware_acceleration_enabled()
        );
        // ...and on x86_64 with the CPU itself. This pins core 0.6's runtime
        // probe — the only executed evidence in any repo that the bool is live
        // detection, not a constant.
        #[cfg(target_arch = "x86_64")]
        assert_eq!(
            layer.hardware_acceleration_enabled(),
            std::arch::is_x86_feature_detected!("aes")
        );
    }

    #[test]
    fn debug_redacts_key() {
        let layer = EncryptionLayer::new(TEST_MASTER_KEY, TEST_TENANT).unwrap();
        let debug = format!("{layer:?}");
        assert!(debug.contains("[REDACTED]"));
        assert!(!debug.contains("test_master_key"));
    }
}