fix(import): forbid bad names in import DEV-2415

kpi/models/import_export_task.py

@@ -58,6 +58,7 @@
 from kpi.exceptions import ConcurrentExportException, XlsFormatException
 from kpi.fields import KpiUidField
 from kpi.models import Asset
+from kpi.utils.autoname import _is_group_end
 from kpi.utils.data_exports import (
     ACCESS_LOGS_EXPORT_FIELDS,
     ASSET_FIELDS,
@@ -76,6 +77,7 @@
     rename_xls_sheet,
     rename_xlsx_sheet,
 )
+from kpi.utils.sluggify import is_valid_node_name
 from kpi.utils.storage import is_filesystem_storage
 from kpi.utils.strings import to_str
 from kpi.zip_importer import HttpContentParse
@@ -362,6 +364,7 @@ def _load_assets_from_url(self, url, messages, **kwargs):
                     kontent = xlsx_to_dict(item.readable)
                 except InvalidFileException:
                     kontent = xls_to_dict(item.readable)
+                self._ensure_valid_node_names(kontent)
 
                 if not destination:
                     extra_args['content'] = _strip_header_keys(kontent)
@@ -394,6 +397,14 @@ def _load_assets_from_url(self, url, messages, **kwargs):
             orm_obj.parent = parent_item
             orm_obj.save()
 
+    @staticmethod
+    def _ensure_valid_node_names(survey_dict):
+        survey_list = survey_dict.get('survey', [])
+        for node in survey_list:
+            name = node.get('name')
+            if bool(name) and not _is_group_end(node) and not is_valid_node_name(name):
+                raise ValueError(f'Invalid node name: {name}')
+
     def _parse_b64_upload(self, base64_encoded_upload, messages, **kwargs):
         filename = kwargs.get('filename', False)
         desired_type = kwargs.get('desired_type')
@@ -404,6 +415,7 @@ def _parse_b64_upload(self, base64_encoded_upload, messages, **kwargs):
             filename = ''
         library = kwargs.get('library')
         survey_dict = _b64_xls_to_dict(base64_encoded_upload)
+        self._ensure_valid_node_names(survey_dict)
         survey_dict_keys = survey_dict.keys()
 
         destination = kwargs.get('destination', False)

kpi/tests/api/v2/test_api_imports.py

@@ -13,6 +13,7 @@
 from kobo.apps.kobo_auth.shortcuts import User
 from kpi.constants import ASSET_TYPE_BLOCK, ASSET_TYPE_QUESTION
 from kpi.models import Asset, ImportTask
+from kpi.models.import_export_task import ImportExportStatusChoices
 from kpi.tests.base_test_case import BaseTestCase
 from kpi.urls.router_api_v2 import URL_NAMESPACE as ROUTER_URL_NAMESPACE
 from kpi.utils.strings import to_str
@@ -1087,3 +1088,29 @@ def test_import_to_existing_asset_base64_xls_records_audit_log_info(self):
                 'project_owner': self.asset.owner.username,
             },
         )
+
+    def test_import_raises_error_with_invalid_names(self):
+        survey_sheet_content = [
+            ['type', 'name', 'label::English (en)'],
+            ['today', 'today', ''],
+            ['begin group', 'this is a bad group name', 'Bad bad bad, no donut'],
+            ['note', 'note_1', 'Hi there 👋'],
+            ['end group', '', ''],
+        ]
+        choices_sheet_content = [
+            ['list_name', 'name', 'label::English (en)'],
+        ]
+        content = (
+            ('survey', survey_sheet_content),
+            ('choices', choices_sheet_content),
+        )
+
+        task_data = self._construct_xls_for_import(content, name='Bad file')
+        encoded_str = task_data['base64Encoded']
+        encoded_substr = encoded_str[encoded_str.index('base64') + 7:]
+        task_data['base64Encoded'] = encoded_substr
+        task = ImportTask.objects.create(user=self.asset.owner, data=task_data)
+        result = task.run()
+        assert result.status == ImportExportStatusChoices.ERROR
+        error_message = result.messages['error']
+        assert 'this is a bad group name' in error_message