From f8a52733fe6b19d359ffa885746f001829a13f0f Mon Sep 17 00:00:00 2001 From: Spyros Date: Mon, 13 Jul 2026 18:19:06 +0100 Subject: [PATCH] fix: rewrite CSV import validation for MyOpenCRE path (#554) Rebase #751 onto current main: UTF-8/triple-quote/csv.Error checks in web_main, CRE XXX-XXX|Name validation in myopencre_parser, plus tests. Co-authored-by: Ayesha Rafaqat Co-authored-by: Cursor --- application/tests/myopencre_parser_test.py | 27 +++++++++++ application/tests/web_main_test.py | 45 +++++++++++++++++++ .../parsers/myopencre_parser.py | 33 ++++++++++++++ application/web/web_main.py | 22 ++++++++- 4 files changed, 125 insertions(+), 2 deletions(-) diff --git a/application/tests/myopencre_parser_test.py b/application/tests/myopencre_parser_test.py index 565192abe..87486e56d 100644 --- a/application/tests/myopencre_parser_test.py +++ b/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() diff --git a/application/tests/web_main_test.py b/application/tests/web_main_test.py index 0fab4fc74..5f0cbf176 100644 --- a/application/tests/web_main_test.py +++ b/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() diff --git a/application/utils/external_project_parsers/parsers/myopencre_parser.py b/application/utils/external_project_parsers/parsers/myopencre_parser.py index 09393aae5..9b87d3ada 100644 --- a/application/utils/external_project_parsers/parsers/myopencre_parser.py +++ b/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 ``|`` (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 diff --git a/application/web/web_main.py b/application/web/web_main.py index b460672b5..4bc681e90 100644 --- a/application/web/web_main.py +++ b/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: