Webhook verification (#49)

* add webhook verification and improve readme

* add pr template

* wip

* wip

Author: Ashutosh-Bhadauriya

.github/pull_request_template.md

@@ -0,0 +1,12 @@
+### Breaking Changes
+<!-- Optional - List any backward incompatible changes -->
+
+### New Features
+<!-- Optional - List new functionality added -->
+
+### Bug Fixes
+<!-- Optional - List bugs fixed in this release -->
+
+### Checklist
+- [ ] Version bumped
+- [ ] Documentation updated 

authsignal/__init__.py

@@ -1 +1,2 @@
 from .client import AuthsignalClient
+from .webhook import Webhook

authsignal/client.py

@@ -9,6 +9,7 @@
 from requests.adapters import HTTPAdapter
 
 from authsignal.version import VERSION
+from authsignal.webhook import Webhook
 
 API_BASE_URL = "https://api.authsignal.com/v1"
 
@@ -98,7 +99,7 @@ class AuthsignalClient(object):
     def __init__(self, api_secret_key, api_url=API_BASE_URL, timeout=2.0):
         """Initialize the client.
         Args:
-            api_key: Your Authsignal Secret API key of your tenant
+            api_secret_key: Your Authsignal Secret API key of your tenant
             api_url: Base URL, including scheme and host, for sending events.
                 Defaults to 'https://api.authsignal.com/v1'.
             timeout: Number of seconds to wait before failing request. Defaults
@@ -112,6 +113,7 @@ def __init__(self, api_secret_key, api_url=API_BASE_URL, timeout=2.0):
 
         self.session = CustomSession(timeout=timeout, api_key=api_secret_key)
         self.version = VERSION
+        self.webhook = Webhook(api_secret_key=api_secret_key)
 
     def track(
         self, user_id: str, action: str, attributes: Dict[str, Any] = None

authsignal/webhook.py

@@ -0,0 +1,51 @@
+import hmac
+import hashlib
+import base64
+import json
+import time
+from typing import List, Dict, Any
+
+DEFAULT_TOLERANCE = 5  # minutes
+VERSION = "v2"
+
+class InvalidSignatureError(Exception):
+    pass
+
+class Webhook:
+    def __init__(self, api_secret_key: str):
+        self.api_secret_key = api_secret_key
+
+    def construct_event(self, payload: str, signature: str, tolerance: int = DEFAULT_TOLERANCE) -> Dict[str, Any]:
+        parsed_signature = self.parse_signature(signature)
+        seconds_since_epoch = int(time.time())
+
+        if tolerance > 0 and parsed_signature["timestamp"] < seconds_since_epoch - tolerance * 60:
+            raise InvalidSignatureError("Timestamp is outside the tolerance zone.")
+
+        hmac_content = f"{parsed_signature['timestamp']}.{payload}"
+        computed_signature = base64.b64encode(
+            hmac.new(
+                self.api_secret_key.encode(),
+                hmac_content.encode(),
+                hashlib.sha256
+            ).digest()
+        ).decode().replace("=", "")
+
+        match = any(sig == computed_signature for sig in parsed_signature["signatures"])
+        if not match:
+            raise InvalidSignatureError("Signature mismatch.")
+
+        return json.loads(payload)
+
+    def parse_signature(self, value: str) -> Dict[str, Any]:
+        timestamp = -1
+        signatures: List[str] = []
+        for item in value.split(","):
+            kv = item.split("=")
+            if kv[0] == "t":
+                timestamp = int(kv[1])
+            if kv[0] == VERSION:
+                signatures.append(kv[1])
+        if timestamp == -1 or not signatures:
+            raise InvalidSignatureError("Signature format is invalid.")
+        return {"timestamp": timestamp, "signatures": signatures} 

authsignal/webhook_tests.py

@@ -0,0 +1,107 @@
+import unittest
+import time
+import base64
+import hmac
+import hashlib
+import json
+
+from .webhook import Webhook, InvalidSignatureError
+
+class TestWebhook(unittest.TestCase):
+    def setUp(self):
+        self.secret = "YOUR_AUTHSIGNAL_SECRET_KEY" 
+        self.webhook = Webhook(self.secret)
+        self.payload_valid_signature = json.dumps({
+            "version": 1,
+            "id": "bc1598bc-e5d6-4c69-9afb-1a6fe3469d6e",
+            "source": "https://authsignal.com",
+            "time": "2025-02-20T01:51:56.070Z",
+            "tenantId": "7752d28e-e627-4b1b-bb81-b45d68d617bc",
+            "type": "email.created",
+            "data": {
+                "to": "[email protected]",
+                "code": "157743",
+                "userId": "b9f74d36-fcfc-4efc-87f1-3664ab5a7fb0",
+                "actionCode": "accountRecovery",
+                "idempotencyKey": "ba8c1a7c-775d-4dff-9abe-be798b7b8bb9",
+                "verificationMethod": "EMAIL_OTP",
+            },
+        })
+        self.payload_multiple_keys = json.dumps({
+            "version": 1,
+            "id": "af7be03c-ea8f-4739-b18e-8b48fcbe4e38",
+            "source": "https://authsignal.com",
+            "time": "2025-02-20T01:47:17.248Z",
+            "tenantId": "7752d28e-e627-4b1b-bb81-b45d68d617bc",
+            "type": "email.created",
+            "data": {
+                "to": "[email protected]",
+                "code": "718190",
+                "userId": "b9f74d36-fcfc-4efc-87f1-3664ab5a7fb0",
+                "actionCode": "accountRecovery",
+                "idempotencyKey": "68d68190-fac9-4e91-b277-c63d31d3c6b1",
+                "verificationMethod": "EMAIL_OTP",
+            },
+        })
+        self.timestamp = int(time.time())
+        self.version = "v2"
+
+    def generate_signature(self, payload, timestamp=None, secret=None, extra_signatures=None):
+        if timestamp is None:
+            timestamp = self.timestamp
+        if secret is None:
+            secret = self.secret
+        hmac_content = f"{timestamp}.{payload}"
+        computed_signature = base64.b64encode(
+            hmac.new(secret.encode(), hmac_content.encode(), hashlib.sha256).digest()
+        ).decode().replace("=", "")
+        sigs = [f"{self.version}={computed_signature}"]
+        if extra_signatures:
+            sigs.extend([f"{self.version}={s}" for s in extra_signatures])
+        return f"t={timestamp}," + ",".join(sigs)
+
+    def test_invalid_signature_format(self):
+        with self.assertRaises(InvalidSignatureError) as cm:
+            self.webhook.construct_event(self.payload_valid_signature, "123")
+        self.assertEqual(str(cm.exception), "Signature format is invalid.")
+
+    def test_timestamp_tolerance_error(self):
+        signature = "t=1630000000,v2=invalid_signature"
+        with self.assertRaises(InvalidSignatureError) as cm:
+            self.webhook.construct_event(self.payload_valid_signature, signature)
+        self.assertEqual(str(cm.exception), "Timestamp is outside the tolerance zone.")
+
+    def test_invalid_computed_signature(self):
+        timestamp = int(time.time())
+        signature = f"t={timestamp},v2=invalid_signature"
+        with self.assertRaises(InvalidSignatureError) as cm:
+            self.webhook.construct_event(self.payload_valid_signature, signature)
+        self.assertEqual(str(cm.exception), "Signature mismatch.")
+
+    def test_valid_signature(self):
+        payload = self.payload_valid_signature
+        timestamp = 1740016316 
+        signature = self.generate_signature(payload, timestamp=timestamp, secret=self.secret)
+        
+        event = self.webhook.construct_event(payload, signature, tolerance=-1)
+        self.assertIsNotNone(event)
+        self.assertEqual(event["version"], 1)
+        self.assertEqual(event["data"]["actionCode"], "accountRecovery")
+
+    def test_valid_signature_multiple_keys(self):
+
+        payload = self.payload_multiple_keys
+        timestamp = 1740016037
+
+        valid_signature = self.generate_signature(payload, timestamp=timestamp, secret=self.secret).split(",")[1]
+        
+
+        signature = f"t={timestamp},{valid_signature},v2=dummyInvalidSignature"
+
+        event = self.webhook.construct_event(payload, signature, tolerance=-1)
+        self.assertIsNotNone(event)
+        self.assertEqual(event["version"], 1)
+        self.assertEqual(event["data"]["actionCode"], "accountRecovery")
+
+if __name__ == "__main__":
+    unittest.main() 

readme.md

@@ -2,11 +2,12 @@
 
 # Authsignal Python SDK
 
-The Authsignal Python library for server-side applications.
+[![PyPI version](https://img.shields.io/pypi/v/authsignal.svg)](https://pypi.org/project/authsignal/)
+[![License](https://img.shields.io/github/license/authsignal/authsignal-python.svg)](https://github.com/authsignal/authsignal-python/blob/main/LICENSE)
 
-## Installation
+The official Authsignal Python library for server-side applications. Use this SDK to easily integrate Authsignal's multi-factor authentication (MFA) and passwordless features into your Python backend.
 
-Python 3
+## Installation
 
 ```bash
 pip3 install authsignal
@@ -18,6 +19,32 @@ or install newest source directly from GitHub:
 pip3 install git+https://github.com/authsignal/authsignal-python
 ```
 
+## Getting Started
+
+Initialize the Authsignal client with your secret key from the [Authsignal Portal](https://portal.authsignal.com/) and the API URL for your region.
+
+```python
+from authsignal import Authsignal
+
+# Initialize the client
+authsignal = Authsignal(
+    api_secret_key="your_secret_key",
+    api_url="https://api.authsignal.com/v1"  # Use region-specific URL
+)
+```
+
+### API URLs by Region
+
+| Region      | API URL                          |
+| ----------- | -------------------------------- |
+| US (Oregon) | https://api.authsignal.com/v1    |
+| AU (Sydney) | https://au.api.authsignal.com/v1 |
+| EU (Dublin) | https://eu.api.authsignal.com/v1 |
+
+## License
+
+This SDK is licensed under the [MIT License](LICENSE).
+
 ## Documentation
 
-Refer to our [SDK documentation](https://docs.authsignal.com/sdks/server) for information on how to use this SDK.
+For more information and advanced usage examples, refer to the official [Authsignal Server-Side SDK documentation](https://docs.authsignal.com/sdks/server/overview).