urls as unique

Author: dogboat

dojo/db_migrations/0259_locations.py

@@ -334,7 +334,7 @@ def endpoint_meta_import(file, product, create_endpoints, create_tags, create_me
         elif object_class == Location:
             endpoints = Location.objects.filter(url__host=host, products__product=product)
             if not endpoints.exists() and create_endpoints:
-                url = URL.objects.create(host=host)
+                url = URL.get_or_create_from_values(host=host)
                 url.location.associate_with_product(product)
                 endpoints = [url.location]
         meta = [(key, row.get(key)) for key in keys]

dojo/endpoint/utils.py

@@ -941,17 +941,7 @@ def get_value(field_name, default=None):
             for endpoint_url in endpoint_urls:
                 try:
                     if settings.V3_FEATURE_LOCATIONS:
-                        unsaved_url = URL.from_value(endpoint_url)
-                        saved_url, _ = URL.objects.get_or_create(
-                            protocol=unsaved_url.protocol,
-                            user_info=unsaved_url.user_info,
-                            host=unsaved_url.host,
-                            port=unsaved_url.port,
-                            path=unsaved_url.path,
-                            query=unsaved_url.query,
-                            fragment=unsaved_url.fragment,
-                            host_validation_failure=unsaved_url.host_validation_failure,
-                        )
+                        saved_url = URL.create_location_from_value(endpoint_url)
                         saved_url.location.associate_with_finding(finding)
                     else:
                         # TODO: Delete this after the move to Locations

dojo/finding/helper.py

@@ -24,23 +24,9 @@
 
 # test_notifications.py: Implement Locations
 class LocationManager:
-
-    def get_or_create_url(self, unsaved_url: URL) -> URL:
-        saved_url, _ = URL.objects.get_or_create(
-            protocol=unsaved_url.protocol,
-            user_info=unsaved_url.user_info,
-            host=unsaved_url.host,
-            port=unsaved_url.port,
-            path=unsaved_url.path,
-            query=unsaved_url.query,
-            fragment=unsaved_url.fragment,
-            host_validation_failure=unsaved_url.host_validation_failure,
-        )
-        return saved_url
-
     def get_or_create_location(self, unsaved_location: AbstractLocation) -> AbstractLocation | None:
         if isinstance(unsaved_location, URL):
-            return self.get_or_create_url(unsaved_location)
+            return URL.get_or_create_from_object(unsaved_location)
         logger.debug(f"IMPORT_SCAN: Unsupported location type: {type(unsaved_location)}")
         return None
 

dojo/importers/location_manager.py

@@ -85,7 +85,7 @@ def url_get_or_create(**kwargs):
     matches = list(qs.order_by("id")[:2])
     if not matches:
         # Most common case: nothing exists yet
-        return URL.objects.create(**kwargs), True
+        return URL.get_or_create_from_values(**kwargs), True
     if len(matches) == 1:
         # Common case: exactly one existing URL
         return matches[0], False

dojo/location/utils.py

@@ -36,9 +36,9 @@ def _endpoint_to_url(self, endpoint: Endpoint) -> Location:
         # Create the raw URL object first
         # This should create the location object as well
         url = URL.objects.get_or_create(
-            protocol=endpoint.protocol or "",
+            protocol=(endpoint.protocol or "").lower(),
             user_info=endpoint.userinfo or "",
-            host=endpoint.host,
+            host=(endpoint.host or "").lower(),
             port=endpoint.port,
             path=endpoint.path or "",
             query=endpoint.query or "",

dojo/management/commands/migrate_endpoints_to_locations.py

@@ -6,7 +6,8 @@
 
 from django.core.exceptions import ValidationError
 from django.core.validators import MaxValueValidator, MinValueValidator
-from django.db.models import BooleanField, CharField, Index, PositiveIntegerField
+from django.db.models import BooleanField, CharField, Index, PositiveIntegerField, UniqueConstraint
+from django.db.models.functions import Lower
 
 # Ignoring the N811 error as this is an external library and we cannot change its name
 # We are already using "URL" in our own code so we need to alias this import
@@ -60,6 +61,10 @@ def parse(self, value: str) -> ParsedUrl:
                 error_message = f"No host provided in URL: {parsed_url}"
                 raise ValidationError(error_message)
 
+        if parsed_url.port is not None and (parsed_url.port < 1 or parsed_url.port > 65535):
+            error_message = f"Invalid port: {parsed_url.port}"
+            raise ValidationError(error_message)
+
         return ParsedUrl(
             raw=value,
             protocol=parsed_url.scheme,
@@ -162,6 +167,19 @@ class Meta:
         verbose_name = "Locations - URL"
         verbose_name_plural = "Locations - URLs"
         indexes = (Index(fields=["host"]),)
+        constraints = [
+            UniqueConstraint(
+                Lower("protocol"),
+                "user_info",
+                Lower("host"),
+                "port",
+                "path",
+                "query",
+                "fragment",
+                "host_validation_failure",
+                name="url_unique",
+            ),
+        ]
 
     def manual_str(self):
         value = ""
@@ -206,10 +224,19 @@ def get_location_type(cls) -> str:
     def get_location_value(self) -> str:
         return str(self)
 
+    def normalize_url_parts(self):
+        self.clean_protocol()
+        self.clean_user_info()
+        self.clean_host()
+        self.clean_port()
+        self.clean_path()
+        self.clean_query()
+        self.clean_fragment()
+        self.clean_host_validation_failure()
+
     def pre_save_logic(self) -> None:
         """Allow for some pre save operations by other classes."""
-        # Set default port based on protocol if not provided
-        self.clean_port()
+        self.normalize_url_parts()
         super().pre_save_logic()
 
     @staticmethod
@@ -220,11 +247,25 @@ def _parse_string_value(value: str) -> ParsedUrl:
     def clean(self, *args: list, **kwargs: dict) -> None:
         """Validate the input supplied."""
         super().clean(*args, **kwargs)
-        # Ensure the full value is correctly parsable. If not, an exception will be raised
-        self.clean_port()
-        self.clean_path()
-        self.clean_query()
-        self.clean_fragment()
+        self.normalize_url_parts()
+
+    def clean_protocol(self) -> None:
+        if not self.protocol:
+            self.protocol = ""
+        else:
+            self.protocol = self.protocol.lower()
+
+    def clean_user_info(self):
+        if not self.user_info:
+            self.user_info = ""
+        else:
+            self.user_info = self.remove_null_bytes(self.user_info.strip())
+
+    def clean_host(self) -> None:
+        if not self.host:
+            self.host = ""
+        else:
+            self.host = self.host.lower()
 
     def clean_port(self) -> None:
         if self.port is None:
@@ -249,15 +290,54 @@ def clean_query(self) -> None:
         else:
             self.query = self.remove_null_bytes(self.query.strip().removeprefix("?"))
 
+    def clean_host_validation_failure(self):
+        self.host_validation_failure = bool(self.host_validation_failure)
+
     def remove_null_bytes(self, value: str) -> str:
         return value.replace("\x00", "%00")
 
+    @staticmethod
+    def get_or_create_from_object(url: URL) -> URL:
+        url.normalize_url_parts()
+        url, _ = URL.objects.get_or_create(
+            protocol=url.protocol,
+            user_info=url.user_info,
+            host=url.host,
+            port=url.port,
+            path=url.path,
+            query=url.query,
+            fragment=url.fragment,
+            host_validation_failure=url.host_validation_failure,
+        )
+        return url
+
+    @staticmethod
+    def get_or_create_from_values(
+        protocol=None,
+        user_info=None,
+        host=None,
+        port=None,
+        path=None,
+        query=None,
+        fragment=None,
+        host_validation_failure=None,
+    ) -> URL:
+        return URL.get_or_create_from_object(URL(
+            protocol=protocol,
+            user_info=user_info,
+            host=host,
+            port=port,
+            path=path,
+            query=query,
+            fragment=fragment,
+            host_validation_failure=host_validation_failure,
+        ))
+
     @staticmethod
     def create_location_from_value(value: str) -> URL:
         """Parse a string URL and return the resulting *persisted* URL Model."""
-        url = URL.from_value(value)
-        url.save()
-        return url
+        unsaved_url = URL.from_value(value)
+        return URL.get_or_create_from_object(unsaved_url)
 
     @staticmethod
     def from_value(value: str) -> URL:
@@ -280,5 +360,5 @@ def from_value(value: str) -> URL:
             query=query,
             fragment=fragment,
         )
-        url.full_clean()
+        url.normalize_url_parts()
         return url

dojo/url/models.py

@@ -305,8 +305,7 @@ def create_and_associate_endpoints(self, finding, *endpoint_defs: dict):
             for e_def in endpoint_defs:
                 e_def.pop("product", None)
                 e_def.pop("finding", None)
-                url = URL(**e_def)
-                url.save()
+                url = URL.get_or_create_from_values(**e_def)
                 url.location.associate_with_finding(finding)
         else:
             # TODO: Delete this after the move to Locations

unittests/test_deduplication_logic.py

@@ -73,7 +73,7 @@ def test_inspector_ec2(self):
             self.assertEqual("CVE-2022-3643", finding.unsaved_vulnerability_ids[0])
             self.assertEqual("- Update kernel-4.14.301\n\t- yum update kernel\n", finding.mitigation)
             location = self.get_unsaved_locations(finding)[0]
-            self.assertEqual("AwsEc2Instance_arn_aws_ec2_us-east-1_XXXXXXXXXXXX_i-11111111111111111", location.host)
+            self.assertEqual("AwsEc2Instance_arn_aws_ec2_us-east-1_XXXXXXXXXXXX_i-11111111111111111".lower(), location.host.lower())
 
     def test_inspector_ec2_with_no_vulnerabilities(self):
         with sample_path("inspector_ec2_cve_no_vulnerabilities.json").open(encoding="utf-8") as test_file:
@@ -98,7 +98,7 @@ def test_inspector_ec2_ghsa(self):
             self.assertSetEqual({"CVE-2023-34256", "GHSA-p98r-538v-jgw5"}, set(finding.unsaved_vulnerability_ids))
             self.assertEqual("https://github.com/bottlerocket-os/bottlerocket/security/advisories/GHSA-p98r-538v-jgw5", finding.references)
             location = self.get_unsaved_locations(finding)[0]
-            self.assertEqual("AwsEc2Instance_arn_aws_ec2_eu-central-1_012345678912_instance_i-07c11cc535d830123", location.host)
+            self.assertEqual("AwsEc2Instance_arn_aws_ec2_eu-central-1_012345678912_instance_i-07c11cc535d830123".lower(), location.host.lower())
 
     def test_inspector_ecr(self):
         with sample_path("inspector_ecr.json").open(encoding="utf-8") as test_file:
@@ -116,7 +116,7 @@ def test_inspector_ecr(self):
             self.assertIn("Repository: repo-os", finding.impact)
             self.assertEqual(0.0014, finding.epss_score)
             location = self.get_unsaved_locations(finding)[0]
-            self.assertEqual("AwsEcrContainerImage_arn_aws_ecr_eu-central-1_123456789012_repository_repo-os_sha256_af965ef68c78374a5f987fce98c0ddfa45801df2395bf012c50b863e65978d74", location.host)
+            self.assertEqual("AwsEcrContainerImage_arn_aws_ecr_eu-central-1_123456789012_repository_repo-os_sha256_af965ef68c78374a5f987fce98c0ddfa45801df2395bf012c50b863e65978d74".lower(), location.host.lower())
 
     def test_guardduty(self):
         with sample_path("guardduty.json").open(encoding="utf-8") as test_file:
@@ -133,7 +133,7 @@ def test_guardduty(self):
             self.assertEqual("User AssumedRole : 123123123 is anomalously invoking APIs commonly used in Discovery tactics. - Resource: 123123123", finding.title)
             self.assertEqual("TTPs/Discovery/IAMUser-AnomalousBehavior\n[https://docs.aws.amazon.com/guardduty/latest/ug/guardduty_finding-types-active.html](https://docs.aws.amazon.com/guardduty/latest/ug/guardduty_finding-types-active.html)", finding.mitigation)
             location = self.get_unsaved_locations(findings[0])[0]
-            self.assertEqual("AwsEc2Instance_arn_aws_ec2_us-east-1_123456789012_instance_i-1234567890", location.host)
+            self.assertEqual("AwsEc2Instance_arn_aws_ec2_us-east-1_123456789012_instance_i-1234567890".lower(), location.host.lower())
             self.assertEqual("This is a GuardDuty Finding\nAPIs commonly used in Discovery tactics were invoked by user AssumedRole : 123123123, under anomalous circumstances. Such activity is not typically seen from this user.\n**AWS Finding ARN:** arn:aws:guardduty:us-east-1:123456789012:detector/123456789/finding/2123123123123\n**SourceURL:** [https://us-east-1.console.aws.amazon.com/guardduty/home?region=us-east-1#/findings?macros=current&fId=2123123123123](https://us-east-1.console.aws.amazon.com/guardduty/home?region=us-east-1#/findings?macros=current&fId=2123123123123)\n**AwsAccountId:** 123456789012\n**Region:** us-east-1\n**Generator ID:** arn:aws:guardduty:us-east-1:123456789012:detector/123456789\n", finding.description)
 
     def test_issue_10956(self):

unittests/tools/test_awssecurityhub_parser.py

@@ -98,7 +98,7 @@ def test_parser_defender_issue_11217(self):
         self.assertEqual(1, len(findings))
         finding = findings[0]
         self.assertEqual("Medium", finding.severity)
-        self.assertEqual("Max_Mustermann_iPadAir_17zoll__2ndgeneration_", self.get_unsaved_locations(finding)[0].host)
+        self.assertEqual("Max_Mustermann_iPadAir_17zoll__2ndgeneration_".lower(), self.get_unsaved_locations(finding)[0].host.lower())
 
     def test_parser_defender_error_handling(self):
         """https://github.com/DefectDojo/django-DefectDojo/issues/11896 handle missing values properly, i.e. defenderAvStatus"""