Repository: airtai/faststream (python)

Issue:
Bug: [ConfluentKafka] Oauth configuration is not passed correctly to Consumer and Producer
**Describe the bug**
Hey, We are currently exploring this package for production usage. We would like to authenticate to an AWS MSK cluster via IAM auth. We got it to work with `confluent_kafka` so that we are sure the configuration is correct. We however did not get the connection with faststream to work. We are following the instructions from [here](https://faststream.airt.ai/latest/confluent/security/#4-sasloauthbearer-object-with-ssltls)


**How to reproduce**
Include source code:

```python
from faststream import FastStream
from faststream.confluent import KafkaBroker
from faststream.security import SASLOAuthBearer


security = SASLOAuthBearer(
    use_ssl=True,
)

broker = KafkaBroker("<bootstrap_servers>", security=security)

app = FastStream(broker)
```

And/Or steps to reproduce the behavior:

Run the code with `faststream run app:app`

**Expected behavior**
Faststream should start up correctly if IAM permissions are correctly set up.

**Observed behavior**
We are seeing a Kafka exception that indicates a wrong `sasl.mechanism` is provided to the consumer:

```
cimpl.KafkaException: KafkaError{code=_INVALID_ARG,val=-186,str="Failed to create producer: No provider for SASL mechanism GSSAPI: recompile librdkafka with libsasl2 or openssl support. Current build options: PLAIN SASL_SCRAM OAUTHBEARER"}
```

**Full tracelog**

```
2025-03-04 09:04:53,408 INFO     - FastStream app starting...
Traceback (most recent call last):

  File "/opt/pysetup/.venv/bin/faststream", line 8, in <module>
    sys.exit(cli())
             ^^^^^

  File "/opt/pysetup/.venv/lib/python3.11/site-packages/faststream/cli/main.py", line 172, in run
    _run_imported_app(

  File "/opt/pysetup/.venv/lib/python3.11/site-packages/faststream/cli/main.py", line 218, in _run_imported_app
    anyio.run(

  File "/opt/pysetup/.venv/lib/python3.11/site-packages/anyio/_core/_eventloop.py", line 74, in run
    return async_backend.run(func, args, {}, backend_options)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

  File "/opt/pysetup/.venv/lib/python3.11/site-packages/anyio/_backends/_asyncio.py", line 2303, in run
    return runner.run(wrapper())
           ^^^^^^^^^^^^^^^^^^^^^

  File "/usr/local/lib/python3.11/asyncio/runners.py", line 118, in run
    return self._loop.run_until_complete(task)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

  File "/usr/local/lib/python3.11/asyncio/base_events.py", line 653, in run_until_complete
    return future.result()
           ^^^^^^^^^^^^^^^

  File "/opt/pysetup/.venv/lib/python3.11/site-packages/anyio/_backends/_asyncio.py", line 2291, in wrapper
    return await func(*args)
           ^^^^^^^^^^^^^^^^^

  File "/opt/pysetup/.venv/lib/python3.11/site-packages/faststream/app.py", line 55, in run
    raise ex from None

  File "/opt/pysetup/.venv/lib/python3.11/site-packages/faststream/_internal/application.py", line 199, in _startup
    await self.start(**(run_extra_options or {}))

  File "/opt/pysetup/.venv/lib/python3.11/site-packages/faststream/_internal/application.py", line 177, in start
    await self.broker.start()

  File "/opt/pysetup/.venv/lib/python3.11/site-packages/faststream/confluent/broker/broker.py", line 456, in start
    await super().start()

  File "/opt/pysetup/.venv/lib/python3.11/site-packages/faststream/broker/core/usecase.py", line 219, in start
    await self.connect()

  File "/opt/pysetup/.venv/lib/python3.11/site-packages/faststream/confluent/broker/broker.py", line 423, in connect
    return await super().connect(**kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

  File "/opt/pysetup/.venv/lib/python3.11/site-packages/faststream/broker/core/usecase.py", line 226, in connect
    self._connection = await self._connect(**connection_kwargs)
                       ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

  File "/opt/pysetup/.venv/lib/python3.11/site-packages/faststream/confluent/broker/broker.py", line 435, in _connect
    native_producer = AsyncConfluentProducer(
                      ^^^^^^^^^^^^^^^^^^^^^^^

  File "/opt/pysetup/.venv/lib/python3.11/site-packages/faststream/confluent/client.py", line 115, in __init__
    self.producer = Producer(final_config, logger=self.logger)  # type: ignore[call-arg]
                    ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

cimpl.KafkaException: KafkaError{code=_INVALID_ARG,val=-186,str="Failed to create producer: No provider for SASL mechanism GSSAPI: recompile librdkafka with libsasl2 or openssl support. Current build options: PLAIN SASL_SCRAM OAUTHBEARER"}
```

**Environment**

```
faststream -v
> Running FastStream 0.5.34 with CPython 3.11.6 on Linux
```

**Additional context**

We were digging a bit into the codebase and think a config parameter is not properly propagated to the consumer and producer class:

```python
def _parse_sasl_oauthbearer(security: SASLOAuthBearer) -> "AnyDict":
    return {
        "security_protocol": "SASL_SSL" if security.use_ssl else "SASL_PLAINTEXT",
        "sasl_mechanism": "OAUTHBEARER",
    }
```

The parsed configuration from the security class looks as expected.

```python
# KafkaBroker class

    @override
    async def _connect(  # type: ignore[override]
        self,
        *,
        client_id: str,
        **kwargs: Any,
    ) -> Callable[..., AsyncConfluentConsumer]:
        security_params = parse_security(self.security)
        kwargs.update(security_params)

        native_producer = AsyncConfluentProducer(
            **kwargs,
            client_id=client_id,
            logger=self.logger,
            config=self.config,
        )
```

Security parameters are added as kwargs to the producer and consumer classes.

```python
# AsyncConfluentProducer class

        config_from_params = {
            # "topic.metadata.refresh.interval.ms": 1000,
            "bootstrap.servers": bootstrap_servers,
            "client.id": client_id,
            "metadata.max.age.ms": metadata_max_age_ms,
            "request.timeout.ms": request_timeout_ms,
            "acks": acks,
            "compression.type": compression_type,
            "partitioner": partitioner,
            "message.max.bytes": max_request_size,
            "linger.ms": linger_ms,
            "enable.idempotence": enable_idempotence,
            "transactional.id": transactional_id,
            "transaction.timeout.ms": transaction_timeout_ms,
            "retry.backoff.ms": retry_backoff_ms,
            "security.protocol": security_protocol.lower(),
            "connections.max.idle.ms": connections_max_idle_ms,
            "allow.auto.create.topics": allow_auto_create_topics,
        }

        final_config = {**config.as_config_dict(), **config_from_params}

        if sasl_mechanism in ["PLAIN", "SCRAM-SHA-256", "SCRAM-SHA-512"]:
            final_config.update(
                {
                    "sasl.mechanism": sasl_mechanism,
                    "sasl.username": sasl_plain_username,
                    "sasl.password": sasl_plain_password,
                }
            )

        self.producer = Producer(final_config, logger=self.logger)
```

`sasl_mechanism` is of value `OAUTHBEARER` and _not_ passed to the Producer so that it falls back to `GSSAPI` which matches the exception.

**Workaround**

The value can be provided via config:

```python
from faststream import FastStream
from faststream.confluent import KafkaBroker
from faststream.confluent.config import ConfluentConfig
from faststream.security import SASLOAuthBearer


security = SASLOAuthBearer(
    use_ssl=True,
)

broker = KafkaBroker(
    "localhost:9098",
    security=security,
    config=ConfluentConfig(
        {
            "sasl.mechanism": "OAUTHBEARER",
        }
    ),
)

app = FastStream(broker)
```

**Suggested solution**

The `sasl.mechanism` should be set based on the provided security class.

Pull request:
fix #2088: respect parsed sasl_mechanism
# Description

Please include a summary of the change and specify which issue is being addressed. Additionally, provide relevant motivation and context.

Fixes #2088 

## Type of change

Please delete options that are not relevant.

- [ ] Documentation (typos, code examples, or any documentation updates)
- [x] Bug fix (a non-breaking change that resolves an issue)
- [ ] New feature (a non-breaking change that adds functionality)
- [ ] Breaking change (a fix or feature that would disrupt existing functionality)
- [ ] This change requires a documentation update

## Checklist

- [ ] My code adheres to the style guidelines of this project (`scripts/lint.sh` shows no errors)
- [ ] I have conducted a self-review of my own code
- [ ] I have made the necessary changes to the documentation
- [ ] My changes do not generate any new warnings
- [ ] I have added tests to validate the effectiveness of my fix or the functionality of my new feature
- [ ] Both new and existing unit tests pass successfully on my local environment by running `scripts/test-cov.sh`
- [ ] I have ensured that static analysis tests are passing by running `scripts/static-analysis.sh`
- [ ] I have included code examples to illustrate the modifications


Patch:
diff --git a/docs/docs/SUMMARY.md b/docs/docs/SUMMARY.md
index ebb3c44a..ce6ca266 100644
--- a/docs/docs/SUMMARY.md
+++ b/docs/docs/SUMMARY.md
@@ -564,6 +564,7 @@ search:
                     - [TopicPartition](api/faststream/confluent/schemas/TopicPartition.md)
                     - params
                         - [ConsumerConnectionParams](api/faststream/confluent/schemas/params/ConsumerConnectionParams.md)
+                        - [SecurityOptions](api/faststream/confluent/schemas/params/SecurityOptions.md)
                     - partition
                         - [TopicPartition](api/faststream/confluent/schemas/partition/TopicPartition.md)
                 - security
diff --git a/docs/docs/en/api/faststream/confluent/schemas/params/SecurityOptions.md b/docs/docs/en/api/faststream/confluent/schemas/params/SecurityOptions.md
new file mode 100644
index 00000000..d8e40ac4
--- /dev/null
+++ b/docs/docs/en/api/faststream/confluent/schemas/params/SecurityOptions.md
@@ -0,0 +1,11 @@
+---
+# 0.5 - API
+# 2 - Release
+# 3 - Contributing
+# 5 - Template Page
+# 10 - Default
+search:
+  boost: 0.5
+---
+
+::: faststream.confluent.schemas.params.SecurityOptions
diff --git a/faststream/__about__.py b/faststream/__about__.py
index 9f10e18e..a0fb0711 100644
--- a/faststream/__about__.py
+++ b/faststream/__about__.py
@@ -1,5 +1,5 @@
 """Simple and fast framework to create message brokers based microservices."""
 
-__version__ = "0.5.35"
+__version__ = "0.5.36"
 
 SERVICE_NAME = f"faststream-{__version__}"
diff --git a/faststream/confluent/client.py b/faststream/confluent/client.py
index eb8ce40a..666f5670 100644
--- a/faststream/confluent/client.py
+++ b/faststream/confluent/client.py
@@ -29,6 +29,7 @@ from faststream.utils.functions import call_or_await
 if TYPE_CHECKING:
     from typing_extensions import NotRequired, TypedDict
 
+    from faststream.confluent.schemas.params import SecurityOptions
     from faststream.types import AnyDict, LoggerProto
 
     class _SendKwargs(TypedDict):
@@ -64,9 +65,7 @@ class AsyncConfluentProducer:
         transactional_id: Optional[Union[str, int]] = None,
         transaction_timeout_ms: int = 60000,
         allow_auto_create_topics: bool = True,
-        sasl_mechanism: Optional[str] = None,
-        sasl_plain_password: Optional[str] = None,
-        sasl_plain_username: Optional[str] = None,
+        security_config: Optional["SecurityOptions"] = None,
     ) -> None:
         self.logger = logger
 
