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)