[PULP-1496] Add repository-specific package blocklist

CHANGES/1166.feature

@@ -0,0 +1 @@
+Added repository-specific package blocklist.

docs/user/guides/upload.md

@@ -135,6 +135,8 @@ as Pulp may contain different content units with the same name.
     }
     ```
 
+TODO: blocklist docs
+
 ## Remove content from a repository
 
 A content unit can be removed from a repository using the `remove` command.

pulp_python/app/migrations/0022_pythonblocklistentry.py

@@ -0,0 +1,48 @@
+# Generated by Django 5.2.10 on 2026-04-09 08:27
+
+import django.db.models.deletion
+import django_lifecycle.mixins
+import pulpcore.app.models.base
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("python", "0021_pythonrepository_upload_duplicate_filenames"),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name="PythonBlocklistEntry",
+            fields=[
+                (
+                    "pulp_id",
+                    models.UUIDField(
+                        default=pulpcore.app.models.base.pulp_uuid,
+                        editable=False,
+                        primary_key=True,
+                        serialize=False,
+                    ),
+                ),
+                ("pulp_created", models.DateTimeField(auto_now_add=True)),
+                ("pulp_last_updated", models.DateTimeField(auto_now=True, null=True)),
+                ("name", models.TextField(default=None, null=True)),
+                ("version", models.TextField(default=None, null=True)),
+                ("filename", models.TextField(default=None, null=True)),
+                ("added_by", models.TextField(blank=True, default="")),
+                (
+                    "repository",
+                    models.ForeignKey(
+                        on_delete=django.db.models.deletion.CASCADE,
+                        related_name="blocklist_entries",
+                        to="python.pythonrepository",
+                    ),
+                ),
+            ],
+            options={
+                "default_related_name": "%(app_label)s_%(model_name)s",
+            },
+            bases=(django_lifecycle.mixins.LifecycleModelMixin, models.Model),
+        ),
+    ]

pulp_python/app/models.py

@@ -14,6 +14,7 @@
 from rest_framework.serializers import ValidationError
 from pulpcore.plugin.models import (
     AutoAddObjPermsMixin,
+    BaseModel,
     Content,
     Publication,
     Distribution,
@@ -399,9 +400,12 @@ def finalize_new_version(self, new_version):
 
         When allow_package_substitution is False, reject any new version that would implicitly
         replace existing content with different checksums (content substitution).
+
+        Also checks newly added content against the repository's blocklist entries.
         """
         if not self.allow_package_substitution:
             self._check_for_package_substitution(new_version)
+        self._check_blocklist(new_version)
         remove_duplicates(new_version)
         validate_repo_version(new_version)
 
@@ -418,3 +422,58 @@ def _check_for_package_substitution(self, new_version):
                 "To allow this, set 'allow_package_substitution' to True on the repository. "
                 f"Conflicting packages: {duplicates}"
             )
+
+    def _check_blocklist(self, new_version):
+        """
+        Check newly added content in a repository version against the blocklist.
+        """
+        added_content = PythonPackageContent.objects.filter(
+            pk__in=new_version.added().values_list("pk", flat=True)
+        )
+        if added_content.exists():
+            self.check_blocklist_for_packages(added_content)
+
+    def check_blocklist_for_packages(self, packages):
+        """
+        Raise a ValidationError if any of the given packages match a blocklist entry.
+        """
+        entries = PythonBlocklistEntry.objects.filter(repository=self)
+        if not entries.exists():
+            return
+
+        blocked = []
+        for pkg in packages:
+            pkg_name_normalized = canonicalize_name(pkg.name) if pkg.name else ""
+            for entry in entries:
+                if entry.filename and entry.filename == pkg.filename:
+                    blocked.append(pkg.filename)
+                    break
+                if entry.name and canonicalize_name(entry.name) == pkg_name_normalized:
+                    if not entry.version or entry.version == pkg.version:
+                        blocked.append(pkg.filename)
+                        break
+        if blocked:
+            raise ValidationError(
+                "Blocklisted packages cannot be added to this repository: "
+                "{}".format(", ".join(blocked))
+            )
+
+
+class PythonBlocklistEntry(BaseModel):
+    """
+    An entry in a PythonRepository's package blocklist.
+
+    Blocks package uploads by exact filename, package name, or package name + version.
+    At least one of `name` or `filename` must be non-empty.
+    """
+
+    name = models.TextField(null=True, default=None)
+    version = models.TextField(null=True, default=None)
+    filename = models.TextField(null=True, default=None)
+    added_by = models.TextField(blank=True, default="")
+    repository = models.ForeignKey(
+        PythonRepository, on_delete=models.CASCADE, related_name="blocklist_entries"
+    )
+
+    class Meta:
+        default_related_name = "%(app_label)s_%(model_name)s"

pulp_python/app/serializers.py

@@ -6,14 +6,17 @@
 from django.db.utils import IntegrityError
 from drf_spectacular.utils import extend_schema_serializer
 from packaging.requirements import Requirement
+from packaging.version import Version, InvalidVersion
 from rest_framework import serializers
+from rest_framework_nested.relations import NestedHyperlinkedIdentityField
+from rest_framework_nested.serializers import NestedHyperlinkedModelSerializer
 from pypi_attestations import AttestationError
 from pydantic import TypeAdapter, ValidationError
 from urllib.parse import urljoin
 
 from pulpcore.plugin import models as core_models
 from pulpcore.plugin import serializers as core_serializers
-from pulpcore.plugin.util import get_domain, get_prn, get_current_authenticated_user
+from pulpcore.plugin.util import get_domain, get_prn, get_current_authenticated_user, reverse
 
 from pulp_python.app import models as python_models
 from pulp_python.app.provenance import (
@@ -53,6 +56,11 @@ class PythonRepositorySerializer(core_serializers.RepositorySerializer):
         default=False,
         required=False,
     )
+    blocklist_entries_href = serializers.SerializerMethodField(
+        help_text=_("URL to the blocklist entries for this repository."),
+        read_only=True,
+    )
+
     allow_package_substitution = serializers.BooleanField(
         help_text=_(
             "Whether to allow package substitution (replacing existing packages with packages "
@@ -65,10 +73,15 @@ class PythonRepositorySerializer(core_serializers.RepositorySerializer):
         required=False,
     )
 
+    def get_blocklist_entries_href(self, obj):
+        repo_href = reverse("repositories-python/python-detail", kwargs={"pk": obj.pk})
+        return f"{repo_href}blocklist_entries/"
+
     class Meta:
         fields = core_serializers.RepositorySerializer.Meta.fields + (
             "autopublish",
             "allow_package_substitution",
+            "blocklist_entries_href",
         )
         model = python_models.PythonRepository
 
@@ -780,6 +793,117 @@ class Meta:
         model = python_models.PythonRemote
 
 
+class _NestedIdentityField(NestedHyperlinkedIdentityField):
+    """
+    NestedHyperlinkedIdentityField that uses pulpcore's reverse for relative URLs.
+    Mimics NestedIdentityField from pulpcore, which is not exposed via the plugin API.
+    """
+
+    def get_url(self, obj, view_name, request, *args, **kwargs):
+        self.reverse = reverse
+        return super().get_url(obj, view_name, request, *args, **kwargs)
+
+
+class PythonBlocklistEntrySerializer(
+    core_serializers.ModelSerializer, NestedHyperlinkedModelSerializer
+):
+    """
+    Serializer for PythonBlocklistEntry.
+
+    The `repository` is supplied by the URL (not the request body) and is injected
+    by the viewset before saving.
+    """
+
+    pulp_href = _NestedIdentityField(
+        view_name="blocklist_entries-detail",
+        parent_lookup_kwargs={"repository_pk": "repository__pk"},
+    )
+    repository = core_serializers.DetailRelatedField(
+        read_only=True,
+        view_name_pattern=r"repositories(-.*/.*)?-detail",
+        help_text=_("Repository this blocklist entry belongs to."),
+    )
+    name = serializers.CharField(
+        required=False,
+        allow_null=True,
+        default=None,
+        help_text=_(
+            "Package name to block (for all versions). Compared after PEP 503 normalization. "
+            "Required when 'filename' is not provided."
+        ),
+    )
+    version = serializers.CharField(
+        required=False,
+        allow_null=True,
+        default=None,
+        help_text=_("Exact version string to block (e.g. '1.0'). Only used when 'name' is set."),
+    )
+    filename = serializers.CharField(
+        required=False,
+        allow_null=True,
+        default=None,
+        help_text=_("Exact filename to block. Required when 'name' is not provided."),
+    )
+    added_by = serializers.CharField(read_only=True)
+
+    def validate(self, data):
+        """
+        Validate that the blocklist entry is well-formed and not a duplicate.
+        """
+        name = data.get("name")
+        filename = data.get("filename")
+        version = data.get("version")
+
+        if version and filename:
+            raise serializers.ValidationError(_("'version' cannot be used with 'filename'."))
+        if version and not name:
+            raise serializers.ValidationError(_("'version' requires 'name' to be provided."))
+        if name and filename:
+            raise serializers.ValidationError(_("'name' and 'filename' are mutually exclusive."))
+        if not name and not filename:
+            raise serializers.ValidationError(_("Either 'name' or 'filename' must be provided."))
+
+        if version:
+            try:
+                Version(version)
+            except InvalidVersion:
+                raise serializers.ValidationError(
+                    {"version": _("'{}' is not a valid version.").format(version)}
+                )
+
+        repository = self.context.get("repository")
+        if repository:
+            qs = python_models.PythonBlocklistEntry.objects.filter(repository=repository)
+            if name and qs.filter(name=name, version=version).exists():
+                raise serializers.ValidationError(
+                    _("A blocklist entry with this name and version already exists.")
+                )
+            if filename and qs.filter(filename=filename).exists():
+                raise serializers.ValidationError(
+                    _("A blocklist entry with this filename already exists.")
+                )
+
+        return data
+
+    def create(self, validated_data):
+        """
+        Create a new blocklist entry, recording the authenticated user in `added_by`.
+        """
+        user = get_current_authenticated_user()
+        validated_data["added_by"] = get_prn(user) if user else ""
+        return super().create(validated_data)
+
+    class Meta:
+        fields = core_serializers.ModelSerializer.Meta.fields + (
+            "repository",
+            "name",
+            "version",
+            "filename",
+            "added_by",
+        )
+        model = python_models.PythonBlocklistEntry
+
+
 class PythonBanderRemoteSerializer(serializers.Serializer):
     """
     A Serializer for the initial step of creating a Python Remote from a Bandersnatch config file