@@ -101,16 +100,11 @@ class AsyncConfluentProducer:
             "allow.auto.create.topics": allow_auto_create_topics,
         }
 
-        final_config = {**config.as_config_dict(), **config_from_params}
-
-        if sasl_mechanism in ["PLAIN", "SCRAM-SHA-256", "SCRAM-SHA-512"]:
-            final_config.update(
-                {
-                    "sasl.mechanism": sasl_mechanism,
-                    "sasl.username": sasl_plain_username,
-                    "sasl.password": sasl_plain_password,
-                }
-            )
+        final_config = {
+            **config.as_config_dict(),
+            **config_from_params,
+            **dict(security_config or {}),
+        }
 
         self.producer = Producer(final_config, logger=self.logger)  # type: ignore[call-arg]
 
@@ -247,9 +241,7 @@ class AsyncConfluentConsumer:
         connections_max_idle_ms: int = 540000,
         isolation_level: str = "read_uncommitted",
         allow_auto_create_topics: bool = True,
-        sasl_mechanism: Optional[str] = None,
-        sasl_plain_password: Optional[str] = None,
-        sasl_plain_username: Optional[str] = None,
+        security_config: Optional["SecurityOptions"] = None,
     ) -> None:
         self.logger = logger
 
@@ -284,7 +276,6 @@ class AsyncConfluentConsumer:
             "fetch.max.bytes": fetch_max_bytes,
             "fetch.min.bytes": fetch_min_bytes,
             "max.partition.fetch.bytes": max_partition_fetch_bytes,
