feat(sdk): add verify_webhook_signature + AgentScoreError.status alias

Closes audit items #12, #17.

verify_webhook_signature: generic HMAC-SHA256 webhook signature verifier,
Stripe-pattern (`t=<unix>,v1=<hex>` header). Useful both when AgentScore
eventually ships outbound webhooks and as a generic helper for merchants
verifying any HMAC-signed webhook source.

Returns VerifyWebhookSignatureResult with `reason` set on failure
(no_signatures / no_timestamp / timestamp_too_old / timestamp_in_future /
signature_mismatch / malformed_header) so callers can differentiate
transient vs permanent failures. Uses hmac.compare_digest for
constant-time comparison.

AgentScoreError.status property mirrors .status_code — polyglot codebases
can use err.status regardless of which SDK raised the error. Both
attributes return the same int.

11 new tests; coverage holds at 97.30% (Tier A).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: vvillait88

agentscore/__init__.py

@@ -30,6 +30,7 @@
     WalletAuthRequiresSigningBody,
     WalletSignerMismatchBody,
 )
+from agentscore.webhooks import VerifyWebhookSignatureResult, verify_webhook_signature
 
 __version__ = _pkg_version("agentscore-py")
 
@@ -60,7 +61,9 @@
     "SessionCreateResponse",
     "SessionPollResponse",
     "VerificationLevel",
+    "VerifyWebhookSignatureResult",
     "WalletAuthRequiresSigningBody",
     "WalletSignerMismatchBody",
     "__version__",
+    "verify_webhook_signature",
 ]

agentscore/errors.py

@@ -3,3 +3,11 @@ def __init__(self, code: str, message: str, status_code: int):
         super().__init__(message)
         self.code = code
         self.status_code = status_code
+
+    @property
+    def status(self) -> int:
+        """Alias for ``status_code`` — parity with node-sdk's attribute name.
+
+        Polyglot codebases can use ``err.status`` regardless of which SDK raised the error.
+        """
+        return self.status_code

agentscore/webhooks.py

