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

@@ -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 %}