diff --git a/README.md b/README.md
index 8e807ac0..191c1ee5 100644
--- a/README.md
+++ b/README.md
@@ -142,29 +142,6 @@ page = client.hris.directory.list()
print(page.page)
```
-## Webhook Verification
-
-We provide helper methods for verifying that a webhook request came from Finch, and not a malicious third party.
-
-You can use `finch.webhooks.verify_signature(body: string, headers, secret?) -> None` or `finch.webhooks.unwrap(body: string, headers, secret?) -> Payload`,
-both of which will raise an error if the signature is invalid.
-
-Note that the "body" parameter must be the raw JSON string sent from the server (do not parse it first).
-The `.unwrap()` method can parse this JSON for you into a `Payload` object.
-
-For example, in [FastAPI](https://fastapi.tiangolo.com/):
-
-```py
-@app.post('/my-webhook-handler')
-async def handler(request: Request):
- body = await request.body()
- secret = os.environ['FINCH_WEBHOOK_SECRET'] # env var used by default; explicit here.
- payload = client.webhooks.unwrap(body, request.headers, secret)
- print(payload)
-
- return {'ok': True}
-```
-
## Handling errors
When the library is unable to connect to the API (for example, due to network connection problems or a timeout), a subclass of `finch.APIConnectionError` is raised.
diff --git a/api.md b/api.md
index 80dfe151..5ca9c362 100644
--- a/api.md
+++ b/api.md
@@ -4,13 +4,6 @@
from finch.types import ConnectionStatusType, OperationSupport, OperationSupportMatrix, Paging
```
-# Finch
-
-Methods:
-
-- client.get_auth_url(\*args) -> str
-- client.with_access_token(\*args) -> Self
-
# AccessTokens
Types:
@@ -195,11 +188,6 @@ from finch.types import (
)
```
-Methods:
-
-- client.webhooks.unwrap(\*args) -> WebhookEvent
-- client.webhooks.verify_signature(\*args) -> None
-
# RequestForwarding
Types:
diff --git a/src/finch/_client.py b/src/finch/_client.py
index cddb281d..9afd1c69 100644
--- a/src/finch/_client.py
+++ b/src/finch/_client.py
@@ -56,7 +56,6 @@ class Finch(SyncAPIClient):
hris: resources.HRIS
providers: resources.Providers
account: resources.Account
- webhooks: resources.Webhooks
request_forwarding: resources.RequestForwarding
jobs: resources.Jobs
sandbox: resources.Sandbox
@@ -157,7 +156,6 @@ def __init__(
self.hris = resources.HRIS(self)
self.providers = resources.Providers(self)
self.account = resources.Account(self)
- self.webhooks = resources.Webhooks(self)
self.request_forwarding = resources.RequestForwarding(self)
self.jobs = resources.Jobs(self)
self.sandbox = resources.Sandbox(self)
@@ -301,70 +299,6 @@ def copy(
# client.with_options(timeout=10).foo.create(...)
with_options = copy
- def get_access_token(
- self,
- code: str,
- *,
- redirect_uri: str | None = None,
- ) -> str:
- """DEPRECATED: use client.access_tokens.create instead."""
- if self.client_id is None:
- raise ValueError("Expected client_id to be set in order to call get_access_token")
-
- if self.client_secret is None:
- raise ValueError("Expected client_secret to be set in order to call get_access_token")
-
- response = self.post(
- "/auth/token",
- body={
- "client_id": self.client_id,
- "client_secret": self.client_secret,
- "code": code,
- "redirect_uri": redirect_uri,
- },
- options={"headers": {"Authorization": Omit()}},
- cast_to=httpx.Response,
- )
- data = response.json()
- return str(data["access_token"])
-
- def get_auth_url(
- self,
- *,
- products: str,
- redirect_uri: str,
- sandbox: bool,
- ) -> str:
- """
- Returns the authorization url which can be visited in order to obtain an
- authorization code from Finch. The authorization code can then be exchanged for
- an access token for the Finch api by calling get_access_token().
- """
- if self.client_id is None:
- raise ValueError("Expected the client_id to be set in order to call get_auth_url")
-
- return str(
- httpx.URL(
- "https://connect.tryfinch.com/authorize",
- params={
- "client_id": self.client_id,
- "products": products,
- "redirect_uri": redirect_uri,
- "sandbox": sandbox,
- },
- )
- )
-
- def with_access_token(
- self,
- access_token: str,
- ) -> Self:
- """
- Returns a copy of the current Finch client with the given access token for
- authentication.
- """
- return self.with_options(access_token=access_token)
-
@override
def _make_status_error(
self,
@@ -404,7 +338,6 @@ class AsyncFinch(AsyncAPIClient):
hris: resources.AsyncHRIS
providers: resources.AsyncProviders
account: resources.AsyncAccount
- webhooks: resources.AsyncWebhooks
request_forwarding: resources.AsyncRequestForwarding
jobs: resources.AsyncJobs
sandbox: resources.AsyncSandbox
@@ -505,7 +438,6 @@ def __init__(
self.hris = resources.AsyncHRIS(self)
self.providers = resources.AsyncProviders(self)
self.account = resources.AsyncAccount(self)
- self.webhooks = resources.AsyncWebhooks(self)
self.request_forwarding = resources.AsyncRequestForwarding(self)
self.jobs = resources.AsyncJobs(self)
self.sandbox = resources.AsyncSandbox(self)
@@ -649,70 +581,6 @@ def copy(
# client.with_options(timeout=10).foo.create(...)
with_options = copy
- async def get_access_token(
- self,
- code: str,
- *,
- redirect_uri: str | None = None,
- ) -> str:
- """DEPRECATED: use client.access_tokens.create instead."""
- if self.client_id is None:
- raise ValueError("Expected client_id to be set in order to call get_access_token")
-
- if self.client_secret is None:
- raise ValueError("Expected client_secret to be set in order to call get_access_token")
-
- response = await self.post(
- "/auth/token",
- body={
- "client_id": self.client_id,
- "client_secret": self.client_secret,
- "code": code,
- "redirect_uri": redirect_uri,
- },
- options={"headers": {"Authorization": Omit()}},
- cast_to=httpx.Response,
- )
- data = response.json()
- return str(data["access_token"])
-
- def get_auth_url(
- self,
- *,
- products: str,
- redirect_uri: str,
- sandbox: bool,
- ) -> str:
- """
- Returns the authorization url which can be visited in order to obtain an
- authorization code from Finch. The authorization code can then be exchanged for
- an access token for the Finch api by calling get_access_token().
- """
- if self.client_id is None:
- raise ValueError("Expected the client_id to be set in order to call get_auth_url")
-
- return str(
- httpx.URL(
- "https://connect.tryfinch.com/authorize",
- params={
- "client_id": self.client_id,
- "products": products,
- "redirect_uri": redirect_uri,
- "sandbox": sandbox,
- },
- )
- )
-
- def with_access_token(
- self,
- access_token: str,
- ) -> Self:
- """
- Returns a copy of the current Finch client with the given access token for
- authentication.
- """
- return self.with_options(access_token=access_token)
-
@override
def _make_status_error(
self,
diff --git a/src/finch/resources/__init__.py b/src/finch/resources/__init__.py
index d5f226e9..04bef46f 100644
--- a/src/finch/resources/__init__.py
+++ b/src/finch/resources/__init__.py
@@ -32,7 +32,6 @@
SandboxWithStreamingResponse,
AsyncSandboxWithStreamingResponse,
)
-from .webhooks import Webhooks, AsyncWebhooks
from .providers import (
Providers,
AsyncProviders,
@@ -83,8 +82,6 @@
"AsyncAccountWithRawResponse",
"AccountWithStreamingResponse",
"AsyncAccountWithStreamingResponse",
- "Webhooks",
- "AsyncWebhooks",
"RequestForwarding",
"AsyncRequestForwarding",
"RequestForwardingWithRawResponse",
diff --git a/src/finch/resources/access_tokens.py b/src/finch/resources/access_tokens.py
index 7e4fad17..8b5086d7 100644
--- a/src/finch/resources/access_tokens.py
+++ b/src/finch/resources/access_tokens.py
@@ -7,7 +7,10 @@
from .. import _legacy_response
from ..types import CreateAccessTokenResponse, access_token_create_params
from .._types import NOT_GIVEN, Body, Query, Headers, NotGiven
-from .._utils import is_given, maybe_transform
+from .._utils import (
+ maybe_transform,
+ async_maybe_transform,
+)
from .._compat import cached_property
from .._resource import SyncAPIResource, AsyncAPIResource
from .._response import to_streamed_response_wrapper, async_to_streamed_response_wrapper
@@ -53,27 +56,13 @@ def create(
timeout: Override the client-level default timeout for this request, in seconds
"""
- if not is_given(client_id):
- if self._client.client_id is None:
- raise ValueError(
- "client_id must be provided as an argument or with the FINCH_CLIENT_ID environment variable"
- )
- client_id = self._client.client_id
-
- if not is_given(client_secret):
- if self._client.client_secret is None:
- raise ValueError(
- "client_secret must be provided as an argument or with the FINCH_CLIENT_SECRET environment variable"
- )
- client_secret = self._client.client_secret
-
return self._post(
"/auth/token",
body=maybe_transform(
{
+ "code": code,
"client_id": client_id,
"client_secret": client_secret,
- "code": code,
"redirect_uri": redirect_uri,
},
access_token_create_params.AccessTokenCreateParams,
@@ -120,27 +109,13 @@ async def create(
timeout: Override the client-level default timeout for this request, in seconds
"""
- if not is_given(client_id):
- if self._client.client_id is None:
- raise ValueError(
- "client_id must be provided as an argument or with the FINCH_CLIENT_ID environment variable"
- )
- client_id = self._client.client_id
-
- if not is_given(client_secret):
- if self._client.client_secret is None:
- raise ValueError(
- "client_secret must be provided as an argument or with the FINCH_CLIENT_SECRET environment variable"
- )
- client_secret = self._client.client_secret
-
return await self._post(
"/auth/token",
- body=maybe_transform(
+ body=await async_maybe_transform(
{
+ "code": code,
"client_id": client_id,
"client_secret": client_secret,
- "code": code,
"redirect_uri": redirect_uri,
},
access_token_create_params.AccessTokenCreateParams,
diff --git a/src/finch/resources/webhooks.py b/src/finch/resources/webhooks.py
deleted file mode 100644
index 0fe50944..00000000
--- a/src/finch/resources/webhooks.py
+++ /dev/null
@@ -1,221 +0,0 @@
-# File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details.
-
-from __future__ import annotations
-
-import hmac
-import json
-import math
-import base64
-import hashlib
-from typing import cast
-from datetime import datetime, timezone, timedelta
-
-from ..types import WebhookEvent
-from .._types import (
- HeadersLike,
-)
-from .._utils import (
- get_required_header,
-)
-from .._models import construct_type
-from .._resource import SyncAPIResource, AsyncAPIResource
-
-__all__ = ["Webhooks", "AsyncWebhooks"]
-
-
-class Webhooks(SyncAPIResource):
- def unwrap(
- self,
- payload: str | bytes,
- headers: HeadersLike,
- *,
- secret: str | None = None,
- ) -> WebhookEvent:
- """Validates that the given payload was sent by Finch and parses the payload."""
- self.verify_signature(payload=payload, headers=headers, secret=secret)
- return cast(
- WebhookEvent,
- construct_type(
- value=json.loads(payload),
- type_=WebhookEvent, # type: ignore[arg-type]
- ),
- )
-
- def verify_signature(
- self,
- payload: str | bytes,
- headers: HeadersLike,
- *,
- secret: str | None = None,
- ) -> None:
- """Validates whether or not the webhook payload was sent by Finch.
-
- An error will be raised if the webhook payload was not sent by Finch.
- """
- if secret is None:
- secret = self._client.webhook_secret
-
- if secret is None:
- raise ValueError(
- "The webhook secret must either be set using the env var, FINCH_WEBHOOK_SECRET, on the client class, Finch(webhook_secret='123'), or passed to this function"
- )
-
- try:
- parsedSecret = base64.b64decode(secret)
- except Exception as err:
- raise ValueError("Bad secret") from err
-
- msg_id = get_required_header(headers, "finch-event-id")
- msg_timestamp = get_required_header(headers, "finch-timestamp")
-
- # validate the timestamp
- webhook_tolerance = timedelta(minutes=5)
- now = datetime.now(tz=timezone.utc)
-
- try:
- timestamp = datetime.fromtimestamp(float(msg_timestamp), tz=timezone.utc)
- except Exception as err:
- raise ValueError("Invalid timestamp header: " + msg_timestamp + ". Could not convert to timestamp") from err
-
- # too old
- if timestamp < (now - webhook_tolerance):
- raise ValueError("Webhook timestamp is too old")
-
- # too new
- if timestamp > (now + webhook_tolerance):
- raise ValueError("Webhook timestamp is too new")
-
- # create the signature
- body = payload.decode("utf-8") if isinstance(payload, bytes) else payload
- if not isinstance(body, str): # pyright: ignore[reportUnnecessaryIsInstance]
- raise ValueError(
- "Webhook body should be a string of JSON (or bytes which can be decoded to a utf-8 string), not a parsed dictionary."
- )
-
- timestamp_str = str(math.floor(timestamp.replace(tzinfo=timezone.utc).timestamp()))
-
- to_sign = f"{msg_id}.{timestamp_str}.{body}".encode()
- expected_signature = hmac.new(parsedSecret, to_sign, hashlib.sha256).digest()
-
- msg_signature = get_required_header(headers, "finch-signature")
-
- # Signature header can contain multiple signatures delimited by spaces
- passed_sigs = msg_signature.split(" ")
-
- for versioned_sig in passed_sigs:
- values = versioned_sig.split(",")
- if len(values) != 2:
- # signature is not formatted like {version},{signature}
- continue
-
- (version, signature) = values
-
- # Only verify prefix v1
- if version != "v1":
- continue
-
- sig_bytes = base64.b64decode(signature)
- if hmac.compare_digest(expected_signature, sig_bytes):
- # valid!
- return None
-
- raise ValueError("None of the given webhook signatures match the expected signature")
-
-
-class AsyncWebhooks(AsyncAPIResource):
- def unwrap(
- self,
- payload: str | bytes,
- headers: HeadersLike,
- *,
- secret: str | None = None,
- ) -> WebhookEvent:
- """Validates that the given payload was sent by Finch and parses the payload."""
- self.verify_signature(payload=payload, headers=headers, secret=secret)
- return cast(
- WebhookEvent,
- construct_type(
- value=json.loads(payload),
- type_=WebhookEvent, # type: ignore[arg-type]
- ),
- )
-
- def verify_signature(
- self,
- payload: str | bytes,
- headers: HeadersLike,
- *,
- secret: str | None = None,
- ) -> None:
- """Validates whether or not the webhook payload was sent by Finch.
-
- An error will be raised if the webhook payload was not sent by Finch.
- """
- if secret is None:
- secret = self._client.webhook_secret
-
- if secret is None:
- raise ValueError(
- "The webhook secret must either be set using the env var, FINCH_WEBHOOK_SECRET, on the client class, Finch(webhook_secret='123'), or passed to this function"
- )
-
- try:
- parsedSecret = base64.b64decode(secret)
- except Exception as err:
- raise ValueError("Bad secret") from err
-
- msg_id = get_required_header(headers, "finch-event-id")
- msg_timestamp = get_required_header(headers, "finch-timestamp")
-
- # validate the timestamp
- webhook_tolerance = timedelta(minutes=5)
- now = datetime.now(tz=timezone.utc)
-
- try:
- timestamp = datetime.fromtimestamp(float(msg_timestamp), tz=timezone.utc)
- except Exception as err:
- raise ValueError("Invalid timestamp header: " + msg_timestamp + ". Could not convert to timestamp") from err
-
- # too old
- if timestamp < (now - webhook_tolerance):
- raise ValueError("Webhook timestamp is too old")
-
- # too new
- if timestamp > (now + webhook_tolerance):
- raise ValueError("Webhook timestamp is too new")
-
- # create the signature
- body = payload.decode("utf-8") if isinstance(payload, bytes) else payload
- if not isinstance(body, str): # pyright: ignore[reportUnnecessaryIsInstance]
- raise ValueError(
- "Webhook body should be a string of JSON (or bytes which can be decoded to a utf-8 string), not a parsed dictionary."
- )
-
- timestamp_str = str(math.floor(timestamp.replace(tzinfo=timezone.utc).timestamp()))
-
- to_sign = f"{msg_id}.{timestamp_str}.{body}".encode()
- expected_signature = hmac.new(parsedSecret, to_sign, hashlib.sha256).digest()
-
- msg_signature = get_required_header(headers, "finch-signature")
-
- # Signature header can contain multiple signatures delimited by spaces
- passed_sigs = msg_signature.split(" ")
-
- for versioned_sig in passed_sigs:
- values = versioned_sig.split(",")
- if len(values) != 2:
- # signature is not formatted like {version},{signature}
- continue
-
- (version, signature) = values
-
- # Only verify prefix v1
- if version != "v1":
- continue
-
- sig_bytes = base64.b64decode(signature)
- if hmac.compare_digest(expected_signature, sig_bytes):
- # valid!
- return None
-
- raise ValueError("None of the given webhook signatures match the expected signature")
diff --git a/tests/api_resources/test_top_level.py b/tests/api_resources/test_top_level.py
deleted file mode 100644
index e1c15ec0..00000000
--- a/tests/api_resources/test_top_level.py
+++ /dev/null
@@ -1,17 +0,0 @@
-# File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details.
-
-from __future__ import annotations
-
-import os
-
-import pytest
-
-base_url = os.environ.get("TEST_API_BASE_URL", "http://127.0.0.1:4010")
-
-
-class TestTopLevel:
- parametrize = pytest.mark.parametrize("client", [False, True], indirect=True, ids=["loose", "strict"])
-
-
-class TestAsyncTopLevel:
- parametrize = pytest.mark.parametrize("async_client", [False, True], indirect=True, ids=["loose", "strict"])
diff --git a/tests/api_resources/test_webhooks.py b/tests/api_resources/test_webhooks.py
deleted file mode 100644
index 72b7d6af..00000000
--- a/tests/api_resources/test_webhooks.py
+++ /dev/null
@@ -1,214 +0,0 @@
-# File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details.
-
-from __future__ import annotations
-
-import os
-import base64
-from typing import Any, cast
-from datetime import datetime, timezone, timedelta
-
-import pytest
-import time_machine
-from pydantic import BaseModel
-
-from finch import Finch, AsyncFinch
-
-base_url = os.environ.get("TEST_API_BASE_URL", "http://127.0.0.1:4010")
-
-
-class TestWebhooks:
- parametrize = pytest.mark.parametrize("client", [False, True], indirect=True, ids=["loose", "strict"])
-
- timestamp = "1676312382"
- fake_now = datetime.fromtimestamp(float(timestamp), tz=timezone.utc)
-
- payload = """{"company_id":"720be419-0293-4d32-a707-32179b0827ab"}"""
- signature = "m7y0TV2C+hlHxU42wCieApTSTaA8/047OAplBqxIV/s="
- headers = {
- "Finch-Event-Id": "msg_2Lh9KRb0pzN4LePd3XiA4v12Axj",
- "finch-timestamp": timestamp,
- "finch-signature": f"v1,{signature}",
- }
- secret = "5WbX5kEWLlfzsGNjH64I8lOOqUB6e8FH"
-
- @time_machine.travel(fake_now)
- def test_unwrap(self, client: Finch) -> None:
- payload = self.payload
- headers = self.headers
- secret = self.secret
-
- obj = client.webhooks.unwrap(payload, headers, secret=secret)
- assert isinstance(obj, BaseModel)
-
- @time_machine.travel(fake_now)
- def test_verify_signature(self, client: Finch) -> None:
- payload = self.payload
- headers = self.headers
- secret = self.secret
- signature = self.signature
- verify = client.webhooks.verify_signature
-
- assert verify(payload=payload, headers=headers, secret=secret) is None
-
- with pytest.raises(ValueError, match="Webhook timestamp is too old"):
- with time_machine.travel(self.fake_now + timedelta(minutes=6)):
- assert verify(payload=payload, headers=headers, secret=secret) is False
-
- with pytest.raises(ValueError, match="Webhook timestamp is too new"):
- with time_machine.travel(self.fake_now - timedelta(minutes=6)):
- assert verify(payload=payload, headers=headers, secret=secret) is False
-
- # wrong secret
- with pytest.raises(ValueError, match=r"Bad secret"):
- verify(payload=payload, headers=headers, secret="invalid secret")
-
- invalid_signature_message = "None of the given webhook signatures match the expected signature"
- with pytest.raises(ValueError, match=invalid_signature_message):
- verify(
- payload=payload,
- headers=headers,
- secret=base64.b64encode(b"foo").decode("utf-8"),
- )
-
- # multiple signatures
- invalid_signature = base64.b64encode(b"my_sig").decode("utf-8")
- assert (
- verify(
- payload=payload,
- headers={**headers, "finch-signature": f"v1,{invalid_signature} v1,{signature}"},
- secret=secret,
- )
- is None
- )
-
- # different signature version
- with pytest.raises(ValueError, match=invalid_signature_message):
- verify(
- payload=payload,
- headers={**headers, "finch-signature": f"v2,{signature}"},
- secret=secret,
- )
-
- assert (
- verify(
- payload=payload,
- headers={**headers, "finch-signature": f"v2,{signature} v1,{signature}"},
- secret=secret,
- )
- is None
- )
-
- # missing version
- with pytest.raises(ValueError, match=invalid_signature_message):
- verify(
- payload=payload,
- headers={**headers, "finch-signature": signature},
- secret=secret,
- )
-
- # non-string payload
- with pytest.raises(ValueError, match=r"Webhook body should be a string"):
- verify(
- payload=cast(Any, {"payload": payload}),
- headers=headers,
- secret=secret,
- )
-
-
-class TestAsyncWebhooks:
- parametrize = pytest.mark.parametrize("async_client", [False, True], indirect=True, ids=["loose", "strict"])
-
- timestamp = "1676312382"
- fake_now = datetime.fromtimestamp(float(timestamp), tz=timezone.utc)
-
- payload = """{"company_id":"720be419-0293-4d32-a707-32179b0827ab"}"""
- signature = "m7y0TV2C+hlHxU42wCieApTSTaA8/047OAplBqxIV/s="
- headers = {
- "Finch-Event-Id": "msg_2Lh9KRb0pzN4LePd3XiA4v12Axj",
- "finch-timestamp": timestamp,
- "finch-signature": f"v1,{signature}",
- }
- secret = "5WbX5kEWLlfzsGNjH64I8lOOqUB6e8FH"
-
- @time_machine.travel(fake_now)
- def test_unwrap(self, async_client: AsyncFinch) -> None:
- payload = self.payload
- headers = self.headers
- secret = self.secret
-
- obj = async_client.webhooks.unwrap(payload, headers, secret=secret)
- assert isinstance(obj, BaseModel)
-
- @time_machine.travel(fake_now)
- def test_verify_signature(self, async_client: AsyncFinch) -> None:
- payload = self.payload
- headers = self.headers
- secret = self.secret
- signature = self.signature
- verify = async_client.webhooks.verify_signature
-
- assert verify(payload=payload, headers=headers, secret=secret) is None
-
- with pytest.raises(ValueError, match="Webhook timestamp is too old"):
- with time_machine.travel(self.fake_now + timedelta(minutes=6)):
- assert verify(payload=payload, headers=headers, secret=secret) is False
-
- with pytest.raises(ValueError, match="Webhook timestamp is too new"):
- with time_machine.travel(self.fake_now - timedelta(minutes=6)):
- assert verify(payload=payload, headers=headers, secret=secret) is False
-
- # wrong secret
- with pytest.raises(ValueError, match=r"Bad secret"):
- verify(payload=payload, headers=headers, secret="invalid secret")
-
- invalid_signature_message = "None of the given webhook signatures match the expected signature"
- with pytest.raises(ValueError, match=invalid_signature_message):
- verify(
- payload=payload,
- headers=headers,
- secret=base64.b64encode(b"foo").decode("utf-8"),
- )
-
- # multiple signatures
- invalid_signature = base64.b64encode(b"my_sig").decode("utf-8")
- assert (
- verify(
- payload=payload,
- headers={**headers, "finch-signature": f"v1,{invalid_signature} v1,{signature}"},
- secret=secret,
- )
- is None
- )
-
- # different signature version
- with pytest.raises(ValueError, match=invalid_signature_message):
- verify(
- payload=payload,
- headers={**headers, "finch-signature": f"v2,{signature}"},
- secret=secret,
- )
-
- assert (
- verify(
- payload=payload,
- headers={**headers, "finch-signature": f"v2,{signature} v1,{signature}"},
- secret=secret,
- )
- is None
- )
-
- # missing version
- with pytest.raises(ValueError, match=invalid_signature_message):
- verify(
- payload=payload,
- headers={**headers, "finch-signature": signature},
- secret=secret,
- )
-
- # non-string payload
- with pytest.raises(ValueError, match=r"Webhook body should be a string"):
- verify(
- payload=cast(Any, {"payload": payload}),
- headers=headers,
- secret=secret,
- )