orbit_auth/
fleet.rs

1use std::num::NonZeroUsize;
2use std::sync::{Arc, Mutex};
3
4use hmac::{Hmac, Mac};
5use lru::LruCache;
6use orbit_core::Fleet;
7use serde::{Deserialize, Serialize};
8use sha2::Sha256;
9use zeroize::Zeroizing;
10
11use crate::keyring::{context, derive};
12use crate::table::{
13    CONSUMED, REVOKED, Record, RecordProtection, SESSION, Snapshot, StoreKey, Table, find, insert
14};
15use crate::{
16    Authority, CacheKey, CachedPrincipal, Claims, Error, MemoryCache, Principal, PrincipalCache,
17    Purpose, Result, SecretKey, StateProtection, ValidationHook, Validator
18};
19
20/// Random session identity. It is not a bearer credential.
21#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)]
22pub struct SessionId(pub [u8; 16]);
23
24#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
25#[serde(deny_unknown_fields)]
26pub(crate) struct SessionBinding {
27    pub(crate) id: SessionId,
28    generation: u64
29}
30
31/// Issuer-side handle, never serialized into SHM. Live validation checks state
32/// even when a caller retained a handle predating refresh or revocation.
33pub struct Session {
34    binding: SessionBinding,
35    subject: String,
36    created_at: u64,
37    expires_at: u64,
38    realm: String,
39    issuer: String
40}
41impl Session {
42    pub fn id(&self) -> SessionId {
43        self.binding.id
44    }
45    pub fn subject(&self) -> &str {
46        &self.subject
47    }
48    pub fn created_at(&self) -> u64 {
49        self.created_at
50    }
51    pub fn expires_at(&self) -> u64 {
52        self.expires_at
53    }
54    pub(crate) fn binding(&self) -> SessionBinding {
55        self.binding.clone()
56    }
57    pub(crate) fn check_issuance(
58        &self,
59        authority: &Authority,
60        claims: &Claims
61    ) -> Result<()> {
62        if self.realm != authority.realm
63            || self.issuer != authority.issuer
64            || claims.subject != self.subject
65            || claims.issued_at < self.created_at
66            || claims.expires_at > self.expires_at
67        {
68            return Err(Error::InvalidInput);
69        }
70        Ok(())
71    }
72}
73
74struct SessionView {
75    snapshot: Snapshot,
76    // Compared once against the shared keyed subject tag, then checked locally.
77    subject: String
78}
79
80/// Local decoded Principals plus fleet-native live sessions and replay state.
81/// Backing follows the supplied Fleet. SHM survives handle drop and restart;
82/// memory state lives as long as a handle retains it. No background driver.
83///
84/// Shared records contain only keyed identifiers and fixed control metadata.
85/// Owner-only OS permissions restrict access. `Blake3State` verifies
86/// integrity; `EncryptedState` also hides metadata. `UnprotectedState` explicitly
87/// trusts every writer. Protected modes still permit deletion/rollback attacks.
88/// `state_key` stays stable across token-key rotations and is never stored in SHM.
89pub struct FleetAuth<P: StateProtection> {
90    _fleet: Arc<Fleet>,
91    table: Table,
92    local: MemoryCache,
93    sessions: Mutex<LruCache<SessionId, SessionView>>,
94    state_key: Zeroizing<[u8; 32]>,
95    protection: RecordProtection<P>,
96    realm: String,
97    issuer: String
98}
99impl<P: StateProtection> FleetAuth<P> {
100    /// Construct with an explicitly selected shared-state protection policy.
101    /// There is no default policy.
102    ///
103    /// `local_capacity` bounds each local Principal/session-view LRU. A cached
104    /// protected view is reused only while every shared byte matches its local copy.
105    /// UnprotectedState trusts the commit revision instead.
106    /// Peers sharing a physical table must select the same policy. Token crypto
107    /// and live session/replay checks are enabled for every policy.
108    pub fn new(
109        fleet: Arc<Fleet>,
110        authority: &Authority,
111        state_key: SecretKey,
112        local_capacity: NonZeroUsize,
113        _policy: P
114    ) -> Result<Self> {
115        let info = context(&[
116            b"orbit-auth/v2/fleet-state",
117            fleet.name().as_bytes(),
118            authority.realm.as_bytes(),
119            authority.issuer.as_bytes()
120        ]);
121        Ok(Self {
122            table: Table::new(fleet.clone(), P::ID)?,
123            // The physical table may contain multiple realm/issuer namespaces.
124            // Its protection key is fleet-scoped; record lookup identifiers below bind
125            // the realm/issuer. All users of one table share this state secret.
126            protection: RecordProtection::new(derive(
127                &state_key,
128                &context(&[
129                    b"orbit-auth/v4/fleet-state-protection",
130                    fleet.name().as_bytes(),
131                    &P::ID.to_le_bytes()
132                ])
133            )),
134            _fleet: fleet,
135            local: MemoryCache::new(local_capacity),
136            sessions: Mutex::new(LruCache::new(local_capacity)),
137            state_key: derive(&state_key, &info),
138            realm: authority.realm.clone(),
139            issuer: authority.issuer.clone()
140        })
141    }
142
143    /// Clear this handle's decoded/validated views; shared authority is preserved.
144    pub fn clear_local(&self) {
145        self.local.clear();
146        self.sessions.lock().unwrap_or_else(|e| e.into_inner()).clear();
147    }
148
149    pub fn create_session(
150        &self,
151        subject: impl Into<String>,
152        now: u64,
153        expires_at: u64
154    ) -> Result<Session> {
155        let subject = subject.into();
156        if !crate::principal::valid_label(&subject) || now >= expires_at {
157            return Err(Error::InvalidInput);
158        }
159        let mut id = [0; 16];
160        getrandom::fill(&mut id).map_err(|_| Error::Randomness)?;
161        let id = SessionId(id);
162        let record = Record {
163            key: self.session_key(id),
164            subject: self.subject_tag(&subject),
165            created_at: now,
166            expires_at,
167            generation: 0,
168            status: SESSION
169        };
170        self.table.write(|slots| {
171            if find(slots, record.key, &self.protection)?.is_some() {
172                return Err(Error::InvalidInput);
173            }
174            insert(slots, record, now, &self.protection)
175        })?;
176        Ok(self.handle(id, subject, record))
177    }
178
179    /// Atomically publish revocation. Every subsequent validation rechecks the
180    /// authenticated record, including warm local cache hits. Already admitted work is
181    /// not cancelled; no notification or polling interval delays the next check.
182    pub fn revoke_session(
183        &self,
184        id: SessionId,
185        now: u64
186    ) -> Result<()> {
187        let key = self.session_key(id);
188        self.table.write(|slots| {
189            let snapshot = find(slots, key, &self.protection)?.ok_or(Error::Revoked)?;
190            let mut record = snapshot.record;
191            if !matches!(record.status, SESSION | REVOKED) {
192                return Err(Error::PolicyUnavailable);
193            }
194            if now < record.created_at {
195                return Err(Error::NotYetValid);
196            }
197            record.status = REVOKED;
198            slots[snapshot.index].store(record, snapshot.index, &self.protection)?;
199            Ok(())
200        })
201    }
202
203    /// The live session behind an id, for the subject that owns it: a handle
204    /// a worker that did not create the session uses to issue its credentials
205    /// or read its horizon. Lock-free; a raced read takes one locked retry.
206    pub fn session(
207        &self,
208        id: SessionId,
209        subject: &str,
210        now: u64
211    ) -> Result<Session> {
212        if !crate::principal::valid_label(subject) {
213            return Err(Error::InvalidInput);
214        }
215        let snapshot =
216            self.table.lookup(self.session_key(id), &self.protection)?.ok_or(Error::Revoked)?;
217        if snapshot.record.subject != self.subject_tag(subject) {
218            return Err(Error::InvalidInput);
219        }
220        self.check_live(snapshot.record, now)?;
221        Ok(self.handle(id, subject.to_owned(), snapshot.record))
222    }
223
224    /// Move a live session's horizon forward on activity the application
225    /// trusts, without a credential and without consuming a refresh generation.
226    /// `subject` must own the session. The horizon never moves back: shortening
227    /// a session is revocation. Credentials issued earlier keep their own
228    /// expiry; the returned handle issues ones that reach the new horizon.
229    pub fn extend_session(
230        &self,
231        id: SessionId,
232        subject: &str,
233        now: u64,
234        expires_at: u64
235    ) -> Result<Session> {
236        if !crate::principal::valid_label(subject) || now >= expires_at {
237            return Err(Error::InvalidInput);
238        }
239        let key = self.session_key(id);
240        let tag = self.subject_tag(subject);
241        self.table.write(|slots| {
242            let snapshot = find(slots, key, &self.protection)?.ok_or(Error::Revoked)?;
243            let mut record = snapshot.record;
244            if record.subject != tag {
245                return Err(Error::InvalidInput);
246            }
247            self.check_live(record, now)?;
248            if expires_at < record.expires_at {
249                return Err(Error::InvalidInput);
250            }
251            if expires_at > record.expires_at {
252                record.expires_at = expires_at;
253                slots[snapshot.index].store(record, snapshot.index, &self.protection)?;
254            }
255            Ok(self.handle(id, subject.to_owned(), record))
256        })
257    }
258
259    /// Validate and consume the current refresh generation atomically. Exactly
260    /// one concurrent caller succeeds. The explicit renewed expiry may preserve
261    /// or extend the previous horizon. The returned handle issues new credentials.
262    /// The application owns renewal/delegation limits. If a response is lost,
263    /// the consumed credential is not retried; reauthentication is required.
264    pub fn refresh(
265        &self,
266        validator: &Validator<'_>,
267        token: &str,
268        now: u64,
269        expires_at: u64
270    ) -> Result<Session> {
271        if validator.validation.purpose != Purpose::Refresh {
272            return Err(Error::InvalidInput);
273        }
274        let principal = validator.validate(token, now, self)?;
275        let binding = principal.session.as_ref().ok_or(Error::InvalidToken)?;
276        let key = self.session_key(binding.id);
277        let subject = self.subject_tag(principal.subject());
278        self.table.write(|slots| {
279            let snapshot = find(slots, key, &self.protection)?.ok_or(Error::Revoked)?;
280            let mut record = snapshot.record;
281            self.check_record(record, subject, &principal, now)?;
282            if expires_at < record.expires_at {
283                return Err(Error::InvalidInput);
284            }
285            record.generation = record.generation.checked_add(1).ok_or(Error::PolicyUnavailable)?;
286            record.expires_at = expires_at;
287            slots[snapshot.index].store(record, snapshot.index, &self.protection)?;
288            Ok(self.handle(binding.id, principal.subject().to_owned(), record))
289        })
290    }
291
292    /// One-use hook. Unexpired consumed IDs never evict; full/unavailable denies.
293    pub fn replay_guard(&self) -> ReplayGuard<'_, P> {
294        ReplayGuard(self)
295    }
296
297    fn handle(
298        &self,
299        id: SessionId,
300        subject: String,
301        record: Record
302    ) -> Session {
303        Session {
304            binding: SessionBinding { id, generation: record.generation },
305            subject,
306            created_at: record.created_at,
307            expires_at: record.expires_at,
308            realm: self.realm.clone(),
309            issuer: self.issuer.clone()
310        }
311    }
312    fn session_key(
313        &self,
314        id: SessionId
315    ) -> StoreKey {
316        StoreKey(mac(self.state_key.as_ref(), &context(&[b"session", &id.0])))
317    }
318    fn subject_tag(
319        &self,
320        subject: &str
321    ) -> [u64; 4] {
322        mac(self.state_key.as_ref(), &context(&[b"subject", subject.as_bytes()]))
323    }
324    fn check_scope(
325        &self,
326        principal: &Principal,
327        now: u64
328    ) -> Result<()> {
329        self.check_scope_lapsed(principal, now)?;
330        if now >= principal.claims.expires_at {
331            return Err(Error::Expired);
332        }
333        Ok(())
334    }
335    fn check_scope_lapsed(
336        &self,
337        principal: &Principal,
338        now: u64
339    ) -> Result<()> {
340        if principal.realm != self.realm || principal.issuer != self.issuer {
341            return Err(Error::InvalidToken);
342        }
343        if now < principal.claims.not_before || now < principal.claims.issued_at {
344            return Err(Error::NotYetValid);
345        }
346        Ok(())
347    }
348    /// The live-session part of `check`, after the scope was accepted.
349    fn check_session(
350        &self,
351        principal: &Principal,
352        now: u64
353    ) -> Result<()> {
354        let binding = principal.session.as_ref().ok_or(Error::InvalidToken)?;
355        {
356            let mut views = self.sessions.lock().map_err(|_| Error::PolicyUnavailable)?;
357            if let Some(view) = views.get(&binding.id) {
358                if self.table.unchanged::<P>(&view.snapshot)? {
359                    if view.subject != principal.subject() {
360                        return Err(Error::InvalidToken);
361                    }
362                    return self.check_metadata(view.snapshot.record, principal, now);
363                }
364                views.pop(&binding.id);
365            }
366        }
367        let key = self.session_key(binding.id);
368        let snapshot = self.table.lookup(key, &self.protection)?.ok_or(Error::Revoked)?;
369        self.check_record(snapshot.record, self.subject_tag(principal.subject()), principal, now)?;
370        self.sessions
371            .lock()
372            .map_err(|_| Error::PolicyUnavailable)?
373            .put(binding.id, SessionView { snapshot, subject: principal.subject().to_owned() });
374        Ok(())
375    }
376    fn check_record(
377        &self,
378        record: Record,
379        subject: [u64; 4],
380        principal: &Principal,
381        now: u64
382    ) -> Result<()> {
383        if record.subject != subject {
384            return Err(Error::InvalidToken);
385        }
386        self.check_metadata(record, principal, now)
387    }
388    fn check_live(
389        &self,
390        record: Record,
391        now: u64
392    ) -> Result<()> {
393        match record.status {
394            REVOKED => return Err(Error::Revoked),
395            SESSION => (),
396            _ => return Err(Error::PolicyUnavailable)
397        }
398        if now >= record.expires_at {
399            return Err(Error::Expired);
400        }
401        if now < record.created_at {
402            return Err(Error::NotYetValid);
403        }
404        Ok(())
405    }
406    fn check_metadata(
407        &self,
408        record: Record,
409        principal: &Principal,
410        now: u64
411    ) -> Result<()> {
412        self.check_live(record, now)?;
413        if principal.claims.issued_at < record.created_at
414            || principal.claims.expires_at > record.expires_at
415        {
416            return Err(Error::InvalidToken);
417        }
418        let binding = principal.session.as_ref().ok_or(Error::InvalidToken)?;
419        if principal.claims.purpose == Purpose::Refresh && binding.generation != record.generation {
420            return Err(Error::Replayed);
421        }
422        Ok(())
423    }
424}
425
426impl<P: StateProtection> PrincipalCache for FleetAuth<P> {
427    fn get(
428        &self,
429        key: &CacheKey
430    ) -> Option<CachedPrincipal> {
431        self.local.get(key)
432    }
433    fn insert(
434        &self,
435        key: CacheKey,
436        entry: CachedPrincipal
437    ) {
438        if key == entry.key {
439            self.local.insert(key, entry);
440        }
441    }
442}
443
444impl<P: StateProtection> ValidationHook for FleetAuth<P> {
445    fn check(
446        &self,
447        principal: &Principal,
448        now: u64
449    ) -> Result<()> {
450        self.check_scope(principal, now)?;
451        self.check_session(principal, now)
452    }
453
454    /// A lapsed credential of a live session passes the session checks; the
455    /// record's own horizon, revocation and subject still decide.
456    fn check_lapsed(
457        &self,
458        principal: &Principal,
459        now: u64
460    ) -> Result<()> {
461        self.check_scope_lapsed(principal, now)?;
462        self.check_session(principal, now)
463    }
464}
465
466pub struct ReplayGuard<'a, P: StateProtection>(&'a FleetAuth<P>);
467impl<P: StateProtection> ValidationHook for ReplayGuard<'_, P> {
468    fn check(
469        &self,
470        principal: &Principal,
471        now: u64
472    ) -> Result<()> {
473        let auth = self.0;
474        auth.check_scope(principal, now)?;
475        let key = StoreKey(mac(auth.state_key.as_ref(), &context(&[b"replay", &principal.jti.0])));
476        // Replay admission and live session check are one locked operation.
477        auth.table.write(|slots| {
478            if let Some(binding) = &principal.session {
479                let snapshot = find(slots, auth.session_key(binding.id), &auth.protection)?
480                    .ok_or(Error::Revoked)?;
481                auth.check_record(
482                    snapshot.record,
483                    auth.subject_tag(principal.subject()),
484                    principal,
485                    now
486                )?;
487            }
488            if find(slots, key, &auth.protection)?.is_some() {
489                return Err(Error::Replayed);
490            }
491            insert(
492                slots,
493                Record {
494                    key,
495                    subject: [0; 4],
496                    created_at: now,
497                    expires_at: principal.claims.expires_at,
498                    generation: 0,
499                    status: CONSUMED
500                },
501                now,
502                &auth.protection
503            )
504        })
505    }
506}
507
508fn mac(
509    key: &[u8],
510    value: &[u8]
511) -> [u64; 4] {
512    let mut mac = <Hmac<Sha256> as hmac::KeyInit>::new_from_slice(key).expect("HMAC key");
513    mac.update(value);
514    let bytes = mac.finalize().into_bytes();
515    std::array::from_fn(|i| {
516        u64::from_le_bytes(bytes[i * 8..i * 8 + 8].try_into().expect("eight bytes"))
517    })
518}
519
520#[cfg(test)]
521#[path = "../tests/unit/fleet.rs"]
522mod tests;
orbit_auth/fleet.rs

orbit_auth/
fleet.rs
