[Fixes #13520] Improved Error Robustness for special characters and whitespaces in filenames and attribute tables (#13535)

* [Fixes #13520] added flow for dynamic vrt file and handle more special character
* [Fixes #13520] update code for only create vrt if there is problem in field name

Author: sijandh35

geonode/upload/handlers/base.py

@@ -22,6 +22,7 @@
 from pathlib import Path
 from typing import List
 import zipfile
+import re
 
 from geonode.assets.utils import create_asset_and_link
 from geonode.resource.enumerator import ExecutionRequestAction as exa
@@ -237,18 +238,14 @@ def fixup_name(self, name):
         prefix = name[0]
         if prefix.isnumeric():
             name = name.replace(name[0], "_")
-        return (
-            name.lower()
-            .replace("-", "_")
-            .replace(" ", "_")
-            .replace("#", "_")
-            .replace("\\", "_")
-            .replace(".", "")
-            .replace(")", "")
-            .replace("(", "")
-            .replace(",", "")
-            .replace("&", "")[:62]
-        )
+        name = name.lower()
+        # Replace specific chars with underscore in one pass
+        name = re.sub(r"[-# \\&]", "_", name)
+
+        # Remove unwanted characters in one pass
+        name = re.sub(r'[.(),!"$%\'*+/:;<=>?@\[\]^`{|}~]', "", name)
+
+        return name[:62]
 
     def extract_resource_to_publish(self, files, layer_name, alternate, **kwargs):
         """

geonode/upload/handlers/common/tests_vector.py

@@ -35,6 +35,7 @@
 from dynamic_models.models import ModelSchema
 from osgeo import ogr
 from django.test.utils import override_settings
+from geonode.upload.utils import create_vrt_file, has_incompatible_field_names
 
 
 class TestBaseVectorFileHandler(TestCase):
@@ -69,6 +70,100 @@ def test_create_error_log(self):
         expected = "Task: foo_task_name raised an error during actions for layer: alternate: my exception"
         self.assertEqual(expected, actual)
 
+    def test_create_vrt_file_with_special_chars(self):
+        """
+        Test that create_vrt_file correctly sanitizes layer and field names
+        with spaces and special characters, and generates a valid VRT file.
+        """
+        source_filepath = "source file.shp"
+
+        mock_layer = MagicMock(spec=ogr.Layer)
+        mock_layer.GetName.return_value = "Layer With Spaces"
+        mock_layer.GetGeomType.return_value = ogr.wkbPolygon
+
+        mock_field_1 = MagicMock()
+        mock_field_1.GetName.return_value = "Field 1 (m)"
+        mock_field_1.GetTypeName.return_value = "String"
+
+        mock_field_2 = MagicMock()
+        mock_field_2.GetName.return_value = "another-field"
+        mock_field_2.GetTypeName.return_value = "Real"
+
+        mock_field_3 = MagicMock()
+        mock_field_3.GetName.return_value = "field/with/slash"
+        mock_field_3.GetTypeName.return_value = "Integer"
+
+        mock_layer_defn = MagicMock()
+        mock_layer_defn.GetFieldCount.return_value = 3
+        mock_layer_defn.GetFieldDefn.side_effect = [mock_field_1, mock_field_2, mock_field_3].__getitem__
+
+        mock_layer.GetLayerDefn.return_value = mock_layer_defn
+
+        vrt_filename, vrt_layer_name = None, None
+        try:
+            vrt_filename, vrt_layer_name = create_vrt_file(mock_layer, source_filepath)
+
+            self.assertIsNotNone(vrt_filename)
+            self.assertTrue(os.path.exists(vrt_filename))
+            self.assertEqual(vrt_layer_name, "layer_with_spaces")
+
+            with open(vrt_filename, "r") as f:
+                vrt_content = f.read()
+
+            self.assertIn('<OGRVRTLayer name="layer_with_spaces">', vrt_content)
+            self.assertIn(f"<SrcDataSource>{source_filepath}</SrcDataSource>", vrt_content)
+            self.assertIn("<SrcLayer>Layer With Spaces</SrcLayer>", vrt_content)
+
+            self.assertIn('<Field name="field_1_m" src="Field 1 (m)" type="String" />', vrt_content)
+            self.assertIn('<Field name="another_field" src="another-field" type="Real" />', vrt_content)
+            self.assertIn('<Field name="fieldwithslash" src="field/with/slash" type="Integer" />', vrt_content)
+
+        finally:
+            if vrt_filename and os.path.exists(vrt_filename):
+                os.remove(vrt_filename)
+
+    def test_has_incompatible_field_names(self):
+        """
+        Test that has_incompatible_field_names correctly identifies layers
+        with field names that need sanitization.
+        """
+
+        # layer with incompatible field names
+        mock_layer_incompatible = MagicMock(spec=ogr.Layer)
+        mock_layer_incompatible.GetName.return_value = "Layer With Spaces"
+
+        mock_field_1 = MagicMock()
+        mock_field_1.GetName.return_value = "Field 1 (m)"
+
+        mock_field_2 = MagicMock()
+        mock_field_2.GetName.return_value = "another-#field"
+
+        mock_layer_defn_incompatible = MagicMock()
+        mock_layer_defn_incompatible.GetFieldCount.return_value = 2
+        mock_layer_defn_incompatible.GetFieldDefn.side_effect = [mock_field_1, mock_field_2].__getitem__
+
+        mock_layer_incompatible.GetLayerDefn.return_value = mock_layer_defn_incompatible
+
+        self.assertTrue(has_incompatible_field_names(mock_layer_incompatible))
+
+        # layer with compatible field names
+        mock_layer_compatible = MagicMock(spec=ogr.Layer)
+        mock_layer_compatible.GetName.return_value = "compatible_layer"
+
+        mock_field_3 = MagicMock()
+        mock_field_3.GetName.return_value = "field_one"
+
+        mock_field_4 = MagicMock()
+        mock_field_4.GetName.return_value = "field_two"
+
+        mock_layer_defn_compatible = MagicMock()
+        mock_layer_defn_compatible.GetFieldCount.return_value = 2
+        mock_layer_defn_compatible.GetFieldDefn.side_effect = [mock_field_3, mock_field_4].__getitem__
+
+        mock_layer_compatible.GetLayerDefn.return_value = mock_layer_defn_compatible
+
+        self.assertFalse(has_incompatible_field_names(mock_layer_compatible))
+
     def test_create_dynamic_model_fields(self):
         try:
             # Prepare the test

geonode/upload/handlers/common/vector.py

@@ -63,6 +63,7 @@
 from geonode.upload.utils import ImporterRequestAction as ira
 from geonode.security.registry import permissions_registry
 from geonode.storage.manager import FileSystemStorageManager
+from geonode.upload.utils import create_vrt_file, has_incompatible_field_names
 
 
 logger = logging.getLogger("importer")
@@ -260,7 +261,10 @@ def create_ogr2ogr_command(files, original_name, ovverwrite_layer, alternate):
                 _datastore["USER"],
                 _datastore["PASSWORD"],
             )
-        options += f'"{files.get("base_file")}"' + " "
+        # vrt file is aready created in import_resource and vrt will be auto detected by ogr2ogr
+        # and also the base_file will work so can be used as alternative for fallback which will also be autodeteced by ogr2ogr.
+        input_file = files.get("temp_vrt_file") or files.get("base_file")
+        options += f'"{input_file}"' + " "
 
         options += f'-nln {alternate} "{original_name}"'
 
@@ -414,6 +418,12 @@ def import_resource(self, files: dict, execution_id: str, **kwargs) -> str:
                     )
                     # and layer.GetGeometryColumn() is not None
                 ):
+                    _files = files.copy()
+                    vrt_layer_name = None
+                    if has_incompatible_field_names(layer):
+                        vrt_filename, vrt_layer_name = create_vrt_file(layer, files.get("base_file"))
+                        _files["temp_vrt_file"] = vrt_filename
+
                     # update the execution request object
                     # setup dynamic model and retrieve the group task needed for tun the async workflow
                     # create the async task for create the resource into geonode_data with ogr2ogr
@@ -433,8 +443,8 @@ def import_resource(self, files: dict, execution_id: str, **kwargs) -> str:
 
                     ogr_res = self.get_ogr2ogr_task_group(
                         execution_id,
-                        files,
-                        layer.GetName().lower(),
+                        _files,
+                        vrt_layer_name or layer.GetName().lower(),
                         should_be_overwritten,
                         alternate,
                     )
@@ -594,7 +604,9 @@ def create_dynamic_model_fields(
         return_celery_group: bool = True,
     ):
         # retrieving the field schema from ogr2ogr and converting the type to Django Types
-        layer_schema = [{"name": x.name.lower(), "class_name": self._get_type(x), "null": True} for x in layer.schema]
+        layer_schema = [
+            {"name": self.fixup_name(x.name), "class_name": self._get_type(x), "null": True} for x in layer.schema
+        ]
         if (
             layer.GetGeometryColumn()
             or self.default_geometry_column_name
@@ -1329,6 +1341,10 @@ def import_with_ogr2ogr(
 
         process = Popen(" ".join(commands), stdout=PIPE, stderr=PIPE, shell=True)
         stdout, stderr = process.communicate()
+
+        if files.get("temp_vrt_file") and os.path.exists(files["temp_vrt_file"]):
+            os.remove(files["temp_vrt_file"])
+
         if (
             stderr is not None
             and stderr != b""
@@ -1345,6 +1361,8 @@ def import_with_ogr2ogr(
             raise Exception(f"{message} for layer {alternate}")
         return "ogr2ogr", alternate, execution_id
     except Exception as e:
+        if files.get("temp_vrt_file") and os.path.exists(files.get("temp_vrt_file")):
+            os.remove(files["temp_vrt_file"])
         call_rollback_function(
             execution_id,
             handlers_module_path=handler_module_path,

geonode/upload/handlers/tests.py

@@ -92,3 +92,9 @@ def test_fixup_name_replace_digits_with_underscore(self):
         expected_name = "_layername"
         actual = BaseHandler().fixup_name(layer_name)
         self.assertEqual(expected_name, actual)
+
+    def test_fixup_name_replace_digits_with_special_character(self):
+        layer_name = "5Test-Name# \\&.()!$%*+@[]~end"
+        expected_name = "_test_name____end"
+        actual = BaseHandler().fixup_name(layer_name)
+        self.assertEqual(expected_name, actual)

geonode/upload/templates/upload/vrt_template.xml

@@ -0,0 +1,9 @@
+<OGRVRTDataSource>
+  <OGRVRTLayer name="{{ layer_name|escape }}">
+    <SrcDataSource>{{ source_filepath|escape }}</SrcDataSource>
+    <SrcLayer>{{ original_layer_name|escape }}</SrcLayer>
+    {% for field in fields %}
+    <Field name="{{ field.name|escape }}" src="{{ field.src|escape }}" type="{{ field.type|escape }}" />
+    {% endfor %}
+  </OGRVRTLayer>
+</OGRVRTDataSource>

geonode/upload/utils.py

@@ -17,6 +17,8 @@
 #
 #########################################################################
 import enum
+import tempfile
+import os
 from geonode.resource.manager import ResourceManager
 from geonode.geoserver.manager import GeoServerResourceManager
 from geonode.base.models import ResourceBase
@@ -30,6 +32,7 @@
 from geonode.resource.models import ExecutionRequest
 from django.core.exceptions import ObjectDoesNotExist
 from django.conf import settings
+from django.template.loader import render_to_string
 
 
 def get_max_upload_size(slug):
@@ -48,6 +51,61 @@ def get_max_upload_parallelism_limit(slug):
     return max_number
 
 
+def has_incompatible_field_names(layer):
+    """
+    Checks if any of the layer's field names contain special
+    characters that need sanitization, Currently used to check if VRT need to be created or not.
+    """
+    from geonode.upload.handlers.base import BaseHandler
+
+    def _name_needs_sanitization(name):
+        # fixup_name lowercases, replaces special chars, and truncates.
+        # If the sanitized name is different from just the lowercased name,
+        # it means special characters were involved that need handling.
+        return BaseHandler().fixup_name(name) != name.lower()
+
+    layer_defn = layer.GetLayerDefn()
+    for i in range(layer_defn.GetFieldCount()):
+        field_name = layer_defn.GetFieldDefn(i).GetName()
+        if _name_needs_sanitization(field_name):
+            return True
+
+    return False
+
+
+def create_vrt_file(layer, source_filepath):
+    """
+    Dynamically creates a VRT file to sanitize field names.
+    """
+    from geonode.upload.handlers.base import BaseHandler
+
+    vrt_layer_name = BaseHandler().fixup_name(layer.GetName())
+
+    layer_defn = layer.GetLayerDefn()
+    fields = [
+        {
+            "name": BaseHandler().fixup_name(layer_defn.GetFieldDefn(i).GetName()),
+            "src": layer_defn.GetFieldDefn(i).GetName(),
+            "type": layer_defn.GetFieldDefn(i).GetTypeName(),
+        }
+        for i in range(layer_defn.GetFieldCount())
+    ]
+
+    context = {
+        "layer_name": vrt_layer_name,
+        "source_filepath": source_filepath,
+        "original_layer_name": layer.GetName(),
+        "fields": fields,
+    }
+    vrt_content = render_to_string("upload/vrt_template.xml", context)
+
+    vrt_fd, vrt_filename = tempfile.mkstemp(suffix=".vrt")
+    with os.fdopen(vrt_fd, "w") as f:
+        f.write(vrt_content)
+
+    return vrt_filename, vrt_layer_name
+
+
 class ImporterRequestAction(enum.Enum):
     ROLLBACK = _("rollback")
     RESOURCE_METADATA_UPLOAD = _("resource_metadata_upload")