-            # "request.timeout.ms": 1000,  # producer only
             "fetch.error.backoff.ms": retry_backoff_ms,
             "auto.offset.reset": auto_offset_reset,
             "enable.auto.commit": enable_auto_commit,
@@ -301,15 +292,7 @@ class AsyncConfluentConsumer:
         }
         self.allow_auto_create_topics = allow_auto_create_topics
         final_config.update(config_from_params)
-
-        if sasl_mechanism in ["PLAIN", "SCRAM-SHA-256", "SCRAM-SHA-512"]:
-            final_config.update(
-                {
-                    "sasl.mechanism": sasl_mechanism,
-                    "sasl.username": sasl_plain_username,
-                    "sasl.password": sasl_plain_password,
-                }
-            )
+        final_config.update(**dict(security_config or {}))
 
         self.config = final_config
         self.consumer = Consumer(final_config, logger=self.logger)  # type: ignore[call-arg]
diff --git a/faststream/confluent/schemas/params.py b/faststream/confluent/schemas/params.py
index 91cb8359..e24a6ad4 100644
--- a/faststream/confluent/schemas/params.py
+++ b/faststream/confluent/schemas/params.py
@@ -1,6 +1,20 @@
 from typing import List, Literal, Union
 
-from typing_extensions import TypedDict
+from typing_extensions import Required, TypedDict
+
+
+class SecurityOptions(TypedDict, total=False):
+    sasl_mechanism: Required[
+        Literal[
+            "PLAIN",
+            "GSSAPI",
+            "SCRAM-SHA-256",
+            "SCRAM-SHA-512",
+            "OAUTHBEARER",
+        ]
+    ]
+    sasl_plain_password: str
+    sasl_plain_username: str
 
 
 class ConsumerConnectionParams(TypedDict, total=False):
