feat: add courseware-access pipeline steps for DataSharingConsentRedirectStep, EnterpriseStartDateAccessFailureStep, ActiveEnterpriseCheckStep, and DataSharingConsentCourseAccessStep

DataSharingConsentRedirectStep supplies a redirect URL when data sharing consent is
required. EnterpriseStartDateAccessFailureStep supplies a start-date error for
enterprise learners. ActiveEnterpriseCheckStep denies access when the
learner's active EnterpriseCustomer differs from the enrollment's customer.
DataSharingConsentCourseAccessStep denies access when consent is required.

ENT-11544

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

Author: pwnage101

consent/filters/__init__.py

@@ -0,0 +1 @@
+"""Pipeline steps registered against openedx-filters by the Consent app."""

consent/filters/courseware.py

@@ -0,0 +1,57 @@
+"""
+Pipeline steps for courseware filters, contributed by the Consent app.
+"""
+from typing import Optional
+
+from crum import get_current_request
+from django.contrib.auth.base_user import AbstractBaseUser
+from django.http import HttpRequest
+from opaque_keys.edx.keys import CourseKey
+from openedx_filters.filters import PipelineStep
+from openedx_filters.learning.filters import CoursewareAccessChecksRequested
+
+from consent.helpers import get_enterprise_consent_url
+
+
+class DataSharingConsentRedirectStep(PipelineStep):
+    """
+    Sets the data-sharing consent redirect URL when the user has not yet consented.
+
+    Registered against ``org.openedx.learning.courseware.view.started.v1``.
+    Earlier steps win: if ``redirect_url`` is already set, this step is a no-op.
+    """
+
+    def run_filter(
+        self,
+        redirect_url: Optional[str],
+        request: HttpRequest,
+        course_key: CourseKey,
+    ) -> dict:  # pylint: disable=arguments-differ
+        if redirect_url is None:
+            consent_url = get_enterprise_consent_url(request, str(course_key))
+            if consent_url:
+                redirect_url = consent_url
+        return {"redirect_url": redirect_url, "request": request, "course_key": course_key}
+
+
+class DataSharingConsentCourseAccessStep(PipelineStep):
+    """
+    Deny courseware access when data sharing consent is required but not granted.
+
+    Registered against ``org.openedx.learning.courseware.access_checks.requested.v1``.
+    Raises ``CoursewareAccessChecksRequested.PreventCoursewareAccess`` to deny
+    access when ``get_enterprise_consent_url`` returns a URL.
+    """
+
+    def run_filter(self, user: AbstractBaseUser, course_key: CourseKey) -> dict:  # pylint: disable=arguments-differ
+        request = get_current_request()
+        if request is None:
+            return {"user": user, "course_key": course_key}
+        consent_url = get_enterprise_consent_url(request, str(course_key))
+        if consent_url:
+            raise CoursewareAccessChecksRequested.PreventCoursewareAccess(
+                error_code="data_sharing_access_required",
+                developer_message=consent_url,
+                user_message="You must give Data Sharing Consent for the course",
+            )
+        return {"user": user, "course_key": course_key}

consent/helpers.py

