Fix pull-through distribution base paths

CHANGES/2494.bugfix

@@ -0,0 +1 @@
+Prevent container distributions from overlapping other distribution base paths. Pull-through distributions **NEED** to be fixed before upgrading pulpcore; run `pulpcore-manager container-repair-pull-through-distributions` to repair existing distributions.

pulp_container/app/authorization.py

@@ -10,7 +10,6 @@
 from cryptography.hazmat.backends import default_backend
 from cryptography.hazmat.primitives import serialization
 from django.conf import settings
-from django.db.models import F, Value
 from django.http import HttpRequest
 from rest_framework.request import Request
 
@@ -21,8 +20,8 @@
 from pulp_container.app.models import (
     ContainerDistribution,
     ContainerNamespace,
-    ContainerPullThroughDistribution,
 )
+from pulp_container.app.pull_through import get_pull_through_distribution
 
 TOKEN_EXPIRATION_TIME = getattr(settings, "TOKEN_EXPIRATION_TIME", 300)
 
@@ -209,15 +208,6 @@ def generate_claim_set(issuer, issued_at, subject, audience, access):
         }
 
 
-def get_pull_through_distribution(path, domain):
-    return (
-        ContainerPullThroughDistribution.objects.annotate(path=Value(path))
-        .filter(pulp_domain=domain, path__startswith=F("base_path"))
-        .order_by("-base_path")
-        .first()
-    )
-
-
 class PermissionChecker:
     def __init__(self, user):
         self.user = user

pulp_container/app/cache.py

@@ -1,11 +1,14 @@
 from django.core.exceptions import ObjectDoesNotExist
-from django.db.models import F, Value
 
 from pulpcore.plugin.cache import AsyncContentCache, CacheKeys, SyncContentCache
 from pulpcore.plugin.util import cache_key, get_domain
 
 from pulp_container.app.exceptions import RepositoryNotFound
-from pulp_container.app.models import ContainerDistribution, ContainerPullThroughDistribution
+from pulp_container.app.models import ContainerDistribution
+from pulp_container.app.pull_through import (
+    get_pull_through_distribution,
+    get_pull_through_distribution_path,
+)
 
 ACCEPT_HEADER_KEY = "accept_header"
 QUERY_KEY = "query"
@@ -83,15 +86,12 @@ def find_base_path_cached(request, cached):
         try:
             distro = ContainerDistribution.objects.get(base_path=path, pulp_domain=domain)
         except ObjectDoesNotExist:
-            distro = (
-                ContainerPullThroughDistribution.objects.annotate(path=Value(path))
-                .filter(path__startswith=F("base_path"), pulp_domain=domain)
-                .order_by("-base_path")
-                .first()
-            )
+            distro = get_pull_through_distribution(path, domain)
             if not distro:
                 raise RepositoryNotFound(name=path)
 
+            return cache_key(get_pull_through_distribution_path(distro))
+
         return cache_key(distro.base_path)
 
 

pulp_container/app/checks.py

@@ -1,6 +1,11 @@
 from django.conf import settings
 from django.core.checks import Error as CheckError
+from django.core.checks import Warning as CheckWarning
 from django.core.checks import register
+from django.db import OperationalError, ProgrammingError
+
+from pulp_container.app.models import ContainerPullThroughDistribution
+from pulp_container.app.pull_through import PULL_THROUGH_DISTRIBUTION_MARKER
 
 
 @register(deploy=True)
@@ -45,3 +50,28 @@ def container_settings_check(app_configs, **kwargs):
         )
 
     return errors
+
+
+@register()
+def pull_through_distribution_check(app_configs, **kwargs):
+    """Warn when legacy pull-through distributions need repair."""
+    try:
+        needs_repair = ContainerPullThroughDistribution.objects.exclude(
+            pulp_labels__contains=PULL_THROUGH_DISTRIBUTION_MARKER
+        ).exists()
+    except (OperationalError, ProgrammingError):
+        return []
+
+    if not needs_repair:
+        return []
+
+    return [
+        CheckWarning(
+            "Pull-through distributions need base-path repair.",
+            hint=(
+                "Run 'pulpcore-manager container-repair-pull-through-distributions' and "
+                "resolve any distributions it reports."
+            ),
+            id="pulp_container.W001",
+        )
+    ]

