znippy-plugin-git 0.1.1

//! `__gunnar_secrets__` — per-repository secret material, on the same
//! one-RecordBatch-per-push log as [`crate::refs`].
//!
//! Deploy keys, webhook tokens and signing keys change with the same shape a ref
//! does: rarely, transactionally, and with a history worth keeping. So they use
//! the same mechanism ([`crate::pushlog`]) rather than a second one — the reason
//! this is a *native* package format and not two bolt-ons stapled together.
//!
//! ## What znippy is and is not responsible for
//!
//! `ciphertext` is **already encrypted by the caller**. znippy never sees
//! plaintext, holds no key, and performs no encryption: it stores an opaque blob
//! and records what it is called and when it arrived. Anything else would put
//! key management inside an archiver.
//!
//! Two consequences follow, and both are enforced here:
//!
//! * ciphertext is high-entropy, so it is **never compressed** — the same law
//!   that keeps znippy off `.pack` files. [`znippy_common::SkipPolicy`] is asked
//!   for the whole batch via [`secret_skip_policy`].
//! * a secret that is stored unencrypted by mistake is a leak that no later fix
//!   undoes, so [`SecretUpdate::new`] refuses empty ciphertext outright rather
//!   than writing a row that *looks* like a stored secret.

use std::collections::BTreeMap;
use std::path::Path;
use std::sync::Arc;

use anyhow::{Result, anyhow, bail};
use znippy_common::GUNNAR_SECRETS_MODULE;
use znippy_common::arrow::array::{
    Array, BinaryArray, BinaryBuilder, StringArray, StringBuilder, UInt64Array, UInt64Builder,
};
use znippy_common::arrow::datatypes::{DataType, Field, Schema};
use znippy_common::arrow::record_batch::RecordBatch;
use znippy_common::precompressed::SkipPolicy;

use crate::pushlog::{PushLog, PushLogScan, read_sealed};

/// Secret material is ciphertext: it never compresses, so it never goes to the
/// codec. One policy for the whole batch — the free, exact answer.
pub fn secret_skip_policy() -> SkipPolicy {
    SkipPolicy::already_compressed()
}

/// One secret update inside a push.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SecretUpdate {
    pub name: String,
    /// Opaque, already-encrypted bytes. `None` revokes the secret.
    pub ciphertext: Option<Vec<u8>>,
    /// Free-form label for the key/recipient that can open it (e.g. an age
    /// recipient). Never the key itself.
    pub recipient: Option<String>,
}

impl SecretUpdate {
    /// Store `ciphertext` under `name`.
    ///
    /// Errors on empty ciphertext: an empty blob is what a caller that forgot to
    /// encrypt produces, and a row that claims to hold a secret but holds
    /// nothing is worse than a refusal.
    pub fn new(name: impl Into<String>, ciphertext: Vec<u8>) -> Result<Self> {
        if ciphertext.is_empty() {
            bail!("refusing to store an empty ciphertext — encrypt before handing it to znippy");
        }
        Ok(Self { name: name.into(), ciphertext: Some(ciphertext), recipient: None })
    }

    pub fn revoke(name: impl Into<String>) -> Self {
        Self { name: name.into(), ciphertext: None, recipient: None }
    }

    pub fn for_recipient(mut self, recipient: impl Into<String>) -> Self {
        self.recipient = Some(recipient.into());
        self
    }
}

/// The state of one secret after replaying the log.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SecretState {
    pub ciphertext: Vec<u8>,
    pub recipient: Option<String>,
    pub push_seq: u64,
    pub updated_ms: u64,
}

pub fn secrets_schema() -> Arc<Schema> {
    Arc::new(Schema::new(vec![
        Field::new("name", DataType::Utf8, false),
        Field::new("ciphertext", DataType::Binary, true),
        Field::new("recipient", DataType::Utf8, true),
        Field::new("push_seq", DataType::UInt64, false),
        Field::new("updated_ms", DataType::UInt64, false),
    ]))
}

