From 68a9286ce5b743480d9ea07524061b78b2fdd712 Mon Sep 17 00:00:00 2001 From: Moshe Levi Date: Mon, 31 Aug 2026 20:02:41 +0300 Subject: [PATCH 1/4] Fix BOLA on transaction-rule endpoints via get_owned_object_or_403 Add apps.common.functions.get_owned_object_or_403, an ownership-enforcing variant of get_object_or_404, and apply it across every rules handler that resolved a TransactionRule / (UpdateOrCreate)TransactionRuleAction from a URL id without an owner check. Mirrors the check in transaction_rule_edit (obj.owner and obj.owner != request.user); objects with no owner remain accessible, preserving existing behaviour. Nested ownership (actions owned via their parent rule) is handled with owner_path='rule.owner'. --- app/apps/common/functions/permissions.py | 36 ++++++++++++++++++++++++ app/apps/rules/views.py | 33 +++++++++++++--------- 2 files changed, 55 insertions(+), 14 deletions(-) create mode 100644 app/apps/common/functions/permissions.py diff --git a/app/apps/common/functions/permissions.py b/app/apps/common/functions/permissions.py new file mode 100644 index 0000000..560c3d7 --- /dev/null +++ b/app/apps/common/functions/permissions.py @@ -0,0 +1,36 @@ +from django.core.exceptions import PermissionDenied +from django.shortcuts import get_object_or_404 + + +def get_owned_object_or_403(klass, request, *args, owner_path="owner", **kwargs): + """Fetch an object like ``get_object_or_404`` while enforcing ownership. + + Returns the object when it has no owner, or when it is owned by + ``request.user``; otherwise raises :class:`~django.core.exceptions.PermissionDenied` + (HTTP 403). This mirrors the owner check used by ``transaction_rule_edit`` + (``if obj.owner and obj.owner != request.user``) so authorization is applied + uniformly across handlers that resolve an object from a URL id. + + An object with no owner stays accessible to everyone, preserving the + existing behaviour for legacy/unowned objects. + + ``owner_path`` is a dotted attribute path to the owning user, so nested + ownership is supported for objects owned through a relation, e.g. a rule + action owned via its parent rule:: + + get_owned_object_or_403( + TransactionRuleAction, request, id=pk, owner_path="rule.owner" + ) + """ + obj = get_object_or_404(klass, *args, **kwargs) + + owner = obj + for attr in owner_path.split("."): + owner = getattr(owner, attr, None) + if owner is None: + break + + if owner is not None and owner != request.user: + raise PermissionDenied + + return obj diff --git a/app/apps/rules/views.py b/app/apps/rules/views.py index 1fb44a6..72e7b97 100644 --- a/app/apps/rules/views.py +++ b/app/apps/rules/views.py @@ -7,6 +7,7 @@ from django.contrib.auth.decorators import login_required from django.db import transaction from django.http import HttpResponse from django.shortcuts import render, get_object_or_404, redirect +from apps.common.functions.permissions import get_owned_object_or_403 from django.utils.translation import gettext_lazy as _ from django.views.decorators.http import require_http_methods @@ -62,7 +63,7 @@ def rules_list(request): @disabled_on_demo @require_http_methods(["GET", "POST"]) def transaction_rule_toggle_activity(request, transaction_rule_id, **kwargs): - transaction_rule = get_object_or_404(TransactionRule, id=transaction_rule_id) + transaction_rule = get_owned_object_or_403(TransactionRule, request, id=transaction_rule_id) current_active = transaction_rule.active transaction_rule.active = not current_active transaction_rule.save(update_fields=["active"]) @@ -151,7 +152,7 @@ def transaction_rule_edit(request, transaction_rule_id): @disabled_on_demo @require_http_methods(["GET", "POST"]) def transaction_rule_view(request, transaction_rule_id): - transaction_rule = get_object_or_404(TransactionRule, id=transaction_rule_id) + transaction_rule = get_owned_object_or_403(TransactionRule, request, id=transaction_rule_id) edit_actions = transaction_rule.transaction_actions.all() update_or_create_actions = ( @@ -200,7 +201,7 @@ def transaction_rule_delete(request, transaction_rule_id): @disabled_on_demo @require_http_methods(["GET"]) def transaction_rule_take_ownership(request, transaction_rule_id): - transaction_rule = get_object_or_404(TransactionRule, id=transaction_rule_id) + transaction_rule = get_owned_object_or_403(TransactionRule, request, id=transaction_rule_id) if not transaction_rule.owner: transaction_rule.owner = request.user @@ -261,7 +262,7 @@ def transaction_rule_share(request, pk): @disabled_on_demo @require_http_methods(["GET", "POST"]) def transaction_rule_action_add(request, transaction_rule_id): - transaction_rule = get_object_or_404(TransactionRule, id=transaction_rule_id) + transaction_rule = get_owned_object_or_403(TransactionRule, request, id=transaction_rule_id) if request.method == "POST": form = TransactionRuleActionForm(request.POST, rule=transaction_rule) @@ -289,8 +290,8 @@ def transaction_rule_action_add(request, transaction_rule_id): @disabled_on_demo @require_http_methods(["GET", "POST"]) def transaction_rule_action_edit(request, transaction_rule_action_id): - transaction_rule_action = get_object_or_404( - TransactionRuleAction, id=transaction_rule_action_id + transaction_rule_action = get_owned_object_or_403( + TransactionRuleAction, request, id=transaction_rule_action_id, owner_path="rule.owner" ) transaction_rule = get_object_or_404( TransactionRule, id=transaction_rule_action.rule.id @@ -327,8 +328,8 @@ def transaction_rule_action_edit(request, transaction_rule_action_id): @disabled_on_demo @require_http_methods(["DELETE"]) def transaction_rule_action_delete(request, transaction_rule_action_id): - transaction_rule_action = get_object_or_404( - TransactionRuleAction, id=transaction_rule_action_id + transaction_rule_action = get_owned_object_or_403( + TransactionRuleAction, request, id=transaction_rule_action_id, owner_path="rule.owner" ) transaction_rule_action.delete() @@ -348,7 +349,7 @@ def transaction_rule_action_delete(request, transaction_rule_action_id): @disabled_on_demo @require_http_methods(["GET", "POST"]) def update_or_create_transaction_rule_action_add(request, transaction_rule_id): - transaction_rule = get_object_or_404(TransactionRule, id=transaction_rule_id) + transaction_rule = get_owned_object_or_403(TransactionRule, request, id=transaction_rule_id) if request.method == "POST": form = UpdateOrCreateTransactionRuleActionForm( @@ -380,7 +381,9 @@ def update_or_create_transaction_rule_action_add(request, transaction_rule_id): @disabled_on_demo @require_http_methods(["GET", "POST"]) def update_or_create_transaction_rule_action_edit(request, pk): - linked_action = get_object_or_404(UpdateOrCreateTransactionRuleAction, id=pk) + linked_action = get_owned_object_or_403( + UpdateOrCreateTransactionRuleAction, request, id=pk, owner_path="rule.owner" + ) transaction_rule = linked_action.rule if request.method == "POST": @@ -415,7 +418,9 @@ def update_or_create_transaction_rule_action_edit(request, pk): @disabled_on_demo @require_http_methods(["DELETE"]) def update_or_create_transaction_rule_action_delete(request, pk): - linked_action = get_object_or_404(UpdateOrCreateTransactionRuleAction, id=pk) + linked_action = get_owned_object_or_403( + UpdateOrCreateTransactionRuleAction, request, id=pk, owner_path="rule.owner" + ) linked_action.delete() @@ -436,7 +441,7 @@ def update_or_create_transaction_rule_action_delete(request, pk): @disabled_on_demo @require_http_methods(["GET", "POST"]) def dry_run_rule_created(request, pk): - rule = get_object_or_404(TransactionRule, id=pk) + rule = get_owned_object_or_403(TransactionRule, request, id=pk) logs = None results = None @@ -481,7 +486,7 @@ def dry_run_rule_created(request, pk): @disabled_on_demo @require_http_methods(["GET", "POST"]) def dry_run_rule_deleted(request, pk): - rule = get_object_or_404(TransactionRule, id=pk) + rule = get_owned_object_or_403(TransactionRule, request, id=pk) logs = None results = None @@ -526,7 +531,7 @@ def dry_run_rule_deleted(request, pk): @disabled_on_demo @require_http_methods(["GET", "POST"]) def dry_run_rule_updated(request, pk): - rule = get_object_or_404(TransactionRule, id=pk) + rule = get_owned_object_or_403(TransactionRule, request, id=pk) logs = None results = None From 18d4ab7d11838c51d1937513a2534b1b2a82f0f6 Mon Sep 17 00:00:00 2001 From: Herculino Trotta Date: Tue, 1 Sep 2026 21:25:46 -0300 Subject: [PATCH 2/4] fix(rules): enforce object-level authorization on rule endpoints SharedObjectManager scopes querysets to what a user may see, which includes other people's public and shared-with-them objects. Several mutating rule endpoints treated that visibility as permission to write. Generalise get_owned_object_or_403 into get_shared_object_or_error, which takes an explicit access level instead of inferring one: - READ requires the object to be visible; denial is 404 so the response does not confirm that an id exists. - EDIT requires ownership; denial is 403, but only after the visibility check, so 403 never leaks the existence of an invisible object. This matters for TransactionRuleAction, whose manager is unscoped. The previous owner_path resolved to a User and discarded the object, so it could not express visibility at all. via= now points at the governing SharedObject and is resolved with a plain getattr, so an unresolvable path raises instead of silently granting access. is_visible_to/is_editable_by replace is_accessible_by, which was never called and tested visibility == "shared", a value that does not exist in Visibility. Also fixes two further holes in the same module: - transaction_rule_delete fell through to delete() whenever the caller was not in shared_with, so any user could delete a public rule. Now only the owner deletes; a shared user revokes their own access. - transaction_rule_view was read-only but is now explicitly READ, so rules shared with a user stay viewable. The activate/deactivate control is hidden for rules the user does not own, instead of rendering a button that always fails. Refs GHSA-83g9-vjqf-2j5q --- app/apps/common/functions/permissions.py | 64 +++++++++----- app/apps/common/models.py | 36 ++++++-- app/apps/rules/views.py | 106 ++++++++++++----------- app/templates/rules/fragments/list.html | 23 +++-- 4 files changed, 143 insertions(+), 86 deletions(-) diff --git a/app/apps/common/functions/permissions.py b/app/apps/common/functions/permissions.py index 560c3d7..713c9f9 100644 --- a/app/apps/common/functions/permissions.py +++ b/app/apps/common/functions/permissions.py @@ -1,36 +1,58 @@ from django.core.exceptions import PermissionDenied +from django.http import Http404 from django.shortcuts import get_object_or_404 +READ = "read" +EDIT = "edit" -def get_owned_object_or_403(klass, request, *args, owner_path="owner", **kwargs): - """Fetch an object like ``get_object_or_404`` while enforcing ownership. - Returns the object when it has no owner, or when it is owned by - ``request.user``; otherwise raises :class:`~django.core.exceptions.PermissionDenied` - (HTTP 403). This mirrors the owner check used by ``transaction_rule_edit`` - (``if obj.owner and obj.owner != request.user``) so authorization is applied - uniformly across handlers that resolve an object from a URL id. +def get_shared_object_or_error(klass, request, *, level=EDIT, via=None, **kwargs): + """Fetch an object like ``get_object_or_404`` while enforcing access control. - An object with no owner stays accessible to everyone, preserving the - existing behaviour for legacy/unowned objects. + ``SharedObjectManager`` scopes querysets to what a user may *see*, which is + not the same as what they may *change*. Views that resolve an object from a + URL id must state which of the two they need, otherwise a shared or public + object becomes writable by anyone who can see it. - ``owner_path`` is a dotted attribute path to the owning user, so nested - ownership is supported for objects owned through a relation, e.g. a rule - action owned via its parent rule:: + ``level`` selects the check applied to the governing ``SharedObject``: - get_owned_object_or_403( - TransactionRuleAction, request, id=pk, owner_path="rule.owner" + ``READ`` + The object must be visible to the user. Denial raises :class:`Http404` + so the response does not confirm that the id exists. + ``EDIT`` + The object must be owned by the user. Denial raises + :class:`~django.core.exceptions.PermissionDenied` (HTTP 403), which the + frontend surfaces as an "Access Denied" dialog. An object the user + cannot even see raises :class:`Http404` instead, so 403 never confirms + the existence of an object they were not allowed to know about. + + Objects with no owner stay accessible to everyone, preserving the existing + behaviour for legacy/unowned objects. + + ``via`` is a dotted path to the ``SharedObject`` that governs access, for + models owned through a relation, e.g. a rule action governed by its parent + rule:: + + get_shared_object_or_error( + TransactionRuleAction, request, id=pk, level=EDIT, via="rule" ) + + The path is resolved with a plain ``getattr``, so a path that does not + resolve raises ``AttributeError`` rather than silently granting access. """ - obj = get_object_or_404(klass, *args, **kwargs) + obj = get_object_or_404(klass, **kwargs) - owner = obj - for attr in owner_path.split("."): - owner = getattr(owner, attr, None) - if owner is None: - break + guard = obj + for attr in via.split(".") if via else []: + guard = getattr(guard, attr) - if owner is not None and owner != request.user: + if level not in (READ, EDIT): + raise ValueError(f"Unknown access level: {level!r}") + + if not guard.is_visible_to(request.user): + raise Http404 + + if level == EDIT and not guard.is_editable_by(request.user): raise PermissionDenied return obj diff --git a/app/apps/common/models.py b/app/apps/common/models.py index 10f79c8..f9a876f 100644 --- a/app/apps/common/models.py +++ b/app/apps/common/models.py @@ -58,13 +58,35 @@ class SharedObject(models.Model): models.Index(fields=["visibility"]), ] - def is_accessible_by(self, user): - """Check if a user can access this object""" - return ( - self.visibility == "public" - or self.owner == user - or (self.visibility == "shared" and user in self.shared_with.all()) - ) + # NOTE: these two predicates must stay in sync with the ``Q`` objects built + # by ``SharedObjectManager.get_queryset`` above. The manager filters at the + # queryset level and these check a single instance, so they cannot share an + # implementation; ``SharedObjectPredicateParityTests`` asserts they agree. + def is_visible_to(self, user): + """Whether ``user`` may read this object. + + Mirrors ``SharedObjectManager``: public objects, objects with no owner, + the owner's own objects, and objects explicitly shared with the user. + """ + if self.owner is None or self.visibility == "public": + return True + + if not user or not user.is_authenticated: + return False + + return self.owner_id == user.pk or self.shared_with.filter(pk=user.pk).exists() + + def is_editable_by(self, user): + """Whether ``user`` may mutate this object. + + Sharing grants read access only; mutation stays with the owner. Objects + with no owner remain editable by everyone, preserving the behaviour of + legacy/unowned objects. + """ + if self.owner is None: + return True + + return bool(user and user.is_authenticated and self.owner_id == user.pk) def save(self, *args, **kwargs): if not self.pk and not self.owner: diff --git a/app/apps/rules/views.py b/app/apps/rules/views.py index 72e7b97..5054c02 100644 --- a/app/apps/rules/views.py +++ b/app/apps/rules/views.py @@ -4,14 +4,19 @@ from copy import deepcopy from django.contrib import messages from django.contrib.auth.decorators import login_required +from django.core.exceptions import PermissionDenied from django.db import transaction from django.http import HttpResponse -from django.shortcuts import render, get_object_or_404, redirect -from apps.common.functions.permissions import get_owned_object_or_403 +from django.shortcuts import render, redirect from django.utils.translation import gettext_lazy as _ from django.views.decorators.http import require_http_methods from apps.common.decorators.htmx import only_htmx +from apps.common.functions.permissions import ( + EDIT, + READ, + get_shared_object_or_error, +) from apps.rules.forms import ( TransactionRuleForm, TransactionRuleActionForm, @@ -63,7 +68,9 @@ def rules_list(request): @disabled_on_demo @require_http_methods(["GET", "POST"]) def transaction_rule_toggle_activity(request, transaction_rule_id, **kwargs): - transaction_rule = get_owned_object_or_403(TransactionRule, request, id=transaction_rule_id) + transaction_rule = get_shared_object_or_error( + TransactionRule, request, id=transaction_rule_id, level=EDIT + ) current_active = transaction_rule.active transaction_rule.active = not current_active transaction_rule.save(update_fields=["active"]) @@ -113,17 +120,9 @@ def transaction_rule_add(request, **kwargs): @disabled_on_demo @require_http_methods(["GET", "POST"]) def transaction_rule_edit(request, transaction_rule_id): - transaction_rule = get_object_or_404(TransactionRule, id=transaction_rule_id) - - if transaction_rule.owner and transaction_rule.owner != request.user: - messages.error(request, _("Only the owner can edit this")) - - return HttpResponse( - status=204, - headers={ - "HX-Trigger": "updated, hide_offcanvas", - }, - ) + transaction_rule = get_shared_object_or_error( + TransactionRule, request, id=transaction_rule_id, level=EDIT + ) if request.method == "POST": form = TransactionRuleForm(request.POST, instance=transaction_rule) @@ -152,7 +151,9 @@ def transaction_rule_edit(request, transaction_rule_id): @disabled_on_demo @require_http_methods(["GET", "POST"]) def transaction_rule_view(request, transaction_rule_id): - transaction_rule = get_owned_object_or_403(TransactionRule, request, id=transaction_rule_id) + transaction_rule = get_shared_object_or_error( + TransactionRule, request, id=transaction_rule_id, level=READ + ) edit_actions = transaction_rule.transaction_actions.all() update_or_create_actions = ( @@ -176,17 +177,20 @@ def transaction_rule_view(request, transaction_rule_id): @disabled_on_demo @require_http_methods(["DELETE"]) def transaction_rule_delete(request, transaction_rule_id): - transaction_rule = get_object_or_404(TransactionRule, id=transaction_rule_id) + transaction_rule = get_shared_object_or_error( + TransactionRule, request, id=transaction_rule_id, level=READ + ) - if ( - transaction_rule.owner != request.user - and request.user in transaction_rule.shared_with.all() - ): + if transaction_rule.is_editable_by(request.user): + transaction_rule.delete() + messages.success(request, _("Rule deleted successfully")) + elif transaction_rule.shared_with.filter(pk=request.user.pk).exists(): + # Someone else's rule shared with us: we can drop our own access to it, + # but never delete it. transaction_rule.shared_with.remove(request.user) messages.success(request, _("Item no longer shared with you")) else: - transaction_rule.delete() - messages.success(request, _("Rule deleted successfully")) + raise PermissionDenied return HttpResponse( status=204, @@ -201,7 +205,9 @@ def transaction_rule_delete(request, transaction_rule_id): @disabled_on_demo @require_http_methods(["GET"]) def transaction_rule_take_ownership(request, transaction_rule_id): - transaction_rule = get_owned_object_or_403(TransactionRule, request, id=transaction_rule_id) + transaction_rule = get_shared_object_or_error( + TransactionRule, request, id=transaction_rule_id, level=EDIT + ) if not transaction_rule.owner: transaction_rule.owner = request.user @@ -223,17 +229,7 @@ def transaction_rule_take_ownership(request, transaction_rule_id): @disabled_on_demo @require_http_methods(["GET", "POST"]) def transaction_rule_share(request, pk): - obj = get_object_or_404(TransactionRule, id=pk) - - if obj.owner and obj.owner != request.user: - messages.error(request, _("Only the owner can edit this")) - - return HttpResponse( - status=204, - headers={ - "HX-Trigger": "updated, hide_offcanvas", - }, - ) + obj = get_shared_object_or_error(TransactionRule, request, id=pk, level=EDIT) if request.method == "POST": form = SharedObjectForm(request.POST, instance=obj, user=request.user) @@ -262,7 +258,9 @@ def transaction_rule_share(request, pk): @disabled_on_demo @require_http_methods(["GET", "POST"]) def transaction_rule_action_add(request, transaction_rule_id): - transaction_rule = get_owned_object_or_403(TransactionRule, request, id=transaction_rule_id) + transaction_rule = get_shared_object_or_error( + TransactionRule, request, id=transaction_rule_id, level=EDIT + ) if request.method == "POST": form = TransactionRuleActionForm(request.POST, rule=transaction_rule) @@ -290,12 +288,14 @@ def transaction_rule_action_add(request, transaction_rule_id): @disabled_on_demo @require_http_methods(["GET", "POST"]) def transaction_rule_action_edit(request, transaction_rule_action_id): - transaction_rule_action = get_owned_object_or_403( - TransactionRuleAction, request, id=transaction_rule_action_id, owner_path="rule.owner" - ) - transaction_rule = get_object_or_404( - TransactionRule, id=transaction_rule_action.rule.id + transaction_rule_action = get_shared_object_or_error( + TransactionRuleAction, + request, + id=transaction_rule_action_id, + level=EDIT, + via="rule", ) + transaction_rule = transaction_rule_action.rule if request.method == "POST": form = TransactionRuleActionForm( @@ -328,8 +328,12 @@ def transaction_rule_action_edit(request, transaction_rule_action_id): @disabled_on_demo @require_http_methods(["DELETE"]) def transaction_rule_action_delete(request, transaction_rule_action_id): - transaction_rule_action = get_owned_object_or_403( - TransactionRuleAction, request, id=transaction_rule_action_id, owner_path="rule.owner" + transaction_rule_action = get_shared_object_or_error( + TransactionRuleAction, + request, + id=transaction_rule_action_id, + level=EDIT, + via="rule", ) transaction_rule_action.delete() @@ -349,7 +353,9 @@ def transaction_rule_action_delete(request, transaction_rule_action_id): @disabled_on_demo @require_http_methods(["GET", "POST"]) def update_or_create_transaction_rule_action_add(request, transaction_rule_id): - transaction_rule = get_owned_object_or_403(TransactionRule, request, id=transaction_rule_id) + transaction_rule = get_shared_object_or_error( + TransactionRule, request, id=transaction_rule_id, level=EDIT + ) if request.method == "POST": form = UpdateOrCreateTransactionRuleActionForm( @@ -381,8 +387,8 @@ def update_or_create_transaction_rule_action_add(request, transaction_rule_id): @disabled_on_demo @require_http_methods(["GET", "POST"]) def update_or_create_transaction_rule_action_edit(request, pk): - linked_action = get_owned_object_or_403( - UpdateOrCreateTransactionRuleAction, request, id=pk, owner_path="rule.owner" + linked_action = get_shared_object_or_error( + UpdateOrCreateTransactionRuleAction, request, id=pk, level=EDIT, via="rule" ) transaction_rule = linked_action.rule @@ -418,8 +424,8 @@ def update_or_create_transaction_rule_action_edit(request, pk): @disabled_on_demo @require_http_methods(["DELETE"]) def update_or_create_transaction_rule_action_delete(request, pk): - linked_action = get_owned_object_or_403( - UpdateOrCreateTransactionRuleAction, request, id=pk, owner_path="rule.owner" + linked_action = get_shared_object_or_error( + UpdateOrCreateTransactionRuleAction, request, id=pk, level=EDIT, via="rule" ) linked_action.delete() @@ -441,7 +447,7 @@ def update_or_create_transaction_rule_action_delete(request, pk): @disabled_on_demo @require_http_methods(["GET", "POST"]) def dry_run_rule_created(request, pk): - rule = get_owned_object_or_403(TransactionRule, request, id=pk) + rule = get_shared_object_or_error(TransactionRule, request, id=pk, level=EDIT) logs = None results = None @@ -486,7 +492,7 @@ def dry_run_rule_created(request, pk): @disabled_on_demo @require_http_methods(["GET", "POST"]) def dry_run_rule_deleted(request, pk): - rule = get_owned_object_or_403(TransactionRule, request, id=pk) + rule = get_shared_object_or_error(TransactionRule, request, id=pk, level=EDIT) logs = None results = None @@ -531,7 +537,7 @@ def dry_run_rule_deleted(request, pk): @disabled_on_demo @require_http_methods(["GET", "POST"]) def dry_run_rule_updated(request, pk): - rule = get_owned_object_or_403(TransactionRule, request, id=pk) + rule = get_shared_object_or_error(TransactionRule, request, id=pk, level=EDIT) logs = None results = None diff --git a/app/templates/rules/fragments/list.html b/app/templates/rules/fragments/list.html index 07cb0f7..998708e 100644 --- a/app/templates/rules/fragments/list.html +++ b/app/templates/rules/fragments/list.html @@ -64,14 +64,21 @@ - - {% if rule.active %}{% else %} - {% endif %} - + {% if not rule.owner or user == rule.owner %} + + {% if rule.active %}{% else %} + {% endif %} + + {% else %} + + {% if rule.active %}{% else %} + {% endif %} + + {% endif %}
{{ rule.order }}
From 039ad225d3901e3bc762d0bceb8b49d7bb8be668 Mon Sep 17 00:00:00 2001 From: Herculino Trotta Date: Tue, 1 Sep 2026 21:25:53 -0300 Subject: [PATCH 3/4] test(rules): cover object-level authorization for transaction rules Regression tests for GHSA-83g9-vjqf-2j5q, one per endpoint the advisory named plus the delete and view paths found alongside them: a non-owner gets 403 on every mutation of a public or shared rule, and the object is asserted unchanged afterwards. Also covers the parts that are easy to regress in the other direction: shared users keep read access, a shared user deleting only revokes their own access, unowned rules stay claimable, and children of invisible rules answer 404 rather than 403. SharedObjectPredicateParityTests asserts is_visible_to agrees with SharedObjectManager across every owner/visibility/shared combination. The manager builds a Q and the predicate tests an instance, so they cannot share an implementation and can otherwise drift apart. --- app/apps/common/tests/test_permissions.py | 269 ++++++++++ app/apps/rules/tests/test_view_permissions.py | 473 ++++++++++++++++++ 2 files changed, 742 insertions(+) create mode 100644 app/apps/common/tests/test_permissions.py create mode 100644 app/apps/rules/tests/test_view_permissions.py diff --git a/app/apps/common/tests/test_permissions.py b/app/apps/common/tests/test_permissions.py new file mode 100644 index 0000000..a803bed --- /dev/null +++ b/app/apps/common/tests/test_permissions.py @@ -0,0 +1,269 @@ +from django.contrib.auth import get_user_model +from django.contrib.auth.models import AnonymousUser +from django.core.exceptions import PermissionDenied +from django.http import Http404 +from django.test import RequestFactory, TestCase, override_settings + +from apps.common.functions.permissions import ( + EDIT, + READ, + get_shared_object_or_error, +) +from apps.common.middleware.thread_local import delete_current_user, write_current_user +from apps.rules.models import TransactionRule, TransactionRuleAction + + +@override_settings( + STORAGES={ + "default": {"BACKEND": "django.core.files.storage.FileSystemStorage"}, + "staticfiles": { + "BACKEND": "django.contrib.staticfiles.storage.StaticFilesStorage" + }, + }, + WHITENOISE_AUTOREFRESH=True, +) +class SharedObjectPredicateTests(TestCase): + """Unit tests for is_visible_to / is_editable_by on SharedObject.""" + + def setUp(self): + User = get_user_model() + self.owner = User.objects.create_user( + email="owner@test.com", password="testpass123" + ) + self.shared_user = User.objects.create_user( + email="shared@test.com", password="testpass123" + ) + self.stranger = User.objects.create_user( + email="stranger@test.com", password="testpass123" + ) + + def _rule(self, **kwargs): + kwargs.setdefault("name", "Rule") + kwargs.setdefault("trigger", "True") + return TransactionRule.all_objects.create(**kwargs) + + def test_owner_can_read_and_edit_own_private_rule(self): + rule = self._rule(owner=self.owner, visibility="private") + + self.assertTrue(rule.is_visible_to(self.owner)) + self.assertTrue(rule.is_editable_by(self.owner)) + + def test_private_rule_is_invisible_to_stranger(self): + rule = self._rule(owner=self.owner, visibility="private") + + self.assertFalse(rule.is_visible_to(self.stranger)) + self.assertFalse(rule.is_editable_by(self.stranger)) + + def test_shared_rule_is_readable_but_not_editable(self): + rule = self._rule(owner=self.owner, visibility="private") + rule.shared_with.add(self.shared_user) + + self.assertTrue(rule.is_visible_to(self.shared_user)) + self.assertFalse(rule.is_editable_by(self.shared_user)) + + def test_public_rule_is_readable_but_not_editable(self): + rule = self._rule(owner=self.owner, visibility="public") + + self.assertTrue(rule.is_visible_to(self.stranger)) + self.assertFalse(rule.is_editable_by(self.stranger)) + + def test_unowned_rule_stays_readable_and_editable_by_everyone(self): + rule = self._rule(owner=None, visibility="private") + + self.assertTrue(rule.is_visible_to(self.stranger)) + self.assertTrue(rule.is_editable_by(self.stranger)) + + def test_anonymous_user_gets_no_access_to_owned_rules(self): + rule = self._rule(owner=self.owner, visibility="private") + + self.assertFalse(rule.is_visible_to(AnonymousUser())) + self.assertFalse(rule.is_editable_by(AnonymousUser())) + + +@override_settings( + STORAGES={ + "default": {"BACKEND": "django.core.files.storage.FileSystemStorage"}, + "staticfiles": { + "BACKEND": "django.contrib.staticfiles.storage.StaticFilesStorage" + }, + }, + WHITENOISE_AUTOREFRESH=True, +) +class SharedObjectPredicateParityTests(TestCase): + """is_visible_to must agree with what SharedObjectManager returns. + + The manager filters at the queryset level and the predicate checks a single + instance, so the two cannot share an implementation. This asserts they do + not drift apart. + """ + + def setUp(self): + User = get_user_model() + self.owner = User.objects.create_user( + email="owner@test.com", password="testpass123" + ) + self.other = User.objects.create_user( + email="other@test.com", password="testpass123" + ) + self.addCleanup(self._clear_current_user) + + def _clear_current_user(self): + try: + delete_current_user() + except AttributeError: + pass + + def test_manager_and_predicate_agree_over_every_combination(self): + combinations = [] + for owner in (self.owner, self.other, None): + for visibility in ("private", "public"): + for shared in (True, False): + rule = TransactionRule.all_objects.create( + name=f"{owner}-{visibility}-{shared}", + trigger="True", + owner=owner, + visibility=visibility, + ) + if shared: + rule.shared_with.add(self.owner) + combinations.append(rule) + + write_current_user(self.owner) + visible_ids = set(TransactionRule.objects.values_list("id", flat=True)) + + for rule in combinations: + with self.subTest(rule=rule.name): + self.assertEqual( + rule.id in visible_ids, + rule.is_visible_to(self.owner), + f"manager and is_visible_to disagree for {rule.name}", + ) + + +@override_settings( + STORAGES={ + "default": {"BACKEND": "django.core.files.storage.FileSystemStorage"}, + "staticfiles": { + "BACKEND": "django.contrib.staticfiles.storage.StaticFilesStorage" + }, + }, + WHITENOISE_AUTOREFRESH=True, +) +class GetSharedObjectOrErrorTests(TestCase): + def setUp(self): + User = get_user_model() + self.owner = User.objects.create_user( + email="owner@test.com", password="testpass123" + ) + self.stranger = User.objects.create_user( + email="stranger@test.com", password="testpass123" + ) + self.factory = RequestFactory() + + self.public_rule = TransactionRule.all_objects.create( + name="Public", trigger="True", owner=self.owner, visibility="public" + ) + self.private_rule = TransactionRule.all_objects.create( + name="Private", trigger="True", owner=self.owner, visibility="private" + ) + self.public_action = TransactionRuleAction.objects.create( + rule=self.public_rule, field="notes", value="x" + ) + self.private_action = TransactionRuleAction.objects.create( + rule=self.private_rule, field="notes", value="x" + ) + + self.addCleanup(self._clear_current_user) + + def _clear_current_user(self): + try: + delete_current_user() + except AttributeError: + pass + + def _request(self, user): + request = self.factory.get("/") + request.user = user + write_current_user(user) + return request + + def test_read_allows_visible_object(self): + request = self._request(self.stranger) + + rule = get_shared_object_or_error( + TransactionRule, request, id=self.public_rule.id, level=READ + ) + + self.assertEqual(rule, self.public_rule) + + def test_edit_denies_visible_but_unowned_object_with_403(self): + request = self._request(self.stranger) + + with self.assertRaises(PermissionDenied): + get_shared_object_or_error( + TransactionRule, request, id=self.public_rule.id, level=EDIT + ) + + def test_edit_allows_owner(self): + request = self._request(self.owner) + + rule = get_shared_object_or_error( + TransactionRule, request, id=self.public_rule.id, level=EDIT + ) + + self.assertEqual(rule, self.public_rule) + + def test_invisible_object_raises_404_not_403(self): + """403 must not confirm the existence of an object the user cannot see.""" + request = self._request(self.stranger) + + with self.assertRaises(Http404): + get_shared_object_or_error( + TransactionRule, request, id=self.private_rule.id, level=EDIT + ) + + def test_via_traverses_to_the_governing_object(self): + request = self._request(self.stranger) + + with self.assertRaises(PermissionDenied): + get_shared_object_or_error( + TransactionRuleAction, + request, + id=self.public_action.id, + level=EDIT, + via="rule", + ) + + def test_via_hides_children_of_invisible_parents_behind_404(self): + """TransactionRuleAction has an unscoped manager, so the id is reachable.""" + request = self._request(self.stranger) + + with self.assertRaises(Http404): + get_shared_object_or_error( + TransactionRuleAction, + request, + id=self.private_action.id, + level=EDIT, + via="rule", + ) + + def test_unresolvable_via_path_fails_closed(self): + """A typo'd path must raise, never silently grant access.""" + request = self._request(self.stranger) + + with self.assertRaises(AttributeError): + get_shared_object_or_error( + TransactionRuleAction, + request, + id=self.public_action.id, + level=EDIT, + via="rulee", + ) + + def test_unknown_level_is_rejected(self): + request = self._request(self.owner) + + with self.assertRaises(ValueError): + get_shared_object_or_error( + TransactionRule, request, id=self.public_rule.id, level="write" + ) diff --git a/app/apps/rules/tests/test_view_permissions.py b/app/apps/rules/tests/test_view_permissions.py new file mode 100644 index 0000000..9506dd0 --- /dev/null +++ b/app/apps/rules/tests/test_view_permissions.py @@ -0,0 +1,473 @@ +"""Object-level authorization tests for the transaction-rule endpoints. + +Regression coverage for GHSA-83g9-vjqf-2j5q: SharedObjectManager scopes rules to +what a user may *see*, which included other people's public and shared-with-them +rules. Several mutating endpoints treated that visibility as permission to write. +""" + +from django.contrib.auth import get_user_model +from django.test import TestCase, override_settings +from django.urls import reverse + +from apps.rules.models import ( + TransactionRule, + TransactionRuleAction, + UpdateOrCreateTransactionRuleAction, +) + +HTMX = {"HTTP_HX_REQUEST": "true"} + + +@override_settings( + STORAGES={ + "default": {"BACKEND": "django.core.files.storage.FileSystemStorage"}, + "staticfiles": { + "BACKEND": "django.contrib.staticfiles.storage.StaticFilesStorage" + }, + }, + WHITENOISE_AUTOREFRESH=True, + DEMO=False, +) +class TransactionRuleObjectPermissionTests(TestCase): + def setUp(self): + User = get_user_model() + self.owner = User.objects.create_user( + email="owner@test.com", password="testpass123" + ) + self.shared_user = User.objects.create_user( + email="shared@test.com", password="testpass123" + ) + self.stranger = User.objects.create_user( + email="stranger@test.com", password="testpass123" + ) + + # Public: visible to everyone through SharedObjectManager. + self.public_rule = TransactionRule.all_objects.create( + name="Public rule", + trigger="True", + owner=self.owner, + visibility="public", + active=True, + ) + # Private but shared with shared_user: visible to them, not to stranger. + self.shared_rule = TransactionRule.all_objects.create( + name="Shared rule", + trigger="True", + owner=self.owner, + visibility="private", + active=True, + ) + self.shared_rule.shared_with.add(self.shared_user) + # Private and unshared: invisible to everyone but the owner. + self.private_rule = TransactionRule.all_objects.create( + name="Private rule", + trigger="True", + owner=self.owner, + visibility="private", + active=True, + ) + + self.public_action = TransactionRuleAction.objects.create( + rule=self.public_rule, field="notes", value="owned by owner" + ) + self.private_action = TransactionRuleAction.objects.create( + rule=self.private_rule, field="notes", value="owned by owner" + ) + self.public_linked_action = UpdateOrCreateTransactionRuleAction.objects.create( + rule=self.public_rule + ) + self.private_linked_action = UpdateOrCreateTransactionRuleAction.objects.create( + rule=self.private_rule + ) + + def login(self, user): + self.client.force_login(user) + + # ------------------------------------------------------------------ + # toggle-active + # ------------------------------------------------------------------ + def test_stranger_cannot_toggle_public_rule(self): + self.login(self.stranger) + + response = self.client.get( + reverse( + "transaction_rule_toggle_activity", + kwargs={"transaction_rule_id": self.public_rule.id}, + ), + **HTMX, + ) + + self.assertEqual(response.status_code, 403) + self.public_rule.refresh_from_db() + self.assertTrue(self.public_rule.active) + + def test_shared_user_cannot_toggle_shared_rule(self): + self.login(self.shared_user) + + response = self.client.get( + reverse( + "transaction_rule_toggle_activity", + kwargs={"transaction_rule_id": self.shared_rule.id}, + ), + **HTMX, + ) + + self.assertEqual(response.status_code, 403) + self.shared_rule.refresh_from_db() + self.assertTrue(self.shared_rule.active) + + def test_owner_can_toggle_own_rule(self): + self.login(self.owner) + + response = self.client.get( + reverse( + "transaction_rule_toggle_activity", + kwargs={"transaction_rule_id": self.public_rule.id}, + ), + **HTMX, + ) + + self.assertEqual(response.status_code, 204) + self.public_rule.refresh_from_db() + self.assertFalse(self.public_rule.active) + + # ------------------------------------------------------------------ + # rule actions + # ------------------------------------------------------------------ + def test_stranger_cannot_add_action_to_public_rule(self): + self.login(self.stranger) + + response = self.client.post( + reverse( + "transaction_rule_action_add", + kwargs={"transaction_rule_id": self.public_rule.id}, + ), + data={"field": "notes", "value": "injected", "order": 0}, + **HTMX, + ) + + self.assertEqual(response.status_code, 403) + self.assertFalse( + TransactionRuleAction.objects.filter(value="injected").exists() + ) + + def test_stranger_cannot_edit_action_on_public_rule(self): + self.login(self.stranger) + + response = self.client.post( + reverse( + "transaction_rule_action_edit", + kwargs={"transaction_rule_action_id": self.public_action.id}, + ), + data={"field": "notes", "value": "rewritten", "order": 0}, + **HTMX, + ) + + self.assertEqual(response.status_code, 403) + self.public_action.refresh_from_db() + self.assertEqual(self.public_action.value, "owned by owner") + + def test_stranger_cannot_delete_action_on_public_rule(self): + self.login(self.stranger) + + response = self.client.delete( + reverse( + "transaction_rule_action_delete", + kwargs={"transaction_rule_action_id": self.public_action.id}, + ), + **HTMX, + ) + + self.assertEqual(response.status_code, 403) + self.assertTrue( + TransactionRuleAction.objects.filter(pk=self.public_action.pk).exists() + ) + + def test_stranger_cannot_delete_action_on_invisible_rule(self): + """The child manager is unscoped, so the id is reachable by guessing. + + The parent rule is invisible to the stranger, so the response must be a + 404 rather than a 403 that confirms the action exists. + """ + self.login(self.stranger) + + response = self.client.delete( + reverse( + "transaction_rule_action_delete", + kwargs={"transaction_rule_action_id": self.private_action.id}, + ), + **HTMX, + ) + + self.assertEqual(response.status_code, 404) + self.assertTrue( + TransactionRuleAction.objects.filter(pk=self.private_action.pk).exists() + ) + + def test_owner_can_delete_own_action(self): + self.login(self.owner) + + response = self.client.delete( + reverse( + "transaction_rule_action_delete", + kwargs={"transaction_rule_action_id": self.public_action.id}, + ), + **HTMX, + ) + + self.assertEqual(response.status_code, 204) + self.assertFalse( + TransactionRuleAction.objects.filter(pk=self.public_action.pk).exists() + ) + + # ------------------------------------------------------------------ + # update-or-create rule actions + # ------------------------------------------------------------------ + def test_stranger_cannot_add_linked_action_to_public_rule(self): + self.login(self.stranger) + + response = self.client.post( + reverse( + "update_or_create_transaction_rule_action_add", + kwargs={"transaction_rule_id": self.public_rule.id}, + ), + data={}, + **HTMX, + ) + + self.assertEqual(response.status_code, 403) + self.assertEqual( + UpdateOrCreateTransactionRuleAction.objects.filter( + rule=self.public_rule + ).count(), + 1, + ) + + def test_stranger_cannot_edit_linked_action_on_public_rule(self): + self.login(self.stranger) + + response = self.client.post( + reverse( + "update_or_create_transaction_rule_action_edit", + kwargs={"pk": self.public_linked_action.id}, + ), + data={"filter": "injected"}, + **HTMX, + ) + + self.assertEqual(response.status_code, 403) + self.public_linked_action.refresh_from_db() + self.assertEqual(self.public_linked_action.filter, "") + + def test_stranger_cannot_delete_linked_action_on_public_rule(self): + self.login(self.stranger) + + response = self.client.delete( + reverse( + "update_or_create_transaction_rule_action_delete", + kwargs={"pk": self.public_linked_action.id}, + ), + **HTMX, + ) + + self.assertEqual(response.status_code, 403) + self.assertTrue( + UpdateOrCreateTransactionRuleAction.objects.filter( + pk=self.public_linked_action.pk + ).exists() + ) + + def test_stranger_cannot_delete_linked_action_on_invisible_rule(self): + self.login(self.stranger) + + response = self.client.delete( + reverse( + "update_or_create_transaction_rule_action_delete", + kwargs={"pk": self.private_linked_action.id}, + ), + **HTMX, + ) + + self.assertEqual(response.status_code, 404) + self.assertTrue( + UpdateOrCreateTransactionRuleAction.objects.filter( + pk=self.private_linked_action.pk + ).exists() + ) + + # ------------------------------------------------------------------ + # edit / share / dry-run + # ------------------------------------------------------------------ + def test_stranger_cannot_edit_public_rule(self): + self.login(self.stranger) + + response = self.client.post( + reverse( + "transaction_rule_edit", + kwargs={"transaction_rule_id": self.public_rule.id}, + ), + data={"name": "hijacked", "trigger": "True", "order": 0}, + **HTMX, + ) + + self.assertEqual(response.status_code, 403) + self.public_rule.refresh_from_db() + self.assertEqual(self.public_rule.name, "Public rule") + + def test_stranger_cannot_change_sharing_of_public_rule(self): + self.login(self.stranger) + + response = self.client.post( + reverse( + "transaction_rule_share_settings", kwargs={"pk": self.public_rule.id} + ), + data={"visibility": "private"}, + **HTMX, + ) + + self.assertEqual(response.status_code, 403) + self.public_rule.refresh_from_db() + self.assertEqual(self.public_rule.visibility, "public") + + def test_stranger_cannot_dry_run_public_rule(self): + self.login(self.stranger) + + response = self.client.get( + reverse( + "transaction_rule_dry_run_created", kwargs={"pk": self.public_rule.id} + ), + **HTMX, + ) + + self.assertEqual(response.status_code, 403) + + # ------------------------------------------------------------------ + # delete: sharing may be revoked, ownership may not be overridden + # ------------------------------------------------------------------ + def test_stranger_cannot_delete_public_rule(self): + """The original condition fell through to delete() for public rules.""" + self.login(self.stranger) + + response = self.client.delete( + reverse( + "transaction_rule_delete", + kwargs={"transaction_rule_id": self.public_rule.id}, + ), + **HTMX, + ) + + self.assertEqual(response.status_code, 403) + self.assertTrue( + TransactionRule.all_objects.filter(pk=self.public_rule.pk).exists() + ) + + def test_shared_user_deleting_only_revokes_their_own_access(self): + self.login(self.shared_user) + + response = self.client.delete( + reverse( + "transaction_rule_delete", + kwargs={"transaction_rule_id": self.shared_rule.id}, + ), + **HTMX, + ) + + self.assertEqual(response.status_code, 204) + self.assertTrue( + TransactionRule.all_objects.filter(pk=self.shared_rule.pk).exists() + ) + self.assertNotIn(self.shared_user, self.shared_rule.shared_with.all()) + + def test_owner_can_delete_own_rule(self): + self.login(self.owner) + + response = self.client.delete( + reverse( + "transaction_rule_delete", + kwargs={"transaction_rule_id": self.public_rule.id}, + ), + **HTMX, + ) + + self.assertEqual(response.status_code, 204) + self.assertFalse( + TransactionRule.all_objects.filter(pk=self.public_rule.pk).exists() + ) + + # ------------------------------------------------------------------ + # reads must keep working for shared users + # ------------------------------------------------------------------ + def test_shared_user_can_still_view_shared_rule(self): + self.login(self.shared_user) + + response = self.client.get( + reverse( + "transaction_rule_view", + kwargs={"transaction_rule_id": self.shared_rule.id}, + ), + **HTMX, + ) + + self.assertEqual(response.status_code, 200) + + def test_stranger_can_view_public_rule(self): + self.login(self.stranger) + + response = self.client.get( + reverse( + "transaction_rule_view", + kwargs={"transaction_rule_id": self.public_rule.id}, + ), + **HTMX, + ) + + self.assertEqual(response.status_code, 200) + + def test_stranger_cannot_view_private_rule(self): + self.login(self.stranger) + + response = self.client.get( + reverse( + "transaction_rule_view", + kwargs={"transaction_rule_id": self.private_rule.id}, + ), + **HTMX, + ) + + self.assertEqual(response.status_code, 404) + + # ------------------------------------------------------------------ + # take ownership stays available for unowned rules only + # ------------------------------------------------------------------ + def test_take_ownership_of_unowned_rule_still_works(self): + unowned = TransactionRule.all_objects.create( + name="Legacy rule", trigger="True", owner=None, visibility="private" + ) + self.login(self.stranger) + + response = self.client.get( + reverse( + "transaction_rule_take_ownership", + kwargs={"transaction_rule_id": unowned.id}, + ), + **HTMX, + ) + + self.assertEqual(response.status_code, 204) + unowned.refresh_from_db() + self.assertEqual(unowned.owner, self.stranger) + + def test_cannot_take_ownership_of_someone_elses_public_rule(self): + self.login(self.stranger) + + response = self.client.get( + reverse( + "transaction_rule_take_ownership", + kwargs={"transaction_rule_id": self.public_rule.id}, + ), + **HTMX, + ) + + self.assertEqual(response.status_code, 403) + self.public_rule.refresh_from_db() + self.assertEqual(self.public_rule.owner, self.owner) From e2f26b3629cd742779fb35c8c536350ee24f0039 Mon Sep 17 00:00:00 2001 From: Herculino Trotta Date: Tue, 1 Sep 2026 23:15:41 -0300 Subject: [PATCH 4/4] 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)