Repository: airtai/faststream (language: python)

Issue:
Feature: add sasl_oauthbearer support
Hi!

Faststream doesn't allow to authenticate with Oauth. According to [librdkafka/CONFIGURATION.md](https://github.com/confluentinc/librdkafka/blob/master/CONFIGURATION.md) `sasl.mechanism` must be set to `OAUTHBEARER`. However this value will get overwritten when specifying it in config and providing it during broker creation. To be precise it will get overwritten by one of `_parse_*` functions in `faststream\confluent\security.py` depending on which security class from `faststream\security.py` will be chosen. Changes should allow broker configuration with oauth:

```
broker = KafkaBroker(
    "host:port",
    config=config,
    security=SASLOAuthBearer(use_ssl=True),
)
```

To achieve this we need to implement a new security class and new parser. I have already created fork with these changes and successfully deployed it.




Discussion / hints:
PR resolving the issue https://github.com/airtai/faststream/pull/1632

Fix patch:
diff --git a/faststream/security.py b/faststream/security.py
index 4bcd4eba99..352acffb22 100644
--- a/faststream/security.py
+++ b/faststream/security.py
@@ -167,7 +167,7 @@ def get_requirement(self) -> List["AnyDict"]:
 
     def get_schema(self) -> Dict[str, Dict[str, str]]:
         """Get the security schema for SASL/OAUTHBEARER authentication."""
-        return {"oauthbearer": {"type": "oauthBearer"}}
+        return {"oauthbearer": {"type": "oauth2", "$ref": ""}}
 
 
 class SASLGSSAPI(BaseSecurity):


Test patch:
diff --git a/tests/asyncapi/confluent/test_security.py b/tests/asyncapi/confluent/test_security.py
index 0621cbf756..915621ab84 100644
--- a/tests/asyncapi/confluent/test_security.py
+++ b/tests/asyncapi/confluent/test_security.py
@@ -192,7 +192,7 @@ async def test_topic(msg: str) -> str:
         {"oauthbearer": []}
     ]
     sasl_oauthbearer_security_schema["components"]["securitySchemes"] = {
-        "oauthbearer": {"type": "oauthBearer"}
+        "oauthbearer": {"type": "oauth2", "$ref": ""}
     }
 
     assert schema == sasl_oauthbearer_security_schema
diff --git a/tests/asyncapi/kafka/test_security.py b/tests/asyncapi/kafka/test_security.py
index 3201bf899c..62e30e9ccf 100644
--- a/tests/asyncapi/kafka/test_security.py
+++ b/tests/asyncapi/kafka/test_security.py
@@ -192,7 +192,7 @@ async def test_topic(msg: str) -> str:
         {"oauthbearer": []}
     ]
     sasl_oauthbearer_security_schema["components"]["securitySchemes"] = {
-        "oauthbearer": {"type": "oauthBearer"}
+        "oauthbearer": {"type": "oauth2", "$ref": ""}
     }
 
     assert schema == sasl_oauthbearer_security_schema