session_guard/guard/
mod.rs

1//! The guard: issues, checks, rotates and revokes sessions.
2
3mod outcome;
4#[cfg(test)]
5mod tests;
6
7use std::{
8    collections::HashMap,
9    marker::PhantomData,
10    net::IpAddr,
11    panic::{AssertUnwindSafe, catch_unwind},
12    sync::{
13        Arc, Mutex, PoisonError,
14        atomic::{AtomicBool, Ordering},
15    },
16    time::Duration,
17};
18
19use chrono::{DateTime, TimeDelta, Utc};
20use uuid::Uuid;
21use zeroize::Zeroizing;
22
23pub use outcome::{
24    AuthSession, Authenticated, IssuedToken, NewSession, Outcome, PurgeReport, Rejection, SessionInfo, StepUpOutcome,
25};
26
27use crate::{
28    checks::{Check, Checks, Decision},
29    client::{ClientContext, HeaderSource, resolve_client_ip},
30    clock::{Clock, SystemClock},
31    config::{GuardConfig, Mode, StepUpPolicy},
32    error::GuardError,
33    event::{EventKind, SessionEvent},
34    fingerprint::{Baseline, Fingerprint, UaCache},
35    intel::{IpIntel, NoIpIntel},
36    store::{Rotation, SessionRecord, SessionStore, SessionUpdate, Supersede},
37    subject::Subject,
38    token::{SessionToken, TokenHash},
39};
40
41use secrecy::SecretString;
42
43type Hook = Arc<dyn Fn(&SessionEvent) + Send + Sync>;
44
45/// Issues, checks, rotates and revokes sessions. Cheap to clone; clones share state.
46///
47/// `St` is the [`SessionStore`]; `S` is your subject type (default [`Uuid`]), stored opaquely.
48pub struct Guard<St, S = Uuid> {
49    inner: Arc<Inner<St>>,
50    _subject: PhantomData<fn() -> S>,
51}
52
53struct Inner<St> {
54    store: St,
55    config: GuardConfig,
56    intel: Box<dyn IpIntel>,
57    clock: Box<dyn Clock>,
58    hook: Option<Hook>,
59    ua_cache: UaCache,
60    xff_warned: AtomicBool,
61    shadow_seen: Mutex<ShadowSeen>,
62}
63
64/// Last time each unenforced event was recorded, for shadow-mode deduplication.
65struct ShadowSeen {
66    seen: HashMap<(Uuid, EventKind), DateTime<Utc>>,
67    pruned_at: DateTime<Utc>,
68}
69
70impl<St, S> Clone for Guard<St, S> {
71    fn clone(&self) -> Self {
72        Self {
73            inner: Arc::clone(&self.inner),
74            _subject: PhantomData,
75        }
76    }
77}
78
79impl<St, S> std::fmt::Debug for Guard<St, S> {
80    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
81        f.debug_struct("Guard")
82            .field("config", &self.inner.config)
83            .finish_non_exhaustive()
84    }
85}
86
87/// Builds a [`Guard`]. Start with [`Guard::builder`].
88#[must_use]
89pub struct GuardBuilder<St, S = Uuid> {
90    store: St,
91    config: GuardConfig,
92    intel: Box<dyn IpIntel>,
93    clock: Box<dyn Clock>,
94    hook: Option<Hook>,
95    _subject: PhantomData<fn() -> S>,
96}
97
98impl<St: SessionStore> Guard<St> {
99    /// A guard with the default configuration and [`Uuid`] subjects.
100    pub fn new(store: St) -> Self {
101        Self::builder(store, GuardConfig::default()).build()
102    }
103
104    /// Starts building a guard. Subjects are [`Uuid`] unless changed with
105    /// [`GuardBuilder::subject`].
106    pub fn builder(store: St, config: GuardConfig) -> GuardBuilder<St> {
107        GuardBuilder {
108            store,
109            config,
110            intel: Box::new(NoIpIntel),
111            clock: Box::new(SystemClock),
112            hook: None,
113            _subject: PhantomData,
114        }
115    }
116}
117
118impl<St: SessionStore, S: Subject> GuardBuilder<St, S> {
119    /// Uses `T` as the subject type.
120    pub fn subject<T: Subject>(self) -> GuardBuilder<St, T> {
121        GuardBuilder {
122            store: self.store,
123            config: self.config,
124            intel: self.intel,
125            clock: self.clock,
126            hook: self.hook,
127            _subject: PhantomData,
128        }
129    }
130
131    /// Country / ASN source. Without one, IP changes are judged by network prefix only.
132    pub fn ip_intel(mut self, intel: impl IpIntel) -> Self {
133        self.intel = Box::new(intel);
134        self
135    }
136
137    /// Called synchronously for every recorded event, after it is stored. Keep it fast; a panic
138    /// in the hook is caught and logged.
139    pub fn on_event(mut self, hook: impl Fn(&SessionEvent) + Send + Sync + 'static) -> Self {
140        self.hook = Some(Arc::new(hook));
141        self
142    }
143
144    /// Time source (default [`SystemClock`]).
145    pub fn clock(mut self, clock: impl Clock) -> Self {
146        self.clock = Box::new(clock);
147        self
148    }
149
150    /// Builds the guard.
151    pub fn build(self) -> Guard<St, S> {
152        Guard {
153            inner: Arc::new(Inner {
154                store: self.store,
155                config: self.config,
156                intel: self.intel,
157                clock: self.clock,
158                hook: self.hook,
159                ua_cache: UaCache::new(UA_CACHE_CAPACITY),
160                xff_warned: AtomicBool::new(false),
161                shadow_seen: Mutex::new(ShadowSeen {
162                    seen: HashMap::new(),
163                    pruned_at: DateTime::<Utc>::MIN_UTC,
164                }),
165            }),
166            _subject: PhantomData,
167        }
168    }
169}
170
171const UA_CACHE_CAPACITY: usize = 1024;
172
173/// Most (session, kind) pairs the local shadow dedup cache holds, about 6 MB at most.
174const SHADOW_SEEN_CAPACITY: usize = 100_000;
175
176/// How a presented token relates to its session's current token.
177#[derive(Clone, Copy, Debug, PartialEq, Eq)]
178enum TokenState {
179    /// The current token.
180    Current,
181    /// Superseded moments ago; a request that raced the rotation.
182    Grace,
183    /// Superseded, but its successor was never used: the rotation response was lost.
184    Orphaned,
185    /// Superseded and its successor is in use: two parties hold the session.
186    Reuse,
187}
188
189/// A token resolved to a live session with a decodable subject.
190struct Resolved<S> {
191    lookup: crate::store::TokenLookup,
192    hash: TokenHash,
193    subject: S,
194    state: TokenState,
195}
196
197impl<St: SessionStore, S: Subject> Guard<St, S> {
198    /// Creates a session for `subject` and returns its first token.
199    ///
200    /// Pass the token the client presented, if any: that session is revoked first, so a token
201    /// planted before login (session fixation) never becomes authenticated. Call this only
202    /// after verifying the subject's credentials, and rate-limit that endpoint yourself.
203    ///
204    /// # Errors
205    ///
206    /// Store failures, an unavailable RNG, or a subject that does not serialize.
207    pub async fn login(
208        &self,
209        subject: &S,
210        ctx: &ClientContext,
211        previous_token: Option<&str>,
212    ) -> Result<NewSession, GuardError> {
213        let inner = &*self.inner;
214        if let Some(hash) = previous_token.and_then(|t| self.presented_hash(t))
215            && let Some(old) = inner.store.find_by_token(&hash).await?
216            && inner.store.delete_session(old.session.id).await?
217        {
218            self.emit(old.session.id, &old.session.subject, EventKind::Logout, true, ctx, None)
219                .await;
220        }
221
222        let encoded = crate::subject::encode(subject)?;
223        let token = SessionToken::generate()?;
224        let now = self.now();
225        let fingerprint = self.fingerprint(ctx);
226        let record = SessionRecord {
227            id: Uuid::new_v4(),
228            subject: encoded,
229            created_at: now,
230            authenticated_at: now,
231            last_seen_at: now,
232            expires_at: now + inner.config.session_lifetime,
233            step_up_required: false,
234            baseline: Baseline::new(fingerprint),
235        };
236        inner.store.insert_session(&record, &token.hash()).await?;
237        self.emit(record.id, &record.subject, EventKind::Login, true, ctx, None)
238            .await;
239        Ok(NewSession {
240            session_id: record.id,
241            token: self.issue(&token, record.expires_at),
242        })
243    }
244
245    /// Checks a request's token with the configured checks.
246    ///
247    /// On success the request may proceed; send [`Authenticated::rotated`] back as
248    /// `Set-Cookie` when present. Every [`Rejection`] maps to 401.
249    ///
250    /// # Errors
251    ///
252    /// Store failures only. Bad tokens and risky clients are [`Outcome::Rejected`].
253    pub async fn authenticate(&self, token: &str, ctx: &ClientContext) -> Result<Outcome<S>, GuardError> {
254        self.authenticate_with(token, ctx, &self.inner.config.checks).await
255    }
256
257    /// Like [`authenticate`](Self::authenticate) with other checks, for routes that need
258    /// stricter (or looser) treatment. The baseline learns only the signals `checks` scored.
259    ///
260    /// # Errors
261    ///
262    /// Store failures only.
263    pub async fn authenticate_with(
264        &self,
265        token: &str,
266        ctx: &ClientContext,
267        checks: &Checks,
268    ) -> Result<Outcome<S>, GuardError> {
269        self.authenticate_inner(token, ctx, checks).await
270    }
271
272    /// The rotation due for a token in `state`, with its replacement.
273    fn plan_rotation(
274        &self,
275        state: TokenState,
276        issued_at: DateTime<Utc>,
277        hash: TokenHash,
278        now: DateTime<Utc>,
279    ) -> Result<Option<(Rotation, SessionToken)>, GuardError> {
280        let config = &self.inner.config;
281        let supersede = match state {
282            TokenState::Current if issued_at <= now - config.rotation_interval => Supersede::Exact(hash),
283            TokenState::Orphaned => Supersede::CurrentIssuedBefore(now - config.rotation_grace),
284            _ => return Ok(None),
285        };
286        let token = SessionToken::generate()?;
287        let rotation = Rotation {
288            new_token: token.hash(),
289            supersede,
290            prune_superseded_before: now - config.reuse_detection,
291        };
292        Ok(Some((rotation, token)))
293    }
294
295    /// Scores the request: the fingerprint against the baseline, plus the token's own state.
296    fn assess(
297        &self,
298        baseline: &Baseline,
299        fingerprint: &Fingerprint,
300        state: TokenState,
301        now: DateTime<Utc>,
302        checks: &Checks,
303    ) -> crate::risk::Assessment {
304        let window = self.inner.config.ping_pong_window;
305        let mut assessment = crate::risk::assess(baseline, fingerprint, now, window, checks);
306        if state == TokenState::Orphaned {
307            assessment.add(Check::OrphanedToken, checks);
308        }
309        assessment
310    }
311
312    async fn authenticate_inner(
313        &self,
314        token: &str,
315        ctx: &ClientContext,
316        checks: &Checks,
317    ) -> Result<Outcome<S>, GuardError> {
318        let inner = &*self.inner;
319        let config = &inner.config;
320        let now = self.now();
321        let resolved = match self.resolve(token, now).await? {
322            Ok(resolved) => resolved,
323            Err(rejection) => return Ok(Outcome::Rejected(rejection)),
324        };
325        let Resolved {
326            lookup,
327            hash,
328            subject,
329            state,
330        } = resolved;
331        let session = &lookup.session;
332
333        if state == TokenState::Reuse {
334            self.revoke_for_reuse(session, ctx).await?;
335            return Ok(Outcome::Rejected(Rejection::Revoked));
336        }
337
338        let fingerprint = self.fingerprint(ctx);
339        let assessment = self.assess(&session.baseline, &fingerprint, state, now, checks);
340        let decision = config.thresholds.decide(assessment.score);
341        crate::telemetry::assessed(decision, &assessment.signals);
342        let shadow = config.mode == Mode::Shadow;
343        let step_up = decision == Decision::StepUp;
344
345        if decision == Decision::Revoke || (step_up && config.step_up_policy == StepUpPolicy::Revoke) {
346            self.emit(
347                session.id,
348                &session.subject,
349                EventKind::RiskRevoked,
350                !shadow,
351                ctx,
352                Some(&assessment),
353            )
354            .await;
355            if !shadow {
356                inner.store.delete_session(session.id).await?;
357                return Ok(Outcome::Rejected(Rejection::Revoked));
358            }
359        } else if session.step_up_required && !shadow {
360            return Ok(Outcome::Rejected(Rejection::StepUpRequired));
361        } else if step_up {
362            let enforced = config.step_up_policy == StepUpPolicy::Require && !shadow;
363            self.emit(
364                session.id,
365                &session.subject,
366                EventKind::StepUpRequired,
367                enforced,
368                ctx,
369                Some(&assessment),
370            )
371            .await;
372            if enforced {
373                let mut update = SessionUpdate::touch(session.id, now);
374                update.step_up_required = Some(true);
375                update.mark_used =
376                    (state == TokenState::Current && lookup.token.first_used_at.is_none()).then_some(hash);
377                inner.store.apply(&update).await?;
378                return Ok(Outcome::Rejected(Rejection::StepUpRequired));
379            }
380        }
381
382        let mut update = SessionUpdate::touch(session.id, now);
383        if decision == Decision::Allow {
384            let mut baseline = session.baseline.clone();
385            if baseline.absorb(&fingerprint, now, checks) {
386                update.baseline = Some(baseline);
387            }
388        }
389
390        let new_token = match self.plan_rotation(state, lookup.token.issued_at, hash, now)? {
391            Some((rotation, token)) => {
392                update.rotation = Some(rotation);
393                Some(token)
394            }
395            None => None,
396        };
397        // A reissued token counts as used from the start, so the orphan it replaces turns into
398        // reuse: one reissue per orphan, or a thief could keep a stolen token alive by
399        // dropping every successor.
400        update.mark_used = match (state, &update.rotation) {
401            (TokenState::Current, _) if lookup.token.first_used_at.is_none() => Some(hash),
402            (TokenState::Orphaned, Some(rotation)) => Some(rotation.new_token),
403            _ => None,
404        };
405
406        let stale = now - session.last_seen_at >= config.last_seen_write_interval;
407        let mut rotated = None;
408        if update.rotation.is_some() || update.mark_used.is_some() || update.baseline.is_some() || stale {
409            let did_rotate = inner.store.apply(&update).await?;
410            if did_rotate && let Some(token) = &new_token {
411                rotated = Some(self.issue(token, session.expires_at));
412                if state == TokenState::Orphaned {
413                    self.emit(session.id, &session.subject, EventKind::TokenReissued, true, ctx, None)
414                        .await;
415                }
416            }
417        }
418
419        Ok(Outcome::Authenticated(Authenticated {
420            session: AuthSession::new(session, subject),
421            rotated,
422            assessment,
423        }))
424    }
425
426    /// Looks a token up without changing anything: no rotation, no `last_seen` write, no
427    /// checks. `None` unless the token is current (or within its grace period) and the session
428    /// is live.
429    ///
430    /// # Errors
431    ///
432    /// Store failures only.
433    pub async fn peek(&self, token: &str) -> Result<Option<SessionInfo<S>>, GuardError> {
434        Ok(match self.resolve(token, self.now()).await? {
435            Ok(resolved) if matches!(resolved.state, TokenState::Current | TokenState::Grace) => {
436                SessionInfo::from_record(&resolved.lookup.session)
437            }
438            _ => None,
439        })
440    }
441
442    /// Clears a pending step-up after the app has re-verified the subject (password, passkey,
443    /// OTP…). The current client becomes the new baseline, `authenticated_at` resets and a new
444    /// token is issued.
445    ///
446    /// `verified_subject` must be the subject the app just verified; a mismatch changes
447    /// nothing. Rate-limit the verification yourself.
448    ///
449    /// # Errors
450    ///
451    /// Store failures or an unavailable RNG.
452    pub async fn complete_step_up(
453        &self,
454        token: &str,
455        ctx: &ClientContext,
456        verified_subject: &S,
457    ) -> Result<StepUpOutcome, GuardError> {
458        let inner = &*self.inner;
459        let now = self.now();
460        let resolved = match self.resolve(token, now).await? {
461            Ok(resolved) => resolved,
462            Err(rejection) => return Ok(StepUpOutcome::Rejected(rejection)),
463        };
464        let session = &resolved.lookup.session;
465        if resolved.state == TokenState::Reuse {
466            self.revoke_for_reuse(session, ctx).await?;
467            return Ok(StepUpOutcome::Rejected(Rejection::Revoked));
468        }
469        if resolved.subject != *verified_subject {
470            return Ok(StepUpOutcome::SubjectMismatch);
471        }
472
473        let token = SessionToken::generate()?;
474        let mut update = SessionUpdate::touch(session.id, now);
475        update.baseline = Some(Baseline::new(self.fingerprint(ctx)));
476        update.step_up_required = Some(false);
477        update.authenticated_at = Some(now);
478        update.rotation = Some(Rotation {
479            new_token: token.hash(),
480            supersede: Supersede::Current,
481            prune_superseded_before: now - inner.config.reuse_detection,
482        });
483        if !inner.store.apply(&update).await? {
484            // The session vanished between lookup and update.
485            return Ok(StepUpOutcome::Rejected(Rejection::Revoked));
486        }
487        self.emit(
488            session.id,
489            &session.subject,
490            EventKind::StepUpCompleted,
491            true,
492            ctx,
493            None,
494        )
495        .await;
496        Ok(StepUpOutcome::Completed(self.issue(&token, session.expires_at)))
497    }
498
499    /// Ends the session the token belongs to. Returns whether there was one.
500    ///
501    /// # Errors
502    ///
503    /// Store failures only.
504    pub async fn logout(&self, token: &str, ctx: &ClientContext) -> Result<bool, GuardError> {
505        let inner = &*self.inner;
506        let Some(hash) = self.presented_hash(token) else {
507            return Ok(false);
508        };
509        let Some(lookup) = inner.store.find_by_token(&hash).await? else {
510            return Ok(false);
511        };
512        let deleted = inner.store.delete_session(lookup.session.id).await?;
513        if deleted {
514            self.emit(
515                lookup.session.id,
516                &lookup.session.subject,
517                EventKind::Logout,
518                true,
519                ctx,
520                None,
521            )
522            .await;
523        }
524        Ok(deleted)
525    }
526
527    /// Deletes one session by id, e.g. from a "your devices" page. Returns whether it existed.
528    ///
529    /// # Errors
530    ///
531    /// Store failures only.
532    pub async fn revoke_session(&self, session_id: Uuid) -> Result<bool, GuardError> {
533        Ok(self.inner.store.delete_session(session_id).await?)
534    }
535
536    /// Deletes every session of `subject`, e.g. after a password change. Returns how many.
537    ///
538    /// # Errors
539    ///
540    /// Store failures or a subject that does not serialize.
541    pub async fn revoke_all_for_subject(&self, subject: &S) -> Result<u64, GuardError> {
542        let encoded = crate::subject::encode(subject)?;
543        Ok(self.inner.store.delete_subject_sessions(&encoded).await?)
544    }
545
546    /// The live sessions of `subject`.
547    ///
548    /// # Errors
549    ///
550    /// Store failures or a subject that does not serialize.
551    pub async fn list_sessions(&self, subject: &S) -> Result<Vec<SessionInfo<S>>, GuardError> {
552        let encoded = crate::subject::encode(subject)?;
553        let now = self.now();
554        let records = self.inner.store.list_subject_sessions(&encoded).await?;
555        Ok(records
556            .iter()
557            .filter(|r| self.is_live(r, now))
558            .filter_map(SessionInfo::from_record)
559            .collect())
560    }
561
562    /// Deletes expired sessions and events older than the retention period. Run it
563    /// periodically (hourly is plenty).
564    ///
565    /// # Errors
566    ///
567    /// Store failures only.
568    pub async fn purge_expired(&self) -> Result<PurgeReport, GuardError> {
569        let config = &self.inner.config;
570        let now = self.now();
571        let sessions = self.inner.store.delete_expired(now, now - config.idle_timeout).await?;
572        let events = self
573            .inner
574            .store
575            .delete_events_before(now - config.event_retention)
576            .await?;
577        Ok(PurgeReport { sessions, events })
578    }
579
580    /// Runs [`purge_expired`](Self::purge_expired) now and then every `every` on the tokio
581    /// runtime, logging each report (`info`) or failure (`warn`) with `tracing`. A failed run
582    /// does not stop the task; abort the handle to stop it.
583    ///
584    /// # Panics
585    ///
586    /// Outside a tokio runtime, or if `every` is zero.
587    #[cfg(feature = "tokio")]
588    #[cfg_attr(docsrs, doc(cfg(feature = "tokio")))]
589    pub fn spawn_purge(&self, every: Duration) -> tokio::task::JoinHandle<()> {
590        let guard = self.clone();
591        tokio::spawn(async move {
592            let mut ticker = tokio::time::interval(every);
593            ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
594            loop {
595                ticker.tick().await;
596                match guard.purge_expired().await {
597                    Ok(report) => tracing::info!(
598                        sessions = report.sessions,
599                        events = report.events,
600                        "purged expired sessions"
601                    ),
602                    Err(error) => tracing::warn!(%error, "purging expired sessions failed"),
603                }
604            }
605        })
606    }
607
608    /// Builds the [`ClientContext`] for a request from the TCP peer address and the request
609    /// headers.
610    ///
611    /// `X-Forwarded-For` is honoured only when `peer` is a configured trusted proxy, walking
612    /// right to left past trusted hops. Header values are capped at the configured length.
613    pub fn client_context<H: HeaderSource + ?Sized>(&self, peer: IpAddr, headers: &H) -> ClientContext {
614        let config = &self.inner.config;
615        // A non-UTF-8 value stays in the chain as an empty, unparseable hop, so the peer is used
616        // instead of whatever valid hops surround it.
617        let forwarded: Vec<&str> = headers
618            .header_values("x-forwarded-for")
619            .map(|v| std::str::from_utf8(v).unwrap_or(""))
620            .collect();
621        if config.trusted_proxies.is_empty()
622            && !forwarded.is_empty()
623            && !self.inner.xff_warned.swap(true, Ordering::Relaxed)
624        {
625            tracing::warn!(
626                "X-Forwarded-For received but no trusted proxies are configured; using the peer address. \
627                 Configure trusted_proxies if the app runs behind a reverse proxy."
628            );
629        }
630        let ip = resolve_client_ip(peer, forwarded, &config.trusted_proxies);
631        ClientContext::from_headers(
632            ip,
633            headers,
634            config.tls_fingerprint_header.as_deref(),
635            config.max_header_len,
636        )
637    }
638
639    /// The session cookie's value from the request's `Cookie` headers, if present.
640    pub fn cookie_token<'a, H: HeaderSource + ?Sized>(&self, headers: &'a H) -> Option<&'a str> {
641        headers
642            .header_values("cookie")
643            .filter_map(|v| std::str::from_utf8(v).ok())
644            .find_map(|v| self.inner.config.cookie.find_in(v))
645    }
646
647    /// A `Set-Cookie` value that deletes the session cookie.
648    pub fn clear_cookie(&self) -> String {
649        self.inner.config.cookie.clear_cookie()
650    }
651
652    #[cfg(feature = "aide")]
653    pub(crate) fn cookie_name(&self) -> &str {
654        self.inner.config.cookie.cookie_name()
655    }
656
657    /// Whether the subject last proved who they are more than `max_age` ago. Use it to demand
658    /// a fresh login before sensitive actions.
659    ///
660    /// With a zero `max_age` it is true as soon as any time has passed since the login or
661    /// step-up, so it is true for practically every request.
662    pub fn requires_reauth(&self, session: &AuthSession<S>, max_age: Duration) -> bool {
663        self.now() - session.authenticated_at > td(max_age)
664    }
665
666    /// Whether `value` has the shape of this guard's tokens (configured prefix, then 43
667    /// base64url characters). Does not touch the store.
668    pub fn is_session_token(&self, value: &str) -> bool {
669        self.presented_hash(value).is_some()
670    }
671
672    fn now(&self) -> DateTime<Utc> {
673        self.inner.clock.now()
674    }
675
676    fn fingerprint(&self, ctx: &ClientContext) -> Fingerprint {
677        Fingerprint::capture_cached(ctx, self.inner.intel.lookup(ctx.ip), &self.inner.ua_cache)
678    }
679
680    fn presented_hash(&self, presented: &str) -> Option<TokenHash> {
681        let bare = match &self.inner.config.token_prefix {
682            Some(prefix) => presented.strip_prefix(prefix.as_str())?,
683            None => presented,
684        };
685        TokenHash::from_presented(bare)
686    }
687
688    fn issue(&self, token: &SessionToken, expires_at: DateTime<Utc>) -> IssuedToken {
689        // Exact capacity, so neither building the value nor boxing it into the secret leaves an
690        // unwiped copy behind in a reallocated buffer.
691        let prefix = self.inner.config.token_prefix.as_deref().unwrap_or("");
692        let mut value = String::with_capacity(prefix.len() + token.expose().len());
693        value.push_str(prefix);
694        value.push_str(token.expose());
695        let max_age = (expires_at - self.now()).num_seconds();
696        let set_cookie = Zeroizing::new(self.inner.config.cookie.set_cookie(&value, max_age));
697        IssuedToken {
698            value: SecretString::from(value),
699            expires_at,
700            set_cookie,
701        }
702    }
703
704    fn is_live(&self, record: &SessionRecord, now: DateTime<Utc>) -> bool {
705        now < record.expires_at && now - record.last_seen_at < self.inner.config.idle_timeout
706    }
707
708    fn classify(&self, lookup: &crate::store::TokenLookup, now: DateTime<Utc>) -> TokenState {
709        match lookup.token.superseded_at {
710            None => TokenState::Current,
711            Some(at) if now - at <= self.inner.config.rotation_grace => TokenState::Grace,
712            Some(_) if !lookup.newer_token_used => TokenState::Orphaned,
713            Some(_) => TokenState::Reuse,
714        }
715    }
716
717    /// Whether an orphaned token was superseded longer ago than the reissue window allows.
718    fn orphan_too_old(&self, lookup: &crate::store::TokenLookup, now: DateTime<Utc>) -> bool {
719        match (self.inner.config.orphan_reissue_window, lookup.token.superseded_at) {
720            (Some(window), Some(at)) => now - at > window,
721            _ => false,
722        }
723    }
724
725    /// Finds a token's live session. Expired sessions are deleted on sight.
726    async fn resolve(&self, token: &str, now: DateTime<Utc>) -> Result<Result<Resolved<S>, Rejection>, GuardError> {
727        let store = &self.inner.store;
728        let Some(hash) = self.presented_hash(token) else {
729            return Ok(Err(Rejection::NotAuthenticated));
730        };
731        let Some(lookup) = store.find_by_token(&hash).await? else {
732            return Ok(Err(Rejection::NotAuthenticated));
733        };
734        if !self.is_live(&lookup.session, now) {
735            store.delete_session(lookup.session.id).await?;
736            return Ok(Err(Rejection::Expired));
737        }
738        let Some(subject) = crate::subject::decode::<S>(&lookup.session.subject) else {
739            tracing::warn!(session_id = %lookup.session.id, "stored subject does not decode as the guard's subject type");
740            return Ok(Err(Rejection::NotAuthenticated));
741        };
742        let state = self.classify(&lookup, now);
743        if state == TokenState::Orphaned && self.orphan_too_old(&lookup, now) {
744            // Treated like a pruned token: the session stays with whoever holds the successor.
745            return Ok(Err(Rejection::NotAuthenticated));
746        }
747        Ok(Ok(Resolved {
748            lookup,
749            hash,
750            subject,
751            state,
752        }))
753    }
754
755    async fn revoke_for_reuse(&self, session: &SessionRecord, ctx: &ClientContext) -> Result<(), GuardError> {
756        self.inner.store.delete_session(session.id).await?;
757        self.emit(session.id, &session.subject, EventKind::TokenReuse, true, ctx, None)
758            .await;
759        Ok(())
760    }
761
762    /// Records an event, calls the hook and updates metrics. Never fails: a store error is
763    /// logged, a hook panic is caught. Unenforced events are recorded at most once per session
764    /// and kind per dedup window.
765    async fn emit(
766        &self,
767        session_id: Uuid,
768        subject: &str,
769        kind: EventKind,
770        enforced: bool,
771        ctx: &ClientContext,
772        assessment: Option<&crate::risk::Assessment>,
773    ) {
774        let inner = &*self.inner;
775        let config = &inner.config;
776        let now = self.now();
777        if !enforced {
778            if self.seen_recently(session_id, kind, now) {
779                return;
780            }
781            match inner
782                .store
783                .claim_event(session_id, kind, now, config.shadow_dedup_window)
784                .await
785            {
786                Ok(true) => {}
787                Ok(false) => return,
788                // Better a duplicate than a lost event.
789                Err(error) => {
790                    tracing::error!(%session_id, kind = kind.as_str(), %error, "failed to claim session event");
791                }
792            }
793        }
794
795        let ip = if config.minimize_event_ip {
796            mask_ip(ctx.ip)
797        } else {
798            ctx.ip
799        };
800        let event = SessionEvent {
801            session_id,
802            subject: subject.to_owned(),
803            kind,
804            enforced,
805            score: assessment.map(|a| a.score),
806            signals: assessment.map(|a| a.signals.clone()).unwrap_or_default(),
807            ip,
808            user_agent: ctx
809                .user_agent
810                .as_deref()
811                .map(|ua| crate::client::truncate(ua, config.max_header_len).to_owned()),
812            at: now,
813        };
814        if kind.is_security_relevant() {
815            tracing::warn!(%session_id, kind = kind.as_str(), enforced, score = ?event.score, "session event");
816        } else {
817            tracing::debug!(%session_id, kind = kind.as_str(), "session event");
818        }
819        if let Err(error) = inner.store.record_event(&event).await {
820            tracing::error!(%session_id, kind = kind.as_str(), %error, "failed to record session event");
821        }
822        if let Some(hook) = &inner.hook
823            && catch_unwind(AssertUnwindSafe(|| hook(&event))).is_err()
824        {
825            tracing::error!(%session_id, kind = kind.as_str(), "on_event hook panicked");
826        }
827        crate::telemetry::event(kind, enforced);
828    }
829
830    /// Whether this process already handled an unenforced `kind` event for the session within
831    /// the dedup window; notes this one if not. A cheap first filter in front of the store's
832    /// cross-process [`SessionStore::claim_event`].
833    fn seen_recently(&self, session_id: Uuid, kind: EventKind, now: DateTime<Utc>) -> bool {
834        let window = self.inner.config.shadow_dedup_window;
835        let mut state = self.inner.shadow_seen.lock().unwrap_or_else(PoisonError::into_inner);
836        if now - state.pruned_at >= window || state.seen.len() >= SHADOW_SEEN_CAPACITY {
837            state.seen.retain(|_, at| now - *at < window);
838            state.pruned_at = now;
839            if state.seen.len() >= SHADOW_SEEN_CAPACITY {
840                // A flood of distinct sessions: forget everything and let the store dedupe.
841                state.seen.clear();
842            }
843        }
844        match state.seen.get(&(session_id, kind)) {
845            Some(at) if now - *at < window => true,
846            _ => {
847                state.seen.insert((session_id, kind), now);
848                false
849            }
850        }
851    }
852}
853
854/// Reduces an address to its network prefix (/24 or /48) for event storage.
855fn mask_ip(ip: IpAddr) -> IpAddr {
856    let len = if ip.is_ipv4() { 24 } else { 48 };
857    ipnet::IpNet::new(ip, len).map_or(ip, |net| net.trunc().addr())
858}
859
860fn td(duration: Duration) -> TimeDelta {
861    TimeDelta::from_std(duration).unwrap_or(TimeDelta::MAX)
862}
session_guard/guard/mod.rs

session_guard/guard/
mod.rs
