Restrict media privacy until a referencing page is published

cms/documents/migrations/0003_customdocument__privacy_and_more.py

@@ -0,0 +1,28 @@
+# Generated by Django 5.1.3 on 2024-11-29 16:19
+
+import django.utils.timezone
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("documents", "0002_alter_customdocument_file_size"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="customdocument",
+            name="_privacy",
+            field=models.CharField(default="public", max_length=10),
+        ),
+        migrations.AddField(
+            model_name="customdocument",
+            name="file_permissions_last_set",
+            field=models.DateTimeField(editable=False, null=True),
+        ),
+        migrations.AddField(
+            model_name="customdocument",
+            name="privacy_last_changed",
+            field=models.DateTimeField(default=django.utils.timezone.now, editable=False),
+        ),
+    ]

cms/documents/migrations/0004_alter_customdocument__privacy.py

@@ -0,0 +1,17 @@
+# Generated by Django 5.1.3 on 2024-11-29 16:20
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("documents", "0003_customdocument__privacy_and_more"),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name="customdocument",
+            name="_privacy",
+            field=models.CharField(default="private", max_length=10),
+        ),
+    ]

cms/documents/models.py

@@ -1,12 +1,16 @@
-from wagtail.documents.models import AbstractDocument
-from wagtail.documents.models import Document as WagtailDocument
+from collections.abc import Sequence
+from typing import ClassVar
 
+from wagtail.documents.models import AbstractDocument, Document
 
-class CustomDocument(AbstractDocument):
+from cms.private_media.models import PrivateDocumentMixin
+
+
+class CustomDocument(PrivateDocumentMixin, AbstractDocument):
     """Custom Wagtail document class.
 
     Using a custom class from the beginning allows us to add
     any customisations we may need.
     """
 
-    admin_form_fields = WagtailDocument.admin_form_fields
+    admin_form_fields: ClassVar[Sequence[str]] = Document.admin_form_fields

cms/images/migrations/0003_customimage__privacy_and_more.py

@@ -0,0 +1,28 @@
+# Generated by Django 5.1.3 on 2024-11-29 16:19
+
+import django.utils.timezone
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("images", "0002_customimage_description"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="customimage",
+            name="_privacy",
+            field=models.CharField(default="public", max_length=10),
+        ),
+        migrations.AddField(
+            model_name="customimage",
+            name="file_permissions_last_set",
+            field=models.DateTimeField(editable=False, null=True),
+        ),
+        migrations.AddField(
+            model_name="customimage",
+            name="privacy_last_changed",
+            field=models.DateTimeField(default=django.utils.timezone.now, editable=False),
+        ),
+    ]

cms/images/migrations/0004_alter_customimage__privacy.py

@@ -0,0 +1,17 @@
+# Generated by Django 5.1.3 on 2024-11-29 16:20
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("images", "0003_customimage__privacy_and_more"),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name="customimage",
+            name="_privacy",
+            field=models.CharField(default="private", max_length=10),
+        ),
+    ]

cms/images/models.py

@@ -1,21 +1,28 @@
+from collections.abc import Sequence
+from typing import ClassVar
+
 from django.db import models
-from wagtail.images.models import AbstractImage, AbstractRendition, Image
+from wagtail.images.models import AbstractImage, Image
+
+from cms.private_media.models import AbstractPrivateRendition, PrivateImageMixin
 
 
-class CustomImage(AbstractImage):
+class CustomImage(PrivateImageMixin, AbstractImage):  # type: ignore[django-manager-missing]
     """Custom Wagtail image class.
 
     Using a custom class from the beginning allows us to add
     any customisations we may need.
     """
 
-    admin_form_fields = Image.admin_form_fields
+    admin_form_fields: ClassVar[Sequence[str]] = Image.admin_form_fields
 
 
-class Rendition(AbstractRendition):
+class Rendition(AbstractPrivateRendition):
     """Our custom rendition class."""
 
-    image = models.ForeignKey["CustomImage"]("CustomImage", related_name="renditions", on_delete=models.CASCADE)
+    image = models.ForeignKey(  # type: ignore[var-annotated]
+        "CustomImage", related_name="renditions", on_delete=models.CASCADE
+    )
 
     class Meta:
         unique_together = (("image", "filter_spec", "focal_point_key"),)

cms/private_media/apps.py

@@ -0,0 +1,11 @@
+from django.apps import AppConfig
+
+
+class PrivateMediaConfig(AppConfig):
+    default_auto_field = "django.db.models.AutoField"
+    name = "cms.private_media"
+
+    def ready(self) -> None:
+        from .signal_handlers import register_signal_handlers  # pylint: disable=import-outside-toplevel
+
+        register_signal_handlers()

cms/private_media/bulk_operations.py

