Repository: airtai/faststream (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.



Pull request:
fix: change oauth type in asyncapi schema
# Description

Added changes to the generation of AsyncAPI documentation in the OAuthBearer part. According to the AsyncAPI specification, the OAuthBearer is mapped to oauth2. Also adds a “$ref” field for mapping and generating documentation. It should be clarified in the future whether requiring this field is a bug or whether additional edits are needed.

Thank you to the following developers for their help:
1) @Lancetnik for help with the bug and support during the fix;
2) @KrySeyt for consultations on AsyncAPI in FastStream;
3) @three-question-marks for spending time together while fixing this problem and studying AsyncAPI documentation at the same time;
4) @dalelane for help in mapping Kafka Security Config to AsyncAPI types

Fixes #1631

## Type of change

Please delete options that are not relevant.

- [ ] Documentation (typos, code examples, or any documentation updates)
- [Х] 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/faststream/security.py b/faststream/security.py
index 4bcd4eba..352acffb 100644
--- a/faststream/security.py
+++ b/faststream/security.py
@@ -167,7 +167,7 @@ class SASLOAuthBearer(BaseSecurity):
 
     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 0621cbf7..915621ab 100644
--- a/tests/asyncapi/confluent/test_security.py
+++ b/tests/asyncapi/confluent/test_security.py
@@ -192,7 +192,7 @@ def test_oauthbearer_security_schema():
         {"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 3201bf89..62e30e9c 100644
--- a/tests/asyncapi/kafka/test_security.py
+++ b/tests/asyncapi/kafka/test_security.py
@@ -192,7 +192,7 @@ def test_oauthbearer_security_schema():
         {"oauthbearer": []}
     ]
     sasl_oauthbearer_security_schema["components"]["securitySchemes"] = {
-        "oauthbearer": {"type": "oauthBearer"}
+        "oauthbearer": {"type": "oauth2", "$ref": ""}
     }
 
     assert schema == sasl_oauthbearer_security_schema