Skip to content

Commit

Permalink
fix: check that DE image url is well-formed
Browse files Browse the repository at this point in the history
  • Loading branch information
jshimkus-rh committed Oct 25, 2024
1 parent 7ede6bc commit 1da7fb5
Show file tree
Hide file tree
Showing 2 changed files with 77 additions and 0 deletions.
27 changes: 27 additions & 0 deletions src/aap_eda/core/validators.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@
import hashlib
import logging
import typing as tp
import urllib

import yaml
from django.conf import settings
Expand Down Expand Up @@ -57,6 +58,32 @@ def check_if_de_exists(decision_environment_id: int) -> int:


def check_if_de_valid(image_url: str, eda_credential_id: int):
parsed_url = urllib.parse.urlparse(image_url)
base_message = f"Image url {image_url} is malformed; "
if parsed_url.scheme:
raise serializers.ValidationError(base_message + "scheme not allowed")
if parsed_url.netloc:
raise serializers.ValidationError(
base_message + "network location not allowed"
)
if parsed_url.params:
raise serializers.ValidationError(
base_message + "parameters not allowed"
)
if parsed_url.query:
raise serializers.ValidationError(base_message + "query not allowed")
if parsed_url.fragment:
raise serializers.ValidationError(
base_message + "fragment not allowed"
)

# Now that we've passed the above and know there's no netloc check if the
# path starts with a "/"; it should not.
if parsed_url.path.startswith("/"):
raise serializers.ValidationError(
base_message + 'must not start with "/"'
)

credential = get_credential_if_exists(eda_credential_id)
inputs = yaml.safe_load(credential.inputs.get_secret_value())
host = inputs.get("host")
Expand Down
50 changes: 50 additions & 0 deletions tests/integration/api/test_decision_environment.py
Original file line number Diff line number Diff line change
Expand Up @@ -97,6 +97,56 @@ def test_create_decision_environment(
assert status_message in response.data["eda_credential_id"]


@pytest.mark.parametrize(
("image_url", "unallowed"),
[
("http://registry.com/img1:tag1", "scheme not allowed"),
("//registry.com/img1:tag1", "network location not allowed"),
("registry.com/img1:tag;parameter", "parameters not allowed"),
("registry.com/img1:tag1?query", "query not allowed"),
("registry.com/img1:tag#fragment", "fragment not allowed"),
("/registry.com/img1:tag1", 'must not start with "/"'),
],
)
@pytest.mark.django_db
def test_create_decision_environment_with_malformed_url(
image_url,
unallowed,
default_organization: models.Organization,
admin_client: APIClient,
preseed_credential_types,
):
credential_type = models.CredentialType.objects.get(
name=enums.DefaultCredentialType.REGISTRY
)
credential = models.EdaCredential.objects.create(
name="eda-credential",
description="Default Credential",
credential_type=credential_type,
organization=default_organization,
inputs=inputs_to_store(
{
"username": "dummy-user",
"password": "dummy-password",
"host": "registry.com",
}
),
)
data_in = {
"name": "de1",
"description": "desc here",
"image_url": image_url,
"organization_id": default_organization.id,
"eda_credential_id": credential.id,
}
response = admin_client.post(
f"{api_url_v1}/decision-environments/", data=data_in
)
assert response.status_code == status.HTTP_400_BAD_REQUEST
errors = response.data.get("non_field_errors")
assert f"Image url {image_url} is malformed; {unallowed}" in str(errors)


@pytest.mark.parametrize(
("credential_inputs", "status_code", "status_message"),
[
Expand Down

0 comments on commit 1da7fb5

Please sign in to comment.