Base path db constaint

pulpcore/app/migrations/0155_create_rel_path_domains.py

@@ -0,0 +1,23 @@
+# Generated by Django 5.2.17 on 2026-08-07 12:29
+
+from django.db import migrations
+
+
+CREATE_REL_PATH_DOMAINS = """
+CREATE DOMAIN "relative_path" AS text CHECK ('/' || VALUE || '/' !~ '[\n\r\s\t\?#]|(/\.{0,2}/)');
+"""
+
+REMOVE_REL_PATH_DOMAINS = """
+DROP DOMAIN IF EXISTS "relative_path";
+"""
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('core', '0154_task_api_version'),
+    ]
+
+    operations = [
+        migrations.RunSQL(sql=CREATE_REL_PATH_DOMAINS, reverse_sql=REMOVE_REL_PATH_DOMAINS),
+    ]

pulpcore/app/migrations/0156_alter_contentartifact_relative_path_and_more.py

@@ -0,0 +1,40 @@
+# Generated by Django 5.2.15 on 2026-08-10 10:26
+
+import django.contrib.postgres.indexes
+import django.db.models.expressions
+import pulpcore.app.models.fields
+from django.db import migrations
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('core', '0155_create_rel_path_domains'),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name='contentartifact',
+            name='relative_path',
+            field=pulpcore.app.models.fields.RelativePathField(),
+        ),
+        migrations.AlterField(
+            model_name='distribution',
+            name='base_path',
+            field=pulpcore.app.models.fields.RelativePathField(),
+        ),
+        migrations.AlterField(
+            model_name='publishedartifact',
+            name='relative_path',
+            field=pulpcore.app.models.fields.RelativePathField(),
+        ),
+        migrations.AlterField(
+            model_name='publishedmetadata',
+            name='relative_path',
+            field=pulpcore.app.models.fields.RelativePathField(),
+        ),
+        migrations.AddIndex(
+            model_name='distribution',
+            index=django.contrib.postgres.indexes.SpGistIndex(django.contrib.postgres.indexes.OpClass(django.db.models.expressions.RawSQL('"base_path" || \'/\'', ()), name='text_ops'), include=('pulp_domain',), name='core_distribution_base_path_slash'),
+        ),
+    ]

pulpcore/app/migrations/0157_distribution_base_path_constraint.py

