fix: enforce GlobalRestrictedCountry on course access, not just registration

lms/djangoapps/course_home_api/outline/tests/test_view.py

@@ -30,6 +30,7 @@
 from openedx.core.djangoapps.content.learning_sequences.api import replace_course_outline
 from openedx.core.djangoapps.content.learning_sequences.data import CourseOutlineData, CourseVisibility
 from openedx.core.djangoapps.course_date_signals.utils import MIN_DURATION
+from openedx.core.djangoapps.embargo.models import Country, GlobalRestrictedCountry
 from openedx.core.djangoapps.user_api.preferences.api import set_user_preference
 from openedx.core.djangoapps.user_api.tests.factories import UserCourseTagFactory
 from openedx.features.course_duration_limits.models import CourseDurationLimitConfig
@@ -53,6 +54,33 @@ def update_course_and_overview(self):
         self.update_course(self.course, self.user.id)
         CourseOverview.load_from_module_store(self.course.id)
 
+    @override_settings(EMBARGO=True)
+    def test_embargo_blocks_access(self):
+        """
+        A `GlobalRestrictedCountry` block applies here too, not just to the legacy
+        courseware page - this is the code path the Learning MFE actually calls,
+        which `EmbargoMiddleware`'s URL-pattern matching does not cover.
+        """
+        CourseEnrollment.enroll(self.user, self.course.id)
+        GlobalRestrictedCountry.objects.create(country=Country.objects.create(country='CU'))
+
+        with patch('openedx.core.djangoapps.embargo.api.country_code_from_ip', return_value='CU'):
+            response = self.client.get(self.url, HTTP_X_FORWARDED_FOR='1.2.3.4')
+
+        assert response.status_code == 403
+        assert response.data['detail'].code == 'embargo'
+
+    @override_settings(EMBARGO=True)
+    def test_embargo_staff_bypass(self):
+        """ Staff should still get access even when their country is globally restricted. """
+        self.switch_to_staff()
+        GlobalRestrictedCountry.objects.create(country=Country.objects.create(country='CU'))
+
+        with patch('openedx.core.djangoapps.embargo.api.country_code_from_ip', return_value='CU'):
+            response = self.client.get(self.url, HTTP_X_FORWARDED_FOR='1.2.3.4')
+
+        assert response.status_code == 200
+
     @override_waffle_flag(ENABLE_COURSE_GOALS, active=True)
     @ddt.data(CourseMode.AUDIT, CourseMode.VERIFIED)
     def test_get_authenticated_enrolled_user(self, enrollment_mode):

lms/djangoapps/courseware/access_response.py

@@ -290,3 +290,15 @@ def __init__(self, courselike):
             course_name=courselike.display_name_with_default,
         )
         super().__init__(error_code, developer_message, user_message)
+
+
+class EmbargoAccessError(AccessError):
+    """
+    Access denied because the user's country is blocked by embargo rules
+    (`GlobalRestrictedCountry` or a per-course `CountryAccessRule`).
+    """
+    def __init__(self):
+        error_code = "embargo"
+        developer_message = "User's location is blocked by country embargo rules"
+        user_message = _("Access to this course is blocked from your current location")
+        super().__init__(error_code, developer_message, user_message)

lms/djangoapps/courseware/access_utils.py

@@ -6,7 +6,9 @@
 from datetime import datetime, timedelta
 from logging import getLogger
 
+from crum import get_current_request
 from django.conf import settings
+from edx_django_utils import ip
 from openedx_filters.learning.filters import CourseStartDateValidationFailed
 from pytz import UTC
 
@@ -15,10 +17,12 @@
 from lms.djangoapps.courseware.access_response import (
     AccessResponse,
     AuthenticationRequiredAccessError,
+    EmbargoAccessError,
     EnrollmentRequiredAccessError,
     StartDateError,
 )
 from lms.djangoapps.courseware.masquerade import get_course_masquerade, is_masquerading_as_student
