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::{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    async fn authenticate_inner(
296        &self,
297        token: &str,
298        ctx: &ClientContext,
299        checks: &Checks,
300    ) -> Result<Outcome<S>, GuardError> {
301        let inner = &*self.inner;
302        let config = &inner.config;
303        let now = self.now();
304        let resolved = match self.resolve(token, now).await? {
305            Ok(resolved) => resolved,
306            Err(rejection) => return Ok(Outcome::Rejected(rejection)),
307        };
308        let Resolved {
309            lookup,
310            hash,
311            subject,
312            state,
313        } = resolved;
314        let session = &lookup.session;
315
316        if state == TokenState::Reuse {
317            self.revoke_for_reuse(session, ctx).await?;
318            return Ok(Outcome::Rejected(Rejection::Revoked));
319        }
320
321        let fingerprint = self.fingerprint(ctx);
322        let assessment = crate::risk::assess(&session.baseline, &fingerprint, now, config.ping_pong_window, checks);
323        let decision = config.thresholds.decide(assessment.score);
324        crate::telemetry::assessed(decision, &assessment.signals);
325        let shadow = config.mode == Mode::Shadow;
326        let step_up = decision == Decision::StepUp;
327
328        if decision == Decision::Revoke || (step_up && config.step_up_policy == StepUpPolicy::Revoke) {
329            self.emit(
330                session.id,
331                &session.subject,
332                EventKind::RiskRevoked,
333                !shadow,
334                ctx,
335                Some(&assessment),
336            )
337            .await;
338            if !shadow {
339                inner.store.delete_session(session.id).await?;
340                return Ok(Outcome::Rejected(Rejection::Revoked));
341            }
342        } else if session.step_up_required && !shadow {
343            return Ok(Outcome::Rejected(Rejection::StepUpRequired));
344        } else if step_up {
345            let enforced = config.step_up_policy == StepUpPolicy::Require && !shadow;
346            self.emit(
347                session.id,
348                &session.subject,
349                EventKind::StepUpRequired,
350                enforced,
351                ctx,
352                Some(&assessment),
353            )
354            .await;
355            if enforced {
356                let mut update = SessionUpdate::touch(session.id, now);
357                update.step_up_required = Some(true);
358                update.mark_used =
359                    (state == TokenState::Current && lookup.token.first_used_at.is_none()).then_some(hash);
360                inner.store.apply(&update).await?;
361                return Ok(Outcome::Rejected(Rejection::StepUpRequired));
362            }
363        }
364
365        let mut update = SessionUpdate::touch(session.id, now);
366        if decision == Decision::Allow {
367            let mut baseline = session.baseline.clone();
368            if baseline.absorb(&fingerprint, now, checks) {
369                update.baseline = Some(baseline);
370            }
371        }
372
373        let new_token = match self.plan_rotation(state, lookup.token.issued_at, hash, now)? {
374            Some((rotation, token)) => {
375                update.rotation = Some(rotation);
376                Some(token)
377            }
378            None => None,
379        };
380        // A reissued token counts as used from the start, so the orphan it replaces turns into
381        // reuse: one reissue per orphan, or a thief could keep a stolen token alive by
382        // dropping every successor.
383        update.mark_used = match (state, &update.rotation) {
384            (TokenState::Current, _) if lookup.token.first_used_at.is_none() => Some(hash),
385            (TokenState::Orphaned, Some(rotation)) => Some(rotation.new_token),
386            _ => None,
387        };
388
389        let stale = now - session.last_seen_at >= config.last_seen_write_interval;
390        let mut rotated = None;
391        if update.rotation.is_some() || update.mark_used.is_some() || update.baseline.is_some() || stale {
392            let did_rotate = inner.store.apply(&update).await?;
393            if did_rotate && let Some(token) = &new_token {
394                rotated = Some(self.issue(token, session.expires_at));
395                if state == TokenState::Orphaned {
396                    self.emit(session.id, &session.subject, EventKind::TokenReissued, true, ctx, None)
397                        .await;
398                }
399            }
400        }
401
402        Ok(Outcome::Authenticated(Authenticated {
403            session: AuthSession::new(session, subject),
404            rotated,
405            assessment,
406        }))
407    }
408
409    /// Looks a token up without changing anything: no rotation, no `last_seen` write, no
410    /// checks. `None` unless the token is current (or within its grace period) and the session
411    /// is live.
412    ///
413    /// # Errors
414    ///
415    /// Store failures only.
416    pub async fn peek(&self, token: &str) -> Result<Option<SessionInfo<S>>, GuardError> {
417        Ok(match self.resolve(token, self.now()).await? {
418            Ok(resolved) if matches!(resolved.state, TokenState::Current | TokenState::Grace) => {
419                SessionInfo::from_record(&resolved.lookup.session)
420            }
421            _ => None,
422        })
423    }
424
425    /// Clears a pending step-up after the app has re-verified the subject (password, passkey,
426    /// OTP…). The current client becomes the new baseline, `authenticated_at` resets and a new
427    /// token is issued.
428    ///
429    /// `verified_subject` must be the subject the app just verified; a mismatch changes
430    /// nothing. Rate-limit the verification yourself.
431    ///
432    /// # Errors
433    ///
434    /// Store failures or an unavailable RNG.
435    pub async fn complete_step_up(
436        &self,
437        token: &str,
438        ctx: &ClientContext,
439        verified_subject: &S,
440    ) -> Result<StepUpOutcome, GuardError> {
441        let inner = &*self.inner;
442        let now = self.now();
443        let resolved = match self.resolve(token, now).await? {
444            Ok(resolved) => resolved,
445            Err(rejection) => return Ok(StepUpOutcome::Rejected(rejection)),
446        };
447        let session = &resolved.lookup.session;
448        if resolved.state == TokenState::Reuse {
449            self.revoke_for_reuse(session, ctx).await?;
450            return Ok(StepUpOutcome::Rejected(Rejection::Revoked));
451        }
452        if resolved.subject != *verified_subject {
453            return Ok(StepUpOutcome::SubjectMismatch);
454        }
455
456        let token = SessionToken::generate()?;
457        let mut update = SessionUpdate::touch(session.id, now);
458        update.baseline = Some(Baseline::new(self.fingerprint(ctx)));
459        update.step_up_required = Some(false);
460        update.authenticated_at = Some(now);
461        update.rotation = Some(Rotation {
462            new_token: token.hash(),
463            supersede: Supersede::Current,
464            prune_superseded_before: now - inner.config.reuse_detection,
465        });
466        if !inner.store.apply(&update).await? {
467            // The session vanished between lookup and update.
468            return Ok(StepUpOutcome::Rejected(Rejection::Revoked));
469        }
470        self.emit(
471            session.id,
472            &session.subject,
473            EventKind::StepUpCompleted,
474            true,
475            ctx,
476            None,
477        )
478        .await;
479        Ok(StepUpOutcome::Completed(self.issue(&token, session.expires_at)))
480    }
481
482    /// Ends the session the token belongs to. Returns whether there was one.
483    ///
484    /// # Errors
485    ///
486    /// Store failures only.
487    pub async fn logout(&self, token: &str, ctx: &ClientContext) -> Result<bool, GuardError> {
488        let inner = &*self.inner;
489        let Some(hash) = self.presented_hash(token) else {
490            return Ok(false);
491        };
492        let Some(lookup) = inner.store.find_by_token(&hash).await? else {
493            return Ok(false);
494        };
495        let deleted = inner.store.delete_session(lookup.session.id).await?;
496        if deleted {
497            self.emit(
498                lookup.session.id,
499                &lookup.session.subject,
500                EventKind::Logout,
501                true,
502                ctx,
503                None,
504            )
505            .await;
506        }
507        Ok(deleted)
508    }
509
510    /// Deletes one session by id, e.g. from a "your devices" page. Returns whether it existed.
511    ///
512    /// # Errors
513    ///
514    /// Store failures only.
515    pub async fn revoke_session(&self, session_id: Uuid) -> Result<bool, GuardError> {
516        Ok(self.inner.store.delete_session(session_id).await?)
517    }
518
519    /// Deletes every session of `subject`, e.g. after a password change. Returns how many.
520    ///
521    /// # Errors
522    ///
523    /// Store failures or a subject that does not serialize.
524    pub async fn revoke_all_for_subject(&self, subject: &S) -> Result<u64, GuardError> {
525        let encoded = crate::subject::encode(subject)?;
526        Ok(self.inner.store.delete_subject_sessions(&encoded).await?)
527    }
528
529    /// The live sessions of `subject`.
530    ///
531    /// # Errors
532    ///
533    /// Store failures or a subject that does not serialize.
534    pub async fn list_sessions(&self, subject: &S) -> Result<Vec<SessionInfo<S>>, GuardError> {
535        let encoded = crate::subject::encode(subject)?;
536        let now = self.now();
537        let records = self.inner.store.list_subject_sessions(&encoded).await?;
538        Ok(records
539            .iter()
540            .filter(|r| self.is_live(r, now))
541            .filter_map(SessionInfo::from_record)
542            .collect())
543    }
544
545    /// Deletes expired sessions and events older than the retention period. Run it
546    /// periodically (hourly is plenty).
547    ///
548    /// # Errors
549    ///
550    /// Store failures only.
551    pub async fn purge_expired(&self) -> Result<PurgeReport, GuardError> {
552        let config = &self.inner.config;
553        let now = self.now();
554        let sessions = self.inner.store.delete_expired(now, now - config.idle_timeout).await?;
555        let events = self
556            .inner
557            .store
558            .delete_events_before(now - config.event_retention)
559            .await?;
560        Ok(PurgeReport { sessions, events })
561    }
562
563    /// Runs [`purge_expired`](Self::purge_expired) now and then every `every` on the tokio
564    /// runtime, logging each report (`info`) or failure (`warn`) with `tracing`. A failed run
565    /// does not stop the task; abort the handle to stop it.
566    ///
567    /// # Panics
568    ///
569    /// Outside a tokio runtime, or if `every` is zero.
570    #[cfg(feature = "tokio")]
571    #[cfg_attr(docsrs, doc(cfg(feature = "tokio")))]
572    pub fn spawn_purge(&self, every: Duration) -> tokio::task::JoinHandle<()> {
573        let guard = self.clone();
574        tokio::spawn(async move {
575            let mut ticker = tokio::time::interval(every);
576            ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
577            loop {
578                ticker.tick().await;
579                match guard.purge_expired().await {
580                    Ok(report) => tracing::info!(
581                        sessions = report.sessions,
582                        events = report.events,
583                        "purged expired sessions"
584                    ),
585                    Err(error) => tracing::warn!(%error, "purging expired sessions failed"),
586                }
587            }
588        })
589    }
590
591    /// Builds the [`ClientContext`] for a request from the TCP peer address and the request
592    /// headers.
593    ///
594    /// `X-Forwarded-For` is honoured only when `peer` is a configured trusted proxy, walking
595    /// right to left past trusted hops. Header values are capped at the configured length.
596    pub fn client_context<H: HeaderSource + ?Sized>(&self, peer: IpAddr, headers: &H) -> ClientContext {
597        let config = &self.inner.config;
598        // A non-UTF-8 value stays in the chain as an empty, unparseable hop, so the peer is used
599        // instead of whatever valid hops surround it.
600        let forwarded: Vec<&str> = headers
601            .header_values("x-forwarded-for")
602            .map(|v| std::str::from_utf8(v).unwrap_or(""))
603            .collect();
604        if config.trusted_proxies.is_empty()
605            && !forwarded.is_empty()
606            && !self.inner.xff_warned.swap(true, Ordering::Relaxed)
607        {
608            tracing::warn!(
609                "X-Forwarded-For received but no trusted proxies are configured; using the peer address. \
610                 Configure trusted_proxies if the app runs behind a reverse proxy."
611            );
612        }
613        let ip = resolve_client_ip(peer, forwarded, &config.trusted_proxies);
614        ClientContext::from_headers(
615            ip,
616            headers,
617            config.tls_fingerprint_header.as_deref(),
618            config.max_header_len,
619        )
620    }
621
622    /// The session cookie's value from the request's `Cookie` headers, if present.
623    pub fn cookie_token<'a, H: HeaderSource + ?Sized>(&self, headers: &'a H) -> Option<&'a str> {
624        headers
625            .header_values("cookie")
626            .filter_map(|v| std::str::from_utf8(v).ok())
627            .find_map(|v| self.inner.config.cookie.find_in(v))
628    }
629
630    /// A `Set-Cookie` value that deletes the session cookie.
631    pub fn clear_cookie(&self) -> String {
632        self.inner.config.cookie.clear_cookie()
633    }
634
635    #[cfg(feature = "aide")]
636    pub(crate) fn cookie_name(&self) -> &str {
637        self.inner.config.cookie.cookie_name()
638    }
639
640    /// Whether the subject last proved who they are more than `max_age` ago. Use it to demand
641    /// a fresh login before sensitive actions.
642    ///
643    /// With a zero `max_age` it is true as soon as any time has passed since the login or
644    /// step-up, so it is true for practically every request.
645    pub fn requires_reauth(&self, session: &AuthSession<S>, max_age: Duration) -> bool {
646        self.now() - session.authenticated_at > td(max_age)
647    }
648
649    /// Whether `value` has the shape of this guard's tokens (configured prefix, then 43
650    /// base64url characters). Does not touch the store.
651    pub fn is_session_token(&self, value: &str) -> bool {
652        self.presented_hash(value).is_some()
653    }
654
655    fn now(&self) -> DateTime<Utc> {
656        self.inner.clock.now()
657    }
658
659    fn fingerprint(&self, ctx: &ClientContext) -> Fingerprint {
660        Fingerprint::capture_cached(ctx, self.inner.intel.lookup(ctx.ip), &self.inner.ua_cache)
661    }
662
663    fn presented_hash(&self, presented: &str) -> Option<TokenHash> {
664        let bare = match &self.inner.config.token_prefix {
665            Some(prefix) => presented.strip_prefix(prefix.as_str())?,
666            None => presented,
667        };
668        TokenHash::from_presented(bare)
669    }
670
671    fn issue(&self, token: &SessionToken, expires_at: DateTime<Utc>) -> IssuedToken {
672        // Exact capacity, so neither building the value nor boxing it into the secret leaves an
673        // unwiped copy behind in a reallocated buffer.
674        let prefix = self.inner.config.token_prefix.as_deref().unwrap_or("");
675        let mut value = String::with_capacity(prefix.len() + token.expose().len());
676        value.push_str(prefix);
677        value.push_str(token.expose());
678        let max_age = (expires_at - self.now()).num_seconds();
679        let set_cookie = Zeroizing::new(self.inner.config.cookie.set_cookie(&value, max_age));
680        IssuedToken {
681            value: SecretString::from(value),
682            expires_at,
683            set_cookie,
684        }
685    }
686
687    fn is_live(&self, record: &SessionRecord, now: DateTime<Utc>) -> bool {
688        now < record.expires_at && now - record.last_seen_at < self.inner.config.idle_timeout
689    }
690
691    fn classify(&self, lookup: &crate::store::TokenLookup, now: DateTime<Utc>) -> TokenState {
692        match lookup.token.superseded_at {
693            None => TokenState::Current,
694            Some(at) if now - at <= self.inner.config.rotation_grace => TokenState::Grace,
695            Some(_) if !lookup.newer_token_used => TokenState::Orphaned,
696            Some(_) => TokenState::Reuse,
697        }
698    }
699
700    /// Finds a token's live session. Expired sessions are deleted on sight.
701    async fn resolve(&self, token: &str, now: DateTime<Utc>) -> Result<Result<Resolved<S>, Rejection>, GuardError> {
702        let store = &self.inner.store;
703        let Some(hash) = self.presented_hash(token) else {
704            return Ok(Err(Rejection::NotAuthenticated));
705        };
706        let Some(lookup) = store.find_by_token(&hash).await? else {
707            return Ok(Err(Rejection::NotAuthenticated));
708        };
709        if !self.is_live(&lookup.session, now) {
710            store.delete_session(lookup.session.id).await?;
711            return Ok(Err(Rejection::Expired));
712        }
713        let Some(subject) = crate::subject::decode::<S>(&lookup.session.subject) else {
714            tracing::warn!(session_id = %lookup.session.id, "stored subject does not decode as the guard's subject type");
715            return Ok(Err(Rejection::NotAuthenticated));
716        };
717        let state = self.classify(&lookup, now);
718        Ok(Ok(Resolved {
719            lookup,
720            hash,
721            subject,
722            state,
723        }))
724    }
725
726    async fn revoke_for_reuse(&self, session: &SessionRecord, ctx: &ClientContext) -> Result<(), GuardError> {
727        self.inner.store.delete_session(session.id).await?;
728        self.emit(session.id, &session.subject, EventKind::TokenReuse, true, ctx, None)
729            .await;
730        Ok(())
731    }
732
733    /// Records an event, calls the hook and updates metrics. Never fails: a store error is
734    /// logged, a hook panic is caught. Unenforced events are recorded at most once per session
735    /// and kind per dedup window.
736    async fn emit(
737        &self,
738        session_id: Uuid,
739        subject: &str,
740        kind: EventKind,
741        enforced: bool,
742        ctx: &ClientContext,
743        assessment: Option<&crate::risk::Assessment>,
744    ) {
745        let inner = &*self.inner;
746        let config = &inner.config;
747        let now = self.now();
748        if !enforced {
749            if self.seen_recently(session_id, kind, now) {
750                return;
751            }
752            match inner
753                .store
754                .claim_event(session_id, kind, now, config.shadow_dedup_window)
755                .await
756            {
757                Ok(true) => {}
758                Ok(false) => return,
759                // Better a duplicate than a lost event.
760                Err(error) => {
761                    tracing::error!(%session_id, kind = kind.as_str(), %error, "failed to claim session event");
762                }
763            }
764        }
765
766        let ip = if config.minimize_event_ip {
767            mask_ip(ctx.ip)
768        } else {
769            ctx.ip
770        };
771        let event = SessionEvent {
772            session_id,
773            subject: subject.to_owned(),
774            kind,
775            enforced,
776            score: assessment.map(|a| a.score),
777            signals: assessment.map(|a| a.signals.clone()).unwrap_or_default(),
778            ip,
779            user_agent: ctx
780                .user_agent
781                .as_deref()
782                .map(|ua| crate::client::truncate(ua, config.max_header_len).to_owned()),
783            at: now,
784        };
785        if kind.is_security_relevant() {
786            tracing::warn!(%session_id, kind = kind.as_str(), enforced, score = ?event.score, "session event");
787        } else {
788            tracing::debug!(%session_id, kind = kind.as_str(), "session event");
789        }
790        if let Err(error) = inner.store.record_event(&event).await {
791            tracing::error!(%session_id, kind = kind.as_str(), %error, "failed to record session event");
792        }
793        if let Some(hook) = &inner.hook
794            && catch_unwind(AssertUnwindSafe(|| hook(&event))).is_err()
795        {
796            tracing::error!(%session_id, kind = kind.as_str(), "on_event hook panicked");
797        }
798        crate::telemetry::event(kind, enforced);
799    }
800
801    /// Whether this process already handled an unenforced `kind` event for the session within
802    /// the dedup window; notes this one if not. A cheap first filter in front of the store's
803    /// cross-process [`SessionStore::claim_event`].
804    fn seen_recently(&self, session_id: Uuid, kind: EventKind, now: DateTime<Utc>) -> bool {
805        let window = self.inner.config.shadow_dedup_window;
806        let mut state = self.inner.shadow_seen.lock().unwrap_or_else(PoisonError::into_inner);
807        if now - state.pruned_at >= window || state.seen.len() >= SHADOW_SEEN_CAPACITY {
808            state.seen.retain(|_, at| now - *at < window);
809            state.pruned_at = now;
810            if state.seen.len() >= SHADOW_SEEN_CAPACITY {
811                // A flood of distinct sessions: forget everything and let the store dedupe.
812                state.seen.clear();
813            }
814        }
815        match state.seen.get(&(session_id, kind)) {
816            Some(at) if now - *at < window => true,
817            _ => {
818                state.seen.insert((session_id, kind), now);
819                false
820            }
821        }
822    }
823}
824
825/// Reduces an address to its network prefix (/24 or /48) for event storage.
826fn mask_ip(ip: IpAddr) -> IpAddr {
827    let len = if ip.is_ipv4() { 24 } else { 48 };
828    ipnet::IpNet::new(ip, len).map_or(ip, |net| net.trunc().addr())
829}
830
831fn td(duration: Duration) -> TimeDelta {
832    TimeDelta::from_std(duration).unwrap_or(TimeDelta::MAX)
833}
session_guard/guard/mod.rs

session_guard/guard/
mod.rs
