mirror of
https://github.com/eitchtee/WYGIWYH.git
synced 2026-09-07 02:17:23 +02:00
Merge pull request #592 from hackking007/fix/owner-scoped-transaction-rule-endpoints
fix: BOLA on transaction-rule endpoints via get_owned_object_or_403
This commit is contained in:
@@ -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)
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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,
|
||||
]
|
||||
|
||||
@@ -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 *
|
||||
|
||||
@@ -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")
|
||||
@@ -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 = {
|
||||
|
||||
@@ -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 = {
|
||||
|
||||
@@ -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 = {
|
||||
|
||||
@@ -0,0 +1,58 @@
|
||||
from django.core.exceptions import PermissionDenied
|
||||
from django.http import Http404
|
||||
from django.shortcuts import get_object_or_404
|
||||
|
||||
READ = "read"
|
||||
EDIT = "edit"
|
||||
|
||||
|
||||
def get_shared_object_or_error(klass, request, *, level=EDIT, via=None, **kwargs):
|
||||
"""Fetch an object like ``get_object_or_404`` while enforcing access control.
|
||||
|
||||
``SharedObjectManager`` scopes querysets to what a user may *see*, which is
|
||||
not the same as what they may *change*. Views that resolve an object from a
|
||||
URL id must state which of the two they need, otherwise a shared or public
|
||||
object becomes writable by anyone who can see it.
|
||||
|
||||
``level`` selects the check applied to the governing ``SharedObject``:
|
||||
|
||||
``READ``
|
||||
The object must be visible to the user. Denial raises :class:`Http404`
|
||||
so the response does not confirm that the id exists.
|
||||
``EDIT``
|
||||
The object must be owned by the user. Denial raises
|
||||
:class:`~django.core.exceptions.PermissionDenied` (HTTP 403), which the
|
||||
frontend surfaces as an "Access Denied" dialog. An object the user
|
||||
cannot even see raises :class:`Http404` instead, so 403 never confirms
|
||||
the existence of an object they were not allowed to know about.
|
||||
|
||||
Objects with no owner stay accessible to everyone, preserving the existing
|
||||
behaviour for legacy/unowned objects.
|
||||
|
||||
``via`` is a dotted path to the ``SharedObject`` that governs access, for
|
||||
models owned through a relation, e.g. a rule action governed by its parent
|
||||
rule::
|
||||
|
||||
get_shared_object_or_error(
|
||||
TransactionRuleAction, request, id=pk, level=EDIT, via="rule"
|
||||
)
|
||||
|
||||
The path is resolved with a plain ``getattr``, so a path that does not
|
||||
resolve raises ``AttributeError`` rather than silently granting access.
|
||||
"""
|
||||
obj = get_object_or_404(klass, **kwargs)
|
||||
|
||||
guard = obj
|
||||
for attr in via.split(".") if via else []:
|
||||
guard = getattr(guard, attr)
|
||||
|
||||
if level not in (READ, EDIT):
|
||||
raise ValueError(f"Unknown access level: {level!r}")
|
||||
|
||||
if not guard.is_visible_to(request.user):
|
||||
raise Http404
|
||||
|
||||
if level == EDIT and not guard.is_editable_by(request.user):
|
||||
raise PermissionDenied
|
||||
|
||||
return obj
|
||||
@@ -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:
|
||||
|
||||
@@ -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"
|
||||
)
|
||||
@@ -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)
|
||||
@@ -1,3 +0,0 @@
|
||||
from django.test import TestCase
|
||||
|
||||
# Create your tests here.
|
||||
@@ -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()
|
||||
)
|
||||
+49
-36
@@ -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()
|
||||
|
||||
|
||||
@@ -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)
|
||||
+58
-47
@@ -4,13 +4,19 @@ from copy import deepcopy
|
||||
|
||||
from django.contrib import messages
|
||||
from django.contrib.auth.decorators import login_required
|
||||
from django.core.exceptions import PermissionDenied
|
||||
from django.db import transaction
|
||||
from django.http import HttpResponse
|
||||
from django.shortcuts import render, get_object_or_404, redirect
|
||||
from django.shortcuts import render, redirect
|
||||
from django.utils.translation import gettext_lazy as _
|
||||
from django.views.decorators.http import require_http_methods
|
||||
|
||||
from apps.common.decorators.htmx import only_htmx
|
||||
from apps.common.functions.permissions import (
|
||||
EDIT,
|
||||
READ,
|
||||
get_shared_object_or_error,
|
||||
)
|
||||
from apps.rules.forms import (
|
||||
TransactionRuleForm,
|
||||
TransactionRuleActionForm,
|
||||
@@ -62,7 +68,9 @@ def rules_list(request):
|
||||
@disabled_on_demo
|
||||
@require_http_methods(["GET", "POST"])
|
||||
def transaction_rule_toggle_activity(request, transaction_rule_id, **kwargs):
|
||||
transaction_rule = get_object_or_404(TransactionRule, id=transaction_rule_id)
|
||||
transaction_rule = get_shared_object_or_error(
|
||||
TransactionRule, request, id=transaction_rule_id, level=EDIT
|
||||
)
|
||||
current_active = transaction_rule.active
|
||||
transaction_rule.active = not current_active
|
||||
transaction_rule.save(update_fields=["active"])
|
||||
@@ -112,17 +120,9 @@ def transaction_rule_add(request, **kwargs):
|
||||
@disabled_on_demo
|
||||
@require_http_methods(["GET", "POST"])
|
||||
def transaction_rule_edit(request, transaction_rule_id):
|
||||
transaction_rule = get_object_or_404(TransactionRule, id=transaction_rule_id)
|
||||
|
||||
if transaction_rule.owner and transaction_rule.owner != request.user:
|
||||
messages.error(request, _("Only the owner can edit this"))
|
||||
|
||||
return HttpResponse(
|
||||
status=204,
|
||||
headers={
|
||||
"HX-Trigger": "updated, hide_offcanvas",
|
||||
},
|
||||
)
|
||||
transaction_rule = get_shared_object_or_error(
|
||||
TransactionRule, request, id=transaction_rule_id, level=EDIT
|
||||
)
|
||||
|
||||
if request.method == "POST":
|
||||
form = TransactionRuleForm(request.POST, instance=transaction_rule)
|
||||
@@ -151,7 +151,9 @@ def transaction_rule_edit(request, transaction_rule_id):
|
||||
@disabled_on_demo
|
||||
@require_http_methods(["GET", "POST"])
|
||||
def transaction_rule_view(request, transaction_rule_id):
|
||||
transaction_rule = get_object_or_404(TransactionRule, id=transaction_rule_id)
|
||||
transaction_rule = get_shared_object_or_error(
|
||||
TransactionRule, request, id=transaction_rule_id, level=READ
|
||||
)
|
||||
|
||||
edit_actions = transaction_rule.transaction_actions.all()
|
||||
update_or_create_actions = (
|
||||
@@ -175,17 +177,20 @@ def transaction_rule_view(request, transaction_rule_id):
|
||||
@disabled_on_demo
|
||||
@require_http_methods(["DELETE"])
|
||||
def transaction_rule_delete(request, transaction_rule_id):
|
||||
transaction_rule = get_object_or_404(TransactionRule, id=transaction_rule_id)
|
||||
transaction_rule = get_shared_object_or_error(
|
||||
TransactionRule, request, id=transaction_rule_id, level=READ
|
||||
)
|
||||
|
||||
if (
|
||||
transaction_rule.owner != request.user
|
||||
and request.user in transaction_rule.shared_with.all()
|
||||
):
|
||||
if transaction_rule.is_editable_by(request.user):
|
||||
transaction_rule.delete()
|
||||
messages.success(request, _("Rule deleted successfully"))
|
||||
elif transaction_rule.shared_with.filter(pk=request.user.pk).exists():
|
||||
# Someone else's rule shared with us: we can drop our own access to it,
|
||||
# but never delete it.
|
||||
transaction_rule.shared_with.remove(request.user)
|
||||
messages.success(request, _("Item no longer shared with you"))
|
||||
else:
|
||||
transaction_rule.delete()
|
||||
messages.success(request, _("Rule deleted successfully"))
|
||||
raise PermissionDenied
|
||||
|
||||
return HttpResponse(
|
||||
status=204,
|
||||
@@ -200,7 +205,9 @@ def transaction_rule_delete(request, transaction_rule_id):
|
||||
@disabled_on_demo
|
||||
@require_http_methods(["GET"])
|
||||
def transaction_rule_take_ownership(request, transaction_rule_id):
|
||||
transaction_rule = get_object_or_404(TransactionRule, id=transaction_rule_id)
|
||||
transaction_rule = get_shared_object_or_error(
|
||||
TransactionRule, request, id=transaction_rule_id, level=EDIT
|
||||
)
|
||||
|
||||
if not transaction_rule.owner:
|
||||
transaction_rule.owner = request.user
|
||||
@@ -222,17 +229,7 @@ def transaction_rule_take_ownership(request, transaction_rule_id):
|
||||
@disabled_on_demo
|
||||
@require_http_methods(["GET", "POST"])
|
||||
def transaction_rule_share(request, pk):
|
||||
obj = get_object_or_404(TransactionRule, id=pk)
|
||||
|
||||
if obj.owner and obj.owner != request.user:
|
||||
messages.error(request, _("Only the owner can edit this"))
|
||||
|
||||
return HttpResponse(
|
||||
status=204,
|
||||
headers={
|
||||
"HX-Trigger": "updated, hide_offcanvas",
|
||||
},
|
||||
)
|
||||
obj = get_shared_object_or_error(TransactionRule, request, id=pk, level=EDIT)
|
||||
|
||||
if request.method == "POST":
|
||||
form = SharedObjectForm(request.POST, instance=obj, user=request.user)
|
||||
@@ -261,7 +258,9 @@ def transaction_rule_share(request, pk):
|
||||
@disabled_on_demo
|
||||
@require_http_methods(["GET", "POST"])
|
||||
def transaction_rule_action_add(request, transaction_rule_id):
|
||||
transaction_rule = get_object_or_404(TransactionRule, id=transaction_rule_id)
|
||||
transaction_rule = get_shared_object_or_error(
|
||||
TransactionRule, request, id=transaction_rule_id, level=EDIT
|
||||
)
|
||||
|
||||
if request.method == "POST":
|
||||
form = TransactionRuleActionForm(request.POST, rule=transaction_rule)
|
||||
@@ -289,12 +288,14 @@ def transaction_rule_action_add(request, transaction_rule_id):
|
||||
@disabled_on_demo
|
||||
@require_http_methods(["GET", "POST"])
|
||||
def transaction_rule_action_edit(request, transaction_rule_action_id):
|
||||
transaction_rule_action = get_object_or_404(
|
||||
TransactionRuleAction, id=transaction_rule_action_id
|
||||
)
|
||||
transaction_rule = get_object_or_404(
|
||||
TransactionRule, id=transaction_rule_action.rule.id
|
||||
transaction_rule_action = get_shared_object_or_error(
|
||||
TransactionRuleAction,
|
||||
request,
|
||||
id=transaction_rule_action_id,
|
||||
level=EDIT,
|
||||
via="rule",
|
||||
)
|
||||
transaction_rule = transaction_rule_action.rule
|
||||
|
||||
if request.method == "POST":
|
||||
form = TransactionRuleActionForm(
|
||||
@@ -327,8 +328,12 @@ def transaction_rule_action_edit(request, transaction_rule_action_id):
|
||||
@disabled_on_demo
|
||||
@require_http_methods(["DELETE"])
|
||||
def transaction_rule_action_delete(request, transaction_rule_action_id):
|
||||
transaction_rule_action = get_object_or_404(
|
||||
TransactionRuleAction, id=transaction_rule_action_id
|
||||
transaction_rule_action = get_shared_object_or_error(
|
||||
TransactionRuleAction,
|
||||
request,
|
||||
id=transaction_rule_action_id,
|
||||
level=EDIT,
|
||||
via="rule",
|
||||
)
|
||||
|
||||
transaction_rule_action.delete()
|
||||
@@ -348,7 +353,9 @@ def transaction_rule_action_delete(request, transaction_rule_action_id):
|
||||
@disabled_on_demo
|
||||
@require_http_methods(["GET", "POST"])
|
||||
def update_or_create_transaction_rule_action_add(request, transaction_rule_id):
|
||||
transaction_rule = get_object_or_404(TransactionRule, id=transaction_rule_id)
|
||||
transaction_rule = get_shared_object_or_error(
|
||||
TransactionRule, request, id=transaction_rule_id, level=EDIT
|
||||
)
|
||||
|
||||
if request.method == "POST":
|
||||
form = UpdateOrCreateTransactionRuleActionForm(
|
||||
@@ -380,7 +387,9 @@ def update_or_create_transaction_rule_action_add(request, transaction_rule_id):
|
||||
@disabled_on_demo
|
||||
@require_http_methods(["GET", "POST"])
|
||||
def update_or_create_transaction_rule_action_edit(request, pk):
|
||||
linked_action = get_object_or_404(UpdateOrCreateTransactionRuleAction, id=pk)
|
||||
linked_action = get_shared_object_or_error(
|
||||
UpdateOrCreateTransactionRuleAction, request, id=pk, level=EDIT, via="rule"
|
||||
)
|
||||
transaction_rule = linked_action.rule
|
||||
|
||||
if request.method == "POST":
|
||||
@@ -415,7 +424,9 @@ def update_or_create_transaction_rule_action_edit(request, pk):
|
||||
@disabled_on_demo
|
||||
@require_http_methods(["DELETE"])
|
||||
def update_or_create_transaction_rule_action_delete(request, pk):
|
||||
linked_action = get_object_or_404(UpdateOrCreateTransactionRuleAction, id=pk)
|
||||
linked_action = get_shared_object_or_error(
|
||||
UpdateOrCreateTransactionRuleAction, request, id=pk, level=EDIT, via="rule"
|
||||
)
|
||||
|
||||
linked_action.delete()
|
||||
|
||||
@@ -436,7 +447,7 @@ def update_or_create_transaction_rule_action_delete(request, pk):
|
||||
@disabled_on_demo
|
||||
@require_http_methods(["GET", "POST"])
|
||||
def dry_run_rule_created(request, pk):
|
||||
rule = get_object_or_404(TransactionRule, id=pk)
|
||||
rule = get_shared_object_or_error(TransactionRule, request, id=pk, level=EDIT)
|
||||
logs = None
|
||||
results = None
|
||||
|
||||
@@ -481,7 +492,7 @@ def dry_run_rule_created(request, pk):
|
||||
@disabled_on_demo
|
||||
@require_http_methods(["GET", "POST"])
|
||||
def dry_run_rule_deleted(request, pk):
|
||||
rule = get_object_or_404(TransactionRule, id=pk)
|
||||
rule = get_shared_object_or_error(TransactionRule, request, id=pk, level=EDIT)
|
||||
logs = None
|
||||
results = None
|
||||
|
||||
@@ -526,7 +537,7 @@ def dry_run_rule_deleted(request, pk):
|
||||
@disabled_on_demo
|
||||
@require_http_methods(["GET", "POST"])
|
||||
def dry_run_rule_updated(request, pk):
|
||||
rule = get_object_or_404(TransactionRule, id=pk)
|
||||
rule = get_shared_object_or_error(TransactionRule, request, id=pk, level=EDIT)
|
||||
logs = None
|
||||
results = None
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -64,14 +64,21 @@
|
||||
</div>
|
||||
</td>
|
||||
<td class="table-col-auto">
|
||||
<a class="no-underline cursor-pointer"
|
||||
role="button"
|
||||
data-tippy-content="
|
||||
{% if rule.active %}{% translate "Deactivate" %}{% else %}{% translate "Activate" %}{% endif %}"
|
||||
hx-get="{% url 'transaction_rule_toggle_activity' transaction_rule_id=rule.id %}">
|
||||
{% if rule.active %}<i class="fa-solid fa-toggle-on text-success"></i>{% else %}
|
||||
<i class="fa-solid fa-toggle-off text-error"></i>{% endif %}
|
||||
</a>
|
||||
{% if not rule.owner or user == rule.owner %}
|
||||
<a class="no-underline cursor-pointer"
|
||||
role="button"
|
||||
data-tippy-content="
|
||||
{% if rule.active %}{% translate "Deactivate" %}{% else %}{% translate "Activate" %}{% endif %}"
|
||||
hx-get="{% url 'transaction_rule_toggle_activity' transaction_rule_id=rule.id %}">
|
||||
{% if rule.active %}<i class="fa-solid fa-toggle-on text-success"></i>{% else %}
|
||||
<i class="fa-solid fa-toggle-off text-error"></i>{% endif %}
|
||||
</a>
|
||||
{% else %}
|
||||
<span data-tippy-content="{% translate "Only the owner can change this" %}">
|
||||
{% if rule.active %}<i class="fa-solid fa-toggle-on text-success opacity-50"></i>{% else %}
|
||||
<i class="fa-solid fa-toggle-off text-error opacity-50"></i>{% endif %}
|
||||
</span>
|
||||
{% endif %}
|
||||
</td>
|
||||
<td class="table-col-auto text-center">
|
||||
<div>{{ rule.order }}</div>
|
||||
|
||||
Reference in New Issue
Block a user