authentik.core.sources.matcher

  1"""Source user and group matching"""
  2
  3from dataclasses import dataclass
  4from enum import Enum
  5from typing import Any
  6
  7from django.db.models import Q
  8from structlog import get_logger
  9
 10from authentik.core.models import (
 11    Group,
 12    GroupSourceConnection,
 13    Source,
 14    SourceGroupMatchingModes,
 15    SourceUserMatchingModes,
 16    User,
 17    UserSourceConnection,
 18)
 19
 20
 21class Action(Enum):
 22    """Actions that can be decided based on the request and source settings"""
 23
 24    LINK = "link"
 25    AUTH = "auth"
 26    ENROLL = "enroll"
 27    DENY = "deny"
 28
 29
 30@dataclass
 31class MatchableProperty:
 32    property: str
 33    link_mode: SourceUserMatchingModes | SourceGroupMatchingModes
 34    deny_mode: SourceUserMatchingModes | SourceGroupMatchingModes
 35
 36
 37class SourceMatcher:
 38    def __init__(
 39        self,
 40        source: Source,
 41        user_connection_type: type[UserSourceConnection],
 42        group_connection_type: type[GroupSourceConnection],
 43    ):
 44        self.source = source
 45        self.user_connection_type = user_connection_type
 46        self.group_connection_type = group_connection_type
 47        self._logger = get_logger().bind(source=self.source)
 48
 49    def get_action(
 50        self,
 51        object_type: type[User | Group],
 52        matchable_properties: list[MatchableProperty],
 53        identifier: str,
 54        properties: dict[str, Any | dict[str, Any]],
 55    ) -> tuple[Action, UserSourceConnection | GroupSourceConnection | None]:
 56        connection_type = None
 57        matching_mode = None
 58        identifier_matching_mode = None
 59        if object_type == User:
 60            connection_type = self.user_connection_type
 61            matching_mode = self.source.user_matching_mode
 62            identifier_matching_mode = SourceUserMatchingModes.IDENTIFIER
 63        if object_type == Group:
 64            connection_type = self.group_connection_type
 65            matching_mode = self.source.group_matching_mode
 66            identifier_matching_mode = SourceGroupMatchingModes.IDENTIFIER
 67        if not connection_type or not matching_mode or not identifier_matching_mode:
 68            return Action.DENY, None
 69
 70        new_connection = connection_type(source=self.source, identifier=identifier)
 71
 72        existing_connections = connection_type.objects.filter(
 73            source=self.source, identifier=identifier
 74        )
 75        if existing_connections.exists():
 76            return Action.AUTH, existing_connections.first()
 77        # No connection exists, but we match on identifier, so enroll
 78        if matching_mode == identifier_matching_mode:
 79            # We don't save the connection here cause it doesn't have a user/group assigned yet
 80            return Action.ENROLL, new_connection
 81
 82        # Check for existing users with matching attributes
 83        query = Q()
 84        for matchable_property in matchable_properties:
 85            property = matchable_property.property
 86            if matching_mode in [matchable_property.link_mode, matchable_property.deny_mode]:
 87                if not properties.get(property, None):
 88                    self._logger.warning(
 89                        "Refusing to use none property", identifier=identifier, property=property
 90                    )
 91                    return Action.DENY, None
 92                query_args = {
 93                    f"{property}__exact": properties[property],
 94                }
 95                query = Q(**query_args)
 96        self._logger.debug(
 97            "Trying to link with existing object", query=query, identifier=identifier
 98        )
 99        matching_objects = object_type.objects.filter(query)
100        # Not matching objects, always enroll
101        if not matching_objects.exists():
102            self._logger.debug("No matching objects found, enrolling")
103            return Action.ENROLL, new_connection
104
105        obj = matching_objects.first()
106        if matching_mode in [mp.link_mode for mp in matchable_properties]:
107            attr = None
108            if object_type == User:
109                attr = "user"
110            if object_type == Group:
111                attr = "group"
112            setattr(new_connection, attr, obj)
113            return Action.LINK, new_connection
114        if matching_mode in [mp.deny_mode for mp in matchable_properties]:
115            self._logger.info("Denying source because object exists", obj=obj)
116            return Action.DENY, None
117
118        # Should never get here as default enroll case is returned above.
119        return Action.DENY, None  # pragma: no cover
120
121    def get_user_action(
122        self, identifier: str, properties: dict[str, Any | dict[str, Any]]
123    ) -> tuple[Action, UserSourceConnection | None]:
124        return self.get_action(
125            User,
126            [
127                MatchableProperty(
128                    "username",
129                    SourceUserMatchingModes.USERNAME_LINK,
130                    SourceUserMatchingModes.USERNAME_DENY,
131                ),
132                MatchableProperty(
133                    "email", SourceUserMatchingModes.EMAIL_LINK, SourceUserMatchingModes.EMAIL_DENY
134                ),
135            ],
136            identifier,
137            properties,
138        )
139
140    def get_group_action(
141        self, identifier: str, properties: dict[str, Any | dict[str, Any]]
142    ) -> tuple[Action, GroupSourceConnection | None]:
143        return self.get_action(
144            Group,
145            [
146                MatchableProperty(
147                    "name", SourceGroupMatchingModes.NAME_LINK, SourceGroupMatchingModes.NAME_DENY
148                ),
149            ],
150            identifier,
151            properties,
152        )