From 1e486a608ee64b663f5da374963e052ba96e58db Mon Sep 17 00:00:00 2001 From: Herculino Trotta Date: Sat, 24 Oct 2026 13:26:37 -0300 Subject: [PATCH] fix(import): enforce object-level authorization on import profiles and runs ImportProfile and ImportRun had no owner, and neither the web views nor the REST viewsets filtered by the requesting user, so any authenticated user could read, edit and delete every other user's import profiles and runs (GHSA-q668-h56g-mfp4). - ImportProfile is now a SharedObject, with share and take-ownership views like transaction rules; the global unique constraint on name is dropped. - ImportRun is now an OwnedObject owned by the uploader, so runs made with a shared profile stay private to whoever ran them. - Existing runs are backfilled with the owner of their imported transactions; existing profiles stay unowned and usable by everyone. --- app/apps/api/serializers/imports.py | 5 + app/apps/api/views/imports.py | 9 +- .../migrations/0003_owner_and_sharing.py | 64 +++ app/apps/import_app/models.py | 18 +- .../import_app/tests/test_view_permissions.py | 440 ++++++++++++++++++ app/apps/import_app/urls.py | 10 + app/apps/import_app/views.py | 100 +++- .../import_app/fragments/profiles/list.html | 18 + .../import_app/fragments/profiles/share.html | 11 + 9 files changed, 660 insertions(+), 15 deletions(-) create mode 100644 app/apps/import_app/migrations/0003_owner_and_sharing.py create mode 100644 app/apps/import_app/tests/test_view_permissions.py create mode 100644 app/templates/import_app/fragments/profiles/share.html diff --git a/app/apps/api/serializers/imports.py b/app/apps/api/serializers/imports.py index 5a00cc2c..93f05587 100644 --- a/app/apps/api/serializers/imports.py +++ b/app/apps/api/serializers/imports.py @@ -39,3 +39,8 @@ class ImportFileSerializer(serializers.Serializer): queryset=ImportProfile.objects.all(), source="profile" ) file = serializers.FileField() + + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + # Reload the queryset to get an updated version with the requesting user + self.fields["profile_id"].queryset = ImportProfile.objects.all() diff --git a/app/apps/api/views/imports.py b/app/apps/api/views/imports.py index 40a5a7f9..f8e6edb7 100644 --- a/app/apps/api/views/imports.py +++ b/app/apps/api/views/imports.py @@ -37,6 +37,9 @@ class ImportProfileViewSet(viewsets.ReadOnlyModelViewSet): ordering_fields = '__all__' ordering = ['name'] + def get_queryset(self): + return ImportProfile.objects.all() + @extend_schema_view( list=extend_schema( @@ -81,7 +84,7 @@ class ImportRunViewSet(viewsets.ReadOnlyModelViewSet): ordering = ['-id'] def get_queryset(self): - queryset = super().get_queryset() + queryset = ImportRun.objects.all().order_by("-id") profile_id = self.request.query_params.get("profile_id") if profile_id: queryset = queryset.filter(profile_id=profile_id) @@ -132,7 +135,9 @@ class ImportViewSet(viewsets.ViewSet): file_path = fs.path(filename) # Create ImportRun record - import_run = ImportRun.objects.create(profile=profile, file_name=filename) + import_run = ImportRun.objects.create( + profile=profile, file_name=filename, owner=request.user + ) # Queue import task process_import.defer( diff --git a/app/apps/import_app/migrations/0003_owner_and_sharing.py b/app/apps/import_app/migrations/0003_owner_and_sharing.py new file mode 100644 index 00000000..f037264b --- /dev/null +++ b/app/apps/import_app/migrations/0003_owner_and_sharing.py @@ -0,0 +1,64 @@ +# Generated by Django 5.2.17 on 2026-10-24 16:17 + +import django.db.models.deletion +from django.conf import settings +from django.db import migrations, models + + +def backfill_import_run_owner(apps, schema_editor): + """Assign existing runs to the user who owns the transactions they imported. + + Runs whose transactions belong to several users, or that imported nothing, + are left without an owner and stay visible to everyone. + """ + ImportRun = apps.get_model("import_app", "ImportRun") + + for run in ImportRun.objects.filter(owner__isnull=True): + owner_ids = set( + run.transactions.exclude(owner__isnull=True).values_list( + "owner_id", flat=True + ) + ) + if len(owner_ids) == 1: + run.owner_id = owner_ids.pop() + run.save(update_fields=["owner"]) + + +class Migration(migrations.Migration): + + dependencies = [ + ('import_app', '0002_alter_importprofile_name_and_more'), + ('transactions', '0038_transaction_owner'), + migrations.swappable_dependency(settings.AUTH_USER_MODEL), + ] + + operations = [ + migrations.AddField( + model_name='importprofile', + name='owner', + field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.CASCADE, related_name='%(class)s_owned', to=settings.AUTH_USER_MODEL, verbose_name='Owner'), + ), + migrations.AddField( + model_name='importprofile', + name='shared_with', + field=models.ManyToManyField(blank=True, related_name='%(class)s_shared', to=settings.AUTH_USER_MODEL, verbose_name='Shared with users'), + ), + migrations.AddField( + model_name='importprofile', + name='visibility', + field=models.CharField(choices=[('private', 'Private'), ('public', 'Public')], default='private', max_length=10, verbose_name='Visibility'), + ), + migrations.AddField( + model_name='importrun', + name='owner', + field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.CASCADE, related_name='%(class)s_owned', to=settings.AUTH_USER_MODEL), + ), + migrations.AlterField( + model_name='importprofile', + name='name', + field=models.CharField(max_length=100, verbose_name='Name'), + ), + migrations.RunPython( + backfill_import_run_owner, migrations.RunPython.noop + ), + ] diff --git a/app/apps/import_app/models.py b/app/apps/import_app/models.py index 3eb739d1..8209688e 100644 --- a/app/apps/import_app/models.py +++ b/app/apps/import_app/models.py @@ -4,14 +4,20 @@ from django.core.exceptions import ValidationError from django.db import models from django.utils.translation import gettext_lazy as _ +from apps.common.models import ( + OwnedObject, + OwnedObjectManager, + SharedObject, + SharedObjectManager, +) from apps.import_app.schemas import version_1 -class ImportProfile(models.Model): +class ImportProfile(SharedObject): class Versions(models.IntegerChoices): VERSION_1 = 1, "Version 1" - name = models.CharField(max_length=100, verbose_name=_("Name"), unique=True) + name = models.CharField(max_length=100, verbose_name=_("Name")) yaml_config = models.TextField(verbose_name=_("YAML Configuration")) version = models.IntegerField( choices=Versions, @@ -19,6 +25,9 @@ class ImportProfile(models.Model): verbose_name=_("Version"), ) + objects = SharedObjectManager() + all_objects = models.Manager() # Unfiltered manager + def __str__(self): return self.name @@ -40,7 +49,7 @@ class ImportProfile(models.Model): ) -class ImportRun(models.Model): +class ImportRun(OwnedObject): class Status(models.TextChoices): QUEUED = "QUEUED", _("Queued") PROCESSING = "PROCESSING", _("Processing") @@ -85,3 +94,6 @@ class ImportRun(models.Model): failed_rows = models.IntegerField(default=0) started_at = models.DateTimeField(null=True) finished_at = models.DateTimeField(null=True) + + objects = OwnedObjectManager() + all_objects = models.Manager() # Unfiltered manager diff --git a/app/apps/import_app/tests/test_view_permissions.py b/app/apps/import_app/tests/test_view_permissions.py new file mode 100644 index 00000000..81f6c19e --- /dev/null +++ b/app/apps/import_app/tests/test_view_permissions.py @@ -0,0 +1,440 @@ +"""Object-level authorization tests for import profiles and import runs. + +Regression coverage for GHSA-q668-h56g-mfp4: ImportProfile and ImportRun had no +owner, and neither the web views nor the REST viewsets filtered by the +requesting user, so any authenticated user could read, edit and delete every +other user's profiles and runs. +""" + +from unittest.mock import patch + +from django.contrib.auth import get_user_model +from django.core.files.uploadedfile import SimpleUploadedFile +from django.test import TestCase, override_settings +from django.urls import reverse +from rest_framework import status +from rest_framework.test import APIClient + +from apps.import_app.models import ImportProfile, ImportRun + +HTMX = {"HTTP_HX_REQUEST": "true"} + +YAML_CONFIG = """ +settings: + file_type: csv + importing: transactions + trigger_transaction_rules: false +mapping: + date_field: + source: date + target: date + format: "%Y-%m-%d" + amount_field: + source: amount + target: amount + description_field: + source: description + target: description + account_field: + source: account + target: account + type: id +""" + + +@override_settings( + STORAGES={ + "default": {"BACKEND": "django.core.files.storage.FileSystemStorage"}, + "staticfiles": { + "BACKEND": "django.contrib.staticfiles.storage.StaticFilesStorage" + }, + }, + WHITENOISE_AUTOREFRESH=True, + DEMO=False, +) +class ImportObjectPermissionTestCase(TestCase): + def setUp(self): + User = get_user_model() + self.owner = User.objects.create_user( + email="owner@test.com", password="testpass123" + ) + self.shared_user = User.objects.create_user( + email="shared@test.com", password="testpass123" + ) + self.stranger = User.objects.create_user( + email="stranger@test.com", password="testpass123" + ) + + # Private and unshared: invisible to everyone but the owner. + self.private_profile = ImportProfile.all_objects.create( + name="Private profile", + yaml_config=YAML_CONFIG, + owner=self.owner, + visibility="private", + ) + # Private but shared with shared_user: usable by them, not by stranger. + self.shared_profile = ImportProfile.all_objects.create( + name="Shared profile", + yaml_config=YAML_CONFIG, + owner=self.owner, + visibility="private", + ) + self.shared_profile.shared_with.add(self.shared_user) + # Legacy profile created before profiles had owners. + self.unowned_profile = ImportProfile.all_objects.create( + name="Unowned profile", yaml_config=YAML_CONFIG, owner=None + ) + + self.owner_run = ImportRun.all_objects.create( + profile=self.shared_profile, + file_name="owner-statement.csv", + logs="owner log", + owner=self.owner, + ) + self.shared_user_run = ImportRun.all_objects.create( + profile=self.shared_profile, + file_name="shared-user-statement.csv", + logs="shared user log", + owner=self.shared_user, + ) + + def login(self, user): + self.client.force_login(user) + + +class ImportProfileViewPermissionTests(ImportObjectPermissionTestCase): + def test_list_only_shows_visible_profiles(self): + self.login(self.stranger) + response = self.client.get(reverse("import_profiles_list"), **HTMX) + + self.assertEqual(response.status_code, 200) + self.assertNotContains(response, "Private profile") + self.assertNotContains(response, "Shared profile") + self.assertContains(response, "Unowned profile") + + def test_list_shows_profiles_shared_with_user(self): + self.login(self.shared_user) + response = self.client.get(reverse("import_profiles_list"), **HTMX) + + self.assertContains(response, "Shared profile") + self.assertNotContains(response, "Private profile") + + def test_stranger_cannot_edit_private_profile(self): + self.login(self.stranger) + response = self.client.post( + reverse( + "import_profile_edit", kwargs={"profile_id": self.private_profile.id} + ), + {"name": "Hijacked", "version": 1, "yaml_config": YAML_CONFIG}, + **HTMX, + ) + + self.assertEqual(response.status_code, 404) + self.private_profile.refresh_from_db() + self.assertEqual(self.private_profile.name, "Private profile") + + def test_shared_user_cannot_edit_shared_profile(self): + self.login(self.shared_user) + response = self.client.post( + reverse( + "import_profile_edit", kwargs={"profile_id": self.shared_profile.id} + ), + {"name": "Hijacked", "version": 1, "yaml_config": YAML_CONFIG}, + **HTMX, + ) + + self.assertEqual(response.status_code, 403) + self.shared_profile.refresh_from_db() + self.assertEqual(self.shared_profile.name, "Shared profile") + + def test_owner_can_edit_profile(self): + self.login(self.owner) + response = self.client.post( + reverse( + "import_profile_edit", kwargs={"profile_id": self.private_profile.id} + ), + { + "name": "Renamed", + "version": 1, + "yaml_config": YAML_CONFIG, + "submit": "Update", + }, + **HTMX, + ) + + self.assertEqual(response.status_code, 204) + self.private_profile.refresh_from_db() + self.assertEqual(self.private_profile.name, "Renamed") + + def test_stranger_cannot_delete_private_profile(self): + self.login(self.stranger) + response = self.client.delete( + reverse( + "import_profile_delete", + kwargs={"profile_id": self.private_profile.id}, + ), + **HTMX, + ) + + self.assertEqual(response.status_code, 404) + self.assertTrue( + ImportProfile.all_objects.filter(id=self.private_profile.id).exists() + ) + + def test_shared_user_delete_only_removes_their_access(self): + self.login(self.shared_user) + response = self.client.delete( + reverse( + "import_profile_delete", kwargs={"profile_id": self.shared_profile.id} + ), + **HTMX, + ) + + self.assertEqual(response.status_code, 204) + self.assertTrue( + ImportProfile.all_objects.filter(id=self.shared_profile.id).exists() + ) + self.assertFalse( + self.shared_profile.shared_with.filter(pk=self.shared_user.pk).exists() + ) + self.assertTrue( + ImportRun.all_objects.filter(id=self.owner_run.id).exists() + ) + + def test_owner_can_delete_profile(self): + self.login(self.owner) + response = self.client.delete( + reverse( + "import_profile_delete", + kwargs={"profile_id": self.private_profile.id}, + ), + **HTMX, + ) + + self.assertEqual(response.status_code, 204) + self.assertFalse( + ImportProfile.all_objects.filter(id=self.private_profile.id).exists() + ) + + def test_shared_user_cannot_change_share_settings(self): + self.login(self.shared_user) + response = self.client.post( + reverse( + "import_profile_share_settings", kwargs={"pk": self.shared_profile.id} + ), + {"owner": self.shared_user.id, "visibility": "public"}, + **HTMX, + ) + + self.assertEqual(response.status_code, 403) + self.shared_profile.refresh_from_db() + self.assertEqual(self.shared_profile.owner, self.owner) + + def test_take_ownership_of_unowned_profile(self): + self.login(self.stranger) + response = self.client.get( + reverse( + "import_profile_take_ownership", + kwargs={"profile_id": self.unowned_profile.id}, + ), + **HTMX, + ) + + self.assertEqual(response.status_code, 204) + self.unowned_profile.refresh_from_db() + self.assertEqual(self.unowned_profile.owner, self.stranger) + + def test_cannot_take_ownership_of_owned_profile(self): + self.login(self.shared_user) + response = self.client.get( + reverse( + "import_profile_take_ownership", + kwargs={"profile_id": self.shared_profile.id}, + ), + **HTMX, + ) + + self.assertEqual(response.status_code, 403) + self.shared_profile.refresh_from_db() + self.assertEqual(self.shared_profile.owner, self.owner) + + +class ImportRunViewPermissionTests(ImportObjectPermissionTestCase): + def test_runs_list_only_shows_own_runs(self): + self.login(self.shared_user) + response = self.client.get( + reverse( + "import_profile_runs_list", + kwargs={"profile_id": self.shared_profile.id}, + ), + **HTMX, + ) + + self.assertEqual(response.status_code, 200) + self.assertContains(response, "shared-user-statement.csv") + self.assertNotContains(response, "owner-statement.csv") + + def test_stranger_cannot_list_runs_of_private_profile(self): + self.login(self.stranger) + response = self.client.get( + reverse( + "import_profile_runs_list", + kwargs={"profile_id": self.shared_profile.id}, + ), + **HTMX, + ) + + self.assertEqual(response.status_code, 404) + + def test_cannot_read_another_users_run_log(self): + self.login(self.shared_user) + response = self.client.get( + reverse( + "import_run_log", + kwargs={ + "profile_id": self.shared_profile.id, + "run_id": self.owner_run.id, + }, + ), + **HTMX, + ) + + self.assertEqual(response.status_code, 404) + + def test_cannot_delete_another_users_run(self): + self.login(self.shared_user) + response = self.client.delete( + reverse( + "import_run_delete", + kwargs={ + "profile_id": self.shared_profile.id, + "run_id": self.owner_run.id, + }, + ), + **HTMX, + ) + + self.assertEqual(response.status_code, 404) + self.assertTrue(ImportRun.all_objects.filter(id=self.owner_run.id).exists()) + + def test_owner_can_delete_own_run(self): + self.login(self.shared_user) + response = self.client.delete( + reverse( + "import_run_delete", + kwargs={ + "profile_id": self.shared_profile.id, + "run_id": self.shared_user_run.id, + }, + ), + **HTMX, + ) + + self.assertEqual(response.status_code, 204) + self.assertFalse( + ImportRun.all_objects.filter(id=self.shared_user_run.id).exists() + ) + + def test_stranger_cannot_import_with_private_profile(self): + self.login(self.stranger) + response = self.client.get( + reverse("import_run_add", kwargs={"profile_id": self.private_profile.id}), + **HTMX, + ) + + self.assertEqual(response.status_code, 404) + + @patch("apps.import_app.views.process_import.defer") + @patch("apps.import_app.views.FileSystemStorage.save", return_value="file.csv") + @patch("apps.import_app.views.FileSystemStorage.path", return_value="/tmp/file.csv") + def test_import_with_shared_profile_is_owned_by_uploader( + self, mock_path, mock_save, mock_defer + ): + self.login(self.shared_user) + response = self.client.post( + reverse("import_run_add", kwargs={"profile_id": self.shared_profile.id}), + {"file": SimpleUploadedFile("file.csv", b"date,amount\n")}, + **HTMX, + ) + + self.assertEqual(response.status_code, 204) + run = ImportRun.all_objects.latest("id") + self.assertEqual(run.owner, self.shared_user) + self.assertEqual(run.profile, self.shared_profile) + + +class ImportAPIPermissionTests(ImportObjectPermissionTestCase): + def setUp(self): + super().setUp() + self.api = APIClient() + + def test_profiles_list_only_returns_visible_profiles(self): + self.api.force_authenticate(user=self.stranger) + response = self.api.get("/api/import/profiles/") + + self.assertEqual(response.status_code, status.HTTP_200_OK) + names = {p["name"] for p in response.data["results"]} + self.assertEqual(names, {"Unowned profile"}) + + def test_cannot_retrieve_private_profile(self): + self.api.force_authenticate(user=self.stranger) + response = self.api.get(f"/api/import/profiles/{self.private_profile.id}/") + + self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND) + + def test_runs_list_only_returns_own_runs(self): + self.api.force_authenticate(user=self.stranger) + response = self.api.get("/api/import/runs/") + + self.assertEqual(response.status_code, status.HTTP_200_OK) + self.assertEqual(response.data["count"], 0) + + def test_runs_list_on_shared_profile_hides_other_users_runs(self): + self.api.force_authenticate(user=self.shared_user) + response = self.api.get( + "/api/import/runs/", {"profile_id": self.shared_profile.id} + ) + + self.assertEqual(response.status_code, status.HTTP_200_OK) + ids = {r["id"] for r in response.data["results"]} + self.assertEqual(ids, {self.shared_user_run.id}) + + def test_cannot_retrieve_another_users_run(self): + self.api.force_authenticate(user=self.stranger) + response = self.api.get(f"/api/import/runs/{self.owner_run.id}/") + + self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND) + + def test_cannot_import_with_private_profile(self): + self.api.force_authenticate(user=self.stranger) + response = self.api.post( + "/api/import/import/", + { + "profile_id": self.private_profile.id, + "file": SimpleUploadedFile("file.csv", b"date,amount\n"), + }, + format="multipart", + ) + + self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) + self.assertIn("profile_id", response.data) + + @patch("apps.import_app.tasks.process_import.defer") + @patch("django.core.files.storage.FileSystemStorage.save", return_value="file.csv") + @patch("django.core.files.storage.FileSystemStorage.path", return_value="/tmp/file.csv") + def test_import_with_shared_profile_is_owned_by_uploader( + self, mock_path, mock_save, mock_defer + ): + self.api.force_authenticate(user=self.shared_user) + response = self.api.post( + "/api/import/import/", + { + "profile_id": self.shared_profile.id, + "file": SimpleUploadedFile("file.csv", b"date,amount\n"), + }, + format="multipart", + ) + + self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED) + run = ImportRun.all_objects.get(id=response.data["import_run_id"]) + self.assertEqual(run.owner, self.shared_user) diff --git a/app/apps/import_app/urls.py b/app/apps/import_app/urls.py index 549e0c8c..df1de9b4 100644 --- a/app/apps/import_app/urls.py +++ b/app/apps/import_app/urls.py @@ -32,6 +32,16 @@ urlpatterns = [ views.import_profile_edit, name="import_profile_edit", ), + path( + "import/profiles//take-ownership/", + views.import_profile_take_ownership, + name="import_profile_take_ownership", + ), + path( + "import/profiles//share/", + views.import_profile_share, + name="import_profile_share_settings", + ), path( "import/profiles//runs/list/", views.import_runs_list, diff --git a/app/apps/import_app/views.py b/app/apps/import_app/views.py index 7309ecc2..adef7679 100644 --- a/app/apps/import_app/views.py +++ b/app/apps/import_app/views.py @@ -2,6 +2,7 @@ import shutil from django.contrib import messages from django.contrib.auth.decorators import login_required +from django.core.exceptions import PermissionDenied from django.core.files.storage import FileSystemStorage from django.http import HttpResponse from django.shortcuts import render, get_object_or_404 @@ -9,6 +10,13 @@ from django.utils.translation import gettext_lazy as _ from django.views.decorators.http import require_http_methods from apps.common.decorators.htmx import only_htmx +from apps.common.forms import SharedObjectForm +from apps.common.functions.permissions import ( + EDIT, + READ, + get_shared_object_or_error, +) +from apps.common.models import SharedObject from apps.import_app.forms import ImportRunFileUploadForm, ImportProfileForm from apps.import_app.models import ImportRun, ImportProfile from apps.import_app.services import PresetService @@ -93,7 +101,9 @@ def import_profile_add(request): @disabled_on_demo @require_http_methods(["GET", "POST"]) def import_profile_edit(request, profile_id): - profile = get_object_or_404(ImportProfile, id=profile_id) + profile = get_shared_object_or_error( + ImportProfile, request, id=profile_id, level=EDIT + ) if request.method == "POST": form = ImportProfileForm(request.POST, instance=profile) @@ -123,11 +133,20 @@ def import_profile_edit(request, profile_id): @disabled_on_demo @require_http_methods(["DELETE"]) def import_profile_delete(request, profile_id): - profile = ImportProfile.objects.get(id=profile_id) + profile = get_shared_object_or_error( + ImportProfile, request, id=profile_id, level=READ + ) - profile.delete() - - messages.success(request, _("Import Profile deleted successfully")) + if profile.is_editable_by(request.user): + profile.delete() + messages.success(request, _("Import Profile deleted successfully")) + elif profile.shared_with.filter(pk=request.user.pk).exists(): + # Someone else's profile shared with us: we can drop our own access to + # it, but never delete it. + profile.shared_with.remove(request.user) + messages.success(request, _("Item no longer shared with you")) + else: + raise PermissionDenied return HttpResponse( status=204, @@ -142,8 +161,12 @@ def import_profile_delete(request, profile_id): @disabled_on_demo @require_http_methods(["GET", "POST"]) def import_runs_list(request, profile_id): - profile = ImportProfile.objects.get(id=profile_id) + profile = get_shared_object_or_error( + ImportProfile, request, id=profile_id, level=READ + ) + # ImportRun.objects only returns the requesting user's runs, so a shared + # profile does not expose the runs other users made with it. runs = ImportRun.objects.filter(profile=profile).order_by("-id") return render( @@ -158,7 +181,7 @@ def import_runs_list(request, profile_id): @disabled_on_demo @require_http_methods(["GET", "POST"]) def import_run_log(request, profile_id, run_id): - run = ImportRun.objects.get(profile__id=profile_id, id=run_id) + run = get_object_or_404(ImportRun, profile__id=profile_id, id=run_id) return render( request, @@ -172,7 +195,9 @@ def import_run_log(request, profile_id, run_id): @disabled_on_demo @require_http_methods(["GET", "POST"]) def import_run_add(request, profile_id): - profile = ImportProfile.objects.get(id=profile_id) + profile = get_shared_object_or_error( + ImportProfile, request, id=profile_id, level=READ + ) if request.method == "POST": form = ImportRunFileUploadForm(request.POST, request.FILES) @@ -183,7 +208,9 @@ def import_run_add(request, profile_id): filename = fs.save(uploaded_file.name, uploaded_file) file_path = fs.path(filename) - import_run = ImportRun.objects.create(profile=profile, file_name=filename) + import_run = ImportRun.objects.create( + profile=profile, file_name=filename, owner=request.user + ) # Defer the procrastinate task process_import.defer( @@ -215,7 +242,7 @@ def import_run_add(request, profile_id): @disabled_on_demo @require_http_methods(["DELETE"]) def import_run_delete(request, profile_id, run_id): - run = ImportRun.objects.get(profile__id=profile_id, id=run_id) + run = get_object_or_404(ImportRun, profile__id=profile_id, id=run_id) run.delete() @@ -227,3 +254,56 @@ def import_run_delete(request, profile_id, run_id): "HX-Trigger": "updated", }, ) + + +@only_htmx +@login_required +@disabled_on_demo +@require_http_methods(["GET"]) +def import_profile_take_ownership(request, profile_id): + profile = get_shared_object_or_error( + ImportProfile, request, id=profile_id, level=EDIT + ) + + if not profile.owner: + profile.owner = request.user + profile.visibility = SharedObject.Visibility.private + profile.save() + + messages.success(request, _("Ownership taken successfully")) + + return HttpResponse( + status=204, + headers={ + "HX-Trigger": "updated, hide_offcanvas", + }, + ) + + +@only_htmx +@login_required +@disabled_on_demo +@require_http_methods(["GET", "POST"]) +def import_profile_share(request, pk): + obj = get_shared_object_or_error(ImportProfile, request, id=pk, level=EDIT) + + if request.method == "POST": + form = SharedObjectForm(request.POST, instance=obj, user=request.user) + if form.is_valid(): + form.save() + messages.success(request, _("Configuration saved successfully")) + + return HttpResponse( + status=204, + headers={ + "HX-Trigger": "updated, hide_offcanvas", + }, + ) + else: + form = SharedObjectForm(instance=obj, user=request.user) + + return render( + request, + "import_app/fragments/profiles/share.html", + {"form": form, "object": obj}, + ) diff --git a/app/templates/import_app/fragments/profiles/list.html b/app/templates/import_app/fragments/profiles/list.html index 96025621..ef8c86d2 100644 --- a/app/templates/import_app/fragments/profiles/list.html +++ b/app/templates/import_app/fragments/profiles/list.html @@ -40,12 +40,30 @@
+ {% if not profile.owner or user == profile.owner %} + {% endif %} + {% if not profile.owner %} + + + {% endif %} + {% if user == profile.owner %} + + + {% endif %} + {% crispy form %} + +{% endblock %}