From 331ee1e0a150541971f1bb07ec4a7e08f9467b98 Mon Sep 17 00:00:00 2001 From: jh-RLI Date: Thu, 1 Oct 2026 13:33:13 +0200 Subject: [PATCH 1/7] Restrict profile routes to their owner Every route under profile// now answers only when the id is the caller's own. The rule lives in login/access.py as a mixin and a decorator: another id answers 404 (the same for an id that does not exist), an anonymous page goes to the login page with next, and an anonymous htmx request gets 401. Platform admins are not exempt. On a match the view works with request.user and no longer loads the user named in the URL. The dataset routes stack the creator check under the rule, the edit view drops its hand-written check, and the account-delete route answers 404 instead of raising. A structural test walks login's URL patterns and fails for a user_id route without the rule. Co-Authored-By: Claude Opus 5.5 (1M context) --- login/access.py | 94 +++++++ login/tests/test_dataset_views.py | 32 ++- login/tests/test_profile_owner_rule.py | 323 +++++++++++++++++++++++++ login/tests/test_views.py | 14 +- login/urls.py | 5 +- login/views.py | 81 ++++--- 6 files changed, 493 insertions(+), 56 deletions(-) create mode 100644 login/access.py create mode 100644 login/tests/test_profile_owner_rule.py diff --git a/login/access.py b/login/access.py new file mode 100644 index 000000000..af27aec53 --- /dev/null +++ b/login/access.py @@ -0,0 +1,94 @@ +""" +SPDX-FileCopyrightText: 2026 Jonas Huber © Reiner Lemoine Institut +SPDX-License-Identifier: AGPL-3.0-or-later + +Who may reach a view under ``profile//``: its owner, and nobody else. + +A profile page is a user's own dashboard. There is no public profile and no +"someone else's dashboard" to show, so every route carrying a ``user_id`` +answers only when that id is the caller's own: + +- anonymous, full page: 302 to the login page, with ``next`` +- anonymous, htmx request: 401, no body +- logged in with another id: 404, page and htmx alike +- the owner: the view runs + +Platform admins are not exempt: support goes through the Django shell, not +through someone's dashboard. + +A foreign id answers 404, never 403, and the same 404 for an id that exists +and one that does not, so the answer does not reveal which accounts exist. On +a match the view is handed ``request.user``; the user named in the URL is +never loaded. + +An htmx request gets a bare 401 rather than the login redirect, because htmx +follows a redirect and would swap the login page into the fragment it was +asked to fill. + +This lives in its own module rather than in ``login/permissions.py`` because +that one holds the permission levels and is imported by ``login/models.py`` at +app-loading time, before the auth views this module needs can be imported. + +Both forms carry ``owner_rule = True``. ``login/tests/test_profile_owner_rule`` +walks the URL patterns and fails for a ``user_id`` route without it. +""" # noqa: 501 + +from functools import wraps + +from django.contrib.auth.views import redirect_to_login +from django.http import Http404, HttpResponse + + +def _is_htmx(request) -> bool: + return "HX-Request" in request.headers + + +def _refusal(request, user_id): + """The response that refuses this caller, or None for the owner. + + Raises Http404 for a logged-in caller with another id. + """ + if not request.user.is_authenticated: + if _is_htmx(request): + return HttpResponse(status=401) + return redirect_to_login(request.get_full_path()) + if str(request.user.pk) != str(user_id): + raise Http404 + return None + + +class ProfileOwnerRequiredMixin: + """The owner rule for class-based views. + + List it FIRST in the bases: ``View.dispatch`` does not call further along + the MRO, so a mixin placed after the view class never runs. + + On a match ``self.profile_user`` is the caller. + """ + + owner_rule = True + + def dispatch(self, request, *args, **kwargs): + refusal = _refusal(request, kwargs.get("user_id")) + if refusal is not None: + return refusal + self.profile_user = request.user + return super().dispatch(request, *args, **kwargs) + + +def profile_owner_required(view_func): + """The owner rule for function views. + + The view is called as ``view_func(request, profile_user, ...)`` without + the ``user_id``, and ``profile_user`` is the caller. + """ + + @wraps(view_func) + def wrapper(request, user_id, *args, **kwargs): + refusal = _refusal(request, user_id) + if refusal is not None: + return refusal + return view_func(request, request.user, *args, **kwargs) + + wrapper.owner_rule = True + return wrapper diff --git a/login/tests/test_dataset_views.py b/login/tests/test_dataset_views.py index a79b4eb88..2346f93a1 100644 --- a/login/tests/test_dataset_views.py +++ b/login/tests/test_dataset_views.py @@ -132,6 +132,7 @@ def test_create_duplicate_normalized_name_shows_inline_error(self): self.assertContains(response, "taken_name") def test_cannot_create_on_foreign_profile(self): + # another user's dashboard does not exist for this caller: 404 response = self.client.post( reverse("login:datasets", args=[self.other_user.id]), { @@ -139,7 +140,7 @@ def test_cannot_create_on_foreign_profile(self): "description": "Posting on someone else's dashboard", }, ) - self.assertEqual(response.status_code, 403) + self.assertEqual(response.status_code, 404) self.assertFalse(Dataset.objects.filter(name="sneaky_dataset").exists()) def test_card_links_to_public_detail_in_new_tab(self): @@ -345,9 +346,10 @@ def test_edit_validation_error_is_shown_inline(self): self.assertContains(response, "invalid-feedback") def test_edit_forbidden_for_non_creator(self): + # through the caller's own dashboard, so the creator check answers self.client.force_login(self.other_user) response = self.client.post( - self.edit_url, + reverse("login:dataset-edit", args=[self.other_user.id, "quick_dataset"]), {"title": "Hijacked", "description": "Should fail"}, ) self.assertEqual(response.status_code, 403) @@ -395,8 +397,11 @@ def test_delete_removes_only_its_own_card(self): self.assertNotContains(response, "Create dataset") def test_delete_forbidden_for_non_creator(self): + # through the caller's own dashboard, so the creator check answers self.client.force_login(self.other_user) - response = self.client.post(self.delete_url) + response = self.client.post( + reverse("login:dataset-delete", args=[self.other_user.id, "quick_dataset"]) + ) self.assertEqual(response.status_code, 403) self.assertTrue(Dataset.objects.filter(name="quick_dataset").exists()) @@ -406,11 +411,11 @@ def test_delete_confirm_copy_mentions_tables_survive(self): self.assertContains(response, "not deleted") def test_actions_not_rendered_for_other_users(self): + # another user's dashboard is refused outright (owner rule) self.client.force_login(self.other_user) response = self.client.get(reverse("login:datasets", args=[self.user.id])) - self.assertEqual(response.status_code, 200) - self.assertNotContains(response, "quick_dataset") - self.assertNotContains(response, "hx-confirm") + self.assertNotContains(response, "quick_dataset", status_code=404) + self.assertNotContains(response, "hx-confirm", status_code=404) class DatasetResourceManagementTests(TestCase): @@ -476,8 +481,13 @@ def test_manage_close_swaps_only_its_own_card(self): self.assertNotContains(response, 'hx-target="#datasets-container"') def test_manage_view_creator_only(self): + # through the caller's own dashboard, so the creator check answers self.client.force_login(self.other_user) - response = self.client.get(self.manage_url) + response = self.client.get( + reverse( + "login:dataset-manage", args=[self.other_user.id, "managed_dataset"] + ) + ) self.assertEqual(response.status_code, 403) def test_manage_lists_resources_with_links_and_status_badges(self): @@ -557,9 +567,15 @@ def test_assign_foreign_draft_forbidden(self): def test_assign_forbidden_for_non_creator(self): self.make_table("t_free_for_all", published=True) + # through the caller's own dashboard, so the creator check answers self.client.force_login(self.other_user) - response = self.client.post(self.assign_url, {"table": "t_free_for_all"}) + response = self.client.post( + reverse( + "login:dataset-assign", args=[self.other_user.id, "managed_dataset"] + ), + {"table": "t_free_for_all"}, + ) self.assertEqual(response.status_code, 403) self.assertFalse(self.dataset.tables.filter(name="t_free_for_all").exists()) diff --git a/login/tests/test_profile_owner_rule.py b/login/tests/test_profile_owner_rule.py new file mode 100644 index 000000000..4b652c07a --- /dev/null +++ b/login/tests/test_profile_owner_rule.py @@ -0,0 +1,323 @@ +""" +SPDX-FileCopyrightText: 2026 Jonas Huber © Reiner Lemoine Institut +SPDX-License-Identifier: AGPL-3.0-or-later + +Every route under ``profile//`` answers its owner only. + +The rule lives in ``login/access.py``. A logged-in caller with another id gets +404 (the same for an id that exists and one that does not, so the answer says +nothing about which accounts exist), an anonymous caller is sent to the login +page, or gets a bare 401 when the request came from htmx. There is no +exemption for platform admins. +""" # noqa: 501 + +import re + +from django.conf import settings +from django.shortcuts import resolve_url +from django.test import TestCase +from django.urls import get_resolver, reverse + +from base.tests import get_urlpattern_params, recursively_get_patterns +from dataedit.models import Dataset, PeerReview, PeerReviewManager, Table +from login.models import ADMIN_PERM, UserPermission, myuser + +HTMX = {"HTTP_HX_REQUEST": "true"} + + +def without_csrf(response) -> str: + """The body with its per-render CSRF tokens blanked out.""" + return re.sub( + r'(csrfmiddlewaretoken" value=|csrfToken = )"[^"]*"', + r"\1", + response.content.decode(), + ) + + +def make_user(name, **extra): + user, _ = myuser.objects.get_or_create( + name=name, + email=f"{name.lower()}@test.com", + did_agree=True, + is_mail_verified=True, + **extra, + ) + return user + + +class OwnerRuleFixture(TestCase): + """Owner A holds a draft and a published table; B holds nothing of A's.""" + + owner_table_draft = "owner_rule_draft_marker" + owner_table_published = "owner_rule_published_marker" + + @classmethod + def setUpTestData(cls): + cls.owner = make_user("OwnerRuleA") + cls.foreign = make_user("OwnerRuleB") + cls.platform_admin = make_user("OwnerRuleAdmin", is_admin=True) + for name, published in ( + (cls.owner_table_draft, False), + (cls.owner_table_published, True), + ): + table = Table.objects.create( + name=name, + is_publish=published, + oemetadata={"resources": [{"name": name}]}, + ) + UserPermission.objects.create( + holder=cls.owner, table=table, level=ADMIN_PERM + ) + + def assert_no_owner_tables(self, response): + body = response.content.decode() + self.assertNotIn(self.owner_table_draft, body) + self.assertNotIn(self.owner_table_published, body) + + +class TablesListIsTheOwnersOnlyTests(OwnerRuleFixture): + """The tables list, page and htmx partial, shows nothing to anyone else.""" + + queries = ( + {}, + {"search": ""}, + {"search": "owner_rule"}, + {"draft_page": "1"}, + {"published_page": "1"}, + ) + + def url(self): + return reverse("login:tables", kwargs={"user_id": self.owner.pk}) + + def test_owner_sees_own_tables_in_the_partial(self): + self.client.force_login(self.owner) + response = self.client.get(self.url(), **HTMX) + self.assertEqual(response.status_code, 200) + self.assertContains(response, self.owner_table_draft) + + def test_foreign_caller_gets_404_and_no_table_names(self): + self.client.force_login(self.foreign) + for query in self.queries: + for headers in ({}, HTMX): + with self.subTest(query=query, htmx=bool(headers)): + response = self.client.get(self.url(), query, **headers) + self.assertEqual(response.status_code, 404) + self.assert_no_owner_tables(response) + + def test_anonymous_caller_gets_no_table_names(self): + for query in self.queries: + for headers, status in (({}, 302), (HTMX, 401)): + with self.subTest(query=query, htmx=bool(headers)): + response = self.client.get(self.url(), query, **headers) + self.assertEqual(response.status_code, status) + self.assert_no_owner_tables(response) + + def test_platform_admin_is_not_exempt(self): + self.client.force_login(self.platform_admin) + response = self.client.get(self.url(), **HTMX) + self.assertEqual(response.status_code, 404) + self.assert_no_owner_tables(response) + + +def carries_owner_rule(callback) -> bool: + view_class = getattr(callback, "view_class", None) + if view_class is not None: + return getattr(view_class, "owner_rule", False) is True + return getattr(callback, "owner_rule", False) is True + + +class EveryProfileRouteCarriesTheRuleTest(TestCase): + """Structural: a new ``profile//`` route without the rule fails here. + + The same pattern as the OEKG API's ``AllowlistTest``: joining the profile + has to be a decision somebody made, not something a route does by + accident of where it was mounted. + """ + + def test_every_user_id_route_carries_the_owner_rule(self): + routes = [ + pattern + for pattern in recursively_get_patterns(get_resolver("login.urls")) + if "user_id" in get_urlpattern_params(pattern) + ] + # a guard against the walk silently finding nothing + self.assertIn("tables", {p.name for p in routes}) + missing = [p.name for p in routes if not carries_owner_rule(p.callback)] + self.assertEqual(missing, [], "profile routes without the owner rule") + + +class ProfileRoutesTests(OwnerRuleFixture): + """owner / foreign / anonymous x page / htmx, for every profile route.""" + + dataset_name = "owner_rule_dataset" + + @classmethod + def setUpTestData(cls): + super().setUpTestData() + Dataset.objects.create( + name=cls.dataset_name, + metadata={"name": cls.dataset_name, "title": "t", "description": ""}, + creator=cls.owner, + ) + + def routes(self): + """(name, extra kwargs, status for the owner) for every GET route.""" + dataset = {"dataset_name": self.dataset_name} + return [ + ("login:profile", {}, 200), + ("login:datasets", {}, 200), + ("login:dataset-card", dataset, 200), + ("login:dataset-edit", dataset, 200), + ("login:dataset-manage", dataset, 200), + ("login:dataset-table-search", dataset, 200), + ("login:tables", {}, 200), + ( + "login:metadata-review-badge-icon", + {"table_name": self.owner_table_draft}, + 200, + ), + ("login:reviews", {}, 200), + ("login:organizations", {}, 200), + ("login:partial-organizations", {}, 200), + ("login:settings", {}, 200), + ("login:edit", {}, 200), + # account deletion is not offered yet; the route answers 404, + # to its owner too, instead of the 500 it used to raise + ("login:account-delete", {}, 404), + ] + + post_routes = ( + "login:datasets", + "login:dataset-edit", + "login:dataset-delete", + "login:dataset-assign", + "login:dataset-unassign", + "login:edit", + ) + + def url(self, name, user_id, extra): + return reverse(name, kwargs={"user_id": user_id, **extra}) + + def test_owner_is_served(self): + self.client.force_login(self.owner) + for name, extra, status in self.routes(): + for headers in ({}, HTMX): + with self.subTest(route=name, htmx=bool(headers)): + response = self.client.get( + self.url(name, self.owner.pk, extra), **headers + ) + self.assertEqual(response.status_code, status) + + def test_foreign_caller_gets_404(self): + self.client.force_login(self.foreign) + for name, extra, _ in self.routes(): + for headers in ({}, HTMX): + with self.subTest(route=name, htmx=bool(headers)): + response = self.client.get( + self.url(name, self.owner.pk, extra), **headers + ) + self.assertEqual(response.status_code, 404) + + def test_anonymous_page_is_sent_to_login_with_next(self): + login_url = resolve_url(settings.LOGIN_URL) + for name, extra, _ in self.routes(): + with self.subTest(route=name): + url = self.url(name, self.owner.pk, extra) + response = self.client.get(url) + self.assertEqual(response.status_code, 302) + location = response["Location"] + self.assertTrue(location.startswith(login_url), location) + self.assertIn("next=", location) + + def test_anonymous_htmx_gets_401_without_body(self): + for name, extra, _ in self.routes(): + with self.subTest(route=name): + response = self.client.get(self.url(name, self.owner.pk, extra), **HTMX) + self.assertEqual(response.status_code, 401) + self.assertEqual(response.content, b"") + + def test_unknown_id_answers_like_a_foreign_one(self): + """No answer tells an existing account from a missing one.""" + self.client.force_login(self.foreign) + missing_id = myuser.objects.order_by("-pk").first().pk + 1000 + for name, extra, _ in self.routes(): + with self.subTest(route=name): + existing = self.client.get(self.url(name, self.owner.pk, extra)) + missing = self.client.get(self.url(name, missing_id, extra)) + self.assertEqual(existing.status_code, 404) + self.assertEqual(missing.status_code, 404) + self.assertEqual(without_csrf(existing), without_csrf(missing)) + + def test_foreign_post_is_refused_before_anything_else(self): + self.client.force_login(self.foreign) + for name in self.post_routes: + extra = ( + {} + if name in ("login:datasets", "login:edit") + else {"dataset_name": self.dataset_name} + ) + with self.subTest(route=name): + response = self.client.post( + self.url(name, self.owner.pk, extra), + {"title": "Hijacked", "table": self.owner_table_published}, + **HTMX, + ) + self.assertEqual(response.status_code, 404) + + def test_anonymous_post_is_refused(self): + for name in self.post_routes: + extra = ( + {} + if name in ("login:datasets", "login:edit") + else {"dataset_name": self.dataset_name} + ) + with self.subTest(route=name): + url = self.url(name, self.owner.pk, extra) + self.assertEqual(self.client.post(url, {}).status_code, 302) + self.assertEqual(self.client.post(url, {}, **HTMX).status_code, 401) + + +class RefusedGetsWriteNothingTests(OwnerRuleFixture): + """Two profile GETs write; a refused caller must not reach either write.""" + + callers = ("foreign", "anonymous") + + def act_as(self, caller): + self.client.logout() + if caller == "foreign": + self.client.force_login(self.foreign) + + def test_refused_settings_get_creates_no_token(self): + from rest_framework.authtoken.models import Token + + Token.objects.filter(user=self.owner).delete() + url = reverse("login:settings", kwargs={"user_id": self.owner.pk}) + for caller in self.callers: + for headers in ({}, HTMX): + with self.subTest(caller=caller, htmx=bool(headers)): + self.act_as(caller) + self.client.get(url, **headers) + self.assertFalse(Token.objects.filter(user=self.owner).exists()) + + def test_refused_reviews_get_leaves_the_review_manager_alone(self): + review = PeerReview.objects.create( + table=self.owner_table_published, + reviewer=self.owner, + contributor=self.foreign, + review={}, + ) + manager = PeerReviewManager.objects.create(opr=review, is_open_since="stale") + url = reverse("login:reviews", kwargs={"user_id": self.owner.pk}) + for caller in self.callers: + for headers in ({}, HTMX): + with self.subTest(caller=caller, htmx=bool(headers)): + self.act_as(caller) + self.client.get(url, **headers) + manager.refresh_from_db() + self.assertEqual(manager.is_open_since, "stale") + + # the same GET by the owner does write, so the check above can fail + self.client.force_login(self.owner) + self.client.get(url) + manager.refresh_from_db() + self.assertNotEqual(manager.is_open_since, "stale") diff --git a/login/tests/test_views.py b/login/tests/test_views.py index 706245904..f0628573b 100644 --- a/login/tests/test_views.py +++ b/login/tests/test_views.py @@ -30,19 +30,21 @@ def test_views(self): kwargs={"organization_id": organization_id}, logged_in=True, ) - self.get("login:organizations", kwargs={"user_id": user_id}) + self.get("login:organizations", kwargs={"user_id": user_id}, logged_in=True) self.get( "login:partial-organization-membership", kwargs={"organization_id": organization_id}, logged_in=True, ) - self.get("login:partial-organizations", kwargs={"user_id": user_id}) + self.get( + "login:partial-organizations", kwargs={"user_id": user_id}, logged_in=True + ) self.get("login:password_reset") self.get("login:password_reset_complete") self.get("login:password_reset_done") - self.get("login:profile", kwargs={"user_id": user_id}) + self.get("login:profile", kwargs={"user_id": user_id}, logged_in=True) self.get("login:redirect") self.get("login:reset-token", logged_in=True) - self.get("login:reviews", kwargs={"user_id": user_id}) - self.get("login:settings", kwargs={"user_id": user_id}) - self.get("login:tables", kwargs={"user_id": user_id}) + self.get("login:reviews", kwargs={"user_id": user_id}, logged_in=True) + self.get("login:settings", kwargs={"user_id": user_id}, logged_in=True) + self.get("login:tables", kwargs={"user_id": user_id}, logged_in=True) diff --git a/login/urls.py b/login/urls.py index 101689c49..05f7ae29d 100644 --- a/login/urls.py +++ b/login/urls.py @@ -18,7 +18,6 @@ ) from django.urls import path, re_path -from base.views import handler404 from login.views import ( DatasetsView, EditUserView, @@ -29,6 +28,7 @@ ReviewsView, SettingsView, TablesView, + account_delete_view, dataset_assign_view, dataset_card_view, dataset_delete_view, @@ -150,8 +150,7 @@ # TODO: implement tests before we allow user deletion re_path( r"^profile/(?P[\d]+)/delete_acc$", - # AccountDeleteView.as_view(), - handler404, + account_delete_view, name="account-delete", ), re_path( diff --git a/login/views.py b/login/views.py index 02a5f1a2e..b5902a1e8 100644 --- a/login/views.py +++ b/login/views.py @@ -26,6 +26,7 @@ from django.core.paginator import Paginator from django.db.models import F, Q from django.http import ( + Http404, HttpResponse, HttpResponseForbidden, HttpResponseNotAllowed, @@ -53,6 +54,7 @@ ) from dataedit.helper import delete_peer_review from dataedit.models import Dataset, PeerReviewManager, Table, Topic +from login.access import ProfileOwnerRequiredMixin, profile_owner_required from login.forms import EditUserForm, OrganizationForm from login.models import Membership, Organization from login.models import myuser as OepUser @@ -71,7 +73,7 @@ ########################################################################### -class TablesView(View): +class TablesView(ProfileOwnerRequiredMixin, View): def _get_filtered_tables(self, user, search_query=""): """Return filtered querysets for draft and published tables.""" @@ -96,7 +98,7 @@ def _get_filtered_tables(self, user, search_query=""): @method_decorator(never_cache) def get(self, request, user_id): - user = get_object_or_404(OepUser, pk=user_id) + user = self.profile_user search_query = request.GET.get("search", "").strip() has_search_param = "search" in request.GET @@ -179,41 +181,39 @@ def _serializer_errors(serializer): def dataset_creator_required(view_func): - """Resolve profile user and dataset for the dataset partial views and - enforce that only the dataset's creator may act (403 otherwise).""" + """Resolve the dataset for the dataset partial views and enforce that + only the dataset's creator may act (403 otherwise). + + Stacks under ``profile_owner_required``, which has already settled that + ``profile_user`` is the caller.""" @wraps(view_func) - def wrapper(request, user_id, dataset_name, *args, **kwargs): + def wrapper(request, profile_user, dataset_name, *args, **kwargs): dataset = get_object_or_404(Dataset, name=dataset_name) if dataset.creator is None or dataset.creator != request.user: return HttpResponseForbidden( "Only the dataset creator may manage this dataset." ) - profile_user = get_object_or_404(OepUser, pk=user_id) return view_func(request, profile_user, dataset, *args, **kwargs) return wrapper -class DatasetsView(LoginRequiredMixin, View): +class DatasetsView(ProfileOwnerRequiredMixin, View): """Dataset-first dashboard view: list the user's datasets and create new ones via HTMX without page reloads. The name is immutable after creation; title and description stay editable.""" @method_decorator(never_cache) def get(self, request, user_id): - user = get_object_or_404(OepUser, pk=user_id) + user = self.profile_user context = _datasets_context(request, user) if "HX-Request" in request.headers: return render(request, "login/partials/datasets_sections.html", context) return render(request, "login/user_datasets.html", context) def post(self, request, user_id): - user = get_object_or_404(OepUser, pk=user_id) - if user != request.user: - return HttpResponseForbidden( - "Datasets can only be created on your own dashboard." - ) + user = self.profile_user # the permanent URL name is derived from the title, so users can # style the title freely without thinking in slugs @@ -245,7 +245,7 @@ def post(self, request, user_id): return render(request, "login/partials/datasets_sections.html", context) -@login_required +@profile_owner_required @dataset_creator_required def dataset_edit_view(request, profile_user, dataset): """Inline edit of a dataset card: title, description and topics; the @@ -288,7 +288,7 @@ def dataset_edit_view(request, profile_user, dataset): ) -@login_required +@profile_owner_required @dataset_creator_required def dataset_card_view(request, profile_user, dataset): """A single dataset card, used to close an open edit or manage panel @@ -301,7 +301,7 @@ def dataset_card_view(request, profile_user, dataset): ) -@login_required +@profile_owner_required @require_POST @dataset_creator_required def dataset_delete_view(request, profile_user, dataset): @@ -338,7 +338,7 @@ def _render_dataset_manage(request, profile_user, dataset, search=""): return render(request, "login/partials/dataset_manage.html", context) -@login_required +@profile_owner_required @dataset_creator_required def dataset_manage_view(request, profile_user, dataset): """Manage panel for a dataset's resources: current tables with draft @@ -346,7 +346,7 @@ def dataset_manage_view(request, profile_user, dataset): return _render_dataset_manage(request, profile_user, dataset) -@login_required +@profile_owner_required @dataset_creator_required def dataset_table_search_view(request, profile_user, dataset): """Picker search: only tables the user may assign under the curation @@ -360,7 +360,7 @@ def dataset_table_search_view(request, profile_user, dataset): return render(request, "login/partials/dataset_table_search_results.html", context) -@login_required +@profile_owner_required @require_POST @dataset_creator_required def dataset_assign_view(request, profile_user, dataset): @@ -373,7 +373,7 @@ def dataset_assign_view(request, profile_user, dataset): return _render_dataset_manage(request, profile_user, dataset) -@login_required +@profile_owner_required @require_POST @dataset_creator_required def dataset_unassign_view(request, profile_user, dataset): @@ -388,7 +388,7 @@ def dataset_unassign_view(request, profile_user, dataset): ############################################################################## -class ReviewsView(View): +class ReviewsView(ProfileOwnerRequiredMixin, View): @method_decorator(never_cache) def get(self, request, user_id): """ @@ -398,7 +398,7 @@ def get(self, request, user_id): :param user_id: An user id :return: Profile renderer """ - user = get_object_or_404(OepUser, pk=user_id) + user = self.profile_user ################################################################## # get reviewer pov reviews @@ -581,7 +581,7 @@ def delete_peer_review_simple_view(request): return delete_peer_review(review_id, request.user) -class SettingsView(View): +class SettingsView(ProfileOwnerRequiredMixin, View): @method_decorator(never_cache) def get(self, request, user_id): """ @@ -596,12 +596,9 @@ def get(self, request, user_id): for user in OepUser.objects.all(): Token.objects.get_or_create(user=user) - user = get_object_or_404(OepUser, pk=user_id) - token = None - user_organizations = None - if request.user.is_authenticated: - token = Token.objects.get(user=request.user) - user_organizations = request.user.memberships + user = self.profile_user + token = Token.objects.get(user=request.user) + user_organizations = request.user.memberships return render( request, "login/user_settings.html", @@ -614,7 +611,7 @@ def get(self, request, user_id): ########################################################################### -class OrganizationsView(View): +class OrganizationsView(ProfileOwnerRequiredMixin, View): @method_decorator(never_cache) def get(self, request, user_id: int): """ @@ -628,7 +625,7 @@ def get(self, request, user_id: int): :return: Profile renderer """ - user = get_object_or_404(OepUser, pk=user_id) + user = self.profile_user return render( request, @@ -688,7 +685,7 @@ def organization_delete_view(request, organization_id: int): return response -class OrganizationListView(View): +class OrganizationListView(ProfileOwnerRequiredMixin, View): @method_decorator(never_cache) def get(self, request, user_id: int): """ @@ -697,7 +694,7 @@ def get(self, request, user_id: int): :param user_id: An user id :return: Profile renderer """ - user = get_object_or_404(OepUser, pk=user_id) + user = self.profile_user return render( request, @@ -940,17 +937,13 @@ def post(self, request, organization_id: int): ############################################################################## -class EditUserView(View): +class EditUserView(ProfileOwnerRequiredMixin, View): @method_decorator(never_cache) def get(self, request, user_id): - if not request.user.id == int(user_id): - raise PermissionDenied form = EditUserForm(instance=request.user) return render(request, "login/oepuser_edit_form.html", {"form": form}) def post(self, request, user_id): - if not request.user.id == int(user_id): - raise PermissionDenied form = EditUserForm( instance=request.user, files=request.FILES or None, @@ -988,6 +981,15 @@ def get(self, request, user_id): return render(request, "login/delete_account.html", {"profile_user": user}) +@profile_owner_required +def account_delete_view(request, profile_user): + """Account deletion is not offered yet (see AccountDeleteView_TODO_UNUSED). + + The route exists so its link resolves; it answers 404, to its owner too. + """ + raise Http404 + + # TODO: should be require_POST? def token_reset_view(request): if request.user.is_authenticated: @@ -1003,8 +1005,9 @@ def token_reset_view(request): return HttpResponseForbidden("You are not authorized to reset the token.") +@profile_owner_required @never_cache -def metadata_review_badge_indicator_icon_file_view(request, user_id, table_name): +def metadata_review_badge_indicator_icon_file_view(request, profile_user, table_name): # is_badge : bool , msg : string -> either error msg or badge name table = get_object_or_404(Table, name=table_name) context = table.get_review_badge_from_table_metadata() From abf14d9c04292d780117972c4e0697019f9dfb4a Mon Sep 17 00:00:00 2001 From: jh-RLI Date: Thu, 1 Oct 2026 13:37:15 +0200 Subject: [PATCH 2/7] Check organization rights before writing The organization views put the login mixin after the view class, where it never ran. It now comes first, so an anonymous request goes to the login page before any lookup or write. Editing an organization checks the caller's membership level before the form is saved, and creating one saves the organization and its first admin in one transaction. The member list answers members only, and removing a member happens only when none of the checks refused it. Redirects after create, leave and delete are reversed from the route instead of spelled out. Co-Authored-By: Claude Opus 5.5 (1M context) --- login/tests/test_organization_write_checks.py | 243 ++++++++++++++++++ login/views.py | 128 ++++----- 2 files changed, 309 insertions(+), 62 deletions(-) create mode 100644 login/tests/test_organization_write_checks.py diff --git a/login/tests/test_organization_write_checks.py b/login/tests/test_organization_write_checks.py new file mode 100644 index 000000000..5c3f7fe60 --- /dev/null +++ b/login/tests/test_organization_write_checks.py @@ -0,0 +1,243 @@ +""" +SPDX-FileCopyrightText: 2026 Jonas Huber © Reiner Lemoine Institut +SPDX-License-Identifier: AGPL-3.0-or-later + +Organization writes are checked before they happen. + +Every organization view requires a login, the caller's membership level is +checked before the form is saved or a membership is changed, and a refused +write leaves the database as it was. Refusal codes follow the views' own +convention: no membership is 404, a membership below the level an action +needs is 403, no login is the login redirect. +""" # noqa: 501 + +from django.test import TestCase +from django.urls import reverse + +from login.models import ( + ADMIN_PERM, + DELETE_PERM, + WRITE_PERM, + Membership, + Organization, + myuser, +) + +HTMX = {"HTTP_HX_REQUEST": "true"} + + +def make_user(name): + user, _ = myuser.objects.get_or_create( + name=name, + email=f"{name.lower()}@test.com", + did_agree=True, + is_mail_verified=True, + ) + return user + + +class OrganizationFixture(TestCase): + """One organization: an admin, a second admin, a Remove-level and an + Invite-level member. The stranger is in no organization.""" + + @classmethod + def setUpTestData(cls): + cls.admin = make_user("OrgCheckAdmin") + cls.second_admin = make_user("OrgCheckSecondAdmin") + cls.remover = make_user("OrgCheckRemover") + cls.inviter = make_user("OrgCheckInviter") + cls.stranger = make_user("OrgCheckStranger") + cls.newcomer = make_user("OrgCheckNewcomer") + cls.organization = Organization.objects.create( + name="org_check_original", description="original description" + ) + for user, level in ( + (cls.admin, ADMIN_PERM), + (cls.second_admin, ADMIN_PERM), + (cls.remover, DELETE_PERM), + (cls.inviter, WRITE_PERM), + ): + Membership.objects.create(user=user, group=cls.organization, level=level) + + def level_of(self, user): + membership = Membership.objects.filter( + group=self.organization, user=user + ).first() + return membership.level if membership else None + + def act_as(self, user): + self.client.logout() + if user is not None: + self.client.force_login(user) + + +class RenameIsCheckedBeforeSavingTests(OrganizationFixture): + def edit_url(self): + return reverse( + "login:organization-edit", + kwargs={"organization_id": self.organization.pk}, + ) + + def rename_as(self, user): + self.act_as(user) + return self.client.post( + self.edit_url(), + {"name": "org_check_renamed", "description": "renamed"}, + **HTMX, + ) + + def assert_unchanged(self): + self.organization.refresh_from_db() + self.assertEqual(self.organization.name, "org_check_original") + self.assertEqual(self.organization.description, "original description") + + def test_admin_renames(self): + response = self.rename_as(self.admin) + self.assertEqual(response.status_code, 200) + self.organization.refresh_from_db() + self.assertEqual(self.organization.name, "org_check_renamed") + + def test_non_member_is_refused_and_nothing_changes(self): + response = self.rename_as(self.stranger) + self.assertEqual(response.status_code, 404) + self.assert_unchanged() + + def test_member_below_admin_is_refused_and_nothing_changes(self): + for member in (self.remover, self.inviter): + with self.subTest(member=member.name): + response = self.rename_as(member) + self.assertEqual(response.status_code, 403) + self.assert_unchanged() + + def test_anonymous_is_sent_to_login_and_nothing_changes(self): + response = self.rename_as(None) + self.assertEqual(response.status_code, 302) + self.assertIn("login", response["Location"]) + self.assert_unchanged() + + def test_unknown_organization_is_404(self): + self.act_as(self.admin) + response = self.client.post( + reverse("login:organization-edit", kwargs={"organization_id": 999999}), + {"name": "org_check_renamed", "description": "renamed"}, + **HTMX, + ) + self.assertEqual(response.status_code, 404) + + +class CreateAlwaysHasAnOwnerTests(OrganizationFixture): + url = reverse("login:organization-create") + + def test_anonymous_create_leaves_no_organization(self): + before = Organization.objects.count() + for headers in ({}, HTMX): + with self.subTest(htmx=bool(headers)): + response = self.client.post( + self.url, + {"name": "org_check_ownerless", "description": "d"}, + **headers, + ) + self.assertEqual(response.status_code, 302) + self.assertEqual(Organization.objects.count(), before) + self.assertFalse( + Organization.objects.filter(name="org_check_ownerless").exists() + ) + + def test_logged_in_create_makes_the_caller_its_admin(self): + self.act_as(self.stranger) + response = self.client.post( + self.url, {"name": "org_check_new", "description": "d"}, **HTMX + ) + organization = Organization.objects.get(name="org_check_new") + self.assertEqual( + Membership.objects.get(group=organization, user=self.stranger).level, + ADMIN_PERM, + ) + # the redirect goes to the caller's own organizations page, which + # the owner rule lets them open + own_page = reverse("login:organizations", kwargs={"user_id": self.stranger.pk}) + self.assertEqual(response["HX-Redirect"], own_page) + self.assertEqual(self.client.get(own_page).status_code, 200) + + +class MemberChangesAreCheckedFirstTests(OrganizationFixture): + def members_url(self): + return reverse( + "login:partial-organization-membership", + kwargs={"organization_id": self.organization.pk}, + ) + + def post_as(self, user, data): + self.act_as(user) + return self.client.post(self.members_url(), data, **HTMX) + + def test_member_list_needs_a_membership(self): + self.act_as(self.inviter) + self.assertContains( + self.client.get(self.members_url(), **HTMX), "OrgCheckAdmin" + ) + + self.act_as(self.stranger) + response = self.client.get(self.members_url(), **HTMX) + self.assertEqual(response.status_code, 404) + self.assertNotIn("OrgCheckAdmin", response.content.decode()) + + self.act_as(None) + response = self.client.get(self.members_url(), **HTMX) + self.assertEqual(response.status_code, 302) + + def test_add_user_refused_for_non_member_and_anonymous(self): + for user, status in ((self.stranger, 404), (None, 302)): + with self.subTest(user=getattr(user, "name", "anonymous")): + response = self.post_as( + user, {"mode": "add_user", "name": self.newcomer.name} + ) + self.assertEqual(response.status_code, status) + self.assertIsNone(self.level_of(self.newcomer)) + + def test_add_user_by_an_inviter(self): + response = self.post_as( + self.inviter, {"mode": "add_user", "name": self.newcomer.name} + ) + self.assertEqual(response.status_code, 200) + self.assertIsNotNone(self.level_of(self.newcomer)) + + def test_remove_refused_below_remove_level(self): + response = self.post_as( + self.inviter, {"mode": "remove_user", "user_id": self.remover.pk} + ) + self.assertEqual(response.status_code, 403) + self.assertEqual(self.level_of(self.remover), DELETE_PERM) + + def test_remove_of_a_higher_level_is_refused_and_kept(self): + response = self.post_as( + self.remover, {"mode": "remove_user", "user_id": self.admin.pk} + ) + self.assertContains(response, "higher permission level") + self.assertEqual(self.level_of(self.admin), ADMIN_PERM) + + def test_removing_yourself_is_refused_and_kept(self): + response = self.post_as( + self.remover, {"mode": "remove_user", "user_id": self.remover.pk} + ) + self.assertContains(response, "leave the organization") + self.assertEqual(self.level_of(self.remover), DELETE_PERM) + + def test_remove_at_or_below_own_level(self): + response = self.post_as( + self.remover, {"mode": "remove_user", "user_id": self.inviter.pk} + ) + self.assertEqual(response.status_code, 200) + self.assertIsNone(self.level_of(self.inviter)) + + def test_level_change_refused_below_admin(self): + response = self.post_as( + self.remover, + { + "mode": "alter_user", + "user_id": self.inviter.pk, + "selected_value": ADMIN_PERM, + }, + ) + self.assertEqual(response.status_code, 403) + self.assertEqual(self.level_of(self.inviter), WRITE_PERM) diff --git a/login/views.py b/login/views.py index b5902a1e8..940d4c834 100644 --- a/login/views.py +++ b/login/views.py @@ -24,6 +24,7 @@ from django.contrib.auth.mixins import LoginRequiredMixin from django.core.exceptions import PermissionDenied from django.core.paginator import Paginator +from django.db import transaction from django.db.models import F, Q from django.http import ( Http404, @@ -662,7 +663,9 @@ def organization_leave_view(request, organization_id: int): membership.delete() response = HttpResponse() - response["HX-Redirect"] = f"/user/profile/{user_id}/organizations" + response["HX-Redirect"] = reverse( + "login:organizations", kwargs={"user_id": user_id} + ) return response @@ -681,7 +684,9 @@ def organization_delete_view(request, organization_id: int): extra_tags="primary", ) response = HttpResponse() - response["HX-Redirect"] = f"/user/profile/{request.user.id}/organizations" + response["HX-Redirect"] = reverse( + "login:organizations", kwargs={"user_id": request.user.id} + ) return response @@ -703,8 +708,12 @@ def get(self, request, user_id: int): ) -class OrganizationManagementView(View, LoginRequiredMixin): - form_is_valid = False +class OrganizationManagementView(LoginRequiredMixin, View): + """Create an organization, or edit one the caller administers. + + The login mixin comes first in the bases: ``View.dispatch`` does not call + further along the MRO, so a mixin after the view class never runs. + """ @method_decorator(never_cache) def get(self, request, organization_id=None): @@ -720,7 +729,7 @@ def get(self, request, organization_id=None): can_edit = False organization = None if organization_id: - organization = Organization.objects.get(id=organization_id) + organization = get_object_or_404(Organization, id=organization_id) membership = get_object_or_404( Membership, group=organization, user=request.user ) @@ -777,60 +786,56 @@ def post(self, request, organization_id=None): :param organization_id: An organization id :return: Profile renderer """ - self.form_is_valid = False - user = request.user.id - organization = ( - Organization.objects.get(id=organization_id) if organization_id else None - ) - form = OrganizationForm(request.POST, instance=organization) - status = None - if form.is_valid(): - self.form_is_valid = True + organization = None + if organization_id: + # who may edit is settled before the form touches the instance + organization = get_object_or_404(Organization, id=organization_id) + membership = get_object_or_404( + Membership, group=organization, user=request.user + ) + if membership.level < ADMIN_PERM: + raise PermissionDenied - if not self.form_is_valid: + form = OrganizationForm(request.POST, instance=organization) + if not form.is_valid(): return render( request, "login/partials/organization_form.html", {"form": form}, ) - if self.form_is_valid: - # status = 201 - if organization_id: - organization = form.save() - membership = get_object_or_404( - Membership, group=organization, user=request.user - ) - if membership.level < ADMIN_PERM: - raise PermissionDenied - return render( - request, - "login/partials/organization_form.html", - {"form": form, "organization": organization}, - status=status, - ) - else: - organization = form.save() - membership = Membership.objects.create( - user=request.user, group=organization, level=ADMIN_PERM - ) - membership.save() - messages.add_message( - request, - level=messages.INFO, - message=( - "Organization created! " - "Edit the organization to invite members." - ), - extra_tags="primary", - ) - response = HttpResponse() - # response["profile_user"] = user - response["HX-Redirect"] = f"/user/profile/{user}/organizations" - return response + if organization_id: + organization = form.save() + return render( + request, + "login/partials/organization_form.html", + {"form": form, "organization": organization}, + ) + # a new organization and its first admin are one write, so an + # organization never exists without an owner + with transaction.atomic(): + organization = form.save() + Membership.objects.create( + user=request.user, group=organization, level=ADMIN_PERM + ) + messages.add_message( + request, + level=messages.INFO, + message="Organization created! Edit the organization to invite members.", + extra_tags="primary", + ) + response = HttpResponse() + response["HX-Redirect"] = reverse( + "login:organizations", kwargs={"user_id": request.user.pk} + ) + return response + + +class OrganizationMembersView(LoginRequiredMixin, TemplateView): + """The member list of an organization, for its members only, and the + member changes their level allows. Login mixin first, as above.""" -class OrganizationMembersView(TemplateView, LoginRequiredMixin): template_name = "login/partials/organization_members.html" def get_context_data(self, **kwargs): @@ -840,12 +845,10 @@ def get_context_data(self, **kwargs): organization = get_object_or_404( Organization, pk=self.kwargs["organization_id"] ) - is_admin = False - membership = Membership.objects.filter( - group=organization, user=self.request.user - ).first() - if membership: - is_admin = membership.level >= ADMIN_PERM + membership = get_object_or_404( + Membership, group=organization, user=self.request.user + ) + is_admin = membership.level >= ADMIN_PERM context["organization"] = organization context["choices"] = Membership.choices @@ -899,7 +902,10 @@ def post(self, request, organization_id: int): error_message = ( "Please leave the organization to remove your own membership." ) - + elif membership.level < target_membership.level: + error_message = ( + "You cant remove memberships with higher permission level." + ) elif target_membership.level >= ADMIN_PERM: admins = ( Membership.objects.filter(group=organization, level=ADMIN_PERM) @@ -908,12 +914,10 @@ def post(self, request, organization_id: int): ) if admins == 0: error_message = "A organization needs at least one admin" - elif membership.level < target_membership.level: - error_message = ( - "You cant remove memberships with higher permission level." - ) - target_membership.delete() + # a refusal above is a refusal: nothing is removed + if error_message is None: + target_membership.delete() elif mode == "alter_user": if membership.level < login.permissions.ADMIN_PERM: From 21272e0221e633056b21c34081f95a559242ae1f Mon Sep 17 00:00:00 2001 From: jh-RLI Date: Thu, 1 Oct 2026 13:40:55 +0200 Subject: [PATCH 3/7] Add changelog entry for #2547 Co-Authored-By: Claude Opus 5.5 (1M context) --- versions/changelogs/current.md | 3 +++ 1 file changed, 3 insertions(+) diff --git a/versions/changelogs/current.md b/versions/changelogs/current.md index 144641db6..31f0f3ad5 100644 --- a/versions/changelogs/current.md +++ b/versions/changelogs/current.md @@ -23,6 +23,9 @@ SPDX-License-Identifier: CC0-1.0 - Deleting a peer review now checks the caller: only its reviewer or a platform admin may delete it [(#2544)](https://github.com/OpenEnergyPlatform/oeplatform/pull/2544) +- Restrict profile pages to their owner and check organization permissions + before saving + [(#2547)](https://github.com/OpenEnergyPlatform/oeplatform/pull/2547) ## Documentation updates From 394bdc81a476fbf0c8345c3ba26b0e76c20c2ffb Mon Sep 17 00:00:00 2001 From: jh-RLI Date: Thu, 1 Oct 2026 13:49:57 +0200 Subject: [PATCH 4/7] Check that the owner rule runs, not mentioned The structural test read an inherited marker, so a class listing the mixin after View passed while View.dispatch skipped the mixin, and so did a dispatch override or a function that copied the marker through functools.wraps. enforces_owner_rule in login/access.py now asks what runs: a class view must resolve dispatch to the mixin's own, a function view must be the decorator's wrapper itself (kept in a registry, which wraps cannot copy). The walk starts at the site's root and carries parameters captured by include() prefixes down to their children. Self-tests pin each rejected case and the include walk on a synthetic resolver. Co-Authored-By: Claude Opus 5.5 (1M context) --- login/access.py | 31 +++++- login/tests/test_profile_owner_rule.py | 131 ++++++++++++++++++++++--- 2 files changed, 142 insertions(+), 20 deletions(-) diff --git a/login/access.py b/login/access.py index af27aec53..3ee5825d6 100644 --- a/login/access.py +++ b/login/access.py @@ -29,11 +29,13 @@ that one holds the permission levels and is imported by ``login/models.py`` at app-loading time, before the auth views this module needs can be imported. -Both forms carry ``owner_rule = True``. ``login/tests/test_profile_owner_rule`` -walks the URL patterns and fails for a ``user_id`` route without it. +``enforces_owner_rule`` tells whether the rule actually runs for a URL +callback. ``login/tests/test_profile_owner_rule`` walks every URL pattern and +fails for a ``user_id`` route on which it does not. """ # noqa: 501 from functools import wraps +from weakref import WeakSet from django.contrib.auth.views import redirect_to_login from django.http import Http404, HttpResponse @@ -66,8 +68,6 @@ class ProfileOwnerRequiredMixin: On a match ``self.profile_user`` is the caller. """ - owner_rule = True - def dispatch(self, request, *args, **kwargs): refusal = _refusal(request, kwargs.get("user_id")) if refusal is not None: @@ -90,5 +90,26 @@ def wrapper(request, user_id, *args, **kwargs): return refusal return view_func(request, request.user, *args, **kwargs) - wrapper.owner_rule = True + _GUARDED_FUNCTIONS.add(wrapper) return wrapper + + +# The wrappers profile_owner_required made. A registry rather than an +# attribute, because functools.wraps copies attributes onto whatever wraps a +# view, so an attribute would also mark a function that never runs the rule. +_GUARDED_FUNCTIONS = WeakSet() + + +def enforces_owner_rule(view) -> bool: + """Whether the owner rule runs first for this URL callback. + + A class-based view must resolve ``dispatch`` to the mixin's own: that + rejects the mixin listed after ``View`` (``View.dispatch`` wins and never + calls along the MRO) and a ``dispatch`` override that could skip it. A + function view must be the decorator's wrapper itself, the outermost layer, + so nothing runs before the rule. + """ + view_class = getattr(view, "view_class", None) + if view_class is not None: + return view_class.dispatch is ProfileOwnerRequiredMixin.dispatch + return view in _GUARDED_FUNCTIONS diff --git a/login/tests/test_profile_owner_rule.py b/login/tests/test_profile_owner_rule.py index 4b652c07a..fc6961de2 100644 --- a/login/tests/test_profile_owner_rule.py +++ b/login/tests/test_profile_owner_rule.py @@ -12,14 +12,25 @@ """ # noqa: 501 import re +from functools import wraps from django.conf import settings +from django.contrib.auth.models import AnonymousUser +from django.http import HttpResponse from django.shortcuts import resolve_url -from django.test import TestCase -from django.urls import get_resolver, reverse +from django.test import RequestFactory, TestCase +from django.urls import URLResolver, get_resolver, re_path, reverse +from django.urls.resolvers import RegexPattern +from django.views.decorators.cache import never_cache +from django.views.generic import View -from base.tests import get_urlpattern_params, recursively_get_patterns +from base.tests import get_urlpattern_params from dataedit.models import Dataset, PeerReview, PeerReviewManager, Table +from login.access import ( + ProfileOwnerRequiredMixin, + enforces_owner_rule, + profile_owner_required, +) from login.models import ADMIN_PERM, UserPermission, myuser HTMX = {"HTTP_HX_REQUEST": "true"} @@ -119,11 +130,18 @@ def test_platform_admin_is_not_exempt(self): self.assert_no_owner_tables(response) -def carries_owner_rule(callback) -> bool: - view_class = getattr(callback, "view_class", None) - if view_class is not None: - return getattr(view_class, "owner_rule", False) is True - return getattr(callback, "owner_rule", False) is True +def routes_capturing(param, resolver, inherited=frozenset()): + """Every URL pattern below ``resolver`` whose address captures ``param``. + + A parameter captured by an ``include()`` prefix counts for every route + below it, so a ``user_id`` in a prefix covers all of its children. + """ + for entry in resolver.url_patterns: + captured = inherited | set(get_urlpattern_params(entry)) + if isinstance(entry, URLResolver): + yield from routes_capturing(param, entry, captured) + elif param in captured: + yield entry class EveryProfileRouteCarriesTheRuleTest(TestCase): @@ -131,21 +149,104 @@ class EveryProfileRouteCarriesTheRuleTest(TestCase): The same pattern as the OEKG API's ``AllowlistTest``: joining the profile has to be a decision somebody made, not something a route does by - accident of where it was mounted. + accident of where it was mounted. The walk starts at the site's root, so + a ``user_id`` route in any app counts, however it is included. """ def test_every_user_id_route_carries_the_owner_rule(self): - routes = [ - pattern - for pattern in recursively_get_patterns(get_resolver("login.urls")) - if "user_id" in get_urlpattern_params(pattern) - ] + routes = list(routes_capturing("user_id", get_resolver())) # a guard against the walk silently finding nothing self.assertIn("tables", {p.name for p in routes}) - missing = [p.name for p in routes if not carries_owner_rule(p.callback)] + missing = [p.name for p in routes if not enforces_owner_rule(p.callback)] self.assertEqual(missing, [], "profile routes without the owner rule") +def _plain_view(request, *args, **kwargs): + return HttpResponse("served") + + +class _MixinAfterView(View, ProfileOwnerRequiredMixin): + """Wrong order: View.dispatch runs and the mixin's never does.""" + + def get(self, request, user_id): + return HttpResponse("served") + + +class _DispatchOverridden(ProfileOwnerRequiredMixin, View): + """Right order, but a dispatch of its own that skips the rule.""" + + def dispatch(self, request, *args, **kwargs): + return View.dispatch(self, request, *args, **kwargs) + + def get(self, request, user_id): + return HttpResponse("served") + + +class _Guarded(ProfileOwnerRequiredMixin, View): + def get(self, request, user_id): + return HttpResponse("served") + + +@profile_owner_required +def _guarded_function(request, profile_user): + return HttpResponse("served") + + +@wraps(_guarded_function) +def _impostor(request, user_id): + """Looks like the guarded view (wraps copies its attributes) but is not.""" + return HttpResponse("served") + + +class OwnerRuleCheckSelfTests(TestCase): + """The structural check must fail for a view on which the rule does not + run, even when the view mentions the rule.""" + + def test_wrong_class_is_rejected_and_really_skips_the_rule(self): + self.assertFalse(enforces_owner_rule(_MixinAfterView.as_view())) + # the reason it must be rejected: the rule does not run + request = RequestFactory().get("/") + request.user = AnonymousUser() + response = _MixinAfterView.as_view()(request, user_id="1") + self.assertEqual(response.status_code, 200) + + def test_dispatch_override_is_rejected(self): + self.assertFalse(enforces_owner_rule(_DispatchOverridden.as_view())) + + def test_impostor_function_is_rejected(self): + self.assertFalse(enforces_owner_rule(_impostor)) + self.assertFalse(enforces_owner_rule(_plain_view)) + + def test_rule_wrapped_by_another_decorator_is_rejected(self): + # the rule must be the outermost layer, so nothing runs before it + self.assertFalse(enforces_owner_rule(never_cache(_guarded_function))) + + def test_guarded_views_are_accepted(self): + self.assertTrue(enforces_owner_rule(_Guarded.as_view())) + self.assertTrue(enforces_owner_rule(_guarded_function)) + + def test_walk_follows_include_prefixes(self): + """A ``user_id`` captured by an include prefix covers its children.""" + prefix = URLResolver( + RegexPattern(r"^profile/(?P\d+)/"), + [ + re_path(r"^unguarded$", _plain_view, name="unguarded"), + re_path(r"^guarded$", _guarded_function, name="guarded"), + ], + ) + root = URLResolver( + RegexPattern(r"^"), + [ + re_path(r"^outside$", _plain_view, name="outside"), + URLResolver(RegexPattern(r"^nested/"), [prefix]), + ], + ) + routes = {p.name: p for p in routes_capturing("user_id", root)} + self.assertEqual(set(routes), {"unguarded", "guarded"}) + self.assertFalse(enforces_owner_rule(routes["unguarded"].callback)) + self.assertTrue(enforces_owner_rule(routes["guarded"].callback)) + + class ProfileRoutesTests(OwnerRuleFixture): """owner / foreign / anonymous x page / htmx, for every profile route.""" From 5e7cfa8deed931cc060e68a791b534994c2a463d Mon Sep 17 00:00:00 2001 From: jh-RLI Date: Thu, 1 Oct 2026 13:51:29 +0200 Subject: [PATCH 5/7] Share access test helpers, pin redirects HTMX, make_user and act_as move into login/tests/helpers.py, used by both access test modules. base.tests.TestViewsTestCase does not fit: it creates one fixed user, and these tests need several in distinct roles. The anonymous redirect test now checks that next is the requested path, the POST routes are data instead of a repeated condition, and the leave and delete redirects are tested like create: HX-Redirect to the caller's own organizations page, followed with a user whose pk is not 1. Co-Authored-By: Claude Opus 5.5 (1M context) --- login/tests/helpers.py | 32 ++++++++ login/tests/test_organization_write_checks.py | 72 ++++++++++++------ login/tests/test_profile_owner_rule.py | 73 +++++++------------ 3 files changed, 108 insertions(+), 69 deletions(-) create mode 100644 login/tests/helpers.py diff --git a/login/tests/helpers.py b/login/tests/helpers.py new file mode 100644 index 000000000..8466893d0 --- /dev/null +++ b/login/tests/helpers.py @@ -0,0 +1,32 @@ +""" +SPDX-FileCopyrightText: 2026 Jonas Huber © Reiner Lemoine Institut +SPDX-License-Identifier: AGPL-3.0-or-later + +Small helpers shared by the profile and organization access tests. + +``base.tests.TestViewsTestCase`` is not used for these: it creates one fixed +user in ``setUpClass``, while every access test needs several users in +distinct roles (owner, stranger, admin, members at each level). +""" # noqa: 501 + +from login.models import myuser + +HTMX = {"HTTP_HX_REQUEST": "true"} + + +def make_user(name, **extra): + user, _ = myuser.objects.get_or_create( + name=name, + email=f"{name.lower()}@test.com", + did_agree=True, + is_mail_verified=True, + **extra, + ) + return user + + +def act_as(client, user): + """Log the test client in as ``user``, or leave it anonymous for None.""" + client.logout() + if user is not None: + client.force_login(user) diff --git a/login/tests/test_organization_write_checks.py b/login/tests/test_organization_write_checks.py index 5c3f7fe60..56ec45494 100644 --- a/login/tests/test_organization_write_checks.py +++ b/login/tests/test_organization_write_checks.py @@ -20,20 +20,8 @@ WRITE_PERM, Membership, Organization, - myuser, ) - -HTMX = {"HTTP_HX_REQUEST": "true"} - - -def make_user(name): - user, _ = myuser.objects.get_or_create( - name=name, - email=f"{name.lower()}@test.com", - did_agree=True, - is_mail_verified=True, - ) - return user +from login.tests.helpers import HTMX, act_as, make_user class OrganizationFixture(TestCase): @@ -66,9 +54,16 @@ def level_of(self, user): return membership.level if membership else None def act_as(self, user): - self.client.logout() - if user is not None: - self.client.force_login(user) + act_as(self.client, user) + + def own_organizations_page(self, user): + return reverse("login:organizations", kwargs={"user_id": user.pk}) + + def assert_redirects_to_own_page(self, response, user): + """HX-Redirect to the caller's own organizations page, which the + owner rule lets them open.""" + self.assertEqual(response["HX-Redirect"], self.own_organizations_page(user)) + self.assertEqual(self.client.get(response["HX-Redirect"]).status_code, 200) class RenameIsCheckedBeforeSavingTests(OrganizationFixture): @@ -153,11 +148,46 @@ def test_logged_in_create_makes_the_caller_its_admin(self): Membership.objects.get(group=organization, user=self.stranger).level, ADMIN_PERM, ) - # the redirect goes to the caller's own organizations page, which - # the owner rule lets them open - own_page = reverse("login:organizations", kwargs={"user_id": self.stranger.pk}) - self.assertEqual(response["HX-Redirect"], own_page) - self.assertEqual(self.client.get(own_page).status_code, 200) + self.assert_redirects_to_own_page(response, self.stranger) + + +class LeaveAndDeleteRedirectTests(OrganizationFixture): + """Leave and delete send the caller to their own organizations page. + + WF-09 found these redirects hard-coded to user 1's page, so the caller + used here is deliberately one whose pk is not 1. + """ + + def caller_other_than_user_one(self, *candidates): + caller = next(user for user in candidates if user.pk != 1) + self.assertNotEqual(caller.pk, 1) + return caller + + def test_leave_redirects_to_own_page(self): + caller = self.caller_other_than_user_one(self.inviter, self.remover) + self.act_as(caller) + response = self.client.post( + reverse( + "login:organization-leave", + kwargs={"organization_id": self.organization.pk}, + ), + **HTMX, + ) + self.assertIsNone(self.level_of(caller)) + self.assert_redirects_to_own_page(response, caller) + + def test_delete_redirects_to_own_page(self): + caller = self.caller_other_than_user_one(self.admin, self.second_admin) + self.act_as(caller) + response = self.client.post( + reverse( + "login:organization-delete", + kwargs={"organization_id": self.organization.pk}, + ), + **HTMX, + ) + self.assertFalse(Organization.objects.filter(pk=self.organization.pk).exists()) + self.assert_redirects_to_own_page(response, caller) class MemberChangesAreCheckedFirstTests(OrganizationFixture): diff --git a/login/tests/test_profile_owner_rule.py b/login/tests/test_profile_owner_rule.py index fc6961de2..2c8acebc2 100644 --- a/login/tests/test_profile_owner_rule.py +++ b/login/tests/test_profile_owner_rule.py @@ -13,6 +13,7 @@ import re from functools import wraps +from urllib.parse import parse_qs, urlsplit from django.conf import settings from django.contrib.auth.models import AnonymousUser @@ -23,6 +24,7 @@ from django.urls.resolvers import RegexPattern from django.views.decorators.cache import never_cache from django.views.generic import View +from rest_framework.authtoken.models import Token from base.tests import get_urlpattern_params from dataedit.models import Dataset, PeerReview, PeerReviewManager, Table @@ -32,8 +34,7 @@ profile_owner_required, ) from login.models import ADMIN_PERM, UserPermission, myuser - -HTMX = {"HTTP_HX_REQUEST": "true"} +from login.tests.helpers import HTMX, act_as, make_user def without_csrf(response) -> str: @@ -45,17 +46,6 @@ def without_csrf(response) -> str: ) -def make_user(name, **extra): - user, _ = myuser.objects.get_or_create( - name=name, - email=f"{name.lower()}@test.com", - did_agree=True, - is_mail_verified=True, - **extra, - ) - return user - - class OwnerRuleFixture(TestCase): """Owner A holds a draft and a published table; B holds nothing of A's.""" @@ -287,14 +277,17 @@ def routes(self): ("login:account-delete", {}, 404), ] - post_routes = ( - "login:datasets", - "login:dataset-edit", - "login:dataset-delete", - "login:dataset-assign", - "login:dataset-unassign", - "login:edit", - ) + def post_routes(self): + """(name, extra kwargs) for every route that takes a POST.""" + dataset = {"dataset_name": self.dataset_name} + return [ + ("login:datasets", {}), + ("login:dataset-edit", dataset), + ("login:dataset-delete", dataset), + ("login:dataset-assign", dataset), + ("login:dataset-unassign", dataset), + ("login:edit", {}), + ] def url(self, name, user_id, extra): return reverse(name, kwargs={"user_id": user_id, **extra}) @@ -326,9 +319,9 @@ def test_anonymous_page_is_sent_to_login_with_next(self): url = self.url(name, self.owner.pk, extra) response = self.client.get(url) self.assertEqual(response.status_code, 302) - location = response["Location"] - self.assertTrue(location.startswith(login_url), location) - self.assertIn("next=", location) + location = urlsplit(response["Location"]) + self.assertEqual(location.path, login_url) + self.assertEqual(parse_qs(location.query)["next"], [url]) def test_anonymous_htmx_gets_401_without_body(self): for name, extra, _ in self.routes(): @@ -351,12 +344,7 @@ def test_unknown_id_answers_like_a_foreign_one(self): def test_foreign_post_is_refused_before_anything_else(self): self.client.force_login(self.foreign) - for name in self.post_routes: - extra = ( - {} - if name in ("login:datasets", "login:edit") - else {"dataset_name": self.dataset_name} - ) + for name, extra in self.post_routes(): with self.subTest(route=name): response = self.client.post( self.url(name, self.owner.pk, extra), @@ -366,12 +354,7 @@ def test_foreign_post_is_refused_before_anything_else(self): self.assertEqual(response.status_code, 404) def test_anonymous_post_is_refused(self): - for name in self.post_routes: - extra = ( - {} - if name in ("login:datasets", "login:edit") - else {"dataset_name": self.dataset_name} - ) + for name, extra in self.post_routes(): with self.subTest(route=name): url = self.url(name, self.owner.pk, extra) self.assertEqual(self.client.post(url, {}).status_code, 302) @@ -381,22 +364,16 @@ def test_anonymous_post_is_refused(self): class RefusedGetsWriteNothingTests(OwnerRuleFixture): """Two profile GETs write; a refused caller must not reach either write.""" - callers = ("foreign", "anonymous") - - def act_as(self, caller): - self.client.logout() - if caller == "foreign": - self.client.force_login(self.foreign) + def callers(self): + return (("foreign", self.foreign), ("anonymous", None)) def test_refused_settings_get_creates_no_token(self): - from rest_framework.authtoken.models import Token - Token.objects.filter(user=self.owner).delete() url = reverse("login:settings", kwargs={"user_id": self.owner.pk}) - for caller in self.callers: + for caller, user in self.callers(): for headers in ({}, HTMX): with self.subTest(caller=caller, htmx=bool(headers)): - self.act_as(caller) + act_as(self.client, user) self.client.get(url, **headers) self.assertFalse(Token.objects.filter(user=self.owner).exists()) @@ -409,10 +386,10 @@ def test_refused_reviews_get_leaves_the_review_manager_alone(self): ) manager = PeerReviewManager.objects.create(opr=review, is_open_since="stale") url = reverse("login:reviews", kwargs={"user_id": self.owner.pk}) - for caller in self.callers: + for caller, user in self.callers(): for headers in ({}, HTMX): with self.subTest(caller=caller, htmx=bool(headers)): - self.act_as(caller) + act_as(self.client, user) self.client.get(url, **headers) manager.refresh_from_db() self.assertEqual(manager.is_open_since, "stale") From 98c6dd0c229e56255d657e5a83c8efae9694386a Mon Sep 17 00:00:00 2001 From: jh-RLI Date: Thu, 1 Oct 2026 13:53:11 +0200 Subject: [PATCH 6/7] Share one membership lookup in access.py membership_or_404 in login/access.py does the organization and membership lookup with a minimum level: not a member 404, level too low 403, as each view did by hand before. The organization views in this change use it. is_htmx replaces the three inline HX-Request checks in the views this change touches, the warning about mixin order is stated once on ProfileOwnerRequiredMixin, _refusal is now _refusal_or_raise_404, and the settings view uses the module-level Token import. Co-Authored-By: Claude Opus 5.5 (1M context) --- login/access.py | 40 +++++++++++++++++++++++++-------- login/views.py | 59 +++++++++++++++++++++---------------------------- 2 files changed, 56 insertions(+), 43 deletions(-) diff --git a/login/access.py b/login/access.py index 3ee5825d6..a22b7119e 100644 --- a/login/access.py +++ b/login/access.py @@ -38,20 +38,25 @@ from weakref import WeakSet from django.contrib.auth.views import redirect_to_login +from django.core.exceptions import PermissionDenied from django.http import Http404, HttpResponse +from django.shortcuts import get_object_or_404 +from login.models import Membership, Organization +from login.permissions import NO_PERM -def _is_htmx(request) -> bool: - return "HX-Request" in request.headers +def is_htmx(request) -> bool: + """Whether the request came from htmx (it sends ``HX-Request``).""" + return "HX-Request" in request.headers -def _refusal(request, user_id): - """The response that refuses this caller, or None for the owner. - Raises Http404 for a logged-in caller with another id. +def _refusal_or_raise_404(request, user_id): + """Settle the caller: None for the owner, a refusal for an anonymous + caller, and ``Http404`` raised for a logged-in caller with another id. """ if not request.user.is_authenticated: - if _is_htmx(request): + if is_htmx(request): return HttpResponse(status=401) return redirect_to_login(request.get_full_path()) if str(request.user.pk) != str(user_id): @@ -63,13 +68,15 @@ class ProfileOwnerRequiredMixin: """The owner rule for class-based views. List it FIRST in the bases: ``View.dispatch`` does not call further along - the MRO, so a mixin placed after the view class never runs. + the MRO, so a mixin placed after the view class never runs. The same holds + for every mixin that guards ``dispatch``, Django's ``LoginRequiredMixin`` + included; the organization views point here for that reason. On a match ``self.profile_user`` is the caller. """ def dispatch(self, request, *args, **kwargs): - refusal = _refusal(request, kwargs.get("user_id")) + refusal = _refusal_or_raise_404(request, kwargs.get("user_id")) if refusal is not None: return refusal self.profile_user = request.user @@ -85,7 +92,7 @@ def profile_owner_required(view_func): @wraps(view_func) def wrapper(request, user_id, *args, **kwargs): - refusal = _refusal(request, user_id) + refusal = _refusal_or_raise_404(request, user_id) if refusal is not None: return refusal return view_func(request, request.user, *args, **kwargs) @@ -113,3 +120,18 @@ def enforces_owner_rule(view) -> bool: if view_class is not None: return view_class.dispatch is ProfileOwnerRequiredMixin.dispatch return view in _GUARDED_FUNCTIONS + + +def membership_or_404(user, organization_id, min_level=NO_PERM): + """The organization and ``user``'s membership in it, or a refusal. + + 404 when the organization does not exist or ``user`` is not a member, so + a non-member learns nothing about it; ``PermissionDenied`` (403) when the + member's level is below ``min_level``. Returns + ``(organization, membership)``. + """ + organization = get_object_or_404(Organization, id=organization_id) + membership = get_object_or_404(Membership, group=organization, user=user) + if membership.level < min_level: + raise PermissionDenied + return organization, membership diff --git a/login/views.py b/login/views.py index 940d4c834..2af238638 100644 --- a/login/views.py +++ b/login/views.py @@ -55,9 +55,14 @@ ) from dataedit.helper import delete_peer_review from dataedit.models import Dataset, PeerReviewManager, Table, Topic -from login.access import ProfileOwnerRequiredMixin, profile_owner_required +from login.access import ( + ProfileOwnerRequiredMixin, + is_htmx, + membership_or_404, + profile_owner_required, +) from login.forms import EditUserForm, OrganizationForm -from login.models import Membership, Organization +from login.models import Membership from login.models import myuser as OepUser from login.permissions import ADMIN_PERM, DELETE_PERM, WRITE_PERM from login.utils import get_tables_for_organization @@ -125,7 +130,7 @@ def get(self, request, user_id): "search_query": search_query, } - if "HX-Request" in request.headers and not has_search_param: + if is_htmx(request) and not has_search_param: return render( request, "login/partials/tables_sections.html", @@ -209,7 +214,7 @@ class DatasetsView(ProfileOwnerRequiredMixin, View): def get(self, request, user_id): user = self.profile_user context = _datasets_context(request, user) - if "HX-Request" in request.headers: + if is_htmx(request): return render(request, "login/partials/datasets_sections.html", context) return render(request, "login/user_datasets.html", context) @@ -593,8 +598,6 @@ def get(self, request, user_id): :return: Profile renderer """ - from rest_framework.authtoken.models import Token - for user in OepUser.objects.all(): Token.objects.get_or_create(user=user) user = self.profile_user @@ -641,8 +644,7 @@ def organization_leave_view(request, organization_id: int): """ """ user: OepUser = request.user user_id: int = request.user.id - organization = get_object_or_404(Organization, id=organization_id) - membership = get_object_or_404(Membership, group=organization, user=request.user) + organization, membership = membership_or_404(request.user, organization_id) members = ( Membership.objects.filter(group=organization).exclude(user=user.pk).count() @@ -672,10 +674,9 @@ def organization_leave_view(request, organization_id: int): @login_required def organization_delete_view(request, organization_id: int): """View to delete an organization.""" - organization = get_object_or_404(Organization, id=organization_id) - membership = get_object_or_404(Membership, group=organization, user=request.user) - if membership.level < login.permissions.ADMIN_PERM: - raise PermissionDenied + organization, _ = membership_or_404( + request.user, organization_id, min_level=ADMIN_PERM + ) organization.delete() messages.add_message( request, @@ -711,8 +712,7 @@ def get(self, request, user_id: int): class OrganizationManagementView(LoginRequiredMixin, View): """Create an organization, or edit one the caller administers. - The login mixin comes first in the bases: ``View.dispatch`` does not call - further along the MRO, so a mixin after the view class never runs. + The login mixin comes first in the bases, see ProfileOwnerRequiredMixin. """ @method_decorator(never_cache) @@ -729,10 +729,7 @@ def get(self, request, organization_id=None): can_edit = False organization = None if organization_id: - organization = get_object_or_404(Organization, id=organization_id) - membership = get_object_or_404( - Membership, group=organization, user=request.user - ) + organization, membership = membership_or_404(request.user, organization_id) # In case the organization is down to one member make sure # the remaining user gets admin permissions @@ -758,7 +755,7 @@ def get(self, request, organization_id=None): organization_tables = get_tables_for_organization(organization=organization) # Redirect if the request is not triggered using htmx methods - if "HX-Request" not in request.headers: + if not is_htmx(request): return redirect("login:organizations", user_id=request.user.id) return render( @@ -789,12 +786,9 @@ def post(self, request, organization_id=None): organization = None if organization_id: # who may edit is settled before the form touches the instance - organization = get_object_or_404(Organization, id=organization_id) - membership = get_object_or_404( - Membership, group=organization, user=request.user + organization, _ = membership_or_404( + request.user, organization_id, min_level=ADMIN_PERM ) - if membership.level < ADMIN_PERM: - raise PermissionDenied form = OrganizationForm(request.POST, instance=organization) if not form.is_valid(): @@ -834,7 +828,10 @@ def post(self, request, organization_id=None): class OrganizationMembersView(LoginRequiredMixin, TemplateView): """The member list of an organization, for its members only, and the - member changes their level allows. Login mixin first, as above.""" + member changes their level allows. + + The login mixin comes first in the bases, see ProfileOwnerRequiredMixin. + """ template_name = "login/partials/organization_members.html" @@ -842,11 +839,8 @@ def get_context_data(self, **kwargs): """Render context.""" context = super(OrganizationMembersView, self).get_context_data(**kwargs) - organization = get_object_or_404( - Organization, pk=self.kwargs["organization_id"] - ) - membership = get_object_or_404( - Membership, group=organization, user=self.request.user + organization, membership = membership_or_404( + self.request.user, self.kwargs["organization_id"] ) is_admin = membership.level >= ADMIN_PERM @@ -871,10 +865,7 @@ def post(self, request, organization_id: int): "Post request required field 'mode' not specified!" ) - organization = get_object_or_404(Organization, id=organization_id) - membership = get_object_or_404( - Membership, group=organization, user=request.user - ) + organization, membership = membership_or_404(request.user, organization_id) error_message = None if mode == "add_user": From 5a6c4aebabd321cf008c04f61b27b152baecd12e Mon Sep 17 00:00:00 2001 From: jh-RLI Date: Thu, 1 Oct 2026 14:20:15 +0200 Subject: [PATCH 7/7] Reject wrapped class views in the rule check functools.wraps copies view_class onto a decorator around as_view(), so enforces_owner_rule accepted such a wrapper although the decorator runs before the mixin. It now accepts the decorator's registered wrapper first and rejects any other callable carrying __wrapped__ before looking at view_class. Self-tests pin login_required and never_cache around as_view(). Also: the registry sits above the decorator, the module docstring covers membership_or_404 and is_htmx, the organization views use the bare permission constants, and the tests call act_as directly. Co-Authored-By: Claude Opus 5.5 (1M context) --- login/access.py | 46 +++++++++++++------ login/tests/helpers.py | 3 ++ login/tests/test_organization_write_checks.py | 27 +++++------ login/tests/test_profile_owner_rule.py | 11 +++++ login/views.py | 7 ++- 5 files changed, 60 insertions(+), 34 deletions(-) diff --git a/login/access.py b/login/access.py index a22b7119e..f54a11284 100644 --- a/login/access.py +++ b/login/access.py @@ -2,9 +2,18 @@ SPDX-FileCopyrightText: 2026 Jonas Huber © Reiner Lemoine Institut SPDX-License-Identifier: AGPL-3.0-or-later -Who may reach a view under ``profile//``: its owner, and nobody else. - -A profile page is a user's own dashboard. There is no public profile and no +Access checks for the ``login`` views: who may reach a profile view, and who +may act on an organization. + +- ``ProfileOwnerRequiredMixin`` / ``profile_owner_required``: the owner rule + for every view under ``profile//`` (below), and + ``enforces_owner_rule`` to tell whether it runs for a URL callback. +- ``membership_or_404``: the caller's membership in an organization, with a + minimum level; not a member answers 404, a level too low 403. +- ``is_htmx``: whether a request came from htmx, which decides between a + page answer and a fragment answer. + +The owner rule: a profile page is a user's own dashboard. There is no public profile and no "someone else's dashboard" to show, so every route carrying a ``user_id`` answers only when that id is the caller's own: @@ -29,9 +38,9 @@ that one holds the permission levels and is imported by ``login/models.py`` at app-loading time, before the auth views this module needs can be imported. -``enforces_owner_rule`` tells whether the rule actually runs for a URL -callback. ``login/tests/test_profile_owner_rule`` walks every URL pattern and -fails for a ``user_id`` route on which it does not. +``login/tests/test_profile_owner_rule`` walks every URL pattern and fails for +a ``user_id`` route on which ``enforces_owner_rule`` says the rule does not +run. """ # noqa: 501 from functools import wraps @@ -83,6 +92,12 @@ def dispatch(self, request, *args, **kwargs): return super().dispatch(request, *args, **kwargs) +# The wrappers profile_owner_required made. A registry rather than an +# attribute, because functools.wraps copies attributes onto whatever wraps a +# view, so an attribute would also mark a function that never runs the rule. +_GUARDED_FUNCTIONS = WeakSet() + + def profile_owner_required(view_func): """The owner rule for function views. @@ -101,25 +116,26 @@ def wrapper(request, user_id, *args, **kwargs): return wrapper -# The wrappers profile_owner_required made. A registry rather than an -# attribute, because functools.wraps copies attributes onto whatever wraps a -# view, so an attribute would also mark a function that never runs the rule. -_GUARDED_FUNCTIONS = WeakSet() - - def enforces_owner_rule(view) -> bool: """Whether the owner rule runs first for this URL callback. A class-based view must resolve ``dispatch`` to the mixin's own: that rejects the mixin listed after ``View`` (``View.dispatch`` wins and never calls along the MRO) and a ``dispatch`` override that could skip it. A - function view must be the decorator's wrapper itself, the outermost layer, - so nothing runs before the rule. + function view must be the decorator's wrapper itself. Either way the rule + is the outermost layer, so nothing runs before it: any other decorated + callable is rejected, including a decorator around ``as_view()``, which + ``functools.wraps`` makes look like the class view by copying + ``view_class`` onto it. """ + if view in _GUARDED_FUNCTIONS: + return True + if hasattr(view, "__wrapped__"): + return False view_class = getattr(view, "view_class", None) if view_class is not None: return view_class.dispatch is ProfileOwnerRequiredMixin.dispatch - return view in _GUARDED_FUNCTIONS + return False def membership_or_404(user, organization_id, min_level=NO_PERM): diff --git a/login/tests/helpers.py b/login/tests/helpers.py index 8466893d0..31ccc2adf 100644 --- a/login/tests/helpers.py +++ b/login/tests/helpers.py @@ -15,6 +15,9 @@ def make_user(name, **extra): + """A verified user who has agreed to the terms, named ``name``; its + email is derived from the name. ``extra`` sets further fields, such as + ``is_admin``. Returns the existing user if one already matches.""" user, _ = myuser.objects.get_or_create( name=name, email=f"{name.lower()}@test.com", diff --git a/login/tests/test_organization_write_checks.py b/login/tests/test_organization_write_checks.py index 56ec45494..b8e33819d 100644 --- a/login/tests/test_organization_write_checks.py +++ b/login/tests/test_organization_write_checks.py @@ -53,9 +53,6 @@ def level_of(self, user): ).first() return membership.level if membership else None - def act_as(self, user): - act_as(self.client, user) - def own_organizations_page(self, user): return reverse("login:organizations", kwargs={"user_id": user.pk}) @@ -74,7 +71,7 @@ def edit_url(self): ) def rename_as(self, user): - self.act_as(user) + act_as(self.client, user) return self.client.post( self.edit_url(), {"name": "org_check_renamed", "description": "renamed"}, @@ -111,7 +108,7 @@ def test_anonymous_is_sent_to_login_and_nothing_changes(self): self.assert_unchanged() def test_unknown_organization_is_404(self): - self.act_as(self.admin) + act_as(self.client, self.admin) response = self.client.post( reverse("login:organization-edit", kwargs={"organization_id": 999999}), {"name": "org_check_renamed", "description": "renamed"}, @@ -139,7 +136,7 @@ def test_anonymous_create_leaves_no_organization(self): ) def test_logged_in_create_makes_the_caller_its_admin(self): - self.act_as(self.stranger) + act_as(self.client, self.stranger) response = self.client.post( self.url, {"name": "org_check_new", "description": "d"}, **HTMX ) @@ -159,13 +156,13 @@ class LeaveAndDeleteRedirectTests(OrganizationFixture): """ def caller_other_than_user_one(self, *candidates): - caller = next(user for user in candidates if user.pk != 1) - self.assertNotEqual(caller.pk, 1) - return caller + callers = [user for user in candidates if user.pk != 1] + self.assertTrue(callers, "every candidate caller has pk 1") + return callers[0] def test_leave_redirects_to_own_page(self): caller = self.caller_other_than_user_one(self.inviter, self.remover) - self.act_as(caller) + act_as(self.client, caller) response = self.client.post( reverse( "login:organization-leave", @@ -178,7 +175,7 @@ def test_leave_redirects_to_own_page(self): def test_delete_redirects_to_own_page(self): caller = self.caller_other_than_user_one(self.admin, self.second_admin) - self.act_as(caller) + act_as(self.client, caller) response = self.client.post( reverse( "login:organization-delete", @@ -198,21 +195,21 @@ def members_url(self): ) def post_as(self, user, data): - self.act_as(user) + act_as(self.client, user) return self.client.post(self.members_url(), data, **HTMX) def test_member_list_needs_a_membership(self): - self.act_as(self.inviter) + act_as(self.client, self.inviter) self.assertContains( self.client.get(self.members_url(), **HTMX), "OrgCheckAdmin" ) - self.act_as(self.stranger) + act_as(self.client, self.stranger) response = self.client.get(self.members_url(), **HTMX) self.assertEqual(response.status_code, 404) self.assertNotIn("OrgCheckAdmin", response.content.decode()) - self.act_as(None) + act_as(self.client, None) response = self.client.get(self.members_url(), **HTMX) self.assertEqual(response.status_code, 302) diff --git a/login/tests/test_profile_owner_rule.py b/login/tests/test_profile_owner_rule.py index 2c8acebc2..51e95261d 100644 --- a/login/tests/test_profile_owner_rule.py +++ b/login/tests/test_profile_owner_rule.py @@ -16,6 +16,7 @@ from urllib.parse import parse_qs, urlsplit from django.conf import settings +from django.contrib.auth.decorators import login_required from django.contrib.auth.models import AnonymousUser from django.http import HttpResponse from django.shortcuts import resolve_url @@ -211,6 +212,16 @@ def test_rule_wrapped_by_another_decorator_is_rejected(self): # the rule must be the outermost layer, so nothing runs before it self.assertFalse(enforces_owner_rule(never_cache(_guarded_function))) + def test_wrapped_class_view_is_rejected(self): + # functools.wraps copies view_class onto the wrapper, so a decorator + # around as_view() looks like the class view while the wrapper runs + # first; the rule must be the outermost layer for class views too + for decorator in (login_required, never_cache): + with self.subTest(decorator=decorator.__name__): + wrapped = decorator(_Guarded.as_view()) + self.assertIs(wrapped.view_class, _Guarded) + self.assertFalse(enforces_owner_rule(wrapped)) + def test_guarded_views_are_accepted(self): self.assertTrue(enforces_owner_rule(_Guarded.as_view())) self.assertTrue(enforces_owner_rule(_guarded_function)) diff --git a/login/views.py b/login/views.py index 2af238638..5b680edf4 100644 --- a/login/views.py +++ b/login/views.py @@ -41,7 +41,6 @@ from django.views.generic.edit import DeleteView from rest_framework.authtoken.models import Token -import login.permissions from api.serializers import DatasetCreateSerializer, DatasetUpdateSerializer from api.services.dataset_creation import ( DatasetNameTaken, @@ -869,7 +868,7 @@ def post(self, request, organization_id: int): error_message = None if mode == "add_user": - if membership.level < login.permissions.WRITE_PERM: + if membership.level < WRITE_PERM: raise PermissionDenied try: user = OepUser.objects.get(name=request.POST["name"]) @@ -881,7 +880,7 @@ def post(self, request, organization_id: int): error_message = "User does not exist" elif mode == "remove_user": - if membership.level < login.permissions.DELETE_PERM: + if membership.level < DELETE_PERM: raise PermissionDenied user_to_remove: OepUser = OepUser.objects.get(id=request.POST["user_id"]) @@ -911,7 +910,7 @@ def post(self, request, organization_id: int): target_membership.delete() elif mode == "alter_user": - if membership.level < login.permissions.ADMIN_PERM: + if membership.level < ADMIN_PERM: raise PermissionDenied user = OepUser.objects.get(id=request.POST["user_id"]) if user == request.user: