[PULP-1496] Add repository-specific package blocklist

CHANGES/1166.feature

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

docs/user/guides/_SUMMARY.md

@@ -4,3 +4,4 @@
 * [Host Python Content](host.md)
 * [Vulnerability Report](vulnerability_report.md)
 * [Attestation Hosting](attestation.md)
+* [Package Blocklist](blocklist.md)

docs/user/guides/blocklist.md

@@ -0,0 +1,73 @@
+# Package Blocklist
+
+A repository can have a blocklist that prevents specific packages from being added.
+Blocklist entries can match by package `name` (all versions), package `name` with an exact `version`, or exact `filename`.
+Exactly one of `name` or `filename` must be provided.
+
+Each entry records the PRN of the user who created it in the `added_by` field.
+
+## Setup
+
+If you do not already have a repository, create one:
+
+```bash
+pulp python repository create --name foo
+```
+
+Set the API base URL and repository HREF for use in the subsequent commands:
+
+```bash
+PULP_API="http://localhost:5001"
+REPO_HREF=$(pulp python repository show --name foo | jq -r ".pulp_href")
+```
+
+## Add a blocklist entry
+
+=== "By name (all versions)"
+
+    ```bash
+    # Block all versions of shelf-reader
+    http POST "${PULP_API}${REPO_HREF}blocklist_entries/" name="shelf-reader"
+    ```
+
+=== "By name and version"
+
+    ```bash
+    # Block only shelf-reader 0.1
+    http POST "${PULP_API}${REPO_HREF}blocklist_entries/" name="shelf-reader" version="0.1"
+    ```
+
+=== "By filename"
+
+    ```bash
+    # Block only shelf-reader-0.1.tar.gz
+    http POST "${PULP_API}${REPO_HREF}blocklist_entries/" filename="shelf-reader-0.1.tar.gz"
+    ```
+
+Set the UUID of a created entry for use in the subsequent commands:
+
+```bash
+ENTRY_UUID=$(http GET "${PULP_API}${REPO_HREF}blocklist_entries/" | jq -r '.results[0].prn | split(":") | .[-1]')
+```
+
+## List blocklist entries
+
+List all entries for a repository:
+
+```bash
+http GET "${PULP_API}${REPO_HREF}blocklist_entries/"
+```
+
+Show a single entry:
+
+```bash
+http GET "${PULP_API}${REPO_HREF}blocklist_entries/${ENTRY_UUID}/"
+```
+
+## Remove a blocklist entry
+
+```bash
+http DELETE "${PULP_API}${REPO_HREF}blocklist_entries/${ENTRY_UUID}/"
+```
+
+Once an entry is removed, packages matching it can be added to the repository again.

pulp_python/app/migrations/0022_pythonblocklistentry.py

@@ -0,0 +1,48 @@
+# Generated by Django 5.2.10 on 2026-04-16 14:00
+
+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(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,63 @@ 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)
+        ).only("filename", "name_normalized", "version")
+        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:
+            for entry in entries:
+                if entry.filename and entry.filename == pkg.filename:
+                    blocked.append(pkg.filename)
+                    break
+                if 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.
+
+    Blocklist entries prevent packages from being added to the repository.
+    Entries can match by package `name` (all versions), package `name` + `version`,
+    or exact `filename`. Exactly one of `name` or `filename` must be provided.
+    """
+
+    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(default="")
+    repository = models.ForeignKey(
+        PythonRepository, on_delete=models.CASCADE, related_name="blocklist_entries"
+    )
+
+    def __str__(self):
+        if self.filename:
+            return f"<{self._meta.object_name}: {self.filename}>"
+        return f"<{self._meta.object_name}: {self.name} [{self.version or 'all'}]>"
+
+    class Meta:
+        default_related_name = "%(app_label)s_%(model_name)s"

pulp_python/app/serializers.py

@@ -6,16 +6,18 @@
 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 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.utils import canonicalize_name
 from pulp_python.app.provenance import (
     Attestation,
     Provenance,
@@ -53,6 +55,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 +72,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 +792,115 @@ class Meta:
         model = python_models.PythonRemote
 
 
+class PythonBlocklistEntrySerializer(core_serializers.ModelSerializer):
+    """
+    Serializer for PythonBlocklistEntry.
+
+    The `repository` is supplied by the URL (not the request body) and is injected
+    by the viewset before saving.
+    """
+
+    pulp_href = serializers.SerializerMethodField(
+        read_only=True,
+        help_text=_("The URL of this blocklist entry."),
+    )
+    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,
+        help_text=_("PRN of the user who added this blocklist entry."),
+    )
+
+    def get_pulp_href(self, obj):
+        repo_href = reverse("repositories-python/python-detail", kwargs={"pk": obj.repository_id})
+        return f"{repo_href}blocklist_entries/{obj.pk}/"
+
+    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 bool(name) == bool(filename):
+            raise serializers.ValidationError(
+                _("Exactly one of 'name' or 'filename' must be provided.")
+            )
+
+        if version:
+            try:
+                Version(version)
+            except InvalidVersion:
+                raise serializers.ValidationError(
+                    {"version": _("'{}' is not a valid version.").format(version)}
+                )
+        if name:
+            data["name"] = canonicalize_name(name)
+            name = data["name"]
+
+        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.")
+                )
+            data["repository"] = repository
+
+        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