pretix_webhook/
service.rs

1use std::{
2    convert::Infallible,
3    fmt::Display,
4    future::Future,
5    pin::Pin,
6    sync::Arc,
7    task::{Context, Poll},
8};
9
10use bytes::Bytes;
11use http::{HeaderMap, Request, Response, StatusCode, header};
12use http_body::Body;
13use http_body_util::{BodyExt, Empty, LengthLimitError, Limited};
14use pretix_webhook_events::WebhookEvent;
15use tower::{BoxError, Service};
16
17use crate::{builder::WebhookServiceBuilder, handler::WebhookHandler};
18
19/// The maximum request body size used by [`WebhookServiceBuilder::new`].
20pub const DEFAULT_BODY_LIMIT: usize = 2 * 1024 * 1024;
21
22/// The empty HTTP response returned by [`WebhookService`].
23pub type WebhookResponse = Response<Empty<Bytes>>;
24
25type ResponseFuture =
26    Pin<Box<dyn Future<Output = Result<WebhookResponse, Infallible>> + Send + 'static>>;
27
28/// An HTTP service that authenticates and processes pretix webhooks.
29///
30/// Routing and HTTP method selection belong to the caller. The handler
31/// receives only authenticated, parsed events that pass the configured filters.
32/// Apply an outer Tower concurrency or load-shedding layer when accepted work
33/// must be bounded.
34pub struct WebhookService<H> {
35    handler: Arc<H>,
36    policy: WebhookServiceBuilder,
37}
38
39impl<H> WebhookService<H> {
40    pub(crate) fn new(handler: H, policy: WebhookServiceBuilder) -> Self {
41        Self {
42            handler: Arc::new(handler),
43            policy,
44        }
45    }
46}
47
48impl<H> Clone for WebhookService<H> {
49    fn clone(&self) -> Self {
50        Self {
51            handler: Arc::clone(&self.handler),
52            policy: self.policy.clone(),
53        }
54    }
55}
56
57impl<H, B> Service<Request<B>> for WebhookService<H>
58where
59    B: Body<Data = Bytes> + Send + 'static,
60    B::Error: Into<BoxError>,
61    H: WebhookHandler,
62{
63    type Response = WebhookResponse;
64    type Error = Infallible;
65    type Future = ResponseFuture;
66
67    fn poll_ready(&mut self, _context: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
68        Poll::Ready(Ok(()))
69    }
70
71    fn call(&mut self, request: Request<B>) -> Self::Future {
72        let policy = self.policy.clone();
73        let handler = Arc::clone(&self.handler);
74        let (parts, body) = request.into_parts();
75        #[cfg(feature = "tracing")]
76        let route = parts.uri.path().to_owned();
77
78        Box::pin(async move {
79            let body = match Limited::new(body, policy.body_limit_bytes())
80                .collect()
81                .await
82            {
83                Ok(body) => body.to_bytes(),
84                Err(error) if error.is::<LengthLimitError>() => {
85                    return Ok(empty_response(StatusCode::PAYLOAD_TOO_LARGE));
86                }
87                Err(_) => return Ok(empty_response(StatusCode::BAD_REQUEST)),
88            };
89
90            #[cfg(feature = "tracing")]
91            let response = tracing::Instrument::instrument(
92                respond(policy, handler, parts.headers, body),
93                request_span(&route),
94            )
95            .await;
96            #[cfg(not(feature = "tracing"))]
97            let response = respond(policy, handler, parts.headers, body).await;
98
99            Ok(response)
100        })
101    }
102}
103
104async fn respond<H>(
105    policy: WebhookServiceBuilder,
106    handler: Arc<H>,
107    headers: HeaderMap,
108    body: Bytes,
109) -> WebhookResponse
110where
111    H: WebhookHandler,
112{
113    if !policy.authenticates(&headers) {
114        #[cfg(feature = "tracing")]
115        tracing::warn!("rejected unauthenticated pretix webhook request");
116        let mut response = empty_response(StatusCode::UNAUTHORIZED);
117        response.headers_mut().insert(
118            header::WWW_AUTHENTICATE,
119            http::HeaderValue::from_static("Basic realm=\"pretix-webhook\""),
120        );
121        return response;
122    }
123
124    let event = match serde_json::from_slice::<WebhookEvent>(&body) {
125        Ok(event) => event,
126        Err(error) => {
127            #[cfg(feature = "tracing")]
128            tracing::warn!(%error, "rejected malformed pretix webhook payload");
129            #[cfg(not(feature = "tracing"))]
130            let _ = error;
131            return empty_response(StatusCode::BAD_REQUEST);
132        }
133    };
134
135    #[cfg(feature = "tracing")]
136    record_identity(&event);
137
138    if !policy.allows(&event) {
139        #[cfg(feature = "tracing")]
140        tracing::debug!("rejected filtered pretix webhook event");
141        return empty_response(StatusCode::NOT_FOUND);
142    }
143
144    #[cfg(feature = "tracing")]
145    tracing::info!("received pretix webhook");
146
147    match handler.handle(event).await {
148        Ok(()) => empty_response(StatusCode::NO_CONTENT),
149        Err(error) => failed_response(error),
150    }
151}
152
153fn failed_response(error: impl Display) -> WebhookResponse {
154    #[cfg(feature = "tracing")]
155    tracing::error!(%error, "pretix webhook handler failed");
156    #[cfg(not(feature = "tracing"))]
157    let _ = error;
158    empty_response(StatusCode::INTERNAL_SERVER_ERROR)
159}
160
161fn empty_response(status: StatusCode) -> WebhookResponse {
162    let mut response = Response::new(Empty::new());
163    *response.status_mut() = status;
164    response
165}
166
167#[cfg(feature = "tracing")]
168fn request_span(route: &str) -> tracing::Span {
169    use tracing::field::Empty;
170
171    tracing::info_span!(
172        "pretix_webhook",
173        route,
174        notification_id = Empty,
175        action = Empty,
176        organizer = Empty,
177        pretix_event = Empty,
178        kind = Empty,
179    )
180}
181
182#[cfg(feature = "tracing")]
183fn record_identity(event: &WebhookEvent) {
184    let span = tracing::Span::current();
185    span.record("notification_id", event.notification_id());
186    span.record("action", event.action());
187    span.record("organizer", event.organizer_slug());
188    span.record("pretix_event", event.event_slug());
189    span.record("kind", tracing::field::debug(event.kind()));
190}
pretix_webhook/service.rs

pretix_webhook/
service.rs
