zulip/zerver/lib/scim.py

from typing import Any, Callable, Dict, List, Optional, Type, Union

import django_scim.constants as scim_constants
import django_scim.exceptions as scim_exceptions
import orjson
from django.conf import settings
from django.contrib.auth.decorators import login_required
from django.core.exceptions import ValidationError
from django.db import models, transaction
from django.http import HttpRequest, HttpResponse
from django.utils.decorators import method_decorator
from django.views.decorators.csrf import csrf_exempt
from django_scim.adapters import SCIMUser
from django_scim.views import SCIMView, SearchView, UserSearchView, UsersView
from django_scim.views import logger as scim_views_logger
from scim2_filter_parser.attr_paths import AttrPath

from zerver.lib.actions import (
    check_change_full_name,
    do_change_user_delivery_email,
    do_create_user,
    do_deactivate_user,
    do_reactivate_user,
)
from zerver.lib.email_validation import email_allowed_for_realm, validate_email_not_already_in_realm
from zerver.lib.request import RequestNotes
from zerver.lib.subdomains import get_subdomain
from zerver.models import (
    DisposableEmailError,
    DomainNotAllowedForRealmError,
    EmailContainsPlusError,
    UserProfile,
)


class ZulipSCIMUser(SCIMUser):
    """With django-scim2, the core of a project's SCIM implementation is
    this user adapter class, which defines how to translate between the
    concepts of users in the SCIM specification and the Zulip users.
    """

    id_field = "id"

    def __init__(self, obj: UserProfile, request: Optional[HttpRequest] = None) -> None:
        # We keep the function signature from the superclass, but this actually
        # shouldn't be called with request being None.
        assert request is not None

        # self.obj is populated appropriately by django-scim2 views with
        # an instance of UserProfile - either fetched from the database
        # or constructed via UserProfile() if the request currently being
        # handled is a User creation request (POST).
        self.obj: UserProfile

        super().__init__(obj, request)
        self.subdomain = get_subdomain(request)
        self.config = settings.SCIM_CONFIG[self.subdomain]

        # These attributes are custom to this class and will be
        # populated with values in handle_replace and similar methods
        # in response to a request for the the corresponding
        # UserProfile fields to change. The .save() method inspects
        # these fields an executes the requested changes.
        self._email_new_value: Optional[str] = None
        self._is_active_new_value: Optional[bool] = None
        self._full_name_new_value: Optional[str] = None
        self._password_set_to: Optional[str] = None

    def is_new_user(self) -> bool:
        return not bool(self.obj.id)

    @property
    def display_name(self) -> str:
        """
        Return the displayName of the user per the SCIM spec.

        Overridden because UserProfile uses the .full_name attribute,
        while the superclass expects .first_name and .last_name.
        """
        return self.obj.full_name

    def to_dict(self) -> Dict[str, Any]:
        """
        Return a ``dict`` conforming to the SCIM User Schema,
        ready for conversion to a JSON object.

        The attribute names appearing in the dict are those defined in the SCIM User Schema:
        https://datatracker.ietf.org/doc/html/rfc7643#section-4.1
        """
        if self.config["name_formatted_included"]:
            name = {
                "formatted": self.obj.full_name,
            }
        else:
            # Some clients (e.g. Okta) operate with a first_name,
            # last_name model and don't support a full name field.
            # While we strive never to do this in the project because
            # not every culture has the first/last name structure,
            # Okta's design means we have to convert our full_name
            # into a first_name/last_name pair to provide to the
            # client.  We do naive conversion with `split`.
            if " " in self.obj.full_name:
                first_name, last_name = self.obj.full_name.split(" ", 1)
            else:
                first_name, last_name = self.obj.full_name, ""
            name = {
                "givenName": first_name,
                "familyName": last_name,
            }
        d = dict(
            {
                "schemas": [scim_constants.SchemaURI.USER],
                "id": self.obj.id,
                "userName": self.obj.delivery_email,
                "name": name,
                "displayName": self.display_name,
                "active": self.obj.is_active,
                # meta is a property implemented in the superclass
                # TODO: The upstream implementation uses `user_profile.date_joined`
                # as the value of the lastModified meta attribute, which is not
                # a correct simplification. We should add proper tracking
                # of this value.
                "meta": self.meta,
            }
        )

        return d

    def from_dict(self, d: Dict[str, Any]) -> None:
        """Consume a dictionary conforming to the SCIM User Schema. The
        dictionary was originally submitted as JSON by the client in
        PUT (update a user) and POST (create a new user) requests.  A
        PUT request tells us to update User attributes to match those
        passed in the dict.  A POST request tells us to create a new
        User with attributes as specified in the dict.

        The superclass implements some very basic default behavior,
        that doesn't support changing attributes via our actions.py
        functions (which update audit logs, send events, etc.) or
        doing application-specific validation.

        Thus, we've completely overridden the upstream implementation
        to store the values of the supported attributes that the
        request would like to change. Actually modifying the database
        is implemented in self.save().

        Given that SCIMUser is an adapter class, this method is meant
        to be completely overridden, and we can expect it remain the
        case that no important django-scim2 logic relies on the
        superclass's implementation of this function.
        """
        email = d.get("userName")
        assert isinstance(email, str)
        self.change_delivery_email(email)

        name_attr_dict = d.get("name", {})
        if self.config["name_formatted_included"]:
            full_name = name_attr_dict.get("formatted", "")
        else:
            # Some providers (e.g. Okta) don't provide name.formatted.
            first_name = name_attr_dict.get("givenName", "")
            last_name = name_attr_dict.get("familyName", "")
            full_name = f"{first_name} {last_name}".strip()

        if full_name:
            assert isinstance(full_name, str)
            self.change_full_name(full_name)

        if self.is_new_user() and not full_name:
            raise scim_exceptions.BadRequestError(
                "Must specify name.formatted, name.givenName or name.familyName when creating a new user"
            )

        active = d.get("active")
        if self.is_new_user() and not active:
            raise scim_exceptions.BadRequestError("New user must have active=True")

        if active is not None:
            assert isinstance(active, bool)
            self.change_is_active(active)

    def change_delivery_email(self, new_value: str) -> None:
        # Note that the email_allowed_for_realm check that usually
        # appears adjacent to validate_email is present in save().
        self.validate_email(new_value)
        if self.obj.delivery_email != new_value:
            self._email_new_value = new_value

    def change_full_name(self, new_value: str) -> None:
        if new_value and self.obj.full_name != new_value:
            self._full_name_new_value = new_value

    def change_is_active(self, new_value: bool) -> None:
        if new_value is not None and new_value != self.obj.is_active:
            self._is_active_new_value = new_value

    def handle_replace(
        self,
        path: Optional[AttrPath],
        value: Union[str, List[object], Dict[AttrPath, object]],
        operation: Any,
    ) -> None:
        """
        PATCH requests specify a list of operations of types "add", "remove", "replace".
        So far we only implement "replace" as that should be sufficient.

        This method is forked from the superclass and is called to handle "replace"
        PATCH operations. Such an operation tells us to change the values
        of a User's attributes as specified. The superclass implements a very basic
        behavior in this method and is meant to be overridden, since this is an adapter class.
        """
        if not isinstance(value, dict):
            # Restructure for use in loop below. Taken from the
            # overridden upstream method.
            assert path is not None
            value = {path: value}

        assert isinstance(value, dict)
        for path, val in (value or {}).items():
            if path.first_path == ("userName", None, None):
                assert isinstance(val, str)
                self.change_delivery_email(val)
            elif path.first_path == ("name", "formatted", None):
                # TODO: Add support name_formatted_included=False config like we do
                # for updates via PUT.
                assert isinstance(val, str)
                self.change_full_name(val)
            elif path.first_path == ("active", None, None):
                assert isinstance(val, bool)
                self.change_is_active(val)
            else:
                raise scim_exceptions.NotImplementedError("Not Implemented")

        self.save()

    def save(self) -> None:
        """
        This method is called at the end of operations modifying a user,
        and is responsible for actually applying the requested changes,
        writing them to the database.
        """
        realm = RequestNotes.get_notes(self._request).realm
        assert realm is not None

        email_new_value = getattr(self, "_email_new_value", None)
        is_active_new_value = getattr(self, "_is_active_new_value", None)
        full_name_new_value = getattr(self, "_full_name_new_value", None)
        password = getattr(self, "_password_set_to", None)

        # Clean up the internal "pending change" state, now that we've
        # fetched the values:
        self._email_new_value = None
        self._is_active_new_value = None
        self._full_name_new_value = None
        self._password_set_to = None

        if email_new_value:
            try:
                # Note that the validate_email check that usually
                # appears adjacent to email_allowed_for_realm is
                # present in save().
                email_allowed_for_realm(email_new_value, realm)
            except DomainNotAllowedForRealmError:
                raise scim_exceptions.BadRequestError(
                    "This email domain isn't allowed in this organization."
                )
            except DisposableEmailError:  # nocoverage
                raise scim_exceptions.BadRequestError(
                    "Disposable email domains are not allowed for this realm."
                )
            except EmailContainsPlusError:  # nocoverage
                raise scim_exceptions.BadRequestError("Email address can't contain + characters.")

            try:
                validate_email_not_already_in_realm(realm, email_new_value)
            except ValidationError as e:
                raise ConflictError("Email address already in use: " + str(e))

        if self.is_new_user():
            assert full_name_new_value is not None
            self.obj = do_create_user(
                email_new_value,
                password,
                realm,
                full_name_new_value,
                acting_user=None,
            )
            return

        with transaction.atomic():
            # We process full_name first here, since it's the only one that can fail.
            if full_name_new_value:
                check_change_full_name(self.obj, full_name_new_value, acting_user=None)

            if email_new_value:
                do_change_user_delivery_email(self.obj, email_new_value)

            if is_active_new_value is not None and is_active_new_value:
                do_reactivate_user(self.obj, acting_user=None)
            elif is_active_new_value is not None and not is_active_new_value:
                do_deactivate_user(self.obj, acting_user=None)

    def delete(self) -> None:
        """
        This is consistent with Okta SCIM - users don't get DELETEd, they're deactivated
        by changing their "active" attr to False.
        """
        raise scim_exceptions.BadRequestError(
            'DELETE operation not supported. Use PUT or PATCH to modify the "active" attribute instead.'
        )


