diff --git a/src/orb/resources/webhooks.py b/src/orb/resources/webhooks.py index 2c5e9c7c..48c321fa 100644 --- a/src/orb/resources/webhooks.py +++ b/src/orb/resources/webhooks.py @@ -56,7 +56,11 @@ def verify_signature( now = datetime.now(tz=timezone.utc) try: - timestamp = datetime.fromisoformat(msg_timestamp).astimezone() + timestamp = datetime.fromisoformat(msg_timestamp) + # If the timestamp doesn't have timezone info, assume it's UTC + if timestamp.tzinfo is None: + timestamp = timestamp.replace(tzinfo=timezone.utc) + timestamp = timestamp.astimezone() except Exception as err: raise ValueError("Invalid signature headers. Could not convert to timestamp") from err @@ -140,7 +144,11 @@ def verify_signature( now = datetime.now(tz=timezone.utc) try: - timestamp = datetime.fromisoformat(msg_timestamp).astimezone() + timestamp = datetime.fromisoformat(msg_timestamp) + # If the timestamp doesn't have timezone info, assume it's UTC + if timestamp.tzinfo is None: + timestamp = timestamp.replace(tzinfo=timezone.utc) + timestamp = timestamp.astimezone() except Exception as err: raise ValueError("Invalid signature headers. Could not convert to timestamp") from err diff --git a/tests/api_resources/test_webhooks.py b/tests/api_resources/test_webhooks.py index ff269d5e..2ac0b7a1 100644 --- a/tests/api_resources/test_webhooks.py +++ b/tests/api_resources/test_webhooks.py @@ -4,7 +4,7 @@ import os from typing import Any, cast -from datetime import datetime, timedelta +from datetime import datetime, timezone, timedelta import pytest import time_machine @@ -18,7 +18,11 @@ class TestWebhooks: parametrize = pytest.mark.parametrize("client", [False, True], indirect=True, ids=["loose", "strict"]) timestamp = "2024-03-27T15:42:29.551" - fake_now = datetime.fromisoformat(timestamp).astimezone() + # Fix: Ensure fake_now matches how webhook timestamps are now parsed (UTC assumption) + fake_now_dt = datetime.fromisoformat(timestamp) + if fake_now_dt.tzinfo is None: + fake_now_dt = fake_now_dt.replace(tzinfo=timezone.utc) + fake_now = fake_now_dt.astimezone() payload = """{"id": "o4mmewpfNNTnjfZc", "created_at": "2024-03-27T15:42:29+00:00", "type": "resource_event.test", "properties": {"message": "A test webhook from Orb. Happy testing!"}}""" signature = "9d25de966891ab0bc18754faf8d83d0980b44ae330fcc130b41a6cf3daf1f391" @@ -106,12 +110,98 @@ def test_verify_signature(self, client: Orb) -> None: secret=secret, ) + def test_microsecond_precision_issue_fixed(self, client: Orb) -> None: + """Test that the webhook timestamp parsing issue is fixed for the reported examples.""" + import hmac + import hashlib + + secret = self.secret + + # Test cases from the reported issue - these should all work now + test_cases = [ + ("2025-08-08T21:35:11.531998+00:00", "2025-08-08T21:35:11.445"), + ("2025-08-08T21:32:02.585239+00:00", "2025-08-08T21:32:02.497"), + ("2025-08-08T21:35:42.810490+00:00", "2025-08-08T21:35:42.660"), + ] + + for _, (system_time_str, webhook_timestamp) in enumerate(test_cases, 1): + system_time = datetime.fromisoformat(system_time_str) + + # Generate the correct signature + to_sign = f"v1:{webhook_timestamp}:{self.payload}".encode("utf-8") + signature = hmac.new(secret.encode("utf-8"), to_sign, hashlib.sha256).hexdigest() + + with time_machine.travel(system_time): + # This should now work without raising "Webhook timestamp is too new" + client.webhooks.verify_signature( + payload=self.payload, + headers={"X-Orb-Timestamp": webhook_timestamp, "X-Orb-Signature": f"v1={signature}"}, + secret=secret, + ) + + # Also test the unwrap method + result = client.webhooks.unwrap( + payload=self.payload, + headers={"X-Orb-Timestamp": webhook_timestamp, "X-Orb-Signature": f"v1={signature}"}, + secret=secret, + ) + assert result is not None + + def test_timezone_aware_timestamps_still_work(self, client: Orb) -> None: + """Test that webhook timestamps with explicit timezone info still work.""" + import hmac + import hashlib + from datetime import timezone + + secret = self.secret + + # Test with explicit UTC timezone + system_time = datetime(2025, 8, 8, 21, 35, 11, 531998, tzinfo=timezone.utc) + webhook_timestamp = "2025-08-08T21:35:11.445+00:00" # Explicit UTC + + to_sign = f"v1:{webhook_timestamp}:{self.payload}".encode("utf-8") + signature = hmac.new(secret.encode("utf-8"), to_sign, hashlib.sha256).hexdigest() + + with time_machine.travel(system_time): + client.webhooks.verify_signature( + payload=self.payload, + headers={"X-Orb-Timestamp": webhook_timestamp, "X-Orb-Signature": f"v1={signature}"}, + secret=secret, + ) + + def test_webhook_timestamp_actually_too_new(self, client: Orb) -> None: + """Test that webhooks that are genuinely too new are still rejected.""" + import hmac + import hashlib + from datetime import timezone + + secret = self.secret + + # Set system time to be much earlier than webhook timestamp (more than 5 minute tolerance) + system_time = datetime(2025, 8, 8, 21, 30, 0, 0, tzinfo=timezone.utc) + webhook_timestamp = "2025-08-08T21:36:00.000" # 6 minutes later - should be rejected + + to_sign = f"v1:{webhook_timestamp}:{self.payload}".encode("utf-8") + signature = hmac.new(secret.encode("utf-8"), to_sign, hashlib.sha256).hexdigest() + + with time_machine.travel(system_time): + with pytest.raises(ValueError, match="Webhook timestamp is too new"): + client.webhooks.verify_signature( + payload=self.payload, + headers={"X-Orb-Timestamp": webhook_timestamp, "X-Orb-Signature": f"v1={signature}"}, + secret=secret, + ) + class TestAsyncWebhooks: parametrize = pytest.mark.parametrize("async_client", [False, True], indirect=True, ids=["loose", "strict"]) timestamp = "2024-03-27T15:42:29.551" - fake_now = datetime.fromisoformat(timestamp).astimezone() + # Fix: Ensure fake_now matches how webhook timestamps are now parsed (UTC assumption) + fake_now_dt = datetime.fromisoformat(timestamp) + if fake_now_dt.tzinfo is None: + fake_now_dt = fake_now_dt.replace(tzinfo=timezone.utc) + fake_now = fake_now_dt.astimezone() payload = """{"id": "o4mmewpfNNTnjfZc", "created_at": "2024-03-27T15:42:29+00:00", "type": "resource_event.test", "properties": {"message": "A test webhook from Orb. Happy testing!"}}""" signature = "9d25de966891ab0bc18754faf8d83d0980b44ae330fcc130b41a6cf3daf1f391"