Feature: GSSAPI (Kerberos) support

docs/docs/SUMMARY.md

@@ -953,6 +953,7 @@ search:
                     - [build_message](api/faststream/redis/testing/build_message.md)
             - security
                 - [BaseSecurity](api/faststream/security/BaseSecurity.md)
+                - [SASLGSSAPI](api/faststream/security/SASLGSSAPI.md)
                 - [SASLPlaintext](api/faststream/security/SASLPlaintext.md)
                 - [SASLScram256](api/faststream/security/SASLScram256.md)
                 - [SASLScram512](api/faststream/security/SASLScram512.md)

docs/docs/en/api/faststream/security/SASLGSSAPI.md

@@ -0,0 +1,11 @@
+---
+# 0.5 - API
+# 2 - Release
+# 3 - Contributing
+# 5 - Template Page
+# 10 - Default
+search:
+  boost: 0.5
+---
+
+::: faststream.security.SASLGSSAPI

docs/docs/en/kafka/security.md

@@ -51,3 +51,13 @@ This chapter discusses the security options available in **FastStream** and how
     ```python linenums="1"
     {!> docs_src/kafka/security/sasl_scram512.py [ln:1-10.25,11-] !}
     ```
+
+### 4. SASLGSSAPI Object with SSL/TLS
+
+**Purpose:** The `SASLGSSAPI` object is used for authentication using Kerberos.
+
+**Usage:**
+
+```python linenums="1"
+{!> docs_src/kafka/security/sasl_gssapi.py [ln:1-10.25,11-] !}
+```

docs/docs_src/kafka/security/sasl_gssapi.py

@@ -0,0 +1,9 @@
+import ssl
+
+from faststream.kafka import KafkaBroker
+from faststream.security import SASLGSSAPI
+
+ssl_context = ssl.create_default_context()
+security = SASLGSSAPI(ssl_context=ssl_context)
+
+broker = KafkaBroker("localhost:9092", security=security)

faststream/kafka/security.py

@@ -1,6 +1,7 @@
 from typing import TYPE_CHECKING, Optional
 
 from faststream.security import (
+    SASLGSSAPI,
     BaseSecurity,
     SASLPlaintext,
     SASLScram256,
@@ -20,6 +21,8 @@ def parse_security(security: Optional[BaseSecurity]) -> "AnyDict":
         return _parse_sasl_scram256(security)
     elif isinstance(security, SASLScram512):
         return _parse_sasl_scram512(security)
+    elif isinstance(security, SASLGSSAPI):
+        return _parse_sasl_gssapi(security)
     elif isinstance(security, BaseSecurity):
         return _parse_base_security(security)
     else:
@@ -61,3 +64,11 @@ def _parse_sasl_scram512(security: SASLScram512) -> "AnyDict":
         "sasl_plain_username": security.username,
         "sasl_plain_password": security.password,
     }
+
+
+def _parse_sasl_gssapi(security: SASLGSSAPI) -> "AnyDict":
+    return {
+        "security_protocol": "SASL_SSL" if security.use_ssl else "SASL_PLAINTEXT",
+        "ssl_context": security.ssl_context,
+        "sasl_mechanism": "GSSAPI",
+    }

faststream/security.py

@@ -151,3 +151,34 @@ def get_requirement(self) -> List["AnyDict"]:
     def get_schema(self) -> Dict[str, Dict[str, str]]:
         """Get the security schema for SASL/SCRAM-SHA-512 authentication."""
         return {"scram512": {"type": "scramSha512"}}
+
+
+class SASLGSSAPI(BaseSecurity):
+    """Security configuration for SASL/GSSAPI authentication.
+
+    This class defines security configuration for SASL/GSSAPI authentication.
+    """
+
+    # TODO: mv to SecretStr
+    __slots__ = (
+        "use_ssl",
+        "ssl_context",
+    )
+
+    def __init__(
+        self,
+        ssl_context: Optional["SSLContext"] = None,
+        use_ssl: Optional[bool] = None,
+    ) -> None:
+        super().__init__(
+            ssl_context=ssl_context,
+            use_ssl=use_ssl,
+        )
+
+    def get_requirement(self) -> List["AnyDict"]:
+        """Get the security requirements for SASL/GSSAPI authentication."""
+        return [{"gssapi": []}]
+
+    def get_schema(self) -> Dict[str, Dict[str, str]]:
+        """Get the security schema for SASL/GSSAPI authentication."""
+        return {"gssapi": {"type": "gssapi"}}

tests/asyncapi/kafka/test_security.py

@@ -5,6 +5,7 @@
 from faststream.asyncapi.generate import get_app_schema
 from faststream.kafka import KafkaBroker
 from faststream.security import (
+    SASLGSSAPI,
     BaseSecurity,
     SASLPlaintext,
     SASLScram256,
@@ -167,3 +168,28 @@ async def test_topic(msg: str) -> str:
     }
 
     assert schema == sasl512_security_schema
+
+
+def test_gssapi_security_schema():
+    ssl_context = ssl.create_default_context()
+    security = SASLGSSAPI(
+        ssl_context=ssl_context,
+    )
+
+    broker = KafkaBroker("localhost:9092", security=security)
+    app = FastStream(broker)
+
+    @broker.publisher("test_2")
+    @broker.subscriber("test_1")
+    async def test_topic(msg: str) -> str:
+        pass
+
+    schema = get_app_schema(app).to_jsonable()
+
+    gssapi_security_schema = deepcopy(basic_schema)
+    gssapi_security_schema["servers"]["development"]["security"] = [{"gssapi": []}]
+    gssapi_security_schema["components"]["securitySchemes"] = {
+        "gssapi": {"type": "gssapi"}
+    }
+
+    assert schema == gssapi_security_schema

tests/docs/kafka/test_security.py

@@ -98,3 +98,24 @@ async def test_plaintext():
             assert call_kwargs.items() <= producer_call_kwargs.items()
 
             assert type(producer_call_kwargs["ssl_context"]) is ssl.SSLContext
+
+
+@pytest.mark.kafka()
+@pytest.mark.asyncio()
+async def test_gssapi():
+    from docs.docs_src.kafka.security.sasl_gssapi import (
+        broker as gssapi_broker,
+    )
+
+    with patch_aio_consumer_and_producer() as producer:
+        async with gssapi_broker:
+            producer_call_kwargs = producer.call_args.kwargs
+
+            call_kwargs = {
+                "sasl_mechanism": "GSSAPI",
+                "security_protocol": "SASL_SSL",
+            }
+
+            assert call_kwargs.items() <= producer_call_kwargs.items()
+
+            assert type(producer_call_kwargs["ssl_context"]) is ssl.SSLContext