def get_extra_model_filter_kwargs_getter(
    model: Type[models.Model],
) -> Callable[[HttpRequest, Any, Any], Dict[str, object]]:
    """Registered as GET_EXTRA_MODEL_FILTER_KWARGS_GETTER in our
    SCIM configuration.

    Returns a function which generates additional kwargs
    to add to QuerySet's .filter() when fetching a UserProfile
    corresponding to the requested SCIM User from the database.

    It's *crucial* for security that we filter by realm_id (based on
    the subdomain of the request) to prevent a SCIM client authorized
    for subdomain X from being able to interact with all of the Users
    on the entire server.

    This should be extended for Groups when implementing them by
    checking the `model` parameter; because we only support
    UserProfiles, such a check is unnecessary.
    """

    def get_extra_filter_kwargs(
        request: HttpRequest, *args: Any, **kwargs: Any
    ) -> Dict[str, object]:
        realm = RequestNotes.get_notes(request).realm
        assert realm is not None
        return {"realm_id": realm.id, "is_bot": False}

    return get_extra_filter_kwargs


def base_scim_location_getter(request: HttpRequest, *args: Any, **kwargs: Any) -> str:
    """Used as the base url for constructing the Location of a SCIM resource.

    Since SCIM synchronization is scoped to an individual realm, we
    need these locations to be namespaced within the realm's domain
    namespace, which is conveniently accessed via realm.uri.
    """

    realm = RequestNotes.get_notes(request).realm
    assert realm is not None

    return realm.uri


