views.py

"""
LTI consumer plugin passthrough views
"""
import logging
import urllib

from django.contrib.auth import get_user_model
from django.core.exceptions import ObjectDoesNotExist, PermissionDenied
from django.http import JsonResponse, Http404
from django.db import transaction
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_http_methods
from django.views.decorators.clickjacking import xframe_options_sameorigin
from django.shortcuts import render
from django_filters.rest_framework import DjangoFilterBackend
from opaque_keys import InvalidKeyError
from opaque_keys.edx.keys import UsageKey
from rest_framework import viewsets, status
from rest_framework.decorators import action
from rest_framework.response import Response
from rest_framework.status import HTTP_400_BAD_REQUEST, HTTP_403_FORBIDDEN, HTTP_404_NOT_FOUND

from lti_consumer.api import get_lti_pii_sharing_state_for_course
from lti_consumer.exceptions import LtiError
from lti_consumer.models import (
    LtiConfiguration,
    LtiAgsLineItem,
    LtiDlContentItem,
)

from lti_consumer.lti_1p3.consumer import LTI_1P3_CONTEXT_TYPE
from lti_consumer.lti_1p3.exceptions import (
    Lti1p3Exception,
    LtiDeepLinkingContentTypeNotSupported,
    UnsupportedGrantType,
    MalformedJwtToken,
    MissingRequiredClaim,
    NoSuitableKeys,
    TokenSignatureExpired,
    UnknownClientId,
)
from lti_consumer.lti_1p3.extensions.rest_framework.constants import LTI_DL_CONTENT_TYPE_SERIALIZER_MAP
from lti_consumer.lti_1p3.extensions.rest_framework.serializers import (
    LtiAgsLineItemSerializer,
    LtiAgsScoreSerializer,
    LtiAgsResultSerializer,
    LtiNrpsContextMembershipBasicSerializer,
    LtiNrpsContextMembershipPIISerializer,
)
from lti_consumer.lti_1p3.extensions.rest_framework.permissions import (
    LtiAgsPermissions,
    LtiNrpsContextMembershipsPermissions,
)
from lti_consumer.lti_1p3.extensions.rest_framework.authentication import Lti1p3ApiAuthentication
from lti_consumer.lti_1p3.extensions.rest_framework.renderers import (
    LineItemsRenderer,
    LineItemRenderer,
    LineItemScoreRenderer,
    LineItemResultsRenderer,
    MembershipResultRenderer,
)
from lti_consumer.lti_1p3.extensions.rest_framework.parsers import (
    LineItemParser,
    LineItemScoreParser,
)
from lti_consumer.lti_1p3.extensions.rest_framework.utils import IgnoreContentNegotiation
from lti_consumer.plugin import compat
from lti_consumer.utils import _
from lti_consumer.track import track_event


log = logging.getLogger(__name__)


def has_block_access(user, block, course_key):
    """
    Checks if a user has access to given xblock.

    ``has_access`` doesn't checks for course enrollment. On the otherhand, ``get_course_with_access``
    only checks for the course itself. There is no way to check access for specific xblock. This function
    has been created to perform a combination of check for both enrollment and access for specific xblock.

    Args:
        user: User Object
        block: xblock Object to check permission for
        course_key: A course_key specifying which course run this access is for.

    Returns:
        bool: True if user has access, False otherwise.
    """
    # Get the course
    course = compat.get_course_by_id(course_key)

    # Check if user is authenticated & enrolled
    course_access = compat.user_course_access(course, user, 'load', check_if_enrolled=True, check_if_authenticated=True)

    # Check if user has access to xblock
    block_access = compat.user_has_access(user, 'load', block, course_key)

    # Return True if the user has access to xblock and is enrolled in that specific course.
    return course_access and block_access


@require_http_methods(["GET"])
def public_keyset_endpoint(request, usage_id=None, lti_config_id=None):
    """
    Gate endpoint to fetch public keysets from a problem

    This is basically a passthrough function that uses the
    OIDC response parameter `login_hint` to locate the block
    and run the proper handler.
    """
    try:
        if usage_id:
            lti_config = LtiConfiguration.objects.get(location=UsageKey.from_string(usage_id))
        elif lti_config_id:
            lti_config = LtiConfiguration.objects.get(config_id=lti_config_id)

        if lti_config.version != lti_config.LTI_1P3:
            raise LtiError(
                "LTI Error: LTI 1.1 blocks do not have a public keyset endpoint."
            )

        # Retrieve block's Public JWK
        # The underlying method will generate a new Private-Public Pair if one does
        # not exist, and retrieve the values.
        response = JsonResponse(lti_config.lti_1p3_public_jwk)
        response['Content-Disposition'] = 'attachment; filename=keyset.json'
        return response
    except (LtiError, InvalidKeyError, ObjectDoesNotExist) as exc:
        log.info(
            "Error while retrieving keyset for usage_id (%r) or lit_config_id (%s): %s",
            usage_id,
            lti_config_id,
            exc,
            exc_info=True,
        )
        raise Http404 from exc


@require_http_methods(["GET", "POST"])
def launch_gate_endpoint(request, suffix=None):  # pylint: disable=unused-argument
    """
    Gate endpoint that triggers LTI launch endpoint XBlock handler

    This uses the location key from the "login_hint" query parameter
    to identify the LtiConfiguration and its consumer to generate the
    LTI 1.3 Launch Form.
    """
    usage_id = request.GET.get('login_hint')
    if not usage_id:
        log.info('The `login_hint` query param in the request is missing or empty.')
        return render(request, 'html/lti_launch_error.html', status=HTTP_400_BAD_REQUEST)

    try:
        usage_key = UsageKey.from_string(usage_id)
    except InvalidKeyError as exc:
        log.error(
            "The login_hint: %s is not a valid block location. Error: %s",
            usage_id,
            exc,
            exc_info=True
        )
        return render(request, 'html/lti_launch_error.html', status=HTTP_404_NOT_FOUND)

    try:
        lti_config = LtiConfiguration.objects.get(
            location=usage_key
        )
    except LtiConfiguration.DoesNotExist as exc:
        log.error("Invalid usage_id '%s' for LTI 1.3 Launch callback", usage_id)
        raise Http404 from exc

    if lti_config.version != LtiConfiguration.LTI_1P3:
        log.error("The LTI Version of configuration %s is not LTI 1.3", lti_config)
        return render(request, 'html/lti_launch_error.html', status=HTTP_404_NOT_FOUND)

    context = {}

    course_key = usage_key.course_key
    course = compat.get_course_by_id(course_key)
    user_role = compat.get_user_role(request.user, course_key)
    external_user_id = compat.get_external_id_for_user(request.user)
    lti_consumer = lti_config.get_lti_consumer()

    try:
        # Pass user data
        # Pass django user role to library
        lti_consumer.set_user_data(user_id=external_user_id, role=user_role)

        # Set launch context
        # Hardcoded for now, but we need to translate from
        # self.launch_target to one of the LTI compliant names,
        # either `iframe`, `frame` or `window`
        # This is optional though
        lti_consumer.set_launch_presentation_claim('iframe')

        # Set context claim
        # This is optional
        context_title = " - ".join([
            course.display_name_with_default,
            course.display_org_with_default
        ])
        # Course ID is the context ID for the LTI for now. This can be changed to be
        # more specific in the future for supporting other tools like discussions, etc.
        lti_consumer.set_context_claim(
            str(course_key),
            context_types=[LTI_1P3_CONTEXT_TYPE.course_offering],
            context_title=context_title,
            context_label=str(course_key)
        )

        # Retrieve preflight response
        preflight_response = request.GET.dict()
        lti_message_hint = preflight_response.get('lti_message_hint', '')

        # Set LTI Launch URL
        context.update({'launch_url': lti_consumer.launch_url})

        # Modify LTI Launch URL dependind on launch type
        # Deep Linking Launch - Configuration flow launched by
        # course creators to set up content.
        if lti_consumer.dl and lti_message_hint == 'deep_linking_launch':
            # Check if the user is staff before LTI doing deep linking launch.
            # If not, raise exception and display error page
            if user_role not in ['instructor', 'staff']:
                raise AssertionError('Deep Linking can only be performed by instructors and staff.')
            # Set deep linking launch
            context.update({'launch_url': lti_consumer.dl.deep_linking_launch_url})

        # Deep Linking ltiResourceLink content presentation
        # When content type is `ltiResourceLink`, the tool will be launched with
        # different parameters, set by instructors when running the DL configuration flow.
        elif lti_consumer.dl and 'deep_linking_content_launch' in lti_message_hint:
            # Retrieve Deep Linking parameters using lti_message_hint parameter.
            deep_linking_id = lti_message_hint.split(':')[1]
            content_item = lti_config.ltidlcontentitem_set.get(pk=deep_linking_id)
            # Only filter DL content item from content item set in the same LTI configuration.
            # This avoids a malicious user to input a random LTI id and perform LTI DL
            # content launches outside the scope of its configuration.
            dl_params = content_item.attributes

            # Modify LTI launch and set ltiResourceLink parameters
            lti_consumer.set_dl_content_launch_parameters(
                url=dl_params.get('url'),
                custom=dl_params.get('custom')
            )

        # Update context with LTI launch parameters
        context.update({
            "preflight_response": preflight_response,
            "launch_request": lti_consumer.generate_launch_request(
                resource_link=usage_id,
                preflight_response=preflight_response
            )
        })
        event = {
            'lti_version': lti_config.version,
            'user_roles': user_role,
            'launch_url': context['launch_url']
        }
        track_event('xblock.launch_request', event)

        return render(request, 'html/lti_1p3_launch.html', context)
    except Lti1p3Exception as exc:
        log.warning(
            "Error preparing LTI 1.3 launch for block %r: %s",
            usage_id,
            exc,
            exc_info=True
        )
        return render(request, 'html/lti_launch_error.html', context, status=HTTP_400_BAD_REQUEST)
    except AssertionError as exc:
        log.warning(
            "Permission on LTI block %r denied for user %r: %s",
            usage_id,
            external_user_id,
            exc,
            exc_info=True
        )
        return render(request, 'html/lti_1p3_permission_error.html', context, status=HTTP_403_FORBIDDEN)


@csrf_exempt
@require_http_methods(["POST"])
def access_token_endpoint(request, lti_config_id=None, usage_id=None):
    """
    Gate endpoint to enable tools to retrieve access tokens for the LTI 1.3 tool.

    This endpoint is only valid when a LTI 1.3 tool is being used.

    Arguments:
        lti_config_id (UUID): config_id of the LtiConfiguration
        usage_id (UsageKey): location of the Block

    Returns:
        JsonResponse or Http404

    References:
        Sucess: https://tools.ietf.org/html/rfc6749#section-4.4.3
        Failure: https://tools.ietf.org/html/rfc6749#section-5.2
    """

    try:
        if lti_config_id:
            lti_config = LtiConfiguration.objects.get(config_id=lti_config_id)
        else:
            usage_key = UsageKey.from_string(usage_id)
            lti_config = LtiConfiguration.objects.get(location=usage_key)
    except LtiConfiguration.DoesNotExist as exc:
        log.warning("Error getting the LTI configuration with id %r: %s", lti_config_id, exc, exc_info=True)
        raise Http404 from exc

    if lti_config.version != lti_config.LTI_1P3:
        return JsonResponse({"error": "invalid_lti_version"}, status=HTTP_404_NOT_FOUND)

    lti_consumer = lti_config.get_lti_consumer()
    try:
        token = lti_consumer.access_token(
            dict(urllib.parse.parse_qsl(
                request.body.decode('utf-8'),
                keep_blank_values=True
            ))
        )
        return JsonResponse(token)

    # Handle errors and return a proper response
    except MissingRequiredClaim:
        # Missing request attibutes
        return JsonResponse({"error": "invalid_request"}, status=HTTP_400_BAD_REQUEST)
    except (MalformedJwtToken, TokenSignatureExpired):
        # Triggered when a invalid grant token is used
        return JsonResponse({"error": "invalid_grant"}, status=HTTP_400_BAD_REQUEST)
    except (NoSuitableKeys, UnknownClientId):
        # Client ID is not registered in the block or
        # isn't possible to validate token using available keys.
        return JsonResponse({"error": "invalid_client"}, status=HTTP_400_BAD_REQUEST)
    except UnsupportedGrantType:
        return JsonResponse({"error": "unsupported_grant_type"}, status=HTTP_400_BAD_REQUEST)


