diff --git a/src/openai/lib/_parsing/_audio.py b/src/openai/lib/_parsing/_audio.py new file mode 100644 index 0000000000..e97e088461 --- /dev/null +++ b/src/openai/lib/_parsing/_audio.py @@ -0,0 +1,57 @@ +from __future__ import annotations + +import logging +from typing import TYPE_CHECKING +from typing_extensions import assert_never + +from ..._types import Omit +from ...types.audio.translation import Translation +from ...types.audio.transcription import Transcription +from ...types.audio_response_format import AudioResponseFormat +from ...types.audio.translation_verbose import TranslationVerbose +from ...types.audio.transcription_verbose import TranscriptionVerbose +from ...types.audio.transcription_diarized import TranscriptionDiarized + + +def get_transcription_response_format_type( + response_format: AudioResponseFormat | Omit, + *, + log: logging.Logger, +) -> type[Transcription | TranscriptionVerbose | TranscriptionDiarized | str]: + if isinstance(response_format, Omit) or response_format is None: # pyright: ignore[reportUnnecessaryComparison] + return Transcription + + if response_format == "json": + return Transcription + elif response_format == "verbose_json": + return TranscriptionVerbose + elif response_format == "diarized_json": + return TranscriptionDiarized + elif response_format == "srt" or response_format == "text" or response_format == "vtt": + return str + elif TYPE_CHECKING: # type: ignore[unreachable] + assert_never(response_format) + else: + log.warn("Unexpected audio response format: %s", response_format) + return Transcription + + +def get_translation_response_format_type( + response_format: AudioResponseFormat | Omit, + *, + log: logging.Logger, +) -> type[Translation | TranslationVerbose | str]: + if isinstance(response_format, Omit) or response_format is None: # pyright: ignore[reportUnnecessaryComparison] + return Translation + + if response_format == "json": + return Translation + elif response_format == "verbose_json": + return TranslationVerbose + elif response_format == "srt" or response_format == "text" or response_format == "vtt": + return str + elif TYPE_CHECKING and response_format != "diarized_json": # type: ignore[unreachable] + assert_never(response_format) + else: + log.warning("Unexpected audio response format: %s", response_format) + return Translation diff --git a/src/openai/lib/_parsing/_embeddings.py b/src/openai/lib/_parsing/_embeddings.py new file mode 100644 index 0000000000..acb890fa52 --- /dev/null +++ b/src/openai/lib/_parsing/_embeddings.py @@ -0,0 +1,35 @@ +from __future__ import annotations + +import array +import base64 +from typing import cast + +from ..._types import Omit, NotGiven +from ..._utils import is_given +from ..._extras import numpy as np, has_numpy +from ...types.create_embedding_response import CreateEmbeddingResponse + + +def parse_embedding_response( + obj: CreateEmbeddingResponse, *, encoding_format: str | Omit | NotGiven +) -> CreateEmbeddingResponse: + if is_given(encoding_format): + # don't modify the response object if a user explicitly asked for a format + return obj + + if not obj.data: + raise ValueError("No embedding data received") + + for embedding in obj.data: + data = cast(object, embedding.embedding) + if not isinstance(data, str): + continue + if not has_numpy(): + # use array for base64 optimisation + embedding.embedding = array.array("f", base64.b64decode(data)).tolist() + else: + embedding.embedding = np.frombuffer( # type: ignore[no-untyped-call] + base64.b64decode(data), dtype="float32" + ).tolist() + + return obj diff --git a/src/openai/lib/_webhooks.py b/src/openai/lib/_webhooks.py new file mode 100644 index 0000000000..a5cb547bc3 --- /dev/null +++ b/src/openai/lib/_webhooks.py @@ -0,0 +1,64 @@ +from __future__ import annotations + +import hmac +import time +import base64 +import hashlib + +from .._types import HeadersLike +from .._utils import get_required_header +from .._exceptions import InvalidWebhookSignatureError + + +def webhook_signature_matches( + payload: str | bytes, + headers: HeadersLike, + *, + secret: str, + tolerance: int, +) -> bool: + """Validate the replay window and compare the supplied signatures.""" + signature_header = get_required_header(headers, "webhook-signature") + timestamp = get_required_header(headers, "webhook-timestamp") + webhook_id = get_required_header(headers, "webhook-id") + + # Validate timestamp to prevent replay attacks + try: + timestamp_seconds = int(timestamp) + except ValueError: + raise InvalidWebhookSignatureError("Invalid webhook timestamp format") from None + + now = int(time.time()) + + if now - timestamp_seconds > tolerance: + raise InvalidWebhookSignatureError("Webhook timestamp is too old") from None + + if timestamp_seconds > now + tolerance: + raise InvalidWebhookSignatureError("Webhook timestamp is too new") from None + + # Extract signatures from v1, format + # The signature header can have multiple values, separated by spaces. + # Each value is in the format v1,. We should accept if any match. + signatures: list[str] = [] + for part in signature_header.split(): + if part.startswith("v1,"): + signatures.append(part[3:]) + else: + signatures.append(part) + + # Decode the secret if it starts with whsec_ + if secret.startswith("whsec_"): + decoded_secret = base64.b64decode(secret[6:]) + else: + decoded_secret = secret.encode() + + body = payload.decode("utf-8") if isinstance(payload, bytes) else payload + + # Prepare the signed payload (OpenAI uses webhookId.timestamp.payload format) + signed_payload = f"{webhook_id}.{timestamp}.{body}" + expected_signature = base64.b64encode( + hmac.new(decoded_secret, signed_payload.encode(), hashlib.sha256).digest() + ).decode() + + # Accept if any signature matches + return any(hmac.compare_digest(expected_signature, sig) for sig in signatures) diff --git a/src/openai/resources/audio/transcriptions.py b/src/openai/resources/audio/transcriptions.py index 80b7cbdcf6..7dd54f9f02 100644 --- a/src/openai/resources/audio/transcriptions.py +++ b/src/openai/resources/audio/transcriptions.py @@ -3,8 +3,8 @@ from __future__ import annotations import logging -from typing import TYPE_CHECKING, List, Union, Mapping, Optional, cast -from typing_extensions import Literal, overload, assert_never +from typing import List, Union, Mapping, Optional, cast +from typing_extensions import Literal, overload import httpx2 @@ -29,6 +29,7 @@ from ...types.audio import transcription_create_params from ..._base_client import make_request_options from ...types.audio_model import AudioModel +from ...lib._parsing._audio import get_transcription_response_format_type as _get_transcription_response_format_type from ...types.audio.transcription import Transcription from ...types.audio_response_format import AudioResponseFormat from ...types.audio.transcription_include import TranscriptionInclude @@ -1078,19 +1079,4 @@ def __init__(self, transcriptions: AsyncTranscriptions) -> None: def _get_response_format_type( response_format: AudioResponseFormat | Omit, ) -> type[Transcription | TranscriptionVerbose | TranscriptionDiarized | str]: - if isinstance(response_format, Omit) or response_format is None: # pyright: ignore[reportUnnecessaryComparison] - return Transcription - - if response_format == "json": - return Transcription - elif response_format == "verbose_json": - return TranscriptionVerbose - elif response_format == "diarized_json": - return TranscriptionDiarized - elif response_format == "srt" or response_format == "text" or response_format == "vtt": - return str - elif TYPE_CHECKING: # type: ignore[unreachable] - assert_never(response_format) - else: - log.warn("Unexpected audio response format: %s", response_format) - return Transcription + return _get_transcription_response_format_type(response_format, log=log) diff --git a/src/openai/resources/audio/translations.py b/src/openai/resources/audio/translations.py index 60b746b5f2..c41ef1fd97 100644 --- a/src/openai/resources/audio/translations.py +++ b/src/openai/resources/audio/translations.py @@ -3,8 +3,8 @@ from __future__ import annotations import logging -from typing import TYPE_CHECKING, Union, Mapping, cast -from typing_extensions import Literal, overload, assert_never +from typing import Union, Mapping, cast +from typing_extensions import Literal, overload import httpx2 @@ -18,6 +18,7 @@ from ...types.audio import translation_create_params from ..._base_client import make_request_options from ...types.audio_model import AudioModel +from ...lib._parsing._audio import get_translation_response_format_type as _get_translation_response_format_type from ...types.audio.translation import Translation from ...types.audio_response_format import AudioResponseFormat from ...types.audio.translation_verbose import TranslationVerbose @@ -370,17 +371,4 @@ def __init__(self, translations: AsyncTranslations) -> None: def _get_response_format_type( response_format: AudioResponseFormat | Omit, ) -> type[Translation | TranslationVerbose | str]: - if isinstance(response_format, Omit) or response_format is None: # pyright: ignore[reportUnnecessaryComparison] - return Translation - - if response_format == "json": - return Translation - elif response_format == "verbose_json": - return TranslationVerbose - elif response_format == "srt" or response_format == "text" or response_format == "vtt": - return str - elif TYPE_CHECKING and response_format != "diarized_json": # type: ignore[unreachable] - assert_never(response_format) - else: - log.warning("Unexpected audio response format: %s", response_format) - return Translation + return _get_translation_response_format_type(response_format, log=log) diff --git a/src/openai/resources/embeddings.py b/src/openai/resources/embeddings.py index 5b5a018de5..1868381372 100644 --- a/src/openai/resources/embeddings.py +++ b/src/openai/resources/embeddings.py @@ -2,9 +2,8 @@ from __future__ import annotations -import array -import base64 -from typing import Union, Iterable, cast +from typing import Union, Iterable +from functools import partial from typing_extensions import Literal import httpx2 @@ -14,11 +13,11 @@ from .._types import Body, Omit, Query, Headers, NotGiven, SequenceNotStr, omit, not_given from .._utils import is_given, maybe_transform from .._compat import cached_property -from .._extras import numpy as np, has_numpy from .._resource import SyncAPIResource, AsyncAPIResource from .._response import to_streamed_response_wrapper, async_to_streamed_response_wrapper from .._base_client import make_request_options from ..types.embedding_model import EmbeddingModel +from ..lib._parsing._embeddings import parse_embedding_response as _parse_embedding_response from ..types.create_embedding_response import CreateEmbeddingResponse __all__ = ["Embeddings", "AsyncEmbeddings"] @@ -111,28 +110,6 @@ def create( if not is_given(encoding_format): params["encoding_format"] = "base64" - def parser(obj: CreateEmbeddingResponse) -> CreateEmbeddingResponse: - if is_given(encoding_format): - # don't modify the response object if a user explicitly asked for a format - return obj - - if not obj.data: - raise ValueError("No embedding data received") - - for embedding in obj.data: - data = cast(object, embedding.embedding) - if not isinstance(data, str): - continue - if not has_numpy(): - # use array for base64 optimisation - embedding.embedding = array.array("f", base64.b64decode(data)).tolist() - else: - embedding.embedding = np.frombuffer( # type: ignore[no-untyped-call] - base64.b64decode(data), dtype="float32" - ).tolist() - - return obj - return self._post( "/embeddings", body=maybe_transform(params, embedding_create_params.EmbeddingCreateParams), @@ -141,7 +118,7 @@ def parser(obj: CreateEmbeddingResponse) -> CreateEmbeddingResponse: extra_query=extra_query, extra_body=extra_body, timeout=timeout, - post_parser=parser, + post_parser=partial(_parse_embedding_response, encoding_format=encoding_format), security={"bearer_auth": True}, ), cast_to=CreateEmbeddingResponse, @@ -235,28 +212,6 @@ async def create( if not is_given(encoding_format): params["encoding_format"] = "base64" - def parser(obj: CreateEmbeddingResponse) -> CreateEmbeddingResponse: - if is_given(encoding_format): - # don't modify the response object if a user explicitly asked for a format - return obj - - if not obj.data: - raise ValueError("No embedding data received") - - for embedding in obj.data: - data = cast(object, embedding.embedding) - if not isinstance(data, str): - continue - if not has_numpy(): - # use array for base64 optimisation - embedding.embedding = array.array("f", base64.b64decode(data)).tolist() - else: - embedding.embedding = np.frombuffer( # type: ignore[no-untyped-call] - base64.b64decode(data), dtype="float32" - ).tolist() - - return obj - return await self._post( "/embeddings", body=maybe_transform(params, embedding_create_params.EmbeddingCreateParams), @@ -265,7 +220,7 @@ def parser(obj: CreateEmbeddingResponse) -> CreateEmbeddingResponse: extra_query=extra_query, extra_body=extra_body, timeout=timeout, - post_parser=parser, + post_parser=partial(_parse_embedding_response, encoding_format=encoding_format), security={"bearer_auth": True}, ), cast_to=CreateEmbeddingResponse, diff --git a/src/openai/resources/webhooks/webhooks.py b/src/openai/resources/webhooks/webhooks.py index 9a6e2ae42a..8080ce875f 100644 --- a/src/openai/resources/webhooks/webhooks.py +++ b/src/openai/resources/webhooks/webhooks.py @@ -2,18 +2,14 @@ from __future__ import annotations -import hmac import json -import time -import base64 -import hashlib from typing import cast from ..._types import HeadersLike -from ..._utils import get_required_header from ..._models import construct_type from ..._resource import SyncAPIResource, AsyncAPIResource from ..._exceptions import InvalidWebhookSignatureError +from ...lib._webhooks import webhook_signature_matches as _webhook_signature_matches from ...types.webhooks.unwrap_webhook_event import UnwrapWebhookEvent __all__ = ["Webhooks", "AsyncWebhooks"] @@ -66,50 +62,7 @@ def verify_signature( "on the client class, OpenAI(webhook_secret='123'), or passed to this function" ) - signature_header = get_required_header(headers, "webhook-signature") - timestamp = get_required_header(headers, "webhook-timestamp") - webhook_id = get_required_header(headers, "webhook-id") - - # Validate timestamp to prevent replay attacks - try: - timestamp_seconds = int(timestamp) - except ValueError: - raise InvalidWebhookSignatureError("Invalid webhook timestamp format") from None - - now = int(time.time()) - - if now - timestamp_seconds > tolerance: - raise InvalidWebhookSignatureError("Webhook timestamp is too old") from None - - if timestamp_seconds > now + tolerance: - raise InvalidWebhookSignatureError("Webhook timestamp is too new") from None - - # Extract signatures from v1, format - # The signature header can have multiple values, separated by spaces. - # Each value is in the format v1,. We should accept if any match. - signatures: list[str] = [] - for part in signature_header.split(): - if part.startswith("v1,"): - signatures.append(part[3:]) - else: - signatures.append(part) - - # Decode the secret if it starts with whsec_ - if secret.startswith("whsec_"): - decoded_secret = base64.b64decode(secret[6:]) - else: - decoded_secret = secret.encode() - - body = payload.decode("utf-8") if isinstance(payload, bytes) else payload - - # Prepare the signed payload (OpenAI uses webhookId.timestamp.payload format) - signed_payload = f"{webhook_id}.{timestamp}.{body}" - expected_signature = base64.b64encode( - hmac.new(decoded_secret, signed_payload.encode(), hashlib.sha256).digest() - ).decode() - - # Accept if any signature matches - if not any(hmac.compare_digest(expected_signature, sig) for sig in signatures): + if not _webhook_signature_matches(payload, headers, secret=secret, tolerance=tolerance): raise InvalidWebhookSignatureError( "The given webhook signature does not match the expected signature" ) from None @@ -163,48 +116,5 @@ def verify_signature( "on the client class, OpenAI(webhook_secret='123'), or passed to this function" ) from None - signature_header = get_required_header(headers, "webhook-signature") - timestamp = get_required_header(headers, "webhook-timestamp") - webhook_id = get_required_header(headers, "webhook-id") - - # Validate timestamp to prevent replay attacks - try: - timestamp_seconds = int(timestamp) - except ValueError: - raise InvalidWebhookSignatureError("Invalid webhook timestamp format") from None - - now = int(time.time()) - - if now - timestamp_seconds > tolerance: - raise InvalidWebhookSignatureError("Webhook timestamp is too old") from None - - if timestamp_seconds > now + tolerance: - raise InvalidWebhookSignatureError("Webhook timestamp is too new") from None - - # Extract signatures from v1, format - # The signature header can have multiple values, separated by spaces. - # Each value is in the format v1,. We should accept if any match. - signatures: list[str] = [] - for part in signature_header.split(): - if part.startswith("v1,"): - signatures.append(part[3:]) - else: - signatures.append(part) - - # Decode the secret if it starts with whsec_ - if secret.startswith("whsec_"): - decoded_secret = base64.b64decode(secret[6:]) - else: - decoded_secret = secret.encode() - - body = payload.decode("utf-8") if isinstance(payload, bytes) else payload - - # Prepare the signed payload (OpenAI uses webhookId.timestamp.payload format) - signed_payload = f"{webhook_id}.{timestamp}.{body}" - expected_signature = base64.b64encode( - hmac.new(decoded_secret, signed_payload.encode(), hashlib.sha256).digest() - ).decode() - - # Accept if any signature matches - if not any(hmac.compare_digest(expected_signature, sig) for sig in signatures): + if not _webhook_signature_matches(payload, headers, secret=secret, tolerance=tolerance): raise InvalidWebhookSignatureError("The given webhook signature does not match the expected signature") diff --git a/tests/api_resources/test_webhooks.py b/tests/api_resources/test_webhooks.py index f03b9780d4..c2f45e64c3 100644 --- a/tests/api_resources/test_webhooks.py +++ b/tests/api_resources/test_webhooks.py @@ -3,282 +3,17 @@ from __future__ import annotations import os -from unittest import mock import pytest -import openai -from openai._exceptions import InvalidWebhookSignatureError - base_url = os.environ.get("TEST_API_BASE_URL", "http://127.0.0.1:4010") -# Standardized test constants (matches TypeScript implementation) -TEST_SECRET = "whsec_RdvaYFYUXuIFuEbvZHwMfYFhUf7aMYjYcmM24+Aj40c=" -TEST_PAYLOAD = '{"id": "evt_685c059ae3a481909bdc86819b066fb6", "object": "event", "created_at": 1750861210, "type": "response.completed", "data": {"id": "resp_123"}}' -TEST_TIMESTAMP = 1750861210 # Fixed timestamp that matches our test signature -TEST_WEBHOOK_ID = "wh_685c059ae39c8190af8c71ed1022a24d" -TEST_SIGNATURE = "v1,gUAg4R2hWouRZqRQG4uJypNS8YK885G838+EHb4nKBY=" - - -def create_test_headers( - timestamp: int | None = None, signature: str | None = None, webhook_id: str | None = None -) -> dict[str, str]: - """Helper function to create test headers""" - return { - "webhook-signature": signature or TEST_SIGNATURE, - "webhook-timestamp": str(timestamp or TEST_TIMESTAMP), - "webhook-id": webhook_id or TEST_WEBHOOK_ID, - } - class TestWebhooks: parametrize = pytest.mark.parametrize("client", [False, True], indirect=True, ids=["loose", "strict"]) - @mock.patch("time.time", mock.MagicMock(return_value=TEST_TIMESTAMP)) - @parametrize - def test_unwrap_with_secret(self, client: openai.OpenAI) -> None: - headers = create_test_headers() - unwrapped = client.webhooks.unwrap(TEST_PAYLOAD, headers, secret=TEST_SECRET) - assert unwrapped.id == "evt_685c059ae3a481909bdc86819b066fb6" - assert unwrapped.created_at == 1750861210 - - @parametrize - def test_unwrap_without_secret(self, client: openai.OpenAI) -> None: - headers = create_test_headers() - with pytest.raises(ValueError, match="The webhook secret must either be set"): - client.webhooks.unwrap(TEST_PAYLOAD, headers) - - @mock.patch("time.time", mock.MagicMock(return_value=TEST_TIMESTAMP)) - @parametrize - def test_verify_signature_valid(self, client: openai.OpenAI) -> None: - headers = create_test_headers() - # Should not raise - this is a truly valid signature for this timestamp - client.webhooks.verify_signature(TEST_PAYLOAD, headers, secret=TEST_SECRET) - - @parametrize - def test_verify_signature_invalid_secret_format(self, client: openai.OpenAI) -> None: - headers = create_test_headers() - with pytest.raises(ValueError, match="The webhook secret must either be set"): - client.webhooks.verify_signature(TEST_PAYLOAD, headers, secret=None) - - @mock.patch("time.time", mock.MagicMock(return_value=TEST_TIMESTAMP)) - @parametrize - def test_verify_signature_invalid(self, client: openai.OpenAI) -> None: - headers = create_test_headers() - with pytest.raises(InvalidWebhookSignatureError, match="The given webhook signature does not match"): - client.webhooks.verify_signature(TEST_PAYLOAD, headers, secret="invalid_secret") - - @parametrize - def test_verify_signature_missing_webhook_signature_header(self, client: openai.OpenAI) -> None: - headers = create_test_headers(signature=None) - del headers["webhook-signature"] - with pytest.raises(ValueError, match="Could not find webhook-signature header"): - client.webhooks.verify_signature(TEST_PAYLOAD, headers, secret=TEST_SECRET) - - @parametrize - def test_verify_signature_missing_webhook_timestamp_header(self, client: openai.OpenAI) -> None: - headers = create_test_headers() - del headers["webhook-timestamp"] - with pytest.raises(ValueError, match="Could not find webhook-timestamp header"): - client.webhooks.verify_signature(TEST_PAYLOAD, headers, secret=TEST_SECRET) - - @parametrize - def test_verify_signature_missing_webhook_id_header(self, client: openai.OpenAI) -> None: - headers = create_test_headers() - del headers["webhook-id"] - with pytest.raises(ValueError, match="Could not find webhook-id header"): - client.webhooks.verify_signature(TEST_PAYLOAD, headers, secret=TEST_SECRET) - - @mock.patch("time.time", mock.MagicMock(return_value=TEST_TIMESTAMP)) - @parametrize - def test_verify_signature_payload_bytes(self, client: openai.OpenAI) -> None: - headers = create_test_headers() - client.webhooks.verify_signature(TEST_PAYLOAD.encode("utf-8"), headers, secret=TEST_SECRET) - - @mock.patch("time.time", mock.MagicMock(return_value=TEST_TIMESTAMP)) - def test_unwrap_with_client_secret(self) -> None: - test_client = openai.OpenAI(base_url=base_url, api_key="test-api-key", webhook_secret=TEST_SECRET) - headers = create_test_headers() - - unwrapped = test_client.webhooks.unwrap(TEST_PAYLOAD, headers) - assert unwrapped.id == "evt_685c059ae3a481909bdc86819b066fb6" - assert unwrapped.created_at == 1750861210 - - @parametrize - def test_verify_signature_timestamp_too_old(self, client: openai.OpenAI) -> None: - # Use a timestamp that's older than 5 minutes from our test timestamp - old_timestamp = TEST_TIMESTAMP - 400 # 6 minutes 40 seconds ago - headers = create_test_headers(timestamp=old_timestamp, signature="v1,dummy_signature") - - with pytest.raises(InvalidWebhookSignatureError, match="Webhook timestamp is too old"): - client.webhooks.verify_signature(TEST_PAYLOAD, headers, secret=TEST_SECRET) - - @mock.patch("time.time", mock.MagicMock(return_value=TEST_TIMESTAMP)) - @parametrize - def test_verify_signature_timestamp_too_new(self, client: openai.OpenAI) -> None: - # Use a timestamp that's in the future beyond tolerance from our test timestamp - future_timestamp = TEST_TIMESTAMP + 400 # 6 minutes 40 seconds in the future - headers = create_test_headers(timestamp=future_timestamp, signature="v1,dummy_signature") - - with pytest.raises(InvalidWebhookSignatureError, match="Webhook timestamp is too new"): - client.webhooks.verify_signature(TEST_PAYLOAD, headers, secret=TEST_SECRET) - - @mock.patch("time.time", mock.MagicMock(return_value=TEST_TIMESTAMP)) - @parametrize - def test_verify_signature_custom_tolerance(self, client: openai.OpenAI) -> None: - # Use a timestamp that's older than default tolerance but within custom tolerance - old_timestamp = TEST_TIMESTAMP - 400 # 6 minutes 40 seconds ago from test timestamp - headers = create_test_headers(timestamp=old_timestamp, signature="v1,dummy_signature") - - # Should fail with default tolerance - with pytest.raises(InvalidWebhookSignatureError, match="Webhook timestamp is too old"): - client.webhooks.verify_signature(TEST_PAYLOAD, headers, secret=TEST_SECRET) - - # Should also fail with custom tolerance of 10 minutes (signature won't match) - with pytest.raises(InvalidWebhookSignatureError, match="The given webhook signature does not match"): - client.webhooks.verify_signature(TEST_PAYLOAD, headers, secret=TEST_SECRET, tolerance=600) - - @mock.patch("time.time", mock.MagicMock(return_value=TEST_TIMESTAMP)) - @parametrize - def test_verify_signature_recent_timestamp_succeeds(self, client: openai.OpenAI) -> None: - # Use a recent timestamp with dummy signature - headers = create_test_headers(signature="v1,dummy_signature") - - # Should fail on signature verification (not timestamp validation) - with pytest.raises(InvalidWebhookSignatureError, match="The given webhook signature does not match"): - client.webhooks.verify_signature(TEST_PAYLOAD, headers, secret=TEST_SECRET) - - @mock.patch("time.time", mock.MagicMock(return_value=TEST_TIMESTAMP)) - @parametrize - def test_verify_signature_multiple_signatures_one_valid(self, client: openai.OpenAI) -> None: - # Test multiple signatures: one invalid, one valid - multiple_signatures = f"v1,invalid_signature {TEST_SIGNATURE}" - headers = create_test_headers(signature=multiple_signatures) - - # Should not raise when at least one signature is valid - client.webhooks.verify_signature(TEST_PAYLOAD, headers, secret=TEST_SECRET) - - @mock.patch("time.time", mock.MagicMock(return_value=TEST_TIMESTAMP)) - @parametrize - def test_verify_signature_multiple_signatures_all_invalid(self, client: openai.OpenAI) -> None: - # Test multiple invalid signatures - multiple_invalid_signatures = "v1,invalid_signature1 v1,invalid_signature2" - headers = create_test_headers(signature=multiple_invalid_signatures) - - with pytest.raises(InvalidWebhookSignatureError, match="The given webhook signature does not match"): - client.webhooks.verify_signature(TEST_PAYLOAD, headers, secret=TEST_SECRET) - class TestAsyncWebhooks: parametrize = pytest.mark.parametrize( "async_client", [False, True, {"http_client": "aiohttp"}], indirect=True, ids=["loose", "strict", "aiohttp"] ) - - @mock.patch("time.time", mock.MagicMock(return_value=TEST_TIMESTAMP)) - @parametrize - async def test_unwrap_with_secret(self, async_client: openai.AsyncOpenAI) -> None: - headers = create_test_headers() - unwrapped = async_client.webhooks.unwrap(TEST_PAYLOAD, headers, secret=TEST_SECRET) - assert unwrapped.id == "evt_685c059ae3a481909bdc86819b066fb6" - assert unwrapped.created_at == 1750861210 - - @parametrize - async def test_unwrap_without_secret(self, async_client: openai.AsyncOpenAI) -> None: - headers = create_test_headers() - with pytest.raises(ValueError, match="The webhook secret must either be set"): - async_client.webhooks.unwrap(TEST_PAYLOAD, headers) - - @mock.patch("time.time", mock.MagicMock(return_value=TEST_TIMESTAMP)) - @parametrize - async def test_verify_signature_valid(self, async_client: openai.AsyncOpenAI) -> None: - headers = create_test_headers() - # Should not raise - this is a truly valid signature for this timestamp - async_client.webhooks.verify_signature(TEST_PAYLOAD, headers, secret=TEST_SECRET) - - @parametrize - async def test_verify_signature_invalid_secret_format(self, async_client: openai.AsyncOpenAI) -> None: - headers = create_test_headers() - with pytest.raises(ValueError, match="The webhook secret must either be set"): - async_client.webhooks.verify_signature(TEST_PAYLOAD, headers, secret=None) - - @mock.patch("time.time", mock.MagicMock(return_value=TEST_TIMESTAMP)) - @parametrize - async def test_verify_signature_invalid(self, async_client: openai.AsyncOpenAI) -> None: - headers = create_test_headers() - with pytest.raises(InvalidWebhookSignatureError, match="The given webhook signature does not match"): - async_client.webhooks.verify_signature(TEST_PAYLOAD, headers, secret="invalid_secret") - - @parametrize - async def test_verify_signature_missing_webhook_signature_header(self, async_client: openai.AsyncOpenAI) -> None: - headers = create_test_headers() - del headers["webhook-signature"] - with pytest.raises(ValueError, match="Could not find webhook-signature header"): - async_client.webhooks.verify_signature(TEST_PAYLOAD, headers, secret=TEST_SECRET) - - @parametrize - async def test_verify_signature_missing_webhook_timestamp_header(self, async_client: openai.AsyncOpenAI) -> None: - headers = create_test_headers() - del headers["webhook-timestamp"] - with pytest.raises(ValueError, match="Could not find webhook-timestamp header"): - async_client.webhooks.verify_signature(TEST_PAYLOAD, headers, secret=TEST_SECRET) - - @parametrize - async def test_verify_signature_missing_webhook_id_header(self, async_client: openai.AsyncOpenAI) -> None: - headers = create_test_headers() - del headers["webhook-id"] - with pytest.raises(ValueError, match="Could not find webhook-id header"): - async_client.webhooks.verify_signature(TEST_PAYLOAD, headers, secret=TEST_SECRET) - - @mock.patch("time.time", mock.MagicMock(return_value=TEST_TIMESTAMP)) - @parametrize - async def test_verify_signature_payload_bytes(self, async_client: openai.AsyncOpenAI) -> None: - headers = create_test_headers() - async_client.webhooks.verify_signature(TEST_PAYLOAD.encode("utf-8"), headers, secret=TEST_SECRET) - - @mock.patch("time.time", mock.MagicMock(return_value=TEST_TIMESTAMP)) - async def test_unwrap_with_client_secret(self) -> None: - test_async_client = openai.AsyncOpenAI(base_url=base_url, api_key="test-api-key", webhook_secret=TEST_SECRET) - headers = create_test_headers() - - unwrapped = test_async_client.webhooks.unwrap(TEST_PAYLOAD, headers) - assert unwrapped.id == "evt_685c059ae3a481909bdc86819b066fb6" - assert unwrapped.created_at == 1750861210 - - @parametrize - async def test_verify_signature_timestamp_too_old(self, async_client: openai.AsyncOpenAI) -> None: - # Use a timestamp that's older than 5 minutes from our test timestamp - old_timestamp = TEST_TIMESTAMP - 400 # 6 minutes 40 seconds ago - headers = create_test_headers(timestamp=old_timestamp, signature="v1,dummy_signature") - - with pytest.raises(InvalidWebhookSignatureError, match="Webhook timestamp is too old"): - async_client.webhooks.verify_signature(TEST_PAYLOAD, headers, secret=TEST_SECRET) - - @mock.patch("time.time", mock.MagicMock(return_value=TEST_TIMESTAMP)) - @parametrize - async def test_verify_signature_timestamp_too_new(self, async_client: openai.AsyncOpenAI) -> None: - # Use a timestamp that's in the future beyond tolerance from our test timestamp - future_timestamp = TEST_TIMESTAMP + 400 # 6 minutes 40 seconds in the future - headers = create_test_headers(timestamp=future_timestamp, signature="v1,dummy_signature") - - with pytest.raises(InvalidWebhookSignatureError, match="Webhook timestamp is too new"): - async_client.webhooks.verify_signature(TEST_PAYLOAD, headers, secret=TEST_SECRET) - - @mock.patch("time.time", mock.MagicMock(return_value=TEST_TIMESTAMP)) - @parametrize - async def test_verify_signature_multiple_signatures_one_valid(self, async_client: openai.AsyncOpenAI) -> None: - # Test multiple signatures: one invalid, one valid - multiple_signatures = f"v1,invalid_signature {TEST_SIGNATURE}" - headers = create_test_headers(signature=multiple_signatures) - - # Should not raise when at least one signature is valid - async_client.webhooks.verify_signature(TEST_PAYLOAD, headers, secret=TEST_SECRET) - - @mock.patch("time.time", mock.MagicMock(return_value=TEST_TIMESTAMP)) - @parametrize - async def test_verify_signature_multiple_signatures_all_invalid(self, async_client: openai.AsyncOpenAI) -> None: - # Test multiple invalid signatures - multiple_invalid_signatures = "v1,invalid_signature1 v1,invalid_signature2" - headers = create_test_headers(signature=multiple_invalid_signatures) - - with pytest.raises(InvalidWebhookSignatureError, match="The given webhook signature does not match"): - async_client.webhooks.verify_signature(TEST_PAYLOAD, headers, secret=TEST_SECRET) diff --git a/tests/lib/test_audio_response_format.py b/tests/lib/test_audio_response_format.py new file mode 100644 index 0000000000..b7e23dc00d --- /dev/null +++ b/tests/lib/test_audio_response_format.py @@ -0,0 +1,236 @@ +from __future__ import annotations + +import logging +from typing import Any, cast +from unittest.mock import Mock +from typing_extensions import Literal + +import httpx2 +import pytest + +from openai import OpenAI, AsyncOpenAI +from tests.respx2 import MockRouter +from openai._types import Omit, omit, not_given +from openai.types.audio import ( + Translation, + Transcription, + TranslationVerbose, + TranscriptionVerbose, + TranscriptionDiarized, + TranscriptionTextDeltaEvent, +) +from openai.resources.audio import translations, transcriptions +from openai.types.audio_response_format import AudioResponseFormat + +AudioResource = Literal["transcriptions", "translations"] +ResponseMode = Literal["normal", "raw", "streaming"] +FORMAT_CASES = [ + pytest.param("transcriptions", omit, Transcription, id="transcription-default"), + pytest.param("transcriptions", "json", Transcription, id="transcription-json"), + pytest.param("transcriptions", "verbose_json", TranscriptionVerbose, id="transcription-verbose"), + pytest.param("transcriptions", "diarized_json", TranscriptionDiarized, id="transcription-diarized"), + pytest.param("transcriptions", "text", str, id="transcription-text"), + pytest.param("transcriptions", "srt", str, id="transcription-srt"), + pytest.param("transcriptions", "vtt", str, id="transcription-vtt"), + pytest.param("translations", omit, Translation, id="translation-default"), + pytest.param("translations", "json", Translation, id="translation-json"), + pytest.param("translations", "verbose_json", TranslationVerbose, id="translation-verbose"), + pytest.param("translations", "text", str, id="translation-text"), + pytest.param("translations", "srt", str, id="translation-srt"), + pytest.param("translations", "vtt", str, id="translation-vtt"), +] + + +def select_response_type(resource: AudioResource, response_format: object) -> type[object]: + if resource == "transcriptions": + return transcriptions._get_response_format_type(cast(Any, response_format)) + return translations._get_response_format_type(cast(Any, response_format)) + + +@pytest.mark.parametrize("resource,response_format,expected_type", FORMAT_CASES) +def test_supported_formats_keep_exact_response_classes( + resource: AudioResource, response_format: AudioResponseFormat | Omit, expected_type: type[object] +) -> None: + assert select_response_type(resource, response_format) is expected_type + + +@pytest.mark.parametrize( + "resource,expected_type", + [("transcriptions", Transcription), ("translations", Translation)], +) +@pytest.mark.parametrize("response_format", [None, Omit()], ids=["none", "new-omit"]) +def test_default_compatibility_values( + resource: AudioResource, response_format: object, expected_type: type[object] +) -> None: + assert select_response_type(resource, response_format) is expected_type + + +@pytest.mark.parametrize( + "resource,response_format,expected_type", + [ + ("transcriptions", "future-format", Transcription), + ("transcriptions", not_given, Transcription), + ("translations", "future-format", Translation), + ("translations", not_given, Translation), + ("translations", "diarized_json", Translation), + ], +) +def test_fallback_keeps_resource_logger_and_method( + resource: AudioResource, response_format: object, expected_type: type[object], monkeypatch: pytest.MonkeyPatch +) -> None: + logger = Mock(spec=logging.Logger) + module = transcriptions if resource == "transcriptions" else translations + monkeypatch.setattr(module, "log", logger) + + assert select_response_type(resource, response_format) is expected_type + + if resource == "transcriptions": + logger.warn.assert_called_once_with("Unexpected audio response format: %s", response_format) + logger.warning.assert_not_called() + else: + logger.warning.assert_called_once_with("Unexpected audio response format: %s", response_format) + logger.warn.assert_not_called() + + +def test_fallback_keeps_historical_logger_name_and_warning(caplog: pytest.LogCaptureFixture) -> None: + logger = logging.getLogger("openai.audio.transcriptions") + assert transcriptions.log is logger + assert translations.log is logger + + with pytest.warns(DeprecationWarning, match="deprecated"): + assert select_response_type("transcriptions", "future-format") is Transcription + assert select_response_type("translations", "future-format") is Translation + + records = [record for record in caplog.records if record.name == logger.name] + assert [(record.levelno, record.getMessage()) for record in records] == [ + (logging.WARNING, "Unexpected audio response format: future-format"), + (logging.WARNING, "Unexpected audio response format: future-format"), + ] + + +def make_response(expected_type: type[object]) -> httpx2.Response: + if expected_type is str: + return httpx2.Response(200, text="test transcript") + return httpx2.Response( + 200, + json={ + "text": "test transcript", + "duration": 1.0, + "language": "english", + "segments": [], + "task": "transcribe", + }, + ) + + +def assert_request_and_response( + request: httpx2.Request, + response: object, + response_format: AudioResponseFormat | Omit, + expected_type: type[object], +) -> None: + assert type(response) is expected_type + assert request.headers["content-type"].startswith("multipart/form-data; boundary=") + if isinstance(response_format, Omit): + assert b'name="response_format"' not in request.content + else: + assert f'name="response_format"\r\n\r\n{response_format}\r\n'.encode() in request.content + + +@pytest.mark.respx2() +@pytest.mark.parametrize("client", [False, True], indirect=True, ids=["loose", "strict"]) +@pytest.mark.parametrize("resource,response_format,expected_type", FORMAT_CASES) +@pytest.mark.parametrize("mode", ["normal", "raw", "streaming"]) +def test_sync_create_uses_selected_response_class( + client: OpenAI, + respx2_mock: MockRouter, + resource: AudioResource, + response_format: AudioResponseFormat | Omit, + expected_type: type[object], + mode: ResponseMode, +) -> None: + route = respx2_mock.post(f"{str(client.base_url).rstrip('/')}/audio/{resource}").mock( + return_value=make_response(expected_type) + ) + audio = cast(Any, client.audio.transcriptions if resource == "transcriptions" else client.audio.translations) + kwargs: dict[str, Any] = {"file": b"fake audio", "model": "whisper-1", "response_format": response_format} + + if mode == "normal": + response = audio.create(**kwargs) + elif mode == "raw": + response = audio.with_raw_response.create(**kwargs).parse() + else: + with audio.with_streaming_response.create(**kwargs) as raw_response: + response = raw_response.parse() + + assert route.call_count == 1 + assert_request_and_response(route.calls.last.request, response, response_format, expected_type) + + +@pytest.mark.respx2() +@pytest.mark.parametrize("async_client", [False, True], indirect=True, ids=["loose", "strict"]) +@pytest.mark.parametrize("resource,response_format,expected_type", FORMAT_CASES) +@pytest.mark.parametrize("mode", ["normal", "raw", "streaming"]) +async def test_async_create_uses_selected_response_class( + async_client: AsyncOpenAI, + respx2_mock: MockRouter, + resource: AudioResource, + response_format: AudioResponseFormat | Omit, + expected_type: type[object], + mode: ResponseMode, +) -> None: + route = respx2_mock.post(f"{str(async_client.base_url).rstrip('/')}/audio/{resource}").mock( + return_value=make_response(expected_type) + ) + audio = cast( + Any, async_client.audio.transcriptions if resource == "transcriptions" else async_client.audio.translations + ) + kwargs: dict[str, Any] = {"file": b"fake audio", "model": "whisper-1", "response_format": response_format} + + if mode == "normal": + response = await audio.create(**kwargs) + elif mode == "raw": + response = (await audio.with_raw_response.create(**kwargs)).parse() + else: + async with audio.with_streaming_response.create(**kwargs) as raw_response: + response = await raw_response.parse() + + assert route.call_count == 1 + assert_request_and_response(route.calls.last.request, response, response_format, expected_type) + + +STREAM_BODY = 'event: transcript.text.delta\ndata: {"type":"transcript.text.delta","delta":"hello"}\n\ndata: [DONE]\n\n' + + +@pytest.mark.respx2() +def test_sync_transcription_stream_still_yields_events(client: OpenAI, respx2_mock: MockRouter) -> None: + respx2_mock.post(f"{str(client.base_url).rstrip('/')}/audio/transcriptions").mock( + return_value=httpx2.Response(200, content=STREAM_BODY, headers={"content-type": "text/event-stream"}) + ) + + with client.audio.transcriptions.create( + file=b"fake audio", model="gpt-4o-transcribe", response_format="json", stream=True + ) as stream: + events = list(stream) + + assert len(events) == 1 + assert isinstance(events[0], TranscriptionTextDeltaEvent) + assert events[0].delta == "hello" + + +@pytest.mark.respx2() +async def test_async_transcription_stream_still_yields_events( + async_client: AsyncOpenAI, respx2_mock: MockRouter +) -> None: + respx2_mock.post(f"{str(async_client.base_url).rstrip('/')}/audio/transcriptions").mock( + return_value=httpx2.Response(200, content=STREAM_BODY, headers={"content-type": "text/event-stream"}) + ) + + async with await async_client.audio.transcriptions.create( + file=b"fake audio", model="gpt-4o-transcribe", response_format="json", stream=True + ) as stream: + events = [event async for event in stream] + + assert len(events) == 1 + assert isinstance(events[0], TranscriptionTextDeltaEvent) + assert events[0].delta == "hello" diff --git a/tests/lib/test_embeddings.py b/tests/lib/test_embeddings.py new file mode 100644 index 0000000000..787d580716 --- /dev/null +++ b/tests/lib/test_embeddings.py @@ -0,0 +1,185 @@ +from __future__ import annotations + +import json +import array +import base64 +import binascii +from typing import Any, cast +from typing_extensions import Literal + +import httpx2 +import pytest + +from openai import OpenAI, AsyncOpenAI +from tests.respx2 import MockRouter +from openai._types import Omit, NotGiven, omit, not_given +from openai._models import construct_type_unchecked +from openai.lib._parsing import _embeddings as embeddings_parser +from openai.types.create_embedding_response import CreateEmbeddingResponse + +VALUES = [0.125, -2.5, 3.75] +ENCODED = base64.b64encode(array.array("f", VALUES).tobytes()).decode("ascii") +EncodingFormat = Literal["float", "base64"] | Omit +ResponseMode = Literal["normal", "raw", "streaming"] + + +def response_body(*vectors: object) -> dict[str, object]: + return { + "data": [{"embedding": vector, "index": index, "object": "embedding"} for index, vector in enumerate(vectors)], + "model": "text-embedding-3-small", + "object": "list", + "usage": {"prompt_tokens": 1, "total_tokens": 1}, + } + + +def make_response(*vectors: object) -> CreateEmbeddingResponse: + return construct_type_unchecked(type_=CreateEmbeddingResponse, value=response_body(*vectors)) + + +@pytest.fixture(params=[False, True], ids=["stdlib", "numpy"]) +def decoder(request: pytest.FixtureRequest, monkeypatch: pytest.MonkeyPatch) -> None: + use_numpy = cast(bool, request.param) + if use_numpy: + pytest.importorskip("numpy") + monkeypatch.setattr(embeddings_parser, "has_numpy", lambda: use_numpy) + + +@pytest.mark.usefixtures("decoder") +@pytest.mark.parametrize("encoding_format", [omit, not_given], ids=["omit", "not-given"]) +def test_decode_preserves_response_and_non_string_vectors(encoding_format: Omit | NotGiven) -> None: + response = make_response(ENCODED, [4.0, 5.0], ENCODED) + data = response.data + unchanged_vector = data[1].embedding + usage = response.usage + + parsed = embeddings_parser.parse_embedding_response(response, encoding_format=encoding_format) + + assert parsed is response + assert parsed.data is data + assert parsed.data[0].embedding == VALUES + assert parsed.data[1].embedding is unchanged_vector + assert parsed.data[2].embedding == VALUES + assert parsed.usage is usage + assert parsed.model == "text-embedding-3-small" + + +@pytest.mark.parametrize("encoding_format", ["float", "base64", None]) +@pytest.mark.parametrize("vectors", [(ENCODED,), ("abc",), ()], ids=["encoded", "invalid", "empty"]) +def test_explicit_format_is_untouched( + encoding_format: object, vectors: tuple[object, ...], monkeypatch: pytest.MonkeyPatch +) -> None: + def unexpected_decoder() -> bool: + raise AssertionError("an explicit format must not inspect the decoder") + + monkeypatch.setattr(embeddings_parser, "has_numpy", unexpected_decoder) + response = make_response(*vectors) + data = response.data + original = [cast(object, item.embedding) for item in data] + + assert embeddings_parser.parse_embedding_response(response, encoding_format=cast(Any, encoding_format)) is response + assert response.data is data + assert [cast(object, item.embedding) for item in data] == original + + +@pytest.mark.parametrize("encoding_format", [omit, not_given], ids=["omit", "not-given"]) +@pytest.mark.parametrize("data", [[], None], ids=["empty", "null"]) +def test_missing_data_keeps_existing_error(encoding_format: Omit | NotGiven, data: object) -> None: + body = response_body() + body["data"] = data + response = construct_type_unchecked(type_=CreateEmbeddingResponse, value=body) + + with pytest.raises(ValueError, match=r"^No embedding data received$"): + embeddings_parser.parse_embedding_response(response, encoding_format=encoding_format) + + +@pytest.mark.usefixtures("decoder") +@pytest.mark.parametrize( + "encoded,error", + [("abc", binascii.Error), (base64.b64encode(b"abc").decode("ascii"), ValueError)], + ids=["invalid-base64", "invalid-float-buffer"], +) +def test_invalid_data_preserves_decoder_errors(encoded: str, error: type[Exception]) -> None: + response = make_response(encoded) + + with pytest.raises(error): + embeddings_parser.parse_embedding_response(response, encoding_format=omit) + + +def assert_request_and_response( + request: httpx2.Request, response: CreateEmbeddingResponse, encoding_format: EncodingFormat +) -> None: + expected_format = encoding_format if isinstance(encoding_format, str) else "base64" + body = json.loads(request.content) + assert body == { + "input": "test input", + "model": "text-embedding-3-small", + "user": "fake-user", + "dimensions": 3, + "encoding_format": expected_format, + } + expected_vector = ENCODED if encoding_format == "base64" else VALUES + assert cast(object, response.data[0].embedding) == expected_vector + + +@pytest.mark.respx2() +@pytest.mark.usefixtures("decoder") +@pytest.mark.parametrize("client", [False], indirect=True) +@pytest.mark.parametrize("encoding_format", [omit, "float", "base64"], ids=["default", "float", "base64"]) +@pytest.mark.parametrize("mode", ["normal", "raw", "streaming"]) +def test_sync_create_uses_decoder( + client: OpenAI, respx2_mock: MockRouter, encoding_format: EncodingFormat, mode: ResponseMode +) -> None: + vector = VALUES if encoding_format == "float" else ENCODED + route = respx2_mock.post(f"{str(client.base_url).rstrip('/')}/embeddings").mock( + return_value=httpx2.Response(200, json=response_body(vector)) + ) + kwargs: dict[str, Any] = { + "input": "test input", + "model": "text-embedding-3-small", + "user": "fake-user", + "dimensions": 3, + "encoding_format": encoding_format, + } + + if mode == "normal": + response = client.embeddings.create(**kwargs) + elif mode == "raw": + response = client.embeddings.with_raw_response.create(**kwargs).parse() + else: + with client.embeddings.with_streaming_response.create(**kwargs) as raw_response: + response = raw_response.parse() + + assert route.call_count == 1 + assert_request_and_response(route.calls.last.request, response, encoding_format) + + +@pytest.mark.respx2() +@pytest.mark.usefixtures("decoder") +@pytest.mark.parametrize("async_client", [False], indirect=True) +@pytest.mark.parametrize("encoding_format", [omit, "float", "base64"], ids=["default", "float", "base64"]) +@pytest.mark.parametrize("mode", ["normal", "raw", "streaming"]) +async def test_async_create_uses_decoder( + async_client: AsyncOpenAI, respx2_mock: MockRouter, encoding_format: EncodingFormat, mode: ResponseMode +) -> None: + vector = VALUES if encoding_format == "float" else ENCODED + route = respx2_mock.post(f"{str(async_client.base_url).rstrip('/')}/embeddings").mock( + return_value=httpx2.Response(200, json=response_body(vector)) + ) + kwargs: dict[str, Any] = { + "input": "test input", + "model": "text-embedding-3-small", + "user": "fake-user", + "dimensions": 3, + "encoding_format": encoding_format, + } + + if mode == "normal": + response = await async_client.embeddings.create(**kwargs) + elif mode == "raw": + response = (await async_client.embeddings.with_raw_response.create(**kwargs)).parse() + else: + async with async_client.embeddings.with_streaming_response.create(**kwargs) as raw_response: + response = await raw_response.parse() + + assert route.call_count == 1 + assert_request_and_response(route.calls.last.request, response, encoding_format) diff --git a/tests/lib/test_webhook_signature.py b/tests/lib/test_webhook_signature.py new file mode 100644 index 0000000000..5ba9724162 --- /dev/null +++ b/tests/lib/test_webhook_signature.py @@ -0,0 +1,211 @@ +from __future__ import annotations + +import hmac +import base64 +import hashlib +import binascii +from typing import Iterator, cast +from unittest import mock + +import pytest + +import openai +from openai._exceptions import InvalidWebhookSignatureError +from openai.lib._webhooks import webhook_signature_matches +from openai.resources.webhooks.webhooks import Webhooks, AsyncWebhooks + +NOW = 1_750_000_000 +PAYLOAD = '{"synthetic": "café"}' +WEBHOOK_ID = "evt_synthetic" +RAW_SECRET = "synthetic-webhook-secret" +PREFIXED_SECRET = "whsec_" + base64.b64encode(RAW_SECRET.encode()).decode() + + +def signed_headers( + payload: str | bytes = PAYLOAD, + *, + timestamp: str = str(NOW), + secret: bytes = RAW_SECRET.encode(), +) -> dict[str, str]: + body = payload.encode() if isinstance(payload, str) else payload + signed = f"{WEBHOOK_ID}.{timestamp}.".encode() + body + signature = base64.b64encode(hmac.new(secret, signed, hashlib.sha256).digest()).decode() + return { + "webhook-id": WEBHOOK_ID, + "webhook-timestamp": timestamp, + "webhook-signature": f"v1,{signature}", + } + + +@pytest.fixture(autouse=True) +def frozen_time() -> Iterator[None]: + with mock.patch("time.time", return_value=NOW): + yield + + +@pytest.fixture(params=["sync", "async"]) +def webhook_resource(request: pytest.FixtureRequest) -> Webhooks | AsyncWebhooks: + client = mock.Mock() + client.webhook_secret = None + if request.param == "sync": + return Webhooks(cast(openai.OpenAI, client)) + return AsyncWebhooks(cast(openai.AsyncOpenAI, client)) + + +@pytest.mark.parametrize( + ("secret", "key"), + [(RAW_SECRET, RAW_SECRET.encode()), (PREFIXED_SECRET, RAW_SECRET.encode()), ("", b"")], + ids=["raw", "prefixed", "empty"], +) +@pytest.mark.parametrize("payload", [PAYLOAD, PAYLOAD.encode()], ids=["text", "bytes"]) +@pytest.mark.parametrize("signature_form", ["prefixed", "bare", "multiple"]) +def test_signature_forms_and_secret_encodings( + webhook_resource: Webhooks | AsyncWebhooks, + secret: str, + key: bytes, + payload: str | bytes, + signature_form: str, +) -> None: + headers = signed_headers(payload, secret=key) + signature = headers["webhook-signature"][3:] + if signature_form == "bare": + headers["webhook-signature"] = signature + elif signature_form == "multiple": + headers["webhook-signature"] = f"v1,invalid\t{signature} v1,also-invalid" + headers = {name.upper(): value for name, value in headers.items()} + + assert webhook_signature_matches(payload, headers, secret=secret, tolerance=300) + assert webhook_resource.verify_signature(payload, headers, secret=secret) is None + + +@pytest.mark.parametrize( + ("delta", "tolerance", "error"), + [ + (0, 0, None), + (-300, 300, None), + (300, 300, None), + (-301, 300, "Webhook timestamp is too old"), + (301, 300, "Webhook timestamp is too new"), + (-1, 0, "Webhook timestamp is too old"), + (1, 0, "Webhook timestamp is too new"), + (0, -1, "Webhook timestamp is too old"), + ], +) +def test_replay_window_boundaries( + webhook_resource: Webhooks | AsyncWebhooks, + delta: int, + tolerance: int, + error: str | None, +) -> None: + headers = signed_headers(timestamp=str(NOW + delta)) + if error is None: + webhook_resource.verify_signature(PAYLOAD, headers, secret=RAW_SECRET, tolerance=tolerance) + else: + with pytest.raises(InvalidWebhookSignatureError, match=error) as caught: + webhook_resource.verify_signature(PAYLOAD, headers, secret=RAW_SECRET, tolerance=tolerance) + assert caught.value.__cause__ is None + assert caught.value.__suppress_context__ + + +@pytest.mark.parametrize("timestamp", [f"0{NOW}", f"+{NOW}", f" {NOW} "]) +def test_signature_uses_original_timestamp_text(webhook_resource: Webhooks | AsyncWebhooks, timestamp: str) -> None: + webhook_resource.verify_signature(PAYLOAD, signed_headers(timestamp=timestamp), secret=RAW_SECRET) + + +@pytest.mark.parametrize("timestamp", ["", "not-a-timestamp", "1.5"]) +def test_invalid_timestamp_exception(webhook_resource: Webhooks | AsyncWebhooks, timestamp: str) -> None: + with pytest.raises(InvalidWebhookSignatureError, match="Invalid webhook timestamp format") as caught: + webhook_resource.verify_signature(PAYLOAD, signed_headers(timestamp=timestamp), secret=RAW_SECRET) + assert caught.value.__cause__ is None + assert caught.value.__suppress_context__ + + +@pytest.mark.parametrize( + ("headers", "missing"), + [ + ({}, "webhook-signature"), + ({"webhook-signature": "invalid"}, "webhook-timestamp"), + ({"webhook-signature": "invalid", "webhook-timestamp": str(NOW)}, "webhook-id"), + ], +) +def test_required_header_order( + webhook_resource: Webhooks | AsyncWebhooks, headers: dict[str, str], missing: str +) -> None: + with pytest.raises(ValueError, match=f"Could not find {missing} header"): + webhook_resource.verify_signature(PAYLOAD, headers, secret=RAW_SECRET) + + +def test_client_secret_fallback_preserves_explicit_empty_secret( + webhook_resource: Webhooks | AsyncWebhooks, +) -> None: + webhook_resource._client.webhook_secret = PREFIXED_SECRET + webhook_resource.verify_signature(PAYLOAD, signed_headers()) + webhook_resource.verify_signature(PAYLOAD, signed_headers(secret=b""), secret="") + + +def test_missing_secret_preserves_wrapper_exception_chaining( + webhook_resource: Webhooks | AsyncWebhooks, +) -> None: + previous = RuntimeError("synthetic prior failure") + try: + raise previous + except RuntimeError: + with pytest.raises(ValueError, match="The webhook secret must either be set") as caught: + webhook_resource.verify_signature(PAYLOAD, {}) + + assert caught.value.__context__ is previous + assert caught.value.__cause__ is None + assert caught.value.__suppress_context__ is isinstance(webhook_resource, AsyncWebhooks) + + +def test_mismatch_preserves_wrapper_exception_chaining( + webhook_resource: Webhooks | AsyncWebhooks, +) -> None: + headers = signed_headers() + headers["webhook-signature"] = "v1,synthetic-invalid-signature" + previous = RuntimeError("synthetic prior failure") + try: + raise previous + except RuntimeError: + with pytest.raises(InvalidWebhookSignatureError, match="does not match the expected signature") as caught: + webhook_resource.verify_signature(PAYLOAD, headers, secret=RAW_SECRET) + + assert caught.value.__context__ is previous + assert caught.value.__cause__ is None + assert caught.value.__suppress_context__ is (not isinstance(webhook_resource, AsyncWebhooks)) + assert RAW_SECRET not in str(caught.value) + assert PAYLOAD not in str(caught.value) + assert headers["webhook-signature"] not in str(caught.value) + + +def test_malformed_secret_keeps_base64_error(webhook_resource: Webhooks | AsyncWebhooks) -> None: + with pytest.raises(binascii.Error): + webhook_resource.verify_signature(PAYLOAD, signed_headers(), secret="whsec_a") + + +def test_invalid_utf8_payload_keeps_decode_error(webhook_resource: Webhooks | AsyncWebhooks) -> None: + with pytest.raises(UnicodeDecodeError): + webhook_resource.verify_signature(b"\xff", signed_headers(), secret=RAW_SECRET) + + +def test_non_ascii_signature_keeps_compare_error(webhook_resource: Webhooks | AsyncWebhooks) -> None: + headers = signed_headers() + headers["webhook-signature"] = "v1,é" + with pytest.raises(TypeError, match="non-ASCII"): + webhook_resource.verify_signature(PAYLOAD, headers, secret=RAW_SECRET) + + +def test_constant_time_comparison_order(webhook_resource: Webhooks | AsyncWebhooks) -> None: + headers = signed_headers() + expected = headers["webhook-signature"][3:] + headers["webhook-signature"] = f"v1,invalid v1,{expected} v1,unused" + with mock.patch("openai.lib._webhooks.hmac.compare_digest", wraps=hmac.compare_digest) as compare: + webhook_resource.verify_signature(PAYLOAD, headers, secret=RAW_SECRET) + assert compare.call_args_list == [mock.call(expected, "invalid"), mock.call(expected, expected)] + + +@pytest.mark.parametrize("signature", ["", "v1,invalid", "invalid v1,also-invalid"]) +def test_helper_returns_false_for_mismatch(signature: str) -> None: + headers = signed_headers() + headers["webhook-signature"] = signature + assert not webhook_signature_matches(PAYLOAD, headers, secret=RAW_SECRET, tolerance=300) diff --git a/tests/lib/test_webhooks.py b/tests/lib/test_webhooks.py new file mode 100644 index 0000000000..09e457aee9 --- /dev/null +++ b/tests/lib/test_webhooks.py @@ -0,0 +1,282 @@ +from __future__ import annotations + +import os +from unittest import mock + +import pytest + +import openai +from openai._exceptions import InvalidWebhookSignatureError + +base_url = os.environ.get("TEST_API_BASE_URL", "http://127.0.0.1:4010") + +# Standardized test constants (matches TypeScript implementation) +TEST_SECRET = "whsec_RdvaYFYUXuIFuEbvZHwMfYFhUf7aMYjYcmM24+Aj40c=" +TEST_PAYLOAD = '{"id": "evt_685c059ae3a481909bdc86819b066fb6", "object": "event", "created_at": 1750861210, "type": "response.completed", "data": {"id": "resp_123"}}' +TEST_TIMESTAMP = 1750861210 # Fixed timestamp that matches our test signature +TEST_WEBHOOK_ID = "wh_685c059ae39c8190af8c71ed1022a24d" +TEST_SIGNATURE = "v1,gUAg4R2hWouRZqRQG4uJypNS8YK885G838+EHb4nKBY=" + + +def create_test_headers( + timestamp: int | None = None, signature: str | None = None, webhook_id: str | None = None +) -> dict[str, str]: + """Helper function to create test headers""" + return { + "webhook-signature": signature or TEST_SIGNATURE, + "webhook-timestamp": str(timestamp or TEST_TIMESTAMP), + "webhook-id": webhook_id or TEST_WEBHOOK_ID, + } + + +class TestWebhooks: + parametrize = pytest.mark.parametrize("client", [False, True], indirect=True, ids=["loose", "strict"]) + + @mock.patch("time.time", mock.MagicMock(return_value=TEST_TIMESTAMP)) + @parametrize + def test_unwrap_with_secret(self, client: openai.OpenAI) -> None: + headers = create_test_headers() + unwrapped = client.webhooks.unwrap(TEST_PAYLOAD, headers, secret=TEST_SECRET) + assert unwrapped.id == "evt_685c059ae3a481909bdc86819b066fb6" + assert unwrapped.created_at == 1750861210 + + @parametrize + def test_unwrap_without_secret(self, client: openai.OpenAI) -> None: + headers = create_test_headers() + with pytest.raises(ValueError, match="The webhook secret must either be set"): + client.webhooks.unwrap(TEST_PAYLOAD, headers) + + @mock.patch("time.time", mock.MagicMock(return_value=TEST_TIMESTAMP)) + @parametrize + def test_verify_signature_valid(self, client: openai.OpenAI) -> None: + headers = create_test_headers() + # Should not raise - this is a truly valid signature for this timestamp + client.webhooks.verify_signature(TEST_PAYLOAD, headers, secret=TEST_SECRET) + + @parametrize + def test_verify_signature_invalid_secret_format(self, client: openai.OpenAI) -> None: + headers = create_test_headers() + with pytest.raises(ValueError, match="The webhook secret must either be set"): + client.webhooks.verify_signature(TEST_PAYLOAD, headers, secret=None) + + @mock.patch("time.time", mock.MagicMock(return_value=TEST_TIMESTAMP)) + @parametrize + def test_verify_signature_invalid(self, client: openai.OpenAI) -> None: + headers = create_test_headers() + with pytest.raises(InvalidWebhookSignatureError, match="The given webhook signature does not match"): + client.webhooks.verify_signature(TEST_PAYLOAD, headers, secret="invalid_secret") + + @parametrize + def test_verify_signature_missing_webhook_signature_header(self, client: openai.OpenAI) -> None: + headers = create_test_headers(signature=None) + del headers["webhook-signature"] + with pytest.raises(ValueError, match="Could not find webhook-signature header"): + client.webhooks.verify_signature(TEST_PAYLOAD, headers, secret=TEST_SECRET) + + @parametrize + def test_verify_signature_missing_webhook_timestamp_header(self, client: openai.OpenAI) -> None: + headers = create_test_headers() + del headers["webhook-timestamp"] + with pytest.raises(ValueError, match="Could not find webhook-timestamp header"): + client.webhooks.verify_signature(TEST_PAYLOAD, headers, secret=TEST_SECRET) + + @parametrize + def test_verify_signature_missing_webhook_id_header(self, client: openai.OpenAI) -> None: + headers = create_test_headers() + del headers["webhook-id"] + with pytest.raises(ValueError, match="Could not find webhook-id header"): + client.webhooks.verify_signature(TEST_PAYLOAD, headers, secret=TEST_SECRET) + + @mock.patch("time.time", mock.MagicMock(return_value=TEST_TIMESTAMP)) + @parametrize + def test_verify_signature_payload_bytes(self, client: openai.OpenAI) -> None: + headers = create_test_headers() + client.webhooks.verify_signature(TEST_PAYLOAD.encode("utf-8"), headers, secret=TEST_SECRET) + + @mock.patch("time.time", mock.MagicMock(return_value=TEST_TIMESTAMP)) + def test_unwrap_with_client_secret(self) -> None: + test_client = openai.OpenAI(base_url=base_url, api_key="test-api-key", webhook_secret=TEST_SECRET) + headers = create_test_headers() + + unwrapped = test_client.webhooks.unwrap(TEST_PAYLOAD, headers) + assert unwrapped.id == "evt_685c059ae3a481909bdc86819b066fb6" + assert unwrapped.created_at == 1750861210 + + @parametrize + def test_verify_signature_timestamp_too_old(self, client: openai.OpenAI) -> None: + # Use a timestamp that's older than 5 minutes from our test timestamp + old_timestamp = TEST_TIMESTAMP - 400 # 6 minutes 40 seconds ago + headers = create_test_headers(timestamp=old_timestamp, signature="v1,dummy_signature") + + with pytest.raises(InvalidWebhookSignatureError, match="Webhook timestamp is too old"): + client.webhooks.verify_signature(TEST_PAYLOAD, headers, secret=TEST_SECRET) + + @mock.patch("time.time", mock.MagicMock(return_value=TEST_TIMESTAMP)) + @parametrize + def test_verify_signature_timestamp_too_new(self, client: openai.OpenAI) -> None: + # Use a timestamp that's in the future beyond tolerance from our test timestamp + future_timestamp = TEST_TIMESTAMP + 400 # 6 minutes 40 seconds in the future + headers = create_test_headers(timestamp=future_timestamp, signature="v1,dummy_signature") + + with pytest.raises(InvalidWebhookSignatureError, match="Webhook timestamp is too new"): + client.webhooks.verify_signature(TEST_PAYLOAD, headers, secret=TEST_SECRET) + + @mock.patch("time.time", mock.MagicMock(return_value=TEST_TIMESTAMP)) + @parametrize + def test_verify_signature_custom_tolerance(self, client: openai.OpenAI) -> None: + # Use a timestamp that's older than default tolerance but within custom tolerance + old_timestamp = TEST_TIMESTAMP - 400 # 6 minutes 40 seconds ago from test timestamp + headers = create_test_headers(timestamp=old_timestamp, signature="v1,dummy_signature") + + # Should fail with default tolerance + with pytest.raises(InvalidWebhookSignatureError, match="Webhook timestamp is too old"): + client.webhooks.verify_signature(TEST_PAYLOAD, headers, secret=TEST_SECRET) + + # Should also fail with custom tolerance of 10 minutes (signature won't match) + with pytest.raises(InvalidWebhookSignatureError, match="The given webhook signature does not match"): + client.webhooks.verify_signature(TEST_PAYLOAD, headers, secret=TEST_SECRET, tolerance=600) + + @mock.patch("time.time", mock.MagicMock(return_value=TEST_TIMESTAMP)) + @parametrize + def test_verify_signature_recent_timestamp_succeeds(self, client: openai.OpenAI) -> None: + # Use a recent timestamp with dummy signature + headers = create_test_headers(signature="v1,dummy_signature") + + # Should fail on signature verification (not timestamp validation) + with pytest.raises(InvalidWebhookSignatureError, match="The given webhook signature does not match"): + client.webhooks.verify_signature(TEST_PAYLOAD, headers, secret=TEST_SECRET) + + @mock.patch("time.time", mock.MagicMock(return_value=TEST_TIMESTAMP)) + @parametrize + def test_verify_signature_multiple_signatures_one_valid(self, client: openai.OpenAI) -> None: + # Test multiple signatures: one invalid, one valid + multiple_signatures = f"v1,invalid_signature {TEST_SIGNATURE}" + headers = create_test_headers(signature=multiple_signatures) + + # Should not raise when at least one signature is valid + client.webhooks.verify_signature(TEST_PAYLOAD, headers, secret=TEST_SECRET) + + @mock.patch("time.time", mock.MagicMock(return_value=TEST_TIMESTAMP)) + @parametrize + def test_verify_signature_multiple_signatures_all_invalid(self, client: openai.OpenAI) -> None: + # Test multiple invalid signatures + multiple_invalid_signatures = "v1,invalid_signature1 v1,invalid_signature2" + headers = create_test_headers(signature=multiple_invalid_signatures) + + with pytest.raises(InvalidWebhookSignatureError, match="The given webhook signature does not match"): + client.webhooks.verify_signature(TEST_PAYLOAD, headers, secret=TEST_SECRET) + + +class TestAsyncWebhooks: + parametrize = pytest.mark.parametrize( + "async_client", [False, True, {"http_client": "aiohttp"}], indirect=True, ids=["loose", "strict", "aiohttp"] + ) + + @mock.patch("time.time", mock.MagicMock(return_value=TEST_TIMESTAMP)) + @parametrize + async def test_unwrap_with_secret(self, async_client: openai.AsyncOpenAI) -> None: + headers = create_test_headers() + unwrapped = async_client.webhooks.unwrap(TEST_PAYLOAD, headers, secret=TEST_SECRET) + assert unwrapped.id == "evt_685c059ae3a481909bdc86819b066fb6" + assert unwrapped.created_at == 1750861210 + + @parametrize + async def test_unwrap_without_secret(self, async_client: openai.AsyncOpenAI) -> None: + headers = create_test_headers() + with pytest.raises(ValueError, match="The webhook secret must either be set"): + async_client.webhooks.unwrap(TEST_PAYLOAD, headers) + + @mock.patch("time.time", mock.MagicMock(return_value=TEST_TIMESTAMP)) + @parametrize + async def test_verify_signature_valid(self, async_client: openai.AsyncOpenAI) -> None: + headers = create_test_headers() + # Should not raise - this is a truly valid signature for this timestamp + async_client.webhooks.verify_signature(TEST_PAYLOAD, headers, secret=TEST_SECRET) + + @parametrize + async def test_verify_signature_invalid_secret_format(self, async_client: openai.AsyncOpenAI) -> None: + headers = create_test_headers() + with pytest.raises(ValueError, match="The webhook secret must either be set"): + async_client.webhooks.verify_signature(TEST_PAYLOAD, headers, secret=None) + + @mock.patch("time.time", mock.MagicMock(return_value=TEST_TIMESTAMP)) + @parametrize + async def test_verify_signature_invalid(self, async_client: openai.AsyncOpenAI) -> None: + headers = create_test_headers() + with pytest.raises(InvalidWebhookSignatureError, match="The given webhook signature does not match"): + async_client.webhooks.verify_signature(TEST_PAYLOAD, headers, secret="invalid_secret") + + @parametrize + async def test_verify_signature_missing_webhook_signature_header(self, async_client: openai.AsyncOpenAI) -> None: + headers = create_test_headers() + del headers["webhook-signature"] + with pytest.raises(ValueError, match="Could not find webhook-signature header"): + async_client.webhooks.verify_signature(TEST_PAYLOAD, headers, secret=TEST_SECRET) + + @parametrize + async def test_verify_signature_missing_webhook_timestamp_header(self, async_client: openai.AsyncOpenAI) -> None: + headers = create_test_headers() + del headers["webhook-timestamp"] + with pytest.raises(ValueError, match="Could not find webhook-timestamp header"): + async_client.webhooks.verify_signature(TEST_PAYLOAD, headers, secret=TEST_SECRET) + + @parametrize + async def test_verify_signature_missing_webhook_id_header(self, async_client: openai.AsyncOpenAI) -> None: + headers = create_test_headers() + del headers["webhook-id"] + with pytest.raises(ValueError, match="Could not find webhook-id header"): + async_client.webhooks.verify_signature(TEST_PAYLOAD, headers, secret=TEST_SECRET) + + @mock.patch("time.time", mock.MagicMock(return_value=TEST_TIMESTAMP)) + @parametrize + async def test_verify_signature_payload_bytes(self, async_client: openai.AsyncOpenAI) -> None: + headers = create_test_headers() + async_client.webhooks.verify_signature(TEST_PAYLOAD.encode("utf-8"), headers, secret=TEST_SECRET) + + @mock.patch("time.time", mock.MagicMock(return_value=TEST_TIMESTAMP)) + async def test_unwrap_with_client_secret(self) -> None: + test_async_client = openai.AsyncOpenAI(base_url=base_url, api_key="test-api-key", webhook_secret=TEST_SECRET) + headers = create_test_headers() + + unwrapped = test_async_client.webhooks.unwrap(TEST_PAYLOAD, headers) + assert unwrapped.id == "evt_685c059ae3a481909bdc86819b066fb6" + assert unwrapped.created_at == 1750861210 + + @parametrize + async def test_verify_signature_timestamp_too_old(self, async_client: openai.AsyncOpenAI) -> None: + # Use a timestamp that's older than 5 minutes from our test timestamp + old_timestamp = TEST_TIMESTAMP - 400 # 6 minutes 40 seconds ago + headers = create_test_headers(timestamp=old_timestamp, signature="v1,dummy_signature") + + with pytest.raises(InvalidWebhookSignatureError, match="Webhook timestamp is too old"): + async_client.webhooks.verify_signature(TEST_PAYLOAD, headers, secret=TEST_SECRET) + + @mock.patch("time.time", mock.MagicMock(return_value=TEST_TIMESTAMP)) + @parametrize + async def test_verify_signature_timestamp_too_new(self, async_client: openai.AsyncOpenAI) -> None: + # Use a timestamp that's in the future beyond tolerance from our test timestamp + future_timestamp = TEST_TIMESTAMP + 400 # 6 minutes 40 seconds in the future + headers = create_test_headers(timestamp=future_timestamp, signature="v1,dummy_signature") + + with pytest.raises(InvalidWebhookSignatureError, match="Webhook timestamp is too new"): + async_client.webhooks.verify_signature(TEST_PAYLOAD, headers, secret=TEST_SECRET) + + @mock.patch("time.time", mock.MagicMock(return_value=TEST_TIMESTAMP)) + @parametrize + async def test_verify_signature_multiple_signatures_one_valid(self, async_client: openai.AsyncOpenAI) -> None: + # Test multiple signatures: one invalid, one valid + multiple_signatures = f"v1,invalid_signature {TEST_SIGNATURE}" + headers = create_test_headers(signature=multiple_signatures) + + # Should not raise when at least one signature is valid + async_client.webhooks.verify_signature(TEST_PAYLOAD, headers, secret=TEST_SECRET) + + @mock.patch("time.time", mock.MagicMock(return_value=TEST_TIMESTAMP)) + @parametrize + async def test_verify_signature_multiple_signatures_all_invalid(self, async_client: openai.AsyncOpenAI) -> None: + # Test multiple invalid signatures + multiple_invalid_signatures = "v1,invalid_signature1 v1,invalid_signature2" + headers = create_test_headers(signature=multiple_invalid_signatures) + + with pytest.raises(InvalidWebhookSignatureError, match="The given webhook signature does not match"): + async_client.webhooks.verify_signature(TEST_PAYLOAD, headers, secret=TEST_SECRET)