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};
pub fn secret_skip_policy() -> SkipPolicy {
SkipPolicy::already_compressed()
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SecretUpdate {
pub name: String,
pub ciphertext: Option<Vec<u8>>,
pub recipient: Option<String>,
}
impl SecretUpdate {
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
}
}
#[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}"))
}
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()
}
pub fn compact(&self) -> Result<crate::pushlog::CompactionReport> {
self.log.compact()
}
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
}
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)
}
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();
}
#[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();
}
#[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"
);
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");
}
#[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}");
}
}