@@ -2,11 +2,168 @@
 Helper functions for the Consent application.
 """
 
+import logging
+from urllib.parse import urlencode
+
 from django.apps import apps
+from django.conf import settings
+from django.contrib.sites.models import Site
+from django.urls import reverse
+from edx_django_utils.cache import TieredCache
 
 from consent.models import ProxyDataSharingConsent
 from enterprise.api_client.discovery import get_course_catalog_api_service_client
-from enterprise.utils import get_enterprise_customer
+from enterprise.utils import (
+    _enterprise_integration_enabled,
+    get_active_enterprise_customer_user,
+    get_enterprise_customer,
+)
+
+try:
+    from openedx.features.enterprise_support.api import (
+        CONSENT_FAILED_PARAMETER,
+        ConsentApiClient,
+        enterprise_customer_uuid_for_request,
+    )
+    from openedx.features.enterprise_support.utils import get_data_consent_share_cache_key
+except ImportError:
+    CONSENT_FAILED_PARAMETER = 'consent_failed'
+    ConsentApiClient = None
+    enterprise_customer_uuid_for_request = None
+    get_data_consent_share_cache_key = None
+
+LOGGER = logging.getLogger(__name__)
+
+
+def consent_needed_for_course(request, user, course_id, enrollment_exists=False):
+    """
+    Determine whether ``user`` must grant data-sharing consent before accessing ``course_id``.
+    """
+    if not _enterprise_integration_enabled():
+        return False
+    if (
+        ConsentApiClient is None
+        or enterprise_customer_uuid_for_request is None
+        or get_data_consent_share_cache_key is None
+    ):
+        return False
+    LOGGER.info(
+        "[ENTERPRISE DSC] Determining if user [%s] must consent to data sharing for course [%s]",
+        user.username, course_id,
+    )
+
+    active_enterprise_learner_info = get_active_enterprise_customer_user(user)
+    if not active_enterprise_learner_info:
+        LOGGER.info(
+            "[ENTERPRISE DSC] Consent from user [%s] is not needed for course [%s]. "
+            "The user is not linked to an enterprise.",
+            user.username, course_id,
+        )
+        return False
+
+    active_enterprise_customer = active_enterprise_learner_info['enterprise_customer']
+
+    consent_cache_key = get_data_consent_share_cache_key(
+        user.id, course_id, active_enterprise_customer['uuid'],
+    )
+    cached = TieredCache.get_cached_response(consent_cache_key)
+    if cached.is_found and cached.value == 0:
+        LOGGER.info(
+            "[ENTERPRISE DSC] Consent from user [%s] is not needed for course [%s]. "
+            "The DSC cache was checked and the value was 0.",
+            user.username, course_id,
+        )
+        return False
+
+    if not active_enterprise_customer['enable_data_sharing_consent']:
+        LOGGER.info(
+            "[ENTERPRISE DSC] DSC is disabled for enterprise customer [%s]. "
+            "Consent from user [%s] is not needed for course [%s]",
+            active_enterprise_customer['slug'], user.username, course_id,
+        )
+        TieredCache.set_all_tiers(consent_cache_key, 0, settings.DATA_CONSENT_SHARE_CACHE_TIMEOUT)
+        return False
+
+    current_enterprise_uuid = enterprise_customer_uuid_for_request(request)
+    if str(current_enterprise_uuid) != str(active_enterprise_customer['uuid']):
+        LOGGER.info(
+            '[ENTERPRISE DSC] Enterprise mismatch. USER: [%s], RequestEnterprise: [%s], '
+            'LearnerEnterprise: [%s]',
+            user.username, current_enterprise_uuid, active_enterprise_customer['uuid'],
+        )
+        TieredCache.set_all_tiers(consent_cache_key, 0, settings.DATA_CONSENT_SHARE_CACHE_TIMEOUT)
+        return False
+
+    enterprise_domain = Site.objects.get(domain=active_enterprise_customer['site']['domain'])
+    if enterprise_domain != request.site:
+        LOGGER.info(
+            '[ENTERPRISE DSC] Site mismatch. USER: [%s], RequestSite: [%s], '
+            'LearnerEnterpriseDomain: [%s]',
+            user.username, request.site, enterprise_domain,
+        )
+        TieredCache.set_all_tiers(consent_cache_key, 0, settings.DATA_CONSENT_SHARE_CACHE_TIMEOUT)
+        return False
+
+    client = ConsentApiClient(user=request.user)
+    consent_required = client.consent_required(
+        username=user.username,
+        course_id=course_id,
+        enterprise_customer_uuid=current_enterprise_uuid,
+        enrollment_exists=enrollment_exists,
+    )
+    if not consent_required:
+        LOGGER.info(
+            "[ENTERPRISE DSC] Consent from user [%s] is not needed for course [%s]. "
+            "The user's current enterprise does not require data sharing consent.",
+            user.username, course_id,
+        )
+        TieredCache.set_all_tiers(consent_cache_key, 0, settings.DATA_CONSENT_SHARE_CACHE_TIMEOUT)
+        return False
+
+    LOGGER.info(
+        "[ENTERPRISE DSC] Consent from user [%s] is needed for course [%s]. "
+        "The user's current enterprise requires data sharing consent, and it has not been given.",
+        user.username, course_id,
+    )
+    return True
+
+
+def get_enterprise_consent_url(request, course_id, user=None, return_to=None, enrollment_exists=False, source='lms'):
+    """
+    Build a URL to redirect the user to the data-sharing consent page for a specific course.
+
+    Arguments:
+        request: Django request object.
+        course_id: Course key/identifier string.
+        user: user to check for consent. If None, uses ``request.user``.
+        return_to: url name for the page to return to after consent is granted; defaults to
+            ``request.path``.
+        enrollment_exists: forwarded to ``consent_needed_for_course``.
+        source: opaque string identifying the caller, recorded on the consent URL.
+    """
+    if enterprise_customer_uuid_for_request is None:
+        return None
+    user = user or request.user
+    LOGGER.info(
+        'Getting enterprise consent url for user [%s] and course [%s].',
+        user.username,
+        course_id,
+    )
+    if not consent_needed_for_course(request, user, course_id, enrollment_exists=enrollment_exists):
+        return None
+    return_path = request.path if return_to is None else reverse(return_to, args=(course_id,))
+    url_params = {
+        'enterprise_customer_uuid': enterprise_customer_uuid_for_request(request),
+        'course_id': course_id,
+        'source': source,
+        'next': request.build_absolute_uri(return_path),
+        'failure_url': request.build_absolute_uri(
+            reverse('dashboard') + '?' + urlencode({CONSENT_FAILED_PARAMETER: course_id})
+        ),
+    }
+    full_url = reverse('grant_data_sharing_permissions') + '?' + urlencode(url_params)
+    LOGGER.info('Redirecting to %s to complete data sharing consent', full_url)
+    return full_url
 
 
 def get_data_sharing_consent(username, enterprise_customer_uuid, course_id=None, program_uuid=None):

consent/settings/common.py

@@ -6,7 +6,16 @@
 # enterprise plugin's settings module.
 from enterprise.settings.common import FiltersConfig, _merge_filters_config
 
-CONSENT_FILTERS_CONFIG: FiltersConfig = {}
+CONSENT_FILTERS_CONFIG: FiltersConfig = {
+    "org.openedx.learning.courseware.view.started.v1": {
+        "fail_silently": True,
+        "pipeline": ["consent.filters.courseware.DataSharingConsentRedirectStep"],
+    },
+    "org.openedx.learning.courseware.access_checks.requested.v1": {
+        "fail_silently": True,
+        "pipeline": ["consent.filters.courseware.DataSharingConsentCourseAccessStep"],
+    },
+}
 
 
 def plugin_settings(settings):

enterprise/filters/courseware.py

@@ -0,0 +1,107 @@
+"""
+Pipeline steps for courseware-related openedx-filters contributed by the Enterprise app.
+"""
+import logging
+from typing import Optional
+
+from django.contrib.auth.base_user import AbstractBaseUser
+from django.http import HttpRequest
+from opaque_keys.edx.keys import CourseKey
+from openedx_filters.filters import PipelineStep
+from openedx_filters.learning.filters import CoursewareAccessChecksRequested
+
+try:
+    from openedx.features.enterprise_support.api import enterprise_customer_from_session_or_learner_data
+except ImportError:
+    enterprise_customer_from_session_or_learner_data = None
+
+from enterprise.models import EnterpriseCourseEnrollment, EnterpriseCustomerUser
+
+log = logging.getLogger(__name__)
+
+
+class EnterpriseStartDateAccessFailureStep(PipelineStep):
+    """
+    Substitutes a more specific start-date access-error payload for enterprise learners.
+
+    Registered against ``org.openedx.learning.course.start_date.validation_failed.v1``.
+    If ``error_code`` is already set (an earlier step won) or the request user is not
+    associated with the active session enterprise customer, this step passes through
+    unchanged.
+    """
+
+    def run_filter(
+        self,
+        error_code: Optional[str],
+        developer_message: Optional[str],
+        user_message: Optional[str],
+        request: HttpRequest,
+        course_key: CourseKey,
+    ) -> dict:  # pylint: disable=arguments-differ
+        if error_code is None and enterprise_customer_from_session_or_learner_data is not None:
+            enterprise_customer = enterprise_customer_from_session_or_learner_data(request)
+            if enterprise_customer:
+                user = request.user
+                is_enterprise_learner = EnterpriseCustomerUser.objects.filter(
+                    user_id=user.id,
+                    enterprise_customer__uuid=enterprise_customer.get('uuid'),
+                ).exists()
+                if is_enterprise_learner:
+                    error_code = "course_not_started_enterprise_learner"
+                    developer_message = (
+                        f"Course {course_key} has not started, and the learner is "
+                        "enrolled via an enterprise subsidy."
+                    )
+                    user_message = "Course has not started"
+        return {
+            "error_code": error_code,
+            "developer_message": developer_message,
+            "user_message": user_message,
+            "request": request,
+            "course_key": course_key,
+        }
+
+
+class ActiveEnterpriseCheckStep(PipelineStep):
+    """
+    Deny access when the learner's active EnterpriseCustomer differs from the
+    EnterpriseCustomer attached to their EnterpriseCourseEnrollment for this course.
+
+    Registered against ``org.openedx.learning.courseware.access_checks.requested.v1``.
+    Raises ``CoursewareAccessChecksRequested.PreventCoursewareAccess`` to deny access.
+    """
+
+    def run_filter(self, user: AbstractBaseUser, course_key: CourseKey) -> dict:  # pylint: disable=arguments-differ
+        enterprise_enrollments = EnterpriseCourseEnrollment.objects.filter(
+            course_id=course_key, enterprise_customer_user__user_id=user.id,
+        )
+        if not enterprise_enrollments.exists():
+            return {"user": user, "course_key": course_key}
+
+        try:
+            active_ecu = EnterpriseCustomerUser.objects.get(user_id=user.id, active=True)
+            if enterprise_enrollments.filter(enterprise_customer_user=active_ecu).exists():
+                return {"user": user, "course_key": course_key}
+            active_enterprise_name = active_ecu.enterprise_customer.name
+        except (EnterpriseCustomerUser.DoesNotExist, EnterpriseCustomerUser.MultipleObjectsReturned):
+            log.error("Multiple or No Active Enterprise found for the user %s.", user.id)
+            active_enterprise_name = "Incorrect"
+
+        enrollment_enterprise_name = (
+            enterprise_enrollments.first().enterprise_customer_user.enterprise_customer.name
+        )
+        user_message = (
+            "You are enrolled in this course with '{enrollment_enterprise_name}'. However, you are "
+            "currently logged in as a '{active_enterprise_name}' user. Please log in with "
+            "'{enrollment_enterprise_name}' to access this course."
+        ).format(
+            enrollment_enterprise_name=enrollment_enterprise_name,
+            active_enterprise_name=active_enterprise_name,
+        )
+        raise CoursewareAccessChecksRequested.PreventCoursewareAccess(
+            error_code="incorrect_active_enterprise",
+            developer_message=(
+                "User active enterprise should be same as EnterpriseCourseEnrollment enterprise."
+            ),
+            user_message=user_message,
+        )

enterprise/settings/common.py

@@ -20,6 +20,14 @@
         "fail_silently": False,
         "pipeline": ["enterprise.filters.grades.GradeEventContextEnricher"],
     },
+    "org.openedx.learning.course.start_date.validation_failed.v1": {
+        "fail_silently": True,
+        "pipeline": ["enterprise.filters.courseware.EnterpriseStartDateAccessFailureStep"],
+    },
+    "org.openedx.learning.courseware.access_checks.requested.v1": {
+        "fail_silently": True,
+        "pipeline": ["enterprise.filters.courseware.ActiveEnterpriseCheckStep"],
+    },
 }