feat: add default content guard auto-assignment for distributions

Add a default_content_guard field to the Domain model that is automatically
assigned to new distributions created within the domain when they do not
specify their own content guard. Use a composite content guard as the
default to apply multiple guards.

Closes: #7988

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: dkliban

CHANGES/7988.feature

@@ -0,0 +1 @@
+Added a ``default_content_guard`` field to domains that is automatically assigned to new distributions created within the domain when they do not specify their own content-guard.

pulp_file/tests/functional/api/test_domains.py

@@ -370,3 +370,54 @@ def test_no_cross_pollination(
     assert error["add_content_units"][0].startswith(
         f"Content units are not a part of the current domain {domain.name}: ["
     )
+
+
+@pytest.mark.parallel
+def test_distribution_default_content_guard_auto_assignment(
+    pulpcore_bindings,
+    file_bindings,
+    gen_object_with_cleanup,
+    monitor_task,
+):
+    """A distribution created in a domain inherits the domain's default_content_guard."""
+    domain = gen_object_with_cleanup(
+        pulpcore_bindings.DomainsApi,
+        {
+            "name": str(uuid.uuid4()),
+            "storage_class": "pulpcore.app.models.storage.FileSystem",
+            "storage_settings": {"MEDIA_ROOT": "/var/lib/pulp/media/"},
+        },
+    )
+    domain_name = domain.name
+
+    # Create a content guard in the domain and set it as the domain default
+    guard = gen_object_with_cleanup(
+        pulpcore_bindings.ContentguardsRbacApi, {"name": str(uuid.uuid4())}, pulp_domain=domain_name
+    )
+    response = pulpcore_bindings.DomainsApi.partial_update(
+        domain.pulp_href, {"default_content_guard": guard.pulp_href}
+    )
+    monitor_task(response.task)
+
+    # A distribution created WITHOUT a content guard inherits the domain default
+    distro = gen_object_with_cleanup(
+        file_bindings.DistributionsFileApi,
+        {"name": str(uuid.uuid4()), "base_path": str(uuid.uuid4())},
+        pulp_domain=domain_name,
+    )
+    assert distro.content_guard == guard.pulp_href
+
+    # A distribution created WITH an explicit content guard keeps its own
+    other_guard = gen_object_with_cleanup(
+        pulpcore_bindings.ContentguardsRbacApi, {"name": str(uuid.uuid4())}, pulp_domain=domain_name
+    )
+    distro_explicit = gen_object_with_cleanup(
+        file_bindings.DistributionsFileApi,
+        {
+            "name": str(uuid.uuid4()),
+            "base_path": str(uuid.uuid4()),
+            "content_guard": other_guard.pulp_href,
+        },
+        pulp_domain=domain_name,
+    )
+    assert distro_explicit.content_guard == other_guard.pulp_href

pulpcore/app/migrations/0157_domain_default_content_guard.py

@@ -0,0 +1,21 @@
+import django.db.models.deletion
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("core", "0156_alter_contentartifact_relative_path_and_more"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="domain",
+            name="default_content_guard",
+            field=models.ForeignKey(
+                null=True,
+                on_delete=django.db.models.deletion.SET_NULL,
+                related_name="+",
+                to="core.contentguard",
+            ),
+        ),
+    ]

pulpcore/app/models/domain.py

@@ -32,6 +32,10 @@ class Domain(BaseModel, AutoAddObjPermsMixin):
         storage_settings (EncryptedJSONField): Settings needed to configure storage backend
         redirect_to_object_storage (models.BooleanField): Redirect to object storage in content app
         hide_guarded_distributions (models.BooleanField): Hide guarded distributions in content app
+
+    Relations:
+        default_content_guard (models.ForeignKey): An optional content-guard automatically
+            assigned to new distributions created within this domain.
     """
 
     name = models.SlugField(null=False, unique=True)
@@ -43,6 +47,9 @@ class Domain(BaseModel, AutoAddObjPermsMixin):
     # Pulp settings that are appropriate to be set on a "per domain" level
     redirect_to_object_storage = models.BooleanField(default=True)
     hide_guarded_distributions = models.BooleanField(default=False)
+    default_content_guard = models.ForeignKey(
+        "ContentGuard", null=True, on_delete=models.SET_NULL, related_name="+"
+    )
 
     def get_storage(self):
         """Returns this domain's instantiated storage class."""

pulpcore/app/models/publication.py

@@ -16,7 +16,7 @@
 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
+from django_lifecycle import AFTER_CREATE, AFTER_UPDATE, BEFORE_CREATE, BEFORE_DELETE, hook
 from rest_framework.exceptions import APIException
 from url_normalize import url_normalize
 
