huskarl_core/crypto/cipher/
scheduled.rs

1use std::{borrow::Cow, pin::Pin, sync::Arc};
2
3use crate::{
4    crypto::{
5        KeyMatchStrength,
6        cipher::{
7            AeadCipherSelector, AeadDecryptor, AeadEncryptor, AeadOutput, AeadSealer,
8            AeadSealerSelector, AeadUnsealer, AeadV1Cipher, CipherMatch, DecryptError,
9        },
10        refreshable::ScheduledRefreshable,
11    },
12    error::Error,
13    platform::{Duration, MaybeSendBoxFuture, MaybeSendFuture, MaybeSendSync},
14};
15
16/// An AEAD cipher that bounds the age of its keys to a TTL by reloading on the
17/// read path — so a decryption key *removed* upstream is dropped within the TTL
18/// even though it never fails to decrypt (you still hold it). The TTL bounds how
19/// long a removed key lingers; key *additions* are handled by the miss-triggered
20/// [`RetryingDecryptor`](super::RetryingDecryptor) layered on top.
21///
22/// On each [`decrypt`](AeadDecryptor::decrypt)/[`unseal`](AeadUnsealer::unseal),
23/// and inside each [`select_cipher`](AeadCipherSelector::select_cipher)/[`select_sealer`](AeadSealerSelector::select_sealer),
24/// the first caller past the TTL reloads single-flight (non-blocking for others),
25/// then proceeds against a frozen snapshot; for the outbound selectors the TTL
26/// bounds how quickly a rotated-in key is discovered rather than how quickly a removed
27/// one is dropped. Using the type directly as
28/// an [`AeadEncryptor`] serves the current snapshot *without* a reload — go through
29/// the selector for the freshness guarantee. The pure swap mechanism without policy
30/// is [`RefreshableCipher`](super::RefreshableCipher).
31///
32/// See [composing crypto strategies](crate::_docs::explanation::crypto_strategies)
33/// for how this layer (removals) pairs with the retrying layer (additions). All
34/// clones share the same underlying state, so a refresh through any clone is
35/// visible to all others.
36#[derive(Debug)]
37pub struct ScheduledRefreshCipher<C> {
38    inner: Arc<ScheduledRefreshable<C>>,
39}
40
41impl<C> Clone for ScheduledRefreshCipher<C> {
42    fn clone(&self) -> Self {
43        Self {
44            inner: Arc::clone(&self.inner),
45        }
46    }
47}
48
49#[bon::bon]
50impl<C: std::fmt::Debug + MaybeSendSync + 'static> ScheduledRefreshCipher<C> {
51    /// Creates a new [`ScheduledRefreshCipher`] using the given factory and policy parameters.
52    ///
53    /// The factory is called immediately to produce the initial cipher.
54    /// The same factory is called on subsequent refreshes.
55    ///
56    /// # Errors
57    ///
58    /// Returns an error if the initial factory call fails.
59    #[builder]
60    pub async fn new(
61        factory: impl Fn() -> Pin<Box<dyn MaybeSendFuture<Output = Result<C, Error>>>>
62        + MaybeSendSync
63        + 'static,
64        /// The time-to-live for the cached cipher.
65        #[builder(default = Duration::from_hours(1))]
66        ttl: Duration,
67        /// The backoff duration after a failed refresh.
68        #[builder(default = Duration::from_secs(30))]
69        failure_backoff: Duration,
70        /// Minimum time between any two refresh attempts, regardless of outcome.
71        #[builder(default = Duration::from_mins(1))]
72        min_refresh_interval: Duration,
73    ) -> Result<Self, Error> {
74        let inner = ScheduledRefreshable::builder()
75            .factory(factory)
76            .ttl(ttl)
77            .failure_backoff(failure_backoff)
78            .min_refresh_interval(min_refresh_interval)
79            .build()
80            .await?;
81        Ok(Self {
82            inner: Arc::new(inner),
83        })
84    }
85
86    /// Reloads the key if it has outlived its TTL and the rate-limit/backoff
87    /// policy permits; a within-TTL key is left in place. Blocking.
88    ///
89    /// A manual staleness poll: the selector paths
90    /// ([`select_cipher`](AeadCipherSelector::select_cipher),
91    /// [`select_sealer`](AeadSealerSelector::select_sealer)) already reload a stale
92    /// key during selection, so this only pre-warms the key ahead of a
93    /// latency-sensitive encrypt/seal. Distinct from the inbound miss path
94    /// [`AeadDecryptor::try_refresh`], which is not TTL-gated;
95    /// [`refresh`](Self::refresh) reloads unconditionally.
96    ///
97    /// Returns `true` if this call refreshed successfully, `false` if the policy
98    /// blocked it or the refresh failed.
99    pub async fn refresh_if_stale(&self) -> bool {
100        self.inner.refresh_if_stale().await
101    }
102
103    /// Forces a refresh bypassing the scheduling policy, but still records the outcome.
104    ///
105    /// Returns `Ok(true)` if new key material was fetched by this call, or
106    /// `Ok(false)` if another task already refreshed concurrently.
107    ///
108    /// # Errors
109    ///
110    /// Returns an error if the factory call fails.
111    pub async fn refresh(&self) -> Result<bool, Error> {
112        self.inner.refresh().await
113    }
114}
115
116impl<C: AeadEncryptor + 'static> AeadEncryptor for ScheduledRefreshCipher<C> {
117    fn enc_algorithm(&self) -> Cow<'_, str> {
118        Cow::Owned(self.inner.load().enc_algorithm().into_owned())
119    }
120
121    fn key_id(&self) -> Option<Cow<'_, str>> {
122        self.inner
123            .load()
124            .key_id()
125            .map(|kid| Cow::Owned(kid.into_owned()))
126    }
127
128    fn encrypt<'a>(
129        &'a self,
130        plaintext: &'a [u8],
131        aad: &'a [u8],
132    ) -> MaybeSendBoxFuture<'a, Result<AeadOutput, Error>> {
133        Box::pin(async move { self.inner.load_full().encrypt(plaintext, aad).await })
134    }
135}
136
137impl<C: AeadDecryptor + 'static> AeadDecryptor for ScheduledRefreshCipher<C> {
138    fn cipher_match(&self, m: &CipherMatch<'_>) -> Option<KeyMatchStrength> {
139        self.inner.load().cipher_match(m)
140    }
141
142    fn decrypt<'a>(
143        &'a self,
144        cipher_match: Option<&'a CipherMatch<'a>>,
145        nonce: &'a [u8],
146        ciphertext: &'a [u8],
147        tag: &'a [u8],
148        aad: &'a [u8],
149    ) -> MaybeSendBoxFuture<'a, Result<Vec<u8>, DecryptError>> {
150        Box::pin(async move {
151            // Bound staleness on the read path: a decryption key removed upstream
152            // yields no decrypt failure (you still hold it), so only a time bound
153            // retires it. The first decrypt past the TTL reloads (single-flight,
154            // non-blocking for others) before decrypting.
155            self.inner.poll_refresh_ahead().await;
156            self.inner
157                .load_full()
158                .decrypt(cipher_match, nonce, ciphertext, tag, aad)
159                .await
160        })
161    }
162
163    fn try_refresh(&self) -> MaybeSendBoxFuture<'_, bool> {
164        // Miss path (`RetryingDecryptor` / `MultiKeyDecryptor` fan-out): **not**
165        // TTL-gated, so an added key is picked up within the TTL. The inherent
166        // `refresh_if_stale` above is the TTL-gated manual staleness poll — a
167        // deliberately distinct operation.
168        Box::pin(self.inner.try_refresh_on_miss())
169    }
170}
171
172impl<C: AeadEncryptor + 'static> AeadCipherSelector for ScheduledRefreshCipher<C> {
173    fn select_cipher(&self) -> MaybeSendBoxFuture<'_, Arc<dyn AeadEncryptor>> {
174        Box::pin(async move {
175            // Reload a stale key during selection, then hand back the fresh key as
176            // one frozen snapshot — the caller encrypts against exactly this key.
177            self.inner.poll_refresh_ahead().await;
178            let encryptor: Arc<dyn AeadEncryptor> = self.inner.load_full();
179            encryptor
180        })
181    }
182}
183
184impl<C: AeadEncryptor + 'static> AeadSealerSelector for ScheduledRefreshCipher<C> {
185    fn select_sealer(&self) -> MaybeSendBoxFuture<'_, Arc<dyn AeadSealer>> {
186        Box::pin(async move {
187            // Reload on the outbound read path, then frame the frozen snapshot as a
188            // v1 sealer whose metadata and `seal` describe and use the same key.
189            self.inner.poll_refresh_ahead().await;
190            let sealer: Arc<dyn AeadSealer> = Arc::new(AeadV1Cipher::new(self.inner.load_full()));
191            sealer
192        })
193    }
194}
195
196impl<C: AeadDecryptor + 'static> AeadUnsealer for ScheduledRefreshCipher<C> {
197    fn unseal<'a>(
198        &'a self,
199        cipher_match: Option<&'a CipherMatch<'a>>,
200        bundle: &'a [u8],
201        aad: &'a [u8],
202    ) -> MaybeSendBoxFuture<'a, Result<Vec<u8>, DecryptError>> {
203        Box::pin(async move {
204            // Same read-path staleness bound as `decrypt`: reload past the TTL,
205            // then parse and open against the current snapshot.
206            self.inner.poll_refresh_ahead().await;
207            AeadV1Cipher::new(self.inner.load_full())
208                .unseal(cipher_match, bundle, aad)
209                .await
210        })
211    }
212}
213
214#[cfg(test)]
215mod tests {
216    use std::sync::atomic::{AtomicUsize, Ordering};
217
218    use rstest::rstest;
219
220    use super::*;
221
222    /// An identity cipher (ciphertext == plaintext) that reports a fixed `kid`
223    /// and stamps that `kid` into the authentication tag, so the generation that
224    /// actually performed an operation is observable in both its metadata and its
225    /// output.
226    #[derive(Debug)]
227    struct FakeCipher {
228        kid: String,
229    }
230
231    impl AeadEncryptor for FakeCipher {
232        fn enc_algorithm(&self) -> Cow<'_, str> {
233            "A256GCM".into()
234        }
235        fn key_id(&self) -> Option<Cow<'_, str>> {
236            Some(Cow::Borrowed(&self.kid))
237        }
238        fn encrypt<'a>(
239            &'a self,
240            plaintext: &'a [u8],
241            _aad: &'a [u8],
242        ) -> MaybeSendBoxFuture<'a, Result<AeadOutput, Error>> {
243            let ciphertext = plaintext.to_vec();
244            let tag = self.kid.clone().into_bytes();
245            Box::pin(async move {
246                Ok(AeadOutput {
247                    nonce: Vec::new(),
248                    ciphertext,
249                    tag,
250                })
251            })
252        }
253    }
254
255    impl AeadDecryptor for FakeCipher {
256        fn cipher_match(&self, _m: &CipherMatch<'_>) -> Option<KeyMatchStrength> {
257            Some(KeyMatchStrength::ByAlgorithm)
258        }
259        fn decrypt<'a>(
260            &'a self,
261            _cipher_match: Option<&'a CipherMatch<'a>>,
262            _nonce: &'a [u8],
263            ciphertext: &'a [u8],
264            _tag: &'a [u8],
265            _aad: &'a [u8],
266        ) -> MaybeSendBoxFuture<'a, Result<Vec<u8>, DecryptError>> {
267            let plaintext = ciphertext.to_vec();
268            Box::pin(async move { Ok(plaintext) })
269        }
270    }
271
272    /// A generational cipher (`gen-0`, `gen-1`, …) with the given policy.
273    async fn generational_cipher(
274        ttl: Duration,
275        min_refresh_interval: Duration,
276    ) -> ScheduledRefreshCipher<FakeCipher> {
277        let counter = Arc::new(AtomicUsize::new(0));
278        ScheduledRefreshCipher::builder()
279            .factory(move || {
280                let n = counter.fetch_add(1, Ordering::SeqCst);
281                Box::pin(async move {
282                    Ok(FakeCipher {
283                        kid: format!("gen-{n}"),
284                    })
285                })
286            })
287            .ttl(ttl)
288            .min_refresh_interval(min_refresh_interval)
289            .build()
290            .await
291            .unwrap()
292    }
293
294    fn current_kid(cipher: &ScheduledRefreshCipher<FakeCipher>) -> Option<String> {
295        cipher.key_id().map(std::borrow::Cow::into_owned)
296    }
297
298    #[tokio::test]
299    async fn delegates_encryptor_metadata_to_inner() {
300        let cipher = generational_cipher(Duration::from_hours(1), Duration::from_secs(0)).await;
301        assert_eq!(cipher.enc_algorithm().as_ref(), "A256GCM");
302        assert_eq!(current_kid(&cipher).as_deref(), Some("gen-0"));
303    }
304
305    #[tokio::test]
306    async fn encrypt_and_decrypt_delegate_to_inner() {
307        let cipher = generational_cipher(Duration::from_hours(1), Duration::from_secs(0)).await;
308        let out = cipher.encrypt(b"plaintext", b"aad").await.unwrap();
309        let recovered = cipher
310            .decrypt(None, &out.nonce, &out.ciphertext, &out.tag, b"aad")
311            .await
312            .unwrap();
313        assert_eq!(recovered, b"plaintext");
314    }
315
316    #[tokio::test]
317    async fn cipher_match_delegates_to_inner() {
318        let cipher = generational_cipher(Duration::from_hours(1), Duration::from_secs(0)).await;
319        let m = CipherMatch {
320            enc: Some("A256GCM"),
321            kid: None,
322        };
323        assert_eq!(cipher.cipher_match(&m), Some(KeyMatchStrength::ByAlgorithm));
324    }
325
326    #[tokio::test]
327    async fn forced_refresh_bypasses_policy() {
328        // A long TTL would block a policy-gated refresh, but `refresh` is forced.
329        let cipher = generational_cipher(Duration::from_hours(1), Duration::from_mins(1)).await;
330        assert_eq!(current_kid(&cipher).as_deref(), Some("gen-0"));
331        assert!(cipher.refresh().await.unwrap());
332        assert_eq!(current_kid(&cipher).as_deref(), Some("gen-1"));
333    }
334
335    /// `refresh_if_stale` is gated by the TTL: a fresh value is left in place, a
336    /// stale one (zero TTL) is swapped for the next generation.
337    #[rstest]
338    #[case::blocked_while_fresh(Duration::from_hours(1), false, "gen-0")]
339    #[case::allowed_when_stale(Duration::from_secs(0), true, "gen-1")]
340    #[tokio::test]
341    async fn refresh_if_stale_respects_ttl_policy(
342        #[case] ttl: Duration,
343        #[case] expected_refreshed: bool,
344        #[case] expected_kid: &str,
345    ) {
346        let cipher = generational_cipher(ttl, Duration::from_secs(0)).await;
347        assert_eq!(cipher.refresh_if_stale().await, expected_refreshed);
348        assert_eq!(current_kid(&cipher).as_deref(), Some(expected_kid));
349    }
350
351    /// The inbound miss path (`AeadDecryptor::try_refresh`) is **not** TTL-gated:
352    /// even a fresh keyset reloads, so a `RetryingDecryptor` picks up an added key
353    /// within the TTL. Its TTL-gated sibling is the inherent `refresh_if_stale`.
354    #[tokio::test]
355    async fn decryptor_try_refresh_is_not_ttl_gated() {
356        // A large TTL — the value is nowhere near stale — yet the miss path still
357        // reloads, swapping gen-0 for gen-1.
358        let cipher = generational_cipher(Duration::from_hours(1), Duration::from_secs(0)).await;
359        assert!(AeadDecryptor::try_refresh(&cipher).await);
360        assert_eq!(current_kid(&cipher).as_deref(), Some("gen-1"));
361    }
362
363    /// A decryptor whose generation 0 decrypts (holds the key) and every later
364    /// generation rejects (the key has been retired upstream).
365    #[derive(Debug)]
366    struct RetiringDecryptor {
367        holds_key: bool,
368    }
369
370    impl AeadDecryptor for RetiringDecryptor {
371        fn cipher_match(&self, _m: &CipherMatch<'_>) -> Option<KeyMatchStrength> {
372            self.holds_key.then_some(KeyMatchStrength::ByAlgorithm)
373        }
374
375        fn decrypt<'a>(
376            &'a self,
377            _cipher_match: Option<&'a CipherMatch<'a>>,
378            _nonce: &'a [u8],
379            ciphertext: &'a [u8],
380            _tag: &'a [u8],
381            _aad: &'a [u8],
382        ) -> MaybeSendBoxFuture<'a, Result<Vec<u8>, DecryptError>> {
383            let holds_key = self.holds_key;
384            let plaintext = ciphertext.to_vec();
385            Box::pin(async move {
386                if holds_key {
387                    Ok(plaintext)
388                } else {
389                    Err(DecryptError::NoMatchingKey)
390                }
391            })
392        }
393    }
394
395    #[allow(clippy::type_complexity)]
396    fn retiring_decryptor() -> (
397        Arc<AtomicUsize>,
398        impl Fn() -> Pin<Box<dyn MaybeSendFuture<Output = Result<RetiringDecryptor, Error>>>>
399        + MaybeSendSync
400        + 'static,
401    ) {
402        let builds = Arc::new(AtomicUsize::new(0));
403        let counter = builds.clone();
404        let factory = move || {
405            let n = counter.fetch_add(1, Ordering::SeqCst);
406            Box::pin(async move { Ok(RetiringDecryptor { holds_key: n == 0 }) })
407                as Pin<Box<dyn MaybeSendFuture<Output = Result<RetiringDecryptor, Error>>>>
408        };
409        (builds, factory)
410    }
411
412    #[tokio::test]
413    async fn stale_keyset_reloads_before_decrypt_and_retires_the_key() {
414        // gen-0 would decrypt, but the keyset is past its TTL, so the read-path
415        // reload swaps in gen-1 (key retired) before decrypting — no decrypt
416        // failure was ever needed to trigger it.
417        let (builds, factory) = retiring_decryptor();
418        let cipher: ScheduledRefreshCipher<RetiringDecryptor> = ScheduledRefreshCipher::builder()
419            .factory(factory)
420            .ttl(Duration::ZERO)
421            .min_refresh_interval(Duration::ZERO)
422            .build()
423            .await
424            .unwrap();
425
426        cipher
427            .decrypt(None, b"", b"data", b"", b"")
428            .await
429            .expect_err("the retired decryption key is dropped by the staleness reload");
430        assert_eq!(
431            builds.load(Ordering::SeqCst),
432            2,
433            "initial build + one reload"
434        );
435    }
436
437    #[tokio::test]
438    async fn fresh_keyset_decrypts_without_reload() {
439        let (builds, factory) = retiring_decryptor();
440        let cipher: ScheduledRefreshCipher<RetiringDecryptor> = ScheduledRefreshCipher::builder()
441            .factory(factory)
442            .ttl(Duration::from_hours(1))
443            .build()
444            .await
445            .unwrap();
446
447        let recovered = cipher.decrypt(None, b"", b"data", b"", b"").await.unwrap();
448        assert_eq!(recovered, b"data");
449        assert_eq!(builds.load(Ordering::SeqCst), 1, "no reload while fresh");
450    }
451
452    #[tokio::test]
453    async fn select_cipher_hands_back_the_current_snapshot() {
454        let cipher = generational_cipher(Duration::from_hours(1), Duration::from_secs(0)).await;
455        assert_eq!(
456            cipher.select_cipher().await.key_id().as_deref(),
457            Some("gen-0")
458        );
459    }
460
461    #[tokio::test]
462    async fn select_cipher_reloads_a_stale_key_during_selection() {
463        // Zero TTL: the outbound read path reloads before handing back the snapshot.
464        let cipher = generational_cipher(Duration::ZERO, Duration::ZERO).await;
465        assert_eq!(
466            cipher.select_cipher().await.key_id().as_deref(),
467            Some("gen-1")
468        );
469    }
470
471    /// The crux of the sealer-selector: a sealer is a *frozen* snapshot, so a
472    /// rotation landing after selection cannot make its `key_id` and its `seal`
473    /// describe different keys. `FakeCipher` tags every bundle with its `kid`, so
474    /// the key that actually sealed is observable in the ciphertext.
475    #[tokio::test]
476    async fn selected_sealer_kid_and_ciphertext_agree_across_a_rotation() {
477        let cipher = generational_cipher(Duration::from_hours(1), Duration::ZERO).await;
478
479        let sealer = cipher.select_sealer().await;
480        let kid = sealer.key_id().map(std::borrow::Cow::into_owned);
481        assert_eq!(kid.as_deref(), Some("gen-0"));
482
483        // Rotate the underlying key out from under the held sealer.
484        assert!(cipher.refresh().await.unwrap());
485        assert_eq!(current_kid(&cipher).as_deref(), Some("gen-1"));
486
487        // The frozen sealer still seals with gen-0 — the key its `kid` named —
488        // not the freshly rotated gen-1.
489        let bundle = sealer.seal(b"payload", b"aad").await.unwrap();
490        // v1 layout: [0x01, nonce_len=0, tag_len=5, ciphertext="payload", tag="gen-0"]
491        let tag = &bundle[bundle.len() - 5..];
492        assert_eq!(tag, b"gen-0");
493    }
494}
huskarl_core/crypto/cipher/scheduled.rs

huskarl_core/crypto/cipher/
scheduled.rs