pub fn build_push_batch(
    updates: &[SecretUpdate],
    push_seq: u64,
    updated_ms: u64,
) -> Result<RecordBatch> {
    let n = updates.len();
    let mut name = StringBuilder::with_capacity(n, n * 32);
    let mut ct = BinaryBuilder::new();
    let mut recipient = StringBuilder::with_capacity(n, n * 32);
    let mut seq = UInt64Builder::with_capacity(n);
    let mut ms = UInt64Builder::with_capacity(n);

    for u in updates {
        name.append_value(&u.name);
        match &u.ciphertext {
            Some(b) => ct.append_value(b),
            None => ct.append_null(),
        }
        match &u.recipient {
            Some(r) => recipient.append_value(r),
            None => recipient.append_null(),
        }
        seq.append_value(push_seq);
        ms.append_value(updated_ms);
    }

    RecordBatch::try_new(
        secrets_schema(),
        vec![
            Arc::new(name.finish()),
            Arc::new(ct.finish()),
            Arc::new(recipient.finish()),
            Arc::new(seq.finish()),
            Arc::new(ms.finish()),
        ],
    )
    .map_err(|e| anyhow!("secrets push batch: {e}"))
}

/// The secrets log of one repository.
pub struct SecretsLog {
    log: PushLog,
}

impl SecretsLog {
    pub fn new(path: impl Into<std::path::PathBuf>) -> Self {
        Self { log: PushLog::new(path, secrets_schema()) }
    }

    pub fn next_push_seq(&self) -> Result<u64> {
        let scan = self.log.scan()?;
        Ok(max_push_seq(&scan.pushes).map_or(0, |m| m + 1))
    }

    pub fn push(&self, updates: &[SecretUpdate]) -> Result<u64> {
        let seq = self.next_push_seq()?;
        let ms = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .map(|d| d.as_millis() as u64)
            .unwrap_or(0);
        let batch = build_push_batch(updates, seq, ms)?;
        self.log.append(&batch)?;
        Ok(seq)
    }

    pub fn scan(&self) -> Result<PushLogScan> {
        self.log.scan()
    }

    /// Fold every frame into one. See [`PushLog::compact`].
    pub fn compact(&self) -> Result<crate::pushlog::CompactionReport> {
        self.log.compact()
    }

    /// Compact if the log has grown past `policy`.
    pub fn maybe_compact(
        &self,
        policy: crate::pushlog::CompactionPolicy,
    ) -> Result<Option<crate::pushlog::CompactionReport>> {
        self.log.maybe_compact(policy)
    }

    pub fn current(&self) -> Result<BTreeMap<String, SecretState>> {
        fold(&self.log.scan()?.pushes)
    }

    pub fn seal_section(&self) -> Result<znippy_common::ReservedSection> {
        self.log.seal_section(GUNNAR_SECRETS_MODULE)
    }
}

fn max_push_seq(batches: &[RecordBatch]) -> Option<u64> {
    let mut max = None;
    for b in batches {
        let seq = b.column_by_name("push_seq")?.as_any().downcast_ref::<UInt64Array>()?;
        for i in 0..seq.len() {
            max = Some(max.map_or(seq.value(i), |m: u64| m.max(seq.value(i))));
        }
    }
    max
}

/// Replay pushes into the current secret set. Last writer wins by `push_seq`; a
/// null ciphertext revokes.
pub fn fold(batches: &[RecordBatch]) -> Result<BTreeMap<String, SecretState>> {
    let mut rows: Vec<(u64, usize, String, Option<SecretState>)> = Vec::new();

    for (bi, b) in batches.iter().enumerate() {
        let name = col::<StringArray>(b, "name")?;
        let ct = col::<BinaryArray>(b, "ciphertext")?;
        let recipient = col::<StringArray>(b, "recipient")?;
        let seq = col::<UInt64Array>(b, "push_seq")?;
        let ms = col::<UInt64Array>(b, "updated_ms")?;

        for i in 0..b.num_rows() {
            let state = (!ct.is_null(i)).then(|| SecretState {
                ciphertext: ct.value(i).to_vec(),
                recipient: (!recipient.is_null(i)).then(|| recipient.value(i).to_string()),
                push_seq: seq.value(i),
                updated_ms: ms.value(i),
            });
            rows.push((seq.value(i), bi, name.value(i).to_string(), state));
        }
    }

    rows.sort_by_key(|(seq, bi, _, _)| (*seq, *bi));

    let mut out: BTreeMap<String, SecretState> = BTreeMap::new();
    for (_, _, name, state) in rows {
        match state {
            Some(s) => {
                out.insert(name, s);
            }
            None => {
                out.remove(&name);
            }
        }
    }
    Ok(out)
}

/// Read the sealed `__gunnar_secrets__` section out of an archive.
pub fn read_secrets(archive: &Path) -> Result<Option<BTreeMap<String, SecretState>>> {
    match read_sealed(archive, GUNNAR_SECRETS_MODULE)? {
        Some(batches) => Ok(Some(fold(&batches)?)),
        None => Ok(None),
    }
}

