Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
@@ -0,0 +1,6 @@
---
title: Improve webhook signature verification error messages
semver_level: patch
---

- When diagnostic metadata is present in the `Stripe-Signature` header, error messages help identify an incorrect signing secret or altered request body.
100 changes: 98 additions & 2 deletions stripe/_webhook.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,8 +2,10 @@
import json
import time
from collections import OrderedDict
from dataclasses import dataclass
from hashlib import sha256
from typing import Any, Dict, Optional, Union
from urllib.parse import quote

# Used for global variables
import stripe # noqa: IMP101
Expand All @@ -19,6 +21,13 @@
"""


@dataclass
class _DebugMetadata:
secret: Optional[str] = None
destination: Optional[str] = None
payload: Optional[str] = None


def build_v1_event(values: Dict[str, Any], requestor: _APIRequestor) -> Event:
"""
Internal helper for centralizing v1 event creation
Expand Down Expand Up @@ -180,8 +189,7 @@ def verify_header(
expected_sig = cls._compute_signature(signed_payload, secret)
if not any(secure_compare(expected_sig, s) for s in signatures):
raise SignatureVerificationError(
"No signatures found matching the expected signature for "
"payload",
cls._signature_error_message(header, str(payload), secret),
header,
payload,
)
Expand All @@ -194,3 +202,91 @@ def verify_header(
)

return True

@classmethod
def _signature_error_message(
cls, header: str, payload: str, secret: str
) -> str:
"""Build an HMAC-mismatch message using optional header metadata."""
metadata = cls._signature_debug_metadata(header)

# If metadata has an asterisk-masked secret, compare against its length and last four characters
masked_secret = metadata.secret or ""
secret_suffix = secret[-4:]
if "*" in masked_secret and (
len(secret) != len(masked_secret)
or secret_suffix != masked_secret[-4:]
):
dashboard_url = "https://dashboard.stripe.com/webhooks"
if metadata.destination is not None:
dashboard_url += "/" + quote(metadata.destination, safe="")
return (
"Webhook signature verification failed. This request may not "
"have originated from Stripe.\n\n"
f"The signing secret you provided is {len(secret)} characters "
f"long and ends in `{secret_suffix}`. However, the signing "
"secret for this destination is "
f"{len(masked_secret)} characters long and looks like "
f"`{masked_secret}`. Ensure you're using the signing secret "
"for this destination, as shown on the Dashboard: "
f"{dashboard_url}\n\n"
"Learn more: https://docs.stripe.com/webhooks/signature"
)

# If metadata includes the truncated payload hash, compare with the payload passed in
expected_hash = (metadata.payload or "").lower()
if len(expected_hash) == 8 and all(
character in "0123456789abcdef" for character in expected_hash
):
actual_hash = sha256(payload.encode("utf-8")).hexdigest()[-8:]
if expected_hash != actual_hash:
return (
"Webhook signature verification failed.\n\n"
"The request body you passed to this function differs "
"from the request body sent by Stripe to this webhook "
"destination.\n"
"The SHA-256 hash of the body should end in "
f"`{expected_hash}`. However, the SHA-256 hash of the body "
f"you provided ends in `{actual_hash}`.\n\n"
"Make sure your code does not reformat or otherwise alter "
"the raw HTTP request body before passing it to this "
"function.\n\n"
"Learn more: https://docs.stripe.com/webhooks/signature"
)

# Fall back to the generic message if no debug metadata available
return (
"No signatures found matching the expected signature for payload"
)

@classmethod
def _signature_debug_metadata(cls, header: str) -> _DebugMetadata:
"""Extract metadata from the stripe-signature header value to help debug verification failures."""
# For each metadata key, extract the first value found in the header
metadata: Dict[str, Optional[str]] = {}
for item in header.split(","):
key, separator, value = item.lstrip().partition("=")
if (
key in ("secret", "destination", "payload")
and key not in metadata
):
metadata[key] = (
cls._safe_metadata_value(value) if separator else None
)
return _DebugMetadata(
metadata.get("secret"),
metadata.get("destination"),
metadata.get("payload"),
)

@staticmethod
def _safe_metadata_value(value: str) -> Optional[str]:
"""Return a bounded UTF-8 metadata value, or None if unusable."""
# Ignore overly long, potentially attacker-provided metadata values
if len(value) > 100:
return None
try:
value.encode("utf-8")
except UnicodeEncodeError:
return None
return value.strip() or None
172 changes: 170 additions & 2 deletions tests/test_webhook.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
import time
from hashlib import sha256

import pytest

Expand Down Expand Up @@ -35,12 +36,19 @@


def generate_header(
payload=DUMMY_WEBHOOK_PAYLOAD, secret=DUMMY_WEBHOOK_SECRET, timestamp=None
payload=DUMMY_WEBHOOK_PAYLOAD,
secret=DUMMY_WEBHOOK_SECRET,
timestamp=None,
debug_metadata=None,
):
"""Thin wrapper around WebhookSignature.generate_signature_header for tests."""
return WebhookSignature.generate_signature_header(
header = WebhookSignature.generate_signature_header(
payload, secret, timestamp
)
for key, value in (debug_metadata or {}).items():
if value is not None:
header += f",{key}={value}"
return header


def _build_header_with_scheme(
Expand Down Expand Up @@ -214,6 +222,166 @@ def test_raise_on_no_valid_signatures_for_payload(self):
DUMMY_WEBHOOK_PAYLOAD, header, DUMMY_WEBHOOK_SECRET
)

def test_diagnose_secret_mismatch(self):
dashboard_url_base = "https://dashboard.stripe.com/webhooks"
cases = [
# Secret and masket secret don't match; show URL with endpoint
{
"masked_secret": "whsec_******************1q84",
"destination": "we_example",
"url": dashboard_url_base + "/we_example",
},
# Missing and blank destinations use the bare Dashboard URL
{
"masked_secret": "whsec_******************1q84",
"destination": "",
"url": dashboard_url_base,
},
# Length mismatch
{"masked_secret": "whsec_******************cret"},
# An attacker-selected mask must not expose more than the last four characters
{"masked_secret": "whsec_*bcdefghijklmnopqrstuvwxyZ"},
]
for case in cases:
secret = case.get("secret", DUMMY_WEBHOOK_SECRET)
secret_suffix = case.get("secret_suffix", secret[-4:])
masked_secret = case.get("masked_secret", "whsec_*******cret")
url = case.get("url", dashboard_url_base)
header = generate_header(
secret="whsec_other_secret",
debug_metadata={
"secret": masked_secret,
"destination": case.get("destination"),
"payload": "00000000",
},
)
with pytest.raises(SignatureVerificationError) as error:
WebhookSignature.verify_header(
DUMMY_WEBHOOK_PAYLOAD, header, secret
)
assert str(error.value) == (
"Webhook signature verification failed. This request may not have originated from Stripe.\n\n"
f"The signing secret you provided is {len(secret)} characters long and ends in `{secret_suffix}`. "
f"However, the signing secret for this destination is {len(masked_secret)} characters long "
f"and looks like `{masked_secret}`. Ensure you're using the signing secret for this "
f"destination, as shown on the Dashboard: {url}\n\n"
"Learn more: https://docs.stripe.com/webhooks/signature"
)

def test_diagnose_payload_mismatch(self):
# User supplies the original body with an extra trailing space.
payload = DUMMY_WEBHOOK_PAYLOAD + " "
expected_hash = sha256(
DUMMY_WEBHOOK_PAYLOAD.encode("utf-8")
).hexdigest()[-8:]
actual_hash = sha256(payload.encode("utf-8")).hexdigest()[-8:]
header = generate_header(
debug_metadata={
"secret": "whsec_*******cret",
"payload": expected_hash.upper(),
}
)
with pytest.raises(SignatureVerificationError) as error:
WebhookSignature.verify_header(
payload, header, DUMMY_WEBHOOK_SECRET
)
assert str(error.value) == (
"Webhook signature verification failed.\n\n"
"The request body you passed to this function differs from the request body sent by Stripe to this webhook destination.\n"
f"The SHA-256 hash of the body should end in `{expected_hash}`. However, the SHA-256 hash of the body you provided ends in `{actual_hash}`.\n\n"
"Make sure your code does not reformat or otherwise alter the raw HTTP request body before passing it to this function.\n\n"
"Learn more: https://docs.stripe.com/webhooks/signature"
)

def test_matching_metadata_falls_back_to_generic_error(self):
for masked_secret, secret in [
("whsec_*******cret", DUMMY_WEBHOOK_SECRET),
# Prefixes and mask widths do not change the fixed four-character comparison.
("other_*******cret", DUMMY_WEBHOOK_SECRET),
(
"whsec_*bcdefghijklmnopqrstuvwxyz",
"whsec_abcdefghijklmnopqrstuvwxyz",
),
("*", "*"),
]:
header = generate_header(
secret="whsec_other_secret",
debug_metadata={"secret": masked_secret},
)
with pytest.raises(SignatureVerificationError) as error:
WebhookSignature.verify_header(
DUMMY_WEBHOOK_PAYLOAD, header, secret
)
assert (
str(error.value)
== "No signatures found matching the expected signature for payload"
)

def test_invalid_secret_metadata_falls_back_to_generic_error(self):
payload_hash = sha256(
DUMMY_WEBHOOK_PAYLOAD.encode("utf-8")
).hexdigest()[-8:]
for masked_secret in [
"unmasked",
"*" * 101,
"*\ud800",
# A blank first secret must not be replaced by a later value.
",secret=whsec_*******abcd",
]:
header = generate_header(
secret="whsec_other_secret",
debug_metadata={
"secret": masked_secret,
"payload": payload_hash,
},
)
with pytest.raises(SignatureVerificationError) as error:
WebhookSignature.verify_header(
DUMMY_WEBHOOK_PAYLOAD, header, DUMMY_WEBHOOK_SECRET
)
assert (
str(error.value)
== "No signatures found matching the expected signature for payload"
)

def test_invalid_payload_metadata_falls_back_to_generic_error(self):
for payload_hash in [
"1234567",
"123456789",
"not-hex!",
"a" * 101,
"1234567\ud800",
# A blank first hash must not be replaced by a later value.
",payload=00000000",
]:
header = generate_header(
secret="whsec_other_secret",
debug_metadata={
"secret": "whsec_*******cret",
"payload": payload_hash,
},
)
with pytest.raises(SignatureVerificationError) as error:
WebhookSignature.verify_header(
DUMMY_WEBHOOK_PAYLOAD, header, DUMMY_WEBHOOK_SECRET
)
assert (
str(error.value)
== "No signatures found matching the expected signature for payload"
)

def test_valid_signature_ignores_conflicting_metadata(self):
header = generate_header(
debug_metadata={
"secret": "*abcd",
"destination": "clidev_example",
"payload": "00000000",
}
)
assert WebhookSignature.verify_header(
DUMMY_WEBHOOK_PAYLOAD, header, DUMMY_WEBHOOK_SECRET
)

def test_raise_on_timestamp_outside_tolerance(self):
header = generate_header(timestamp=int(time.time()) - 15)
with pytest.raises(
Expand Down
Loading