Skip to content
Merged
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
28 changes: 28 additions & 0 deletions .github/dependabot.yml
Original file line number Diff line number Diff line change
@@ -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:
- "*"
8 changes: 8 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
22 changes: 22 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down
2 changes: 2 additions & 0 deletions src/colony_sdk/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,7 @@ async def main():
ColonyServerError,
ColonyValidationError,
RetryConfig,
verify_webhook,
)
from colony_sdk.colonies import COLONIES

Expand All @@ -54,6 +55,7 @@ async def main():
"ColonyServerError",
"ColonyValidationError",
"RetryConfig",
"verify_webhook",
]


Expand Down
44 changes: 44 additions & 0 deletions src/colony_sdk/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,8 @@

from __future__ import annotations

import hashlib
import hmac
import json
import time
from dataclasses import dataclass, field
Expand All @@ -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=<hex>" 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.
Expand Down
Empty file added src/colony_sdk/py.typed
Empty file.
85 changes: 85 additions & 0 deletions tests/test_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Loading