diff --git a/CHANGELOG.md b/CHANGELOG.md index d0274498..4c823dc2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,10 @@ and this project adheres to ## [Unreleased] +### Added + +- ✨(backend) add file specific admin #1387 + ### Changed - 🐛(agents) fix bug when closing metadata-collector diff --git a/src/backend/core/admin.py b/src/backend/core/admin.py index c8374782..a8ee7568 100644 --- a/src/backend/core/admin.py +++ b/src/backend/core/admin.py @@ -3,17 +3,55 @@ from django import forms from django.contrib import admin, messages from django.contrib.auth import admin as auth_admin +from django.db import transaction +from django.utils.html import format_html from django.utils.translation import gettext_lazy as _ from core.recording.event import notification from . import models +from .tasks.file import process_file_deletion +from .utils import generate_download_file_url + + +def hard_delete_file(file): + """Hard delete a file, soft deleting it first when needed.""" + if file.deleted_at is None: + file.soft_delete() + file.hard_delete() + transaction.on_commit(lambda: process_file_deletion.delay(file.id)) + + +class FileInlineFormSet(forms.BaseInlineFormSet): + """Inline formset overriding delete behavior for files.""" + + def delete_existing(self, obj, commit=True): + """Hard delete files instead of calling model.delete().""" + hard_delete_file(obj) + + +class FileInline(admin.TabularInline): + """Inline class for the File model.""" + + model = models.File + formset = FileInlineFormSet + fk_name = "creator" + extra = 0 + fields = ("id", "title", "type", "upload_state", "created_at") + readonly_fields = ("id", "created_at", "upload_state", "type") + show_change_link = True + + def get_queryset(self, request): + """Hide hard deleted files in the inline.""" + return super().get_queryset(request).filter(hard_deleted_at__isnull=True) @admin.register(models.User) class UserAdmin(auth_admin.UserAdmin): """Admin class for the User model""" + inlines = (FileInline,) + fieldsets = ( ( None, @@ -97,6 +135,136 @@ class UserAdmin(auth_admin.UserAdmin): search_fields = ("id", "sub", "admin_email", "email", "full_name") +@admin.register(models.File) +class FileAdmin(admin.ModelAdmin): + """Admin class for the File model.""" + + list_display = ( + "id", + "title", + "type", + "creator", + "upload_state", + "deleted_at", + "hard_deleted_at", + "created_at", + "updated_at", + ) + list_filter = ( + "type", + "upload_state", + "created_at", + "updated_at", + "deleted_at", + "hard_deleted_at", + ) + search_fields = ( + "id", + "title", + "filename", + "mimetype", + "description", + "creator__email", + "creator__admin_email", + "creator__full_name", + ) + ordering = ("-created_at",) + readonly_fields = ( + "id", + "created_at", + "updated_at", + "deleted_at", + "hard_deleted_at", + "description", + "malware_detection_info", + "is_pending_upload", + "preview_url", + "extension", + "key_base", + "file_key", + "upload_state", + "type", + "mimetype", + "size", + ) + autocomplete_fields = ("creator",) + fieldsets = ( + ( + None, + { + "fields": ( + "id", + "title", + "type", + "creator", + "filename", + "upload_state", + ) + }, + ), + ( + _("Content"), + { + "fields": ( + "mimetype", + "size", + "description", + "malware_detection_info", + ) + }, + ), + ( + _("Deletion"), + { + "fields": ( + "deleted_at", + "hard_deleted_at", + ) + }, + ), + ( + _("Derived info"), + { + "fields": ( + "is_pending_upload", + "extension", + "key_base", + "file_key", + "preview_url", + ) + }, + ), + (_("Timestamps"), {"fields": ("created_at", "updated_at")}), + ) + + @admin.display(description=_("File preview")) + def preview_url(self, obj): + """Return a clickable preview URL for the file.""" + if obj.is_pending_upload: + return "-" + url = generate_download_file_url(obj, expires_in=60 * 60) + + return format_html( + 'Open File', url + ) + + def get_queryset(self, request): + """Hide hard deleted files in admin listing and lookups.""" + return super().get_queryset(request).filter(hard_deleted_at__isnull=True) + + def delete_model(self, request, obj): + """Hard delete instead of calling model.delete().""" + hard_delete_file(obj) + + def delete_queryset(self, request, queryset): + """Hard delete all selected files.""" + for file in queryset: + hard_delete_file(file) + + def has_add_permission(self, request): + return False + + class ResourceAccessInline(admin.TabularInline): """Admin class for the room user access model""" diff --git a/src/backend/core/utils.py b/src/backend/core/utils.py index c7280fc4..6951a99d 100644 --- a/src/backend/core/utils.py +++ b/src/backend/core/utils.py @@ -459,6 +459,41 @@ def generate_upload_policy(file): return policy +def generate_download_file_url(file, *, expires_in: int, override_domain: bool = True): + """ + Generate a S3 signed download url for a given file. + """ + + key = file.file_key + + # This setting should be used if the backend application and the frontend application + # can't connect to the object storage with the same domain. This is the case in the + # docker compose stack used in development. The frontend application will use localhost + # to connect to the object storage while the backend application will use the object storage + # service name declared in the docker compose stack. + # This is needed because the domain name is used to compute the signature. So it can't be + # changed dynamically by the frontend application. + if settings.AWS_S3_DOMAIN_REPLACE and override_domain: + s3_client = boto3.client( + "s3", + aws_access_key_id=settings.AWS_S3_ACCESS_KEY_ID, + aws_secret_access_key=settings.AWS_S3_SECRET_ACCESS_KEY, + endpoint_url=settings.AWS_S3_DOMAIN_REPLACE, + config=botocore.client.Config( + region_name=settings.AWS_S3_REGION_NAME, + signature_version=settings.AWS_S3_SIGNATURE_VERSION, + ), + ) + else: + s3_client = default_storage.connection.meta.client + + return s3_client.generate_presigned_url( + ClientMethod="get_object", + Params={"Bucket": default_storage.bucket_name, "Key": key}, + ExpiresIn=expires_in, + ) + + @lru_cache(maxsize=1) def _format_telephony_phone_number(raw_number, default_country): """Parse a configured phone number and return (country, international_format).