cheers_axum/
knock.rs

1//! Knock routes (`knock.md`; R734-F4) over [`KnockAuthority`].
2//!
3//! | Route | Caller | Does |
4//! |---|---|---|
5//! | `POST /knock` `{knock}` | anyone holding a key (rate-limited) | verifies the signed `Knock`, queues it; a second from the same key replaces the first |
6//! | `GET /knock?resource_kind=&resource_id=` | bearer with the admit ability | lists pending knocks |
7//! | `POST /knock/{id}/admit` `{confirmation?, lease_seconds?}` | bearer with the admit ability | writes the tuple, `granted_by` = caller |
8//! | `POST /offer` `{resource_kind, resource_id, relation, max_uses}` | bearer with the admit ability | mints an issuer-signed `Offer` |
9//! | `POST /offer/redeem` `{offer, knock}` | anyone holding a key | redeems the offer for the knock's key |
10//! | `POST /admit` `{admit}` | anyone (the artifact authenticates itself) | reconciles an offline `Admit` |
11//!
12//! Bearers are the access tokens [`EdgeVerifier`] accepts; the caller is the
13//! token's user. `POST /knock` is limited per source (the connection's peer
14//! address from axum's `ConnectInfo`; without one every request shares a
15//! single bucket, which fails tight) and per resource, by the in-memory
16//! [`RateLimiter`] at the [`KnockConfig`] rates.
17
18use std::collections::HashMap;
19use std::net::SocketAddr;
20use std::sync::{Arc, Mutex, PoisonError};
21
22use axum::extract::{ConnectInfo, Path, Query, Request, State};
23use axum::http::{HeaderMap, StatusCode};
24use axum::response::{IntoResponse, Response};
25use axum::routing::post;
26use axum::{Json, Router};
27use serde::{Deserialize, Serialize};
28
29use cheers_core::{Confirmation, Knock, PrincipalId, TokenVerifier};
30use cheers_server::{
31    Admitted, EdgeVerifier, KnockAuthority, KnockFlowError, OwnershipRow, PendingKnock, Queued, Reconciled, RevocationReader,
32};
33use cheers_verify::verify_device_artifact;
34
35use crate::error::RouteError;
36use crate::me::bearer_from_headers;
37
38/// A fixed-window counter per key: at most `limit` hits per 60-second window.
39/// In memory, per process; a multi-node deployment limits per node.
40#[derive(Debug, Default)]
41pub struct RateLimiter {
42    windows: Mutex<HashMap<String, (i64, u32)>>,
43}
44
45impl RateLimiter {
46    pub fn new() -> Self {
47        Self::default()
48    }
49
50    /// Count one hit on `key` at `now`; `false` once the window is full.
51    pub fn allow(&self, key: &str, limit: u32, now: i64) -> bool {
52        let window = now.div_euclid(60);
53        let mut w = self.windows.lock().unwrap_or_else(PoisonError::into_inner);
54        if w.len() > 10_000 {
55            w.retain(|_, (at, _)| *at == window);
56        }
57        let slot = w.entry(key.to_owned()).or_insert((window, 0));
58        if slot.0 != window {
59            *slot = (window, 0);
60        }
61        if slot.1 >= limit {
62            return false;
63        }
64        slot.1 += 1;
65        true
66    }
67}
68
69pub struct KnockState<V, Rd, S> {
70    /// Verifies the bearer access tokens of approvers.
71    pub edge: Arc<EdgeVerifier<V, Rd>>,
72    pub authority: Arc<KnockAuthority<S>>,
73    pub limiter: Arc<RateLimiter>,
74}
75
76impl<V, Rd, S> std::fmt::Debug for KnockState<V, Rd, S> {
77    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
78        f.debug_struct("KnockState").finish_non_exhaustive()
79    }
80}
81
82/// A knock route's failure.
83#[derive(Debug, thiserror::Error)]
84pub enum KnockRouteError {
85    #[error(transparent)]
86    Route(#[from] RouteError),
87    #[error(transparent)]
88    Flow(#[from] KnockFlowError),
89    #[error("too many knocks; slow down")]
90    RateLimited,
91    #[error("bad request body: {0}")]
92    BadBody(String),
93}
94
95#[derive(Serialize)]
96struct ErrorBody {
97    error: &'static str,
98    message: String,
99}
100
101impl IntoResponse for KnockRouteError {
102    fn into_response(self) -> Response {
103        use KnockFlowError as F;
104        let flow = match self {
105            KnockRouteError::Route(e) => return e.into_response(),
106            other => other,
107        };
108        let (status, code) = match &flow {
109            KnockRouteError::Route(_) => unreachable!("returned above"),
110            KnockRouteError::RateLimited => (StatusCode::TOO_MANY_REQUESTS, "rate_limited"),
111            KnockRouteError::BadBody(_) => (StatusCode::BAD_REQUEST, "bad_body"),
112            KnockRouteError::Flow(f) => match f {
113                F::Artifact(_) | F::Invalid(_) | F::Standing(_) | F::WrongAuthority { .. } | F::OfferMismatch(_) => {
114                    (StatusCode::BAD_REQUEST, "bad_artifact")
115                }
116                F::Expired { .. } => (StatusCode::GONE, "expired"),
117                F::NotAnApprover(_) => (StatusCode::FORBIDDEN, "not_an_approver"),
118                F::Refused(_) => (StatusCode::FORBIDDEN, "admission_refused"),
119                F::PendingFull => (StatusCode::TOO_MANY_REQUESTS, "pending_full"),
120                F::UnknownKnock => (StatusCode::NOT_FOUND, "unknown_knock"),
121                F::UnknownOffer => (StatusCode::NOT_FOUND, "unknown_offer"),
122                F::OfferExhausted => (StatusCode::GONE, "offer_exhausted"),
123                F::OfferExpired => (StatusCode::GONE, "offer_expired"),
124                F::Store(_) => (StatusCode::INTERNAL_SERVER_ERROR, "store"),
125            },
126        };
127        (status, Json(ErrorBody { error: code, message: flow.to_string() })).into_response()
128    }
129}
130
131#[derive(Debug, Clone, Deserialize)]
132pub struct KnockBody {
133    /// The signed `Knock`.
134    pub knock: String,
135}
136
137#[derive(Debug, Clone, Serialize, Deserialize)]
138pub struct KnockQueued {
139    pub id: String,
140    pub expires_at: i64,
141    /// The key's earlier pending knock on this resource was replaced.
142    pub replaced: bool,
143}
144
145#[derive(Debug, Clone, Deserialize)]
146pub struct PendingQuery {
147    pub resource_kind: String,
148    pub resource_id: String,
149}
150
151#[derive(Debug, Clone, Default, Deserialize)]
152#[serde(deny_unknown_fields)]
153pub struct AdmitKnockBody {
154    /// Defaults to `accept`.
155    #[serde(default)]
156    pub confirmation: Option<Confirmation>,
157    /// Lease length; the path default when absent.
158    #[serde(default)]
159    pub lease_seconds: Option<i64>,
160}
161
162#[derive(Debug, Clone, Serialize, Deserialize)]
163pub struct AdmittedBody {
164    pub jti: String,
165    pub ownership: OwnershipRow,
166}
167
168impl From<Admitted> for AdmittedBody {
169    fn from(a: Admitted) -> Self {
170        Self { jti: a.jti, ownership: a.row }
171    }
172}
173
174#[derive(Debug, Clone, Deserialize)]
175pub struct CreateOfferBody {
176    pub resource_kind: String,
177    pub resource_id: String,
178    pub relation: String,
179    pub max_uses: u32,
180}
181
182#[derive(Debug, Clone, Serialize, Deserialize)]
183pub struct OfferCreated {
184    pub offer: cheers_core::Offer,
185    pub token: String,
186}
187
188#[derive(Debug, Clone, Deserialize)]
189pub struct RedeemBody {
190    pub offer: String,
191    pub knock: String,
192}
193
194#[derive(Debug, Clone, Deserialize)]
195pub struct AdmitUploadBody {
196    pub admit: String,
197}
198
199#[derive(Debug, Clone, Serialize, Deserialize)]
200#[serde(tag = "outcome", rename_all = "snake_case")]
201pub enum AdmitUploadOutcome {
202    Accepted { jti: String, ownership: OwnershipRow },
203    Refused { jti: String, refusal: String },
204}
205
206async fn caller<V, Rd, S>(headers: &HeaderMap, state: &KnockState<V, Rd, S>, now: i64) -> Result<PrincipalId, RouteError>
207where
208    V: TokenVerifier + Send + Sync,
209    Rd: RevocationReader,
210{
211    let token = bearer_from_headers(headers)?;
212    match state.edge.verify_at(token, now).await {
213        Ok(claims) => Ok(PrincipalId::user(claims.sub.as_str())),
214        Err(cheers_core::Error::Store(e)) => Err(RouteError::Store(e.to_string())),
215        Err(_) => Err(RouteError::Unauthorized),
216    }
217}
218
219async fn post_knock<V, Rd, S>(
220    State(state): State<Arc<KnockState<V, Rd, S>>>,
221    request: Request,
222) -> Result<(StatusCode, Json<KnockQueued>), KnockRouteError>
223where
224    V: TokenVerifier + Send + Sync,
225    Rd: RevocationReader,
226    S: RevocationReader + Send + Sync,
227{
228    let now = now_unix();
229    let config = *state.authority.config();
230    let source = request
231        .extensions()
232        .get::<ConnectInfo<SocketAddr>>()
233        .map(|c| c.0.ip().to_string())
234        .unwrap_or_else(|| "unknown".into());
235    if !state.limiter.allow(&format!("source:{source}"), config.knocks_per_source_per_minute, now) {
236        return Err(KnockRouteError::RateLimited);
237    }
238    let bytes = axum::body::to_bytes(request.into_body(), 64 * 1024)
239        .await
240        .map_err(|e| KnockRouteError::BadBody(e.to_string()))?;
241    let body: KnockBody = serde_json::from_slice(&bytes).map_err(|e| KnockRouteError::BadBody(e.to_string()))?;
242    let knock: Knock = verify_device_artifact(&body.knock).map_err(KnockFlowError::from)?;
243    let resource = format!("resource:{}/{}", knock.kind, knock.id);
244    if !state.limiter.allow(&resource, config.knocks_per_resource_per_minute, now) {
245        return Err(KnockRouteError::RateLimited);
246    }
247    let (pending, queued) = state.authority.knock(&body.knock, now).await?;
248    Ok((
249        StatusCode::ACCEPTED,
250        Json(KnockQueued { id: pending.id, expires_at: pending.expires_at, replaced: queued == Queued::Replaced }),
251    ))
252}
253
254async fn list_knocks<V, Rd, S>(
255    State(state): State<Arc<KnockState<V, Rd, S>>>,
256    headers: HeaderMap,
257    Query(q): Query<PendingQuery>,
258) -> Result<Json<Vec<PendingKnock>>, KnockRouteError>
259where
260    V: TokenVerifier + Send + Sync,
261    Rd: RevocationReader,
262    S: RevocationReader + Send + Sync,
263{
264    let now = now_unix();
265    let who = caller(&headers, &state, now).await?;
266    Ok(Json(state.authority.pending(&who, &q.resource_kind, &q.resource_id, now).await?))
267}
268
269async fn admit_knock<V, Rd, S>(
270    State(state): State<Arc<KnockState<V, Rd, S>>>,
271    headers: HeaderMap,
272    Path(id): Path<String>,
273    body: Option<Json<AdmitKnockBody>>,
274) -> Result<(StatusCode, Json<AdmittedBody>), KnockRouteError>
275where
276    V: TokenVerifier + Send + Sync,
277    Rd: RevocationReader,
278    S: RevocationReader + Send + Sync,
279{
280    let now = now_unix();
281    let who = caller(&headers, &state, now).await?;
282    let body = body.map(|Json(b)| b).unwrap_or_default();
283    let confirmation = body.confirmation.unwrap_or(Confirmation::Accept);
284    let admitted = state.authority.admit_knock(&who, &id, confirmation, body.lease_seconds, now).await?;
285    Ok((StatusCode::CREATED, Json(admitted.into())))
286}
287
288async fn create_offer<V, Rd, S>(
289    State(state): State<Arc<KnockState<V, Rd, S>>>,
290    headers: HeaderMap,
291    Json(body): Json<CreateOfferBody>,
292) -> Result<(StatusCode, Json<OfferCreated>), KnockRouteError>
293where
294    V: TokenVerifier + Send + Sync,
295    Rd: RevocationReader,
296    S: RevocationReader + Send + Sync,
297{
298    let now = now_unix();
299    let who = caller(&headers, &state, now).await?;
300    let (offer, token) = state
301        .authority
302        .offer(&who, &body.resource_kind, &body.resource_id, &body.relation, body.max_uses, now)
303        .await?;
304    Ok((StatusCode::CREATED, Json(OfferCreated { offer, token })))
305}
306
307async fn redeem_offer<V, Rd, S>(
308    State(state): State<Arc<KnockState<V, Rd, S>>>,
309    Json(body): Json<RedeemBody>,
310) -> Result<(StatusCode, Json<AdmittedBody>), KnockRouteError>
311where
312    S: RevocationReader + Send + Sync,
313{
314    let admitted = state.authority.redeem_offer(&body.offer, &body.knock, now_unix()).await?;
315    Ok((StatusCode::CREATED, Json(admitted.into())))
316}
317
318async fn upload_admit<V, Rd, S>(
319    State(state): State<Arc<KnockState<V, Rd, S>>>,
320    Json(body): Json<AdmitUploadBody>,
321) -> Result<(StatusCode, Json<AdmitUploadOutcome>), KnockRouteError>
322where
323    S: RevocationReader + Send + Sync,
324{
325    Ok(match state.authority.reconcile(&body.admit, now_unix()).await? {
326        Reconciled::Accepted(a) => (StatusCode::OK, Json(AdmitUploadOutcome::Accepted { jti: a.jti, ownership: a.row })),
327        Reconciled::Refused { jti, refusal } => (StatusCode::FORBIDDEN, Json(AdmitUploadOutcome::Refused { jti, refusal })),
328    })
329}
330
331/// The knock routes (module table). Serve under `ConnectInfo` (e.g.
332/// `into_make_service_with_connect_info::<SocketAddr>()`) so `POST /knock`
333/// is limited per client address.
334pub fn router<V, Rd, S>(state: Arc<KnockState<V, Rd, S>>) -> Router
335where
336    V: TokenVerifier + Send + Sync + 'static,
337    Rd: RevocationReader + Send + Sync + 'static,
338    S: RevocationReader + Send + Sync + 'static,
339{
340    Router::new()
341        .route("/knock", post(post_knock::<V, Rd, S>).get(list_knocks::<V, Rd, S>))
342        .route("/knock/{id}/admit", post(admit_knock::<V, Rd, S>))
343        .route("/offer", post(create_offer::<V, Rd, S>))
344        .route("/offer/redeem", post(redeem_offer::<V, Rd, S>))
345        .route("/admit", post(upload_admit::<V, Rd, S>))
346        .with_state(state)
347}
348
349fn now_unix() -> i64 {
350    use std::time::{SystemTime, UNIX_EPOCH};
351    SystemTime::now().duration_since(UNIX_EPOCH).map(|d| d.as_secs() as i64).unwrap_or_default()
352}
353
354#[cfg(test)]
355mod tests {
356    use super::*;
357
358    #[test]
359    fn the_limiter_counts_per_key_per_minute() {
360        let l = RateLimiter::new();
361        assert!(l.allow("a", 2, 600));
362        assert!(l.allow("a", 2, 610));
363        assert!(!l.allow("a", 2, 619));
364        assert!(l.allow("b", 2, 619));
365        assert!(l.allow("a", 2, 660), "a new window resets");
366    }
367}
cheers_axum/knock.rs

cheers_axum/
knock.rs