@@ -792,6 +792,19 @@ def get_fallback_ca(self, path):
                     return pa.content_artifact
         return None
 
+    @hook(BEFORE_CREATE)
+    def _set_default_content_guard(self):
+        """Apply the domain's default content guard when none is explicitly set.
+
+        If the distribution is created without a ``content_guard`` and its domain has a
+        ``default_content_guard`` configured, that guard is assigned automatically. An
+        explicitly provided ``content_guard`` always takes precedence.
+        """
+        if self.content_guard_id is None:
+            default_content_guard_id = self.pulp_domain.default_content_guard_id
+            if default_content_guard_id is not None:
+                self.content_guard_id = default_content_guard_id
+
     @hook(AFTER_CREATE)
     @hook(
         AFTER_UPDATE,

pulpcore/app/serializers/domain.py

@@ -9,13 +9,15 @@
 from rest_framework import serializers
 from rest_framework.validators import UniqueValidator
 
-from pulpcore.app.models import Domain
+from pulpcore.app.models import ContentGuard, Domain
 from pulpcore.app.serializers import (
+    DetailRelatedField,
     HiddenFieldsMixin,
     IdentityField,
     ModelSerializer,
     pulp_labels_validator,
 )
+from pulpcore.app.util import get_prn
 
 BACKEND_CHOICES = (
     ("pulpcore.app.models.storage.FileSystem", "Use local filesystem as storage"),
@@ -454,6 +456,23 @@ class DomainSerializer(BackendSettingsValidator, ModelSerializer):
         help_text=_("Boolean to hide distributions with a content guard in the content app."),
         default=False,
     )
+    default_content_guard = DetailRelatedField(
+        required=False,
+        allow_null=True,
+        help_text=_(
+            "An optional content-guard that is automatically assigned to new distributions "
+            "created within this domain when they do not specify their own content-guard. To "
+            "apply multiple guards by default, use a composite content-guard."
+        ),
+        view_name_pattern=r"contentguards(-.*/.*)?-detail",
+        queryset=ContentGuard.objects.all(),
+    )
+    default_content_guard_prn = serializers.SerializerMethodField(
+        help_text=_("The Pulp Resource Name (PRN) of the domain's default content-guard."),
+    )
+
+    def get_default_content_guard_prn(self, obj):
+        return get_prn(obj.default_content_guard) if obj.default_content_guard else None
 
     def validate_name(self, value):
         """Ensure name is not 'api' or 'content'."""
@@ -463,6 +482,26 @@ def validate_name(self, value):
 
     def validate(self, data):
         """Ensure that Domain settings are valid."""
+        # A default content-guard must live in the same domain it is being assigned to.
+        # Checked before the "default" domain short-circuit so it is never silently skipped.
+        default_content_guard = data.get("default_content_guard")
+        if default_content_guard is not None:
+            if self.instance is None:
+                raise serializers.ValidationError(
+                    detail={
+                        "default_content_guard": _(
+                            "A default content-guard can only be set on an existing domain. "
+                            "Create the domain first, then set this field with an update."
+                        )
+                    }
+                )
+            if default_content_guard.pulp_domain_id != self.instance.pulp_id:
+                raise serializers.ValidationError(
+                    detail={
+                        "default_content_guard": _("The content-guard must belong to this domain.")
+                    }
+                )
+
         # Validate for update gets called before ViewSet default check
         if self.instance and self.instance.name == "default":
             return data
@@ -495,6 +534,8 @@ class Meta:
             "storage_settings",
             "redirect_to_object_storage",
             "hide_guarded_distributions",
+            "default_content_guard",
+            "default_content_guard_prn",
         )
 
 

pulpcore/app/viewsets/domain.py

@@ -108,6 +108,13 @@ class DomainViewSet(
         "core.domain_viewer": ["core.view_domain"],
     }
 
