diff --git a/app/apps/accounts/views/account_groups.py b/app/apps/accounts/views/account_groups.py index d9b0825..75687e0 100644 --- a/app/apps/accounts/views/account_groups.py +++ b/app/apps/accounts/views/account_groups.py @@ -1,13 +1,19 @@ from django.contrib import messages from django.contrib.auth.decorators import login_required +from django.core.exceptions import PermissionDenied from django.http import HttpResponse -from django.shortcuts import render, get_object_or_404 +from django.shortcuts import render from django.utils.translation import gettext_lazy as _ from django.views.decorators.http import require_http_methods from apps.accounts.forms import AccountGroupForm from apps.accounts.models import AccountGroup from apps.common.decorators.htmx import only_htmx +from apps.common.functions.permissions import ( + EDIT, + READ, + get_shared_object_or_error, +) from apps.common.models import SharedObject from apps.common.forms import SharedObjectForm @@ -63,17 +69,7 @@ def account_group_add(request, **kwargs): @login_required @require_http_methods(["GET", "POST"]) def account_group_edit(request, pk): - account_group = get_object_or_404(AccountGroup, id=pk) - - if account_group.owner and account_group.owner != request.user: - messages.error(request, _("Only the owner can edit this")) - - return HttpResponse( - status=204, - headers={ - "HX-Trigger": "updated, hide_offcanvas", - }, - ) + account_group = get_shared_object_or_error(AccountGroup, request, id=pk, level=EDIT) if request.method == "POST": form = AccountGroupForm(request.POST, instance=account_group) @@ -101,17 +97,18 @@ def account_group_edit(request, pk): @login_required @require_http_methods(["DELETE"]) def account_group_delete(request, pk): - account_group = get_object_or_404(AccountGroup, id=pk) + account_group = get_shared_object_or_error(AccountGroup, request, id=pk, level=READ) - if ( - account_group.owner != request.user - and request.user in account_group.shared_with.all() - ): + if account_group.is_editable_by(request.user): + account_group.delete() + messages.success(request, _("Account Group deleted successfully")) + elif account_group.shared_with.filter(pk=request.user.pk).exists(): + # Someone else's object shared with us: we can drop our own access + # to it, but never delete it. account_group.shared_with.remove(request.user) messages.success(request, _("Item no longer shared with you")) else: - account_group.delete() - messages.success(request, _("Account Group deleted successfully")) + raise PermissionDenied return HttpResponse( status=204, @@ -125,7 +122,7 @@ def account_group_delete(request, pk): @login_required @require_http_methods(["GET"]) def account_group_take_ownership(request, pk): - account_group = get_object_or_404(AccountGroup, id=pk) + account_group = get_shared_object_or_error(AccountGroup, request, id=pk, level=EDIT) if not account_group.owner: account_group.owner = request.user @@ -146,17 +143,7 @@ def account_group_take_ownership(request, pk): @login_required @require_http_methods(["GET", "POST"]) def account_group_share(request, pk): - obj = get_object_or_404(AccountGroup, id=pk) - - if obj.owner and obj.owner != request.user: - messages.error(request, _("Only the owner can edit this")) - - return HttpResponse( - status=204, - headers={ - "HX-Trigger": "updated, hide_offcanvas", - }, - ) + obj = get_shared_object_or_error(AccountGroup, request, id=pk, level=EDIT) if request.method == "POST": form = SharedObjectForm(request.POST, instance=obj, user=request.user) diff --git a/app/apps/accounts/views/accounts.py b/app/apps/accounts/views/accounts.py index 4908bb7..bfab255 100644 --- a/app/apps/accounts/views/accounts.py +++ b/app/apps/accounts/views/accounts.py @@ -1,13 +1,19 @@ from django.contrib import messages from django.contrib.auth.decorators import login_required +from django.core.exceptions import PermissionDenied from django.http import HttpResponse -from django.shortcuts import render, get_object_or_404 +from django.shortcuts import render from django.utils.translation import gettext_lazy as _ from django.views.decorators.http import require_http_methods from apps.accounts.forms import AccountForm from apps.accounts.models import Account from apps.common.decorators.htmx import only_htmx +from apps.common.functions.permissions import ( + EDIT, + READ, + get_shared_object_or_error, +) from apps.common.models import SharedObject from apps.common.forms import SharedObjectForm @@ -63,16 +69,7 @@ def account_add(request, **kwargs): @login_required @require_http_methods(["GET", "POST"]) def account_edit(request, pk): - account = get_object_or_404(Account, id=pk) - if account.owner and account.owner != request.user: - messages.error(request, _("Only the owner can edit this")) - - return HttpResponse( - status=204, - headers={ - "HX-Trigger": "updated, hide_offcanvas", - }, - ) + account = get_shared_object_or_error(Account, request, id=pk, level=EDIT) if request.method == "POST": form = AccountForm(request.POST, instance=account) @@ -100,17 +97,7 @@ def account_edit(request, pk): @login_required @require_http_methods(["GET", "POST"]) def account_share(request, pk): - obj = get_object_or_404(Account, id=pk) - - if obj.owner and obj.owner != request.user: - messages.error(request, _("Only the owner can edit this")) - - return HttpResponse( - status=204, - headers={ - "HX-Trigger": "updated, hide_offcanvas", - }, - ) + obj = get_shared_object_or_error(Account, request, id=pk, level=EDIT) if request.method == "POST": form = SharedObjectForm(request.POST, instance=obj, user=request.user) @@ -138,14 +125,18 @@ def account_share(request, pk): @login_required @require_http_methods(["DELETE"]) def account_delete(request, pk): - account = get_object_or_404(Account, id=pk) + account = get_shared_object_or_error(Account, request, id=pk, level=READ) - if account.owner != request.user and request.user in account.shared_with.all(): + if account.is_editable_by(request.user): + account.delete() + messages.success(request, _("Account deleted successfully")) + elif account.shared_with.filter(pk=request.user.pk).exists(): + # Someone else's object shared with us: we can drop our own access + # to it, but never delete it. account.shared_with.remove(request.user) messages.success(request, _("Item no longer shared with you")) else: - account.delete() - messages.success(request, _("Account deleted successfully")) + raise PermissionDenied return HttpResponse( status=204, @@ -159,7 +150,9 @@ def account_delete(request, pk): @login_required @require_http_methods(["GET"]) def account_toggle_untracked(request, pk): - account = get_object_or_404(Account, id=pk) + # Only flips the calling user's own row in untracked_by, so visibility -- + # not ownership -- is the right bar here. + account = get_shared_object_or_error(Account, request, id=pk, level=READ) if account.is_untracked_by(): account.untracked_by.remove(request.user) messages.success(request, _("Account is now tracked")) @@ -179,7 +172,7 @@ def account_toggle_untracked(request, pk): @login_required @require_http_methods(["GET"]) def account_take_ownership(request, pk): - account = get_object_or_404(Account, id=pk) + account = get_shared_object_or_error(Account, request, id=pk, level=EDIT) if not account.owner: account.owner = request.user diff --git a/app/apps/api/permissions.py b/app/apps/api/permissions.py index c8e19c5..b3d3995 100644 --- a/app/apps/api/permissions.py +++ b/app/apps/api/permissions.py @@ -1,4 +1,8 @@ -from rest_framework.permissions import BasePermission +from rest_framework.permissions import ( + SAFE_METHODS, + BasePermission, + DjangoModelPermissions, +) from django.conf import settings @@ -8,3 +12,37 @@ class NotInDemoMode(BasePermission): return False else: return True + + +class SharedObjectPermission(BasePermission): + """Object-level ownership check for SharedObject-backed viewsets. + + DjangoModelPermissions is model-level: a user holding ``change_account`` + may write any object the viewset's queryset returns, and for SharedObject + that queryset includes other people's public and shared-with-them objects. + Sharing grants read access only, so writes are restricted to the owner + here as well. + + Set ``shared_object_via`` on the viewset when the governing SharedObject is + reached through a relation (e.g. ``"strategy"`` for a DCA entry). + """ + + def has_object_permission(self, request, view, obj): + if request.method in SAFE_METHODS: + return True + + guard = obj + via = getattr(view, "shared_object_via", None) + for attr in via.split(".") if via else []: + guard = getattr(guard, attr) + + return guard.is_editable_by(request.user) + + +#: Default permissions plus the object-level ownership check. Assigning +#: ``permission_classes`` replaces the defaults, so they are repeated here. +SHARED_OBJECT_PERMISSIONS = [ + NotInDemoMode, + DjangoModelPermissions, + SharedObjectPermission, +] diff --git a/app/apps/api/tests/__init__.py b/app/apps/api/tests/__init__.py index 9ca8b36..d67d54c 100644 --- a/app/apps/api/tests/__init__.py +++ b/app/apps/api/tests/__init__.py @@ -3,3 +3,4 @@ from .test_imports import * from .test_accounts import * from .test_data_isolation import * from .test_shared_access import * +from .test_object_permissions import * diff --git a/app/apps/api/tests/test_object_permissions.py b/app/apps/api/tests/test_object_permissions.py new file mode 100644 index 0000000..54d4fb9 --- /dev/null +++ b/app/apps/api/tests/test_object_permissions.py @@ -0,0 +1,171 @@ +"""Object-level ownership on the SharedObject API viewsets. + +DjangoModelPermissions is model-level: a user holding ``change_account`` could +write any object the viewset's queryset returned, and for SharedObject that +queryset includes other people's public and shared-with-them objects. Ordinary +users hold no model permissions, so this was not reachable for them, but the +object-level check was missing entirely. + +Reads are unaffected -- they stay governed by SharedObjectManager. +""" + +from datetime import date +from decimal import Decimal + +from django.contrib.auth import get_user_model +from django.contrib.auth.models import Permission +from django.test import TestCase, override_settings +from rest_framework import status +from rest_framework.test import APIClient + +from apps.accounts.models import Account +from apps.currencies.models import Currency +from apps.dca.models import DCAEntry, DCAStrategy +from apps.transactions.models import TransactionCategory + + +@override_settings( + STORAGES={ + "default": {"BACKEND": "django.core.files.storage.FileSystemStorage"}, + "staticfiles": { + "BACKEND": "django.contrib.staticfiles.storage.StaticFilesStorage" + }, + }, + WHITENOISE_AUTOREFRESH=True, + DEMO=False, +) +class SharedObjectAPIPermissionTests(TestCase): + """The attacker here deliberately HOLDS the Django model permissions. + + Without them DjangoModelPermissions already answers 403 and the + object-level check is never consulted, so the test would pass whether or + not it exists. + """ + + def setUp(self): + User = get_user_model() + self.owner = User.objects.create_user( + email="owner@test.com", password="testpass123" + ) + self.attacker = User.objects.create_user( + email="attacker@test.com", password="testpass123" + ) + self.attacker.user_permissions.set( + Permission.objects.filter( + codename__in=[ + "add_account", + "change_account", + "delete_account", + "add_transactioncategory", + "change_transactioncategory", + "delete_transactioncategory", + "add_dcaentry", + "change_dcaentry", + "delete_dcaentry", + ] + ) + ) + # Permissions are cached on the user instance. + self.attacker = User.objects.get(pk=self.attacker.pk) + + self.currency = Currency.objects.create( + code="USD", name="US Dollar", decimal_places=2 + ) + + self.api = APIClient() + self.api.force_authenticate(user=self.attacker) + + def test_cannot_modify_public_account(self): + account = Account.all_objects.create( + name="Public account", + currency=self.currency, + owner=self.owner, + visibility="public", + ) + + response = self.api.patch( + f"/api/accounts/{account.id}/", {"name": "HIJACKED"}, format="json" + ) + + self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) + account.refresh_from_db() + self.assertEqual(account.name, "Public account") + + def test_cannot_delete_account_shared_with_them(self): + account = Account.all_objects.create( + name="Shared account", + currency=self.currency, + owner=self.owner, + visibility="private", + ) + account.shared_with.add(self.attacker) + + response = self.api.delete(f"/api/accounts/{account.id}/") + + self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) + self.assertTrue(Account.all_objects.filter(pk=account.pk).exists()) + + def test_cannot_delete_public_category(self): + category = TransactionCategory.all_objects.create( + name="Public category", owner=self.owner, visibility="public" + ) + + response = self.api.delete(f"/api/categories/{category.id}/") + + self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) + self.assertTrue(TransactionCategory.all_objects.filter(pk=category.pk).exists()) + + def test_cannot_delete_entry_on_public_strategy(self): + strategy = DCAStrategy.all_objects.create( + name="Public strategy", + owner=self.owner, + visibility="public", + target_currency=self.currency, + payment_currency=self.currency, + ) + entry = DCAEntry.objects.create( + strategy=strategy, + date=date(2025, 1, 1), + amount_paid=Decimal("100"), + amount_received=Decimal("1"), + ) + + response = self.api.delete(f"/api/dca/entries/{entry.id}/") + + self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) + self.assertTrue(DCAEntry.objects.filter(pk=entry.pk).exists()) + + def test_reads_of_shared_objects_still_work(self): + account = Account.all_objects.create( + name="Public account", + currency=self.currency, + owner=self.owner, + visibility="public", + ) + + response = self.api.get(f"/api/accounts/{account.id}/") + + self.assertEqual(response.status_code, status.HTTP_200_OK) + + def test_owner_can_still_write_their_own_objects(self): + owner_client = APIClient() + self.owner.user_permissions.set( + Permission.objects.filter(codename__in=["change_account", "delete_account"]) + ) + owner = get_user_model().objects.get(pk=self.owner.pk) + owner_client.force_authenticate(user=owner) + + account = Account.all_objects.create( + name="Own account", + currency=self.currency, + owner=owner, + visibility="private", + ) + + response = owner_client.patch( + f"/api/accounts/{account.id}/", {"name": "Renamed"}, format="json" + ) + + self.assertEqual(response.status_code, status.HTTP_200_OK) + account.refresh_from_db() + self.assertEqual(account.name, "Renamed") diff --git a/app/apps/api/views/accounts.py b/app/apps/api/views/accounts.py index 757001c..f7e44e1 100644 --- a/app/apps/api/views/accounts.py +++ b/app/apps/api/views/accounts.py @@ -6,6 +6,7 @@ from rest_framework.response import Response from apps.accounts.models import AccountGroup, Account from apps.accounts.services import get_account_balance +from apps.api.permissions import SHARED_OBJECT_PERMISSIONS from apps.api.serializers import ( AccountGroupSerializer, AccountSerializer, @@ -16,6 +17,7 @@ from apps.api.serializers import ( class AccountGroupViewSet(viewsets.ModelViewSet): """ViewSet for managing account groups.""" + permission_classes = SHARED_OBJECT_PERMISSIONS queryset = AccountGroup.objects.all() serializer_class = AccountGroupSerializer filterset_fields = { @@ -40,6 +42,7 @@ class AccountGroupViewSet(viewsets.ModelViewSet): class AccountViewSet(viewsets.ModelViewSet): """ViewSet for managing accounts.""" + permission_classes = SHARED_OBJECT_PERMISSIONS queryset = Account.objects.all() serializer_class = AccountSerializer filterset_fields = { diff --git a/app/apps/api/views/dca.py b/app/apps/api/views/dca.py index 9360682..4fc2e8d 100644 --- a/app/apps/api/views/dca.py +++ b/app/apps/api/views/dca.py @@ -2,10 +2,12 @@ from rest_framework import viewsets from rest_framework.decorators import action from rest_framework.response import Response from apps.dca.models import DCAStrategy, DCAEntry +from apps.api.permissions import SHARED_OBJECT_PERMISSIONS from apps.api.serializers import DCAStrategySerializer, DCAEntrySerializer class DCAStrategyViewSet(viewsets.ModelViewSet): + permission_classes = SHARED_OBJECT_PERMISSIONS queryset = DCAStrategy.objects.all() serializer_class = DCAStrategySerializer filterset_fields = { @@ -43,6 +45,8 @@ class DCAStrategyViewSet(viewsets.ModelViewSet): class DCAEntryViewSet(viewsets.ModelViewSet): + permission_classes = SHARED_OBJECT_PERMISSIONS + shared_object_via = "strategy" queryset = DCAEntry.objects.all() serializer_class = DCAEntrySerializer filterset_fields = { diff --git a/app/apps/api/views/transactions.py b/app/apps/api/views/transactions.py index 068ef0d..fdc0003 100644 --- a/app/apps/api/views/transactions.py +++ b/app/apps/api/views/transactions.py @@ -19,6 +19,7 @@ from apps.transactions.models import ( RecurringTransaction, ) from apps.rules.signals import transaction_updated, transaction_created +from apps.api.permissions import SHARED_OBJECT_PERMISSIONS class TransactionViewSet(viewsets.ModelViewSet): @@ -68,6 +69,7 @@ class TransactionViewSet(viewsets.ModelViewSet): class TransactionCategoryViewSet(viewsets.ModelViewSet): + permission_classes = SHARED_OBJECT_PERMISSIONS queryset = TransactionCategory.objects.all() serializer_class = TransactionCategorySerializer filterset_fields = { @@ -85,6 +87,7 @@ class TransactionCategoryViewSet(viewsets.ModelViewSet): class TransactionTagViewSet(viewsets.ModelViewSet): + permission_classes = SHARED_OBJECT_PERMISSIONS queryset = TransactionTag.objects.all() serializer_class = TransactionTagSerializer filterset_fields = { @@ -101,6 +104,7 @@ class TransactionTagViewSet(viewsets.ModelViewSet): class TransactionEntityViewSet(viewsets.ModelViewSet): + permission_classes = SHARED_OBJECT_PERMISSIONS queryset = TransactionEntity.objects.all() serializer_class = TransactionEntitySerializer filterset_fields = { diff --git a/app/apps/common/functions/permissions.py b/app/apps/common/functions/permissions.py new file mode 100644 index 0000000..713c9f9 --- /dev/null +++ b/app/apps/common/functions/permissions.py @@ -0,0 +1,58 @@ +from django.core.exceptions import PermissionDenied +from django.http import Http404 +from django.shortcuts import get_object_or_404 + +READ = "read" +EDIT = "edit" + + +def get_shared_object_or_error(klass, request, *, level=EDIT, via=None, **kwargs): + """Fetch an object like ``get_object_or_404`` while enforcing access control. + + ``SharedObjectManager`` scopes querysets to what a user may *see*, which is + not the same as what they may *change*. Views that resolve an object from a + URL id must state which of the two they need, otherwise a shared or public + object becomes writable by anyone who can see it. + + ``level`` selects the check applied to the governing ``SharedObject``: + + ``READ`` + The object must be visible to the user. Denial raises :class:`Http404` + so the response does not confirm that the id exists. + ``EDIT`` + The object must be owned by the user. Denial raises + :class:`~django.core.exceptions.PermissionDenied` (HTTP 403), which the + frontend surfaces as an "Access Denied" dialog. An object the user + cannot even see raises :class:`Http404` instead, so 403 never confirms + the existence of an object they were not allowed to know about. + + Objects with no owner stay accessible to everyone, preserving the existing + behaviour for legacy/unowned objects. + + ``via`` is a dotted path to the ``SharedObject`` that governs access, for + models owned through a relation, e.g. a rule action governed by its parent + rule:: + + get_shared_object_or_error( + TransactionRuleAction, request, id=pk, level=EDIT, via="rule" + ) + + The path is resolved with a plain ``getattr``, so a path that does not + resolve raises ``AttributeError`` rather than silently granting access. + """ + obj = get_object_or_404(klass, **kwargs) + + guard = obj + for attr in via.split(".") if via else []: + guard = getattr(guard, attr) + + if level not in (READ, EDIT): + raise ValueError(f"Unknown access level: {level!r}") + + if not guard.is_visible_to(request.user): + raise Http404 + + if level == EDIT and not guard.is_editable_by(request.user): + raise PermissionDenied + + return obj diff --git a/app/apps/common/models.py b/app/apps/common/models.py index 10f79c8..f9a876f 100644 --- a/app/apps/common/models.py +++ b/app/apps/common/models.py @@ -58,13 +58,35 @@ class SharedObject(models.Model): models.Index(fields=["visibility"]), ] - def is_accessible_by(self, user): - """Check if a user can access this object""" - return ( - self.visibility == "public" - or self.owner == user - or (self.visibility == "shared" and user in self.shared_with.all()) - ) + # NOTE: these two predicates must stay in sync with the ``Q`` objects built + # by ``SharedObjectManager.get_queryset`` above. The manager filters at the + # queryset level and these check a single instance, so they cannot share an + # implementation; ``SharedObjectPredicateParityTests`` asserts they agree. + def is_visible_to(self, user): + """Whether ``user`` may read this object. + + Mirrors ``SharedObjectManager``: public objects, objects with no owner, + the owner's own objects, and objects explicitly shared with the user. + """ + if self.owner is None or self.visibility == "public": + return True + + if not user or not user.is_authenticated: + return False + + return self.owner_id == user.pk or self.shared_with.filter(pk=user.pk).exists() + + def is_editable_by(self, user): + """Whether ``user`` may mutate this object. + + Sharing grants read access only; mutation stays with the owner. Objects + with no owner remain editable by everyone, preserving the behaviour of + legacy/unowned objects. + """ + if self.owner is None: + return True + + return bool(user and user.is_authenticated and self.owner_id == user.pk) def save(self, *args, **kwargs): if not self.pk and not self.owner: diff --git a/app/apps/common/tests/test_permissions.py b/app/apps/common/tests/test_permissions.py new file mode 100644 index 0000000..a803bed --- /dev/null +++ b/app/apps/common/tests/test_permissions.py @@ -0,0 +1,269 @@ +from django.contrib.auth import get_user_model +from django.contrib.auth.models import AnonymousUser +from django.core.exceptions import PermissionDenied +from django.http import Http404 +from django.test import RequestFactory, TestCase, override_settings + +from apps.common.functions.permissions import ( + EDIT, + READ, + get_shared_object_or_error, +) +from apps.common.middleware.thread_local import delete_current_user, write_current_user +from apps.rules.models import TransactionRule, TransactionRuleAction + + +@override_settings( + STORAGES={ + "default": {"BACKEND": "django.core.files.storage.FileSystemStorage"}, + "staticfiles": { + "BACKEND": "django.contrib.staticfiles.storage.StaticFilesStorage" + }, + }, + WHITENOISE_AUTOREFRESH=True, +) +class SharedObjectPredicateTests(TestCase): + """Unit tests for is_visible_to / is_editable_by on SharedObject.""" + + 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" + ) + + def _rule(self, **kwargs): + kwargs.setdefault("name", "Rule") + kwargs.setdefault("trigger", "True") + return TransactionRule.all_objects.create(**kwargs) + + def test_owner_can_read_and_edit_own_private_rule(self): + rule = self._rule(owner=self.owner, visibility="private") + + self.assertTrue(rule.is_visible_to(self.owner)) + self.assertTrue(rule.is_editable_by(self.owner)) + + def test_private_rule_is_invisible_to_stranger(self): + rule = self._rule(owner=self.owner, visibility="private") + + self.assertFalse(rule.is_visible_to(self.stranger)) + self.assertFalse(rule.is_editable_by(self.stranger)) + + def test_shared_rule_is_readable_but_not_editable(self): + rule = self._rule(owner=self.owner, visibility="private") + rule.shared_with.add(self.shared_user) + + self.assertTrue(rule.is_visible_to(self.shared_user)) + self.assertFalse(rule.is_editable_by(self.shared_user)) + + def test_public_rule_is_readable_but_not_editable(self): + rule = self._rule(owner=self.owner, visibility="public") + + self.assertTrue(rule.is_visible_to(self.stranger)) + self.assertFalse(rule.is_editable_by(self.stranger)) + + def test_unowned_rule_stays_readable_and_editable_by_everyone(self): + rule = self._rule(owner=None, visibility="private") + + self.assertTrue(rule.is_visible_to(self.stranger)) + self.assertTrue(rule.is_editable_by(self.stranger)) + + def test_anonymous_user_gets_no_access_to_owned_rules(self): + rule = self._rule(owner=self.owner, visibility="private") + + self.assertFalse(rule.is_visible_to(AnonymousUser())) + self.assertFalse(rule.is_editable_by(AnonymousUser())) + + +@override_settings( + STORAGES={ + "default": {"BACKEND": "django.core.files.storage.FileSystemStorage"}, + "staticfiles": { + "BACKEND": "django.contrib.staticfiles.storage.StaticFilesStorage" + }, + }, + WHITENOISE_AUTOREFRESH=True, +) +class SharedObjectPredicateParityTests(TestCase): + """is_visible_to must agree with what SharedObjectManager returns. + + The manager filters at the queryset level and the predicate checks a single + instance, so the two cannot share an implementation. This asserts they do + not drift apart. + """ + + def setUp(self): + User = get_user_model() + self.owner = User.objects.create_user( + email="owner@test.com", password="testpass123" + ) + self.other = User.objects.create_user( + email="other@test.com", password="testpass123" + ) + self.addCleanup(self._clear_current_user) + + def _clear_current_user(self): + try: + delete_current_user() + except AttributeError: + pass + + def test_manager_and_predicate_agree_over_every_combination(self): + combinations = [] + for owner in (self.owner, self.other, None): + for visibility in ("private", "public"): + for shared in (True, False): + rule = TransactionRule.all_objects.create( + name=f"{owner}-{visibility}-{shared}", + trigger="True", + owner=owner, + visibility=visibility, + ) + if shared: + rule.shared_with.add(self.owner) + combinations.append(rule) + + write_current_user(self.owner) + visible_ids = set(TransactionRule.objects.values_list("id", flat=True)) + + for rule in combinations: + with self.subTest(rule=rule.name): + self.assertEqual( + rule.id in visible_ids, + rule.is_visible_to(self.owner), + f"manager and is_visible_to disagree for {rule.name}", + ) + + +@override_settings( + STORAGES={ + "default": {"BACKEND": "django.core.files.storage.FileSystemStorage"}, + "staticfiles": { + "BACKEND": "django.contrib.staticfiles.storage.StaticFilesStorage" + }, + }, + WHITENOISE_AUTOREFRESH=True, +) +class GetSharedObjectOrErrorTests(TestCase): + def setUp(self): + User = get_user_model() + self.owner = User.objects.create_user( + email="owner@test.com", password="testpass123" + ) + self.stranger = User.objects.create_user( + email="stranger@test.com", password="testpass123" + ) + self.factory = RequestFactory() + + self.public_rule = TransactionRule.all_objects.create( + name="Public", trigger="True", owner=self.owner, visibility="public" + ) + self.private_rule = TransactionRule.all_objects.create( + name="Private", trigger="True", owner=self.owner, visibility="private" + ) + self.public_action = TransactionRuleAction.objects.create( + rule=self.public_rule, field="notes", value="x" + ) + self.private_action = TransactionRuleAction.objects.create( + rule=self.private_rule, field="notes", value="x" + ) + + self.addCleanup(self._clear_current_user) + + def _clear_current_user(self): + try: + delete_current_user() + except AttributeError: + pass + + def _request(self, user): + request = self.factory.get("/") + request.user = user + write_current_user(user) + return request + + def test_read_allows_visible_object(self): + request = self._request(self.stranger) + + rule = get_shared_object_or_error( + TransactionRule, request, id=self.public_rule.id, level=READ + ) + + self.assertEqual(rule, self.public_rule) + + def test_edit_denies_visible_but_unowned_object_with_403(self): + request = self._request(self.stranger) + + with self.assertRaises(PermissionDenied): + get_shared_object_or_error( + TransactionRule, request, id=self.public_rule.id, level=EDIT + ) + + def test_edit_allows_owner(self): + request = self._request(self.owner) + + rule = get_shared_object_or_error( + TransactionRule, request, id=self.public_rule.id, level=EDIT + ) + + self.assertEqual(rule, self.public_rule) + + def test_invisible_object_raises_404_not_403(self): + """403 must not confirm the existence of an object the user cannot see.""" + request = self._request(self.stranger) + + with self.assertRaises(Http404): + get_shared_object_or_error( + TransactionRule, request, id=self.private_rule.id, level=EDIT + ) + + def test_via_traverses_to_the_governing_object(self): + request = self._request(self.stranger) + + with self.assertRaises(PermissionDenied): + get_shared_object_or_error( + TransactionRuleAction, + request, + id=self.public_action.id, + level=EDIT, + via="rule", + ) + + def test_via_hides_children_of_invisible_parents_behind_404(self): + """TransactionRuleAction has an unscoped manager, so the id is reachable.""" + request = self._request(self.stranger) + + with self.assertRaises(Http404): + get_shared_object_or_error( + TransactionRuleAction, + request, + id=self.private_action.id, + level=EDIT, + via="rule", + ) + + def test_unresolvable_via_path_fails_closed(self): + """A typo'd path must raise, never silently grant access.""" + request = self._request(self.stranger) + + with self.assertRaises(AttributeError): + get_shared_object_or_error( + TransactionRuleAction, + request, + id=self.public_action.id, + level=EDIT, + via="rulee", + ) + + def test_unknown_level_is_rejected(self): + request = self._request(self.owner) + + with self.assertRaises(ValueError): + get_shared_object_or_error( + TransactionRule, request, id=self.public_rule.id, level="write" + ) diff --git a/app/apps/common/tests/test_shared_object_deletion.py b/app/apps/common/tests/test_shared_object_deletion.py new file mode 100644 index 0000000..a7bbf70 --- /dev/null +++ b/app/apps/common/tests/test_shared_object_deletion.py @@ -0,0 +1,161 @@ +"""Delete semantics for every SharedObject-backed model. + +Each of these views carried the same inverted condition:: + + if obj.owner != request.user and request.user in obj.shared_with.all(): + obj.shared_with.remove(request.user) # unshare + else: + obj.delete() # <- public objects landed here + +so an object its owner had made public could be destroyed by any authenticated +user. The rule is: the owner deletes, a shared user revokes only their own +access, and nobody else may do either. +""" + +from django.contrib.auth import get_user_model +from django.test import TestCase, override_settings +from django.urls import reverse + +from apps.accounts.models import Account, AccountGroup +from apps.currencies.models import Currency +from apps.dca.models import DCAStrategy +from apps.rules.models import TransactionRule +from apps.transactions.models import ( + TransactionCategory, + TransactionEntity, + TransactionTag, +) + +HTMX = {"HTTP_HX_REQUEST": "true"} + + +@override_settings( + STORAGES={ + "default": {"BACKEND": "django.core.files.storage.FileSystemStorage"}, + "staticfiles": { + "BACKEND": "django.contrib.staticfiles.storage.StaticFilesStorage" + }, + }, + WHITENOISE_AUTOREFRESH=True, + DEMO=False, +) +class SharedObjectDeletionTests(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" + ) + self.currency = Currency.objects.create( + code="USD", name="US Dollar", decimal_places=2 + ) + + def cases(self): + """(label, model, delete url name, url kwarg, extra create kwargs).""" + return [ + ( + "account", + Account, + "account_delete", + "pk", + {"currency": self.currency}, + ), + ("account group", AccountGroup, "account_group_delete", "pk", {}), + ( + "category", + TransactionCategory, + "category_delete", + "category_id", + {}, + ), + ("tag", TransactionTag, "tag_delete", "tag_id", {}), + ("entity", TransactionEntity, "entity_delete", "entity_id", {}), + ( + "rule", + TransactionRule, + "transaction_rule_delete", + "transaction_rule_id", + {"trigger": "True"}, + ), + ( + "dca strategy", + DCAStrategy, + "dca_strategy_delete", + "strategy_id", + { + "target_currency": self.currency, + "payment_currency": self.currency, + }, + ), + ] + + def make(self, model, visibility, extra): + return model.all_objects.create( + name="Target", owner=self.owner, visibility=visibility, **extra + ) + + def delete(self, url_name, kwarg, obj): + return self.client.delete(reverse(url_name, kwargs={kwarg: obj.id}), **HTMX) + + def test_stranger_cannot_delete_a_public_object(self): + for label, model, url_name, kwarg, extra in self.cases(): + with self.subTest(model=label): + obj = self.make(model, "public", extra) + self.client.force_login(self.stranger) + + response = self.delete(url_name, kwarg, obj) + + self.assertEqual(response.status_code, 403, label) + self.assertTrue( + model.all_objects.filter(pk=obj.pk).exists(), + f"{label} was deleted by a non-owner", + ) + + def test_shared_user_deleting_only_revokes_their_own_access(self): + for label, model, url_name, kwarg, extra in self.cases(): + with self.subTest(model=label): + obj = self.make(model, "private", extra) + obj.shared_with.add(self.shared_user) + self.client.force_login(self.shared_user) + + response = self.delete(url_name, kwarg, obj) + + self.assertEqual(response.status_code, 204, label) + self.assertTrue( + model.all_objects.filter(pk=obj.pk).exists(), + f"{label} was deleted by a shared user", + ) + self.assertNotIn(self.shared_user, obj.shared_with.all()) + + def test_owner_can_delete(self): + for label, model, url_name, kwarg, extra in self.cases(): + with self.subTest(model=label): + obj = self.make(model, "private", extra) + self.client.force_login(self.owner) + + response = self.delete(url_name, kwarg, obj) + + self.assertEqual(response.status_code, 204, label) + self.assertFalse( + model.all_objects.filter(pk=obj.pk).exists(), + f"{label} was not deleted by its owner", + ) + + def test_unowned_objects_stay_deletable(self): + """Legacy objects with no owner are editable by everyone by design.""" + for label, model, url_name, kwarg, extra in self.cases(): + with self.subTest(model=label): + obj = model.all_objects.create( + name="Legacy", owner=None, visibility="private", **extra + ) + self.client.force_login(self.stranger) + + response = self.delete(url_name, kwarg, obj) + + self.assertEqual(response.status_code, 204, label) + self.assertFalse(model.all_objects.filter(pk=obj.pk).exists(), label) diff --git a/app/apps/dca/tests.py b/app/apps/dca/tests.py deleted file mode 100644 index 7ce503c..0000000 --- a/app/apps/dca/tests.py +++ /dev/null @@ -1,3 +0,0 @@ -from django.test import TestCase - -# Create your tests here. diff --git a/app/apps/dca/tests/__init__.py b/app/apps/dca/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/apps/dca/tests/test_view_permissions.py b/app/apps/dca/tests/test_view_permissions.py new file mode 100644 index 0000000..83f0d49 --- /dev/null +++ b/app/apps/dca/tests/test_view_permissions.py @@ -0,0 +1,248 @@ +"""Object-level authorization tests for the DCA views. + +DCAEntry has an unscoped default manager, so filtering an entry by +``strategy__id`` alone reached entries belonging to strategies the caller could +not see at all -- a strictly wider hole than the one reported for rules in +GHSA-83g9-vjqf-2j5q, since it needs no public or shared strategy. +""" + +from datetime import date +from decimal import Decimal + +from django.contrib.auth import get_user_model +from django.test import TestCase, override_settings +from django.urls import reverse + +from apps.currencies.models import Currency +from apps.dca.models import DCAEntry, DCAStrategy + +HTMX = {"HTTP_HX_REQUEST": "true"} + + +@override_settings( + STORAGES={ + "default": {"BACKEND": "django.core.files.storage.FileSystemStorage"}, + "staticfiles": { + "BACKEND": "django.contrib.staticfiles.storage.StaticFilesStorage" + }, + }, + WHITENOISE_AUTOREFRESH=True, + DEMO=False, +) +class DCAObjectPermissionTests(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" + ) + self.currency = Currency.objects.create( + code="USD", name="US Dollar", decimal_places=2 + ) + + self.private_strategy = self._strategy("Private", visibility="private") + self.public_strategy = self._strategy("Public", visibility="public") + self.shared_strategy = self._strategy("Shared", visibility="private") + self.shared_strategy.shared_with.add(self.shared_user) + + self.private_entry = self._entry(self.private_strategy) + self.public_entry = self._entry(self.public_strategy) + self.shared_entry = self._entry(self.shared_strategy) + + def _strategy(self, name, visibility): + return DCAStrategy.all_objects.create( + name=name, + owner=self.owner, + visibility=visibility, + target_currency=self.currency, + payment_currency=self.currency, + ) + + def _entry(self, strategy): + return DCAEntry.objects.create( + strategy=strategy, + date=date(2025, 1, 1), + amount_paid=Decimal("100"), + amount_received=Decimal("1"), + ) + + # ------------------------------------------------------------------ + # entries on an invisible strategy: must not even confirm they exist + # ------------------------------------------------------------------ + def test_stranger_cannot_delete_entry_on_private_strategy(self): + self.client.force_login(self.stranger) + + response = self.client.delete( + reverse( + "dca_entry_delete", + kwargs={ + "strategy_id": self.private_strategy.id, + "entry_id": self.private_entry.id, + }, + ), + **HTMX, + ) + + self.assertEqual(response.status_code, 404) + self.assertTrue(DCAEntry.objects.filter(pk=self.private_entry.pk).exists()) + + def test_stranger_cannot_open_entry_edit_form_on_private_strategy(self): + self.client.force_login(self.stranger) + + response = self.client.get( + reverse( + "dca_entry_edit", + kwargs={ + "strategy_id": self.private_strategy.id, + "entry_id": self.private_entry.id, + }, + ), + **HTMX, + ) + + self.assertEqual(response.status_code, 404) + + def test_stranger_cannot_edit_entry_on_private_strategy(self): + self.client.force_login(self.stranger) + + response = self.client.post( + reverse( + "dca_entry_edit", + kwargs={ + "strategy_id": self.private_strategy.id, + "entry_id": self.private_entry.id, + }, + ), + data={ + "date": "2030-01-01", + "amount_paid": "999", + "amount_received": "999", + }, + **HTMX, + ) + + self.assertEqual(response.status_code, 404) + self.private_entry.refresh_from_db() + self.assertEqual(self.private_entry.amount_paid, Decimal("100")) + + # ------------------------------------------------------------------ + # entries on a visible-but-unowned strategy: 403, not 404 + # ------------------------------------------------------------------ + def test_stranger_cannot_delete_entry_on_public_strategy(self): + self.client.force_login(self.stranger) + + response = self.client.delete( + reverse( + "dca_entry_delete", + kwargs={ + "strategy_id": self.public_strategy.id, + "entry_id": self.public_entry.id, + }, + ), + **HTMX, + ) + + self.assertEqual(response.status_code, 403) + self.assertTrue(DCAEntry.objects.filter(pk=self.public_entry.pk).exists()) + + def test_shared_user_cannot_add_entry_to_shared_strategy(self): + self.client.force_login(self.shared_user) + + response = self.client.post( + reverse("dca_entry_add", kwargs={"strategy_id": self.shared_strategy.id}), + data={ + "date": "2030-01-01", + "amount_paid": "5", + "amount_received": "5", + }, + **HTMX, + ) + + self.assertEqual(response.status_code, 403) + self.assertEqual(self.shared_strategy.entries.count(), 1) + + def test_owner_can_still_manage_own_entries(self): + self.client.force_login(self.owner) + + response = self.client.delete( + reverse( + "dca_entry_delete", + kwargs={ + "strategy_id": self.private_strategy.id, + "entry_id": self.private_entry.id, + }, + ), + **HTMX, + ) + + self.assertEqual(response.status_code, 204) + self.assertFalse(DCAEntry.objects.filter(pk=self.private_entry.pk).exists()) + + # ------------------------------------------------------------------ + # reads stay open to shared users + # ------------------------------------------------------------------ + def test_shared_user_can_still_view_strategy_detail(self): + self.client.force_login(self.shared_user) + + response = self.client.get( + reverse( + "dca_strategy_detail", kwargs={"strategy_id": self.shared_strategy.id} + ), + **HTMX, + ) + + self.assertEqual(response.status_code, 200) + + # ------------------------------------------------------------------ + # strategy delete: the same inversion the rules views had + # ------------------------------------------------------------------ + def test_stranger_cannot_delete_public_strategy(self): + self.client.force_login(self.stranger) + + response = self.client.delete( + reverse( + "dca_strategy_delete", kwargs={"strategy_id": self.public_strategy.id} + ), + **HTMX, + ) + + self.assertEqual(response.status_code, 403) + self.assertTrue( + DCAStrategy.all_objects.filter(pk=self.public_strategy.pk).exists() + ) + + def test_shared_user_deleting_only_revokes_their_own_access(self): + self.client.force_login(self.shared_user) + + response = self.client.delete( + reverse( + "dca_strategy_delete", kwargs={"strategy_id": self.shared_strategy.id} + ), + **HTMX, + ) + + self.assertEqual(response.status_code, 204) + self.assertTrue( + DCAStrategy.all_objects.filter(pk=self.shared_strategy.pk).exists() + ) + self.assertNotIn(self.shared_user, self.shared_strategy.shared_with.all()) + + def test_owner_can_delete_own_strategy(self): + self.client.force_login(self.owner) + + response = self.client.delete( + reverse( + "dca_strategy_delete", kwargs={"strategy_id": self.public_strategy.id} + ), + **HTMX, + ) + + self.assertEqual(response.status_code, 204) + self.assertFalse( + DCAStrategy.all_objects.filter(pk=self.public_strategy.pk).exists() + ) diff --git a/app/apps/dca/views.py b/app/apps/dca/views.py index 62de916..c27979f 100644 --- a/app/apps/dca/views.py +++ b/app/apps/dca/views.py @@ -1,13 +1,19 @@ from django.contrib import messages from django.contrib.auth.decorators import login_required +from django.core.exceptions import PermissionDenied from django.db.models import Sum, Avg from django.db.models.functions import TruncMonth from django.http import HttpResponse -from django.shortcuts import render, get_object_or_404 +from django.shortcuts import render 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.functions.permissions import ( + EDIT, + READ, + get_shared_object_or_error, +) from apps.dca.forms import DCAEntryForm, DCAStrategyForm from apps.dca.models import DCAStrategy, DCAEntry from apps.common.models import SharedObject @@ -56,17 +62,9 @@ def strategy_add(request): @only_htmx @login_required def strategy_edit(request, strategy_id): - dca_strategy = get_object_or_404(DCAStrategy, id=strategy_id) - - if dca_strategy.owner and dca_strategy.owner != request.user: - messages.error(request, _("Only the owner can edit this")) - - return HttpResponse( - status=204, - headers={ - "HX-Trigger": "updated, hide_offcanvas", - }, - ) + dca_strategy = get_shared_object_or_error( + DCAStrategy, request, id=strategy_id, level=EDIT + ) if request.method == "POST": form = DCAStrategyForm(request.POST, instance=dca_strategy) @@ -94,17 +92,20 @@ def strategy_edit(request, strategy_id): @login_required @require_http_methods(["DELETE"]) def strategy_delete(request, strategy_id): - dca_strategy = get_object_or_404(DCAStrategy, id=strategy_id) + dca_strategy = get_shared_object_or_error( + DCAStrategy, request, id=strategy_id, level=READ + ) - if ( - dca_strategy.owner != request.user - and request.user in dca_strategy.shared_with.all() - ): + if dca_strategy.is_editable_by(request.user): + dca_strategy.delete() + messages.success(request, _("DCA strategy deleted successfully")) + elif dca_strategy.shared_with.filter(pk=request.user.pk).exists(): + # Someone else's object shared with us: we can drop our own access + # to it, but never delete it. dca_strategy.shared_with.remove(request.user) messages.success(request, _("Item no longer shared with you")) else: - dca_strategy.delete() - messages.success(request, _("DCA strategy deleted successfully")) + raise PermissionDenied return HttpResponse( status=204, @@ -118,7 +119,9 @@ def strategy_delete(request, strategy_id): @login_required @require_http_methods(["GET"]) def strategy_take_ownership(request, strategy_id): - dca_strategy = get_object_or_404(DCAStrategy, id=strategy_id) + dca_strategy = get_shared_object_or_error( + DCAStrategy, request, id=strategy_id, level=EDIT + ) if not dca_strategy.owner: dca_strategy.owner = request.user @@ -139,17 +142,7 @@ def strategy_take_ownership(request, strategy_id): @login_required @require_http_methods(["GET", "POST"]) def strategy_share(request, pk): - obj = get_object_or_404(DCAStrategy, id=pk) - - if obj.owner and obj.owner != request.user: - messages.error(request, _("Only the owner can edit this")) - - return HttpResponse( - status=204, - headers={ - "HX-Trigger": "updated, hide_offcanvas", - }, - ) + obj = get_shared_object_or_error(DCAStrategy, request, id=pk, level=EDIT) if request.method == "POST": form = SharedObjectForm(request.POST, instance=obj, user=request.user) @@ -175,7 +168,9 @@ def strategy_share(request, pk): @login_required def strategy_detail_index(request, strategy_id): - strategy = get_object_or_404(DCAStrategy, id=strategy_id) + strategy = get_shared_object_or_error( + DCAStrategy, request, id=strategy_id, level=READ + ) return render( request, @@ -187,7 +182,9 @@ def strategy_detail_index(request, strategy_id): @only_htmx @login_required def strategy_detail(request, strategy_id): - strategy = get_object_or_404(DCAStrategy, id=strategy_id) + strategy = get_shared_object_or_error( + DCAStrategy, request, id=strategy_id, level=READ + ) entries = strategy.entries.all() # Calculate monthly aggregates @@ -229,7 +226,9 @@ def strategy_detail(request, strategy_id): @only_htmx @login_required def strategy_entry_add(request, strategy_id): - strategy = get_object_or_404(DCAStrategy, id=strategy_id) + strategy = get_shared_object_or_error( + DCAStrategy, request, id=strategy_id, level=EDIT + ) if request.method == "POST": form = DCAEntryForm(request.POST, strategy=strategy) if form.is_valid(): @@ -255,7 +254,14 @@ def strategy_entry_add(request, strategy_id): @only_htmx @login_required def strategy_entry_edit(request, strategy_id, entry_id): - dca_entry = get_object_or_404(DCAEntry, id=entry_id, strategy__id=strategy_id) + dca_entry = get_shared_object_or_error( + DCAEntry, + request, + id=entry_id, + strategy__id=strategy_id, + level=EDIT, + via="strategy", + ) if request.method == "POST": form = DCAEntryForm(request.POST, instance=dca_entry) @@ -283,7 +289,14 @@ def strategy_entry_edit(request, strategy_id, entry_id): @login_required @require_http_methods(["DELETE"]) def strategy_entry_delete(request, entry_id, strategy_id): - dca_entry = get_object_or_404(DCAEntry, id=entry_id, strategy__id=strategy_id) + dca_entry = get_shared_object_or_error( + DCAEntry, + request, + id=entry_id, + strategy__id=strategy_id, + level=EDIT, + via="strategy", + ) dca_entry.delete() diff --git a/app/apps/rules/tests/test_view_permissions.py b/app/apps/rules/tests/test_view_permissions.py new file mode 100644 index 0000000..9506dd0 --- /dev/null +++ b/app/apps/rules/tests/test_view_permissions.py @@ -0,0 +1,473 @@ +"""Object-level authorization tests for the transaction-rule endpoints. + +Regression coverage for GHSA-83g9-vjqf-2j5q: SharedObjectManager scopes rules to +what a user may *see*, which included other people's public and shared-with-them +rules. Several mutating endpoints treated that visibility as permission to write. +""" + +from django.contrib.auth import get_user_model +from django.test import TestCase, override_settings +from django.urls import reverse + +from apps.rules.models import ( + TransactionRule, + TransactionRuleAction, + UpdateOrCreateTransactionRuleAction, +) + +HTMX = {"HTTP_HX_REQUEST": "true"} + + +@override_settings( + STORAGES={ + "default": {"BACKEND": "django.core.files.storage.FileSystemStorage"}, + "staticfiles": { + "BACKEND": "django.contrib.staticfiles.storage.StaticFilesStorage" + }, + }, + WHITENOISE_AUTOREFRESH=True, + DEMO=False, +) +class TransactionRuleObjectPermissionTests(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" + ) + + # Public: visible to everyone through SharedObjectManager. + self.public_rule = TransactionRule.all_objects.create( + name="Public rule", + trigger="True", + owner=self.owner, + visibility="public", + active=True, + ) + # Private but shared with shared_user: visible to them, not to stranger. + self.shared_rule = TransactionRule.all_objects.create( + name="Shared rule", + trigger="True", + owner=self.owner, + visibility="private", + active=True, + ) + self.shared_rule.shared_with.add(self.shared_user) + # Private and unshared: invisible to everyone but the owner. + self.private_rule = TransactionRule.all_objects.create( + name="Private rule", + trigger="True", + owner=self.owner, + visibility="private", + active=True, + ) + + self.public_action = TransactionRuleAction.objects.create( + rule=self.public_rule, field="notes", value="owned by owner" + ) + self.private_action = TransactionRuleAction.objects.create( + rule=self.private_rule, field="notes", value="owned by owner" + ) + self.public_linked_action = UpdateOrCreateTransactionRuleAction.objects.create( + rule=self.public_rule + ) + self.private_linked_action = UpdateOrCreateTransactionRuleAction.objects.create( + rule=self.private_rule + ) + + def login(self, user): + self.client.force_login(user) + + # ------------------------------------------------------------------ + # toggle-active + # ------------------------------------------------------------------ + def test_stranger_cannot_toggle_public_rule(self): + self.login(self.stranger) + + response = self.client.get( + reverse( + "transaction_rule_toggle_activity", + kwargs={"transaction_rule_id": self.public_rule.id}, + ), + **HTMX, + ) + + self.assertEqual(response.status_code, 403) + self.public_rule.refresh_from_db() + self.assertTrue(self.public_rule.active) + + def test_shared_user_cannot_toggle_shared_rule(self): + self.login(self.shared_user) + + response = self.client.get( + reverse( + "transaction_rule_toggle_activity", + kwargs={"transaction_rule_id": self.shared_rule.id}, + ), + **HTMX, + ) + + self.assertEqual(response.status_code, 403) + self.shared_rule.refresh_from_db() + self.assertTrue(self.shared_rule.active) + + def test_owner_can_toggle_own_rule(self): + self.login(self.owner) + + response = self.client.get( + reverse( + "transaction_rule_toggle_activity", + kwargs={"transaction_rule_id": self.public_rule.id}, + ), + **HTMX, + ) + + self.assertEqual(response.status_code, 204) + self.public_rule.refresh_from_db() + self.assertFalse(self.public_rule.active) + + # ------------------------------------------------------------------ + # rule actions + # ------------------------------------------------------------------ + def test_stranger_cannot_add_action_to_public_rule(self): + self.login(self.stranger) + + response = self.client.post( + reverse( + "transaction_rule_action_add", + kwargs={"transaction_rule_id": self.public_rule.id}, + ), + data={"field": "notes", "value": "injected", "order": 0}, + **HTMX, + ) + + self.assertEqual(response.status_code, 403) + self.assertFalse( + TransactionRuleAction.objects.filter(value="injected").exists() + ) + + def test_stranger_cannot_edit_action_on_public_rule(self): + self.login(self.stranger) + + response = self.client.post( + reverse( + "transaction_rule_action_edit", + kwargs={"transaction_rule_action_id": self.public_action.id}, + ), + data={"field": "notes", "value": "rewritten", "order": 0}, + **HTMX, + ) + + self.assertEqual(response.status_code, 403) + self.public_action.refresh_from_db() + self.assertEqual(self.public_action.value, "owned by owner") + + def test_stranger_cannot_delete_action_on_public_rule(self): + self.login(self.stranger) + + response = self.client.delete( + reverse( + "transaction_rule_action_delete", + kwargs={"transaction_rule_action_id": self.public_action.id}, + ), + **HTMX, + ) + + self.assertEqual(response.status_code, 403) + self.assertTrue( + TransactionRuleAction.objects.filter(pk=self.public_action.pk).exists() + ) + + def test_stranger_cannot_delete_action_on_invisible_rule(self): + """The child manager is unscoped, so the id is reachable by guessing. + + The parent rule is invisible to the stranger, so the response must be a + 404 rather than a 403 that confirms the action exists. + """ + self.login(self.stranger) + + response = self.client.delete( + reverse( + "transaction_rule_action_delete", + kwargs={"transaction_rule_action_id": self.private_action.id}, + ), + **HTMX, + ) + + self.assertEqual(response.status_code, 404) + self.assertTrue( + TransactionRuleAction.objects.filter(pk=self.private_action.pk).exists() + ) + + def test_owner_can_delete_own_action(self): + self.login(self.owner) + + response = self.client.delete( + reverse( + "transaction_rule_action_delete", + kwargs={"transaction_rule_action_id": self.public_action.id}, + ), + **HTMX, + ) + + self.assertEqual(response.status_code, 204) + self.assertFalse( + TransactionRuleAction.objects.filter(pk=self.public_action.pk).exists() + ) + + # ------------------------------------------------------------------ + # update-or-create rule actions + # ------------------------------------------------------------------ + def test_stranger_cannot_add_linked_action_to_public_rule(self): + self.login(self.stranger) + + response = self.client.post( + reverse( + "update_or_create_transaction_rule_action_add", + kwargs={"transaction_rule_id": self.public_rule.id}, + ), + data={}, + **HTMX, + ) + + self.assertEqual(response.status_code, 403) + self.assertEqual( + UpdateOrCreateTransactionRuleAction.objects.filter( + rule=self.public_rule + ).count(), + 1, + ) + + def test_stranger_cannot_edit_linked_action_on_public_rule(self): + self.login(self.stranger) + + response = self.client.post( + reverse( + "update_or_create_transaction_rule_action_edit", + kwargs={"pk": self.public_linked_action.id}, + ), + data={"filter": "injected"}, + **HTMX, + ) + + self.assertEqual(response.status_code, 403) + self.public_linked_action.refresh_from_db() + self.assertEqual(self.public_linked_action.filter, "") + + def test_stranger_cannot_delete_linked_action_on_public_rule(self): + self.login(self.stranger) + + response = self.client.delete( + reverse( + "update_or_create_transaction_rule_action_delete", + kwargs={"pk": self.public_linked_action.id}, + ), + **HTMX, + ) + + self.assertEqual(response.status_code, 403) + self.assertTrue( + UpdateOrCreateTransactionRuleAction.objects.filter( + pk=self.public_linked_action.pk + ).exists() + ) + + def test_stranger_cannot_delete_linked_action_on_invisible_rule(self): + self.login(self.stranger) + + response = self.client.delete( + reverse( + "update_or_create_transaction_rule_action_delete", + kwargs={"pk": self.private_linked_action.id}, + ), + **HTMX, + ) + + self.assertEqual(response.status_code, 404) + self.assertTrue( + UpdateOrCreateTransactionRuleAction.objects.filter( + pk=self.private_linked_action.pk + ).exists() + ) + + # ------------------------------------------------------------------ + # edit / share / dry-run + # ------------------------------------------------------------------ + def test_stranger_cannot_edit_public_rule(self): + self.login(self.stranger) + + response = self.client.post( + reverse( + "transaction_rule_edit", + kwargs={"transaction_rule_id": self.public_rule.id}, + ), + data={"name": "hijacked", "trigger": "True", "order": 0}, + **HTMX, + ) + + self.assertEqual(response.status_code, 403) + self.public_rule.refresh_from_db() + self.assertEqual(self.public_rule.name, "Public rule") + + def test_stranger_cannot_change_sharing_of_public_rule(self): + self.login(self.stranger) + + response = self.client.post( + reverse( + "transaction_rule_share_settings", kwargs={"pk": self.public_rule.id} + ), + data={"visibility": "private"}, + **HTMX, + ) + + self.assertEqual(response.status_code, 403) + self.public_rule.refresh_from_db() + self.assertEqual(self.public_rule.visibility, "public") + + def test_stranger_cannot_dry_run_public_rule(self): + self.login(self.stranger) + + response = self.client.get( + reverse( + "transaction_rule_dry_run_created", kwargs={"pk": self.public_rule.id} + ), + **HTMX, + ) + + self.assertEqual(response.status_code, 403) + + # ------------------------------------------------------------------ + # delete: sharing may be revoked, ownership may not be overridden + # ------------------------------------------------------------------ + def test_stranger_cannot_delete_public_rule(self): + """The original condition fell through to delete() for public rules.""" + self.login(self.stranger) + + response = self.client.delete( + reverse( + "transaction_rule_delete", + kwargs={"transaction_rule_id": self.public_rule.id}, + ), + **HTMX, + ) + + self.assertEqual(response.status_code, 403) + self.assertTrue( + TransactionRule.all_objects.filter(pk=self.public_rule.pk).exists() + ) + + def test_shared_user_deleting_only_revokes_their_own_access(self): + self.login(self.shared_user) + + response = self.client.delete( + reverse( + "transaction_rule_delete", + kwargs={"transaction_rule_id": self.shared_rule.id}, + ), + **HTMX, + ) + + self.assertEqual(response.status_code, 204) + self.assertTrue( + TransactionRule.all_objects.filter(pk=self.shared_rule.pk).exists() + ) + self.assertNotIn(self.shared_user, self.shared_rule.shared_with.all()) + + def test_owner_can_delete_own_rule(self): + self.login(self.owner) + + response = self.client.delete( + reverse( + "transaction_rule_delete", + kwargs={"transaction_rule_id": self.public_rule.id}, + ), + **HTMX, + ) + + self.assertEqual(response.status_code, 204) + self.assertFalse( + TransactionRule.all_objects.filter(pk=self.public_rule.pk).exists() + ) + + # ------------------------------------------------------------------ + # reads must keep working for shared users + # ------------------------------------------------------------------ + def test_shared_user_can_still_view_shared_rule(self): + self.login(self.shared_user) + + response = self.client.get( + reverse( + "transaction_rule_view", + kwargs={"transaction_rule_id": self.shared_rule.id}, + ), + **HTMX, + ) + + self.assertEqual(response.status_code, 200) + + def test_stranger_can_view_public_rule(self): + self.login(self.stranger) + + response = self.client.get( + reverse( + "transaction_rule_view", + kwargs={"transaction_rule_id": self.public_rule.id}, + ), + **HTMX, + ) + + self.assertEqual(response.status_code, 200) + + def test_stranger_cannot_view_private_rule(self): + self.login(self.stranger) + + response = self.client.get( + reverse( + "transaction_rule_view", + kwargs={"transaction_rule_id": self.private_rule.id}, + ), + **HTMX, + ) + + self.assertEqual(response.status_code, 404) + + # ------------------------------------------------------------------ + # take ownership stays available for unowned rules only + # ------------------------------------------------------------------ + def test_take_ownership_of_unowned_rule_still_works(self): + unowned = TransactionRule.all_objects.create( + name="Legacy rule", trigger="True", owner=None, visibility="private" + ) + self.login(self.stranger) + + response = self.client.get( + reverse( + "transaction_rule_take_ownership", + kwargs={"transaction_rule_id": unowned.id}, + ), + **HTMX, + ) + + self.assertEqual(response.status_code, 204) + unowned.refresh_from_db() + self.assertEqual(unowned.owner, self.stranger) + + def test_cannot_take_ownership_of_someone_elses_public_rule(self): + self.login(self.stranger) + + response = self.client.get( + reverse( + "transaction_rule_take_ownership", + kwargs={"transaction_rule_id": self.public_rule.id}, + ), + **HTMX, + ) + + self.assertEqual(response.status_code, 403) + self.public_rule.refresh_from_db() + self.assertEqual(self.public_rule.owner, self.owner) diff --git a/app/apps/rules/views.py b/app/apps/rules/views.py index 1fb44a6..5054c02 100644 --- a/app/apps/rules/views.py +++ b/app/apps/rules/views.py @@ -4,13 +4,19 @@ from copy import deepcopy from django.contrib import messages from django.contrib.auth.decorators import login_required +from django.core.exceptions import PermissionDenied from django.db import transaction from django.http import HttpResponse -from django.shortcuts import render, get_object_or_404, redirect +from django.shortcuts import render, redirect 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.functions.permissions import ( + EDIT, + READ, + get_shared_object_or_error, +) from apps.rules.forms import ( TransactionRuleForm, TransactionRuleActionForm, @@ -62,7 +68,9 @@ def rules_list(request): @disabled_on_demo @require_http_methods(["GET", "POST"]) def transaction_rule_toggle_activity(request, transaction_rule_id, **kwargs): - transaction_rule = get_object_or_404(TransactionRule, id=transaction_rule_id) + transaction_rule = get_shared_object_or_error( + TransactionRule, request, id=transaction_rule_id, level=EDIT + ) current_active = transaction_rule.active transaction_rule.active = not current_active transaction_rule.save(update_fields=["active"]) @@ -112,17 +120,9 @@ def transaction_rule_add(request, **kwargs): @disabled_on_demo @require_http_methods(["GET", "POST"]) def transaction_rule_edit(request, transaction_rule_id): - transaction_rule = get_object_or_404(TransactionRule, id=transaction_rule_id) - - if transaction_rule.owner and transaction_rule.owner != request.user: - messages.error(request, _("Only the owner can edit this")) - - return HttpResponse( - status=204, - headers={ - "HX-Trigger": "updated, hide_offcanvas", - }, - ) + transaction_rule = get_shared_object_or_error( + TransactionRule, request, id=transaction_rule_id, level=EDIT + ) if request.method == "POST": form = TransactionRuleForm(request.POST, instance=transaction_rule) @@ -151,7 +151,9 @@ def transaction_rule_edit(request, transaction_rule_id): @disabled_on_demo @require_http_methods(["GET", "POST"]) def transaction_rule_view(request, transaction_rule_id): - transaction_rule = get_object_or_404(TransactionRule, id=transaction_rule_id) + transaction_rule = get_shared_object_or_error( + TransactionRule, request, id=transaction_rule_id, level=READ + ) edit_actions = transaction_rule.transaction_actions.all() update_or_create_actions = ( @@ -175,17 +177,20 @@ def transaction_rule_view(request, transaction_rule_id): @disabled_on_demo @require_http_methods(["DELETE"]) def transaction_rule_delete(request, transaction_rule_id): - transaction_rule = get_object_or_404(TransactionRule, id=transaction_rule_id) + transaction_rule = get_shared_object_or_error( + TransactionRule, request, id=transaction_rule_id, level=READ + ) - if ( - transaction_rule.owner != request.user - and request.user in transaction_rule.shared_with.all() - ): + if transaction_rule.is_editable_by(request.user): + transaction_rule.delete() + messages.success(request, _("Rule deleted successfully")) + elif transaction_rule.shared_with.filter(pk=request.user.pk).exists(): + # Someone else's rule shared with us: we can drop our own access to it, + # but never delete it. transaction_rule.shared_with.remove(request.user) messages.success(request, _("Item no longer shared with you")) else: - transaction_rule.delete() - messages.success(request, _("Rule deleted successfully")) + raise PermissionDenied return HttpResponse( status=204, @@ -200,7 +205,9 @@ def transaction_rule_delete(request, transaction_rule_id): @disabled_on_demo @require_http_methods(["GET"]) def transaction_rule_take_ownership(request, transaction_rule_id): - transaction_rule = get_object_or_404(TransactionRule, id=transaction_rule_id) + transaction_rule = get_shared_object_or_error( + TransactionRule, request, id=transaction_rule_id, level=EDIT + ) if not transaction_rule.owner: transaction_rule.owner = request.user @@ -222,17 +229,7 @@ def transaction_rule_take_ownership(request, transaction_rule_id): @disabled_on_demo @require_http_methods(["GET", "POST"]) def transaction_rule_share(request, pk): - obj = get_object_or_404(TransactionRule, id=pk) - - if obj.owner and obj.owner != request.user: - messages.error(request, _("Only the owner can edit this")) - - return HttpResponse( - status=204, - headers={ - "HX-Trigger": "updated, hide_offcanvas", - }, - ) + obj = get_shared_object_or_error(TransactionRule, request, id=pk, level=EDIT) if request.method == "POST": form = SharedObjectForm(request.POST, instance=obj, user=request.user) @@ -261,7 +258,9 @@ def transaction_rule_share(request, pk): @disabled_on_demo @require_http_methods(["GET", "POST"]) def transaction_rule_action_add(request, transaction_rule_id): - transaction_rule = get_object_or_404(TransactionRule, id=transaction_rule_id) + transaction_rule = get_shared_object_or_error( + TransactionRule, request, id=transaction_rule_id, level=EDIT + ) if request.method == "POST": form = TransactionRuleActionForm(request.POST, rule=transaction_rule) @@ -289,12 +288,14 @@ def transaction_rule_action_add(request, transaction_rule_id): @disabled_on_demo @require_http_methods(["GET", "POST"]) def transaction_rule_action_edit(request, transaction_rule_action_id): - transaction_rule_action = get_object_or_404( - TransactionRuleAction, id=transaction_rule_action_id - ) - transaction_rule = get_object_or_404( - TransactionRule, id=transaction_rule_action.rule.id + transaction_rule_action = get_shared_object_or_error( + TransactionRuleAction, + request, + id=transaction_rule_action_id, + level=EDIT, + via="rule", ) + transaction_rule = transaction_rule_action.rule if request.method == "POST": form = TransactionRuleActionForm( @@ -327,8 +328,12 @@ def transaction_rule_action_edit(request, transaction_rule_action_id): @disabled_on_demo @require_http_methods(["DELETE"]) def transaction_rule_action_delete(request, transaction_rule_action_id): - transaction_rule_action = get_object_or_404( - TransactionRuleAction, id=transaction_rule_action_id + transaction_rule_action = get_shared_object_or_error( + TransactionRuleAction, + request, + id=transaction_rule_action_id, + level=EDIT, + via="rule", ) transaction_rule_action.delete() @@ -348,7 +353,9 @@ def transaction_rule_action_delete(request, transaction_rule_action_id): @disabled_on_demo @require_http_methods(["GET", "POST"]) def update_or_create_transaction_rule_action_add(request, transaction_rule_id): - transaction_rule = get_object_or_404(TransactionRule, id=transaction_rule_id) + transaction_rule = get_shared_object_or_error( + TransactionRule, request, id=transaction_rule_id, level=EDIT + ) if request.method == "POST": form = UpdateOrCreateTransactionRuleActionForm( @@ -380,7 +387,9 @@ def update_or_create_transaction_rule_action_add(request, transaction_rule_id): @disabled_on_demo @require_http_methods(["GET", "POST"]) def update_or_create_transaction_rule_action_edit(request, pk): - linked_action = get_object_or_404(UpdateOrCreateTransactionRuleAction, id=pk) + linked_action = get_shared_object_or_error( + UpdateOrCreateTransactionRuleAction, request, id=pk, level=EDIT, via="rule" + ) transaction_rule = linked_action.rule if request.method == "POST": @@ -415,7 +424,9 @@ def update_or_create_transaction_rule_action_edit(request, pk): @disabled_on_demo @require_http_methods(["DELETE"]) def update_or_create_transaction_rule_action_delete(request, pk): - linked_action = get_object_or_404(UpdateOrCreateTransactionRuleAction, id=pk) + linked_action = get_shared_object_or_error( + UpdateOrCreateTransactionRuleAction, request, id=pk, level=EDIT, via="rule" + ) linked_action.delete() @@ -436,7 +447,7 @@ def update_or_create_transaction_rule_action_delete(request, pk): @disabled_on_demo @require_http_methods(["GET", "POST"]) def dry_run_rule_created(request, pk): - rule = get_object_or_404(TransactionRule, id=pk) + rule = get_shared_object_or_error(TransactionRule, request, id=pk, level=EDIT) logs = None results = None @@ -481,7 +492,7 @@ def dry_run_rule_created(request, pk): @disabled_on_demo @require_http_methods(["GET", "POST"]) def dry_run_rule_deleted(request, pk): - rule = get_object_or_404(TransactionRule, id=pk) + rule = get_shared_object_or_error(TransactionRule, request, id=pk, level=EDIT) logs = None results = None @@ -526,7 +537,7 @@ def dry_run_rule_deleted(request, pk): @disabled_on_demo @require_http_methods(["GET", "POST"]) def dry_run_rule_updated(request, pk): - rule = get_object_or_404(TransactionRule, id=pk) + rule = get_shared_object_or_error(TransactionRule, request, id=pk, level=EDIT) logs = None results = None diff --git a/app/apps/transactions/views/categories.py b/app/apps/transactions/views/categories.py index 43bd5ff..c539696 100644 --- a/app/apps/transactions/views/categories.py +++ b/app/apps/transactions/views/categories.py @@ -1,11 +1,17 @@ from django.contrib import messages from django.contrib.auth.decorators import login_required +from django.core.exceptions import PermissionDenied from django.http import HttpResponse -from django.shortcuts import render, get_object_or_404 +from django.shortcuts import render 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.functions.permissions import ( + EDIT, + READ, + get_shared_object_or_error, +) from apps.transactions.forms import TransactionCategoryForm from apps.transactions.models import TransactionCategory from apps.common.models import SharedObject @@ -85,17 +91,9 @@ def category_add(request, **kwargs): @login_required @require_http_methods(["GET", "POST"]) def category_edit(request, category_id): - category = get_object_or_404(TransactionCategory, id=category_id) - - if category.owner and category.owner != request.user: - messages.error(request, _("Only the owner can edit this")) - - return HttpResponse( - status=204, - headers={ - "HX-Trigger": "updated, hide_offcanvas", - }, - ) + category = get_shared_object_or_error( + TransactionCategory, request, id=category_id, level=EDIT + ) if request.method == "POST": form = TransactionCategoryForm(request.POST, instance=category) @@ -123,17 +121,7 @@ def category_edit(request, category_id): @login_required @require_http_methods(["GET", "POST"]) def category_share(request, pk): - obj = get_object_or_404(TransactionCategory, id=pk) - - if obj.owner and obj.owner != request.user: - messages.error(request, _("Only the owner can edit this")) - - return HttpResponse( - status=204, - headers={ - "HX-Trigger": "updated, hide_offcanvas", - }, - ) + obj = get_shared_object_or_error(TransactionCategory, request, id=pk, level=EDIT) if request.method == "POST": form = SharedObjectForm(request.POST, instance=obj, user=request.user) @@ -161,14 +149,20 @@ def category_share(request, pk): @login_required @require_http_methods(["DELETE"]) def category_delete(request, category_id): - category = get_object_or_404(TransactionCategory, id=category_id) + category = get_shared_object_or_error( + TransactionCategory, request, id=category_id, level=READ + ) - if category.owner != request.user and request.user in category.shared_with.all(): + if category.is_editable_by(request.user): + category.delete() + messages.success(request, _("Category deleted successfully")) + elif category.shared_with.filter(pk=request.user.pk).exists(): + # Someone else's object shared with us: we can drop our own access + # to it, but never delete it. category.shared_with.remove(request.user) messages.success(request, _("Item no longer shared with you")) else: - category.delete() - messages.success(request, _("Category deleted successfully")) + raise PermissionDenied return HttpResponse( status=204, @@ -182,7 +176,9 @@ def category_delete(request, category_id): @login_required @require_http_methods(["GET"]) def category_take_ownership(request, category_id): - category = get_object_or_404(TransactionCategory, id=category_id) + category = get_shared_object_or_error( + TransactionCategory, request, id=category_id, level=EDIT + ) if not category.owner: category.owner = request.user diff --git a/app/apps/transactions/views/entities.py b/app/apps/transactions/views/entities.py index 300d3e2..8500e61 100644 --- a/app/apps/transactions/views/entities.py +++ b/app/apps/transactions/views/entities.py @@ -1,11 +1,17 @@ from django.contrib import messages from django.contrib.auth.decorators import login_required +from django.core.exceptions import PermissionDenied from django.http import HttpResponse -from django.shortcuts import render, get_object_or_404 +from django.shortcuts import render 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.functions.permissions import ( + EDIT, + READ, + get_shared_object_or_error, +) from apps.transactions.forms import TransactionEntityForm from apps.transactions.models import TransactionEntity from apps.common.models import SharedObject @@ -85,17 +91,9 @@ def entity_add(request, **kwargs): @login_required @require_http_methods(["GET", "POST"]) def entity_edit(request, entity_id): - entity = get_object_or_404(TransactionEntity, id=entity_id) - - if entity.owner and entity.owner != request.user: - messages.error(request, _("Only the owner can edit this")) - - return HttpResponse( - status=204, - headers={ - "HX-Trigger": "updated, hide_offcanvas", - }, - ) + entity = get_shared_object_or_error( + TransactionEntity, request, id=entity_id, level=EDIT + ) if request.method == "POST": form = TransactionEntityForm(request.POST, instance=entity) @@ -123,14 +121,20 @@ def entity_edit(request, entity_id): @login_required @require_http_methods(["DELETE"]) def entity_delete(request, entity_id): - entity = get_object_or_404(TransactionEntity, id=entity_id) + entity = get_shared_object_or_error( + TransactionEntity, request, id=entity_id, level=READ + ) - if entity.owner != request.user and request.user in entity.shared_with.all(): + if entity.is_editable_by(request.user): + entity.delete() + messages.success(request, _("Entity deleted successfully")) + elif entity.shared_with.filter(pk=request.user.pk).exists(): + # Someone else's object shared with us: we can drop our own access + # to it, but never delete it. entity.shared_with.remove(request.user) messages.success(request, _("Item no longer shared with you")) else: - entity.delete() - messages.success(request, _("Entity deleted successfully")) + raise PermissionDenied return HttpResponse( status=204, @@ -144,7 +148,9 @@ def entity_delete(request, entity_id): @login_required @require_http_methods(["GET"]) def entity_take_ownership(request, entity_id): - entity = get_object_or_404(TransactionEntity, id=entity_id) + entity = get_shared_object_or_error( + TransactionEntity, request, id=entity_id, level=EDIT + ) if not entity.owner: entity.owner = request.user @@ -165,17 +171,7 @@ def entity_take_ownership(request, entity_id): @login_required @require_http_methods(["GET", "POST"]) def entity_share(request, pk): - obj = get_object_or_404(TransactionEntity, id=pk) - - if obj.owner and obj.owner != request.user: - messages.error(request, _("Only the owner can edit this")) - - return HttpResponse( - status=204, - headers={ - "HX-Trigger": "updated, hide_offcanvas", - }, - ) + obj = get_shared_object_or_error(TransactionEntity, request, id=pk, level=EDIT) if request.method == "POST": form = SharedObjectForm(request.POST, instance=obj, user=request.user) diff --git a/app/apps/transactions/views/tags.py b/app/apps/transactions/views/tags.py index f7709c9..cc0e487 100644 --- a/app/apps/transactions/views/tags.py +++ b/app/apps/transactions/views/tags.py @@ -1,11 +1,17 @@ from django.contrib import messages from django.contrib.auth.decorators import login_required +from django.core.exceptions import PermissionDenied from django.http import HttpResponse -from django.shortcuts import render, get_object_or_404 +from django.shortcuts import render 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.functions.permissions import ( + EDIT, + READ, + get_shared_object_or_error, +) from apps.transactions.forms import TransactionTagForm from apps.transactions.models import TransactionTag from apps.common.models import SharedObject @@ -85,17 +91,7 @@ def tag_add(request, **kwargs): @login_required @require_http_methods(["GET", "POST"]) def tag_edit(request, tag_id): - tag = get_object_or_404(TransactionTag, id=tag_id) - - if tag.owner and tag.owner != request.user: - messages.error(request, _("Only the owner can edit this")) - - return HttpResponse( - status=204, - headers={ - "HX-Trigger": "updated, hide_offcanvas", - }, - ) + tag = get_shared_object_or_error(TransactionTag, request, id=tag_id, level=EDIT) if request.method == "POST": form = TransactionTagForm(request.POST, instance=tag) @@ -123,14 +119,18 @@ def tag_edit(request, tag_id): @login_required @require_http_methods(["DELETE"]) def tag_delete(request, tag_id): - tag = get_object_or_404(TransactionTag, id=tag_id) + tag = get_shared_object_or_error(TransactionTag, request, id=tag_id, level=READ) - if tag.owner != request.user and request.user in tag.shared_with.all(): + if tag.is_editable_by(request.user): + tag.delete() + messages.success(request, _("Tag deleted successfully")) + elif tag.shared_with.filter(pk=request.user.pk).exists(): + # Someone else's object shared with us: we can drop our own access + # to it, but never delete it. tag.shared_with.remove(request.user) messages.success(request, _("Item no longer shared with you")) else: - tag.delete() - messages.success(request, _("Tag deleted successfully")) + raise PermissionDenied return HttpResponse( status=204, @@ -144,7 +144,7 @@ def tag_delete(request, tag_id): @login_required @require_http_methods(["GET"]) def tag_take_ownership(request, tag_id): - tag = get_object_or_404(TransactionTag, id=tag_id) + tag = get_shared_object_or_error(TransactionTag, request, id=tag_id, level=EDIT) if not tag.owner: tag.owner = request.user @@ -165,17 +165,7 @@ def tag_take_ownership(request, tag_id): @login_required @require_http_methods(["GET", "POST"]) def tag_share(request, pk): - obj = get_object_or_404(TransactionTag, id=pk) - - if obj.owner and obj.owner != request.user: - messages.error(request, _("Only the owner can edit this")) - - return HttpResponse( - status=204, - headers={ - "HX-Trigger": "updated, hide_offcanvas", - }, - ) + obj = get_shared_object_or_error(TransactionTag, request, id=pk, level=EDIT) if request.method == "POST": form = SharedObjectForm(request.POST, instance=obj, user=request.user) diff --git a/app/templates/rules/fragments/list.html b/app/templates/rules/fragments/list.html index 07cb0f7..998708e 100644 --- a/app/templates/rules/fragments/list.html +++ b/app/templates/rules/fragments/list.html @@ -64,14 +64,21 @@