Feature: GSSAPI (Kerberos) support

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/brokers/kafka/test_security.py

@@ -0,0 +1,38 @@
+from contextlib import contextmanager
+from typing import Tuple
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+
+@contextmanager
+def _patch_aio_producer() -> Tuple[MagicMock, MagicMock]:
+    try:
+        producer = MagicMock(return_value=AsyncMock())
+
+        with patch(
+            "aiokafka.AIOKafkaProducer",
+            new=producer,
+        ):
+            yield producer
+    finally:
+        pass
+
+
+@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_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()