Add config param to pass additional parameters to confluent-kafka-python

docs/docs_src/confluent/security/basic.py

@@ -1,9 +1,6 @@
-import ssl
-
 from faststream.confluent import KafkaBroker
 from faststream.security import BaseSecurity
 
-ssl_context = ssl.create_default_context()
-security = BaseSecurity(ssl_context=ssl_context)
+security = BaseSecurity(use_ssl=True)
 
 broker = KafkaBroker("localhost:9092", security=security)

docs/docs_src/confluent/security/plaintext.py

@@ -1,13 +1,10 @@
-import ssl
-
 from faststream.confluent import KafkaBroker
 from faststream.security import SASLPlaintext
 
-ssl_context = ssl.create_default_context()
 security = SASLPlaintext(
-    ssl_context=ssl_context,
     username="admin",
     password="password",  # pragma: allowlist secret
+    use_ssl=True,
 )
 
 broker = KafkaBroker("localhost:9092", security=security)

docs/docs_src/confluent/security/sasl_scram256.py

@@ -1,13 +1,10 @@
-import ssl
-
 from faststream.confluent import KafkaBroker
 from faststream.security import SASLScram256
 
-ssl_context = ssl.create_default_context()
 security = SASLScram256(
-    ssl_context=ssl_context,
     username="admin",
     password="password",  # pragma: allowlist secret
+    use_ssl=True,
 )
 
 broker = KafkaBroker("localhost:9092", security=security)

docs/docs_src/confluent/security/sasl_scram512.py

@@ -1,13 +1,10 @@
-import ssl
-
 from faststream.confluent import KafkaBroker
 from faststream.security import SASLScram512
 
-ssl_context = ssl.create_default_context()
 security = SASLScram512(
-    ssl_context=ssl_context,
     username="admin",
     password="password",  # pragma: allowlist secret
+    use_ssl=True,
 )
 
 broker = KafkaBroker("localhost:9092", security=security)

faststream/confluent/security.py

@@ -1,3 +1,4 @@
+import ssl
 import warnings
 from typing import TYPE_CHECKING, Optional
 
@@ -14,6 +15,9 @@
 
 
 def parse_security(security: Optional[BaseSecurity]) -> "AnyDict":
+    if security and isinstance(security.ssl_context, ssl.SSLContext):
+        raise ValueError("ssl_context in not supported by confluent-kafka-python, please use config instead.")
+
     if security is None:
         return {}
     elif type(security) == BaseSecurity:
@@ -35,7 +39,7 @@ def _parse_base_security(security: BaseSecurity) -> "AnyDict":
 
 
 def _parse_sasl_plaintext(security: SASLPlaintext) -> "AnyDict":
-    if security.ssl_context is None:
+    if not security.use_ssl:
         warnings.warn(
             message=ssl_not_set_error_msg,
             category=RuntimeWarning,

tests/brokers/confluent/test_security.py

@@ -33,14 +33,30 @@ async def test_base_security():
             producer_call_kwargs = producer.call_args.kwargs
 
             call_kwargs = {}
-            call_kwargs["security_protocol"] = "SSL"
 
             assert call_kwargs.items() <= producer_call_kwargs.items()
 
-            assert (
-                producer_call_kwargs["security_protocol"]
-                == call_kwargs["security_protocol"]
-            )
+
+@pytest.mark.asyncio()
+@pytest.mark.confluent()
+async def test_base_security_pass_ssl_context():
+    import ssl
+
+    from faststream.confluent import KafkaBroker
+    from faststream.security import BaseSecurity
+
+    ssl_context = ssl.create_default_context()
+    security = BaseSecurity(ssl_context=ssl_context)
+
+    basic_broker = KafkaBroker("localhost:9092", security=security)
+
+
+    with patch_aio_consumer_and_producer() as producer:
+        with pytest.raises(ValueError) as e:
+            async with basic_broker:
+                pass
+
+    assert str(e.value) == "ssl_context in not supported by confluent-kafka-python, please use config instead."
 
 
 @pytest.mark.asyncio()