rudderstack

e2e_test/webhook/check_1.slt.part

@@ -6,4 +6,9 @@ github sha256
 query TT
 select data ->> 'source', data->> 'auth_algo' from github_sha1;
 ----
-github sha1
+github sha1
+
+query TT
+select data ->> 'source', data->> 'auth_algo' from rudderstack;
+----
+rudderstack plain

e2e_test/webhook/check_2.slt.part

@@ -8,4 +8,10 @@ query TT
 select data ->> 'source', data->> 'auth_algo' from github_sha1;
 ----
 github sha1
-github sha1
+github sha1
+
+query TT
+select data ->> 'source', data->> 'auth_algo' from rudderstack;
+----
+rudderstack plain
+rudderstack plain

e2e_test/webhook/check_3.slt.part

@@ -10,4 +10,11 @@ select data ->> 'source', data->> 'auth_algo' from github_sha1;
 ----
 github sha1
 github sha1
-github sha1
+github sha1
+
+query TT
+select data ->> 'source', data->> 'auth_algo' from rudderstack;
+----
+rudderstack plain
+rudderstack plain
+rudderstack plain

e2e_test/webhook/create_table.slt.part

@@ -1,3 +1,12 @@
+statement ok
+create table rudderstack (
+  data JSONB
+) WITH (
+  connector = 'webhook',
+) VALIDATE SECRET test_secret AS secure_compare(
+  headers->>'Authorization',
+  test_secret
+);
 
 statement ok
 create table github_sha1 (

e2e_test/webhook/drop_table.slt.part

@@ -5,3 +5,5 @@ DROP TABLE github_sha256;
 statement ok
 DROP TABLE github_sha1;
 
+statement ok
+DROP TABLE rudderstack;

e2e_test/webhook/sender.py

@@ -78,6 +78,22 @@ def send_github_sha256(secret):
     send_webhook(url, headers, payload_json)
 
 
+def send_rudderstack(secret):
+    payload = message
+    payload['source'] = "rudderstack"
+    payload['auth_algo'] = "plain"
+    url = SERVER_URL + "rudderstack"
+
+    payload_json = json.dumps(payload)
+    signature = secret
+    # Webhook message headers
+    headers = {
+        "Content-Type": "application/json",
+        "Authorization": signature  # Custom signature header
+    }
+    send_webhook(url, headers, payload_json)
+
+
 if __name__ == "__main__":
     parser = argparse.ArgumentParser(description="Simulate sending Webhook messages")
     parser.add_argument("--secret", required=True, help="Secret key for generating signature")
@@ -86,3 +102,4 @@ def send_github_sha256(secret):
     # send data
     send_github_sha1(secret)
     send_github_sha256(secret)
+    send_rudderstack(secret)

src/expr/impl/src/scalar/hmac.rs

@@ -18,8 +18,8 @@ use risingwave_expr::function;
 use sha1::Sha1;
 use sha2::Sha256;
 
-#[function("hmac(varchar, bytea, varchar) -> bytea")]
-pub fn hmac(secret: &str, payload: &[u8], sha_type: &str) -> Box<[u8]> {
+#[function("hmac(bytea, bytea, varchar) -> bytea")]
+pub fn hmac(secret: &[u8], payload: &[u8], sha_type: &str) -> Box<[u8]> {
     if sha_type == "sha1" {
         sha1_hmac(secret, payload)
     } else if sha_type == "sha256" {
@@ -29,9 +29,8 @@ pub fn hmac(secret: &str, payload: &[u8], sha_type: &str) -> Box<[u8]> {
     }
 }
 
-fn sha256_hmac(secret: &str, payload: &[u8]) -> Box<[u8]> {
-    let mut mac =
-        Hmac::<Sha256>::new_from_slice(secret.as_bytes()).expect("HMAC can take key of any size");
+fn sha256_hmac(secret: &[u8], payload: &[u8]) -> Box<[u8]> {
+    let mut mac = Hmac::<Sha256>::new_from_slice(secret).expect("HMAC can take key of any size");
 
     mac.update(payload);
 
@@ -41,9 +40,8 @@ fn sha256_hmac(secret: &str, payload: &[u8]) -> Box<[u8]> {
     computed_signature.as_bytes().into()
 }
 
-fn sha1_hmac(secret: &str, payload: &[u8]) -> Box<[u8]> {
-    let mut mac =
-        Hmac::<Sha1>::new_from_slice(secret.as_bytes()).expect("HMAC can take key of any size");
+fn sha1_hmac(secret: &[u8], payload: &[u8]) -> Box<[u8]> {
+    let mut mac = Hmac::<Sha1>::new_from_slice(secret).expect("HMAC can take key of any size");
 
     mac.update(payload);
 

src/frontend/src/binder/expr/mod.rs

@@ -91,7 +91,7 @@ impl Binder {
                     }
                 } else if let Some(ctx) = self.secure_compare_context.as_ref() {
                     if ident.real_value() == ctx.secret_name {
-                        Ok(InputRef::new(0, DataType::Varchar).into())
+                        Ok(InputRef::new(0, DataType::Bytea).into())
                     } else if ident.real_value() == ctx.column_name {
                         Ok(InputRef::new(1, DataType::Bytea).into())
                     } else {

src/frontend/src/webhook/mod.rs

@@ -60,6 +60,10 @@ pub(super) mod handlers {
         Path((user, database, schema, table)): Path<(String, String, String, String)>,
         body: Bytes,
     ) -> Result<()> {
+        println!(
+            "WKXLOG receive something: {:?}, \ndatabase: {}, \ntable: {}, \nheaders: {:?}",
+            body, database, table, headers
+        );
         let session_mgr = SESSION_MANAGER
             .get()
             .expect("session manager has been initialized");
@@ -113,6 +117,8 @@ pub(super) mod handlers {
             .fill_secret(secret_ref.unwrap())
             .map_err(|e| err(e, StatusCode::NOT_FOUND))?;
 
+        println!("WKXLOG header_key: {:?}", header_key);
+
         let signature = headers
             .get(header_key)
             .ok_or_else(|| {
@@ -122,14 +128,18 @@ pub(super) mod handlers {
                 )
             })?
             .as_bytes();
+        println!("WKXLOG signature: {:?}", signature);
+        println!("WKXLOG secret_string: {:?}", secret_string);
 
         let is_valid = verify_signature(
-            secret_string.as_str(),
+            secret_string.as_bytes(),
             body.as_ref(),
             signature_expr.unwrap(),
             signature,
         )
         .await?;
+        println!("WKXLOG is_valid: {:?}", is_valid);
+
         if !is_valid {
             return Err(err(
                 anyhow!("Signature verification failed"),
@@ -162,12 +172,7 @@ pub(super) mod handlers {
 
         let _rsp = handle(session, insert_stmt, Arc::from(""), vec![])
             .await
-            .map_err(|e| {
-                err(
-                    anyhow!(e).context("Failed to insert to table"),
-                    StatusCode::INTERNAL_SERVER_ERROR,
-                )
-            })?;
+            .map_err(|e| anyhow!("failed to insert: {:?}", e))?;
 
         Ok(())
     }

src/frontend/src/webhook/utils.rs

@@ -57,15 +57,17 @@ impl IntoResponse for WebhookError {
 }
 
 pub async fn verify_signature(
-    secret: &str,
+    secret: &[u8],
     payload: &[u8],
     signature_expr: ExprNode,
     signature: &[u8],
 ) -> Result<bool> {
     let row = OwnedRow::new(vec![Some(secret.into()), Some(payload.into())]);
+    println!("WKXLOG signature_expr: {:?}", signature_expr);
 
     let signature_expr_impl = ExprImpl::from_expr_proto(&signature_expr)
         .map_err(|e| err(e, StatusCode::INTERNAL_SERVER_ERROR))?;
+    println!("WKXLOG signature_expr_impl: {:?}", signature_expr_impl);
 
     let result = signature_expr_impl
         .eval_row(&row)