+    def get_queryset(self):
+        """Prefetch the default content guard to avoid N+1 queries on the list endpoint."""
+        qs = super().get_queryset()
+        if getattr(self, "action", "") == "list":
+            qs = qs.select_related("default_content_guard")
+        return qs
+
     @extend_schema(
         description="Trigger an asynchronous update task",
         responses={200: DomainSerializer, 202: AsyncOperationResponseSerializer},

pulpcore/tests/functional/api/test_crud_domains.py

@@ -323,6 +323,61 @@ def test_special_domain_creation(pulpcore_bindings, gen_object_with_cleanup, pul
     assert random_name not in domain.pulp_href
 
 
+@pytest.mark.parallel
+def test_domain_default_content_guard(pulpcore_bindings, monitor_task, pulp_settings):
+    """Set, read, clear, and reject cross-domain values for a domain's default_content_guard."""
+    if not pulp_settings.DOMAIN_ENABLED:
+        pytest.skip("Domains not enabled")
+    name = str(uuid.uuid4())
+    body = {
+        "name": name,
+        "storage_class": "pulpcore.app.models.storage.FileSystem",
+        "storage_settings": {"MEDIA_ROOT": ""},
+    }
+    domain = pulpcore_bindings.DomainsApi.create(body)
+    try:
+        # A new domain has no default content guard
+        assert domain.default_content_guard is None
+        assert domain.default_content_guard_prn is None
+
+        # Create a content guard within the domain and set it as the default
+        guard = pulpcore_bindings.ContentguardsRbacApi.create({"name": name}, pulp_domain=name)
+        response = pulpcore_bindings.DomainsApi.partial_update(
+            domain.pulp_href, {"default_content_guard": guard.pulp_href}
+        )
+        monitor_task(response.task)
+
+        domain = pulpcore_bindings.DomainsApi.read(domain.pulp_href)
+        assert domain.default_content_guard == guard.pulp_href
+        assert domain.default_content_guard_prn is not None
+
+        # A content guard from a different domain (default) is rejected
+        other_guard = pulpcore_bindings.ContentguardsRbacApi.create({"name": str(uuid.uuid4())})
+        try:
+            with pytest.raises(ApiException) as e:
+                pulpcore_bindings.DomainsApi.partial_update(
+                    domain.pulp_href, {"default_content_guard": other_guard.pulp_href}
+                )
+            assert e.value.status == 400
+            assert "default_content_guard" in e.value.body
+        finally:
+            pulpcore_bindings.ContentguardsRbacApi.delete(other_guard.pulp_href)
+
+        # Clear the default content guard
+        response = pulpcore_bindings.DomainsApi.partial_update(
+            domain.pulp_href, {"default_content_guard": None}
+        )
+        monitor_task(response.task)
+        domain = pulpcore_bindings.DomainsApi.read(domain.pulp_href)
+        assert domain.default_content_guard is None
+        assert domain.default_content_guard_prn is None
+
+        pulpcore_bindings.ContentguardsRbacApi.delete(guard.pulp_href)
+    finally:
+        response = pulpcore_bindings.DomainsApi.delete(domain.pulp_href)
+        monitor_task(response.task)
+
+
 @pytest.mark.parallel
 def test_filter_domains_by_label(pulpcore_bindings, domain_factory):
     """Test filtering domains by label."""

pulpcore/tests/unit/serializers/test_domain.py

@@ -159,6 +159,47 @@ def test_cloudfront_s3_storage_settings(storage_class, required_settings):
     assert serializer.is_valid(raise_exception=True)
 
 
+DOMAIN_ID = "00000000-0000-0000-0000-000000000001"
+OTHER_DOMAIN_ID = "00000000-0000-0000-0000-000000000002"
+
+
+def test_default_content_guard_cross_domain_rejected():
+    """A content-guard from another domain cannot be a domain's default_content_guard."""
+    domain = SimpleNamespace(pulp_id=DOMAIN_ID, name="doma")
+    other_domain_guard = SimpleNamespace(pulp_domain_id=OTHER_DOMAIN_ID)
+    serializer = DomainSerializer(instance=domain)
+
+    with pytest.raises(serializers.ValidationError) as exc_info:
+        serializer.validate({"default_content_guard": other_domain_guard})
+    assert "default_content_guard" in str(exc_info.value)
+
+
+def test_default_content_guard_rejected_on_create():
+    """default_content_guard cannot be set while creating a domain (no instance yet)."""
+    guard = SimpleNamespace(pulp_domain_id=OTHER_DOMAIN_ID)
+    serializer = DomainSerializer(data={})
+
+    with pytest.raises(serializers.ValidationError) as exc_info:
+        serializer.validate({"default_content_guard": guard})
+    assert "existing domain" in str(exc_info.value)
+
+
+def test_default_content_guard_same_domain_accepted():
+    """A content-guard from the same domain passes validation."""
+    domain = SimpleNamespace(
+        pulp_id=DOMAIN_ID,
+        name="doma",
+        storage_class="pulpcore.app.models.storage.FileSystem",
+        storage_settings={"location": "/var/lib/pulp/media/"},
+        redirect_to_object_storage=True,
+    )
+    guard = SimpleNamespace(pulp_domain_id=DOMAIN_ID)
+    serializer = DomainSerializer(instance=domain)
+
+    # Should not raise (storage backend check is monkeypatched out by the autouse fixture).
+    serializer.validate({"default_content_guard": guard})
+
+
 class DomainSettingsBaseMixin:
     storage_class = None
     serializer_class = None