basic artifact identifier validation

andmat900 · andmat900 · commit 3baf7ac54733 · 2025-10-03T07:38:59.000+02:00
Change-Id: Ie3ce15c2aedba33643f74066eb872fb5269261ec
diff --git a/python/src/etos_api/library/validator.py b/python/src/etos_api/library/validator.py
@@ -22,6 +22,7 @@
 from pydantic import BaseModel  # pylint:disable=no-name-in-module
 from pydantic import ValidationError, conlist, constr, field_validator
 from pydantic.fields import PrivateAttr
+from packageurl import PackageURL
 
 from etos_api.library.docker import Docker
 
@@ -176,3 +177,51 @@ async def validate(self, test_suite):
                 assert (
                     await docker.digest(test_runner) is not None
                 ), f"Test runner {test_runner} not found"
+
+
+class ArtifactValidator:
+    """Validator for artifact identities and IDs."""
+
+    def validate_artifact_identity_or_id(
+        self, artifact_identity: str = None, artifact_id: str = None
+    ) -> None:
+        """Validate that artifact_identity or artifact_id is a valid PURL or UUID.
+
+        :param artifact_identity: The artifact identity to validate (should be PURL if provided).
+        :param artifact_id: The artifact ID to validate (should be UUID if provided).
+        :raises ValueError: If validation fails.
+        """
+        if artifact_identity:
+            if not self.validate_purl(artifact_identity):
+                raise ValueError(
+                    f"Invalid artifact_identity: '{artifact_identity}' is not a valid PURL. "
+                    "PURL must start with 'pkg:'"
+                )
+
+        if artifact_id:
+            if not self.validate_uuid(artifact_id):
+                raise ValueError(f"Invalid artifact_id: '{artifact_id}' is not a valid UUID.")
+
+    def validate_purl(self, purl_string: str) -> bool:
+        """Validate if a string is a valid Package URL (PURL).
+
+        :param purl_string: The string to validate as a PURL.
+        :return: True if valid PURL, False otherwise.
+        """
+        try:
+            PackageURL.from_string(purl_string)
+            return True
+        except (ValueError, TypeError):
+            return False
+
+    def validate_uuid(self, uuid_string: str) -> bool:
+        """Validate if a string is a valid UUID.
+
+        :param uuid_string: The string to validate as a UUID.
+        :return: True if valid UUID, False otherwise.
+        """
+        try:
+            UUID(str(uuid_string))
+            return True
+        except (ValueError, TypeError):
+            return False
diff --git a/python/src/etos_api/routers/v0/router.py b/python/src/etos_api/routers/v0/router.py
@@ -30,9 +30,8 @@
 
 from etos_api.library.environment import Configuration, configure_testrun
 from etos_api.library.utilities import sync_to_async
-
 from .schemas import AbortEtosResponse, StartEtosRequest, StartEtosResponse
-from .utilities import wait_for_artifact_created, validate_suite
+from .utilities import wait_for_artifact_created, validate_suite, validate_artifact
 
 ETOSv0 = FastAPI(
     title="ETOS",
@@ -108,6 +107,11 @@ async def _start(etos: StartEtosRequest, span: Span) -> dict:  # pylint:disable=
     await validate_suite(etos.test_suite_url)
     LOGGER.info("Test suite validated.")
 
+    # Validate artifact identity and ID before proceeding
+    LOGGER.info("Validating artifact identity and ID.")
+    await validate_artifact(artifact_identity=etos.artifact_identity, artifact_id=etos.artifact_id)
+    LOGGER.info("Artifact identity and ID validated.")
+
     etos_library = ETOS("ETOS API", os.getenv("HOSTNAME"), "ETOS API")
     await sync_to_async(etos_library.config.rabbitmq_publisher_from_environment)
 
diff --git a/python/src/etos_api/routers/v0/utilities.py b/python/src/etos_api/routers/v0/utilities.py
@@ -27,7 +27,7 @@
     ARTIFACT_IDENTITY_QUERY,
     VERIFY_ARTIFACT_ID_EXISTS,
 )