+from openedx.core.djangoapps.embargo import api as embargo_api
 from openedx.features.course_experience import (
     COURSE_ENABLE_UNENROLLED_ACCESS_FLAG,
     COURSE_PRE_START_ACCESS_FLAG,
@@ -159,6 +163,28 @@ def check_authentication(user, course):
     return AuthenticationRequiredAccessError()
 
 
+def check_embargo_access(user, course):
+    """
+    Deny access if the user's country is blocked by embargo rules.
+
+    This is the single choke point shared by every caller of `check_course_access` -
+    both legacy courseware pages (also covered separately by `EmbargoMiddleware` for
+    URLs it recognizes) and the course-home BFF APIs used by the Learning MFE, which
+    the middleware's URL-based course-id detection does not cover.
+
+    Returns:
+        AccessResponse: Either ACCESS_GRANTED or EmbargoAccessError.
+    """
+    request = get_current_request()
+    ip_addresses = ip.get_all_client_ips(request) if request is not None else None
+    url = request.path if request is not None else None
+
+    if embargo_api.check_course_access(course.id, user=user, ip_addresses=ip_addresses, url=url):
+        return ACCESS_GRANTED
+
+    return EmbargoAccessError()
+
+
 def check_public_access(course, visibilities):
     """
     This checks if the unenrolled access waffle flag for the course is set

lms/djangoapps/courseware/courses.py

@@ -32,6 +32,7 @@
 from lms.djangoapps.courseware.access_response import (
     AuthenticationRequiredAccessError,
     CatalogVisibilityError,
+    EmbargoAccessError,
     EnrollmentRequiredAccessError,
     MilestoneAccessError,
     OldMongoAccessError,
@@ -40,6 +41,7 @@
 )
 from lms.djangoapps.courseware.access_utils import (
     check_authentication,
+    check_embargo_access,
     check_enrollment,
 )
 from lms.djangoapps.courseware.block_render import get_block
@@ -63,6 +65,7 @@
 from lms.djangoapps.survey.utils import SurveyRequiredAccessError, check_survey_required_and_unanswered
 from openedx.core.djangoapps.content.block_structure.api import get_block_structure_manager
 from openedx.core.djangoapps.content.course_overviews.models import CourseOverview
+from openedx.core.djangoapps.embargo.api import message_url_path
 from openedx.core.djangoapps.enrollments.api import get_course_enrollment_details
 from openedx.core.djangoapps.site_configuration import helpers as configuration_helpers
 from openedx.core.lib.api.view_utils import LazySequence
@@ -189,6 +192,13 @@ def _check_nonstaff_access():
         if not access_response:
             return access_response
 
+        # Embargo applies only to viewing a course's content, not to secondary
+        # actions like 'staff' access checks performed on the same course.
+        if action == 'load':
+            embargo_access_response = check_embargo_access(user, course)
+            if not embargo_access_response:
+                return embargo_access_response
+
         if check_if_authenticated:
             authentication_access_response = check_authentication(user, course)
             if not authentication_access_response:
@@ -307,6 +317,10 @@ def check_course_access_with_redirect(
         if isinstance(access_response, SurveyRequiredAccessError):
             raise CourseAccessRedirect(reverse('course_survey', args=[str(course.id)]))
 
+        # Redirect if the user's country is blocked by embargo rules
+        if isinstance(access_response, EmbargoAccessError):
+            raise CourseAccessRedirect(message_url_path(course.id, 'courseware'), access_response)
+
         # Deliberately return a non-specific error message to avoid
         # leaking info about access control settings
         raise CoursewareAccessException(access_response)

openedx/core/djangoapps/embargo/api.py

@@ -7,7 +7,7 @@
 """
 
 import logging
-from typing import List, Optional  # noqa: UP035
+from typing import List, NamedTuple, Optional  # noqa: UP035
 
 from django.conf import settings
 from django.core.cache import cache
@@ -21,7 +21,7 @@
 from openedx.core import types
 from openedx.core.djangoapps.geoinfo.api import country_code_from_ip
 
-from .models import CountryAccessRule, RestrictedCourse
+from .models import CountryAccessRule, GlobalRestrictedCountry, RestrictedCourse
 
 log = logging.getLogger(__name__)
 
@@ -36,7 +36,8 @@ def redirect_if_blocked(
     Redirect if the user does not have access to the course.
 
     Even if the user would normally be blocked, if the given access_point is 'courseware' and the course has enabled
-    the `is_disabled_access_check` flag, then the user can still view that course.
+    the `is_disabled_access_check` flag, then the user can still view that course - unless the block is coming from
+    `GlobalRestrictedCountry`, which `is_disabled_access_check` (a per-course override) can never bypass.
 
     Arguments:
         request: The current request to be checked.
@@ -50,10 +51,10 @@ def redirect_if_blocked(
     if settings.EMBARGO:
         client_ips = ip.get_all_client_ips(request)
         user = user or request.user
-        is_blocked = not check_course_access(course_key, user=user, ip_addresses=client_ips, url=request.path)
-        if is_blocked:
+        result = _check_course_access(course_key, user=user, ip_addresses=client_ips, url=request.path)
+        if not result.allowed:
             if access_point == "courseware":
-                if not RestrictedCourse.is_disabled_access_check(course_key):
+                if not RestrictedCourse.is_disabled_access_check(course_key) or result.blocked_globally:
                     return message_url_path(course_key, access_point)
             else:
                 return message_url_path(course_key, access_point)
@@ -68,6 +69,10 @@ def check_course_access(
     """
     Check is the user with this ip_addresses chain has access to the given course
 
+    A country listed in `GlobalRestrictedCountry` blocks every course, regardless
+    of whether the course has a `RestrictedCourse` entry. `CountryAccessRule`
+    checks only apply on top of that for courses that do have one.
+
     Arguments:
         course_key: Location of the course the user is trying to access.
         user: The user making the request. Can be None, in which case the user's profile country will not be checked.
@@ -77,60 +82,111 @@ def check_course_access(
     Returns:
         True if the user has access to the course; False otherwise
 
+    """
+    return _check_course_access(course_key, user=user, ip_addresses=ip_addresses, url=url).allowed
+
+
+class _AccessCheckResult(NamedTuple):
+    """
+    Result of `_check_course_access`.
+
+    `blocked_globally` reflects only whether the request's country matched
+    `GlobalRestrictedCountry` - it's set independently of `allowed` (a staff
+    user can have `allowed=True` and `blocked_globally=True` at once, since
+    staff bypass every block). Callers that care "was this actually denied,
+    and for which reason" should check `blocked_globally` together with
+    `not allowed`, not on its own.
+    """
+    allowed: bool
+    blocked_globally: bool
+
+
+def _check_course_access(
+        course_key: CourseKey,
+        user: Optional[types.User] = None,  # noqa: UP045
+        ip_addresses: Optional[List[str]] = None,  # noqa: UP006, UP045
+        url: Optional[str] = None,  # noqa: UP045
+) -> _AccessCheckResult:
+    """
+    Does the real work for `check_course_access`, also reporting whether a block came from
+    `GlobalRestrictedCountry` - `redirect_if_blocked` needs that to know whether a per-course
+    `disable_access_check` override may apply (it may only override a `CountryAccessRule` block,
+    never a global one).
+
+    The global check runs across every IP address and the profile country before any
+    per-course `CountryAccessRule` check is considered, so a global match is never missed
+    just because an earlier signal happened to also fail a per-course rule first.
     """
     # No-op if the country access feature is not enabled
     if not settings.EMBARGO:
-        return True
+        return _AccessCheckResult(True, False)
 
-    # First, check whether there are any restrictions on the course.
-    # If not, then we do not need to do any further checks
+    # Check whether there are any per-course or global restrictions at all.
+    # If neither applies, skip the (non-free) IP/profile country lookups below.
     course_is_restricted = RestrictedCourse.is_restricted_course(course_key)
+    globally_restricted_countries = GlobalRestrictedCountry.get_countries()
 
-    if not course_is_restricted:
-        return True
+    if not course_is_restricted and not globally_restricted_countries:
+        return _AccessCheckResult(True, False)
 
-    # Always give global and course staff access, regardless of embargo settings.
-    if user is not None and has_course_author_access(user, course_key):
-        return True
+    # Resolve each IP's country exactly once, so the global pass below and the
+    # per-course pass further down don't repeat the same GeoIP lookups.
+    ip_countries = [(ip_address, country_code_from_ip(ip_address)) for ip_address in (ip_addresses or [])]
 
-    if ip_addresses is not None:
-        # Check every IP address provided and deny access if ANY of them fail our country checks
-        for ip_address in ip_addresses:
-            # Retrieve the country code from the IP address
-            # and check it against the allowed countries list for a course
-            user_country_from_ip = country_code_from_ip(ip_address)
-
-            if not CountryAccessRule.check_country_access(course_key, user_country_from_ip):
-                log.info(
-                    (
-                        "Blocking user %s from accessing course %s at %s "
-                        "because the user's IP address %s appears to be "
-                        "located in %s."
-                    ),
-                    getattr(user, 'id', '<Not Authenticated>'),
-                    course_key,
-                    url,
-                    ip_address,
-                    user_country_from_ip
-                )
-                return False
-
-    if user is not None:
-        # Retrieve the country code from the user's profile
-        # and check it against the allowed countries list for a course.
-        user_country_from_profile = _get_user_country_from_profile(user)
-
-        if not CountryAccessRule.check_country_access(course_key, user_country_from_profile):
-            log.info(
+    # Global pass: check every IP before considering any per-course rule.
+    for ip_address, country in ip_countries:
+        if country in globally_restricted_countries:
+            return _AccessCheckResult(_deny_unless_staff(
+                user, course_key,
                 (
                     "Blocking user %s from accessing course %s at %s "
-                    "because the user's profile country is %s."
+                    "because the user's IP address %s appears to be "
+                    "located in globally restricted country %s."
                 ),
-                user.id, course_key, url, user_country_from_profile
-            )
-            return False
+                getattr(user, 'id', '<Not Authenticated>'), course_key, url, ip_address, country,
+            ), True)
+
+    # The profile country is only resolved here (not alongside `ip_countries` above),
+    # so a request already decided by the IP pass doesn't pay for it needlessly.
+    profile_country = _get_user_country_from_profile(user) if user is not None else None
+
+    if profile_country is not None and profile_country in globally_restricted_countries:
+        return _AccessCheckResult(_deny_unless_staff(
+            user, course_key,
+            (
+                "Blocking user %s from accessing course %s at %s "
+                "because the user's profile country %s is globally restricted."
+            ),
+            user.id, course_key, url, profile_country,
+        ), True)
 
-    return True
+    if not course_is_restricted:
+        return _AccessCheckResult(True, False)
+
+    # Per-course pass: only relevant once we know the request isn't globally blocked.
+    for ip_address, country in ip_countries:
+        if not CountryAccessRule.check_country_access(course_key, country):
+            return _AccessCheckResult(_deny_unless_staff(
+                user, course_key,
+                (
+                    "Blocking user %s from accessing course %s at %s "
+                    "because the user's IP address %s appears to be "
+                    "located in %s."
+                ),
+                getattr(user, 'id', '<Not Authenticated>'), course_key, url, ip_address, country,
+            ), False)
+
+    if profile_country is not None and not CountryAccessRule.check_country_access(course_key, profile_country):
+        return _AccessCheckResult(_deny_unless_staff(
+            user, course_key,
+            (
+                "Blocking user %s from accessing course %s at %s "
+                "because the user's profile country is %s."
+            ),
+            user.id, course_key, url, profile_country,
+        ), False)
+
+    return _AccessCheckResult(True, False)
 
 
 def message_url_path(course_key: CourseKey, access_point: str) -> str:
@@ -154,6 +210,25 @@ def message_url_path(course_key: CourseKey, access_point: str) -> str:
     return RestrictedCourse.message_url_path(course_key, access_point)
 
 
+def _deny_unless_staff(
+    user: Optional[types.User],  # noqa: UP045
+    course_key: CourseKey,
+    log_message: str,
+    *log_args,
+) -> bool:
+    """
+    Deny access (after logging why), unless the user is global or course staff.
+
+    Global and course staff always get access, regardless of embargo settings.
+    Callers should only invoke this once a block would otherwise occur - the
+    underlying role lookup is not free, and most requests are never blocked.
+    """
+    if user is not None and has_course_author_access(user, course_key):
+        return True
+    log.info(log_message, *log_args)
+    return False
+
+
 def _get_user_country_from_profile(user: types.User) -> str:
     """
     Check whether the user is embargoed based on the country code in the user's profile.