risingwave_frontend/webhook/
utils.rs

1// Copyright 2024 RisingWave Labs
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use std::collections::HashMap;
16use std::sync::LazyLock;
17
18use anyhow::anyhow;
19use axum::Json;
20use axum::http::{HeaderMap, StatusCode};
21use axum::response::IntoResponse;
22use risingwave_common::log::LogSuppressor;
23use risingwave_common::row::OwnedRow;
24use risingwave_common::secret::LocalSecretManager;
25use risingwave_common::types::JsonbVal;
26use risingwave_pb::expr::ExprNode;
27use serde_json::json;
28use thiserror_ext::AsReport;
29
30use crate::expr::ExprImpl;
31
32static WEBHOOK_ERROR_LOG_SUPPRESSOR: LazyLock<LogSuppressor> =
33    LazyLock::new(LogSuppressor::default);
34
35#[derive(Debug)]
36pub struct WebhookError {
37    err: anyhow::Error,
38    code: StatusCode,
39}
40
41pub(crate) type Result<T> = std::result::Result<T, WebhookError>;
42
43pub(crate) fn err(err: impl Into<anyhow::Error>, code: StatusCode) -> WebhookError {
44    WebhookError {
45        err: err.into(),
46        code,
47    }
48}
49
50impl WebhookError {
51    #[cfg(test)]
52    pub(crate) fn code(&self) -> StatusCode {
53        self.code
54    }
55}
56
57impl From<anyhow::Error> for WebhookError {
58    fn from(value: anyhow::Error) -> Self {
59        WebhookError {
60            err: value,
61            code: StatusCode::INTERNAL_SERVER_ERROR,
62        }
63    }
64}
65
66impl std::fmt::Display for WebhookError {
67    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
68        write!(f, "{}", self.err.as_report())
69    }
70}
71
72impl std::error::Error for WebhookError {}
73
74impl IntoResponse for WebhookError {
75    fn into_response(self) -> axum::response::Response {
76        if let Ok(suppressed_count) = WEBHOOK_ERROR_LOG_SUPPRESSOR.check() {
77            tracing::error!(
78                error = %self.err.as_report(),
79                status = self.code.as_u16(),
80                suppressed_count,
81                "webhook request failed",
82            );
83        }
84
85        let mut resp = Json(json!({
86            "error": format!("{}", self.err.as_report()),
87        }))
88        .into_response();
89        *resp.status_mut() = self.code;
90        resp
91    }
92}
93
94pub(crate) fn header_map_to_json(headers: &HeaderMap) -> JsonbVal {
95    let mut header_map = HashMap::new();
96
97    for (key, value) in headers {
98        let key = key.as_str().to_owned();
99        let value = value.to_str().unwrap_or("").to_owned();
100        header_map.insert(key, value);
101    }
102
103    let json_value = json!(header_map);
104    JsonbVal::from(json_value)
105}
106
107pub(crate) async fn authenticate_webhook_payload(
108    headers_jsonb: JsonbVal,
109    payload: &[u8],
110    webhook_source_info: &risingwave_pb::catalog::WebhookSourceInfo,
111) -> Result<()> {
112    let is_valid = if let Some(signature_expr) = webhook_source_info.signature_expr.clone() {
113        let secret = if let Some(secret_ref) = webhook_source_info.secret_ref {
114            LocalSecretManager::global()
115                .fill_secret(secret_ref)
116                .map_err(|e| err(e, StatusCode::NOT_FOUND))?
117        } else {
118            String::new()
119        };
120        verify_signature(headers_jsonb, secret.as_str(), payload, signature_expr).await?
121    } else {
122        true
123    };
124
125    if !is_valid {
126        return Err(err(
127            anyhow!("Signature verification failed"),
128            StatusCode::UNAUTHORIZED,
129        ));
130    }
131
132    Ok(())
133}
134
135pub(crate) async fn verify_signature(
136    headers_jsonb: JsonbVal,
137    secret: &str,
138    payload: &[u8],
139    signature_expr: ExprNode,
140) -> Result<bool> {
141    let row = OwnedRow::new(vec![
142        Some(headers_jsonb.into()),
143        Some(secret.into()),
144        Some(payload.into()),
145    ]);
146
147    let signature_expr_impl = ExprImpl::from_expr_proto(&signature_expr)
148        .map_err(|e| err(e, StatusCode::INTERNAL_SERVER_ERROR))?;
149
150    let result = signature_expr_impl
151        .eval_row(&row)
152        .await
153        .map_err(|e| err(e, StatusCode::INTERNAL_SERVER_ERROR))?
154        .ok_or_else(|| {
155            err(
156                anyhow!("`SECURE_COMPARE()` failed"),
157                StatusCode::BAD_REQUEST,
158            )
159        })?;
160    Ok(*result.as_bool())
161}
162
163#[cfg(test)]
164mod tests {
165    use axum::body::to_bytes;
166    use axum::http::header::HeaderName;
167
168    use super::*;
169
170    #[tokio::test]
171    async fn test_webhook_error_response() {
172        let response = err(
173            anyhow!("failed to decode webhook payload"),
174            StatusCode::UNPROCESSABLE_ENTITY,
175        )
176        .into_response();
177
178        assert_eq!(response.status(), StatusCode::UNPROCESSABLE_ENTITY);
179        let body = to_bytes(response.into_body(), usize::MAX).await.unwrap();
180        assert_eq!(
181            serde_json::from_slice::<serde_json::Value>(&body).unwrap(),
182            json!({"error": "failed to decode webhook payload"}),
183        );
184    }
185
186    #[test]
187    fn test_header_map_to_json_preserves_header_names() {
188        let mut headers = HeaderMap::new();
189        headers.insert(
190            HeaderName::from_static("x-custom-token"),
191            "abc".parse().unwrap(),
192        );
193
194        let headers_json = header_map_to_json(&headers);
195        let json_value: serde_json::Value =
196            serde_json::from_str(&headers_json.to_string()).unwrap();
197
198        assert_eq!(json_value["x-custom-token"], "abc");
199    }
200}
risingwave_frontend/webhook/utils.rs

risingwave_frontend/webhook/
utils.rs
