Webhook 和 Slack 集成

import hmac
import hashlib
from typing import Optional
 
 
def verify_langfuse_signature(
    raw_body: str,
    signature_header: str,
    secret: str,
) -> bool:
    """
    Validate a Langfuse webhook/event signature.
 
    Parameters
    ----------
    raw_body : str
        The request body exactly as received (no decoding or reformatting).
    signature_header : str
        The value of the `Langfuse-Signature` header, e.g. "t=1720701136,s=0123abcd...".
    secret : str
        Your Langfuse signing secret.
 
    Returns
    -------
    bool
        True if the signature is valid, otherwise False.
    """
    # Split "t=timestamp,s=signature" into the two expected key/value chunks
    try:
        ts_pair, sig_pair = signature_header.split(",", 1)
    except ValueError:  # wrong format / missing comma
        return False
 
    # Extract values (everything after the first "=")
    if "=" not in ts_pair or "=" not in sig_pair:
        return False
    timestamp = ts_pair.split("=", 1)[1]
    received_sig_hex = sig_pair.split("=", 1)[1]
 
    # Recreate the message and compute the expected HMAC-SHA256 hex digest
    message = f"{timestamp}.{raw_body}".encode("utf-8")
    expected_sig_hex = hmac.new(
        secret.encode("utf-8"), message, hashlib.sha256
    ).hexdigest()
 
    # Use constant-time comparison on the *decoded* byte strings
    try:
        return hmac.compare_digest(
            bytes.fromhex(received_sig_hex), bytes.fromhex(expected_sig_hex)
        )
    except ValueError:  # received_sig_hex isn't valid hex
        return False

import crypto from "crypto";
 
export function verifyLangfuseSignature(
  rawBody: string,
  signatureHeader: string,
  secret: string
): boolean {
  const [tsPair, sigPair] = signatureHeader.split(",");
  if (!tsPair || !sigPair) return false;
 
  const timestamp = tsPair.split("=")[1];
  const receivedSig = sigPair.split("=")[1];
  const expectedSig = crypto
    .createHmac("sha256", secret)
    .update(`${timestamp}.${rawBody}`, "utf8")
    .digest("hex");
 
  return crypto.timingSafeEqual(
    Buffer.from(receivedSig, "hex"),
    Buffer.from(expectedSig, "hex")
  );
}