@@ -0,0 +1,127 @@
+"""Webhook signature verification — HMAC-SHA256, Stripe-pattern.
+
+Use this when AgentScore (or any service that signs outbound webhooks with this
+convention) sends a webhook to your endpoint. Validates the
+``X-AgentScore-Signature`` (or compatible) header before trusting the payload.
+
+Generic enough to cover any HMAC-signed webhook source: pass the right secret + header
+name. Tolerant of multiple signature versions in the same header
+(``t=...,v1=...`` style).
+"""
+
+from __future__ import annotations
+
+import hashlib
+import hmac
+import time
+from dataclasses import dataclass
+from typing import Literal
+
+
+@dataclass(frozen=True)
+class VerifyWebhookSignatureResult:
+    """Result of :func:`verify_webhook_signature`."""
+
+    valid: bool
+    reason: Literal[
+        "no_signatures",
+        "no_timestamp",
+        "timestamp_too_old",
+        "timestamp_in_future",
+        "signature_mismatch",
+        "malformed_header",
+    ] | None = None
+
+
+def verify_webhook_signature(
+    payload: str | bytes,
+    signature_header: str,
+    secret: str,
+    tolerance_seconds: int = 300,
+    timestamp_key: str = "t",
+    signature_key: str = "v1",
+) -> VerifyWebhookSignatureResult:
+    """Verify an HMAC-SHA256 signed webhook signature, Stripe-compatible.
+
+    Header format: ``t=<unix_seconds>,v1=<hex_hmac>``. The signed payload is
+    ``f"{timestamp}.{raw_body}"``. Returns a result with ``reason`` set on failure so
+    callers can differentiate transient (timestamp drift) from permanent (mismatch).
+
+    Args:
+        payload: Raw request body. MUST be the unparsed body — even one byte of
+            re-serialization breaks the signature. Capture before any JSON parse.
+        signature_header: Value of the signature header from the incoming request.
+        secret: Shared secret the sender uses to sign.
+        tolerance_seconds: Tolerance in seconds for timestamp-replay protection.
+            Default 300 (5 min) per Stripe convention. Set to 0 to disable.
+        timestamp_key: Override the timestamp parameter name. Default ``"t"``.
+        signature_key: Override the signature parameter name. Default ``"v1"``.
+
+    Example::
+
+        from flask import request
+        from agentscore.webhooks import verify_webhook_signature
+
+        @app.post("/webhooks/agentscore")
+        def handle_webhook():
+            result = verify_webhook_signature(
+                payload=request.get_data(),  # raw bytes — DO NOT parse JSON first
+                signature_header=request.headers.get("X-AgentScore-Signature", ""),
+                secret=os.environ["AGENTSCORE_WEBHOOK_SECRET"],
+            )
+            if not result.valid:
+                return {"error": result.reason}, 400
+            event = request.get_json(force=True)
+            # ... handle event ...
+    """
+    parts = [p.strip() for p in signature_header.split(",") if p.strip()]
+    if not parts:
+        return VerifyWebhookSignatureResult(valid=False, reason="no_signatures")
+
+    params: dict[str, list[str]] = {}
+    for p in parts:
+        if "=" not in p:
+            return VerifyWebhookSignatureResult(valid=False, reason="malformed_header")
+        key, _, value = p.partition("=")
+        params.setdefault(key, []).append(value)
+
+    timestamp_str = params.get(timestamp_key, [None])[0]
+    if tolerance_seconds > 0:
+        if not timestamp_str:
+            return VerifyWebhookSignatureResult(valid=False, reason="no_timestamp")
+        try:
+            ts = int(timestamp_str)
+        except ValueError:
+            return VerifyWebhookSignatureResult(valid=False, reason="no_timestamp")
+        now_sec = int(time.time())
+        if ts < now_sec - tolerance_seconds:
+            return VerifyWebhookSignatureResult(valid=False, reason="timestamp_too_old")
+        if ts > now_sec + tolerance_seconds:
+            return VerifyWebhookSignatureResult(valid=False, reason="timestamp_in_future")
+
+    signatures = params.get(signature_key, [])
+    if not signatures:
+        return VerifyWebhookSignatureResult(valid=False, reason="no_signatures")
+
+    payload_bytes = payload.encode("utf-8") if isinstance(payload, str) else payload
+    signed_payload = (
+        f"{timestamp_str}.".encode() + payload_bytes if timestamp_str else payload_bytes
+    )
+
+    expected_hex = hmac.new(secret.encode("utf-8"), signed_payload, hashlib.sha256).hexdigest()
+    expected_bytes = bytes.fromhex(expected_hex)
+
+    for sig_hex in signatures:
+        try:
+            actual_bytes = bytes.fromhex(sig_hex)
+        except ValueError:
+            continue
+        if len(actual_bytes) != len(expected_bytes):
+            continue
+        if hmac.compare_digest(actual_bytes, expected_bytes):
+            return VerifyWebhookSignatureResult(valid=True)
+
+    return VerifyWebhookSignatureResult(valid=False, reason="signature_mismatch")
+
+
+__all__ = ["VerifyWebhookSignatureResult", "verify_webhook_signature"]

tests/test_webhooks.py

