Potential fix for code scanning alert no. 3: Polynomial regular expression used on uncontrolled data

src/aethermesh_core/job_result_schema.py

@@ -374,7 +374,7 @@ def _artifact_reference_list(value: object, context: str) -> None:
             _content_addressed_id(reference, f"{context} entries")
         elif (
             reference.startswith(("/", "~"))
-            or re.match(r"[A-Za-z][A-Za-z0-9+.-]*:", reference) is not None
+            or _starts_with_uri_scheme(reference)
             or "\\" in reference
             or any(part == ".." for part in reference.split("/"))
         ):
@@ -383,6 +383,17 @@ def _artifact_reference_list(value: object, context: str) -> None:
             )
 
 
+def _starts_with_uri_scheme(value: str) -> bool:
+    scheme, separator, _remainder = value.partition(":")
+    return bool(
+        separator
+        and scheme
+        and scheme.isascii()
+        and scheme[0].isalpha()
+        and all(character.isalnum() or character in "+.-" for character in scheme)
+    )
+
+
 def _immutable_local_manifest_ref(value: object) -> None:
     """Accept only the atomic local work-manifest record as hash fallback."""
 

tests/test_job_result_schema.py

@@ -3,6 +3,7 @@
 import tempfile
 import unittest
 from pathlib import Path
+from unittest.mock import patch
 
 from aethermesh_core.job_result_schema import (
     JobResultSchemaError,
@@ -319,6 +320,17 @@ def test_outcome_statuses_and_local_artifact_reference_rules(self) -> None:
                 ):
                     validate_job_result_document(document)
 
+        document = copy.deepcopy(self.success)
+        document["references"]["artifact_refs"] = ["A" * 100_000]
+        document["result_hash"] = canonical_result_document_hash(document)
+        with patch(
+            "aethermesh_core.job_result_schema.re.match",
+            side_effect=AssertionError(
+                "artifact references must not use regex matching"
+            ),
+        ):
+            self.assertIs(validate_job_result_document(document), document)
+
         document = copy.deepcopy(self.failed)
         document["error_summary"] = None
         with self.assertRaisesRegex(