macula_rust/
statement_issuer.rs

1//! A client's status statement issuer, the client side of macula's
2//! macula_statement_issuer (D22), as macula-go's StatementIssuer. It holds the
3//! identity key, the node's CONNECT bindings with the newest status statement
4//! for each, and the current CONNECT key. Each tick issues a statement valid
5//! for an hour for each binding whose not_after has not passed, and hands it
6//! to that binding's subscribers, the links that connected with it. Every 5
7//! days it rotates the CONNECT key: the new key's binding and statement exist
8//! before [`StatementIssuer::connect_material`] hands the key out, and the
9//! rotated-out binding keeps its statements until its not_after.
10//! `connect_material` does work that is due itself, so a dial after missed
11//! ticks, a sleep or a clock step still carries material in force. Nothing is
12//! written to disk, so a new issuer starts with a new CONNECT key.
13
14use std::collections::HashMap;
15use std::fmt;
16use std::sync::{Arc, Mutex, Weak};
17
18use sha2::{Digest, Sha384};
19use tokio::sync::Notify;
20
21use crate::binding::{connect_binding, status_statement, BindingError, SignedTbs};
22use crate::node_key::{KeyError, NodeKey, Purpose};
23
24/// How often statements are reissued.
25pub const STATEMENT_EVERY_MS: i64 = 15 * 60 * 1000;
26/// How long a status statement is valid.
27pub const STATEMENT_VALID_MS: i64 = 60 * 60 * 1000;
28/// How long a CONNECT binding is valid.
29pub const CONNECT_BINDING_VALID_MS: i64 = 7 * 24 * 60 * 60 * 1000;
30/// How often the CONNECT key rotates.
31pub const CONNECT_ROTATE_EVERY_MS: i64 = 5 * 24 * 60 * 60 * 1000;
32/// How long before the current binding's not_after a failed rotation is
33/// reported as overdue.
34pub const ROTATION_MARGIN_MS: i64 = 24 * 60 * 60 * 1000;
35
36const TOLERANCE_MS: i64 = 5 * 60 * 1000;
37
38/// Why the issuer could not do what was asked.
39#[derive(Debug, Clone, PartialEq, Eq)]
40pub enum IssuerError {
41    /// The key given is not an identity key.
42    NotAnIdentityKey,
43    /// A subscription to a binding the issuer does not hold, or whose
44    /// not_after has passed.
45    UnknownBinding,
46    /// No CONNECT binding and status statement in force, because the work
47    /// that renews them failed.
48    NoConnectMaterial(String),
49    /// A rotation failed while the current binding expires within the
50    /// rotation margin.
51    RotationOverdue { failures: u64, left_ms: i64 },
52    /// A key could not be made or could not sign.
53    Key(KeyError),
54    /// A binding or statement could not be issued.
55    Binding(BindingError),
56}
57
58impl fmt::Display for IssuerError {
59    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
60        match self {
61            IssuerError::NotAnIdentityKey => f.write_str("the issuer needs an identity key"),
62            IssuerError::UnknownBinding => f.write_str("no binding in force has that hash"),
63            IssuerError::NoConnectMaterial(why) => write!(f, "no CONNECT binding and status statement in force: {why}"),
64            IssuerError::RotationOverdue { failures, left_ms } => write!(
65                f,
66                "the CONNECT key has not rotated: {failures} failed rotations, {left_ms} ms left on its binding"
67            ),
68            IssuerError::Key(e) => write!(f, "{e}"),
69            IssuerError::Binding(e) => write!(f, "{e}"),
70        }
71    }
72}
73
74impl std::error::Error for IssuerError {}
75
76impl From<KeyError> for IssuerError {
77    fn from(e: KeyError) -> Self {
78        IssuerError::Key(e)
79    }
80}
81
82impl From<BindingError> for IssuerError {
83    fn from(e: BindingError) -> Self {
84        IssuerError::Binding(e)
85    }
86}
87
88/// What a new dial carries: the CONNECT key, its binding, and a status
89/// statement for that binding.
90#[derive(Debug, Clone)]
91pub struct ConnectMaterial {
92    pub key: Arc<NodeKey>,
93    pub binding: SignedTbs,
94    pub status: SignedTbs,
95}
96
97/// A clock in Unix milliseconds.
98pub type Clock = Box<dyn Fn() -> i64 + Send + Sync>;
99
100/// A binding the issuer holds, with its newest statement, and its key while
101/// it is the current binding.
102struct StatedBinding {
103    key: Option<Arc<NodeKey>>,
104    binding: SignedTbs,
105    statement: SignedTbs,
106    bound_at: i64,
107    stated_at: i64,
108    not_after: i64,
109}
110
111impl StatedBinding {
112    fn rotation_due(&self, now: i64) -> bool {
113        now < self.bound_at - TOLERANCE_MS || now >= self.bound_at + CONNECT_ROTATE_EVERY_MS
114    }
115
116    fn restatement_due(&self, now: i64) -> bool {
117        now < self.stated_at - TOLERANCE_MS || now >= self.stated_at + STATEMENT_EVERY_MS
118    }
119
120    fn in_force(&self, now: i64) -> bool {
121        self.bound_at - TOLERANCE_MS <= now
122            && now <= self.not_after
123            && self.stated_at - TOLERANCE_MS <= now
124            && now < self.stated_at + STATEMENT_VALID_MS
125    }
126}
127
128/// A subscription's slot: at most the newest statement, whether it closed,
129/// and the waker of a reader waiting for one.
130struct Slot {
131    newest: Mutex<(Option<SignedTbs>, bool)>,
132    notify: Notify,
133}
134
135struct State {
136    identity: Arc<NodeKey>,
137    clock: Clock,
138    current: [u8; 48],
139    bindings: HashMap<[u8; 48], StatedBinding>,
140    subscribers: HashMap<[u8; 48], Vec<Arc<Slot>>>,
141    rotation_failures: u64,
142}
143
144/// A client's statement issuer, shared by every link of one node.
145#[derive(Clone)]
146pub struct StatementIssuer {
147    state: Arc<Mutex<State>>,
148}
149
150impl StatementIssuer {
151    /// An issuer for `identity`, reading the time from `clock`. It starts with
152    /// a new CONNECT key, bound and stated.
153    pub fn new(identity: Arc<NodeKey>, clock: Clock) -> Result<StatementIssuer, IssuerError> {
154        if identity.purpose() != Purpose::Identity {
155            return Err(IssuerError::NotAnIdentityKey);
156        }
157        let now = clock();
158        let mut state = State {
159            identity,
160            clock,
161            current: [0; 48],
162            bindings: HashMap::new(),
163            subscribers: HashMap::new(),
164            rotation_failures: 0,
165        };
166        state.rotate_connect(now)?;
167        Ok(StatementIssuer {
168            state: Arc::new(Mutex::new(state)),
169        })
170    }
171
172    /// An issuer on the wall clock.
173    pub fn with_wall_clock(identity: Arc<NodeKey>) -> Result<StatementIssuer, IssuerError> {
174        StatementIssuer::new(identity, Box::new(|| crate::uuid_v7::now_ms() as i64))
175    }
176
177    /// The current CONNECT key with its binding and a statement for it, both
178    /// in force at the clock's time. Work that is due is done first.
179    pub fn connect_material(&self) -> Result<ConnectMaterial, IssuerError> {
180        let mut state = self.lock();
181        let now = (state.clock)();
182        let mut work = Ok(());
183        let due = state
184            .bindings
185            .get(&state.current)
186            .is_some_and(|b| b.rotation_due(now) || b.restatement_due(now));
187        if due {
188            work = state.tick(now);
189        }
190        let current = &state.bindings[&state.current];
191        match &current.key {
192            Some(key) if current.in_force(now) => Ok(ConnectMaterial {
193                key: key.clone(),
194                binding: current.binding.clone(),
195                status: current.statement.clone(),
196            }),
197            _ => Err(IssuerError::NoConnectMaterial(match work {
198                Err(e) => e.to_string(),
199                Ok(()) => "the current binding is out of force".into(),
200            })),
201        }
202    }
203
204    /// How many rotations have failed since the last that succeeded.
205    pub fn rotation_failures(&self) -> u64 {
206        self.lock().rotation_failures
207    }
208
209    /// Hands over the newest statement for `binding` at every reissue. The
210    /// subscription closes once a tick finds the binding's not_after passed,
211    /// and unsubscribes when dropped.
212    pub fn subscribe(&self, binding: &SignedTbs) -> Result<StatementSubscription, IssuerError> {
213        let hash: [u8; 48] = Sha384::digest(&binding.tbs).into();
214        let mut state = self.lock();
215        let now = (state.clock)();
216        match state.bindings.get(&hash) {
217            Some(held) if held.not_after >= now => {}
218            _ => return Err(IssuerError::UnknownBinding),
219        }
220        let slot = Arc::new(Slot {
221            newest: Mutex::new((None, false)),
222            notify: Notify::new(),
223        });
224        state
225            .subscribers
226            .entry(hash)
227            .or_default()
228            .push(slot.clone());
229        Ok(StatementSubscription {
230            slot,
231            hash,
232            issuer: Arc::downgrade(&self.state),
233        })
234    }
235
236    /// The periodic work at the clock's time: a statement for each binding in
237    /// force, a rotation when one is due, and the expired bindings let go.
238    pub fn tick(&self) -> Result<(), IssuerError> {
239        let mut state = self.lock();
240        let now = (state.clock)();
241        state.tick(now)
242    }
243
244    /// Ticks every 15 minutes until every handle to the issuer is dropped,
245    /// handing a failed tick's error to `on_error`.
246    pub fn spawn_ticks(
247        &self,
248        on_error: impl Fn(IssuerError) + Send + 'static,
249    ) -> tokio::task::JoinHandle<()> {
250        let weak = Arc::downgrade(&self.state);
251        tokio::spawn(async move {
252            let mut ticks =
253                tokio::time::interval(std::time::Duration::from_millis(STATEMENT_EVERY_MS as u64));
254            ticks.tick().await;
255            loop {
256                ticks.tick().await;
257                let Some(state) = weak.upgrade() else { return };
258                let issuer = StatementIssuer { state };
259                if let Err(e) = issuer.tick() {
260                    on_error(e);
261                }
262            }
263        })
264    }
265
266    fn lock(&self) -> std::sync::MutexGuard<'_, State> {
267        self.state
268            .lock()
269            .unwrap_or_else(|poisoned| poisoned.into_inner())
270    }
271}
272
273impl State {
274    fn tick(&mut self, now: i64) -> Result<(), IssuerError> {
275        let reissued = self.reissue(now);
276        let mut rotated = Ok(());
277        if self
278            .bindings
279            .get(&self.current)
280            .is_some_and(|b| b.rotation_due(now))
281        {
282            rotated = self.rotate(now);
283        }
284        self.drop_expired(now);
285        reissued.and(rotated)
286    }
287
288    fn rotate(&mut self, now: i64) -> Result<(), IssuerError> {
289        match self.rotate_connect(now) {
290            Ok(()) => {
291                self.rotation_failures = 0;
292                Ok(())
293            }
294            Err(e) => {
295                self.rotation_failures += 1;
296                let left = self
297                    .bindings
298                    .get(&self.current)
299                    .map_or(0, |b| b.not_after - now);
300                if left < ROTATION_MARGIN_MS {
301                    return Err(IssuerError::RotationOverdue {
302                        failures: self.rotation_failures,
303                        left_ms: left,
304                    });
305                }
306                Err(e)
307            }
308        }
309    }
310
311    fn reissue(&mut self, now: i64) -> Result<(), IssuerError> {
312        let mut first_error = Ok(());
313        for (hash, held) in self.bindings.iter_mut() {
314            if held.not_after < now {
315                continue;
316            }
317            match status_statement(&self.identity, &held.binding, now, now + STATEMENT_VALID_MS) {
318                Ok(statement) => {
319                    held.statement = statement.clone();
320                    held.stated_at = now;
321                    for slot in self.subscribers.get(hash).into_iter().flatten() {
322                        deliver(slot, statement.clone());
323                    }
324                }
325                Err(e) => {
326                    if first_error.is_ok() {
327                        first_error = Err(e.into());
328                    }
329                }
330            }
331        }
332        first_error
333    }
334
335    fn drop_expired(&mut self, now: i64) {
336        let expired: Vec<[u8; 48]> = self
337            .bindings
338            .iter()
339            .filter(|(_, b)| b.not_after < now)
340            .map(|(h, _)| *h)
341            .collect();
342        for hash in expired {
343            for slot in self.subscribers.remove(&hash).into_iter().flatten() {
344                close(&slot);
345            }
346            if hash != self.current {
347                self.bindings.remove(&hash);
348            }
349        }
350    }
351
352    fn rotate_connect(&mut self, now: i64) -> Result<(), IssuerError> {
353        let key = NodeKey::generate(Purpose::Connect, self.identity.profile())?;
354        let not_after = now + CONNECT_BINDING_VALID_MS;
355        let binding = connect_binding(&self.identity, &key.public_key(), now, not_after)?;
356        let statement = status_statement(&self.identity, &binding, now, now + STATEMENT_VALID_MS)?;
357        if let Some(previous) = self.bindings.get_mut(&self.current) {
358            previous.key = None;
359        }
360        let hash: [u8; 48] = Sha384::digest(&binding.tbs).into();
361        self.bindings.insert(
362            hash,
363            StatedBinding {
364                key: Some(Arc::new(key)),
365                binding,
366                statement,
367                bound_at: now,
368                stated_at: now,
369                not_after,
370            },
371        );
372        self.current = hash;
373        Ok(())
374    }
375}
376
377fn deliver(slot: &Slot, statement: SignedTbs) {
378    let mut newest = slot.newest.lock().unwrap_or_else(|p| p.into_inner());
379    newest.0 = Some(statement);
380    drop(newest);
381    slot.notify.notify_one();
382}
383
384fn close(slot: &Slot) {
385    slot.newest.lock().unwrap_or_else(|p| p.into_inner()).1 = true;
386    slot.notify.notify_one();
387}
388
389/// Why a subscription handed over no statement.
390#[derive(Debug, Clone, Copy, PartialEq, Eq)]
391pub enum SubscriptionEmpty {
392    /// None has been issued since the last taken.
393    Empty,
394    /// The subscription has closed.
395    Closed,
396}
397
398/// A subscription to one binding's statements, holding at most the newest.
399pub struct StatementSubscription {
400    slot: Arc<Slot>,
401    hash: [u8; 48],
402    issuer: Weak<Mutex<State>>,
403}
404
405impl StatementSubscription {
406    /// The newest statement not yet taken.
407    pub fn try_recv(&mut self) -> Result<SignedTbs, SubscriptionEmpty> {
408        let mut newest = self.slot.newest.lock().unwrap_or_else(|p| p.into_inner());
409        match newest.0.take() {
410            Some(statement) => Ok(statement),
411            None if newest.1 => Err(SubscriptionEmpty::Closed),
412            None => Err(SubscriptionEmpty::Empty),
413        }
414    }
415
416    /// The next statement, or `None` once the subscription has closed.
417    pub async fn recv(&mut self) -> Option<SignedTbs> {
418        let slot = self.slot.clone();
419        loop {
420            let notified = slot.notify.notified();
421            match self.try_recv() {
422                Ok(statement) => return Some(statement),
423                Err(SubscriptionEmpty::Closed) => return None,
424                Err(SubscriptionEmpty::Empty) => notified.await,
425            }
426        }
427    }
428}
429
430impl Drop for StatementSubscription {
431    fn drop(&mut self) {
432        let Some(state) = self.issuer.upgrade() else {
433            return;
434        };
435        let mut state = state.lock().unwrap_or_else(|p| p.into_inner());
436        if let Some(slots) = state.subscribers.get_mut(&self.hash) {
437            slots.retain(|s| !Arc::ptr_eq(s, &self.slot));
438        }
439    }
440}
macula_rust/statement_issuer.rs

macula_rust/
statement_issuer.rs