@@ -0,0 +1,131 @@
+"""Tests for verify_webhook_signature."""
+
+import hashlib
+import hmac
+import time
+
+from agentscore import verify_webhook_signature
+from agentscore.errors import AgentScoreError
+
+SECRET = "whsec_testsecret"
+
+
+def _sign(payload: str, ts: int, secret: str = SECRET) -> str:
+    signed = f"{ts}.{payload}".encode()
+    return hmac.new(secret.encode("utf-8"), signed, hashlib.sha256).hexdigest()
+
+
+def test_accepts_valid_signature_with_current_timestamp():
+    payload = '{"event":"test"}'
+    ts = int(time.time())
+    sig = _sign(payload, ts)
+    result = verify_webhook_signature(
+        payload=payload,
+        signature_header=f"t={ts},v1={sig}",
+        secret=SECRET,
+    )
+    assert result.valid is True
+
+
+def test_accepts_bytes_payload():
+    payload = b'{"event":"test"}'
+    ts = int(time.time())
+    sig = _sign(payload.decode("utf-8"), ts)
+    result = verify_webhook_signature(
+        payload=payload,
+        signature_header=f"t={ts},v1={sig}",
+        secret=SECRET,
+    )
+    assert result.valid is True
+
+
+def test_rejects_timestamp_older_than_tolerance():
+    payload = "{}"
+    ts = int(time.time()) - 600
+    sig = _sign(payload, ts)
+    result = verify_webhook_signature(
+        payload=payload,
+        signature_header=f"t={ts},v1={sig}",
+        secret=SECRET,
+        tolerance_seconds=300,
+    )
+    assert result.valid is False
+    assert result.reason == "timestamp_too_old"
+
+
+def test_rejects_timestamp_in_future():
+    payload = "{}"
+    ts = int(time.time()) + 600
+    sig = _sign(payload, ts)
+    result = verify_webhook_signature(
+        payload=payload,
+        signature_header=f"t={ts},v1={sig}",
+        secret=SECRET,
+    )
+    assert result.valid is False
+    assert result.reason == "timestamp_in_future"
+
+
+def test_rejects_signature_mismatch():
+    payload = "{}"
+    ts = int(time.time())
+    sig = _sign(payload, ts, secret="wrong_secret")
+    result = verify_webhook_signature(
+        payload=payload,
+        signature_header=f"t={ts},v1={sig}",
+        secret=SECRET,
+    )
+    assert result.valid is False
+    assert result.reason == "signature_mismatch"
+
+
+def test_no_signatures_for_empty_header():
+    result = verify_webhook_signature(payload="{}", signature_header="", secret=SECRET)
+    assert result.valid is False
+    assert result.reason == "no_signatures"
+
+
+def test_malformed_header():
+    result = verify_webhook_signature(payload="{}", signature_header="just_a_value", secret=SECRET)
+    assert result.valid is False
+    assert result.reason == "malformed_header"
+
+
+def test_no_timestamp_when_missing_and_tolerance_positive():
+    payload = "{}"
+    sig = hmac.new(SECRET.encode(), payload.encode(), hashlib.sha256).hexdigest()
+    result = verify_webhook_signature(payload=payload, signature_header=f"v1={sig}", secret=SECRET)
+    assert result.valid is False
+    assert result.reason == "no_timestamp"
+
+
+def test_tolerance_zero_skips_timestamp_check():
+    payload = '{"event":"test"}'
+    sig = hmac.new(SECRET.encode(), payload.encode(), hashlib.sha256).hexdigest()
+    result = verify_webhook_signature(
+        payload=payload,
+        signature_header=f"v1={sig}",
+        secret=SECRET,
+        tolerance_seconds=0,
+    )
+    assert result.valid is True
+
+
+def test_multiple_signatures_any_match():
+    payload = "{}"
+    ts = int(time.time())
+    sig_good = _sign(payload, ts)
+    sig_bad = hmac.new(b"wrong", f"{ts}.{payload}".encode(), hashlib.sha256).hexdigest()
+    result = verify_webhook_signature(
+        payload=payload,
+        signature_header=f"t={ts},v1={sig_bad},v1={sig_good}",
+        secret=SECRET,
+    )
+    assert result.valid is True
+
+
+def test_status_alias_matches_status_code():
+    """AgentScoreError.status property mirrors .status_code (parity with node-sdk)."""
+    err = AgentScoreError(code="rate_limited", message="too many", status_code=429)
+    assert err.status == 429
+    assert err.status == err.status_code