authentik.sources.kerberos.auth

  1"""authentik Kerberos Authentication Backend"""
  2
  3import gssapi
  4from django.http import HttpRequest
  5from structlog.stdlib import get_logger
  6
  7from authentik.core.auth import InbuiltBackend
  8from authentik.core.models import User
  9from authentik.lib.generators import generate_id
 10from authentik.sources.kerberos.models import (
 11    KerberosSource,
 12    Krb5ConfContext,
 13    UserKerberosSourceConnection,
 14)
 15
 16LOGGER = get_logger()
 17
 18
 19class KerberosBackend(InbuiltBackend):
 20    """Authenticate users against Kerberos realm"""
 21
 22    def authenticate(self, request: HttpRequest, **kwargs):
 23        """Try to authenticate a user via kerberos"""
 24        if "password" not in kwargs or "username" not in kwargs:
 25            return None
 26        username = kwargs.pop("username")
 27        realm = None
 28        if "@" in username:
 29            username, realm = username.rsplit("@", 1)
 30
 31        user, source = self.auth_user(request, username, realm, **kwargs)
 32        if user:
 33            self.set_method("kerberos", request, source=source)
 34            return user
 35        return None
 36
 37    def auth_user(
 38        self, request: HttpRequest, username: str, realm: str | None, password: str, **filters
 39    ) -> tuple[User | None, KerberosSource | None]:
 40        sources = KerberosSource.objects.filter(enabled=True)
 41        user = User.objects.filter(
 42            usersourceconnection__source__in=sources, username=username, **filters
 43        ).first()
 44
 45        if user is not None:
 46            # User found, let's get its connections for the sources that are available
 47            user_source_connections = UserKerberosSourceConnection.objects.filter(
 48                user=user, source__in=sources
 49            )
 50        elif realm is not None:
 51            user_source_connections = UserKerberosSourceConnection.objects.filter(
 52                source__in=sources, identifier=f"{username}@{realm}"
 53            )
 54        # no realm specified, we can't do anything
 55        else:
 56            user_source_connections = UserKerberosSourceConnection.objects.none()
 57
 58        if not user_source_connections.exists():
 59            LOGGER.debug("no kerberos source found for user", username=username)
 60            return None, None
 61
 62        for user_source_connection in user_source_connections.prefetch_related().select_related(
 63            "source__kerberossource"
 64        ):
 65            # User either has an unusable password,
 66            # or has a password, but couldn't be authenticated by ModelBackend
 67            # This means we check with a kinit to see if the Kerberos password has changed
 68            if self.auth_user_by_kinit(user_source_connection, password):
 69                # Password was successful in kinit to Kerberos, so we save it in database
 70                if (
 71                    user_source_connection.source.kerberossource.password_login_update_internal_password
 72                ):
 73                    LOGGER.debug(
 74                        "Updating user's password in DB",
 75                        source=user_source_connection.source,
 76                        user=user_source_connection.user,
 77                    )
 78                    user_source_connection.user.set_password(
 79                        password, sender=user_source_connection.source, request=request
 80                    )
 81                    user_source_connection.user.save()
 82                return user_source_connection.user, user_source_connection.source
 83            # Password doesn't match, onto next source
 84            LOGGER.debug(
 85                "failed to kinit, password invalid",
 86                source=user_source_connection.source,
 87                user=user_source_connection.user,
 88            )
 89        # No source with valid password found
 90        LOGGER.debug("no valid kerberos source found for user", user=user)
 91        return None, None
 92
 93    def auth_user_by_kinit(
 94        self, user_source_connection: UserKerberosSourceConnection, password: str
 95    ) -> bool:
 96        """Attempt authentication by kinit to the source."""
 97        LOGGER.debug(
 98            "Attempting to kinit as user",
 99            user=user_source_connection.user,
100            source=user_source_connection.source,
101            principal=user_source_connection.identifier,
102        )
103
104        with Krb5ConfContext(user_source_connection.source.kerberossource):
105            name = gssapi.raw.import_name(
106                user_source_connection.identifier.encode(), gssapi.raw.NameType.kerberos_principal
107            )
108            try:
109                # Use a temporary credentials cache to not interfere with whatever is defined
110                # elsewhere
111                gssapi.raw.ext_krb5.krb5_ccache_name(f"MEMORY:{generate_id(12)}".encode())
112                gssapi.raw.ext_password.acquire_cred_with_password(name, password.encode())
113                # Restore the credentials cache to what it was before
114                gssapi.raw.ext_krb5.krb5_ccache_name(None)
115                return True
116            except gssapi.exceptions.GSSError as exc:
117                LOGGER.warning("failed to kinit", exc=exc)
118        return False