@@ -0,0 +1,97 @@
+# Generated by Django 5.2.17 on 2026-08-11 10:34
+
+from django.db import migrations
+
+
+ADD_TRIGGER = """
+CREATE OR REPLACE FUNCTION path_prefixes("path" text)
+RETURNS text[] AS $$
+DECLARE
+        segment text;
+        segments text[] := string_to_array(path, '/');
+        prefix text := '';
+        prefixes text[];
+BEGIN
+        prefix := segments[1];
+        FOREACH segment IN ARRAY segments[2:]
+        LOOP
+                prefixes := prefixes || prefix;
+                prefix := prefix || '/' || segment;
+        END LOOP;
+        RETURN prefixes;
+END;
+$$ LANGUAGE plpgsql
+IMMUTABLE
+RETURNS NULL ON NULL INPUT
+PARALLEL SAFE;
+
+CREATE OR REPLACE FUNCTION "check_core_distribution_base_path_prefix_free" ()
+RETURNS TRIGGER AS $$
+DECLARE
+        base_path_slash text := new.base_path || '/';
+BEGIN
+        -- Check that no base_path is a prefix of another.
+        -- The normalization of "/" ensures that a simple string comparison actually suffice.
+
+        -- ^@ only uses an index when the index expression is on the left (pg18).
+        PERFORM 1 FROM "core_distribution"
+                WHERE
+                        "pulp_id" != new."pulp_id"
+                        AND
+                        "pulp_domain_id" = new."pulp_domain_id"
+                        AND
+                        ("base_path" || '/') ^@ base_path_slash
+                LIMIT 1;
+        IF FOUND THEN
+                RAISE EXCEPTION '"%" overlaps with existing base_path.', new."base_path";
+        END IF;
+
+        -- This variant however uses the existing uniqueness index.
+        PERFORM 1 FROM "core_distribution"
+                WHERE
+                        "pulp_id" != new."pulp_id"
+                        AND
+                        "pulp_domain_id" = new."pulp_domain_id"
+                        AND
+                        "base_path" = ANY(path_prefixes(new."base_path"))
+                LIMIT 1;
+        IF FOUND THEN
+                RAISE EXCEPTION '"%" overlaps with existing base_path.', new."base_path";
+        END IF;
+
+        RETURN new;
+END
+$$ LANGUAGE plpgsql;
+
+CREATE CONSTRAINT TRIGGER "insert_base_path_overlap_constraint"
+        AFTER INSERT
+        ON "core_distribution"
+        FOR EACH ROW
+        WHEN (new."base_path" IS NOT NULL)
+        EXECUTE FUNCTION "check_core_distribution_base_path_prefix_free" ();
+
+CREATE CONSTRAINT TRIGGER "update_base_path_overlap_constraint"
+        AFTER UPDATE
+        ON "core_distribution"
+        FOR EACH ROW
+        WHEN (new."base_path" IS NOT NULL AND new."base_path" != old."base_path")
+        EXECUTE FUNCTION "check_core_distribution_base_path_prefix_free" ();
+"""
+
+REMOVE_TRIGGER = """
+DROP TRIGGER IF EXISTS "update_base_path_overlap_constraint" ON "core_distribution";
+DROP TRIGGER IF EXISTS "insert_base_path_overlap_constraint" ON "core_distribution";
+DROP FUNCTION IF EXISTS "check_core_distribution_base_path_prefix_free";
+DROP FUNCTION IF EXISTS "path_prefixes";
+"""
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('core', '0156_alter_contentartifact_relative_path_and_more'),
+    ]
+
+    operations = [
+        migrations.RunSQL(sql=ADD_TRIGGER, reverse_sql=REMOVE_TRIGGER, elidable=False),
+    ]

pulpcore/app/models/content.py

@@ -26,6 +26,7 @@
 
 from pulpcore.app import pulp_hashlib
 from pulpcore.app.models import BaseModel, MasterModel, fields, storage
+from pulpcore.app.models.fields import RelativePathField
 from pulpcore.app.util import get_domain_pk, gpg_verify
 from pulpcore.constants import ALL_KNOWN_CONTENT_CHECKSUMS
 from pulpcore.exceptions import (
@@ -656,7 +657,7 @@ class ContentArtifact(BaseModel, QueryMixin):
         Artifact, on_delete=models.PROTECT, null=True, related_name="content_memberships"
     )
     content = models.ForeignKey(Content, on_delete=models.CASCADE)
-    relative_path = models.TextField()
+    relative_path = RelativePathField()
 
     objects = BulkCreateManager()
 

pulpcore/app/models/fields.py

@@ -155,6 +155,11 @@ def from_db_value(self, value, expression, connection):
         return value
 
 
+class RelativePathField(TextField):
+    def db_type(self, connection):
+        return "relative_path"
+
+
 @Field.register_lookup
 class NotEqualLookup(Lookup):
     # this is copied from https://docs.djangoproject.com/en/3.2/howto/custom-lookups/

pulpcore/app/models/publication.py

@@ -13,6 +13,7 @@
 from aiohttp.web_exceptions import HTTPNotFound
 from django.conf import settings
 from django.contrib.postgres.fields import HStoreField
+from django.contrib.postgres.indexes import OpClass, SpGistIndex
 from django.db import DatabaseError, IntegrityError, models, transaction
 from django.utils import timezone
 from django_lifecycle import AFTER_CREATE, AFTER_UPDATE, BEFORE_DELETE, hook
@@ -21,6 +22,7 @@
 
 from pulpcore.app.files import PulpTemporaryUploadedFile
 from pulpcore.app.models import AutoAddObjPermsMixin