# Post from external tool that doesn't
# have access to CSRF tokens
@csrf_exempt
# This URL should work inside an iframe
@xframe_options_sameorigin
# Post only, as required by LTI-DL Specification
@require_http_methods(["POST"])
def deep_linking_response_endpoint(request, lti_config_id=None):
    """
    Deep Linking response endpoint where tool can send back Deep Linking
    content selected by instructions in the tool's UI.

    For this feature to work, the LMS session cookies need to be Secure
    and have the `SameSite` attribute set to `None`, otherwise we won't
    be able to check user permissions.
    """
    try:
        # Retrieve LTI configuration
        lti_config = LtiConfiguration.objects.get(id=lti_config_id)

        # Get LTI consumer
        lti_consumer = lti_config.get_lti_consumer()

        # Validate Deep Linking return message and return decoded message
        content_items = lti_consumer.check_and_decode_deep_linking_token(
            request.POST.get("JWT")
        )

        # Check if the user has sufficient permissions to
        # save LTI Deep Linking content through the student.auth API.
        course_key = lti_config.location.course_key
        if not compat.user_has_studio_write_access(request.user, course_key):
            raise PermissionDenied()

        # On a transaction, clear older DeepLinking selections, then
        # verify and save each content item passed from the tool.
        with transaction.atomic():
            # Erase older deep linking selection
            LtiDlContentItem.objects.filter(lti_configuration=lti_config).delete()

            for content_item in content_items:
                content_type = content_item.get('type')

                # Retrieve serializer (or raise)
                # pylint: disable=consider-iterating-dictionary
                if content_type not in LTI_DL_CONTENT_TYPE_SERIALIZER_MAP.keys():
                    raise LtiDeepLinkingContentTypeNotSupported()
                serializer_cls = LTI_DL_CONTENT_TYPE_SERIALIZER_MAP[content_type]

                # Validate content item data
                serializer = serializer_cls(data=content_item)
                serializer.is_valid(True)

                # Save content item
                LtiDlContentItem.objects.create(
                    lti_configuration=lti_config,
                    content_type=content_type,
                    attributes=serializer.validated_data,
                )

        # Display sucess page to indicate that LTI DL Content was
        # saved successfully and auto-close after a few seconds.
        return render(request, 'html/lti-dl/dl_response_saved.html')

    # If LtiConfiguration doesn't exist, error with 404 status.
    except LtiConfiguration.DoesNotExist as exc:
        log.info("LtiConfiguration %r does not exist: %s", lti_config_id, exc)
        raise Http404 from exc
    # If the deep linking content type is not supported
    except LtiDeepLinkingContentTypeNotSupported as exc:
        log.info("One of the selected LTI Content Types is not supported: %s", exc)
        return render(
            request,
            'html/lti-dl/dl_response_error.html',
            {"error": _("The selected content type is not supported by Open edX.")},
            status=400
        )
    # Bad JWT message, invalid token, or any other message validation issues
    except (Lti1p3Exception, PermissionDenied) as exc:
        log.warning(
            "Permission on LTI Config %r denied for user %r: %s",
            lti_config,
            request.user,
            exc,
        )
        return render(
            request,
            'html/lti-dl/dl_response_error.html',
            {
                "error": _("You don't have access to save LTI Content Items."),
                "explanation": _("Please check that you have course staff permissions "
                                 "and double check this block's LTI settings."),
            },
            status=403
        )