@@ -16,12 +30,4 @@ class ConsumerConnectionParams(TypedDict, total=False):
     ]
     connections_max_idle_ms: int
     allow_auto_create_topics: bool
-    sasl_mechanism: Literal[
-        "PLAIN",
-        "GSSAPI",
-        "SCRAM-SHA-256",
-        "SCRAM-SHA-512",
-        "OAUTHBEARER",
-    ]
-    sasl_plain_password: str
-    sasl_plain_username: str
+    security_config: SecurityOptions
diff --git a/faststream/confluent/security.py b/faststream/confluent/security.py
index 4e3e0c0e..6324b4c4 100644
--- a/faststream/confluent/security.py
+++ b/faststream/confluent/security.py
@@ -48,39 +48,49 @@ def _parse_base_security(security: BaseSecurity) -> "AnyDict":
 def _parse_sasl_plaintext(security: SASLPlaintext) -> "AnyDict":
     return {
         "security_protocol": "SASL_SSL" if security.use_ssl else "SASL_PLAINTEXT",
-        "sasl_mechanism": "PLAIN",
-        "sasl_plain_username": security.username,
-        "sasl_plain_password": security.password,
+        "security_config": {
+            "sasl_mechanism": "PLAIN",
+            "sasl_plain_username": security.username,
+            "sasl_plain_password": security.password,
+        },
     }
 
 
 def _parse_sasl_scram256(security: SASLScram256) -> "AnyDict":
     return {
         "security_protocol": "SASL_SSL" if security.use_ssl else "SASL_PLAINTEXT",
-        "sasl_mechanism": "SCRAM-SHA-256",
-        "sasl_plain_username": security.username,
-        "sasl_plain_password": security.password,
+        "security_config": {
+            "sasl_mechanism": "SCRAM-SHA-256",
+            "sasl_plain_username": security.username,
+            "sasl_plain_password": security.password,
+        },
     }
 
 
 def _parse_sasl_scram512(security: SASLScram512) -> "AnyDict":
     return {
         "security_protocol": "SASL_SSL" if security.use_ssl else "SASL_PLAINTEXT",
-        "sasl_mechanism": "SCRAM-SHA-512",
-        "sasl_plain_username": security.username,
-        "sasl_plain_password": security.password,
+        "security_config": {
+            "sasl_mechanism": "SCRAM-SHA-512",
+            "sasl_plain_username": security.username,
+            "sasl_plain_password": security.password,
+        },
     }
 
 
 def _parse_sasl_oauthbearer(security: SASLOAuthBearer) -> "AnyDict":
     return {
         "security_protocol": "SASL_SSL" if security.use_ssl else "SASL_PLAINTEXT",
-        "sasl_mechanism": "OAUTHBEARER",
+        "security_config": {
+            "sasl_mechanism": "OAUTHBEARER",
+        },
     }
 
 
 def _parse_sasl_gssapi(security: SASLGSSAPI) -> "AnyDict":
     return {
         "security_protocol": "SASL_SSL" if security.use_ssl else "SASL_PLAINTEXT",
-        "sasl_mechanism": "GSSAPI",
+        "security_config": {
+            "sasl_mechanism": "GSSAPI",
+        },
     }


