Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions src/typesense/async_/api_call.py
Original file line number Diff line number Diff line change
Expand Up @@ -160,6 +160,7 @@ def __init__(self, config: Configuration):
self.request_handler = RequestHandler(config)
self._client = httpx.AsyncClient(
timeout=config.connection_timeout_seconds,
verify=config.verify,
)

async def __aenter__(self) -> "AsyncApiCall":
Expand Down
1 change: 1 addition & 0 deletions src/typesense/sync/api_call.py
Original file line number Diff line number Diff line change
Expand Up @@ -160,6 +160,7 @@ def __init__(self, config: Configuration):
self.request_handler = RequestHandler(config)
self._client = httpx.Client(
timeout=config.connection_timeout_seconds,
verify=config.verify,
)

def __enter__(self) -> "ApiCall":
Expand Down
75 changes: 75 additions & 0 deletions tests/api_call_ssl_test.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,75 @@
"""Regression tests for TLS configuration in both HTTP clients."""

import ssl
import warnings
from pathlib import Path

import certifi
import pytest
from pytest_mock import MockerFixture

from typesense.async_.api_call import AsyncApiCall
from typesense.configuration import Configuration
from typesense.sync.api_call import ApiCall


@pytest.fixture(params=[ApiCall, AsyncApiCall])
def api_call_class(request):
"""Exercise both the async source and the generated sync client."""
return request.param


async def close_api_call(api_call):
if isinstance(api_call, AsyncApiCall):
await api_call.aclose()
else:
api_call.close()


@pytest.mark.parametrize("verify", [True, False])
async def test_verification_mode(fake_config, api_call_class, verify):
"""The effective TLS context must honor explicit verification settings."""
fake_config.verify = verify
api_call = api_call_class(fake_config)
try:
ssl_context = api_call._client._transport._pool._ssl_context
assert ssl_context.verify_mode == (
ssl.CERT_REQUIRED if verify else ssl.CERT_NONE
)
assert ssl_context.check_hostname is verify
finally:
await close_api_call(api_call)


async def test_custom_ca_bundle(
fake_config: Configuration,
api_call_class,
tmp_path: Path,
mocker: MockerFixture,
):
"""A configured CA bundle must reach the real SSL context builder."""
ca_bundle = tmp_path / "custom-ca.pem"
ca_bundle.write_bytes(Path(certifi.where()).read_bytes())
fake_config.verify = str(ca_bundle)
create_context = mocker.spy(ssl, "create_default_context")

with warnings.catch_warnings():
warnings.filterwarnings(
"ignore", message="`verify=<str>`", category=DeprecationWarning
)
api_call = api_call_class(fake_config)
try:
create_context.assert_any_call(cafile=str(ca_bundle))
finally:
await close_api_call(api_call)


async def test_missing_ca_bundle(fake_config, api_call_class, tmp_path):
"""An invalid CA path must fail instead of silently using default trust roots."""
fake_config.verify = str(tmp_path / "missing-ca.pem")
with warnings.catch_warnings():
warnings.filterwarnings(
"ignore", message="`verify=<str>`", category=DeprecationWarning
)
with pytest.raises(FileNotFoundError):
api_call_class(fake_config)