Verifying Webhook Signatures for Security

import hmac
import hashlib
import json
import os

# In a real application, SECRET_KEY would be loaded from environment variables
# or a secure configuration management system, NOT hardcoded.
WEBHOOK_SECRET = os.environ.get("WEBHOOK_SECRET", "your_strong_webhook_secret_key")

def verify_webhook_signature(payload_raw: bytes, signature_header: str, secret: str) -> bool:
    """
    Verifies the incoming webhook signature against a calculated signature.

    Args:
        payload_raw: The raw request body as bytes.
        signature_header: The signature provided in the request header (e.g., 'X-Hub-Signature' or 'X-Stripe-Signature').
                          Expected format: 't=<timestamp>,v1=<signature>' or 'sha256=<signature>'
        secret: The shared secret key used to sign the webhook.

    Returns:
        True if the signature is valid, False otherwise.
    """
    if not signature_header:
        print("No signature header provided.")
        return False

    # Example for 'sha256=<signature>' style headers (e.g., GitHub, some custom)
    # signature_parts = signature_header.split('=')
    # if len(signature_parts) != 2 or signature_parts[0] != 'sha256':
    #     print(f"Invalid signature header format: {signature_header}")
    #     return False
    # expected_signature = signature_parts[1]

    # Example for 't=<timestamp>,v1=<signature>' style headers (e.g., Stripe)
    # For simplicity, this example directly uses the signature part.
    # In a real scenario, you'd parse timestamp and compare it to prevent replay attacks.
    # This assumes the signature_header directly contains the signature value to check against,
    # or that you've parsed it to get the relevant part.
    # Let's assume for this snippet, signature_header IS the signature value to compare.
    # For a robust solution, you'd parse:
    # e.g., 't=1678886400,v1=abcde12345...'
    # header_parts = dict(part.split('=', 1) for part in signature_header.split(','))
    # timestamp = header_parts.get('t')
    # provided_signature = header_parts.get('v1')
    # if not provided_signature or not timestamp:
    #     print("Missing timestamp or signature in header.")
    #     return False
    provided_signature = signature_header # Simplified for demonstration

    # Calculate the expected signature
    hmac_obj = hmac.new(secret.encode('utf-8'), payload_raw, hashlib.sha256)
    calculated_signature = hmac_obj.hexdigest()

    # Compare the provided signature with the calculated one
    # Use hmac.compare_digest to prevent timing attacks
    is_valid = hmac.compare_digest(provided_signature, calculated_signature)

    if not is_valid:
        print(f"Signature mismatch. Provided: {provided_signature}, Calculated: {calculated_signature}")
    return is_valid

# --- Example Usage (simulating a request) ---
if __name__ == "__main__":
    # Simulate an incoming webhook payload
    event_payload = {
        "id": "evt_test_123",
        "type": "payment_succeeded",
        "data": {"amount": 1000}
    }
    raw_payload_bytes = json.dumps(event_payload, separators=(',', ':')).encode('utf-8')
    # IMPORTANT: The actual raw payload from the request body must be used for signing.
    # Do not re-serialize after parsing, as whitespace changes can invalidate the signature.

    # Calculate a sample signature for demonstration purposes
    # In a real scenario, the sending service would provide this.
    hmac_obj_demo = hmac.new(WEBHOOK_SECRET.encode('utf-8'), raw_payload_bytes, hashlib.sha256)
    simulated_signature = hmac_obj_demo.hexdigest()

    print(f"Simulated payload: {raw_payload_bytes.decode('utf-8')}")
    print(f"Simulated signature: {simulated_signature}")

    # Test with a valid signature
    print("
--- Test Valid Signature ---")
    if verify_webhook_signature(raw_payload_bytes, simulated_signature, WEBHOOK_SECRET):
        print("Webhook signature is VALID.")
    else:
        print("Webhook signature is INVALID.")

    # Test with an invalid payload (tampering)
    print("
--- Test Invalid Payload ---")
    tampered_payload_bytes = json.dumps({"id": "evt_test_123", "type": "payment_failed"}, separators=(',', ':')).encode('utf-8')
    if verify_webhook_signature(tampered_payload_bytes, simulated_signature, WEBHOOK_SECRET):
        print("Webhook signature is VALID (ERROR - should be invalid).")
    else:
        print("Webhook signature is INVALID (Correct).")

    # Test with an invalid secret
    print("
--- Test Invalid Secret ---")
    if verify_webhook_signature(raw_payload_bytes, simulated_signature, "wrong_secret"):
        print("Webhook signature is VALID (ERROR - should be invalid).")
    else:
        print("Webhook signature is INVALID (Correct).")