-from etos_api.library.validator import SuiteValidator
+from etos_api.library.validator import SuiteValidator, ArtifactValidator
 
 LOGGER = logging.getLogger(__name__)
 
@@ -106,3 +106,23 @@ async def validate_suite(test_suite_url: str) -> None:
         raise HTTPException(
             status_code=400, detail=f"Test suite validation failed. {exception}"
         ) from exception
+
+
+async def validate_artifact(artifact_identity: str = None, artifact_id=None) -> None:
+    """Validate the artifact identity and ID through the ArtifactValidator.
+
+    :param artifact_identity: The artifact identity to validate (should be PURL if provided).
+    :param artifact_id: The artifact ID to validate (can be UUID object or string).
+    """
+    span = trace.get_current_span()
+
+    try:
+        # Convert artifact_id to string if it's not None
+        artifact_id_str = str(artifact_id) if artifact_id is not None else None
+        ArtifactValidator().validate_artifact_identity_or_id(artifact_identity, artifact_id_str)
+    except ValueError as exception:
+        msg = "Artifact validation failed!"
+        LOGGER.error(msg)
+        LOGGER.error(exception)
+        span.add_event(msg)
+        raise HTTPException(status_code=400, detail=msg) from exception
diff --git a/python/src/etos_api/routers/v1alpha/router.py b/python/src/etos_api/routers/v1alpha/router.py
@@ -35,12 +35,12 @@
 from opentelemetry import trace
 from opentelemetry.trace import Span
 
-
 from .schemas import AbortTestrunResponse, StartTestrunRequest, StartTestrunResponse
 from .utilities import (
     wait_for_artifact_created,
     download_suite,
     validate_suite,
+    validate_artifact,
     convert_to_rfc1123,
     recipes_from_tests,
 )
@@ -124,6 +124,11 @@ async def _create_testrun(etos: StartTestrunRequest, span: Span) -> dict:
     await validate_suite(test_suite)
     LOGGER.info("Test suite validated.")
 
+    # Validate artifact identity and ID before proceeding
+    LOGGER.info("Validating artifact identity and ID.")
+    await validate_artifact(artifact_identity=etos.artifact_identity, artifact_id=etos.artifact_id)
+    LOGGER.info("Artifact identity and ID validated.")
+
     etos_library = ETOS("ETOS API", os.getenv("HOSTNAME", "localhost"), "ETOS API")
 
     LOGGER.info("Get artifact created %r", (etos.artifact_identity or str(etos.artifact_id)))
diff --git a/python/src/etos_api/routers/v1alpha/utilities.py b/python/src/etos_api/routers/v1alpha/utilities.py
@@ -28,7 +28,7 @@
     ARTIFACT_IDENTITY_QUERY,
     VERIFY_ARTIFACT_ID_EXISTS,
 )
-from etos_api.library.validator import SuiteValidator
+from etos_api.library.validator import SuiteValidator, ArtifactValidator
 
 LOGGER = logging.getLogger(__name__)
 
@@ -136,6 +136,26 @@ def convert_to_rfc1123(value: str) -> str:
     return result.lower()
 
 
+async def validate_artifact(artifact_identity: str = None, artifact_id=None) -> None:
+    """Validate the artifact identity and ID through the ArtifactValidator.
+
+    :param artifact_identity: The artifact identity to validate (should be PURL if provided).
+    :param artifact_id: The artifact ID to validate (can be UUID object or string).
+    """
+    span = trace.get_current_span()
+
+    try:
+        # Convert artifact_id to string if it's not None
+        artifact_id_str = str(artifact_id) if artifact_id is not None else None
+        ArtifactValidator().validate_artifact_identity_or_id(artifact_identity, artifact_id_str)
+    except ValueError as exception:
+        msg = "Failed to validate artifact identifier"
+        LOGGER.error(msg)
+        LOGGER.error(exception)
+        span.add_event(msg)
+        raise HTTPException(status_code=400, detail=msg) from exception
+
+
 async def recipes_from_tests(tests: list[dict]) -> list[dict]:
     """Load Eiffel TERCC recipes from test.
 
diff --git a/python/tests/library/test_validator.py b/python/tests/library/test_validator.py
@@ -20,7 +20,11 @@
 
 import pytest
 
-from etos_api.library.validator import SuiteValidator, ValidationError
+from etos_api.library.validator import (
+    ArtifactValidator,
+    SuiteValidator,
+    ValidationError,
+)
 
 logging.basicConfig(level=logging.DEBUG, stream=sys.stdout)
 
@@ -356,3 +360,171 @@ async def test_validate_empty_constraints(self):
             self.logger.info("STEP: Validate a suite without the required key.")
             with pytest.raises(ValidationError):
                 await validator.validate([base_suite])
+
+
+class TestArtifactValidator:
+    """Test the artifact validation functions."""
+
+    def setup_method(self):
+        """Set up test fixtures."""
+        self.validator = ArtifactValidator()
+
+    def test_validate_purl_valid(self):
+        """Test that valid PURL strings are accepted."""
+        valid_purls = [
+            "pkg:npm/lodash@4.17.21",
+            "pkg:pypi/requests@2.25.1",
+            "pkg:maven/org.apache.commons/commons-lang3@3.12.0",
+            "pkg:golang/github.com/gorilla/mux@v1.8.0",
+            "pkg:docker/nginx@latest",
+            "pkg:generic/openssl@1.1.1k",
+        ]
+
+        for purl in valid_purls:
+            assert self.validator.validate_purl(purl) is True
+
+    def test_validate_purl_invalid(self):
+        """Test that invalid PURL strings are rejected."""
+        invalid_purls = [
+            "",
+            None,
+            "not-a-purl",
+            "http://example.com",
+            "pkg:",  # Missing parts
+            "pkg:npm/",  # Incomplete
+            "npm/lodash@4.17.21",  # Missing pkg: prefix
+            "PKG:npm/lodash@4.17.21",  # Wrong case
+        ]
+
+        for purl in invalid_purls:
+            assert self.validator.validate_purl(purl) is False
+
+    def test_validate_uuid_valid(self):
+        """Test that valid UUID strings are accepted."""
+        valid_uuids = [
+            "550e8400-e29b-41d4-a716-446655440000",
+            "6ba7b810-9dad-11d1-80b4-00c04fd430c8",
+            "6ba7b811-9dad-11d1-80b4-00c04fd430c8",
+            "6ba7b812-9dad-11d1-80b4-00c04fd430c8",
+            "6ba7b814-9dad-11d1-80b4-00c04fd430c8",
+            "f47ac10b-58cc-4372-a567-0e02b2c3d479",
+        ]
+
+        for uuid_str in valid_uuids:
+            assert self.validator.validate_uuid(uuid_str) is True
+
+    def test_validate_uuid_invalid(self):
+        """Test that invalid UUID strings are rejected."""
+        invalid_uuids = [
+            "",
+            None,
+            "not-a-uuid",
+            "550e8400-e29b-41d4-a716",  # Too short
+            "550e8400-e29b-41d4-a716-446655440000-extra",  # Too long
+            "550e8400-e29b-41d4-a716-44665544000g",  # Invalid character (g)
+        ]
+
+        for uuid_str in invalid_uuids:
+            assert self.validator.validate_uuid(uuid_str) is False
+
+    def test_validate_artifact_identity_or_id_valid_purl(self):
+        """Test validation with valid PURL artifact_identity."""
+        # Should not raise any exception
+        self.validator.validate_artifact_identity_or_id(
+            artifact_identity="pkg:npm/lodash@4.17.21", artifact_id=None
+        )
+
+    def test_validate_artifact_identity_or_id_valid_uuid(self):
+        """Test validation with valid UUID artifact_id."""
+        # Should not raise any exception
+        self.validator.validate_artifact_identity_or_id(
+            artifact_identity=None, artifact_id="550e8400-e29b-41d4-a716-446655440000"
+        )
+
+    def test_validate_artifact_identity_or_id_invalid_purl(self):
+        """Test validation with invalid PURL artifact_identity raises ValueError."""
+        with pytest.raises(ValueError) as exc_info:
+            self.validator.validate_artifact_identity_or_id(
+                artifact_identity="not-a-purl", artifact_id=None
+            )
+        assert "Invalid artifact_identity" in str(exc_info.value)
+        assert "is not a valid PURL" in str(exc_info.value)
+
+    def test_validate_artifact_identity_or_id_invalid_uuid(self):
+        """Test validation with invalid UUID artifact_id raises ValueError."""
+        with pytest.raises(ValueError) as exc_info:
+            self.validator.validate_artifact_identity_or_id(
+                artifact_identity=None, artifact_id="not-a-uuid"
+            )
+        assert "Invalid artifact_id" in str(exc_info.value)
+        assert "is not a valid UUID" in str(exc_info.value)
+
+    def test_validate_artifact_identity_or_id_both_provided(self):
+        """Test validation with both valid values provided."""
+        # Should not raise any exception when both are valid
+        self.validator.validate_artifact_identity_or_id(
+            artifact_identity="pkg:npm/lodash@4.17.21",
+            artifact_id="550e8400-e29b-41d4-a716-446655440000",
+        )
+
+    def test_validate_artifact_identity_or_id_neither_provided(self):
+        """Test validation with neither value provided."""
+        # Should not raise any exception - no validation is performed when both are None
+        self.validator.validate_artifact_identity_or_id(artifact_identity=None, artifact_id=None)
+
+    # Tests for ArtifactValidator class methods (should raise ValueError)
+    def test_artifact_validator_validate_purl_valid(self):
+        """Test ArtifactValidator.validate_purl with valid PURL strings."""
+        valid_purls = [
+            "pkg:npm/lodash@4.17.21",
+            "pkg:pypi/requests@2.25.1",
+            "pkg:maven/org.apache.commons/commons-lang3@3.12.0",
+        ]
+
+        for purl in valid_purls:
+            assert self.validator.validate_purl(purl) is True
+
+    def test_artifact_validator_validate_uuid_valid(self):
+        """Test ArtifactValidator.validate_uuid with valid UUID strings."""
+        valid_uuids = [
+            "550e8400-e29b-41d4-a716-446655440000",
+            "6ba7b810-9dad-11d1-80b4-00c04fd430c8",
+        ]
+
+        for uuid_str in valid_uuids:
+            assert self.validator.validate_uuid(uuid_str) is True
+
+    def test_artifact_validator_validate_artifact_identity_or_id_valid(self):
+        """Test ArtifactValidator.validate_artifact_identity_or_id with valid inputs."""
+        # Should not raise any exception
+        self.validator.validate_artifact_identity_or_id(
+            artifact_identity="pkg:npm/lodash@4.17.21", artifact_id=None
+        )
+
+        self.validator.validate_artifact_identity_or_id(
+            artifact_identity=None, artifact_id="550e8400-e29b-41d4-a716-446655440000"
+        )
+
+    def test_artifact_validator_validate_identity_or_id_invalid_purl(self):
+        """Test ArtifactValidator.validate_artifact_identity_or_id with invalid PURL.
+
+        Should raise ValueError.
+        """
+        with pytest.raises(ValueError) as exc_info:
+            self.validator.validate_artifact_identity_or_id(
+                artifact_identity="not-a-purl", artifact_id=None
+            )
+        assert "Invalid artifact_identity" in str(exc_info.value)
+        assert "is not a valid PURL" in str(exc_info.value)
+
+    def test_artifact_validator_validate_identity_or_id_invalid_uuid(self):
+        """Test ArtifactValidator.validate_artifact_identity_or_id with invalid UUID.
+
+        Should raise ValueError.
+        """
+        with pytest.raises(ValueError) as exc_info:
+            self.validator.validate_artifact_identity_or_id(
+                artifact_identity=None, artifact_id="not-a-uuid"
+            )
+        assert "Invalid artifact_id" in str(exc_info.value)
+        assert "is not a valid UUID" in str(exc_info.value)
