Add url validaton for Rpm and Uln remote creation

Closes: pulp#3520

Author: pedro-psb

CHANGES/3520.misc

@@ -0,0 +1 @@
+Added url validation for Rpm and Uln Remotes creation.

pulp_rpm/app/serializers/repository.py

@@ -1,44 +1,32 @@
-from gettext import gettext as _
-
 import logging
+from gettext import gettext as _
 
 from django.conf import settings
 from jsonschema import Draft7Validator
-from rest_framework import serializers
-
-from pulpcore.plugin.util import get_domain
-from pulpcore.plugin.models import (
-    AsciiArmoredDetachedSigningService,
-    Remote,
-    Publication,
-)
+from pulpcore.plugin.models import AsciiArmoredDetachedSigningService, Publication, Remote
 from pulpcore.plugin.serializers import (
-    RelatedField,
     DetailRelatedField,
     DistributionSerializer,
     PublicationSerializer,
+    RelatedField,
     RemoteSerializer,
     RepositorySerializer,
     RepositorySyncURLSerializer,
     ValidateFieldsMixin,
 )
+from pulpcore.plugin.util import get_domain
+from rest_framework import serializers
 
 from pulp_rpm.app.constants import (
     ALLOWED_CHECKSUM_ERROR_MSG,
-    ALLOWED_PUBLISH_CHECKSUMS,
     ALLOWED_PUBLISH_CHECKSUM_ERROR_MSG,
+    ALLOWED_PUBLISH_CHECKSUMS,
     CHECKSUM_CHOICES,
+    COMPRESSION_CHOICES,
     SKIP_TYPES,
     SYNC_POLICY_CHOICES,
-    COMPRESSION_CHOICES,
-)
-from pulp_rpm.app.models import (
-    RpmDistribution,
-    RpmRemote,
-    RpmRepository,
-    RpmPublication,
-    UlnRemote,
 )
+from pulp_rpm.app.models import RpmDistribution, RpmPublication, RpmRemote, RpmRepository, UlnRemote
 from pulp_rpm.app.schema import COPY_CONFIG_SCHEMA
 
 
@@ -270,6 +258,15 @@ class RpmRemoteSerializer(RpmBaseRemoteSerializer):
         allow_null=True,
     )
 
+    def validate_url(self, value):
+        protocol = value.strip(":")[0]
+        allowed = ("http", "https", "file")
+        if protocol not in allowed:
+            raise serializers.ValidationError(
+                f"The url for RPM Remotes must start with: {allowed}."
+            )
+        return value
+
     class Meta:
         fields = RemoteSerializer.Meta.fields + ("sles_auth_token",)
         model = RpmRemote
@@ -309,6 +306,11 @@ class UlnRemoteSerializer(RpmBaseRemoteSerializer):
         allow_null=True,
     )
 
+    def validate_url(self, value):
+        if not value.startswith("uln"):
+            raise serializers.ValidationError("The url for ULN Remotes must start with: uln.")
+        return value
+
     class Meta:
         fields = RemoteSerializer.Meta.fields + ("uln_server_base_url",)
         model = UlnRemote

pulp_rpm/tests/functional/api/test_crud_remotes.py

@@ -1,15 +1,14 @@
 """Tests that CRUD rpm remotes."""
 
-import pytest
-
 from random import choice
 from uuid import uuid4
 
+import pytest
+from pulpcore.client.pulp_rpm.exceptions import ApiException
+
 from pulp_rpm.tests.functional.constants import DOWNLOAD_POLICIES
 from pulp_rpm.tests.functional.utils import gen_rpm_remote
 
-from pulpcore.client.pulp_rpm.exceptions import ApiException
-
 
 @pytest.mark.parallel
 def test_basic_crud_remote(rpm_rpmremote_api, rpm_rpmremote_factory, monitor_task):
@@ -127,6 +126,24 @@ def test_policy_update_changes(rpm_rpmremote_api, rpm_rpmremote_factory, monitor
         rpm_rpmremote_api.partial_update(remote["pulp_href"], {"policy": str(uuid4())})
 
 
+def test_raise_on_invalid_remote_url(
+    rpm_rpmremote_api, rpm_ulnremote_api, gen_object_with_cleanup, monitor_task
+):
+    # Cant create invalid RpmRemote
+    for protocol in ("uln", "sftp", "grpc"):
+        with pytest.raises(ApiException, match="The url for RPM Remotes must start with"):
+            body = _gen_verbose_remote()
+            body["url"] = f"{protocol}://some/rpm/remote"
+            gen_object_with_cleanup(rpm_rpmremote_api, body)
+
+    # Cant create invalid UlnRemote
+    for protocol in ("http", "https", "file"):
+        with pytest.raises(ApiException, match="The url for ULN Remotes must start with"):
+            body = _gen_verbose_remote()
+            body["url"] = f"{protocol}://some/uln/remote"
+            gen_object_with_cleanup(rpm_ulnremote_api, body)
+
+
 def _gen_verbose_remote():
     """Return a semi-random dict for use in defining a remote.