Add SPNEGO support 664

app/ldap_protocol/ldap_requests/bind.py

@@ -25,6 +25,9 @@
     get_bad_response,
     sasl_mechanism_map,
 )
+from ldap_protocol.ldap_requests.bind_methods.sasl_spnego import (
+    SaslSPNEGOAuthentication,
+)
 from ldap_protocol.ldap_responses import BaseResponse, BindResponse
 from ldap_protocol.multifactor import MultifactorAPI
 from ldap_protocol.objects import ProtocolRequests
@@ -235,7 +238,14 @@ async def handle(
         await ctx.ldap_session.set_user(user)
         await set_user_logon_attrs(user, ctx.session, ctx.settings.TIMEZONE)
 
-        yield BindResponse(result_code=LDAPCodes.SUCCESS)
+        server_sasl_creds = None
+        if isinstance(self.authentication_choice, SaslSPNEGOAuthentication):
+            server_sasl_creds = self.authentication_choice.server_sasl_creds
+
+        yield BindResponse(
+            result_code=LDAPCodes.SUCCESS,
+            server_sasl_creds=server_sasl_creds,
+        )
 
 
 class UnbindRequest(BaseRequest):

app/ldap_protocol/ldap_requests/bind_methods/__init__.py

@@ -13,11 +13,13 @@
 )
 from .sasl_gssapi import GSSAPISL, GSSAPIAuthStatus, SaslGSSAPIAuthentication
 from .sasl_plain import SaslPLAINAuthentication