pulp_container/app/management/commands/container-repair-pull-through-distributions.py

@@ -0,0 +1,80 @@
+from gettext import gettext as _
+from uuid import uuid4
+
+from django.conf import settings
+from django.core.management import BaseCommand
+from django.db import transaction
+
+from pulpcore.plugin.cache import SyncContentCache
+
+from pulp_container.app.models import ContainerPullThroughDistribution
+from pulp_container.app.pull_through import (
+    PULL_THROUGH_DISTRIBUTION_MARKER,
+    find_container_distribution_overlaps,
+)
+
+
+class Command(BaseCommand):
+    """Repair pull-through distributions whose registry path is stored as their base path."""
+
+    help = _("Repair pull-through distribution base paths and report remaining overlaps.")
+
+    def handle(self, *args, **options):
+        """Move eligible pull-through base paths and report unresolved overlaps."""
+        repaired = 0
+        old_base_paths = []
+        distributions = ContainerPullThroughDistribution.objects.exclude(
+            pulp_labels__contains=PULL_THROUGH_DISTRIBUTION_MARKER
+        ).select_related("pulp_domain")
+
+        for distribution in distributions.iterator():
+            if distribution.name != distribution.base_path:
+                continue
+
+            old_base_paths.append(distribution.base_path)
+            distribution.base_path = str(uuid4())
+            distribution.pulp_labels = {
+                **(distribution.pulp_labels or {}),
+                **PULL_THROUGH_DISTRIBUTION_MARKER,
+            }
+            with transaction.atomic():
+                distribution.save(update_fields=["base_path", "pulp_labels"])
+            repaired += 1
+
+        self.stdout.write(self.style.SUCCESS(f"Repaired {repaired} pull-through distributions."))
+
+        unresolved = ContainerPullThroughDistribution.objects.exclude(
+            pulp_labels__contains=PULL_THROUGH_DISTRIBUTION_MARKER
+        ).select_related("pulp_domain")
+        if unresolved:
+            self.stdout.write(
+                self.style.WARNING(
+                    "The following pull-through distributions require manual repair:"
+                )
+            )
+            for distribution in unresolved:
+                self.stdout.write(self._format_distribution(distribution))
+
+        overlaps = find_container_distribution_overlaps()
+        if overlaps:
+            self.stdout.write(
+                self.style.WARNING(
+                    "The following container distributions have overlapping base paths:"
+                )
+            )
+            for distribution, other in overlaps:
+                self.stdout.write(
+                    f"  {self._format_distribution(distribution).strip()} overlaps "
+                    f"{self._format_distribution(other).strip()}"
+                )
+
+        if settings.CACHE_ENABLED and old_base_paths:
+            SyncContentCache().delete(base_key=old_base_paths)
+            self.stdout.write(self.style.SUCCESS("Flushed repaired distribution cache entries."))
+
+    @staticmethod
+    def _format_distribution(distribution):
+        return (
+            f"  domain={distribution.pulp_domain.name!r} pk={distribution.pk} "
+            f"name={distribution.name!r} base_path={distribution.base_path!r}"
+        )

pulp_container/app/pull_through.py