+from pulpcore.app.models.fields import RelativePathField
 from pulpcore.app.util import cache_key, get_domain_pk, get_url, retain_distributed_pub_enabled
 from pulpcore.cache import Cache
 from pulpcore.responses import ArtifactResponse
@@ -270,7 +272,7 @@ class PublishedArtifact(BaseModel):
         publication (models.ForeignKey): The publication in which the artifact is included.
     """
 
-    relative_path = models.TextField()
+    relative_path = RelativePathField()
 
     content_artifact = models.ForeignKey("ContentArtifact", on_delete=models.CASCADE)
     publication = models.ForeignKey(Publication, on_delete=models.CASCADE)
@@ -293,7 +295,7 @@ class PublishedMetadata(Content):
 
     TYPE = "publishedmetadata"
 
-    relative_path = models.TextField()
+    relative_path = RelativePathField()
 
     publication = models.ForeignKey(Publication, on_delete=models.CASCADE)
 
@@ -642,7 +644,7 @@ class Distribution(MasterModel):
 
     name = models.TextField(db_index=True)
     pulp_labels = HStoreField(default=dict)
-    base_path = models.TextField()
+    base_path = RelativePathField()
     pulp_domain = models.ForeignKey("Domain", default=get_domain_pk, on_delete=models.PROTECT)
     hidden = models.BooleanField(default=False, null=True)
     checkpoint = models.BooleanField(default=False)
@@ -657,6 +659,15 @@ class Distribution(MasterModel):
 
     class Meta:
         unique_together = (("name", "pulp_domain"), ("base_path", "pulp_domain"))
+        # Do not use `models.functions.Concat`!
+        # The index expression needs to match exactly and the ORM is trying to be clever.
+        indexes = [
+            SpGistIndex(
+                OpClass(models.expressions.RawSQL("\"base_path\" || '/'", ()), name="text_ops"),
+                include=("pulp_domain",),
+                name="core_distribution_base_path_slash",
+            ),
+        ]
 
     def get_repository_publication_and_version(self):
         """

pulpcore/app/serializers/__init__.py

@@ -26,6 +26,8 @@
 )
 from .fields import (
     BaseURLField,
+    ContentArtifactChecksumField,
+    ContentArtifactsField,
     ExportsIdentityFromExporterField,
     ExportRelatedField,
     ExportIdentityField,
@@ -37,6 +39,7 @@
     LatestVersionField,
     PgpKeyFingerprintField,
     PulpLabelsField,
+    RelativePathField,
     SingleContentArtifactField,
     RepositoryVersionsIdentityFromRepositoryField,
     RepositoryVersionRelatedField,

pulpcore/app/serializers/base.py

@@ -6,12 +6,10 @@
 from gettext import gettext as _
 from logging import getLogger
 from typing import List, TypedDict
-from urllib.parse import urljoin
 
 from cryptography.x509 import load_pem_x509_certificate
 from django.conf import settings
 from django.core.exceptions import ObjectDoesNotExist
-from django.core.validators import URLValidator
 from django.db import IntegrityError
 from django.db.models import Model
 from django.urls.exceptions import NoReverseMatch
@@ -475,37 +473,6 @@ class Meta:
         read_only=True,
     )
 
-    def _validate_relative_path(self, path):
-        """
-        Validate a relative path (eg from a url) to ensure it forms a valid url and does not begin
-        or end with slashes nor contain spaces
-
-        Args:
-            path (str): A relative path to validate
-
-        Returns:
-            str: the validated path
-
-        Raises:
-            django.core.exceptions.ValidationError: if the relative path is invalid
-
-        """
-        # in order to use django's URLValidator we need to construct a full url
-        base = "http://localhost"  # use a scheme/hostname we know are valid
-
-        if " " in path:
-            raise serializers.ValidationError(detail=_("Relative path cannot contain spaces."))
-
-        validate = URLValidator()
-        validate(urljoin(base, path))
-
-        if path != path.strip("/"):
-            raise serializers.ValidationError(
-                detail=_("Relative path cannot begin or end with slashes.")
-            )
-
-        return path
-
     def save(self, **kwargs):
         try:
             return super().save(**kwargs)