fix: add input validation for CSV import (#554)

application/tests/myopencre_parser_test.py

@@ -46,5 +46,32 @@ def test_raises_on_same_name_different_id_conflict(self) -> None:
                 myopencre_parser._reconcile_cre_identities(cres)
 
 
+class MyOpenCreCsvValidationTest(unittest.TestCase):
+    def test_accepts_well_formed_cre_cells(self) -> None:
+        myopencre_parser.validate_cre_csv_rows(
+            [{"CRE 0": "123-456|Access Control", "standard|name": "ASVS"}]
+        )
+
+    def test_rejects_short_cre_id(self) -> None:
+        with self.assertRaises(ValueError) as cm:
+            myopencre_parser.validate_cre_csv_rows(
+                [{"CRE 0": "12-456|Bad Id", "standard|name": "ASVS"}]
+            )
+        self.assertIn("Expected XXX-XXX|Name", str(cm.exception))
+        self.assertIn("row 2", str(cm.exception))
+
+    def test_rejects_missing_separator(self) -> None:
+        with self.assertRaises(ValueError) as cm:
+            myopencre_parser.validate_cre_csv_rows(
+                [{"CRE 0": "123-456 Access Control", "standard|name": "ASVS"}]
+            )
+        self.assertIn("Expected XXX-XXX|Name", str(cm.exception))
+
+    def test_skips_empty_cre_cells(self) -> None:
+        myopencre_parser.validate_cre_csv_rows(
+            [{"CRE 0": "", "CRE 1": "n/a", "standard|name": "ASVS"}]
+        )
+
+
 if __name__ == "__main__":
     unittest.main()

application/tests/web_main_test.py

@@ -1401,6 +1401,51 @@ def test_import_from_cre_csv(self) -> None:
                 self.assertGreaterEqual(data.get("new_standards"), 2)
                 self.assertIsInstance(data.get("new_cres"), list)
 
+    def test_import_from_cre_csv_rejects_bad_cre_format(self) -> None:
+        """Malformed CRE cells must 400 with a clear message (#554 / #751)."""
+        workspace = tempfile.mkdtemp()
+        path = os.path.join(workspace, "bad_cre.csv")
+        with open(path, "w", encoding="utf-8") as f:
+            writer = csv.DictWriter(
+                f, fieldnames=["CRE 0", "standard|name", "standard|id"]
+            )
+            writer.writeheader()
+            writer.writerow(
+                {
+                    "CRE 0": "12-456|Too Short",
+                    "standard|name": "ASVS",
+                    "standard|id": "1.1.1",
+                }
+            )
+        with patch.dict(os.environ, {"CRE_ALLOW_IMPORT": "True"}):
+            with self.app.test_client() as client:
+                response = client.post(
+                    "/rest/v1/cre_csv_import",
+                    data={"cre_csv": open(path, "rb")},
+                    buffered=True,
+                    content_type="multipart/form-data",
+                )
+                self.assertEqual(400, response.status_code)
+                self.assertIn(b"Expected XXX-XXX|Name", response.data)
+
+    def test_import_from_cre_csv_rejects_triple_quotes(self) -> None:
+        """Triple-quoted content must not create bogus root CREs (#554)."""
+        workspace = tempfile.mkdtemp()
+        path = os.path.join(workspace, "triple.csv")
+        with open(path, "w", encoding="utf-8") as f:
+            f.write('CRE 0,standard|name,standard|id\n')
+            f.write('"""123-456|Quoted""",ASVS,1.1.1\n')
+        with patch.dict(os.environ, {"CRE_ALLOW_IMPORT": "True"}):
+            with self.app.test_client() as client:
+                response = client.post(
+                    "/rest/v1/cre_csv_import",
+                    data={"cre_csv": open(path, "rb")},
+                    buffered=True,
+                    content_type="multipart/form-data",
+                )
+                self.assertEqual(400, response.status_code)
+                self.assertIn(b"triple-quoted", response.data)
+
     def test_get_cre_csv(self) -> None:
         # empty string means temporary db
         collection = db.Node_collection().with_graph()

application/utils/external_project_parsers/parsers/myopencre_parser.py

@@ -13,6 +13,38 @@
 logger.setLevel(logging.INFO)
 
 _CRE_ID_TOKEN = re.compile(r"^\d{3}-\d{3}$")
+# Exported CRE cells are ``<id>|<name>`` (see ExportFormat.separator).
+_CRE_CELL_PATTERN = re.compile(r"^\d{3}-\d{3}\|.+$")
+
+
+def _is_empty_cell(value: Any) -> bool:
+    if value is None:
+        return True
+    text = str(value).strip()
+    return text == "" or text.lower() in {"none", "nan", "n/a", "no", "tbd", "0"}
+
+
+def validate_cre_csv_rows(rows: List[Dict[str, Any]]) -> None:
+    """Fail fast on CRE cells that are not ``XXX-XXX|Name`` (#554).
+
+    Raises ``ValueError`` with a row/column message so the web layer can
+    return HTTP 400 instead of a 500 / silent bad parse.
+    """
+    if not rows:
+        return
+    cre_headers = [h for h in rows[0].keys() if str(h).startswith("CRE")]
+    for row_index, row in enumerate(rows, start=2):
+        for header in cre_headers:
+            value = row.get(header)
+            if _is_empty_cell(value):
+                continue
+            normalized = str(value).strip()
+            if not _CRE_CELL_PATTERN.match(normalized):
+                raise ValueError(
+                    "Invalid CRE column format in row "
+                    f"{row_index}, column '{header}'. Expected XXX-XXX|Name "
+                    f"(got '{normalized}')."
+                )
 
 
 def _load_existing_cre_identity_maps() -> Tuple[Dict[str, str], Dict[str, str]]:
@@ -90,6 +122,7 @@ def parse_rows_to_documents(rows: List[Dict[str, Any]]) -> ParseResult:
     conventions. It reuses export_format_parser.parse_export_format to
     understand the CSV structure.
     """
+    validate_cre_csv_rows(rows)
     documents = export_format_parser.parse_export_format(rows)
 
     # CREs are under the Credoctypes.CRE value key

application/web/web_main.py

@@ -1292,9 +1292,27 @@ def import_from_cre_csv() -> Any:
     if file is None:
         abort(400, "No file provided")
     contents = file.read()
-    csv_read = csv.DictReader(contents.decode("utf-8").splitlines())
     try:
-        parse_result = myopencre_parser.parse_rows_to_documents(list(csv_read))
+        contents_text = contents.decode("utf-8")
+    except UnicodeDecodeError:
+        abort(400, "Invalid CSV encoding. Expected UTF-8.")
+
+    # Triple-quoted blobs were previously mis-parsed into bogus root CREs (#554).
+    if '"""' in contents_text:
+        abort(
+            400,
+            "Invalid CSV content: triple-quoted text detected. "
+            "Use standard CSV quoting (double quotes) instead.",
+        )
+
+    try:
+        csv_read = csv.DictReader(contents_text.splitlines())
+        rows = list(csv_read)
+    except csv.Error as exc:
+        abort(400, f"Invalid CSV content: {exc}")
+
+    try:
+        parse_result = myopencre_parser.parse_rows_to_documents(rows)
     except cre_exceptions.DuplicateLinkException as dle:
         abort(500, f"error during parsing of the incoming CSV, err:{dle}")
     except ValueError as ve: