fix: add validation of host input for registry credential (#1140)

Author: jshimkus-rh

src/aap_eda/api/serializers/eda_credential.py

@@ -120,7 +120,11 @@ def validate(self, data):
         )
 
         inputs = data.get("inputs", {})
-        errors = validate_inputs(credential_type.inputs, inputs)
+        errors = validate_inputs(
+            credential_type,
+            credential_type.inputs,
+            inputs,
+        )
         if bool(errors):
             raise serializers.ValidationError(errors)
 
@@ -153,7 +157,11 @@ def validate(self, data):
         if self.partial and not bool(inputs):
             return data
 
-        errors = validate_inputs(credential_type.inputs, inputs)
+        errors = validate_inputs(
+            credential_type,
+            credential_type.inputs,
+            inputs,
+        )
         if bool(errors):
             raise serializers.ValidationError(errors)
 

src/aap_eda/core/utils/credentials.py

@@ -14,14 +14,21 @@
 
 import re
 import tempfile
+import typing
 
 import gnupg
 import jinja2
+import validators
 import yaml
 from django.core.exceptions import ValidationError
 from jinja2.nativetypes import NativeTemplate
+from rest_framework import serializers
 
 from aap_eda.api.constants import EDA_SERVER_VAULT_LABEL
+from aap_eda.core import enums
+
+if typing.TYPE_CHECKING:
+    from aap_eda.core import models
 from aap_eda.core.utils.awx import validate_ssh_private_key
 
 ENCRYPTED_STRING = "$encrypted$"
@@ -80,7 +87,11 @@ def inputs_from_store(inputs: str) -> dict:
     return yaml.safe_load(inputs)
 
 
-def validate_inputs(schema: dict, inputs: dict) -> dict:
+def validate_inputs(
+    credential_type: "models.CredentialType",
+    schema: dict,
+    inputs: dict,
+) -> dict:
     """Validate user inputs against credential schema.
 
     Sample output:
@@ -137,6 +148,17 @@ def validate_inputs(schema: dict, inputs: dict) -> dict:
                 else:
                     errors[display_field] = result
 
+        # We apply particular requirements on "host" when it is
+        # associated with a container registry.
+        if (
+            (credential_type.name == enums.DefaultCredentialType.REGISTRY)
+            and (field == "host")
+            and user_input
+        ):
+            result = _validate_registry_host_name(user_input)
+            if bool(result):
+                errors[display_field] = result
+
         if field == "gpg_public_key":
             result = _validate_gpg_public_key(user_input)
             if bool(result):
@@ -278,6 +300,27 @@ def validate_injectors(schema: dict, injectors: dict) -> dict:
     return {"injectors": errors} if bool(errors) else {}
 
 
+def validate_registry_host_name(host: str) -> None:
+    errors = _validate_registry_host_name(host)
+    if bool(errors):
+        raise serializers.ValidationError(f"invalid host name: {host}")
+
+
+def _validate_registry_host_name(host: str) -> list[str]:
+    errors = []
+
+    # Yes, validators returns (not throws) an exception if the argument doesn't
+    # pass muster (it returns True otherwise).  Consequently we have to check
+    # the class of the return to know what happened and if it's not validators'
+    # validation exception raise whatever the heck it is.
+    validity = validators.hostname(host)
+    if isinstance(validity, Exception):
+        if not isinstance(validity, validators.ValidationError):
+            raise
+        errors.append("Host format invalid")
+    return errors
+
+
 def _get_id_fields(schema: dict) -> list[str]:
     fields = schema.get("fields", [])
 

src/aap_eda/core/validators.py

@@ -17,14 +17,16 @@
 import typing as tp
 import urllib
 
-import validators
 import yaml
 from django.conf import settings
 from django.utils.translation import gettext_lazy as _
 from rest_framework import serializers
 
 from aap_eda.core import enums, models
-from aap_eda.core.utils.credentials import validate_schema
+from aap_eda.core.utils.credentials import (
+    validate_registry_host_name,
+    validate_schema,
+)
 from aap_eda.core.utils.k8s_service_name import is_rfc_1035_compliant
 
 logger = logging.getLogger(__name__)
@@ -69,8 +71,7 @@ def check_if_de_valid(image_url: str, eda_credential_id: int):
     # We split the image url on the first slash into the host and path.  The
     # path is further split into a name and tag on the rightmost colon.
     #
-    # THe host portion is validated using the validators package and the path
-    # and tag are validated using the OCI regexes for each.
+    # The path and tag are validated using the OCI regexes for each.
     split = image_url.split("/", 1)
     host = split[0]
     path = split[1] if len(split) > 1 else None
@@ -81,14 +82,11 @@ def check_if_de_valid(image_url: str, eda_credential_id: int):
             % {"image_url": image_url}
         )
 
-    # Yes, validators returns (not throws) an exception if the argument doesn't
-    # pass muster (it returns True otherwise).  Consequently we have to check
-    # the class of the return to know what happened and if it's not validators'
-    # validation exception raise whatever the heck it is.
-    validity = validators.hostname(host)
-    if isinstance(validity, Exception):
-        if not isinstance(validity, validators.ValidationError):
-            raise
+    try:
+        validate_registry_host_name(host)
+    except serializers.ValidationError:
+        # We raise our own instance of this exception in order to assert
+        # control over the format of the message.
         raise serializers.ValidationError(
             _(
                 "Image url %(image_url)s is malformed; "

tests/conftest.py

@@ -17,7 +17,11 @@
 import pytest
 from django.conf import settings
 
-from aap_eda.core import models
+from aap_eda.core import enums, models
+from aap_eda.core.management.commands.create_initial_data import (
+    CREDENTIAL_TYPES,
+    populate_credential_types,
+)
 from aap_eda.settings import default
 
 
@@ -46,6 +50,24 @@ def default_organization() -> models.Organization:
     )[0]
 
 
+#################################################################
+# DB
+#################################################################
+@pytest.fixture
+def preseed_credential_types(
+    default_organization: models.Organization,
+) -> list[models.CredentialType]:
+    """Preseed Credential Types."""
+    return populate_credential_types(CREDENTIAL_TYPES)
+
+
+@pytest.fixture
+def aap_credential_type(preseed_credential_types):
+    return models.CredentialType.objects.get(
+        name=enums.DefaultCredentialType.AAP
+    )
+
+
 #################################################################
 # Redis
 #################################################################

tests/integration/api/test_eda_credential.py

@@ -206,6 +206,67 @@ def test_create_eda_credential_with_gpg_key_data(
         assert message.startswith(status_message)
 
 
+@pytest.mark.parametrize(
+    ("status_code", "hostname"),
+    [
+        (
+            status.HTTP_201_CREATED,
+            "validname",
+        ),
+        (
+            status.HTTP_201_CREATED,
+            "valid-name",
+        ),
+        (
+            status.HTTP_201_CREATED,
+            "valid.name",
+        ),
+        (
+            status.HTTP_400_BAD_REQUEST,
+            "invalid name",
+        ),
+        (
+            status.HTTP_400_BAD_REQUEST,
+            "invalid/name",
+        ),
+        (
+            status.HTTP_400_BAD_REQUEST,
+            "invalid@name",
+        ),
+        (
+            status.HTTP_400_BAD_REQUEST,
+            "https://invalid.name",
+        ),
+    ],
+)
+@pytest.mark.django_db
+def test_create_registry_eda_credential_with_various_host_names(
+    admin_client: APIClient,
+    default_organization: models.Organization,
+    preseed_credential_types,
+    status_code,
+    hostname,
+):
+    credential_type = models.CredentialType.objects.get(
+        name=enums.DefaultCredentialType.REGISTRY
+    )
+
+    data_in = {
+        "name": f"eda-credential-{credential_type}",
+        "inputs": {
+            "host": hostname,
+        },
+        "credential_type_id": credential_type.id,
+        "organization_id": default_organization.id,
+    }
+    response = admin_client.post(
+        f"{api_url_v1}/eda-credentials/", data=data_in
+    )
+    assert response.status_code == status_code
+    if status_code == status.HTTP_400_BAD_REQUEST:
+        assert "Host format invalid" in response.data["inputs.host"]
+
+
 @pytest.mark.parametrize(
     ("credential_type", "status_code", "key", "error_message"),
     [
@@ -551,11 +612,11 @@ def test_partial_update_eda_credential_type_not_changed(
                 "verify_ssl": True,
             },
             {
-                "host": "new host",
+                "host": "new-host",
                 "verify_ssl": False,
             },
             {
-                "host": "new host",
+                "host": "new-host",
                 "username": "user name",
                 "password": ENCRYPTED_STRING,
                 "oauth_token": ENCRYPTED_STRING,
@@ -605,12 +666,12 @@ def test_partial_update_eda_credential_type_not_changed(
                 "verify_ssl": True,
             },
             {
-                "host": "new host",
+                "host": "new-host",
                 "username": "new user name",
                 "verify_ssl": False,
             },
             {
-                "host": "new host",
+                "host": "new-host",
                 "username": "new user name",
                 "password": ENCRYPTED_STRING,
                 "verify_ssl": False,
@@ -625,12 +686,12 @@ def test_partial_update_eda_credential_type_not_changed(
                 "verify_ssl": True,
             },
             {
-                "host": "new host",
+                "host": "new-host",
                 "username": "new user name",
                 "password": "password",
             },
             {
-                "host": "new host",
+                "host": "new-host",
                 "username": "new user name",
                 "password": ENCRYPTED_STRING,
                 "verify_ssl": True,

tests/integration/conftest.py

@@ -27,10 +27,6 @@
 
 from aap_eda.core import enums, models
 from aap_eda.core.management.commands import create_initial_data
-from aap_eda.core.management.commands.create_initial_data import (
-    CREDENTIAL_TYPES,
-    populate_credential_types,
-)
 from aap_eda.core.tasking import Queue, get_redis_client
 from aap_eda.core.utils.credentials import inputs_to_store
 from aap_eda.services.activation.engine.common import ContainerEngine
@@ -1035,14 +1031,6 @@ def credential_payload(
     }
 
 
-@pytest.fixture
-def preseed_credential_types(
-    default_organization: models.Organization,
-) -> list[models.CredentialType]:
-    """Preseed Credential Types."""
-    return populate_credential_types(CREDENTIAL_TYPES)
-
-
 @pytest.fixture
 def default_gpg_credential(
     default_organization: models.Organization,