[Backport 5.1.x] Xml upload download improvements

geonode/assets/local.py

@@ -22,6 +22,9 @@
     concrete_storage_manager=FileSystemStorageManager(location=os.path.dirname(settings.ASSETS_ROOT))
 )
 
+# extensions served as an attachment (download) instead of rendered inline
+FORCE_DOWNLOAD_EXTENSIONS = {"xml", "sld"}
+
 
 class DefaultLocalLinkUrlHandler:
     def get_link_url(self, asset: LocalAsset):
@@ -303,9 +306,17 @@ def create_response(
                     )
                 case False:
                     logger.info(f"Returning file '{localfile}' with name '{outname}'")
-                    return DownloadResponse(
-                        _asset_storage_manager.open(localfile).file, basename=f"{outname}", attachment=False
+                    force_download = ext.lower().lstrip(".") in FORCE_DOWNLOAD_EXTENSIONS
+                    response = DownloadResponse(
+                        _asset_storage_manager.open(localfile).file,
+                        basename=f"{outname}",
+                        attachment=force_download,
+                    )
+                    response.headers["X-Content-Type-Options"] = "nosniff"
+                    response.headers["Content-Security-Policy"] = (
+                        "default-src 'none'; style-src 'unsafe-inline'; sandbox"
                     )
+                    return response
         else:
             logger.warning(f"Internal file {localfile} not found for asset {asset.id}")
             return HttpResponse(f"Internal file not found for asset {asset.id}", status=404 if path else 500)

geonode/base/api/views.py

@@ -112,6 +112,7 @@
 from geonode.assets.utils import create_asset_and_link, unlink_asset
 from geonode.assets.handlers import asset_handler_registry
 from geonode.utils import get_supported_datasets_file_types
+from geonode.utils import assert_safe_xml, UnsafeXMLError
 
 logger = logging.getLogger(__name__)
 
@@ -1356,6 +1357,17 @@ def asset(self, request, pk=None, *args, **kwargs):
                     {"message": f"The uploaded file type {file_ext} is not allowed."},
                     status=status.HTTP_400_BAD_REQUEST,
                 )
