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 expires_at(&self) -> u64 {
49        self.expires_at
50    }
51    pub(crate) fn binding(&self) -> SessionBinding {
52        self.binding.clone()
53    }
54    pub(crate) fn check_issuance(
55        &self,
56        authority: &Authority,
57        claims: &Claims
58    ) -> Result<()> {
59        if self.realm != authority.realm
60            || self.issuer != authority.issuer
61            || claims.subject != self.subject
62            || claims.issued_at < self.created_at
63            || claims.expires_at > self.expires_at
64        {
65            return Err(Error::InvalidInput);
66        }
67        Ok(())
68    }
69}
70
71struct SessionView {
72    snapshot: Snapshot,
73    // Compared once against the shared keyed subject tag, then checked locally.
74    subject: String
75}
76
77/// Local decoded Principals plus fleet-native live sessions and replay state.
78/// Backing follows the supplied Fleet. SHM survives handle drop and restart;
79/// memory state lives as long as a handle retains it. No background driver.
80///
81/// Shared records contain only keyed identifiers and fixed control metadata.
82/// Owner-only OS permissions restrict access. `Blake3State` verifies
83/// integrity; `EncryptedState` also hides metadata. `UnprotectedState` explicitly
84/// trusts every writer. Protected modes still permit deletion/rollback attacks.
85/// `state_key` stays stable across token-key rotations and is never stored in SHM.
86pub struct FleetAuth<P: StateProtection> {
87    _fleet: Arc<Fleet>,
88    table: Table,
89    local: MemoryCache,
90    sessions: Mutex<LruCache<SessionId, SessionView>>,
91    state_key: Zeroizing<[u8; 32]>,
92    protection: RecordProtection<P>,
93    realm: String,
94    issuer: String
95}
96impl<P: StateProtection> FleetAuth<P> {
97    /// Construct with an explicitly selected shared-state protection policy.
98    /// There is no default policy.
99    ///
100    /// `local_capacity` bounds each local Principal/session-view LRU. A cached
101    /// protected view is reused only while every shared byte matches its local copy.
102    /// UnprotectedState trusts the commit revision instead.
103    /// Peers sharing a physical table must select the same policy. Token crypto
104    /// and live session/replay checks are enabled for every policy.
105    pub fn new(
106        fleet: Arc<Fleet>,
107        authority: &Authority,
108        state_key: SecretKey,
109        local_capacity: NonZeroUsize,
110        _policy: P
111    ) -> Result<Self> {
112        let info = context(&[
113            b"orbit-auth/v2/fleet-state",
114            fleet.name().as_bytes(),
115            authority.realm.as_bytes(),
116            authority.issuer.as_bytes()
117        ]);
118        Ok(Self {
119            table: Table::new(fleet.clone(), P::ID)?,
120            // The physical table may contain multiple realm/issuer namespaces.
121            // Its protection key is fleet-scoped; record lookup identifiers below bind
122            // the realm/issuer. All users of one table share this state secret.
123            protection: RecordProtection::new(derive(
124                &state_key,
125                &context(&[
126                    b"orbit-auth/v4/fleet-state-protection",
127                    fleet.name().as_bytes(),
128                    &P::ID.to_le_bytes()
129                ])
130            )),
131            _fleet: fleet,
132            local: MemoryCache::new(local_capacity),
133            sessions: Mutex::new(LruCache::new(local_capacity)),
134            state_key: derive(&state_key, &info),
135            realm: authority.realm.clone(),
136            issuer: authority.issuer.clone()
137        })
138    }
139
140    /// Clear this handle's decoded/validated views; shared authority is preserved.
141    pub fn clear_local(&self) {
142        self.local.clear();
143        self.sessions.lock().unwrap_or_else(|e| e.into_inner()).clear();
144    }
145
146    pub fn create_session(
147        &self,
148        subject: impl Into<String>,
149        now: u64,
150        expires_at: u64
151    ) -> Result<Session> {
152        let subject = subject.into();
153        if !crate::principal::valid_label(&subject) || now >= expires_at {
154            return Err(Error::InvalidInput);
155        }
156        let mut id = [0; 16];
157        getrandom::fill(&mut id).map_err(|_| Error::Randomness)?;
158        let id = SessionId(id);
159        let record = Record {
160            key: self.session_key(id),
161            subject: self.subject_tag(&subject),
162            created_at: now,
163            expires_at,
164            generation: 0,
165            status: SESSION
166        };
167        self.table.write(|slots| {
168            if find(slots, record.key, &self.protection)?.is_some() {
169                return Err(Error::InvalidInput);
170            }
171            insert(slots, record, now, &self.protection)
172        })?;
173        Ok(self.session(id, subject, record))
174    }
175
176    /// Atomically publish revocation. Every subsequent validation rechecks the
177    /// authenticated record, including warm local cache hits. Already admitted work is
178    /// not cancelled; no notification or polling interval delays the next check.
179    pub fn revoke_session(
180        &self,
181        id: SessionId,
182        now: u64
183    ) -> Result<()> {
184        let key = self.session_key(id);
185        self.table.write(|slots| {
186            let snapshot = find(slots, key, &self.protection)?.ok_or(Error::Revoked)?;
187            let mut record = snapshot.record;
188            if !matches!(record.status, SESSION | REVOKED) {
189                return Err(Error::PolicyUnavailable);
190            }
191            if now < record.created_at {
192                return Err(Error::NotYetValid);
193            }
194            record.status = REVOKED;
195            slots[snapshot.index].store(record, snapshot.index, &self.protection)?;
196            Ok(())
197        })
198    }
199
200    /// Validate and consume the current refresh generation atomically. Exactly
201    /// one concurrent caller succeeds. The explicit renewed expiry may preserve
202    /// or extend the previous horizon. The returned handle issues new credentials.
203    /// The application owns renewal/delegation limits. If a response is lost,
204    /// the consumed credential is not retried; reauthentication is required.
205    pub fn refresh(
206        &self,
207        validator: &Validator<'_>,
208        token: &str,
209        now: u64,
210        expires_at: u64
211    ) -> Result<Session> {
212        if validator.validation.purpose != Purpose::Refresh {
213            return Err(Error::InvalidInput);
214        }
215        let principal = validator.validate(token, now, self)?;
216        let binding = principal.session.as_ref().ok_or(Error::InvalidToken)?;
217        let key = self.session_key(binding.id);
218        let subject = self.subject_tag(principal.subject());
219        self.table.write(|slots| {
220            let snapshot = find(slots, key, &self.protection)?.ok_or(Error::Revoked)?;
221            let mut record = snapshot.record;
222            self.check_record(record, subject, &principal, now)?;
223            if expires_at < record.expires_at {
224                return Err(Error::InvalidInput);
225            }
226            record.generation = record.generation.checked_add(1).ok_or(Error::PolicyUnavailable)?;
227            record.expires_at = expires_at;
228            slots[snapshot.index].store(record, snapshot.index, &self.protection)?;
229            Ok(self.session(binding.id, principal.subject().to_owned(), record))
230        })
231    }
232
233    /// One-use hook. Unexpired consumed IDs never evict; full/unavailable denies.
234    pub fn replay_guard(&self) -> ReplayGuard<'_, P> {
235        ReplayGuard(self)
236    }
237
238    fn session(
239        &self,
240        id: SessionId,
241        subject: String,
242        record: Record
243    ) -> Session {
244        Session {
245            binding: SessionBinding { id, generation: record.generation },
246            subject,
247            created_at: record.created_at,
248            expires_at: record.expires_at,
249            realm: self.realm.clone(),
250            issuer: self.issuer.clone()
251        }
252    }
253    fn session_key(
254        &self,
255        id: SessionId
256    ) -> StoreKey {
257        StoreKey(mac(self.state_key.as_ref(), &context(&[b"session", &id.0])))
258    }
259    fn subject_tag(
260        &self,
261        subject: &str
262    ) -> [u64; 4] {
263        mac(self.state_key.as_ref(), &context(&[b"subject", subject.as_bytes()]))
264    }
265    fn check_scope(
266        &self,
267        principal: &Principal,
268        now: u64
269    ) -> Result<()> {
270        if principal.realm != self.realm || principal.issuer != self.issuer {
271            return Err(Error::InvalidToken);
272        }
273        if now >= principal.claims.expires_at {
274            return Err(Error::Expired);
275        }
276        if now < principal.claims.not_before || now < principal.claims.issued_at {
277            return Err(Error::NotYetValid);
278        }
279        Ok(())
280    }
281    fn check_record(
282        &self,
283        record: Record,
284        subject: [u64; 4],
285        principal: &Principal,
286        now: u64
287    ) -> Result<()> {
288        if record.subject != subject {
289            return Err(Error::InvalidToken);
290        }
291        self.check_metadata(record, principal, now)
292    }
293    fn check_metadata(
294        &self,
295        record: Record,
296        principal: &Principal,
297        now: u64
298    ) -> Result<()> {
299        match record.status {
300            REVOKED => return Err(Error::Revoked),
301            SESSION => (),
302            _ => return Err(Error::PolicyUnavailable)
303        }
304        if now >= record.expires_at {
305            return Err(Error::Expired);
306        }
307        if now < record.created_at {
308            return Err(Error::NotYetValid);
309        }
310        if principal.claims.issued_at < record.created_at
311            || principal.claims.expires_at > record.expires_at
312        {
313            return Err(Error::InvalidToken);
314        }
315        let binding = principal.session.as_ref().ok_or(Error::InvalidToken)?;
316        if principal.claims.purpose == Purpose::Refresh && binding.generation != record.generation {
317            return Err(Error::Replayed);
318        }
319        Ok(())
320    }
321}
322
323impl<P: StateProtection> PrincipalCache for FleetAuth<P> {
324    fn get(
325        &self,
326        key: &CacheKey
327    ) -> Option<CachedPrincipal> {
328        self.local.get(key)
329    }
330    fn insert(
331        &self,
332        key: CacheKey,
333        entry: CachedPrincipal
334    ) {
335        if key == entry.key {
336            self.local.insert(key, entry);
337        }
338    }
339}
340
341impl<P: StateProtection> ValidationHook for FleetAuth<P> {
342    fn check(
343        &self,
344        principal: &Principal,
345        now: u64
346    ) -> Result<()> {
347        self.check_scope(principal, now)?;
348        let binding = principal.session.as_ref().ok_or(Error::InvalidToken)?;
349        {
350            let mut views = self.sessions.lock().map_err(|_| Error::PolicyUnavailable)?;
351            if let Some(view) = views.get(&binding.id) {
352                if self.table.unchanged::<P>(&view.snapshot)? {
353                    if view.subject != principal.subject() {
354                        return Err(Error::InvalidToken);
355                    }
356                    return self.check_metadata(view.snapshot.record, principal, now);
357                }
358                views.pop(&binding.id);
359            }
360        }
361        let key = self.session_key(binding.id);
362        let snapshot = self.table.lookup(key, &self.protection)?.ok_or(Error::Revoked)?;
363        self.check_record(snapshot.record, self.subject_tag(principal.subject()), principal, now)?;
364        self.sessions
365            .lock()
366            .map_err(|_| Error::PolicyUnavailable)?
367            .put(binding.id, SessionView { snapshot, subject: principal.subject().to_owned() });
368        Ok(())
369    }
370}
371
372pub struct ReplayGuard<'a, P: StateProtection>(&'a FleetAuth<P>);
373impl<P: StateProtection> ValidationHook for ReplayGuard<'_, P> {
374    fn check(
375        &self,
376        principal: &Principal,
377        now: u64
378    ) -> Result<()> {
379        let auth = self.0;
380        auth.check_scope(principal, now)?;
381        let key = StoreKey(mac(auth.state_key.as_ref(), &context(&[b"replay", &principal.jti.0])));
382        // Replay admission and live session check are one locked operation.
383        auth.table.write(|slots| {
384            if let Some(binding) = &principal.session {
385                let snapshot = find(slots, auth.session_key(binding.id), &auth.protection)?
386                    .ok_or(Error::Revoked)?;
387                auth.check_record(
388                    snapshot.record,
389                    auth.subject_tag(principal.subject()),
390                    principal,
391                    now
392                )?;
393            }
394            if find(slots, key, &auth.protection)?.is_some() {
395                return Err(Error::Replayed);
396            }
397            insert(
398                slots,
399                Record {
400                    key,
401                    subject: [0; 4],
402                    created_at: now,
403                    expires_at: principal.claims.expires_at,
404                    generation: 0,
405                    status: CONSUMED
406                },
407                now,
408                &auth.protection
409            )
410        })
411    }
412}
413
414fn mac(
415    key: &[u8],
416    value: &[u8]
417) -> [u64; 4] {
418    let mut mac = <Hmac<Sha256> as hmac::KeyInit>::new_from_slice(key).expect("HMAC key");
419    mac.update(value);
420    let bytes = mac.finalize().into_bytes();
421    std::array::from_fn(|i| {
422        u64::from_le_bytes(bytes[i * 8..i * 8 + 8].try_into().expect("eight bytes"))
423    })
424}
425
426#[cfg(test)]
427#[path = "../tests/unit/fleet.rs"]
428mod tests;
orbit_auth/fleet.rs

orbit_auth/
fleet.rs