@@ -0,0 +1,77 @@
+from django.db.models import F, Q, TextField, Value
+from django.db.models.functions import Concat
+
+from pulpcore.plugin.models import Distribution
+
+from pulp_container.app.models import ContainerDistribution, ContainerPullThroughDistribution
+from pulp_container.constants import (
+    PULL_THROUGH_DISTRIBUTION_LABEL,
+    PULL_THROUGH_DISTRIBUTION_LABEL_VALUE,
+)
+
+PULL_THROUGH_DISTRIBUTION_MARKER = {
+    PULL_THROUGH_DISTRIBUTION_LABEL: PULL_THROUGH_DISTRIBUTION_LABEL_VALUE
+}
+
+
+def is_pull_through_distribution_marked(distribution):
+    """Return whether a pull-through distribution uses its name as its registry path."""
+    return all(
+        (distribution.pulp_labels or {}).get(key) == value
+        for key, value in PULL_THROUGH_DISTRIBUTION_MARKER.items()
+    )
+
+
+def get_pull_through_distribution_path(distribution):
+    """Return the registry path represented by a pull-through distribution."""
+    if is_pull_through_distribution_marked(distribution):
+        return distribution.name
+    return distribution.base_path
+
+
+def _match_path(queryset, path, field):
+    """Return the longest segment-aware path match from a queryset."""
+    prefix_field = f"{field}_prefix"
+    return (
+        queryset.annotate(
+            requested_path=Value(path, output_field=TextField()),
+            **{prefix_field: Concat(F(field), Value("/"), output_field=TextField())},
+        )
+        .filter(Q(requested_path=F(field)) | Q(requested_path__startswith=F(prefix_field)))
+        .order_by(f"-{field}")
+        .first()
+    )
+
+
+def get_pull_through_distribution(path, domain):
+    """Find a pull-through distribution using the repaired scheme, then the legacy scheme."""
+    distributions = ContainerPullThroughDistribution.objects.filter(pulp_domain=domain)
+    marked = distributions.filter(pulp_labels__contains=PULL_THROUGH_DISTRIBUTION_MARKER)
+    if distribution := _match_path(marked, path, "name"):
+        return distribution
+
+    legacy = distributions.exclude(pulp_labels__contains=PULL_THROUGH_DISTRIBUTION_MARKER)
+    return _match_path(legacy, path, "base_path")
+
+
+def find_container_distribution_overlaps():
+    """Return pairs where a container distribution overlaps any distribution."""
+    container_ids = set(ContainerDistribution.objects.values_list("pk", flat=True))
+    distributions = Distribution.objects.select_related("pulp_domain").order_by(
+        "pulp_domain_id", "base_path", "pk"
+    )
+    by_domain_and_path = {
+        (distribution.pulp_domain_id, distribution.base_path): distribution
+        for distribution in distributions.iterator()
+    }
+    overlaps = []
+
+    for distribution in by_domain_and_path.values():
+        parts = distribution.base_path.split("/")
+        for index in range(1, len(parts)):
+            parent_path = "/".join(parts[:index])
+            if parent := by_domain_and_path.get((distribution.pulp_domain_id, parent_path)):
+                if parent.pk in container_ids or distribution.pk in container_ids:
+                    overlaps.append((parent, distribution))
+
+    return overlaps

pulp_container/app/registry_api.py

@@ -75,6 +75,10 @@
     RepositoryInvalid,
     RepositoryNotFound,
 )
+from pulp_container.app.pull_through import (
+    get_pull_through_distribution,
+    get_pull_through_distribution_path,
+)
 from pulp_container.app.redirects import (
     AzureStorageRedirects,
     FileStorageRedirects,
@@ -366,18 +370,12 @@ def get_drv_pull(self, path):
 
     def get_pull_through_drv(self, path):
         domain = get_domain()
-        pull_through_cache_distribution = (
-            models.ContainerPullThroughDistribution.objects.annotate(path=Value(path))
-            .filter(path__startswith=F("base_path"), pulp_domain=domain)
-            .order_by("-base_path")
-            .first()
-        )
+        pull_through_cache_distribution = get_pull_through_distribution(path, domain)
         if not pull_through_cache_distribution or not self.request.user.is_authenticated:
             raise RepositoryNotFound(name=path)
 
-        upstream_name = path.split(pull_through_cache_distribution.base_path, maxsplit=1)[1].strip(
-            "/"
-        )
+        pull_through_path = get_pull_through_distribution_path(pull_through_cache_distribution)
+        upstream_name = path.split(pull_through_path, maxsplit=1)[1].strip("/")
         try:
             pull_through_remote = models.ContainerPullThroughRemote.objects.get(
                 pk=pull_through_cache_distribution.remote_id