class ConflictError(scim_exceptions.IntegrityError):
    """
    Per https://datatracker.ietf.org/doc/html/rfc7644#section-3.3

    If the service provider determines that the creation of the requested
    resource conflicts with existing resources (e.g., a "User" resource
    with a duplicate "userName"), the service provider MUST return HTTP
    status code 409 (Conflict) with a "scimType" error code of
    "uniqueness"

    scim_exceptions.IntegrityError class omits to include the scimType.
    """

    scim_type = "uniqueness"


class ZulipSCIMViewMixin(SCIMView):
    """
    Default django-scim2 behavior is to convert any exception that occurs while processing
    the request within the view code to a string and put it
    in the HttpResponse. We don't want that due to the risk of leaking sensitive information
    through the error message.

    The way we implement this override is by having this mixin override the main dispatch()
    method - and then all the specific view classes are re-defined to inherit from this mixin
    and the original django-scim2 class. This means that we have to also re-register all
    the URL patterns so that our View classes are used.
    """

    @method_decorator(csrf_exempt)
    @method_decorator(login_required)
    def dispatch(self, request: HttpRequest, *args: Any, **kwargs: Any) -> HttpResponse:
        """
        This method through which all SCIM views are processed needs to be forked
        to change its logic of how exceptions are handled.
        """
        if not self.implemented:
            return self.status_501(request, *args, **kwargs)

        try:
            return super(SCIMView, self).dispatch(request, *args, **kwargs)
        except Exception as e:
            if not isinstance(e, scim_exceptions.SCIMException):
                # This is where we adjust the exception-handling behavior. Instead of
                # putting str(e) in the response, we use a generic error that won't leak
                # information.
                scim_views_logger.exception("Unable to complete SCIM call.")
                e = scim_exceptions.SCIMException("Exception while processing SCIM request.")

            content = orjson.dumps(e.to_dict())
            return HttpResponse(
                content=content, content_type=scim_constants.SCIM_CONTENT_TYPE, status=e.status
            )


class ZulipSCIMView(ZulipSCIMViewMixin, SCIMView):
    pass


class ZulipSCIMUsersView(ZulipSCIMViewMixin, UsersView):
    pass


class ZulipSCIMSearchView(ZulipSCIMViewMixin, SearchView):
    pass


class ZulipSCIMUserSearchView(ZulipSCIMViewMixin, UserSearchView):
    pass