fn col<'a, T: Array + 'static>(b: &'a RecordBatch, name: &str) -> Result<&'a T> {
    b.column_by_name(name)
        .ok_or_else(|| anyhow!("secrets: no `{name}` column"))?
        .as_any()
        .downcast_ref::<T>()
        .ok_or_else(|| anyhow!("secrets: `{name}` has an unexpected type"))
}

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

    fn tmpdir(tag: &str) -> std::path::PathBuf {
        let ns = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_nanos();
        let d = std::env::temp_dir().join(format!("znippy_secrets_{tag}_{ns}"));
        std::fs::create_dir_all(&d).unwrap();
        d
    }

    #[test]
    fn rotation_keeps_the_newest_and_revocation_removes() {
        let dir = tmpdir("rotate");
        let log = SecretsLog::new(dir.join("secrets.log"));
        log.push(&[
            SecretUpdate::new("deploy-key", b"CIPHER-v1".to_vec()).unwrap(),
            SecretUpdate::new("webhook", b"CIPHER-hook".to_vec()).unwrap(),
        ])
        .unwrap();
        log.push(&[SecretUpdate::new("deploy-key", b"CIPHER-v2".to_vec()).unwrap()]).unwrap();
        log.push(&[SecretUpdate::revoke("webhook")]).unwrap();

        let now = log.current().unwrap();
        assert_eq!(now["deploy-key"].ciphertext, b"CIPHER-v2".to_vec(), "rotation must win");
        assert!(!now.contains_key("webhook"), "revocation must remove the secret");
        assert_eq!(now.len(), 1);

        std::fs::remove_dir_all(&dir).ok();
    }

    /// A rotation interrupted by a crash must leave the OLD secret intact and
    /// usable — never a half-written new one, and never neither.
    #[test]
    fn a_crash_during_rotation_leaves_the_old_secret_usable() {
        let dir = tmpdir("crash");
        let path = dir.join("secrets.log");
        let log = SecretsLog::new(&path);
        log.push(&[SecretUpdate::new("deploy-key", b"CIPHER-v1".to_vec()).unwrap()]).unwrap();
        let before = std::fs::metadata(&path).unwrap().len();
        log.push(&[SecretUpdate::new("deploy-key", vec![0xAB; 512]).unwrap()]).unwrap();
        let after = std::fs::metadata(&path).unwrap().len();
        let intact = std::fs::read(&path).unwrap();

        for cut in (before + 1)..after {
            std::fs::write(&path, &intact).unwrap();
            truncate_for_test(&path, cut).unwrap();
            let now = log.current().unwrap();
            assert_eq!(
                now["deploy-key"].ciphertext,
                b"CIPHER-v1".to_vec(),
                "cut at {cut}: a torn rotation must leave the previous secret in force"
            );
        }

        std::fs::write(&path, &intact).unwrap();
        assert_eq!(log.current().unwrap()["deploy-key"].ciphertext, vec![0xAB; 512]);
        std::fs::remove_dir_all(&dir).ok();
    }

    /// znippy must never compress ciphertext. The policy is asked once for the
    /// whole batch and must answer "skip" without needing to look at any bytes.
    #[test]
    fn ciphertext_is_never_offered_to_the_codec() {
        let p = secret_skip_policy();
        assert!(
            p.skip_by_path(Path::new("secrets/deploy-key")),
            "secret material must skip the codec on the path decision alone"
        );
        // High-entropy bytes with no recognisable magic: the byte probe alone
        // would NOT skip these, which is exactly why the batch policy exists.
        let entropy: Vec<u8> = (0..64u32).map(|i| (i.wrapping_mul(167) ^ 0x5A) as u8).collect();
        assert!(
            !SkipPolicy::resolve().skip_by_bytes(&entropy),
            "control: the default probe does not recognise raw ciphertext — \
             if this ever passes, this test has stopped proving anything"
        );
        assert!(p.skip_by_bytes(&entropy), "the secrets policy must skip it regardless");
    }

    /// Empty ciphertext is what a caller who forgot to encrypt produces.
    #[test]
    fn an_empty_ciphertext_is_refused() {
        let err = SecretUpdate::new("oops", Vec::new()).unwrap_err().to_string();
        assert!(err.contains("empty ciphertext"), "expected a refusal, got: {err}");
    }
}