+            if file_ext in ("xml", "sld"):
+                try:
+                    assert_safe_xml(file.read())
+                except UnsafeXMLError:
+                    logger.warning("XML validation failed for uploaded asset.", exc_info=True)
+                    return Response(
+                        {"message": f"The uploaded {file_ext} file is invalid or unsafe."},
+                        status=status.HTTP_400_BAD_REQUEST,
+                    )
+                finally:
+                    file.seek(0)
         try:
             handler = asset_handler_registry.get_default_handler()
             asset, link = create_asset_and_link(

geonode/documents/forms.py

@@ -31,9 +31,12 @@
 from geonode.upload.models import UploadSizeLimit
 from geonode.upload.api.exceptions import FileUploadLimitException
 from geonode.upload.zip_validation import ZipValidationError, is_zip_extension, validate_safe_zip
+from geonode.utils import assert_safe_xml, UnsafeXMLError
 
 logger = logging.getLogger(__name__)
 
+XML_LIKE_DOCUMENT_EXTENSIONS = {"xml", "sld"}
+
 
 class SizeRestrictedFileField(forms.FileField):
     """
@@ -160,4 +163,17 @@ def clean_doc_file(self):
                     except (OSError, ValueError):
                         pass
 
+        if doc_file and os.path.splitext(doc_file.name)[1].lower()[1:] in XML_LIKE_DOCUMENT_EXTENSIONS:
+            try:
+                assert_safe_xml(doc_file.read())
+            except UnsafeXMLError as err:
+                logger.warning("Unsafe XML content rejected on document upload: %s", err)
+                raise forms.ValidationError(_("Uploaded XML contains unsafe content."))
+            finally:
+                if hasattr(doc_file, "seek"):
+                    try:
+                        doc_file.seek(0)
+                    except (OSError, ValueError):
+                        pass
+
         return doc_file

geonode/documents/tests.py

@@ -704,6 +704,14 @@ def setUp(self):
         self.perm_spec = self.__class__.perm_spec
         self.doc_link_url = self.__class__.doc_link_url
 
+    def test_document_link_sets_anti_xss_headers(self):
+        self.test_doc.set_permissions(self.perm_spec)
+        self.client.login(username=self.not_admin.username, password="very-secret")
+        response = self.client.get(self.doc_link_url)
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(response.headers.get("X-Content-Type-Options"), "nosniff")
+        self.assertIn("sandbox", response.headers.get("Content-Security-Policy", ""))
+
     def test_document_link_with_permissions(self):
         self.test_doc.set_permissions(self.perm_spec)
         # Get link as Anonymous user

geonode/layers/metadata.py

@@ -26,7 +26,6 @@
 # OWSLib functionality
 from owslib import iso, util
 from owslib.csw import CswRecord
-from owslib.etree import etree as dlxml
 from owslib.fgdc import Metadata
 
 from django.conf import settings
@@ -40,10 +39,14 @@
 
 def set_metadata(xml, identifier="", vals={}, regions=[], keywords=[], custom={}):
     """Generate dict of model properties based on XML metadata"""
+    from geonode.utils import assert_safe_xml, UnsafeXMLError
 
     # check if document is XML
     try:
-        exml = dlxml.fromstring(xml.encode())
+        exml = assert_safe_xml(xml)
+    except UnsafeXMLError as err:
+        LOGGER.warning("Unsafe XML content rejected in metadata: %s", err)
+        raise GeoNodeException("Uploaded XML document contains unsafe content")
     except Exception as err:
         raise GeoNodeException(f"Uploaded XML document is not XML: {str(err)}")
 

geonode/tests/test_utils.py

@@ -32,7 +32,14 @@
 from geonode.geoserver.helpers import set_attributes
 from geonode.tests.base import GeoNodeBaseTestSupport
 from geonode.br.management.commands.utils.utils import ignore_time
-from geonode.utils import copy_tree, bbox_to_wkt, is_safe_url, is_safe_url_with_redirects
+from geonode.utils import (
+    copy_tree,
+    bbox_to_wkt,
+    is_safe_url,
+    is_safe_url_with_redirects,
+    assert_safe_xml,
+    UnsafeXMLError,
+)
 from unittest.mock import MagicMock
 
 
@@ -192,6 +199,25 @@ def test_set_attributes_creates_attributes(self):
             self.assertIn([a.attribute, a.attribute_type], expected_results)
 
 
+class TestAssertSafeXml(TestCase):
+    def test_rejects_xslt_stylesheet(self):
+        payload = (
+            b'<?xml version="1.0"?>'
+            b'<?xml-stylesheet type="text/xsl" href="#s"?>'
+            b'<doc><xsl:stylesheet version="1.0" xmlns:xsl="http://www.w3.org/1999/XSL/Transform"/></doc>'
+        )
+        with self.assertRaises(UnsafeXMLError):
+            assert_safe_xml(payload)
+
+    def test_rejects_script_element(self):
+        with self.assertRaises(UnsafeXMLError):
+            assert_safe_xml(b"<doc><script>alert(1)</script></doc>")
+
+    def test_accepts_safe_xml(self):
+        payload = b"<doc><title>safe content</title></doc>"
+        self.assertIsNotNone(assert_safe_xml(payload))
+
+
 class TestSupportedTypes(TestCase):
     def setUp(self):
         self.replaced = [

geonode/upload/api/serializer.py

@@ -24,6 +24,7 @@
 from geonode.upload.models import UploadParallelismLimit, UploadSizeLimit
 from geonode.resource.enumerator import ExecutionRequestAction as exa
 from geonode.upload.zip_validation import ZipValidationError, is_zip_extension, validate_safe_zip
+from geonode.utils import assert_safe_xml, UnsafeXMLError
 
 
 logger = logging.getLogger(__name__)
@@ -65,6 +66,28 @@ def validate_base_file(self, f):
                     pass
         return f
 
+    def _validate_xml_payload(self, f, label):
+        if f is None:
+            return f
+        try:
+            assert_safe_xml(f.read())
+        except UnsafeXMLError:
+            logger.warning("%s validation failed for uploaded file.", label, exc_info=True)
+            raise serializers.ValidationError(f"Invalid or unsafe {label} file.")
+        finally:
+            if hasattr(f, "seek"):
+                try:
+                    f.seek(0)
+                except (OSError, ValueError):
+                    pass
+        return f
+
+    def validate_xml_file(self, value):
+        return self._validate_xml_payload(value, "XML")
+
+    def validate_sld_file(self, value):
+        return self._validate_xml_payload(value, "SLD")
+
 
 class ImporterSerializer(BaseImporterSerializer):
     class Meta:

geonode/upload/api/tests.py

@@ -156,6 +156,63 @@ def test_importer_upload_rejects_geojson_with_binary_content(self):
         self.assertEqual(400, response.status_code)
         self.assertFalse(Dataset.objects.filter(name="fake").exists())
 
+    def test_importer_upload_rejects_unsafe_xml_file(self):
+        self.client.force_login(get_user_model().objects.get(username="admin"))
+        payload = {
+            "base_file": SimpleUploadedFile(name="test.gpkg", content=GPKG_VALID_HEADER),
+            "xml_file": SimpleUploadedFile(
+                name="meta.xml",
+                content=b'<?xml version="1.0"?><!DOCTYPE x SYSTEM "http://evil.example.com/x.dtd"><x/>',
+                content_type="application/xml",
+            ),
+            "action": "upload",
+        }
+
+        response = self.client.post(self.url, data=payload)
+
+        self.assertEqual(400, response.status_code)
+        self.assertIn(b"Invalid or unsafe XML file", response.content)
+
+    def test_importer_upload_rejects_unsafe_sld_file(self):
+        self.client.force_login(get_user_model().objects.get(username="admin"))
+        payload = {
+            "base_file": SimpleUploadedFile(name="test.gpkg", content=GPKG_VALID_HEADER),
+            "sld_file": SimpleUploadedFile(
+                name="style.sld",
+                content=b'<?xml version="1.0"?><StyledLayerDescriptor><x onload="alert(1)"/>'
+                b"</StyledLayerDescriptor>",
+                content_type="application/xml",
+            ),
+            "action": "upload",
+        }
+
+        response = self.client.post(self.url, data=payload)
+
+        self.assertEqual(400, response.status_code)
+        self.assertIn(b"Invalid or unsafe SLD file", response.content)
+
+    @patch("geonode.upload.api.views.import_orchestrator")
+    def test_importer_upload_accepts_safe_xml_and_sld(self, mock_orchestrator):
+        self.client.force_login(get_user_model().objects.get(username="admin"))
+        payload = {
+            "base_file": SimpleUploadedFile(name="test.gpkg", content=GPKG_VALID_HEADER),
+            "xml_file": SimpleUploadedFile(
+                name="meta.xml",
+                content=b'<?xml version="1.0"?><metadata><title>ok</title></metadata>',
+                content_type="application/xml",
+            ),
+            "sld_file": SimpleUploadedFile(
+                name="style.sld",
+                content=b'<?xml version="1.0"?><StyledLayerDescriptor version="1.0.0"/>',
+                content_type="application/xml",
+            ),
+            "action": "upload",
+        }
+
+        response = self.client.post(self.url, data=payload)
+
+        self.assertEqual(201, response.status_code)
+
     def test_importer_upload_rejects_zip_with_path_traversal(self):
         self.client.force_login(get_user_model().objects.get(username="admin"))
         buf = io.BytesIO()

geonode/upload/handlers/sld/handler.py

@@ -21,7 +21,6 @@
 from geonode.resource.registry import resource_manager_registry
 from geonode.upload.handlers.common.metadata import MetadataFileHandler
 from geonode.upload.handlers.sld.exceptions import InvalidSldException
-from owslib.etree import etree as dlxml
 from geonode.upload.utils import ImporterRequestAction as ira
 
 logger = logging.getLogger("importer")
@@ -89,10 +88,16 @@ def is_valid(files, user, **kwargs):
         """
         Define basic validation steps
         """
+        from geonode.utils import assert_safe_xml, UnsafeXMLError
+
         # calling base validation checks
         try:
             with open(files.get("base_file")) as _xml:
-                dlxml.fromstring(_xml.read().encode())
+                content = _xml.read()
+                assert_safe_xml(content)
+        except UnsafeXMLError as err:
+            logger.warning("Unsafe SLD content rejected: %s", err)
+            raise InvalidSldException("Uploaded document contains unsafe content")
         except Exception as err:
             raise InvalidSldException(f"Uploaded document is not SLD or is invalid: {str(err)}")
         return True

geonode/upload/handlers/xml/handler.py

@@ -21,7 +21,6 @@
 from geonode.resource.registry import resource_manager_registry
 from geonode.upload.handlers.common.metadata import MetadataFileHandler
 from geonode.upload.handlers.xml.exceptions import InvalidXmlException
-from owslib.etree import etree as dlxml
 from geonode.upload.utils import ImporterRequestAction as ira
 
 logger = logging.getLogger("importer")
@@ -89,10 +88,16 @@ def is_valid(files, user=None, **kwargs):
         """
         Define basic validation steps
         """
+        from geonode.utils import assert_safe_xml, UnsafeXMLError
+
         # calling base validation checks
         try:
             with open(files.get("base_file")) as _xml:
-                dlxml.fromstring(_xml.read().encode())
+                content = _xml.read()
+                assert_safe_xml(content)
+        except UnsafeXMLError as err:
+            logger.warning("Unsafe XML content rejected: %s", err)
+            raise InvalidXmlException("Uploaded document contains unsafe content")
         except Exception as err:
             raise InvalidXmlException(f"Uploaded document is not XML or is invalid: {str(err)}")
         return True