diff --git a/src/omnibase_infra/event_bus/kafka_auth.py b/src/omnibase_infra/event_bus/kafka_auth.py index 56027a060d..5e9059f5f3 100644 --- a/src/omnibase_infra/event_bus/kafka_auth.py +++ b/src/omnibase_infra/event_bus/kafka_auth.py @@ -135,7 +135,7 @@ def build_aiokafka_auth_kwargs(config: ModelKafkaEventBusConfig) -> dict[str, ob kwargs["sasl_mechanism"] = "OAUTHBEARER" kwargs["sasl_oauth_token_provider"] = MSKTokenProvider(region=config.msk_region) - if config.ssl_ca_file is not None: + if config.security_protocol in ("SSL", "SASL_SSL"): kwargs["ssl_context"] = ssl.create_default_context(cafile=config.ssl_ca_file) return kwargs diff --git a/tests/unit/event_bus/test_msk_auth_direct_clients.py b/tests/unit/event_bus/test_msk_auth_direct_clients.py index eea0748301..717e16614d 100644 --- a/tests/unit/event_bus/test_msk_auth_direct_clients.py +++ b/tests/unit/event_bus/test_msk_auth_direct_clients.py @@ -82,6 +82,23 @@ def test_ssl_ca_file_builds_aiokafka_ssl_context() -> None: create_context.assert_called_once_with(cafile="/etc/ssl/certs/custom-ca.pem") +def test_msk_iam_builds_default_aiokafka_ssl_context() -> None: + with patch("ssl.create_default_context") as create_context: + context = object() + create_context.return_value = context + config = ModelKafkaEventBusConfig( + bootstrap_servers="b-1.example:9098", + security_protocol="SASL_SSL", + sasl_mechanism="AWS_MSK_IAM", + msk_region="us-east-1", + ) + + kwargs = build_aiokafka_auth_kwargs(config) + + assert kwargs["ssl_context"] is context + create_context.assert_called_once_with(cafile=None) + + @pytest.mark.asyncio async def test_provider_kafka_producer_passes_msk_auth_kwargs() -> None: auth_kwargs = {"security_protocol": "SASL_SSL", "sasl_mechanism": "OAUTHBEARER"}