From e2f26b3629cd742779fb35c8c536350ee24f0039 Mon Sep 17 00:00:00 2001 From: Herculino Trotta Date: Tue, 1 Sep 2026 23:15:41 -0300 Subject: [PATCH] fix(sharing): enforce object-level authorization outside the rules app The rules endpoints were one instance of a pattern repeated across every SharedObject-backed app. Route the rest through the same helper. DCA entries were the worst case and are strictly wider than the reported rules bug: DCAEntry has an unscoped default manager, so filtering by strategy__id alone reached entries on strategies the caller could not see at all. No public or shared strategy was needed -- only a guessable integer. strategy_entry_add/edit/delete now resolve through the parent strategy with via="strategy". Every SharedObject delete view carried an inverted condition: if obj.owner != request.user and request.user in obj.shared_with.all(): obj.shared_with.remove(request.user) else: obj.delete() An object its owner had made public matched neither branch's intent and fell through to delete(), so any authenticated user could destroy it. Confirmed reachable for accounts, account groups, categories, tags, entities and DCA strategies. The owner now deletes, a shared user revokes only their own access, and anyone else gets a 403. The API viewsets had no object-level check at all. DjangoModelPermissions gated them shut for ordinary users, who hold no model permissions, so this was not reachable in a default install -- but the check belongs there regardless, and a user granted change_account in the admin could write any visible account. SharedObjectPermission adds it for the SharedObject viewsets, leaving reads to SharedObjectManager. account_toggle_untracked only flips the calling user's own row in the untracked_by m2m, so it takes READ rather than EDIT. Refs GHSA-83g9-vjqf-2j5q --- app/apps/accounts/views/account_groups.py | 49 ++-- app/apps/accounts/views/accounts.py | 49 ++-- app/apps/api/permissions.py | 40 ++- app/apps/api/tests/__init__.py | 1 + app/apps/api/tests/test_object_permissions.py | 171 ++++++++++++ app/apps/api/views/accounts.py | 3 + app/apps/api/views/dca.py | 4 + app/apps/api/views/transactions.py | 4 + .../tests/test_shared_object_deletion.py | 161 ++++++++++++ app/apps/dca/tests.py | 3 - app/apps/dca/tests/__init__.py | 0 app/apps/dca/tests/test_view_permissions.py | 248 ++++++++++++++++++ app/apps/dca/views.py | 85 +++--- app/apps/transactions/views/categories.py | 52 ++-- app/apps/transactions/views/entities.py | 52 ++-- app/apps/transactions/views/tags.py | 46 ++-- 16 files changed, 785 insertions(+), 183 deletions(-) create mode 100644 app/apps/api/tests/test_object_permissions.py create mode 100644 app/apps/common/tests/test_shared_object_deletion.py delete mode 100644 app/apps/dca/tests.py create mode 100644 app/apps/dca/tests/__init__.py create mode 100644 app/apps/dca/tests/test_view_permissions.py 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/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/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)