diff --git a/.readthedocs.yaml b/.readthedocs.yaml index fdef59eb56b2..eaa59d1f301a 100644 --- a/.readthedocs.yaml +++ b/.readthedocs.yaml @@ -10,6 +10,10 @@ sphinx: python: install: + # Need to install this to set the correct version of setuptools for now + # because it is needed by fs + # See https://github.com/openedx/openedx-platform/issues/38068 for details. + - requirements: "requirements/pip.txt" - requirements: "requirements/edx/doc.txt" - method: pip path: . diff --git a/common/djangoapps/third_party_auth/samlproviderdata/views.py b/common/djangoapps/third_party_auth/samlproviderdata/views.py index b5d044bd0498..92d409aca8a5 100644 --- a/common/djangoapps/third_party_auth/samlproviderdata/views.py +++ b/common/djangoapps/third_party_auth/samlproviderdata/views.py @@ -15,6 +15,7 @@ from rest_framework.response import Response from common.djangoapps.third_party_auth.utils import ( + SAMLMetadataURLError, convert_saml_slug_provider_id, create_or_update_bulk_saml_provider_data, fetch_metadata_xml, @@ -121,7 +122,7 @@ def sync_provider_data(self, request): # part 1: fetch information from remote metadata based on metadataUrl in samlproviderconfig try: xml = fetch_metadata_xml(metadata_url) - except (SSLError, MissingSchema, HTTPError) as ex: + except (SSLError, MissingSchema, HTTPError, SAMLMetadataURLError) as ex: msg = f'Could not verify provider metadata url. Exc type: {type(ex).__name__}' log.warning(msg) return Response(msg, status.HTTP_406_NOT_ACCEPTABLE) diff --git a/common/djangoapps/third_party_auth/tasks.py b/common/djangoapps/third_party_auth/tasks.py index 5778db4b7252..7d271e38e5d7 100644 --- a/common/djangoapps/third_party_auth/tasks.py +++ b/common/djangoapps/third_party_auth/tasks.py @@ -16,8 +16,10 @@ from common.djangoapps.third_party_auth.models import SAMLConfiguration, SAMLProviderConfig from common.djangoapps.third_party_auth.utils import ( MetadataParseError, + SAMLMetadataURLError, create_or_update_bulk_saml_provider_data, parse_metadata_xml, + validate_saml_metadata_url, ) log = logging.getLogger(__name__) @@ -74,10 +76,9 @@ def fetch_saml_metadata(): failure_messages = [] # We return the length of this array for num_failed for url, entity_ids in url_map.items(): try: + validate_saml_metadata_url(url) log.info("Fetching %s", url) - if not url.lower().startswith('https'): - log.warning("This SAML metadata URL is not secure! It should use HTTPS. (%s)", url) - response = requests.get(url, verify=True) # May raise HTTPError or SSLError or ConnectionError + response = requests.get(url, verify=True, timeout=30) # May raise HTTPError or SSLError or ConnectionError response.raise_for_status() # May raise an HTTPError try: @@ -96,13 +97,20 @@ def fetch_saml_metadata(): num_updated += 1 else: log.info(f"→ Updated existing SAMLProviderData. Nothing has changed for entityID {entity_id}") - except (exceptions.SSLError, exceptions.HTTPError, exceptions.RequestException, MetadataParseError) as error: + except ( + exceptions.SSLError, + exceptions.HTTPError, + exceptions.RequestException, + MetadataParseError, + SAMLMetadataURLError, + ) as error: # Catch and process exception in case of errors during fetching and processing saml metadata. # Here is a description of each exception. # SSLError is raised in case of errors caused by SSL (e.g. SSL cer verification failure etc.) # HTTPError is raised in case of unexpected status code (e.g. 500 error etc.) # RequestException is the base exception for any request related error that "requests" lib raises. # MetadataParseError is raised if there is error in the fetched meta data (e.g. missing @entityID etc.) + # SAMLMetadataURLError is raised if the URL fails security validation. log.exception(str(error)) failure_messages.append( diff --git a/common/djangoapps/third_party_auth/tests/test_utils.py b/common/djangoapps/third_party_auth/tests/test_utils.py index c1b3fd98b545..b08d7033dd09 100644 --- a/common/djangoapps/third_party_auth/tests/test_utils.py +++ b/common/djangoapps/third_party_auth/tests/test_utils.py @@ -6,18 +6,22 @@ from unittest.mock import MagicMock import ddt +import pytest +from django.test import override_settings from lxml import etree from common.djangoapps.student.tests.factories import UserFactory from common.djangoapps.third_party_auth.tests.testutil import TestCase from common.djangoapps.third_party_auth.utils import ( + SAMLMetadataURLError, + convert_saml_slug_provider_id, get_associated_user_by_email_response, get_user_from_email, is_enterprise_customer_user, is_oauth_provider, parse_metadata_xml, user_exists, - convert_saml_slug_provider_id, + validate_saml_metadata_url, ) from openedx.core.djangolib.testing.utils import skip_unless_lms from openedx.features.enterprise_support.tests.factories import ( @@ -216,3 +220,63 @@ def test_parse_metadata_with_use_attribute_missing(self): public_keys, sso_url, _ = parse_metadata_xml(xml, entity_id) assert public_keys == ['abc+hkIuUktxkg='] assert sso_url == 'https://idp/SSOService.php' + + +@ddt.ddt +class TestValidateSAMLMetadataURL(TestCase): + """ + Tests for validate_saml_metadata_url — the SSRF-prevention validator. + """ + + @ddt.data( + 'https://idp.example.com/metadata', + 'https://1.1.1.1/metadata', + ) + def test_valid_urls_pass(self, url): + # Should not raise + validate_saml_metadata_url(url) + + @ddt.data( + ('http://idp.example.com/metadata', 'must use HTTPS'), + ('ftp://idp.example.com/metadata', 'must use HTTPS'), + ('https://', 'no hostname'), + ) + @ddt.unpack + def test_invalid_scheme_or_missing_hostname(self, url, match): + with pytest.raises(SAMLMetadataURLError, match=match): + validate_saml_metadata_url(url) + + @ddt.data( + # Loopback + ('https://127.0.0.1/metadata', False), + ('https://127.0.0.1/metadata', True), + # Link-local (includes cloud metadata endpoints like 169.254.169.254) + ('https://169.254.169.254/metadata', False), + ('https://169.254.169.254/metadata', True), + ) + @ddt.unpack + def test_always_blocked_regardless_of_setting(self, url, allow_private): + with override_settings(SAML_METADATA_URL_ALLOW_PRIVATE_IPS=allow_private): + with pytest.raises(SAMLMetadataURLError, match='blocked address'): + validate_saml_metadata_url(url) + + @ddt.data( + 'https://10.0.0.1/metadata', + 'https://172.16.0.1/metadata', + 'https://192.168.1.1/metadata', + 'https://[fc00::1]/metadata', + ) + @override_settings(SAML_METADATA_URL_ALLOW_PRIVATE_IPS=False) + def test_private_ip_blocked_by_default(self, url): + with pytest.raises(SAMLMetadataURLError, match='private address'): + validate_saml_metadata_url(url) + + @ddt.data( + 'https://10.0.0.1/metadata', + 'https://172.16.0.1/metadata', + 'https://192.168.1.1/metadata', + ) + @override_settings(SAML_METADATA_URL_ALLOW_PRIVATE_IPS=True) + def test_private_ip_allowed_with_setting(self, url): + # Should not raise when private IPs are explicitly allowed + validate_saml_metadata_url(url) diff --git a/common/djangoapps/third_party_auth/utils.py b/common/djangoapps/third_party_auth/utils.py index 720c52ea5beb..31a6c52a3161 100644 --- a/common/djangoapps/third_party_auth/utils.py +++ b/common/djangoapps/third_party_auth/utils.py @@ -3,12 +3,15 @@ """ import datetime +import ipaddress import logging +from urllib.parse import urlparse from uuid import UUID import dateutil.parser import pytz import requests +from django.conf import settings from django.contrib.auth.models import User # lint-amnesty, pylint: disable=imported-auth-user from django.utils.timezone import now from enterprise.models import EnterpriseCustomerIdentityProvider, EnterpriseCustomerUser @@ -36,16 +39,60 @@ class MetadataParseError(Exception): pass # lint-amnesty, pylint: disable=unnecessary-pass +class SAMLMetadataURLError(Exception): + """ The SAML metadata URL failed security validation """ + pass # lint-amnesty, pylint: disable=unnecessary-pass + + +def validate_saml_metadata_url(url): + """ + Validate that a SAML metadata URL is safe to fetch. + + Enforces HTTPS and blocks requests to loopback, link-local, and reserved + IP addresses. RFC 1918 private ranges are blocked by default but can be + allowed via SAML_METADATA_URL_ALLOW_PRIVATE_IPS for deployments where the + IdP lives on the same private network as the Open edX server. + + Note: validation is IP-based and only applies when the URL contains a + literal IP address. Hostname-based URLs are not resolved here — operators + should enforce network-level egress filtering (e.g. firewall rules or a + dedicated egress proxy) as a complementary control to guard against + DNS-based bypasses. + + Raises SAMLMetadataURLError if the URL fails any check. + """ + parsed = urlparse(url) + if parsed.scheme != 'https': + raise SAMLMetadataURLError(f"SAML metadata URL must use HTTPS, got: {parsed.scheme!r}") + if not parsed.hostname: + raise SAMLMetadataURLError("SAML metadata URL has no hostname") + + try: + addr = ipaddress.ip_address(parsed.hostname) + except ValueError: + # Not a literal IP — hostname-based, allow it through + return + + if addr.is_loopback or addr.is_link_local or addr.is_reserved: + raise SAMLMetadataURLError(f"SAML metadata URL resolves to a blocked address: {addr}") + + allow_private = getattr(settings, 'SAML_METADATA_URL_ALLOW_PRIVATE_IPS', False) + if not allow_private and addr.is_private: + raise SAMLMetadataURLError( + f"SAML metadata URL resolves to a private address: {addr}. " + "Set SAML_METADATA_URL_ALLOW_PRIVATE_IPS=True to allow this." + ) + + def fetch_metadata_xml(url): """ Fetches IDP metadata from provider url Returns: xml document """ + validate_saml_metadata_url(url) try: log.info("Fetching %s", url) - if not url.lower().startswith('https'): - log.warning("This SAML metadata URL is not secure! It should use HTTPS. (%s)", url) - response = requests.get(url, verify=True) # May raise HTTPError or SSLError or ConnectionError + response = requests.get(url, verify=True, timeout=30) # May raise HTTPError or SSLError or ConnectionError response.raise_for_status() # May raise an HTTPError try: @@ -55,13 +102,15 @@ def fetch_metadata_xml(url): raise # TODO: Can use OneLogin_Saml2_Utils to validate signed XML if anyone is using that return xml - except (exceptions.SSLError, exceptions.HTTPError, exceptions.RequestException, MetadataParseError) as error: + except (exceptions.SSLError, exceptions.HTTPError, exceptions.RequestException, + MetadataParseError, SAMLMetadataURLError) as error: # Catch and process exception in case of errors during fetching and processing saml metadata. # Here is a description of each exception. # SSLError is raised in case of errors caused by SSL (e.g. SSL cer verification failure etc.) # HTTPError is raised in case of unexpected status code (e.g. 500 error etc.) # RequestException is the base exception for any request related error that "requests" lib raises. # MetadataParseError is raised if there is error in the fetched meta data (e.g. missing @entityID etc.) + # SAMLMetadataURLError is raised if the URL fails security validation. log.exception(str(error), exc_info=error) raise error except etree.XMLSyntaxError as error: diff --git a/lms/djangoapps/course_blocks/transformers/library_content.py b/lms/djangoapps/course_blocks/transformers/library_content.py index 10ef8c2138b6..3252738e1372 100644 --- a/lms/djangoapps/course_blocks/transformers/library_content.py +++ b/lms/djangoapps/course_blocks/transformers/library_content.py @@ -1,5 +1,8 @@ """ -Content Library Transformer. +Item Bank Transformer. + +Transformers for handling item bank blocks (library_content, itembank, etc.) +that use ItemBankMixin for randomized content selection. """ @@ -7,6 +10,7 @@ import logging from eventtracking import tracker +from xblock.core import XBlock from common.djangoapps.track import contexts from lms.djangoapps.courseware.models import StudentModule @@ -14,7 +18,7 @@ BlockStructureTransformer, FilteringTransformerMixin ) -from xmodule.library_content_block import LegacyLibraryContentBlock # lint-amnesty, pylint: disable=wrong-import-order +from xmodule.item_bank_block import ItemBankMixin # lint-amnesty, pylint: disable=wrong-import-order from xmodule.modulestore.django import modulestore # lint-amnesty, pylint: disable=wrong-import-order from ..utils import get_student_module_as_dict @@ -25,10 +29,13 @@ class ContentLibraryTransformer(FilteringTransformerMixin, BlockStructureTransformer): """ A transformer that manipulates the block structure by removing all - blocks within a library_content block to which a user should not - have access. + blocks within item bank blocks (library_content, itembank, etc.) + to which a user should not have access. + + This transformer works with any XBlock that inherits from ItemBankMixin, + filtering children based on the selection logic defined by each block type. - Staff users are not to be exempted from library content pathways. + Staff users are not to be exempted from item bank pathways. """ WRITE_VERSION = 1 READ_VERSION = 1 @@ -61,10 +68,10 @@ def summarize_block(usage_key): "original_usage_version": str(orig_version) if orig_version else None, } - # For each block check if block is library_content. - # If library_content add children array to content_library_children field + # For each block check if block uses ItemBankMixin (e.g., library_content, itembank). + # If so add block analytics summary for each of its children. for block_key in block_structure.topological_traversal( - filter_func=lambda block_key: block_key.block_type == 'library_content', + filter_func=lambda block_key: issubclass(XBlock.load_class(block_key.block_type), ItemBankMixin), yield_descendants_of_unyielded=True, ): xblock = block_structure.get_xblock(block_key) @@ -76,7 +83,9 @@ def transform_block_filters(self, usage_info, block_structure): all_library_children = set() all_selected_children = set() for block_key in block_structure: - if block_key.block_type != 'library_content': + block_class = XBlock.load_class(block_key.block_type) + + if block_class is None or not issubclass(block_class, ItemBankMixin): continue library_children = block_structure.get_children(block_key) if library_children: @@ -98,7 +107,12 @@ def transform_block_filters(self, usage_info, block_structure): # Update selected previous_count = len(selected) - block_keys = LegacyLibraryContentBlock.make_selection(selected, library_children, max_count) + # Get the cached block class to call make_selection + block_class = XBlock.load_class(block_key.block_type) + if block_class is None: + logger.error('Failed to load block class for %s', block_key) + continue + block_keys = block_class.make_selection(selected, library_children, max_count) selected = block_keys['selected'] # Save back any changes @@ -128,7 +142,7 @@ def check_child_removal(block_key): """ Return True if selected block should be removed. - Block is removed if it is part of library_content, but has + Block is removed if it is a child of an item bank block, but has not been selected for current user. """ if block_key not in all_library_children: @@ -156,6 +170,12 @@ def format_block_keys(keys): json_result.append(info) return json_result + # Get the cached block class to call publish_selected_children_events + block_class = XBlock.load_class(location.block_type) + if block_class is None: + logger.error('Failed to load block class for publishing events: %s', location) + return + def publish_event(event_name, result, **kwargs): """ Helper function to publish an event for analytics purposes @@ -170,11 +190,12 @@ def publish_event(event_name, result, **kwargs): context = contexts.course_context_from_course_id(location.course_key) if user_id: context['user_id'] = user_id - full_event_name = f"edx.librarycontentblock.content.{event_name}" + event_prefix = block_class.get_selected_event_prefix() + full_event_name = f"{event_prefix}.{event_name}" with tracker.get_tracker().context(full_event_name, context): tracker.emit(full_event_name, event_data) - LegacyLibraryContentBlock.publish_selected_children_events( + block_class.publish_selected_children_events( block_keys, format_block_keys, publish_event, @@ -184,12 +205,14 @@ def publish_event(event_name, result, **kwargs): class ContentLibraryOrderTransformer(BlockStructureTransformer): """ A transformer that manipulates the block structure by modifying the order of the - selected blocks within a library_content block to match the order of the selections - made by the ContentLibraryTransformer or the corresponding XBlock. So this transformer - requires the selections for the randomized content block to be already - made either by the ContentLibraryTransformer or the XBlock. + selected blocks within item bank blocks (library_content, itembank, etc.) + to match the order of the selections made by the ContentLibraryTransformer or the + corresponding XBlock. This transformer requires the selections for the item bank block + to be already made either by the ContentLibraryTransformer or the XBlock. + + This transformer works with any XBlock that inherits from ItemBankMixin. - Staff users are *not* exempted from library content pathways. + Staff users are *not* exempted from item bank pathways. """ WRITE_VERSION = 1 READ_VERSION = 1 @@ -217,7 +240,8 @@ def transform(self, usage_info, block_structure): to match the order of the selections made and stored in the XBlock 'selected' field. """ for block_key in block_structure: - if block_key.block_type != 'library_content': + block_class = XBlock.load_class(block_key.block_type) + if block_class is None or not issubclass(block_class, ItemBankMixin): continue library_children = block_structure.get_children(block_key) @@ -228,7 +252,7 @@ def transform(self, usage_info, block_structure): current_selected_blocks = {item[1] for item in state_dict.get('selected', [])} # As the selections should have already been made by the ContentLibraryTransformer, - # the current children of the library_content block should be the same as the stored + # the current children of the item bank block should be the same as the stored # selections. If they aren't, some other transformer that ran before this transformer # has modified those blocks (for example, content gating may have affected this). So do not # transform the order in that case. diff --git a/lms/djangoapps/course_blocks/transformers/tests/test_library_content.py b/lms/djangoapps/course_blocks/transformers/tests/test_library_content.py index 5a4d7a0de11a..37cf7edd5658 100644 --- a/lms/djangoapps/course_blocks/transformers/tests/test_library_content.py +++ b/lms/djangoapps/course_blocks/transformers/tests/test_library_content.py @@ -4,11 +4,13 @@ from unittest import mock +from ddt import data, ddt + +import openedx.core.djangoapps.content.block_structure.api as bs_api from common.djangoapps.student.tests.factories import CourseEnrollmentFactory from openedx.core.djangoapps.content.block_structure.api import clear_course_from_cache from openedx.core.djangoapps.content.block_structure.transformers import BlockStructureTransformers -import openedx.core.djangoapps.content.block_structure.api as bs_api from ...api import get_course_blocks from ..library_content import ContentLibraryOrderTransformer, ContentLibraryTransformer from .helpers import CourseStructureTestCase @@ -26,6 +28,7 @@ def __init__(self, state): self.state = state +@ddt class ContentLibraryTransformerTestCase(CourseStructureTestCase): """ ContentLibraryTransformer Test @@ -37,9 +40,14 @@ def setUp(self): Setup course structure and create user for content library transformer test. """ super().setUp() + self._initialize_course_hierarchy() + def _initialize_course_hierarchy(self, block_type='library_content'): + """ + Initialize course hierarchy with the given block type. + """ # Build course. - self.course_hierarchy = self.get_course_hierarchy() + self.course_hierarchy = self.get_course_hierarchy(block_type) self.blocks = self.build_course(self.course_hierarchy) self.course = self.blocks['course'] # Do this manually because publish signals are not fired by default in tests. @@ -49,14 +57,14 @@ def setUp(self): # Enroll user in course. CourseEnrollmentFactory.create(user=self.user, course_id=self.course.id, is_active=True) - def get_course_hierarchy(self): + def get_course_hierarchy(self, block_type='library_content'): """ Get a course hierarchy to test with. """ return [{ 'org': 'ContentLibraryTransformer', 'course': 'CL101F', - 'run': 'test_run', + 'run': f'test_run_{block_type}', '#type': 'course', '#ref': 'course', '#children': [ @@ -73,8 +81,8 @@ def get_course_hierarchy(self): '#ref': 'vertical1', '#children': [ { - '#type': 'library_content', - '#ref': 'library_content1', + '#type': block_type, + '#ref': f'{block_type}1', '#children': [ { 'metadata': {'display_name': "CL Vertical 2"}, @@ -111,13 +119,18 @@ def get_course_hierarchy(self): ] }] - def test_content_library(self): + @data('library_content', 'itembank') + def test_content_library(self, block_type): """ Test when course has content library section. First test user can't see any content library section, and after that mock response from MySQL db. Check user can see mocked sections in content library. """ + # Re-initialize if testing with a different block type + if block_type != 'library_content': + self._initialize_course_hierarchy(block_type) + raw_block_structure = get_course_blocks( self.user, self.course.location, @@ -136,7 +149,7 @@ def test_content_library(self): # Should dynamically assign a block to student trans_keys = set(trans_block_structure.get_block_keys()) block_key_set = self.get_block_key_set( - self.blocks, 'course', 'chapter1', 'lesson1', 'vertical1', 'library_content1' + self.blocks, 'course', 'chapter1', 'lesson1', 'vertical1', f'{block_type}1' ) for key in block_key_set: assert key in trans_keys @@ -160,11 +173,12 @@ def test_content_library(self): assert set(trans_block_structure.get_block_keys()) == self.get_block_key_set(self.blocks, 'course', 'chapter1', 'lesson1', 'vertical1', - 'library_content1', + f'{block_type}1', selected_vertical, selected_child), f"Expected 'selected' equality failed in iteration {i}." # pylint: disable=line-too-long +@ddt class ContentLibraryOrderTransformerTestCase(CourseStructureTestCase): """ ContentLibraryOrderTransformer Test @@ -176,7 +190,13 @@ def setUp(self): Setup course structure and create user for content library order transformer test. """ super().setUp() - self.course_hierarchy = self.get_course_hierarchy() + self._initialize_course_hierarchy() + + def _initialize_course_hierarchy(self, block_type='library_content'): + """ + Initialize course hierarchy with the given block type. + """ + self.course_hierarchy = self.get_course_hierarchy(block_type) self.blocks = self.build_course(self.course_hierarchy) self.course = self.blocks['course'] bs_api.update_course_in_cache(self.course.id) @@ -185,14 +205,14 @@ def setUp(self): # Enroll user in course. CourseEnrollmentFactory.create(user=self.user, course_id=self.course.id, is_active=True) - def get_course_hierarchy(self): + def get_course_hierarchy(self, block_type='library_content'): """ Get a course hierarchy to test with. """ return [{ 'org': 'ContentLibraryTransformer', 'course': 'CL101F', - 'run': 'test_run', + 'run': f'test_run_{block_type}', '#type': 'course', '#ref': 'course', '#children': [ @@ -209,8 +229,8 @@ def get_course_hierarchy(self): '#ref': 'vertical1', '#children': [ { - '#type': 'library_content', - '#ref': 'library_content1', + '#type': block_type, + '#ref': f'{block_type}1', '#children': [ { 'metadata': {'display_name': "CL Vertical 2"}, @@ -260,11 +280,15 @@ def get_course_hierarchy(self): }] @mock.patch('lms.djangoapps.course_blocks.transformers.library_content.get_student_module_as_dict') - def test_content_library_randomize(self, mocked): + @data('library_content', 'itembank') + def test_content_library_randomize(self, block_type, mocked): """ Test whether the order of the children blocks matches the order of the selected blocks when course has content library section """ + # Re-initialize if testing with a different block type + if block_type != 'library_content': + self._initialize_course_hierarchy(block_type) mocked.return_value = { 'selected': [ ['vertical', 'vertical_vertical3'], @@ -280,7 +304,7 @@ def test_content_library_randomize(self, mocked): ) children = [] for block_key in trans_block_structure.topological_traversal(): - if block_key.block_type == 'library_content': + if block_key.block_type == block_type: children = trans_block_structure.get_children(block_key) break diff --git a/lms/djangoapps/discussion/rest_api/discussions_notifications.py b/lms/djangoapps/discussion/rest_api/discussions_notifications.py index 8efb2e8acb83..531090732781 100644 --- a/lms/djangoapps/discussion/rest_api/discussions_notifications.py +++ b/lms/djangoapps/discussion/rest_api/discussions_notifications.py @@ -452,6 +452,14 @@ def clean_thread_html_body(html_body): truncated_body = html.unescape(truncated_body) html_body = BeautifulSoup(truncated_body, 'html.parser') + # Remove tags including their content (decompose, not unwrap) + tags_to_decompose = [ + "style", # CSS injection + ] + for tag in tags_to_decompose: + for match in html_body.find_all(tag): + match.decompose() + tags_to_remove = [ "a", "link", # Link Tags "img", "picture", "source", # Image Tags diff --git a/lms/djangoapps/discussion/rest_api/tests/test_discussions_notifications.py b/lms/djangoapps/discussion/rest_api/tests/test_discussions_notifications.py index aaa920a0ff78..5b2bc471928e 100644 --- a/lms/djangoapps/discussion/rest_api/tests/test_discussions_notifications.py +++ b/lms/djangoapps/discussion/rest_api/tests/test_discussions_notifications.py @@ -214,4 +214,17 @@ def test_strip_empty_tags(self): """ html_body = '

content

' result = clean_thread_html_body(html_body) - self.assertEqual(result, '

content

') + self.assertEqual(result, '

content

') # noqa: PT009 + + def test_style_tag_removed_with_content(self): + """ + Test that style tags and their CSS content are fully removed (CSS injection prevention). + """ + html_body = ( + '

Hello

' + '' + ) + result = clean_thread_html_body(html_body) + self.assertNotIn('