1use std::collections::BTreeMap;
26use std::path::Path;
27use std::sync::Arc;
28
29use anyhow::{Result, anyhow, bail};
30use znippy_common::GUNNAR_SECRETS_MODULE;
31use znippy_common::arrow::array::{
32 Array, BinaryArray, BinaryBuilder, StringArray, StringBuilder, UInt64Array, UInt64Builder,
33};
34use znippy_common::arrow::datatypes::{DataType, Field, Schema};
35use znippy_common::arrow::record_batch::RecordBatch;
36use znippy_common::precompressed::SkipPolicy;
37
38use crate::pushlog::{PushLog, PushLogScan, read_sealed};
39
40pub fn secret_skip_policy() -> SkipPolicy {
43 SkipPolicy::already_compressed()
44}
45
46#[derive(Debug, Clone, PartialEq, Eq)]
48pub struct SecretUpdate {
49 pub name: String,
50 pub ciphertext: Option<Vec<u8>>,
52 pub recipient: Option<String>,
55}
56
57impl SecretUpdate {
58 pub fn new(name: impl Into<String>, ciphertext: Vec<u8>) -> Result<Self> {
64 if ciphertext.is_empty() {
65 bail!("refusing to store an empty ciphertext — encrypt before handing it to znippy");
66 }
67 Ok(Self { name: name.into(), ciphertext: Some(ciphertext), recipient: None })
68 }
69
70 pub fn revoke(name: impl Into<String>) -> Self {
71 Self { name: name.into(), ciphertext: None, recipient: None }
72 }
73
74 pub fn for_recipient(mut self, recipient: impl Into<String>) -> Self {
75 self.recipient = Some(recipient.into());
76 self
77 }
78}
79
80#[derive(Debug, Clone, PartialEq, Eq)]
82pub struct SecretState {
83 pub ciphertext: Vec<u8>,
84 pub recipient: Option<String>,
85 pub push_seq: u64,
86 pub updated_ms: u64,
87}
88
89pub fn secrets_schema() -> Arc<Schema> {
90 Arc::new(Schema::new(vec![
91 Field::new("name", DataType::Utf8, false),
92 Field::new("ciphertext", DataType::Binary, true),
93 Field::new("recipient", DataType::Utf8, true),
94 Field::new("push_seq", DataType::UInt64, false),
95 Field::new("updated_ms", DataType::UInt64, false),
96 ]))
97}
98
99pub fn build_push_batch(
100 updates: &[SecretUpdate],
101 push_seq: u64,
102 updated_ms: u64,
103) -> Result<RecordBatch> {
104 let n = updates.len();
105 let mut name = StringBuilder::with_capacity(n, n * 32);
106 let mut ct = BinaryBuilder::new();
107 let mut recipient = StringBuilder::with_capacity(n, n * 32);
108 let mut seq = UInt64Builder::with_capacity(n);
109 let mut ms = UInt64Builder::with_capacity(n);
110
111 for u in updates {
112 name.append_value(&u.name);
113 match &u.ciphertext {
114 Some(b) => ct.append_value(b),
115 None => ct.append_null(),
116 }
117 match &u.recipient {
118 Some(r) => recipient.append_value(r),
119 None => recipient.append_null(),
120 }
121 seq.append_value(push_seq);
122 ms.append_value(updated_ms);
123 }
124
125 RecordBatch::try_new(
126 secrets_schema(),
127 vec![
128 Arc::new(name.finish()),
129 Arc::new(ct.finish()),
130 Arc::new(recipient.finish()),
131 Arc::new(seq.finish()),
132 Arc::new(ms.finish()),
133 ],
134 )
135 .map_err(|e| anyhow!("secrets push batch: {e}"))
136}
137
138pub struct SecretsLog {
140 log: PushLog,
141}
142
143impl SecretsLog {
144 pub fn new(path: impl Into<std::path::PathBuf>) -> Self {
145 Self { log: PushLog::new(path, secrets_schema()) }
146 }
147
148 pub fn next_push_seq(&self) -> Result<u64> {
149 let scan = self.log.scan()?;
150 Ok(max_push_seq(&scan.pushes).map_or(0, |m| m + 1))
151 }
152
153 pub fn push(&self, updates: &[SecretUpdate]) -> Result<u64> {
154 let seq = self.next_push_seq()?;
155 let ms = std::time::SystemTime::now()
156 .duration_since(std::time::UNIX_EPOCH)
157 .map(|d| d.as_millis() as u64)
158 .unwrap_or(0);
159 let batch = build_push_batch(updates, seq, ms)?;
160 self.log.append(&batch)?;
161 Ok(seq)
162 }
163
164 pub fn scan(&self) -> Result<PushLogScan> {
165 self.log.scan()
166 }
167
168 pub fn compact(&self) -> Result<crate::pushlog::CompactionReport> {
170 self.log.compact()
171 }
172
173 pub fn maybe_compact(
175 &self,
176 policy: crate::pushlog::CompactionPolicy,
177 ) -> Result<Option<crate::pushlog::CompactionReport>> {
178 self.log.maybe_compact(policy)
179 }
180
181 pub fn current(&self) -> Result<BTreeMap<String, SecretState>> {
182 fold(&self.log.scan()?.pushes)
183 }
184
185 pub fn seal_section(&self) -> Result<znippy_common::ReservedSection> {
186 self.log.seal_section(GUNNAR_SECRETS_MODULE)
187 }
188}
189
190fn max_push_seq(batches: &[RecordBatch]) -> Option<u64> {
191 let mut max = None;
192 for b in batches {
193 let seq = b.column_by_name("push_seq")?.as_any().downcast_ref::<UInt64Array>()?;
194 for i in 0..seq.len() {
195 max = Some(max.map_or(seq.value(i), |m: u64| m.max(seq.value(i))));
196 }
197 }
198 max
199}
200
201pub fn fold(batches: &[RecordBatch]) -> Result<BTreeMap<String, SecretState>> {
204 let mut rows: Vec<(u64, usize, String, Option<SecretState>)> = Vec::new();
205
206 for (bi, b) in batches.iter().enumerate() {
207 let name = col::<StringArray>(b, "name")?;
208 let ct = col::<BinaryArray>(b, "ciphertext")?;
209 let recipient = col::<StringArray>(b, "recipient")?;
210 let seq = col::<UInt64Array>(b, "push_seq")?;
211 let ms = col::<UInt64Array>(b, "updated_ms")?;
212
213 for i in 0..b.num_rows() {
214 let state = (!ct.is_null(i)).then(|| SecretState {
215 ciphertext: ct.value(i).to_vec(),
216 recipient: (!recipient.is_null(i)).then(|| recipient.value(i).to_string()),
217 push_seq: seq.value(i),
218 updated_ms: ms.value(i),
219 });
220 rows.push((seq.value(i), bi, name.value(i).to_string(), state));
221 }
222 }
223
224 rows.sort_by_key(|(seq, bi, _, _)| (*seq, *bi));
225
226 let mut out: BTreeMap<String, SecretState> = BTreeMap::new();
227 for (_, _, name, state) in rows {
228 match state {
229 Some(s) => {
230 out.insert(name, s);
231 }
232 None => {
233 out.remove(&name);
234 }
235 }
236 }
237 Ok(out)
238}
239
240pub fn read_secrets(archive: &Path) -> Result<Option<BTreeMap<String, SecretState>>> {
242 match read_sealed(archive, GUNNAR_SECRETS_MODULE)? {
243 Some(batches) => Ok(Some(fold(&batches)?)),
244 None => Ok(None),
245 }
246}
247
248fn col<'a, T: Array + 'static>(b: &'a RecordBatch, name: &str) -> Result<&'a T> {
249 b.column_by_name(name)
250 .ok_or_else(|| anyhow!("secrets: no `{name}` column"))?
251 .as_any()
252 .downcast_ref::<T>()
253 .ok_or_else(|| anyhow!("secrets: `{name}` has an unexpected type"))
254}
255
256#[cfg(test)]
257mod tests {
258 use super::*;
259 use crate::pushlog::truncate_for_test;
260
261 fn tmpdir(tag: &str) -> std::path::PathBuf {
262 let ns = std::time::SystemTime::now()
263 .duration_since(std::time::UNIX_EPOCH)
264 .unwrap()
265 .as_nanos();
266 let d = std::env::temp_dir().join(format!("znippy_secrets_{tag}_{ns}"));
267 std::fs::create_dir_all(&d).unwrap();
268 d
269 }
270
271 #[test]
272 fn rotation_keeps_the_newest_and_revocation_removes() {
273 let dir = tmpdir("rotate");
274 let log = SecretsLog::new(dir.join("secrets.log"));
275 log.push(&[
276 SecretUpdate::new("deploy-key", b"CIPHER-v1".to_vec()).unwrap(),
277 SecretUpdate::new("webhook", b"CIPHER-hook".to_vec()).unwrap(),
278 ])
279 .unwrap();
280 log.push(&[SecretUpdate::new("deploy-key", b"CIPHER-v2".to_vec()).unwrap()]).unwrap();
281 log.push(&[SecretUpdate::revoke("webhook")]).unwrap();
282
283 let now = log.current().unwrap();
284 assert_eq!(now["deploy-key"].ciphertext, b"CIPHER-v2".to_vec(), "rotation must win");
285 assert!(!now.contains_key("webhook"), "revocation must remove the secret");
286 assert_eq!(now.len(), 1);
287
288 std::fs::remove_dir_all(&dir).ok();
289 }
290
291 #[test]
294 fn a_crash_during_rotation_leaves_the_old_secret_usable() {
295 let dir = tmpdir("crash");
296 let path = dir.join("secrets.log");
297 let log = SecretsLog::new(&path);
298 log.push(&[SecretUpdate::new("deploy-key", b"CIPHER-v1".to_vec()).unwrap()]).unwrap();
299 let before = std::fs::metadata(&path).unwrap().len();
300 log.push(&[SecretUpdate::new("deploy-key", vec![0xAB; 512]).unwrap()]).unwrap();
301 let after = std::fs::metadata(&path).unwrap().len();
302 let intact = std::fs::read(&path).unwrap();
303
304 for cut in (before + 1)..after {
305 std::fs::write(&path, &intact).unwrap();
306 truncate_for_test(&path, cut).unwrap();
307 let now = log.current().unwrap();
308 assert_eq!(
309 now["deploy-key"].ciphertext,
310 b"CIPHER-v1".to_vec(),
311 "cut at {cut}: a torn rotation must leave the previous secret in force"
312 );
313 }
314
315 std::fs::write(&path, &intact).unwrap();
316 assert_eq!(log.current().unwrap()["deploy-key"].ciphertext, vec![0xAB; 512]);
317 std::fs::remove_dir_all(&dir).ok();
318 }
319
320 #[test]
323 fn ciphertext_is_never_offered_to_the_codec() {
324 let p = secret_skip_policy();
325 assert!(
326 p.skip_by_path(Path::new("secrets/deploy-key")),
327 "secret material must skip the codec on the path decision alone"
328 );
329 let entropy: Vec<u8> = (0..64u32).map(|i| (i.wrapping_mul(167) ^ 0x5A) as u8).collect();
332 assert!(
333 !SkipPolicy::resolve().skip_by_bytes(&entropy),
334 "control: the default probe does not recognise raw ciphertext — \
335 if this ever passes, this test has stopped proving anything"
336 );
337 assert!(p.skip_by_bytes(&entropy), "the secrets policy must skip it regardless");
338 }
339
340 #[test]
342 fn an_empty_ciphertext_is_refused() {
343 let err = SecretUpdate::new("oops", Vec::new()).unwrap_err().to_string();
344 assert!(err.contains("empty ciphertext"), "expected a refusal, got: {err}");
345 }
346}