@require_http_methods(['GET'])
@xframe_options_sameorigin
def deep_linking_content_endpoint(request, lti_config_id=None):
    """
    Deep Linking endpoint for rendering Deep Linking Content Items.
    """
    try:
        # Get LTI Configuration
        lti_config = LtiConfiguration.objects.get(id=lti_config_id)
    except LtiConfiguration.DoesNotExist as exc:
        log.info("LtiConfiguration %r does not exist: %s", lti_config_id, exc)
        raise Http404 from exc

    # check if user has proper access
    if not has_block_access(request.user, lti_config.block, lti_config.location.course_key):
        log.warning(
            "Permission on LTI Config %r denied for user %r.",
            lti_config_id,
            request.user,
        )
        raise PermissionDenied

    # Get all LTI-DL contents
    content_items = LtiDlContentItem.objects.filter(lti_configuration=lti_config)

    # If no LTI-DL contents found for current configuration, throw 404 error
    if not content_items.exists():
        log.info("There's no Deep linking content for LTI configuration %s.", lti_config)
        raise Http404

    # Render LTI-DL contents
    return render(request, 'html/lti-dl/render_dl_content.html', {
        'content_items': content_items,
        'block': lti_config.block,
    })


class LtiAgsLineItemViewset(viewsets.ModelViewSet):
    """
    LineItem endpoint implementation from LTI Advantage.

    See full documentation at:
    https://www.imsglobal.org/spec/lti-ags/v2p0#line-item-service
    """
    serializer_class = LtiAgsLineItemSerializer
    pagination_class = None

    # Custom permission classes for LTI APIs
    authentication_classes = [Lti1p3ApiAuthentication]
    permission_classes = [LtiAgsPermissions]

    # Renderer/parser classes to accept LTI AGS content types
    renderer_classes = [
        LineItemsRenderer,
        LineItemRenderer,
    ]
    parser_classes = [LineItemParser]

    # Filters
    filter_backends = [DjangoFilterBackend]
    filterset_fields = [
        'resource_link_id',
        'resource_id',
        'tag'
    ]

    def get_queryset(self):
        lti_configuration = self.request.lti_configuration

        # Return all LineItems related to the LTI configuration.
        # TODO:
        # Note that each configuration currently maps 1:1
        # to each resource link (block), and this filter needs
        # improved once we start reusing LTI configurations.
        return LtiAgsLineItem.objects.filter(
            lti_configuration=lti_configuration
        )

    def perform_create(self, serializer):
        lti_configuration = self.request.lti_configuration
        serializer.save(lti_configuration=lti_configuration)

    @action(
        detail=True,
        methods=['GET'],
        url_path='results/(?P<user_id>[^/.]+)?',
        renderer_classes=[LineItemResultsRenderer],
        content_negotiation_class=IgnoreContentNegotiation,
    )
    def results(self, request, user_id=None, **kwargs):
        """
        Return a Result list for an LtiAgsLineItem

        URL Parameters:
          * user_id (string): String external user id representation.

        Query Parameters:
          * limit (integer): The maximum number of records to return. Records are
                sorted with most recent timestamp first

        Returns:
          * An array of Result records, formatted by LtiAgsResultSerializer
                and returned with the media-type for LineItemResultsRenderer
        """
        line_item = self.get_object()
        scores = line_item.scores.filter(score_given__isnull=False).order_by('-timestamp')

        if user_id:
            scores = scores.filter(user_id=user_id)

        if request.query_params.get('limit'):
            scores = scores[:int(request.query_params.get('limit'))]

        serializer = LtiAgsResultSerializer(
            list(scores),
            context={'request': self.request},
            many=True,
        )

        return Response(serializer.data)

    @action(
        detail=True,
        methods=['POST'],
        parser_classes=[LineItemScoreParser],
        renderer_classes=[LineItemScoreRenderer],
        content_negotiation_class=IgnoreContentNegotiation,
    )
    def scores(self, request, *args, **kwargs):
        """
        Create a Score record for an LtiAgsLineItem

        Data:
          * A JSON object capable of being serialized by LtiAgsScoreSerializer

        Returns:
          * An copy of the saved record, formatted by LtiAgsScoreSerializer
                and returned with the media-type for LineItemScoreRenderer
        """
        line_item = self.get_object()

        user_id = request.data.get('userId')

        # Using `filter` and `first` so that when a score does not exist,
        # `existing_score` is set to `None`. Using `get` will raise `DoesNotExist`
        existing_score = line_item.scores.filter(user_id=user_id).first()

        serializer = LtiAgsScoreSerializer(
            instance=existing_score,
            data=request.data,
            context={'request': self.request},
        )
        serializer.is_valid(raise_exception=True)
        serializer.save(line_item=line_item)
        headers = self.get_success_headers(serializer.data)
        return Response(
            serializer.data,
            status=status.HTTP_201_CREATED,
            headers=headers
        )


