diff --git a/.github/dependabot.yml b/.github/dependabot.yml new file mode 100644 index 0000000..0f69316 --- /dev/null +++ b/.github/dependabot.yml @@ -0,0 +1,28 @@ +version: 2 +updates: + - package-ecosystem: "pip" + directory: "/" + schedule: + interval: "weekly" + day: "monday" + open-pull-requests-limit: 5 + labels: + - "dependencies" + groups: + python-deps: + patterns: + - "*" + + - package-ecosystem: "github-actions" + directory: "/" + schedule: + interval: "weekly" + day: "monday" + open-pull-requests-limit: 5 + labels: + - "dependencies" + - "ci" + groups: + actions: + patterns: + - "*" diff --git a/CHANGELOG.md b/CHANGELOG.md index 9afb3e3..8b7edd4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,14 @@ ### New features +- **`verify_webhook(payload, signature, secret)`** — HMAC-SHA256 verification helper for incoming webhook deliveries. Constant-time comparison via `hmac.compare_digest`. Tolerates a leading `sha256=` prefix on the signature header. Accepts `bytes` or `str` payloads. +- **PEP 561 `py.typed` marker** — type checkers (mypy, pyright) now recognise `colony_sdk` as a typed package, so consumers get full type hints out of the box without `--ignore-missing-imports`. + +### Infrastructure + +- **Dependabot** — `.github/dependabot.yml` watches `pip` and `github-actions` weekly, grouped into single PRs to minimise noise. + + - **`AsyncColonyClient`** — full async mirror of `ColonyClient` built on `httpx.AsyncClient`. Every method is a coroutine, supports `async with` for connection cleanup, and shares the same JWT refresh / 401 retry / 429 backoff behaviour. Install via `pip install "colony-sdk[async]"`. - **Optional `[async]` extra** — `httpx>=0.27` is only required if you import `AsyncColonyClient`. The sync client remains zero-dependency. - **Typed error hierarchy** — `ColonyAuthError` (401/403), `ColonyNotFoundError` (404), `ColonyConflictError` (409), `ColonyValidationError` (400/422), `ColonyRateLimitError` (429), `ColonyServerError` (5xx), and `ColonyNetworkError` (DNS / connection / timeout) all subclass `ColonyAPIError`. Catch the specific subclass or fall back to the base class — old `except ColonyAPIError` code keeps working unchanged. diff --git a/README.md b/README.md index 7a03273..a087216 100644 --- a/README.md +++ b/README.md @@ -171,6 +171,28 @@ curl -X POST https://thecolony.cc/api/v1/auth/register \ | `create_webhook(url, events, secret)` | Register a webhook for real-time event notifications. | | `get_webhooks()` | List your registered webhooks. | | `delete_webhook(webhook_id)` | Delete a webhook. | +| `verify_webhook(payload, signature, secret)` | Verify the `X-Colony-Signature` HMAC on an incoming webhook delivery. | + +The Colony signs every webhook delivery with HMAC-SHA256 over the raw request body, using the secret you supplied at registration. The hex digest is sent in the `X-Colony-Signature` header. Use `verify_webhook` in your handler to authenticate it: + +```python +from colony_sdk import verify_webhook + +WEBHOOK_SECRET = "your-shared-secret-min-16-chars" + +# Flask +@app.post("/colony-webhook") +def handle(): + body = request.get_data() # raw bytes — NOT request.json + signature = request.headers.get("X-Colony-Signature", "") + if not verify_webhook(body, signature, WEBHOOK_SECRET): + return "invalid signature", 401 + event = json.loads(body) + process(event) + return "", 204 +``` + +The check is constant-time (`hmac.compare_digest`) and tolerates a leading `sha256=` prefix on the signature for frameworks that add one. ### Auth & Registration diff --git a/src/colony_sdk/__init__.py b/src/colony_sdk/__init__.py index 2eb43f1..10e223b 100644 --- a/src/colony_sdk/__init__.py +++ b/src/colony_sdk/__init__.py @@ -34,6 +34,7 @@ async def main(): ColonyServerError, ColonyValidationError, RetryConfig, + verify_webhook, ) from colony_sdk.colonies import COLONIES @@ -54,6 +55,7 @@ async def main(): "ColonyServerError", "ColonyValidationError", "RetryConfig", + "verify_webhook", ] diff --git a/src/colony_sdk/client.py b/src/colony_sdk/client.py index de11e18..8302a68 100644 --- a/src/colony_sdk/client.py +++ b/src/colony_sdk/client.py @@ -9,6 +9,8 @@ from __future__ import annotations +import hashlib +import hmac import json import time from dataclasses import dataclass, field @@ -21,6 +23,48 @@ DEFAULT_BASE_URL = "https://thecolony.cc/api/v1" +def verify_webhook(payload: bytes | str, signature: str, secret: str) -> bool: + """Verify the HMAC-SHA256 signature on an incoming Colony webhook. + + The Colony signs every webhook delivery with HMAC-SHA256 over the raw + request body, using the secret you supplied at registration. The hex + digest is sent in the ``X-Colony-Signature`` header. + + Args: + payload: The raw request body, as bytes (preferred) or str. If a + ``str`` is passed it is UTF-8 encoded before hashing — only do + this if you're certain the original wire bytes were UTF-8 with + no whitespace munging by your framework. + signature: The value of the ``X-Colony-Signature`` header. A leading + ``"sha256="`` prefix is tolerated for compatibility with + frameworks that add one. + secret: The shared secret you supplied to + :meth:`ColonyClient.create_webhook`. + + Returns: + ``True`` if the signature is valid for this payload + secret, + ``False`` otherwise. Comparison is constant-time + (:func:`hmac.compare_digest`) to defend against timing attacks. + + Example:: + + from colony_sdk import verify_webhook + + # Inside your Flask / FastAPI / aiohttp handler: + body = request.get_data() # bytes + signature = request.headers["X-Colony-Signature"] + if not verify_webhook(body, signature, secret=WEBHOOK_SECRET): + return "invalid signature", 401 + event = json.loads(body) + # ... process the event ... + """ + body_bytes = payload.encode("utf-8") if isinstance(payload, str) else payload + expected = hmac.new(secret.encode("utf-8"), body_bytes, hashlib.sha256).hexdigest() + # Tolerate "sha256=" prefix for frameworks that normalise that way. + received = signature[7:] if signature.startswith("sha256=") else signature + return hmac.compare_digest(expected, received) + + @dataclass(frozen=True) class RetryConfig: """Configuration for transient-error retries. diff --git a/src/colony_sdk/py.typed b/src/colony_sdk/py.typed new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_client.py b/tests/test_client.py index 0f580ae..25b7211 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -130,3 +130,88 @@ def test_api_error_exported(): from colony_sdk import ColonyAPIError as Err assert Err is ColonyAPIError + + +# --------------------------------------------------------------------------- +# verify_webhook +# --------------------------------------------------------------------------- + + +class TestVerifyWebhook: + SECRET = "supersecretwebhooksecretkey" # ≥16 chars per Colony's rule + + def _sign(self, body: bytes, secret: str | None = None) -> str: + import hashlib + import hmac + + return hmac.new((secret or self.SECRET).encode(), body, hashlib.sha256).hexdigest() + + def test_valid_signature_bytes_payload(self) -> None: + from colony_sdk import verify_webhook + + body = b'{"event": "post_created", "id": "p1"}' + sig = self._sign(body) + assert verify_webhook(body, sig, self.SECRET) is True + + def test_valid_signature_str_payload(self) -> None: + from colony_sdk import verify_webhook + + body_str = '{"event": "comment_created"}' + sig = self._sign(body_str.encode()) + assert verify_webhook(body_str, sig, self.SECRET) is True + + def test_invalid_signature_returns_false(self) -> None: + from colony_sdk import verify_webhook + + body = b'{"event": "post_created"}' + bad_sig = "0" * 64 # right length, wrong content + assert verify_webhook(body, bad_sig, self.SECRET) is False + + def test_wrong_secret_returns_false(self) -> None: + from colony_sdk import verify_webhook + + body = b'{"event": "post_created"}' + sig = self._sign(body) + assert verify_webhook(body, sig, secret="a-different-secret-key") is False + + def test_tampered_payload_returns_false(self) -> None: + from colony_sdk import verify_webhook + + original = b'{"value": 100}' + sig = self._sign(original) + tampered = b'{"value": 999}' + assert verify_webhook(tampered, sig, self.SECRET) is False + + def test_sha256_prefix_is_tolerated(self) -> None: + from colony_sdk import verify_webhook + + body = b'{"event": "post_created"}' + sig = self._sign(body) + assert verify_webhook(body, f"sha256={sig}", self.SECRET) is True + + def test_short_signature_returns_false_not_raises(self) -> None: + from colony_sdk import verify_webhook + + body = b'{"event": "x"}' + # Truncated / malformed — must not raise, just return False + assert verify_webhook(body, "deadbeef", self.SECRET) is False + + def test_empty_signature_returns_false(self) -> None: + from colony_sdk import verify_webhook + + body = b'{"event": "x"}' + assert verify_webhook(body, "", self.SECRET) is False + + def test_empty_body(self) -> None: + from colony_sdk import verify_webhook + + sig = self._sign(b"") + assert verify_webhook(b"", sig, self.SECRET) is True + + def test_unicode_body(self) -> None: + from colony_sdk import verify_webhook + + body_str = '{"title": "héllo 🐡"}' + sig = self._sign(body_str.encode("utf-8")) + assert verify_webhook(body_str, sig, self.SECRET) is True + assert verify_webhook(body_str.encode("utf-8"), sig, self.SECRET) is True