Merge pull request #629 from eitchtee/fix/import-object-authorization

fix(import): enforce object-level authorization on import profiles and runs

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()

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(

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
+        ),
+    ]

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

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)

app/apps/import_app/urls.py

@@ -32,6 +32,16 @@ urlpatterns = [
         views.import_profile_edit,
         name="import_profile_edit",
     ),
+    path(
+        "import/profiles/<int:profile_id>/take-ownership/",
+        views.import_profile_take_ownership,
+        name="import_profile_take_ownership",
+    ),
+    path(
+        "import/profiles/<int:pk>/share/",
+        views.import_profile_share,
+        name="import_profile_share_settings",
+    ),
     path(
         "import/profiles/<int:profile_id>/runs/list/",
         views.import_runs_list,

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},
+    )

app/templates/import_app/fragments/profiles/list.html

@@ -40,12 +40,30 @@
             <tr class="profile">
               <td class="table-col-auto">
                 <div class="join" role="group" aria-label="{% translate 'Actions' %}">
+                  {% if not profile.owner or user == profile.owner %}
                   <a class="btn btn-secondary btn-sm join-item"
                      role="button"
                      data-tippy-content="{% translate "Edit" %}"
                      hx-get="{% url 'import_profile_edit' profile_id=profile.id %}"
                      hx-target="#generic-offcanvas">
                     <i class="fa-solid fa-pencil fa-fw"></i></a>
+                  {% endif %}
+                  {% if not profile.owner %}
+                  <a class="btn btn-secondary btn-sm join-item"
+                     role="button"
+                     data-tippy-content="{% translate "Take ownership" %}"
+                     hx-get="{% url 'import_profile_take_ownership' profile_id=profile.id %}">
+                    <i class="fa-solid fa-crown fa-fw"></i></a>
+                  {% endif %}
+                  {% if user == profile.owner %}
+                  <a class="btn btn-secondary btn-sm join-item"
+                     role="button"
+                     hx-target="#generic-offcanvas"
+                     hx-swap="innerHTML"
+                     data-tippy-content="{% translate "Share" %}"
+                     hx-get="{% url 'import_profile_share_settings' pk=profile.id %}">
+                    <i class="fa-solid fa-share fa-fw"></i></a>
+                  {% endif %}
                   <a class="btn btn-secondary btn-sm join-item"
                      role="button"
                      data-tippy-content="{% translate "Runs" %}"

app/templates/import_app/fragments/profiles/share.html

@@ -0,0 +1,11 @@
+{% extends 'extends/offcanvas.html' %}
+{% load i18n %}
+{% load crispy_forms_tags %}
+
+{% block title %}{% translate 'Share settings' %}{% endblock %}
+
+{% block body %}
+<form hx-post="{% url "import_profile_share_settings" pk=object.id %}" hx-target="#generic-offcanvas" novalidate>
+    {% crispy form %}
+</form>
+{% endblock %}