class LtiNrpsContextMembershipViewSet(viewsets.ReadOnlyModelViewSet):
    """
    LTI NRPS Context Membership Service endpoint.

    See full documentation at:
    http://imsglobal.org/spec/lti-nrps/v2p0
    """

    # Custom permission classes for LTI APIs
    authentication_classes = [Lti1p3ApiAuthentication]
    permission_classes = [LtiNrpsContextMembershipsPermissions]

    # Renderer classes to accept LTI NRPS content types
    renderer_classes = [
        MembershipResultRenderer,
    ]

    def attach_external_user_ids(self, data):
        """
        Preprocess the output of `get_membership` method amd appends external ids to each user.
        """

        # batch get or create external ids for all users
        user_ids = data.keys()
        users = get_user_model().objects.prefetch_related('profile').filter(id__in=user_ids)

        # get external ids
        external_ids = compat.batch_get_or_create_externalids(users)

        for userid in user_ids:
            # append external ids to user
            data[userid]['external_id'] = external_ids[userid].external_user_id

    def get_serializer_class(self):
        """
        Overrides ModelViewSet's `get_serializer_class` method.
        Checks if PII fields can be exposed and returns appropiate serializer.
        """
        if get_lti_pii_sharing_state_for_course(self.request.lti_configuration.location.course_key):
            return LtiNrpsContextMembershipPIISerializer
        else:
            return LtiNrpsContextMembershipBasicSerializer

    def list(self, *args, **kwargs):
        """
        Overrides default list method of ModelViewSet. Calls LMS `get_course_members`
        API and returns result.
        """

        # get course key
        course_key = self.request.lti_configuration.location.course_key

        try:
            data = compat.get_course_members(course_key)
            self.attach_external_user_ids(data)

            # build correct format for the serializer
            result = {
                'id': self.request.build_absolute_uri(),
                'context': {
                    'id': course_key
                },
                'members': data.values(),
            }

            # Serialize and return data NRPS reponse.
            serializer = self.get_serializer_class()(result)
            return Response(serializer.data)

        except LtiError as ex:
            log.warning("LTI NRPS Error: %s", ex)
            return Response({
                "error": "above_response_limit",
                "explanation": "The number of retrieved users is bigger than the maximum allowed in the configuration.",
            }, status=HTTP_403_FORBIDDEN)