use secure_compare

e2e_test/webhook/check_1.slt.part

@@ -12,3 +12,8 @@ query TT
 select data ->> 'source', data->> 'auth_algo' from rudderstack;
 ----
 rudderstack plain
+
+query TT
+select data ->> 'source', data->> 'auth_algo' from segment_encode_hmac;
+----
+segment encode_hmac

e2e_test/webhook/check_2.slt.part

@@ -15,3 +15,9 @@ select data ->> 'source', data->> 'auth_algo' from rudderstack;
 ----
 rudderstack plain
 rudderstack plain
+
+query TT
+select data ->> 'source', data->> 'auth_algo' from segment_encode_hmac;
+----
+segment encode_hmac
+segment encode_hmac

e2e_test/webhook/check_3.slt.part

@@ -17,4 +17,11 @@ select data ->> 'source', data->> 'auth_algo' from rudderstack;
 ----
 rudderstack plain
 rudderstack plain
-rudderstack plain
+rudderstack plain
+
+query TT
+select data ->> 'source', data->> 'auth_algo' from segment_encode_hmac;
+----
+segment encode_hmac
+segment encode_hmac
+segment encode_hmac

e2e_test/webhook/create_table.slt.part

@@ -4,7 +4,7 @@ create table rudderstack (
 ) WITH (
   connector = 'webhook',
 ) VALIDATE SECRET test_secret AS secure_compare(
-  headers->>'Authorization',
+  headers->>'authorization',
   test_secret
 );
 
@@ -14,8 +14,8 @@ create table github_sha1 (
 ) WITH (
   connector = 'webhook',
 ) VALIDATE SECRET test_secret AS secure_compare(
-  headers->>'X-Hub-Signature',
-  hmac(test_secret, data, 'sha1')
+  headers->>'x-hub-signature',
+  hmac(test_secret, data, 'sha1', 'sha1=')
 );
 
 statement ok
@@ -24,7 +24,16 @@ create table github_sha256 (
 ) WITH (
   connector = 'webhook',
 ) VALIDATE SECRET test_secret AS secure_compare(
-  headers->>'X-Hub-Signature-256',
-  hmac(test_secret, data, 'sha256')
+  headers->>'x-hub-signature-256',
+  hmac(test_secret, data, 'sha256', 'sha256=')
 );
 
+statement ok
+create table segment_sha1 (
+  data JSONB
+) WITH (
+  connector = 'webhook',
+) VALIDATE SECRET test_secret AS secure_compare(
+  headers->>'x-signature',
+  hmac(test_secret, data, 'sha1')
+);

e2e_test/webhook/drop_table.slt.part

@@ -1,9 +1,12 @@
 
+statement ok
+DROP TABLE segment_encode_hmac;
+
 statement ok
 DROP TABLE github_sha256;
 
 statement ok
 DROP TABLE github_sha1;
 
 statement ok
-DROP TABLE rudderstack;
+DROP TABLE rudderstack;

e2e_test/webhook/sender.py

@@ -21,14 +21,14 @@
 SERVER_URL = "http://127.0.0.1:8080/message/root/dev/public/"
 
 
-def generate_signature_hmac(secret, payload, auth_algo):
+def generate_signature_hmac(secret, payload, auth_algo, prefix):
     secret_bytes = bytes(secret, 'utf-8')
     payload_bytes = bytes(payload, 'utf-8')
     signature = ""
     if auth_algo == "sha1":
-        signature = "sha1=" + hmac.new(secret_bytes, payload_bytes, digestmod=hashlib.sha1).hexdigest()
+        signature = prefix + hmac.new(secret_bytes, payload_bytes, digestmod=hashlib.sha1).hexdigest()
     elif auth_algo == "sha256":
-        signature = "sha256=" + hmac.new(secret_bytes, payload_bytes, digestmod=hashlib.sha256).hexdigest()
+        signature = prefix + hmac.new(secret_bytes, payload_bytes, digestmod=hashlib.sha256).hexdigest()
     else:
         print("Unsupported auth type")
         sys.exit(1)
@@ -53,7 +53,7 @@ def send_github_sha1(secret):
     url = SERVER_URL + "github_sha1"
 
     payload_json = json.dumps(payload)
