kellnr_webhooks/
endpoints.rs

1use axum::Json;
2use axum::extract::{Path, State};
3use axum::http::StatusCode;
4use kellnr_common::webhook::Webhook;
5use kellnr_http::authz::require_admin;
6use kellnr_http::error::{ApiError, ApiResult};
7use kellnr_http::identity::TokenUser;
8use kellnr_http::state::{DbState, SettingsState};
9use tracing::{trace, warn};
10use uuid::Uuid;
11
12use crate::client::{Delivery, DeliveryError, WebhookClient, truncated_body};
13use crate::signature::decode_secret;
14use crate::target::{is_blocked_url, validate_callback_url};
15use crate::types;
16
17/// Register a new webhook (admin only)
18#[utoipa::path(
19    post,
20    path = "/",
21    tag = "webhooks",
22    request_body = types::RegisterWebhookRequest,
23    responses(
24        (status = 200, description = "Webhook registered successfully", body = types::RegisterWebhookResponse),
25        (status = 400, description = "Invalid callback URL, target address not permitted or invalid secret"),
26        (status = 401, description = "Admin access required")
27    ),
28    security(("cargo_token" = []))
29)]
30pub async fn register_webhook(
31    user: TokenUser,
32    State(db): DbState,
33    State(settings): SettingsState,
34    Json(input): Json<types::RegisterWebhookRequest>,
35) -> ApiResult<Json<types::RegisterWebhookResponse>> {
36    trace!(user = %user.name, event = ?input.event, "Registering webhook");
37    require_admin(&user)?;
38
39    let url = validate_callback_url(&input.callback_url)
40        .map_err(|e| ApiError::new(&e, StatusCode::BAD_REQUEST))?;
41    if is_blocked_url(&url, settings.webhooks.block_private_targets) {
42        return Err(ApiError::new(
43            "Callback URL target address is not permitted",
44            StatusCode::BAD_REQUEST,
45        ));
46    }
47    if let Some(secret) = &input.secret {
48        decode_secret(secret).map_err(|e| ApiError::new(&e, StatusCode::BAD_REQUEST))?;
49    }
50
51    let id = db
52        .register_webhook(Webhook {
53            id: None,
54            event: input.event,
55            callback_url: input.callback_url,
56            name: input.name,
57            secret: input.secret,
58        })
59        .await?;
60
61    Ok(Json(types::RegisterWebhookResponse { id }))
62}
63
64/// Get a webhook by ID (admin only)
65#[utoipa::path(
66    get,
67    path = "/{id}",
68    tag = "webhooks",
69    params(
70        ("id" = String, Path, description = "Webhook ID")
71    ),
72    responses(
73        (status = 200, description = "Webhook details", body = types::GetWebhookResponse),
74        (status = 401, description = "Admin access required")
75    ),
76    security(("cargo_token" = []))
77)]
78pub async fn get_webhook(
79    user: TokenUser,
80    Path(id): Path<String>,
81    State(db): DbState,
82) -> ApiResult<Json<types::GetWebhookResponse>> {
83    trace!(user = %user.name, webhook_id = %id, "Getting webhook");
84    require_admin(&user)?;
85
86    Ok(Json(db.get_webhook(&id).await?.into()))
87}
88
89/// List all webhooks (admin only)
90#[utoipa::path(
91    get,
92    path = "/",
93    tag = "webhooks",
94    responses(
95        (status = 200, description = "List of all webhooks", body = types::GetAllWebhooksResponse),
96        (status = 401, description = "Admin access required")
97    ),
98    security(("cargo_token" = []))
99)]
100pub async fn get_all_webhooks(
101    user: TokenUser,
102    State(db): DbState,
103) -> ApiResult<Json<types::GetAllWebhooksResponse>> {
104    trace!(user = %user.name, "Listing all webhooks");
105    require_admin(&user)?;
106
107    let w = db.get_all_webhooks().await?;
108    Ok(Json(types::GetAllWebhooksResponse(
109        w.into_iter().map(Into::into).collect(),
110    )))
111}
112
113/// Delete a webhook (admin only)
114#[utoipa::path(
115    delete,
116    path = "/{id}",
117    tag = "webhooks",
118    params(
119        ("id" = String, Path, description = "Webhook ID")
120    ),
121    responses(
122        (status = 200, description = "Webhook deleted successfully"),
123        (status = 401, description = "Admin access required")
124    ),
125    security(("cargo_token" = []))
126)]
127pub async fn delete_webhook(
128    user: TokenUser,
129    Path(id): Path<String>,
130    State(db): DbState,
131) -> ApiResult<()> {
132    trace!(user = %user.name, webhook_id = %id, "Deleting webhook");
133    require_admin(&user)?;
134
135    db.delete_webhook(&id).await?;
136    Ok(())
137}
138
139/// Test a webhook by sending a test payload (admin only)
140#[utoipa::path(
141    post,
142    path = "/{id}/test",
143    tag = "webhooks",
144    params(
145        ("id" = String, Path, description = "Webhook ID")
146    ),
147    responses(
148        (status = 200, description = "Test payload sent successfully"),
149        (status = 400, description = "Invalid callback URL or target address not permitted"),
150        (status = 401, description = "Admin access required"),
151        (status = 500, description = "Webhook callback request failed"),
152        (status = 502, description = "Webhook callback returned a status other than 2xx, which is contained in the error message")
153    ),
154    security(("cargo_token" = []))
155)]
156pub async fn test_webhook(
157    user: TokenUser,
158    Path(id): Path<String>,
159    State(db): DbState,
160    State(settings): SettingsState,
161) -> ApiResult<()> {
162    trace!(user = %user.name, webhook_id = %id, "Testing webhook");
163    require_admin(&user)?;
164
165    let w = db.get_webhook(&id).await?;
166    let client = WebhookClient::new(&settings.webhooks)
167        .map_err(|e| ApiError::from_err(&e, StatusCode::INTERNAL_SERVER_ERROR))?;
168    let msg_id = format!("msg_test_{}", Uuid::new_v4().simple());
169    let delivery = Delivery {
170        id: &msg_id,
171        payload: "Test Payload",
172        secret: w.secret.as_deref(),
173    };
174    let resp = client
175        .post(&w.callback_url, &delivery)
176        .await
177        .map_err(|e| delivery_error_to_api_error(&id, &e))?;
178
179    let status = resp.status();
180    if status.as_u16() < 300 {
181        return Ok(());
182    }
183
184    warn!(
185        webhook_id = %id,
186        "Webhook test callback returned status {status}. Body: {:?}",
187        truncated_body(resp).await
188    );
189    // Only the status code is passed on in the message, the response body of
190    // the callback host is never returned to the caller. The callback status
191    // is not used as the response status, as e.g. a 401 of the callback would
192    // look like a failed login to Kellnr.
193    Err(ApiError::new(
194        &format!("Webhook callback returned status {}", status.as_u16()),
195        StatusCode::BAD_GATEWAY,
196    ))
197}
198
199fn delivery_error_to_api_error(webhook_id: &str, err: &DeliveryError) -> ApiError {
200    match err {
201        DeliveryError::InvalidUrl(_)
202        | DeliveryError::BlockedTarget
203        | DeliveryError::InvalidSecret(_) => {
204            ApiError::new(&err.to_string(), StatusCode::BAD_REQUEST)
205        }
206        DeliveryError::Request(_) | DeliveryError::Payload(_) => {
207            warn!(webhook_id, "Webhook test callback failed: {err}");
208            ApiError::new(
209                "Webhook callback request failed",
210                StatusCode::INTERNAL_SERVER_ERROR,
211            )
212        }
213    }
214}
215
216#[cfg(test)]
217mod endpoint_tests {
218    use std::sync::Arc;
219
220    use axum::Router;
221    use axum::body::{Body, to_bytes};
222    use axum::http::{HeaderMap, Request};
223    use axum::response::Response;
224    use axum::routing::{delete, get, post};
225    use hyper::header;
226    use kellnr_common::account_name::UserName;
227    use kellnr_common::plain_token::PlainToken;
228    use kellnr_common::webhook::WebhookEvent;
229    use kellnr_db::{ConString, Database, DbProvider, SqliteConString, UserFlags};
230    use kellnr_http::state::AppStateData;
231    use kellnr_http::test_support::closed_port_url;
232    use serde::de::DeserializeOwned;
233    use tower::ServiceExt;
234
235    use super::*;
236    use crate::tests::{get_test_listener, get_test_listener_with_response};
237    use crate::types::{GetAllWebhooksResponse, GetWebhookResponse, RegisterWebhookResponse};
238
239    const ADMIN_TOKEN: &str = "jkjkashd09128u3019283o1i3j";
240    const NON_ADMIN_TOKEN: &str = "kjas09ed8o1i23k1jh";
241
242    #[tokio::test]
243    async fn test_register_webhook() {
244        let (router, db) = get_app().await;
245
246        let payload = "{\"type\": \"crate_add\", \"callback_url\": \"http://my-service:8000\"}";
247
248        let response = router
249            .clone()
250            .oneshot(
251                Request::post("/api/v1/webhook")
252                    .header(header::CONTENT_TYPE, "application/json")
253                    .header(header::AUTHORIZATION, ADMIN_TOKEN)
254                    .body(Body::from(payload))
255                    .unwrap(),
256            )
257            .await
258            .unwrap();
259
260        assert_eq!(200, response.status().as_u16());
261
262        let response: RegisterWebhookResponse = parse_response(response).await;
263
264        let webhook = db.get_webhook(&response.id).await.unwrap();
265        assert_eq!(webhook.event, WebhookEvent::CrateAdd);
266        assert_eq!(webhook.callback_url, "http://my-service:8000".to_string());
267    }
268
269    #[tokio::test]
270    async fn test_register_webhook_invalid_callback_url() {
271        let (router, db) = get_app().await;
272
273        for url in [
274            "not a url",
275            "file:///etc/passwd",
276            "ftp://my-service",
277            "http://169.254.169.254/latest/meta-data",
278        ] {
279            let payload = format!("{{\"type\": \"crate_add\", \"callback_url\": \"{url}\"}}");
280
281            let response = router
282                .clone()
283                .oneshot(
284                    Request::post("/api/v1/webhook")
285                        .header(header::CONTENT_TYPE, "application/json")
286                        .header(header::AUTHORIZATION, ADMIN_TOKEN)
287                        .body(Body::from(payload))
288                        .unwrap(),
289                )
290                .await
291                .unwrap();
292
293            assert_eq!(400, response.status().as_u16(), "{url}");
294        }
295
296        assert!(db.get_all_webhooks().await.unwrap().is_empty());
297    }
298
299    #[tokio::test]
300    async fn test_register_webhook_with_secret_is_never_returned() {
301        let (router, db) = get_app().await;
302        let secret = "whsec_MfKQ9r8GKYqrTwjUPD8ILPZIo2LaLaSw";
303
304        let payload = format!(
305            "{{\"type\": \"crate_add\", \"callback_url\": \"http://my-service:8000\", \"secret\": \"{secret}\"}}"
306        );
307        let response = router
308            .clone()
309            .oneshot(
310                Request::post("/api/v1/webhook")
311                    .header(header::CONTENT_TYPE, "application/json")
312                    .header(header::AUTHORIZATION, ADMIN_TOKEN)
313                    .body(Body::from(payload))
314                    .unwrap(),
315            )
316            .await
317            .unwrap();
318        assert_eq!(200, response.status().as_u16());
319        let response: RegisterWebhookResponse = parse_response(response).await;
320        let id = response.id;
321        assert_eq!(
322            Some(secret.to_string()),
323            db.get_webhook(&id).await.unwrap().secret
324        );
325
326        for uri in [
327            format!("/api/v1/webhook/{id}"),
328            "/api/v1/webhook".to_string(),
329        ] {
330            let response = router
331                .clone()
332                .oneshot(
333                    Request::get(uri)
334                        .header(header::AUTHORIZATION, ADMIN_TOKEN)
335                        .body(Body::empty())
336                        .unwrap(),
337                )
338                .await
339                .unwrap();
340            assert_eq!(200, response.status().as_u16());
341            let body = to_bytes(response.into_body(), usize::MAX).await.unwrap();
342            let body = String::from_utf8_lossy(&body);
343            assert!(!body.contains("MfKQ9r8GKYqrTwjUPD8ILPZIo2LaLaSw"));
344            assert!(!body.contains("\"secret\""));
345            assert!(body.contains("\"has_secret\":true"));
346        }
347    }
348
349    #[tokio::test]
350    async fn test_register_webhook_invalid_secret() {
351        let (router, db) = get_app().await;
352
353        let payload = "{\"type\": \"crate_add\", \"callback_url\": \"http://my-service:8000\", \"secret\": \"too-short\"}";
354        let response = router
355            .clone()
356            .oneshot(
357                Request::post("/api/v1/webhook")
358                    .header(header::CONTENT_TYPE, "application/json")
359                    .header(header::AUTHORIZATION, ADMIN_TOKEN)
360                    .body(Body::from(payload))
361                    .unwrap(),
362            )
363            .await
364            .unwrap();
365
366        assert_eq!(400, response.status().as_u16());
367        assert!(db.get_all_webhooks().await.unwrap().is_empty());
368    }
369
370    #[tokio::test]
371    async fn test_register_webhook_non_admin() {
372        let (router, _) = get_app().await;
373
374        let payload = "{\"type\": \"crate_add\", \"callback_url\": \"http://my-service:8000\"}";
375
376        let response = router
377            .clone()
378            .oneshot(
379                Request::post("/api/v1/webhook")
380                    .header(header::CONTENT_TYPE, "application/json")
381                    .header(header::AUTHORIZATION, NON_ADMIN_TOKEN)
382                    .body(Body::from(payload))
383                    .unwrap(),
384            )
385            .await
386            .unwrap();
387
388        assert_eq!(403, response.status().as_u16());
389    }
390
391    #[tokio::test]
392    async fn test_get_webhook() {
393        let (router, db) = get_app().await;
394
395        let id = db.register_webhook(sample_webhook()).await.unwrap();
396
397        let response = router
398            .clone()
399            .oneshot(
400                Request::get(format!("/api/v1/webhook/{id}"))
401                    .header(header::AUTHORIZATION, ADMIN_TOKEN)
402                    .body(Body::empty())
403                    .unwrap(),
404            )
405            .await
406            .unwrap();
407
408        assert_eq!(200, response.status().as_u16());
409
410        let response: GetWebhookResponse = parse_response(response).await;
411
412        assert_eq!(response.event, WebhookEvent::CrateUpdate);
413        assert_eq!(response.callback_url, sample_webhook().callback_url);
414        assert_eq!(response.name, sample_webhook().name);
415    }
416
417    #[tokio::test]
418    async fn test_get_webhook_non_admin() {
419        let (router, db) = get_app().await;
420
421        let id = db.register_webhook(sample_webhook()).await.unwrap();
422
423        let response = router
424            .clone()
425            .oneshot(
426                Request::get(format!("/api/v1/webhook/{id}"))
427                    .header(header::AUTHORIZATION, NON_ADMIN_TOKEN)
428                    .body(Body::empty())
429                    .unwrap(),
430            )
431            .await
432            .unwrap();
433
434        assert_eq!(403, response.status().as_u16());
435    }
436
437    #[tokio::test]
438    async fn test_get_all_webhooks() {
439        let (router, db) = get_app().await;
440
441        let mut ids = vec![];
442        for _ in 0..5 {
443            ids.push(db.register_webhook(sample_webhook()).await.unwrap());
444        }
445
446        let response = router
447            .clone()
448            .oneshot(
449                Request::get("/api/v1/webhook")
450                    .header(header::AUTHORIZATION, ADMIN_TOKEN)
451                    .body(Body::empty())
452                    .unwrap(),
453            )
454            .await
455            .unwrap();
456
457        assert_eq!(200, response.status().as_u16());
458        let response: GetAllWebhooksResponse = parse_response(response).await;
459
460        assert_eq!(5, response.0.len());
461    }
462
463    #[tokio::test]
464    async fn test_get_all_webhooks_non_admin() {
465        let (router, db) = get_app().await;
466
467        let mut ids = vec![];
468        for _ in 0..2 {
469            ids.push(db.register_webhook(sample_webhook()).await.unwrap());
470        }
471
472        let response = router
473            .clone()
474            .oneshot(
475                Request::get("/api/v1/webhook")
476                    .header(header::AUTHORIZATION, NON_ADMIN_TOKEN)
477                    .body(Body::empty())
478                    .unwrap(),
479            )
480            .await
481            .unwrap();
482
483        assert_eq!(403, response.status().as_u16());
484    }
485
486    #[tokio::test]
487    async fn test_delete_webhook() {
488        let (router, db) = get_app().await;
489
490        let id = db.register_webhook(sample_webhook()).await.unwrap();
491
492        let response = router
493            .clone()
494            .oneshot(
495                Request::delete(format!("/api/v1/webhook/{id}"))
496                    .header(header::AUTHORIZATION, ADMIN_TOKEN)
497                    .body(Body::empty())
498                    .unwrap(),
499            )
500            .await
501            .unwrap();
502
503        assert_eq!(200, response.status().as_u16());
504
505        let result = db.get_webhook(&id).await;
506        assert!(result.is_err());
507    }
508
509    #[tokio::test]
510    async fn test_delete_webhook_non_admin() {
511        let (router, db) = get_app().await;
512
513        let id = db.register_webhook(sample_webhook()).await.unwrap();
514
515        let response = router
516            .clone()
517            .oneshot(
518                Request::delete(format!("/api/v1/webhook/{id}"))
519                    .header(header::AUTHORIZATION, NON_ADMIN_TOKEN)
520                    .body(Body::empty())
521                    .unwrap(),
522            )
523            .await
524            .unwrap();
525
526        assert_eq!(403, response.status().as_u16());
527
528        let result = db.get_webhook(&id).await;
529        assert!(result.is_ok());
530    }
531
532    #[tokio::test]
533    async fn test_test_webhook() {
534        let (router, db) = get_app().await;
535        let mut listener = get_test_listener(200).await;
536
537        let mut webhook = sample_webhook();
538        webhook.callback_url = listener.url();
539        let id = db.register_webhook(webhook).await.unwrap();
540
541        let response = router
542            .clone()
543            .oneshot(
544                Request::get(format!("/api/v1/webhook/{id}/test"))
545                    .header(header::AUTHORIZATION, ADMIN_TOKEN)
546                    .body(Body::empty())
547                    .unwrap(),
548            )
549            .await
550            .unwrap();
551
552        assert_eq!(200, response.status().as_u16());
553
554        listener.rx.recv().await.unwrap();
555    }
556
557    #[tokio::test]
558    async fn test_test_webhook_does_not_return_callback_body() {
559        let (router, db) = get_app().await;
560        let body = "internal-callback-details";
561        let listener =
562            get_test_listener_with_response(503, HeaderMap::new(), body.to_string()).await;
563
564        let mut webhook = sample_webhook();
565        webhook.callback_url = listener.url();
566        let id = db.register_webhook(webhook).await.unwrap();
567
568        let response = router
569            .clone()
570            .oneshot(
571                Request::get(format!("/api/v1/webhook/{id}/test"))
572                    .header(header::AUTHORIZATION, ADMIN_TOKEN)
573                    .body(Body::empty())
574                    .unwrap(),
575            )
576            .await
577            .unwrap();
578
579        assert_eq!(502, response.status().as_u16());
580        let response_body = to_bytes(response.into_body(), usize::MAX).await.unwrap();
581        let response_body = String::from_utf8_lossy(&response_body);
582        assert!(!response_body.contains(body));
583        assert!(response_body.contains("503"));
584    }
585
586    #[tokio::test]
587    async fn test_test_webhook_returns_502_for_unauthorized_callback() {
588        let (router, db) = get_app().await;
589        let listener = get_test_listener(401).await;
590        let mut webhook = sample_webhook();
591        webhook.callback_url = listener.url();
592        let id = db.register_webhook(webhook).await.unwrap();
593
594        let response = router
595            .clone()
596            .oneshot(
597                Request::get(format!("/api/v1/webhook/{id}/test"))
598                    .header(header::AUTHORIZATION, ADMIN_TOKEN)
599                    .body(Body::empty())
600                    .unwrap(),
601            )
602            .await
603            .unwrap();
604
605        assert_eq!(502, response.status().as_u16());
606        let body = to_bytes(response.into_body(), usize::MAX).await.unwrap();
607        assert!(String::from_utf8_lossy(&body).contains("401"));
608    }
609
610    #[tokio::test]
611    async fn test_test_webhook_does_not_return_request_error() {
612        let (router, db) = get_app().await;
613
614        let mut webhook = sample_webhook();
615        webhook.callback_url = closed_port_url().await;
616        let id = db.register_webhook(webhook).await.unwrap();
617
618        let response = router
619            .clone()
620            .oneshot(
621                Request::get(format!("/api/v1/webhook/{id}/test"))
622                    .header(header::AUTHORIZATION, ADMIN_TOKEN)
623                    .body(Body::empty())
624                    .unwrap(),
625            )
626            .await
627            .unwrap();
628
629        assert_eq!(500, response.status().as_u16());
630        let response_body = to_bytes(response.into_body(), usize::MAX).await.unwrap();
631        let response_body = String::from_utf8_lossy(&response_body);
632        assert!(!response_body.contains("127.0.0.1"));
633        assert!(response_body.contains("Webhook callback request failed"));
634    }
635
636    #[tokio::test]
637    async fn test_test_webhook_non_admin() {
638        let (router, db) = get_app().await;
639        let mut listener = get_test_listener(200).await;
640
641        let mut webhook = sample_webhook();
642        webhook.callback_url = listener.url();
643        let id = db.register_webhook(webhook).await.unwrap();
644
645        let response = router
646            .clone()
647            .oneshot(
648                Request::get(format!("/api/v1/webhook/{id}/test"))
649                    .header(header::AUTHORIZATION, NON_ADMIN_TOKEN)
650                    .body(Body::empty())
651                    .unwrap(),
652            )
653            .await
654            .unwrap();
655
656        assert_eq!(403, response.status().as_u16());
657
658        assert!(listener.rx.try_recv().is_err());
659    }
660
661    async fn get_app() -> (Router, Arc<Database>) {
662        let con_string = ConString::Sqlite(SqliteConString::new(
663            std::path::Path::new(":memory:"),
664            "admin",
665            Some("token".to_string()),
666            kellnr_settings::SessionAge::from_secs(10),
667        ));
668        let db = Arc::new(Database::new(&con_string, 1).await.unwrap());
669
670        db.add_auth_token(
671            "wh_test_admin",
672            &PlainToken::from(ADMIN_TOKEN),
673            &UserName::new("admin").unwrap(),
674        )
675        .await
676        .unwrap();
677        db.add_user(
678            &UserName::new("wh_non_admin").unwrap(),
679            "na",
680            UserFlags::normal(),
681        )
682        .await
683        .unwrap();
684        db.add_auth_token(
685            "wh_non_admin",
686            &PlainToken::from(NON_ADMIN_TOKEN),
687            &UserName::new("wh_non_admin").unwrap(),
688        )
689        .await
690        .unwrap();
691
692        let state = AppStateData {
693            db: db.clone(),
694            ..kellnr_http::test_support::test_state()
695        };
696
697        let routes = Router::new()
698            .route("/", get(get_all_webhooks))
699            .route("/", post(register_webhook))
700            .route("/{id}", get(get_webhook))
701            .route("/{id}", delete(delete_webhook))
702            .route("/{id}/test", get(test_webhook));
703
704        (
705            Router::new()
706                .nest("/api/v1/webhook", routes)
707                .with_state(state),
708            db,
709        )
710    }
711
712    async fn parse_response<T: DeserializeOwned>(response: Response<Body>) -> T {
713        serde_json::from_slice(&to_bytes(response.into_body(), usize::MAX).await.unwrap()).unwrap()
714    }
715
716    fn sample_webhook() -> Webhook {
717        Webhook {
718            id: None,
719            event: WebhookEvent::CrateUpdate,
720            callback_url: "https://some-callback:8000".to_string(),
721            name: Some("My callback".to_string()),
722            secret: None,
723        }
724    }
725}
kellnr_webhooks/endpoints.rs

kellnr_webhooks/
endpoints.rs