+from .sasl_spnego import SaslSPNEGOAuthentication
 from .simple import SimpleAuthentication
 
 sasl_mechanism: list[type[SaslAuthentication]] = [
     SaslPLAINAuthentication,
     SaslGSSAPIAuthentication,
+    SaslSPNEGOAuthentication,
 ]
 
 sasl_mechanism_map: dict[SASLMethod, type[SaslAuthentication]] = {
@@ -31,6 +33,7 @@
     "SASLMethod",
     "SaslAuthentication",
     "SaslGSSAPIAuthentication",
+    "SaslSPNEGOAuthentication",
     "SaslPLAINAuthentication",
     "SimpleAuthentication",
     "GSSAPIAuthStatus",

app/ldap_protocol/ldap_requests/bind_methods/base.py

@@ -23,6 +23,7 @@ class SASLMethod(StrEnum):
     PLAIN = "PLAIN"
     EXTERNAL = "EXTERNAL"
     GSSAPI = "GSSAPI"
+    GSS_SPNEGO = "GSS-SPNEGO"
     CRAM_MD5 = "CRAM-MD5"
     DIGEST_MD5 = "DIGEST-MD5"
     SCRAM_SHA_1 = "SCRAM-SHA-1"

app/ldap_protocol/ldap_requests/bind_methods/sasl_gssapi.py

@@ -127,7 +127,6 @@ async def _init_security_context(
             name=server_name,
             usage="accept",
             store={"keytab": settings.KRB5_LDAP_KEYTAB},
-            mechs=[gssapi.MechType.kerberos],
         )
 
         self._ldap_session.gssapi_security_context = gssapi.SecurityContext(

app/ldap_protocol/ldap_requests/bind_methods/sasl_spnego.py

@@ -0,0 +1,81 @@
+"""Sasl SPNEGO auth method.
+
+Copyright (c) 2024 MultiFactor
+License: https://github.com/MultiDirectoryLab/MultiDirectory/blob/main/LICENSE
+"""
+
+from typing import ClassVar
+
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from config import Settings
+from ldap_protocol.dialogue import LDAPSession
+from ldap_protocol.ldap_codes import LDAPCodes
+from ldap_protocol.ldap_responses import BindResponse
+
+from .base import LDAPBindErrors, SASLMethod, get_bad_response
+from .sasl_gssapi import GSSAPISL, GSSAPIAuthStatus, SaslGSSAPIAuthentication
+
+
+class SaslSPNEGOAuthentication(SaslGSSAPIAuthentication):
+    """Sasl SPNEGO auth method.
+
+    Implements SPNEGO (RFC 4178) as a negotiation wrapper around GSS-API.
+    In practice the negotiated mechanism is Kerberos.
+
+    Flow:
+
+    1. Negotiation & Context Initialization:
+        - The server acquires acceptor credentials from keytab using the
+            ldap/{REALM} service principal.
+        - Creates a GSS-API acceptor security context with the SPNEGO
+            mechanism (which in turn negotiates Kerberos).
+        - Stores the context in the LDAP session.
+
+    2. Intermediate Request:
+        - The client and server may exchange several SPNEGO tokens
+            (NegTokenResp with responseToken/mechListMIC) until the wrapped
+            GSS (Kerberos) context becomes established.
+        - When `server_ctx.complete` becomes true, the initiator principal
+            is available via `ctx.initiator_name` and server sends NegTokenResp
+            with success.
+
+    """
+
+    mechanism: ClassVar[SASLMethod] = SASLMethod.GSS_SPNEGO
+
+    async def step(
+        self,
+        session: AsyncSession,
+        ldap_session: LDAPSession,
+        settings: Settings,
+    ) -> BindResponse | None:
+        """SPNEGO step.
+
+        :param AsyncSession session: db session
+        :param LDAPSession ldap_session: ldap session
+        :param Settings settings: settings
+        """
+        self._ldap_session = ldap_session
+
+        if not self._ldap_session.gssapi_security_context:
+            await self._init_security_context(session, settings)
+
+        server_ctx = self._ldap_session.gssapi_security_context
+        if server_ctx is None or self.ticket == b"":
+            return get_bad_response(LDAPBindErrors.LOGON_FAILURE)
+
+        status = self._handle_ticket(server_ctx)
+
+        if not server_ctx.complete:
+            return BindResponse(
+                result_code=LDAPCodes.SASL_BIND_IN_PROGRESS,
+                server_sasl_creds=self.server_sasl_creds,
+            )
+
+        if status == GSSAPIAuthStatus.SEND_TO_CLIENT:
+            self._ldap_session.gssapi_authenticated = True
+            self._ldap_session.gssapi_security_layer = GSSAPISL.CONFIDENTIALITY
+            return None
+
+        return get_bad_response(LDAPBindErrors.LOGON_FAILURE)

app/ldap_protocol/ldap_requests/search.py

@@ -217,7 +217,12 @@ async def get_root_dse(
         data["currentTime"].append(get_generalized_now(settings.TIMEZONE))
         data["subschemaSubentry"].append(schema)
         data["schemaNamingContext"].append(schema)
-        data["supportedSASLMechanisms"] = ["ANONYMOUS", "PLAIN", "GSSAPI"]
+        data["supportedSASLMechanisms"] = [
+            "ANONYMOUS",
+            "PLAIN",
+            "GSSAPI",
+            "GSS-SPNEGO",
+        ]
         data["highestCommittedUSN"].append("126991")
         data["supportedExtension"] = [
             "1.3.6.1.4.1.4203.1.11.3",  # whoami

tests/test_ldap/test_bind.py

@@ -23,6 +23,9 @@
     SimpleAuthentication,
     UnbindRequest,
 )
+from ldap_protocol.ldap_requests.bind_methods.sasl_spnego import (
+    SaslSPNEGOAuthentication,
+)
 from ldap_protocol.ldap_requests.contexts import (
     LDAPBindRequestContext,
     LDAPUnbindRequestContext,
@@ -199,6 +202,80 @@ async def mock_init_security_context(
         )
 
 
+@pytest.mark.asyncio
+@pytest.mark.usefixtures("session")
+@pytest.mark.usefixtures("setup_session")
+async def test_spnego_bind_ok(
+    creds: TestCreds,
+    container: AsyncContainer,
+) -> None:
+    """Test spnego bind ok."""
+    mock_security_context = Mock(spec=gssapi.SecurityContext)
+    mock_security_context.step.return_value = b"server_ticket"
+    mock_security_context.complete = False
+    mock_security_context.initiator_name = f"{creds.un}@domain"
+
+    async def mock_init_security_context(
+        session: AsyncSession,  # noqa: ARG001
+        settings: Settings,  # noqa: ARG001
+    ) -> None:
+        auth_choice._ldap_session.gssapi_security_context = (
+            mock_security_context
+        )
+
+    auth_choice = SaslSPNEGOAuthentication(ticket=b"client_ticket")
+    auth_choice._init_security_context = mock_init_security_context  # type: ignore
+
+    first_bind = BindRequest(
+        version=0,
+        name=creds.un,
+        AuthenticationChoice=auth_choice,
+    )
+
+    second_bind = MutePolicyBindRequest(
+        version=0,
+        name=creds.un,
+        AuthenticationChoice=SaslSPNEGOAuthentication(ticket=b"client_ticket"),
+    )
+
+    async with container(scope=Scope.REQUEST) as container:
+        kwargs = await resolve_deps(first_bind.handle, container)
+        result = await anext(first_bind.handle(**kwargs))
+        assert result == BindResponse(
+            result_code=LDAPCodes.SASL_BIND_IN_PROGRESS,
+            serverSaslCreds=b"server_ticket",
+        )
+
+        mock_security_context.complete = True
+
+        kwargs = await resolve_deps(second_bind.handle, container)
+        result = await anext(second_bind.handle(**kwargs))
+        assert result == BindResponse(
+            result_code=LDAPCodes.SUCCESS,
+            serverSaslCreds=b"server_ticket",
+        )
+
+
+@pytest.mark.asyncio
+@pytest.mark.usefixtures("session")
+@pytest.mark.usefixtures("setup_session")
+async def test_spnego_bind_missing_credentials(
+    creds: TestCreds,
+    container: AsyncContainer,
+) -> None:
+    """Test spnego bind with missing credentials."""
+    bind = BindRequest(
+        version=0,
+        name=creds.un,
+        AuthenticationChoice=SaslSPNEGOAuthentication(),
+    )
+
+    async with container(scope=Scope.REQUEST) as container:
+        kwargs = await resolve_deps(bind.handle, container)
+        with pytest.raises(gssapi.exceptions.MissingCredentialsError):
+            await anext(bind.handle(**kwargs))
+
+
 @pytest.mark.asyncio
 @pytest.mark.usefixtures("session")
 async def test_bind_invalid_password_or_user(