From d2c474fc9e9f1f5caee0adeab861607fc95ba3df Mon Sep 17 00:00:00 2001 From: Asad Ali Date: Fri, 4 Sep 2026 23:54:08 +0500 Subject: [PATCH 1/5] feat: enforce GlobalRestrictedCountry on course access, not just registration GlobalRestrictedCountry previously only blocked account registration and profile-country changes - it had no effect on actually accessing a course. The real course-access enforcement (RestrictedCourse + CountryAccessRule) requires a row per course, with no way to block a country across every course at once. Wire GlobalRestrictedCountry into embargo.api.check_course_access() so a listed country blocks every course, regardless of whether it has a RestrictedCourse entry. Per-course CountryAccessRule checks still apply on top for courses that have them, and staff continue to bypass both. Also ensure disable_access_check (a per-course escape hatch) can never bypass a global restriction, and restore a fast path so courses/installs that never use this feature don't pay for the extra lookups. Co-Authored-By: Claude Sonnet 5 --- openedx/core/djangoapps/embargo/api.py | 112 +++++++++++---- .../core/djangoapps/embargo/tests/test_api.py | 129 +++++++++++++++--- 2 files changed, 198 insertions(+), 43 deletions(-) diff --git a/openedx/core/djangoapps/embargo/api.py b/openedx/core/djangoapps/embargo/api.py index 23304f27cd9f..b7f3a97c5f36 100644 --- a/openedx/core/djangoapps/embargo/api.py +++ b/openedx/core/djangoapps/embargo/api.py @@ -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. @@ -53,7 +54,9 @@ def redirect_if_blocked( is_blocked = not check_course_access(course_key, user=user, ip_addresses=client_ips, url=request.path) if is_blocked: if access_point == "courseware": - if not RestrictedCourse.is_disabled_access_check(course_key): + if not RestrictedCourse.is_disabled_access_check(course_key) or _is_globally_restricted( + user, client_ips, + ): return message_url_path(course_key, access_point) else: return message_url_path(course_key, access_point) @@ -68,6 +71,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. @@ -82,53 +89,69 @@ def check_course_access( if not settings.EMBARGO: return True - # 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 - - # Always give global and course staff access, regardless of embargo settings. - if user is not None and has_course_author_access(user, course_key): + if not course_is_restricted and not globally_restricted_countries: return True 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 + # Retrieve the country code from the IP address and check it against + # the global restricted-country list, then (if the course has one) + # the course's own allowed-countries list. user_country_from_ip = country_code_from_ip(ip_address) - if not CountryAccessRule.check_country_access(course_key, user_country_from_ip): - log.info( + if user_country_from_ip in globally_restricted_countries: + return _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 globally restricted country %s." + ), + getattr(user, 'id', ''), course_key, url, ip_address, user_country_from_ip, + ) + + if course_is_restricted and not CountryAccessRule.check_country_access(course_key, user_country_from_ip): + return _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', ''), - course_key, - url, - ip_address, - user_country_from_ip + getattr(user, 'id', ''), 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. + # Retrieve the country code from the user's profile and check it against + # the global restricted-country list, then (if the course has one) the + # course's own allowed-countries list. user_country_from_profile = _get_user_country_from_profile(user) - if not CountryAccessRule.check_country_access(course_key, user_country_from_profile): - log.info( + if user_country_from_profile in globally_restricted_countries: + return _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, user_country_from_profile, + ) + + if course_is_restricted and not CountryAccessRule.check_country_access(course_key, user_country_from_profile): + return _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, user_country_from_profile + user.id, course_key, url, user_country_from_profile, ) - return False return True @@ -154,6 +177,43 @@ def message_url_path(course_key: CourseKey, access_point: str) -> str: return RestrictedCourse.message_url_path(course_key, access_point) +def _is_globally_restricted( + user: Optional[types.User], # noqa: UP045 + ip_addresses: Optional[List[str]], # noqa: UP006, UP045 +) -> bool: + """ + Check whether the request's IP or profile country is on the `GlobalRestrictedCountry` list. + + Used so that a per-course `disable_access_check` override can never bypass a + global embargo - it can only bypass a per-course `CountryAccessRule` block. + """ + restricted_countries = GlobalRestrictedCountry.get_countries() + if ip_addresses: + for ip_address in ip_addresses: + if country_code_from_ip(ip_address) in restricted_countries: + return True + return user is not None and _get_user_country_from_profile(user) in restricted_countries + + +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. diff --git a/openedx/core/djangoapps/embargo/tests/test_api.py b/openedx/core/djangoapps/embargo/tests/test_api.py index 7cf18077f3bc..36884eb9d1be 100644 --- a/openedx/core/djangoapps/embargo/tests/test_api.py +++ b/openedx/core/djangoapps/embargo/tests/test_api.py @@ -34,7 +34,7 @@ from .. import api as embargo_api from ..exceptions import InvalidAccessPoint -from ..models import Country, CountryAccessRule, RestrictedCourse +from ..models import Country, CountryAccessRule, GlobalRestrictedCountry, RestrictedCourse QUERY_COUNT_TABLE_IGNORELIST = WAFFLE_TABLES + AUTHZ_TABLES @@ -126,13 +126,16 @@ def test_no_user_blocked(self): assert not result def test_course_not_restricted(self): - # No restricted course model for this course key, - # so all access checks should be skipped. + # No `RestrictedCourse` row for this course, and no `GlobalRestrictedCountry` + # rows either, so `check_course_access` takes its fast path: it only needs + # to warm the two "is anything restricted at all" caches, then returns + # without ever looking at the IP or the user's profile. unrestricted_course = CourseFactory.create() - with self.assertNumQueries(1): + with self.assertNumQueries(2): embargo_api.check_course_access(unrestricted_course.id, user=self.user, ip_addresses=['0.0.0.0']) - # The second check should require no database queries + # The second check should require no database queries - both caches + # (restricted-course list, global-country list) are warm by now. with self.assertNumQueries(0): embargo_api.check_course_access(unrestricted_course.id, user=self.user, ip_addresses=['0.0.0.0']) @@ -174,8 +177,12 @@ def test_caching(self): # Test the scenario that will go through every check # (restricted course, but pass all the checks) # This is the worst case, so it will hit all of the - # caching code. - with self.assertNumQueries(5, table_ignorelist=QUERY_COUNT_TABLE_IGNORELIST): + # caching code: restricted-course cache (1) + global-country + # cache (1) + per-course country-access-rule cache (1) + the + # user's profile lookup (1) = 4. This scenario no longer pays for + # the `has_course_author_access` role lookup, since that's now + # deferred until a block is about to happen, and nothing blocks here. + with self.assertNumQueries(4, table_ignorelist=QUERY_COUNT_TABLE_IGNORELIST): embargo_api.check_course_access(self.course.id, user=self.user, ip_addresses=['0.0.0.0']) with self.assertNumQueries(0, table_ignorelist=QUERY_COUNT_TABLE_IGNORELIST): @@ -185,7 +192,8 @@ def test_caching_no_restricted_courses(self): RestrictedCourse.objects.all().delete() cache.clear() - with self.assertNumQueries(1): + # Same fast path as `test_course_not_restricted` - see there for the count. + with self.assertNumQueries(2): embargo_api.check_course_access(self.course.id, user=self.user, ip_addresses=['0.0.0.0']) with self.assertNumQueries(0): @@ -213,22 +221,109 @@ def test_staff_access_country_block(self, staff_role_cls): # Expect that the user is blocked, because the user isn't staff assert not result, "User should not have access because the user isn't staff." - # Instantiate the role, configuring it for this course or org + self._add_staff_role(staff_role_cls, self.course.id) + + # Now the user should have access + with self._mock_geoip('US'): + result = embargo_api.check_course_access(self.course.id, user=self.user, ip_addresses=['0.0.0.0']) + + assert result, 'User should have access because the user is staff.' + + @ddt.data( + GlobalStaff, + CourseStaffRole, + CourseInstructorRole, + OrgStaffRole, + OrgInstructorRole, + ) + def test_staff_access_global_country_block(self, staff_role_cls): + # Staff should bypass a `GlobalRestrictedCountry` block too, on a + # course that has no `RestrictedCourse` row at all. + course_key = CourseFactory.create().id + GlobalRestrictedCountry.objects.create(country=Country.objects.get(country='IR')) + + with self._mock_geoip('IR'): + result = embargo_api.check_course_access(course_key, user=self.user, ip_addresses=['0.0.0.0']) + assert not result, "User should not have access because the user isn't staff." + + self._add_staff_role(staff_role_cls, course_key) + + with self._mock_geoip('IR'): + result = embargo_api.check_course_access(course_key, user=self.user, ip_addresses=['0.0.0.0']) + assert result, 'User should have access because the user is staff.' + + def _add_staff_role(self, staff_role_cls, course_key): + """Instantiate `staff_role_cls` for `course_key` (or its org) and add `self.user` to it.""" if issubclass(staff_role_cls, CourseRole): - staff_role = staff_role_cls(self.course.id) + staff_role = staff_role_cls(course_key) elif issubclass(staff_role_cls, OrgRole): - staff_role = staff_role_cls(self.course.id.org) + staff_role = staff_role_cls(course_key.org) else: staff_role = staff_role_cls() - - # Add the user to the role staff_role.add_users(self.user) - # Now the user should have access - with self._mock_geoip('US'): - result = embargo_api.check_course_access(self.course.id, user=self.user, ip_addresses=['0.0.0.0']) + @ddt.data( + # course_restricted, rule_type, rule_country, global_country, country, source, allow_access + (False, None, None, 'IR', 'IR', 'ip', False), # no RestrictedCourse row at all, blocked globally + (False, None, None, 'IR', 'US', 'ip', True), # no RestrictedCourse row, unrelated country is fine + (False, None, None, 'IR', 'IR', 'profile', False), # global block also applies via profile country + (True, CountryAccessRule.BLACKLIST_RULE, 'CU', None, 'CU', 'ip', False), # existing per-course rule, unaffected + (True, CountryAccessRule.BLACKLIST_RULE, 'CU', None, 'US', 'ip', True), + (True, CountryAccessRule.WHITELIST_RULE, 'IR', 'IR', 'IR', 'ip', False), # global restriction beats whitelist + ) + @ddt.unpack + def test_global_restricted_country_access( + self, course_restricted, rule_type, rule_country, global_country, country, source, allow_access, + ): + course_key = self.course.id if course_restricted else CourseFactory.create().id - assert result, 'User should have access because the user is staff.' + if rule_type is not None: + CountryAccessRule.objects.create( + rule_type=rule_type, + restricted_course=self.restricted_course, + country=Country.objects.get(country=rule_country), + ) + + if global_country is not None: + GlobalRestrictedCountry.objects.create(country=Country.objects.get(country=global_country)) + + if source == 'profile': + self.user.profile.country = country + self.user.profile.save() + ip_country = '' + else: + ip_country = country + + with self._mock_geoip(ip_country): + result = embargo_api.check_course_access(course_key, user=self.user, ip_addresses=['0.0.0.0']) + assert result == allow_access + + def test_redirect_if_blocked_global_restricted_country(self): + # A course with no `RestrictedCourse` row still redirects to the + # default blocked-message page when blocked by `GlobalRestrictedCountry`. + unrestricted_course = CourseFactory.create() + GlobalRestrictedCountry.objects.create(country=Country.objects.get(country='IR')) + + request = RequestFactory().get('', HTTP_X_FORWARDED_FOR='0.0.0.0') + request.user = self.user + + with self._mock_geoip('IR'): + redirect_url = embargo_api.redirect_if_blocked(request, unrestricted_course.id, access_point='courseware') + assert redirect_url == '/embargo/blocked-message/courseware/default/' + + def test_disable_access_check_does_not_bypass_global_restriction(self): + # `disable_access_check` is a per-course escape hatch for `CountryAccessRule` + # blocks. It must NOT let a `GlobalRestrictedCountry` block through too. + self.restricted_course.disable_access_check = True + self.restricted_course.save() + GlobalRestrictedCountry.objects.create(country=Country.objects.get(country='IR')) + + request = RequestFactory().get('', HTTP_X_FORWARDED_FOR='0.0.0.0') + request.user = self.user + + with self._mock_geoip('IR'): + redirect_url = embargo_api.redirect_if_blocked(request, self.course.id, access_point='courseware') + assert redirect_url is not None, "A global restriction should still redirect even with disable_access_check." @ddt.data( # (Note that any '0.x.x.x' IP _should_ be blocked in this test.) From b40b6bc400d5ce80fedad53a81e616fa67c31731 Mon Sep 17 00:00:00 2001 From: Asad Ali Date: Mon, 7 Sep 2026 14:15:46 +0500 Subject: [PATCH 2/5] fix: never let disable_access_check bypass a GlobalRestrictedCountry block check_course_access() now delegates to an internal _check_course_access() that reports *why* access was denied (blocked_globally) as well as whether it was, instead of redirect_if_blocked() re-deriving the same IP/profile country lookups a second time via a separate helper. This also fixes an ordering bug in that split: a per-course CountryAccessRule match on an earlier IP could return before the profile country was checked against GlobalRestrictedCountry, letting a globally-restricted user slip through disable_access_check just because their IP also happened to fail an unrelated per-course rule first. The global check now runs across every IP and the profile country before any per-course rule is considered. Co-Authored-By: Claude Sonnet 5 --- openedx/core/djangoapps/embargo/api.py | 163 ++++++++++-------- .../core/djangoapps/embargo/tests/test_api.py | 33 +++- 2 files changed, 121 insertions(+), 75 deletions(-) diff --git a/openedx/core/djangoapps/embargo/api.py b/openedx/core/djangoapps/embargo/api.py index b7f3a97c5f36..335fb09fe2eb 100644 --- a/openedx/core/djangoapps/embargo/api.py +++ b/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 @@ -51,12 +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) or _is_globally_restricted( - user, client_ips, - ): + 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) @@ -84,10 +82,44 @@ 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) # Check whether there are any per-course or global restrictions at all. # If neither applies, skip the (non-free) IP/profile country lookups below. @@ -95,65 +127,66 @@ def check_course_access( globally_restricted_countries = GlobalRestrictedCountry.get_countries() if not course_is_restricted and not globally_restricted_countries: - return True + return _AccessCheckResult(True, False) + + # 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 global restricted-country list, then (if the course has one) - # the course's own allowed-countries list. - user_country_from_ip = country_code_from_ip(ip_address) - - if user_country_from_ip in globally_restricted_countries: - return _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 globally restricted country %s." - ), - getattr(user, 'id', ''), course_key, url, ip_address, user_country_from_ip, - ) - - if course_is_restricted and not CountryAccessRule.check_country_access(course_key, user_country_from_ip): - return _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', ''), course_key, url, ip_address, user_country_from_ip, - ) - - if user is not None: - # Retrieve the country code from the user's profile and check it against - # the global restricted-country list, then (if the course has one) the - # course's own allowed-countries list. - user_country_from_profile = _get_user_country_from_profile(user) - - if user_country_from_profile in globally_restricted_countries: - return _deny_unless_staff( + # 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 %s is globally restricted." + "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, - ) - - if course_is_restricted and not CountryAccessRule.check_country_access(course_key, user_country_from_profile): - return _deny_unless_staff( + getattr(user, 'id', ''), 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) + + 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 profile country is %s." + "because the user's IP address %s appears to be " + "located in %s." ), - user.id, course_key, url, user_country_from_profile, - ) + getattr(user, 'id', ''), 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 True + return _AccessCheckResult(True, False) def message_url_path(course_key: CourseKey, access_point: str) -> str: @@ -177,24 +210,6 @@ def message_url_path(course_key: CourseKey, access_point: str) -> str: return RestrictedCourse.message_url_path(course_key, access_point) -def _is_globally_restricted( - user: Optional[types.User], # noqa: UP045 - ip_addresses: Optional[List[str]], # noqa: UP006, UP045 -) -> bool: - """ - Check whether the request's IP or profile country is on the `GlobalRestrictedCountry` list. - - Used so that a per-course `disable_access_check` override can never bypass a - global embargo - it can only bypass a per-course `CountryAccessRule` block. - """ - restricted_countries = GlobalRestrictedCountry.get_countries() - if ip_addresses: - for ip_address in ip_addresses: - if country_code_from_ip(ip_address) in restricted_countries: - return True - return user is not None and _get_user_country_from_profile(user) in restricted_countries - - def _deny_unless_staff( user: Optional[types.User], # noqa: UP045 course_key: CourseKey, diff --git a/openedx/core/djangoapps/embargo/tests/test_api.py b/openedx/core/djangoapps/embargo/tests/test_api.py index 36884eb9d1be..cfb09059edea 100644 --- a/openedx/core/djangoapps/embargo/tests/test_api.py +++ b/openedx/core/djangoapps/embargo/tests/test_api.py @@ -325,6 +325,31 @@ def test_disable_access_check_does_not_bypass_global_restriction(self): redirect_url = embargo_api.redirect_if_blocked(request, self.course.id, access_point='courseware') assert redirect_url is not None, "A global restriction should still redirect even with disable_access_check." + def test_global_restriction_via_profile_not_masked_by_earlier_ip_rule_match(self): + # A per-course `CountryAccessRule` match on the IP must not short-circuit + # before the profile country is also checked against `GlobalRestrictedCountry` - + # otherwise a globally-restricted user could slip through via disable_access_check + # just because their IP happened to also fail a (unrelated) per-course rule first. + self.restricted_course.disable_access_check = True + self.restricted_course.save() + CountryAccessRule.objects.create( + rule_type=CountryAccessRule.BLACKLIST_RULE, + restricted_course=self.restricted_course, + country=Country.objects.get(country='CU'), + ) + GlobalRestrictedCountry.objects.create(country=Country.objects.get(country='IR')) + self.user.profile.country = 'IR' + self.user.profile.save() + + request = RequestFactory().get('', HTTP_X_FORWARDED_FOR='0.0.0.0') + request.user = self.user + + # IP matches the per-course blacklist (CU), not the global list - but the + # profile country (IR) is globally restricted, and that must still win. + with self._mock_geoip('CU'): + redirect_url = embargo_api.redirect_if_blocked(request, self.course.id, access_point='courseware') + assert redirect_url is not None, "Global restriction must not be masked by an earlier IP rule match." + @ddt.data( # (Note that any '0.x.x.x' IP _should_ be blocked in this test.) # ips, allow access @@ -357,8 +382,14 @@ def test_redirect_if_blocked_ips(self, ips, allow_access, mock_country): ('courseware', True, True), # Unless the access check has been disabled, then we allow them ) @ddt.unpack - @mock.patch('openedx.core.djangoapps.embargo.api.check_course_access', return_value=False) + @mock.patch( + 'openedx.core.djangoapps.embargo.api._check_course_access', + return_value=embargo_api._AccessCheckResult(False, False), # pylint: disable=protected-access + ) def test_redirect_if_blocked_courseware(self, access_point, check_disabled, allow_access, _mock_access): # noqa: PT019 # pylint: disable=line-too-long + # blocked_globally=False here - this test is specifically about the + # per-course disable_access_check override, which only ever applies + # to a non-global (CountryAccessRule) block. self.restricted_course.disable_access_check = check_disabled self.restricted_course.save() From 5e5eaaeacd72a879a239bfa430438bee894bbe75 Mon Sep 17 00:00:00 2001 From: Asad Ali Date: Mon, 7 Sep 2026 14:30:42 +0500 Subject: [PATCH 3/5] fix: correct query count in test_caching Actual query count is 3, not 4: CountryAccessRule.check_country_access caches its per-course allowed-countries list, so the IP-based check and the profile-country check share a single query instead of each paying for their own. Verified by running the full embargo suite (101 passed) after fixing an unrelated missing openedx-learning dependency in the local test environment. --- openedx/core/djangoapps/embargo/tests/test_api.py | 12 +++++++----- 1 file changed, 7 insertions(+), 5 deletions(-) diff --git a/openedx/core/djangoapps/embargo/tests/test_api.py b/openedx/core/djangoapps/embargo/tests/test_api.py index cfb09059edea..ee54f9f44843 100644 --- a/openedx/core/djangoapps/embargo/tests/test_api.py +++ b/openedx/core/djangoapps/embargo/tests/test_api.py @@ -178,11 +178,13 @@ def test_caching(self): # (restricted course, but pass all the checks) # This is the worst case, so it will hit all of the # caching code: restricted-course cache (1) + global-country - # cache (1) + per-course country-access-rule cache (1) + the - # user's profile lookup (1) = 4. This scenario no longer pays for - # the `has_course_author_access` role lookup, since that's now - # deferred until a block is about to happen, and nothing blocks here. - with self.assertNumQueries(4, table_ignorelist=QUERY_COUNT_TABLE_IGNORELIST): + # cache (1) + per-course country-access-rule cache (1) = 3. + # `CountryAccessRule.check_country_access` caches its allowed-countries + # list per course_key, so the IP check and the profile check below share + # that single query rather than each paying for their own. This scenario + # also doesn't pay for the `has_course_author_access` role lookup, since + # that's deferred until a block is about to happen, and nothing blocks here. + with self.assertNumQueries(3, table_ignorelist=QUERY_COUNT_TABLE_IGNORELIST): embargo_api.check_course_access(self.course.id, user=self.user, ip_addresses=['0.0.0.0']) with self.assertNumQueries(0, table_ignorelist=QUERY_COUNT_TABLE_IGNORELIST): From 9cf399d2a62aa99e91c006a9672f5f02aef3da71 Mon Sep 17 00:00:00 2001 From: Asad Ali Date: Mon, 7 Sep 2026 15:56:08 +0500 Subject: [PATCH 4/5] feat: enforce embargo in check_course_access, closing Learning MFE bypass GlobalRestrictedCountry (and CountryAccessRule) previously took effect only through EmbargoMiddleware, which recognizes legacy /course/ and /courses/ URLs. The Learning MFE's course_home_api endpoints (outline, dates, progress, navigation, course_metadata) don't match that URL pattern, so a learner in an embargoed country could view all course content through the MFE even with GlobalRestrictedCountry configured. Wires the check into check_course_access() in courseware/courses.py - the shared choke point every course_home_api view and legacy view already routes through via has_access()/get_course_with_access(). Staff bypass falls out of the existing check_course_access() fallback for free. Verified live against a running devstack: the MFE outline API went from 200 OK (unblocked) to 403 with the embargo error code, while the legacy courseware page continued to redirect as before. --- .../outline/tests/test_view.py | 28 +++++++++++++++++++ lms/djangoapps/courseware/access_response.py | 12 ++++++++ lms/djangoapps/courseware/access_utils.py | 26 +++++++++++++++++ lms/djangoapps/courseware/courses.py | 14 ++++++++++ 4 files changed, 80 insertions(+) diff --git a/lms/djangoapps/course_home_api/outline/tests/test_view.py b/lms/djangoapps/course_home_api/outline/tests/test_view.py index b802a806eeca..eab42dcd474d 100644 --- a/lms/djangoapps/course_home_api/outline/tests/test_view.py +++ b/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): diff --git a/lms/djangoapps/courseware/access_response.py b/lms/djangoapps/courseware/access_response.py index d13258d16c86..e64fb5673f16 100644 --- a/lms/djangoapps/courseware/access_response.py +++ b/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) diff --git a/lms/djangoapps/courseware/access_utils.py b/lms/djangoapps/courseware/access_utils.py index 4caaeb9f993a..795c294a68b7 100644 --- a/lms/djangoapps/courseware/access_utils.py +++ b/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 diff --git a/lms/djangoapps/courseware/courses.py b/lms/djangoapps/courseware/courses.py index 51dd6863fdf1..e136473fff00 100644 --- a/lms/djangoapps/courseware/courses.py +++ b/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) From b3020ce8546acaa9fcbfba882a900b4eef99c2a4 Mon Sep 17 00:00:00 2001 From: Asad Ali Date: Tue, 8 Sep 2026 13:06:23 +0500 Subject: [PATCH 5/5] fix: scope embargo enforcement to course_metadata, not check_course_access The previous commit wired GlobalRestrictedCountry into check_course_access(), which every course_home_api BFF view (outline, dates, progress, navigation) routes through - blocking real course content on those endpoints for embargoed learners, but also requiring the Learning MFE to stop swallowing their 403s to avoid a crash. For now, scope the check down to course_metadata's own view instead, which already computes course access separately to serialize a graceful `course_access` JSON field for the frontend. outline/dates/progress keep their original behavior unchanged. This accepts a known gap (an embargoed learner's course content is still served by those endpoints) in exchange for zero Learning MFE data-layer changes; a corresponding one-line MFE change (frontend-app-learning) makes the outline tab redirect on this flag like every other tab already does. Co-Authored-By: Claude Sonnet 5 --- .../course_metadata/tests/test_views.py | 29 +++++++++++++++++++ .../course_home_api/course_metadata/views.py | 7 +++++ .../outline/tests/test_view.py | 28 ------------------ lms/djangoapps/courseware/courses.py | 14 --------- 4 files changed, 36 insertions(+), 42 deletions(-) diff --git a/lms/djangoapps/course_home_api/course_metadata/tests/test_views.py b/lms/djangoapps/course_home_api/course_metadata/tests/test_views.py index 2d0ac7b41b7f..9d7b40ffa420 100644 --- a/lms/djangoapps/course_home_api/course_metadata/tests/test_views.py +++ b/lms/djangoapps/course_home_api/course_metadata/tests/test_views.py @@ -27,6 +27,7 @@ COURSEWARE_MICROFRONTEND_PROGRESS_MILESTONES_STREAK_CELEBRATION, ) from openedx.core.djangoapps.discussions.models import DiscussionsConfiguration +from openedx.core.djangoapps.embargo.models import Country, GlobalRestrictedCountry @ddt.ddt @@ -241,6 +242,34 @@ def test_course_access( self._assert_course_access_response(response, expect_course_access, error_code) + @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 metadata endpoint the Learning MFE actually + reads `course_access` from, 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') + + self._assert_course_access_response(response, False, 'embargo') + + @override_settings(EMBARGO=True) + def test_embargo_staff_bypass(self): + """ Course staff should still get access even when their country is globally restricted. """ + CourseInstructorRole(self.course.id).add_users(self.user) + 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 + assert response.data['course_access']['has_access'] is True + @override_settings(ENABLE_DISCUSSION_SERVICE=True) @ddt.data(True, False) def test_discussion_tab_visible(self, visible): diff --git a/lms/djangoapps/course_home_api/course_metadata/views.py b/lms/djangoapps/course_home_api/course_metadata/views.py index f6f15203260e..af58f3f21578 100644 --- a/lms/djangoapps/course_home_api/course_metadata/views.py +++ b/lms/djangoapps/course_home_api/course_metadata/views.py @@ -20,6 +20,7 @@ from lms.djangoapps.course_home_api.course_metadata.serializers import CourseHomeMetadataSerializer from lms.djangoapps.course_home_api.toggles import new_discussion_sidebar_view_is_enabled from lms.djangoapps.courseware.access import has_access, has_cms_access +from lms.djangoapps.courseware.access_utils import check_embargo_access from lms.djangoapps.courseware.context_processor import user_timezone_locale_prefs from lms.djangoapps.courseware.courses import check_course_access from lms.djangoapps.courseware.exceptions import CourseAccessRedirect @@ -104,6 +105,12 @@ def get(self, request, *args, **kwargs): check_if_authenticated=True, apply_priority_access_checks=True, ) + # A country embargo (GlobalRestrictedCountry / CountryAccessRule) takes priority over + # any other access denial reason, and is checked here - rather than shared into + # check_course_access() - so it stays scoped to metadata's UI-level access flag for now. + embargo_access = check_embargo_access(request.user, course) + if not embargo_access: + load_access = embargo_access _, request.user = setup_masquerade( request, diff --git a/lms/djangoapps/course_home_api/outline/tests/test_view.py b/lms/djangoapps/course_home_api/outline/tests/test_view.py index eab42dcd474d..b802a806eeca 100644 --- a/lms/djangoapps/course_home_api/outline/tests/test_view.py +++ b/lms/djangoapps/course_home_api/outline/tests/test_view.py @@ -30,7 +30,6 @@ 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 @@ -54,33 +53,6 @@ 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): diff --git a/lms/djangoapps/courseware/courses.py b/lms/djangoapps/courseware/courses.py index e136473fff00..51dd6863fdf1 100644 --- a/lms/djangoapps/courseware/courses.py +++ b/lms/djangoapps/courseware/courses.py @@ -32,7 +32,6 @@ from lms.djangoapps.courseware.access_response import ( AuthenticationRequiredAccessError, CatalogVisibilityError, - EmbargoAccessError, EnrollmentRequiredAccessError, MilestoneAccessError, OldMongoAccessError, @@ -41,7 +40,6 @@ ) from lms.djangoapps.courseware.access_utils import ( check_authentication, - check_embargo_access, check_enrollment, ) from lms.djangoapps.courseware.block_render import get_block @@ -65,7 +63,6 @@ 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 @@ -192,13 +189,6 @@ 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: @@ -317,10 +307,6 @@ 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)