Test patch:
diff --git a/tests/a_docs/confluent/test_security.py b/tests/a_docs/confluent/test_security.py
index b77c91eb..b0ba4c86 100644
--- a/tests/a_docs/confluent/test_security.py
+++ b/tests/a_docs/confluent/test_security.py
@@ -28,11 +28,14 @@ async def test_scram256():
         async with scram256_broker:
             producer_call_kwargs = producer.call_args.kwargs
 
-            call_kwargs = {}
-            call_kwargs["sasl_mechanism"] = "SCRAM-SHA-256"
-            call_kwargs["sasl_plain_username"] = "admin"
-            call_kwargs["sasl_plain_password"] = "password"  # pragma: allowlist secret
-            call_kwargs["security_protocol"] = "SASL_SSL"
+            call_kwargs = {
+                "security_config": {
+                    "sasl_mechanism": "SCRAM-SHA-256",
+                    "sasl_plain_username": "admin",
+                    "sasl_plain_password": "password",  # pragma: allowlist secret
+                },
+                "security_protocol": "SASL_SSL",
+            }
 
             assert call_kwargs.items() <= producer_call_kwargs.items()
 
@@ -53,11 +56,14 @@ async def test_scram512():
         async with scram512_broker:
             producer_call_kwargs = producer.call_args.kwargs
 
-            call_kwargs = {}
-            call_kwargs["sasl_mechanism"] = "SCRAM-SHA-512"
-            call_kwargs["sasl_plain_username"] = "admin"
-            call_kwargs["sasl_plain_password"] = "password"  # pragma: allowlist secret
-            call_kwargs["security_protocol"] = "SASL_SSL"
+            call_kwargs = {
+                "security_config": {
+                    "sasl_mechanism": "SCRAM-SHA-512",
+                    "sasl_plain_username": "admin",
+                    "sasl_plain_password": "password",  # pragma: allowlist secret
+                },
+                "security_protocol": "SASL_SSL",
+            }
 
             assert call_kwargs.items() <= producer_call_kwargs.items()
 
@@ -78,11 +84,14 @@ async def test_plaintext():
         async with plaintext_broker:
             producer_call_kwargs = producer.call_args.kwargs
 
-            call_kwargs = {}
-            call_kwargs["sasl_mechanism"] = "PLAIN"
-            call_kwargs["sasl_plain_username"] = "admin"
-            call_kwargs["sasl_plain_password"] = "password"  # pragma: allowlist secret
-            call_kwargs["security_protocol"] = "SASL_SSL"
+            call_kwargs = {
+                "security_config": {
+                    "sasl_mechanism": "PLAIN",
+                    "sasl_plain_username": "admin",
+                    "sasl_plain_password": "password",  # pragma: allowlist secret
+                },
+                "security_protocol": "SASL_SSL",
+            }
 
             assert call_kwargs.items() <= producer_call_kwargs.items()
 
@@ -103,9 +112,12 @@ async def test_oathbearer():
         async with oauthbearer_broker:
             producer_call_kwargs = producer.call_args.kwargs
 
-            call_kwargs = {}
-            call_kwargs["sasl_mechanism"] = "OAUTHBEARER"
-            call_kwargs["security_protocol"] = "SASL_SSL"
+            call_kwargs = {
+                "security_config": {
+                    "sasl_mechanism": "OAUTHBEARER",
+                },
+                "security_protocol": "SASL_SSL",
+            }
 
             assert call_kwargs.items() <= producer_call_kwargs.items()
 
@@ -127,7 +139,9 @@ async def test_gssapi():
             producer_call_kwargs = producer.call_args.kwargs
 
             call_kwargs = {
-                "sasl_mechanism": "GSSAPI",
+                "security_config": {
+                    "sasl_mechanism": "GSSAPI",
+                },
                 "security_protocol": "SASL_SSL",
             }