@@ -0,0 +1,57 @@
+import concurrent.futures
+import logging
+from collections.abc import Callable, Iterable
+from typing import TYPE_CHECKING
+
+from django.conf import settings
+
+from cms.private_media.constants import Privacy
+
+if TYPE_CHECKING:
+    from django.db.models.fields.files import FieldFile
+
+
+logger = logging.getLogger(__name__)
+
+
+def bulk_set_file_permissions(files: Iterable["FieldFile"], intended_privacy: Privacy) -> dict["FieldFile", bool]:
+    """Set file permissions for an iterable of FieldFile objects, using the
+    make_private() or make_public() methods of the storage backend.
+
+    Uses a thread pool to set the file permissions in parallel where possible.
+
+    Args:
+        files: An iterable of FieldFile objects to set permissions for
+        intended_privacy: The intended privacy status for the supplied files
+
+    Returns:
+        dict[FieldFile, bool]: A mapping of files to a boolean indicating whether
+        the permission setting request was successful
+    """
+    results: dict[FieldFile, bool] = {}
+
+    def set_file_permission_and_report(file: "FieldFile") -> None:
+        storage = file.storage
+        handler: Callable[[FieldFile], bool] | None
+        if intended_privacy is Privacy.PRIVATE:
+            handler = getattr(storage, "make_private", None)
+        elif intended_privacy is Privacy.PUBLIC:
+            handler = getattr(storage, "make_public", None)
+
+        if handler is None:
+            logger.info(
+                "%s does not support setting of individual file permissions to %s, so skipping for: %s.",
+                storage.__class__.__name__,
+                intended_privacy,
+                file.name,
+            )
+            results[file] = True
+        else:
+            results[file] = handler(file)
+
+    with concurrent.futures.ThreadPoolExecutor(
+        max_workers=int(settings.PRIVATE_MEDIA_PERMISSION_SETTING_MAX_WORKERS)
+    ) as executor:
+        executor.map(set_file_permission_and_report, files)
+
+    return results

cms/private_media/constants.py

@@ -0,0 +1,7 @@
+from django.db.models import TextChoices
+from django.utils.translation import gettext_lazy as _
+
+
+class Privacy(TextChoices):
+    PRIVATE = "private", _("Private")
+    PUBLIC = "public", _("Public")

cms/private_media/management/commands/retry_file_permission_set_attempts.py

@@ -0,0 +1,62 @@
+from typing import TYPE_CHECKING, Any
+
+from django.core.management.base import BaseCommand
+from django.db.models import F
+
+from cms.private_media.constants import Privacy
+from cms.private_media.models.images import PrivateImageMixin
+from cms.private_media.utils import get_private_media_models
+
+if TYPE_CHECKING:
+    from cms.private_media.models import PrivateMediaMixin
+
+
+class Command(BaseCommand):
+    def add_arguments(self, parser: Any) -> None:
+        parser.add_argument(
+            "--dry-run",
+            action="store_true",
+            dest="dry_run",
+            default=False,
+            help="Dry run -- don't change anything.",
+        )
+
+    def handle(self, *args: Any, **options: Any) -> None:
+        self.dry_run = options["dry_run"]  # pylint: disable=attribute-defined-outside-init
+        if self.dry_run:
+            self.stdout.write("This is a dry run.")
+
+        for model in get_private_media_models():
+            queryset = model.objects.filter(file_permissions_last_set__lt=F("privacy_last_changed"))
+            if not self.dry_run and issubclass(model, PrivateImageMixin):
+                queryset = queryset.prefetch_related("renditions")
+            permissions_outdated = list(queryset)
+            self.stdout.write(f"{len(permissions_outdated)} {model.__name__} instances have outdated file permissions.")
+            if permissions_outdated:
+                make_private = []
+                make_public = []
+                for obj in permissions_outdated:
+                    if obj.privacy is Privacy.PRIVATE:
+                        make_private.append(obj)
+                    elif obj.privacy is Privacy.PUBLIC:
+                        make_public.append(obj)
+
+                self.update_file_permissions(model, make_private, Privacy.PRIVATE)
+                self.update_file_permissions(model, make_public, Privacy.PUBLIC)
+
+    def update_file_permissions(
+        self, model_class: type["PrivateMediaMixin"], items: list["PrivateMediaMixin"], privacy: Privacy
+    ) -> None:
+        """Update the file permissions for the provided items to reflect the provided privacy status."""
+        plural = model_class._meta.verbose_name_plural
+        if self.dry_run:
+            self.stdout.write(f"Would update file permissions for {len(items)} {privacy} {plural}.")
+        else:
+            updated_count = 0
+            for item in model_class.objects.bulk_set_file_permissions(  # type: ignore[attr-defined]
+                items, privacy, save_changes=True
+            ):
+                if not item.file_permissions_are_outdated():
+                    updated_count += 1
+
+            self.stdout.write(f"File permissions successfully updated for {updated_count} {privacy} {plural}.")