-    signature = generate_signature_hmac(secret, payload_json, 'sha1')
+    signature = generate_signature_hmac(secret, payload_json, 'sha1', "sha1=")
     # Webhook message headers
     headers = {
         "Content-Type": "application/json",
@@ -69,7 +69,7 @@ def send_github_sha256(secret):
     url = SERVER_URL + "github_sha256"
 
     payload_json = json.dumps(payload)
-    signature = generate_signature_hmac(secret, payload_json, 'sha256')
+    signature = generate_signature_hmac(secret, payload_json, 'sha256', "sha256=")
     # Webhook message headers
     headers = {
         "Content-Type": "application/json",
@@ -79,6 +79,7 @@ def send_github_sha256(secret):
 
 
 def send_rudderstack(secret):
+    # apply to both rudderstack and AWS EventBridge
     payload = message
     payload['source'] = "rudderstack"
     payload['auth_algo'] = "plain"
@@ -94,6 +95,23 @@ def send_rudderstack(secret):
     send_webhook(url, headers, payload_json)
 
 
+def send_segment_encode_hmac(secret):
+    # apply to both rudderstack and AWS EventBridge
+    payload = message
+    payload['source'] = "segment"
+    payload['auth_algo'] = "encode_hmac"
+    url = SERVER_URL + "segment_encode_hmac"
+
+    payload_json = json.dumps(payload)
+    signature = generate_signature_hmac(secret, payload_json, 'sha1', '')
+    # Webhook message headers
+    headers = {
+        "Content-Type": "application/json",
+        "x-signature": 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")

e2e_test/webhook/webhook_source.slt

@@ -16,6 +16,7 @@ python3 e2e_test/webhook/sender.py --secret TEST_WEBHOOK
 
 sleep 3s
 
+# TODO(kexiang): will use a script to take place of check_1, check_2, check_3
 include ./check_1.slt.part
 
 # insert again

proto/catalog.proto

@@ -96,8 +96,7 @@ message StreamSourceInfo {
 
 message WebhookSourceInfo {
   secret.SecretRef secret_ref = 1;
-  string header_key = 2;
-  expr.ExprNode signature_expr = 3;
+  expr.ExprNode signature_expr = 2;
 }
 
 message Source {

proto/expr.proto

@@ -198,6 +198,7 @@ message ExprNode {
     QUOTE_LITERAL = 330;
     QUOTE_NULLABLE = 331;
     HMAC = 332;
+    SECURE_COMPARE = 333;
 
     // Unary operators
     NEG = 401;

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

@@ -20,35 +20,53 @@ use sha2::Sha256;
 
 #[function("hmac(bytea, bytea, varchar) -> bytea")]
 pub fn hmac(secret: &[u8], payload: &[u8], sha_type: &str) -> Box<[u8]> {
+    hmac_with_prefix(secret, payload, sha_type, "")
+}
+
+#[function("hmac(bytea, bytea, varchar, varchar) -> bytea")]
+pub fn hmac_with_prefix(secret: &[u8], payload: &[u8], sha_type: &str, prefix: &str) -> Box<[u8]> {
     if sha_type == "sha1" {
-        sha1_hmac(secret, payload)
+        sha1_hmac(secret, payload, prefix)
     } else if sha_type == "sha256" {
-        sha256_hmac(secret, payload)
+        sha256_hmac(secret, payload, prefix)
     } else {
         panic!("Unsupported SHA type: {}", sha_type)
     }
 }
 
-fn sha256_hmac(secret: &[u8], payload: &[u8]) -> Box<[u8]> {
+fn sha256_hmac(secret: &[u8], payload: &[u8], prefix: &str) -> Box<[u8]> {
     let mut mac = Hmac::<Sha256>::new_from_slice(secret).expect("HMAC can take key of any size");
 
     mac.update(payload);
 
     let result = mac.finalize();
     let code_bytes = result.into_bytes();
-    let computed_signature = format!("sha256={}", encode(code_bytes));
-    computed_signature.as_bytes().into()
+    if prefix.is_empty() {
+        code_bytes.as_slice().into()
+    } else {
+        let computed_signature = format!("{}{}", prefix, encode(code_bytes));
+        computed_signature.as_bytes().into()
+    }
 }
 
-fn sha1_hmac(secret: &[u8], payload: &[u8]) -> Box<[u8]> {
+fn sha1_hmac(secret: &[u8], payload: &[u8], prefix: &str) -> Box<[u8]> {
     let mut mac = Hmac::<Sha1>::new_from_slice(secret).expect("HMAC can take key of any size");
 
     mac.update(payload);
 
     let result = mac.finalize();
     let code_bytes = result.into_bytes();
-    let computed_signature = format!("sha1={}", encode(code_bytes));
-    computed_signature.as_bytes().into()
+    if prefix.is_empty() {
+        code_bytes.as_slice().into()
+    } else {
+        let computed_signature = format!("{}{}", prefix, encode(code_bytes));
+        computed_signature.as_bytes().into()
+    }
+}
+
+#[function("secure_compare(bytea, bytea) -> boolean")]
+pub fn secure_compare(left: &[u8], right: &[u8]) -> bool {
+    left == right
 }
 
 #[cfg(test)]

src/frontend/src/binder/expr/function/builtin_scalar.rs

@@ -295,6 +295,16 @@ impl Binder {
                 ("encrypt", raw_call(ExprType::Encrypt)),
                 ("decrypt", raw_call(ExprType::Decrypt)),
                 ("hmac", raw_call(ExprType::Hmac)),
+                ("secure_compare",guard_by_len(2, raw(|_binder, mut inputs| {
+                    // Similar to `cast` from string, return type is set explicitly rather than inferred.
+                    if !inputs[0].is_untyped() && inputs[0].return_type() == DataType::Varchar {
+                        FunctionCall::cast_mut(&mut inputs[0], DataType::Bytea, CastContext::Explicit)?;
+                    };
+                    if !inputs[1].is_untyped() && inputs[1].return_type() == DataType::Varchar {
+                        FunctionCall::cast_mut(&mut inputs[1], DataType::Bytea, CastContext::Explicit)?;
+                    };
+                    Ok(FunctionCall::new_unchecked(ExprType::SecureCompare , inputs, DataType::Boolean).into())
+                }))),
                 ("left", raw_call(ExprType::Left)),
                 ("right", raw_call(ExprType::Right)),
                 ("inet_aton", raw_call(ExprType::InetAton)),

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

@@ -90,10 +90,12 @@ 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::Bytea).into())
-                    } else if ident.real_value() == ctx.column_name {
+                    if ident.real_value() == "headers".to_string() {
+                        Ok(InputRef::new(0, DataType::Jsonb).into())
+                    } else if ident.real_value() == ctx.secret_name {
                         Ok(InputRef::new(1, DataType::Bytea).into())
+                    } else if ident.real_value() == ctx.column_name {
+                        Ok(InputRef::new(2, DataType::Bytea).into())
                     } else {
                         Err(
                             ErrorCode::ItemNotFound(format!("Unknown arg: {}", ident.real_value()))

src/frontend/src/expr/pure.rs

@@ -233,6 +233,7 @@ impl ExprVisitor for ImpureAnalyzer {
             | Type::Sha384
             | Type::Sha512
             | Type::Hmac
+            | Type::SecureCompare
             | Type::Decrypt
             | Type::Encrypt
             | Type::Tand

src/frontend/src/handler/create_table.rs

@@ -1482,7 +1482,6 @@ fn bind_webhook_info(
 
     let WebhookSourceInfo {
         secret_ref,
-        header_key,
         signature_expr,
     } = webhook_info;
 
@@ -1509,7 +1508,6 @@ fn bind_webhook_info(
 
     let pb_webhook_info = PbWebhookSourceInfo {
         secret_ref: Some(pb_secret_ref),
-        header_key,
         signature_expr: Some(expr.to_expr_proto()),
     };
 

src/frontend/src/optimizer/plan_expr_visitor/strong.rs

@@ -222,6 +222,7 @@ impl Strong {
             | ExprType::Sha384
             | ExprType::Sha512
             | ExprType::Hmac
+            | ExprType::SecureCompare
             | ExprType::Left
             | ExprType::Right
             | ExprType::Format