diff --git a/app/CHANGELOG.md b/app/CHANGELOG.md index f80fa7a1..8a431628 100644 --- a/app/CHANGELOG.md +++ b/app/CHANGELOG.md @@ -1,6 +1,8 @@ # Changelog ## [Unreleased] +### Fixed +- Fix issues with read and edit rights on shared projects [[#478](https://github.com/open-plan-tool/gui/pull/478)] ## [v2.1.4] – 2026-07-01 ### Fixed diff --git a/app/dashboard/views.py b/app/dashboard/views.py index c1a64f76..2fcf9e7e 100644 --- a/app/dashboard/views.py +++ b/app/dashboard/views.py @@ -36,6 +36,7 @@ get_selected_scenarios_in_cache, ) +from projects.decorators import user_has_edit_rights, user_has_read_rights from projects.forms import BusForm, AssetCreateForm, StorageForm from projects.constants import COMPARE_VIEW, STEP_LIST, MAX_STEP @@ -72,13 +73,9 @@ @login_required @json_view @require_http_methods(["GET"]) +@user_has_read_rights def scenario_available_results(request, scen_id): scenario = get_object_or_404(Scenario, pk=scen_id) - if (scenario.project.user != request.user) and ( - scenario.project.viewers.filter(user__email=request.user.email).exists() - is False - ): - raise PermissionDenied try: assets_results_obj = AssetsResults.objects.get(simulation=scenario.simulation) @@ -174,9 +171,7 @@ def scenario_visualize_results( ) else: project = get_object_or_404(Project, id=proj_id) - if (project.user != request.user) and ( - project.viewers.filter(user__email=request.user.email).exists() is False - ): + if not request.user.has_read_rights(project): raise PermissionDenied selected_scenarios = get_selected_scenarios_in_cache(request, proj_id) @@ -217,12 +212,6 @@ def scenario_visualize_results( scenario = get_object_or_404(Scenario, id=scen_id) # TODO: change this when multi-scenario selection is allowed - if (scenario.project.user != request.user) and ( - scenario.project.viewers.filter(user__email=request.user.email).exists() - is False - ): - raise PermissionDenied - qs = FancyResults.objects.filter(simulation=scenario.simulation) if qs.exists() and scenario in user_scenarios: @@ -271,6 +260,7 @@ def scenario_visualize_results( @login_required @require_http_methods(["POST", "GET"]) +@user_has_read_rights def project_compare_results(request, proj_id, step_id=5, max_step=MAX_STEP): request.session[COMPARE_VIEW] = True user_projects = fetch_user_projects(request.user) @@ -281,10 +271,6 @@ def project_compare_results(request, proj_id, step_id=5, max_step=MAX_STEP): ) project = get_object_or_404(Project, id=proj_id) - if (project.user != request.user) and ( - project.viewers.filter(user__email=request.user.email).exists() is False - ): - raise PermissionDenied user_scenarios = project.get_scenarios_with_results() report_items_data = [ @@ -340,10 +326,7 @@ def project_sensitivity_analysis(request, proj_id, sa_id=None): ) else: project = get_object_or_404(Project, id=proj_id) - if (scenario.project.user != request.user) and ( - scenario.project.viewers.filter(user__email=request.user.email).exists() - is False - ): + if not request.user.has_read_rights(project): raise PermissionDenied user_sa = get_project_sensitivity_analysis(project) @@ -1002,6 +985,7 @@ def view_asset_parameters(request, scen_id, asset_type_name, asset_uuid): @login_required @json_view @require_http_methods(["GET"]) +@user_has_read_rights def scenario_economic_results(request, scen_id=None): """ This view gathers all simulation specific cost matrix KPI results @@ -1017,14 +1001,6 @@ def scenario_economic_results(request, scen_id=None): scenario = get_object_or_404(Scenario, pk=scen_id) - # if scenario.project.user != request.user: - # return HttpResponseForbidden() - if (scenario.project.user != request.user) and ( - scenario.project.viewers.filter(user__email=request.user.email).exists() - is False - ): - raise PermissionDenied - try: kpi_cost_results_obj = KPICostsMatrixResults.objects.get( simulation=scenario.simulation @@ -1093,6 +1069,7 @@ def scenario_economic_results(request, scen_id=None): @login_required @json_view @require_http_methods(["GET"]) +@user_has_read_rights def scenario_visualize_timeseries(request, proj_id=None, scen_id=None): if scen_id is None: selected_scenario = get_selected_scenarios_in_cache(request, proj_id) @@ -1103,11 +1080,6 @@ def scenario_visualize_timeseries(request, proj_id=None, scen_id=None): for scen_id in selected_scenario: scenario = get_object_or_404(Scenario, pk=scen_id) - if (scenario.project.user != request.user) and ( - scenario.project.viewers.filter(user__email=request.user.email).exists() - is False - ): - raise PermissionDenied simulations.append(scenario.simulation) results_json = report_item_render_to_json( @@ -1122,13 +1094,10 @@ def scenario_visualize_timeseries(request, proj_id=None, scen_id=None): ) +@login_required +@user_has_read_rights def scenario_visualize_stacked_timeseries(request, scen_id): scenario = get_object_or_404(Scenario, pk=scen_id) - if (scenario.project.user != request.user) and ( - scenario.project.viewers.filter(user__email=request.user.email).exists() - is False - ): - raise PermissionDenied results_json = [] for energy_vector in scenario.energy_vectors: @@ -1151,6 +1120,7 @@ def scenario_visualize_stacked_timeseries(request, scen_id): # TODO exclude sink components +@user_has_read_rights def scenario_visualize_capacities(request, proj_id, scen_id=None): if scen_id is None: selected_scenario = get_selected_scenarios_in_cache(request, proj_id) @@ -1161,11 +1131,6 @@ def scenario_visualize_capacities(request, proj_id, scen_id=None): qs = Scenario.objects.filter(id__in=selected_scenario).order_by("name") for scenario in qs: - if (scenario.project.user != request.user) and ( - scenario.project.viewers.filter(user__email=request.user.email).exists() - is False - ): - raise PermissionDenied simulations.append(scenario.simulation) results_json = report_item_render_to_json( @@ -1180,6 +1145,7 @@ def scenario_visualize_capacities(request, proj_id, scen_id=None): ) +@user_has_read_rights def scenario_visualize_costs(request, proj_id, scen_id=None): if scen_id is None: selected_scenario = get_selected_scenarios_in_cache(request, proj_id) @@ -1190,11 +1156,6 @@ def scenario_visualize_costs(request, proj_id, scen_id=None): qs = Scenario.objects.filter(id__in=selected_scenario).order_by("name") for scenario in qs: - if (scenario.project.user != request.user) and ( - scenario.project.viewers.filter(user__email=request.user.email).exists() - is False - ): - raise PermissionDenied simulations.append(scenario.simulation) results_json = [] @@ -1216,13 +1177,11 @@ def scenario_visualize_costs(request, proj_id, scen_id=None): # TODO: Sector coupling must be refined (including transformer flows) +@login_required +@user_has_read_rights def scenario_visualize_sankey(request, scen_id, ts=None): scenario = get_object_or_404(Scenario, pk=scen_id) - if (scenario.project.user != request.user) and ( - scenario.project.viewers.filter(user__email=request.user.email).exists() - is False - ): - raise PermissionDenied + if ts is not None: ts = int(ts) results_json = report_item_render_to_json( @@ -1243,15 +1202,10 @@ def scenario_visualize_sankey(request, scen_id, ts=None): @login_required @require_http_methods(["GET"]) +@user_has_read_rights def download_scalar_results(request, scen_id): scenario = get_object_or_404(Scenario, pk=scen_id) - if (scenario.project.user != request.user) and ( - scenario.project.viewers.filter(user__email=request.user.email).exists() - is False - ): - raise PermissionDenied - try: kpi_scalar_results_obj = KPIScalarResults.objects.get( simulation=scenario.simulation @@ -1290,15 +1244,10 @@ def download_scalar_results(request, scen_id): @login_required @require_http_methods(["GET"]) +@user_has_read_rights def download_cost_results(request, scen_id): scenario = get_object_or_404(Scenario, pk=scen_id) - if (scenario.project.user != request.user) and ( - scenario.project.viewers.filter(user__email=request.user.email).exists() - is False - ): - raise PermissionDenied - try: kpi_cost_results_obj = KPICostsMatrixResults.objects.get( simulation=scenario.simulation @@ -1336,15 +1285,10 @@ def download_cost_results(request, scen_id): @login_required @require_http_methods(["GET"]) +@user_has_read_rights def download_timeseries_results(request, scen_id): scenario = get_object_or_404(Scenario, pk=scen_id) - if (scenario.project.user != request.user) and ( - scenario.project.viewers.filter(user__email=request.user.email).exists() - is False - ): - raise PermissionDenied - try: assets_results_obj = AssetsResults.objects.get(simulation=scenario.simulation) assets_results_json = json.loads(assets_results_obj.assets_list) diff --git a/app/locale/de/LC_MESSAGES/django.po b/app/locale/de/LC_MESSAGES/django.po index deacd4d3..851e5c03 100644 --- a/app/locale/de/LC_MESSAGES/django.po +++ b/app/locale/de/LC_MESSAGES/django.po @@ -381,6 +381,21 @@ msgstr "" "Wählen Sie den/die Benutzer aus, für den/die Sie die Zugriffsrechte " "entziehen möchten." +msgid "Revoke access" +msgstr "Zugriffsrechte entziehen" + +msgid "Share" +msgstr "Teilen" + +msgid "You need to be the owner of the project to access this." +msgstr "Sie müssen der Eigentümer des Projekts sein, um darauf zugreifen zu können." + +msgid "You need to be the owner of the project or have read rights to access this." +msgstr "Sie müssen entweder der Eigentümer des Projekts sein oder über Leserechte verfügen, um darauf zugreifen zu können." + +msgid "You need to be the owner of the project or have edit rights to access this." +msgstr "Sie müssen der Eigentümer des Projekts sein oder über Bearbeitungsrechte verfügen, um darauf zugreifen zu können." + #: projects/forms.py:367 msgid "Users currently having access to the project" msgstr "Benutzer, die derzeit Zugriff auf das Projekt haben" diff --git a/app/projects/decorators.py b/app/projects/decorators.py new file mode 100644 index 00000000..98c7fbb8 --- /dev/null +++ b/app/projects/decorators.py @@ -0,0 +1,106 @@ +from django.shortcuts import get_object_or_404, render +from django.utils.translation import gettext as _ + +from functools import wraps + +from projects.models import Project, Scenario + + +def get_project_from_proj_or_scen_id(proj_id, scen_id): + if proj_id is not None: + return get_object_or_404(Project, pk=proj_id) + + if scen_id is not None: + scenario = get_object_or_404(Scenario, pk=scen_id) + return scenario.project + + raise ValueError("this decorator requires proj_id or scen_id") + + +def user_is_owner(view_func): + @wraps(view_func) + def _wrapped_view(request, proj_id=None, scen_id=None, *args, **kwargs): + project = get_project_from_proj_or_scen_id(proj_id, scen_id) + + if project.user != request.user: + reason = _("You need to be the owner of the project to access this.") + return render( + request, + "error_403.html", + { + "reason": reason, + }, + ) + + # check for existing parameters to handle the different view parameters + if proj_id is not None and scen_id is not None: + return view_func(request, proj_id, scen_id, *args, **kwargs) + elif proj_id is not None: + return view_func(request, proj_id, *args, **kwargs) + elif scen_id is not None: + return view_func(request, scen_id, *args, **kwargs) + + return _wrapped_view + + +def user_has_read_rights(view_func): + @wraps(view_func) + def _wrapped_view(request, proj_id=None, scen_id=None, *args, **kwargs): + project = get_project_from_proj_or_scen_id(proj_id, scen_id) + + if (project.user != request.user) and ( + project.viewers.filter(user__email=request.user.email).exists() is False + ): + reason = _( + "You need to be the owner of the project or have read rights to access this." + ) + return render( + request, + "error_403.html", + { + "reason": reason, + }, + ) + + # check for existing parameters to handle the different view parameters + if proj_id is not None and scen_id is not None: + return view_func(request, proj_id, scen_id, *args, **kwargs) + elif proj_id is not None: + return view_func(request, proj_id, *args, **kwargs) + elif scen_id is not None: + return view_func(request, scen_id, *args, **kwargs) + + return _wrapped_view + + +def user_has_edit_rights(view_func): + @wraps(view_func) + def _wrapped_view(request, proj_id=None, scen_id=None, *args, **kwargs): + project = get_project_from_proj_or_scen_id(proj_id, scen_id) + + if (project.user != request.user) and ( + project.viewers.filter( + user__email=request.user.email, share_rights="edit" + ).exists() + is False + ): + reason = _( + "You need to be the owner of the project or have edit rights to access this." + ) + return render( + request, + "error_403.html", + { + "reason": reason, + }, + ) + + # check for existing parameters to handle the different view parameters + if proj_id is not None and scen_id is not None: + return view_func(request, proj_id, scen_id, *args, **kwargs) + elif proj_id is not None: + return view_func(request, proj_id, *args, **kwargs) + elif scen_id is not None: + return view_func(request, scen_id, *args, **kwargs) + + return _wrapped_view diff --git a/app/projects/templatetags/custom_filters.py b/app/projects/templatetags/custom_filters.py index 7c54cf49..b5eacccf 100644 --- a/app/projects/templatetags/custom_filters.py +++ b/app/projects/templatetags/custom_filters.py @@ -57,9 +57,7 @@ def has_viewer_read_rights(proj_id, user): if project.user == user: answer = True else: - qs = project.viewers.filter( - Q(user__email=user.email) & Q(share_rights="read") - ) + qs = project.viewers.filter(Q(user__email=user.email)) if qs.exists(): answer = True return answer diff --git a/app/projects/tests.py b/app/projects/tests.py index b3046bdf..5b162971 100644 --- a/app/projects/tests.py +++ b/app/projects/tests.py @@ -33,6 +33,56 @@ def test_delete_project_redirects(self): self.assertEqual(response.status_code, 200) self.assertEqual(Project.objects.all().count(), 0) + def test_delete_project_as_owner_removes_project(self): + """Make sure when you are the owner of a project, it gets deleted from the database.""" + response = self.client.post(reverse("project_delete", args=[self.project.id])) + self.assertRedirects(response, reverse("project_search")) + response = self.client.get(response.url) + self.assertEqual(response.status_code, 200) + self.assertEqual(Project.objects.filter(id=self.project.id).count(), 0) + + def test_delete_project_as_read_viewer_removes_access_only(self): + """Make sure when you have read rights, project stays in database and you lose rights.""" + read_user = CustomUser.objects.last() + success, _ = self.project.add_viewer_if_not_exist( + email=read_user.email, share_rights="read" + ) + self.assertTrue(success) + + self.client.logout() + self.client.force_login(read_user) + + response = self.client.post(reverse("project_delete", args=[self.project.id])) + self.assertRedirects(response, reverse("project_search")) + response = self.client.get(response.url) + self.assertEqual(response.status_code, 200) + self.project.refresh_from_db() + self.assertEqual( + Project.objects.filter(id=self.project.id).count(), 1 + ) # project still in database + self.assertFalse(self.project.viewers.filter(user=read_user).exists()) + + def test_delete_project_as_edit_viewer_removes_access_only(self): + """Make sure when you have edit rights, project stays in database and you lose rights.""" + edit_user = CustomUser.objects.last() + success, _ = self.project.add_viewer_if_not_exist( + email=edit_user.email, share_rights="edit" + ) + self.assertTrue(success) + + self.client.logout() + self.client.force_login(edit_user) + + response = self.client.post(reverse("project_delete", args=[self.project.id])) + self.assertRedirects(response, reverse("project_search")) + response = self.client.get(response.url) + self.assertEqual(response.status_code, 200) + self.project.refresh_from_db() + self.assertEqual( + Project.objects.filter(id=self.project.id).count(), 1 + ) # project still in database + self.assertFalse(self.project.viewers.filter(user=edit_user).exists()) + def test_duplicate_project_redirects(self): """Make sure we are redirected to project page once duplicating a project""" response = self.client.post( @@ -149,6 +199,44 @@ def test_visit_create_scenario_link_from_landing_page_links_to_right_view(self): self.assertEqual(response.status_code, 200) self.assertTemplateUsed(response, "scenario/scenario_step1.html") + def test_duplicate_scenario_as_edit_viewer(self): + self.scenario = self.project.scenario_set.first() + edit_user = CustomUser.objects.last() + success, _ = self.project.add_viewer_if_not_exist( + email=edit_user.email, share_rights="edit" + ) + self.assertTrue(success) + + scenario_count = Scenario.objects.count() + self.client.logout() + self.client.force_login(edit_user) + + response = self.client.get( + reverse("scenario_duplicate", args=[self.scenario.id]) + ) + self.assertRedirects( + response, reverse("project_search", args=[self.project.id]) + ) + self.assertEqual(Scenario.objects.count(), scenario_count + 1) + + def test_duplicate_scenario_as_read_viewer_raises_permission_error(self): + self.scenario = self.project.scenario_set.first() + read_user = CustomUser.objects.last() + success, _ = self.project.add_viewer_if_not_exist( + email=read_user.email, share_rights="read" + ) + self.assertTrue(success) + + scenario_count = Scenario.objects.count() + self.client.logout() + self.client.force_login(read_user) + + response = self.client.get( + reverse("scenario_duplicate", args=[self.scenario.id]) + ) + self.assertTemplateUsed(response, "error_403.html") + self.assertEqual(Scenario.objects.count(), scenario_count) + def test_logout(self): response = self.client.post(reverse("logout"), follow=True) self.assertEqual(response.status_code, 200) diff --git a/app/projects/views.py b/app/projects/views.py index 4acf2662..10f0c550 100644 --- a/app/projects/views.py +++ b/app/projects/views.py @@ -69,6 +69,11 @@ MaxEmissionConstraint, NZEConstraint, ) +from projects.decorators import ( + user_is_owner, + user_has_read_rights, + user_has_edit_rights, +) from dashboard.models import FancyResults from .scenario_topology_helpers import ( handle_storage_unit_form_post, @@ -126,16 +131,13 @@ def landing_default(request, version=1): @login_required @require_http_methods(["POST"]) +@user_is_owner def scenario_upload(request, proj_id): # read the scenario file to a dict scenario_data = request.FILES["file"].read() scenario_data = json.loads(scenario_data) - project = get_object_or_404(Project, id=proj_id) - if project.user != request.user: - raise PermissionDenied - answer = HttpResponseRedirect(reverse("project_search")) file_format_error = False # make a single scenario within a list @@ -214,16 +216,10 @@ def sponsor_feature(request): @login_required @json_view @require_http_methods(["GET"]) +@user_is_owner def project_members_list(request, proj_id): project = get_object_or_404(Project, pk=proj_id) - if project.user != request.user: - return JsonResponse( - {"status": "error", "message": "Project does not belong to you."}, - status=403, - content_type="application/json", - ) - viewers = project.viewers.values_list("email", flat=True) return JsonResponse( {"status": "success", "viewers": list(viewers)}, @@ -234,14 +230,11 @@ def project_members_list(request, proj_id): @login_required @require_http_methods(["POST"]) +@user_is_owner def project_share(request, proj_id): qs = request.POST - project = get_object_or_404(Project, id=proj_id) - if project.user != request.user: - raise PermissionDenied - form_item = ProjectShareForm(qs) if form_item.is_valid(): @@ -257,13 +250,11 @@ def project_share(request, proj_id): @login_required @json_view @require_http_methods(["POST"]) +@user_is_owner def project_revoke_access(request, proj_id=None): qs = request.POST - project = get_object_or_404(Project, id=proj_id) - if project.user != request.user: - raise PermissionDenied form_item = ProjectRevokeForm(qs, proj_id=proj_id) if form_item.is_valid(): success, message = project.revoke_access(**form_item.cleaned_data) @@ -278,13 +269,10 @@ def project_revoke_access(request, proj_id=None): @login_required @json_view @require_http_methods(["POST"]) +@user_is_owner def ajax_project_viewers_form(request): if request.headers.get("x-requested-with") == "XMLHttpRequest": proj_id = int(request.POST.get("proj_id")) - project = get_object_or_404(Project, id=proj_id) - - if project.user != request.user: - raise PermissionDenied form_item = ProjectRevokeForm(proj_id=proj_id) return render( @@ -296,14 +284,9 @@ def ajax_project_viewers_form(request): @login_required @require_http_methods(["GET"]) +@user_has_read_rights def project_detail(request, proj_id): project = get_object_or_404(Project, pk=proj_id) - - if (project.user != request.user) and ( - project.viewers.filter(user__email=request.user.email).exists() is False - ): - raise PermissionDenied - logger.info(f"Populating project and economic details in forms.") project_form = ProjectDetailForm(None, instance=project) economic_data_form = EconomicDataDetailForm(None, instance=project.economic_data) @@ -346,17 +329,10 @@ def project_create(request): @login_required @require_http_methods(["GET", "POST"]) +@user_has_edit_rights def project_update(request, proj_id): project = get_object_or_404(Project, id=proj_id) - if (project.user != request.user) and ( - project.viewers.filter( - user__email=request.user.email, share_rights="edit" - ).exists() - is False - ): - raise PermissionDenied - project_form = ProjectUpdateForm(request.POST or None, instance=project) economic_data_form = EconomicDataUpdateForm( request.POST or None, instance=project.economic_data @@ -434,12 +410,10 @@ def project_update(request, proj_id): @login_required @require_http_methods(["GET", "POST"]) +@user_is_owner def project_export(request, proj_id): project = get_object_or_404(Project, id=proj_id) - if project.user != request.user: - raise PermissionDenied - if request.method == "POST": bind_scenario_data = request.POST.get("bind_scenario_data", True) if bind_scenario_data == "True": @@ -515,15 +489,18 @@ def usecase_export(request, usecase_id): @login_required @require_http_methods(["POST"]) +@user_has_read_rights def project_delete(request, proj_id): project = get_object_or_404(Project, id=proj_id) - if project.user != request.user: - raise PermissionDenied - if request.method == "POST": - project.delete() - messages.success(request, "Project successfully deleted!") + if request.user.is_owner(project): + project.delete() + messages.success(request, "Project successfully deleted!") + else: + viewer = project.viewers.filter(user=request.user) + project.revoke_access(viewer) + messages.success(request, "Project successfully removed from project list!") return HttpResponseRedirect(reverse("project_search")) @@ -596,28 +573,14 @@ def project_search(request, proj_id=None, scen_id=None): @login_required @require_http_methods(["POST"]) +@user_has_edit_rights def project_duplicate(request, proj_id): """Duplicates the selected project along with its associated scenarios""" project = get_object_or_404(Project, pk=proj_id) # duplicate the project dm = project.export(bind_scenario_data=True) - if (project.user == request.user) or ( - project.viewers.filter( - user__email=request.user.email, share_rights="edit" - ).exists() - is True - ): - new_proj_id = load_project_from_dict(dm, user=request.user) - else: - messages.error( - request, - _( - "You cannot duplicate a shared project without the owner granting you 'edit' rights" - ), - ) - new_proj_id = project.id - + new_proj_id = load_project_from_dict(dm, user=request.user) return HttpResponseRedirect(reverse("project_search", args=[new_proj_id])) @@ -779,6 +742,7 @@ def scenario_select_project(request, step_id=0, max_step=1): @login_required @require_http_methods(["GET", "POST"]) +@user_has_read_rights def scenario_create_parameters(request, proj_id, scen_id=None, step_id=1, max_step=2): project = get_object_or_404(Project, pk=proj_id) # all projects which the user is able to select (the one the user created) @@ -793,13 +757,6 @@ def scenario_create_parameters(request, proj_id, scen_id=None, step_id=1, max_st if request.method == "GET": if scen_id is not None: scenario = get_object_or_404(Scenario, id=scen_id) - - if (scenario.project.user != request.user) and ( - scenario.project.viewers.filter(user__email=request.user.email).exists() - is False - ): - raise PermissionDenied - form = ScenarioUpdateForm( None, instance=scenario, project_queryset=user_projects ) @@ -845,12 +802,7 @@ def scenario_create_parameters(request, proj_id, scen_id=None, step_id=1, max_st # Only allow edition in DB for owner or share with edit rights selected_project = form.cleaned_data["project"] - if (selected_project.user == request.user) or ( - selected_project.viewers.filter( - user__email=request.user.email, share_rights="edit" - ).exists() - is True - ): + if request.user.has_edit_rights(selected_project): qs_sim = Simulation.objects.filter(scenario__id=scenario.id) # update the parameter values which are different from existing values for name, value in form.cleaned_data.items(): @@ -960,13 +912,9 @@ def scenario_create_topology(request, proj_id, scen_id, step_id=2, max_step=3): # TODO: if the scenario exists, load it, otherwise default form scenario = get_object_or_404(Scenario, pk=scen_id) + project = get_object_or_404(Project, pk=proj_id) - if (scenario.project.user != request.user) and ( - scenario.project.viewers.filter( - user__email=request.user.email, share_rights="edit" - ).exists() - is False - ): + if request.user.has_edit_rights(project): user_has_right_to_save = False # raise PermissionDenied else: @@ -1024,6 +972,7 @@ def scenario_create_topology(request, proj_id, scen_id, step_id=2, max_step=3): @login_required @require_http_methods(["GET", "POST"]) +@user_has_read_rights def scenario_create_constraints(request, proj_id, scen_id, step_id=3, max_step=4): constraints_labels = { "minimal_degree_of_autonomy": _("Minimal degree of autonomy"), @@ -1046,19 +995,9 @@ def scenario_create_constraints(request, proj_id, scen_id, step_id=3, max_step=4 } scenario = get_object_or_404(Scenario, pk=scen_id) + project = get_object_or_404(Project, pk=proj_id) - if (scenario.project.user != request.user) and ( - scenario.project.viewers.filter(user__email=request.user.email).exists() - is False - ): - raise PermissionDenied - - if (scenario.project.user != request.user) and ( - scenario.project.viewers.filter( - user__email=request.user.email, share_rights="edit" - ).exists() - is False - ): + if request.user.has_edit_rights(project): user_has_right_to_save = False else: user_has_right_to_save = True @@ -1132,12 +1071,7 @@ def scenario_create_constraints(request, proj_id, scen_id, step_id=3, max_step=4 if constraint_type == "net_zero_energy": constraint_instance.value = constraint_instance.activated - if (scenario.project.user == request.user) or ( - scenario.project.viewers.filter( - user__email=request.user.email, share_rights="edit" - ).exists() - is True - ): + if request.user.has_edit_rights(scenario.project): constraint_instance.save() return HttpResponseRedirect(reverse("scenario_review", args=[proj_id, scen_id])) @@ -1145,15 +1079,10 @@ def scenario_create_constraints(request, proj_id, scen_id, step_id=3, max_step=4 @login_required @require_http_methods(["GET", "POST"]) +@user_has_read_rights def scenario_review(request, proj_id, scen_id, step_id=4, max_step=MAX_STEP): scenario = get_object_or_404(Scenario, pk=scen_id) - if (scenario.project.user != request.user) and ( - scenario.project.viewers.filter(user__email=request.user.email).exists() - is False - ): - raise PermissionDenied - if request.method == "GET": html_template = f"scenario/simulation/no-status.html" context = { @@ -1295,27 +1224,20 @@ def scenario_steps(request, proj_id, step_id=None, scen_id=None): # TODO delete this useless code here @login_required @require_http_methods(["GET"]) +@user_has_read_rights def scenario_view(request, scen_id, step_id): """Scenario View. GET request only.""" scenario = get_object_or_404(Scenario, pk=scen_id) - - if (scenario.project.user != request.user) and ( - scenario.project.viewers.filter(user__email=request.user.email).exists() - is False - ): - raise PermissionDenied - return HttpResponseRedirect(reverse("project_search", args=[scenario.project.id])) # TODO delete this useless code here @login_required @require_http_methods(["GET"]) +@user_is_owner def scenario_update(request, scen_id, step_id): """Scenario Update View. POST request only.""" scenario = get_object_or_404(Scenario, pk=scen_id) - if scenario.project.user != request.user: - raise PermissionDenied if request.POST: form = ScenarioUpdateForm(request.POST) if form.is_valid(): @@ -1333,13 +1255,11 @@ def scenario_update(request, scen_id, step_id): @login_required @require_http_methods(["GET"]) +@user_has_edit_rights def scenario_duplicate(request, scen_id): """duplicates the selected scenario and all of its associated components (topology data included)""" scenario = get_object_or_404(Scenario, pk=scen_id) - if scenario.project.user != request.user: - raise PermissionDenied - # We need to iterate over all the objects related to this scenario and duplicate them # and associate them with the new scenario id. asset_list = Asset.objects.filter(scenario=scenario) @@ -1385,12 +1305,10 @@ def scenario_export(request, proj_id): @login_required @require_http_methods(["GET"]) +@user_is_owner def scenario_export_as_datapackage(request, scen_id, n_timestamps=None): scenario = get_object_or_404(Scenario, id=int(scen_id)) - if scenario.project.user != request.user: - raise PermissionDenied - with tempfile.TemporaryDirectory() as temp_dir: destination_path = Path(temp_dir) # write the content of the scenario into a temp directory @@ -1418,10 +1336,9 @@ def scenario_export_as_datapackage(request, scen_id, n_timestamps=None): @login_required @require_http_methods(["GET"]) +@user_is_owner def scenario_export_as_jsonified_datapackage(request, scen_id, n_timestamps=None): scenario = get_object_or_404(Scenario, id=int(scen_id)) - if scenario.project.user != request.user: - raise PermissionDenied json_dp = scenario.to_jsonified_datapackage(number=n_timestamps) response = JsonResponse(json_dp, status=200, content_type="application/json") @@ -1431,12 +1348,10 @@ def scenario_export_as_jsonified_datapackage(request, scen_id, n_timestamps=None @login_required @require_http_methods(["GET"]) +@user_is_owner def project_export_as_datapackage(request, proj_id, n_timestamps=None): project = get_object_or_404(Project, id=int(proj_id)) - if project.user != request.user: - raise PermissionDenied - with tempfile.TemporaryDirectory() as temp_dir: destination_path = Path(temp_dir) @@ -1465,13 +1380,9 @@ def project_export_as_datapackage(request, proj_id, n_timestamps=None): @login_required @require_http_methods(["POST"]) +@user_is_owner def scenario_delete(request, scen_id): scenario = get_object_or_404(Scenario, id=scen_id) - if scenario.project.user != request.user: - logger.warning( - f"Unauthorized user tried to delete project scenario with db id = {scen_id}." - ) - raise PermissionDenied if request.POST: scenario.delete() messages.success(request, "scenario successfully deleted!") @@ -1487,17 +1398,10 @@ def scenario_delete(request, scen_id): @login_required @require_http_methods(["POST"]) +@user_has_edit_rights def reset_scenario_changes(request, scen_id): scenario = get_object_or_404(Scenario, id=scen_id) - if (scenario.project.user != request.user) and ( - scenario.project.viewers.filter( - user__email=request.user.email, share_rights="edit" - ).exists() - is False - ): - raise PermissionDenied - if request.POST: qs = ParameterChangeTracker.objects.filter(simulation=scenario.simulation) # TODO reverse the changes here @@ -1514,11 +1418,10 @@ def reset_scenario_changes(request, scen_id): @login_required @require_http_methods(["GET", "POST"]) +@user_is_owner def sensitivity_analysis_create(request, scen_id, sa_id=None, step_id=5): excuses_design_under_development(request) scenario = get_object_or_404(Scenario, id=scen_id) - if scenario.project.user != request.user: - raise PermissionDenied if request.method == "GET": if sa_id is not None: @@ -1663,8 +1566,6 @@ def get_timeseries(request, ts_id=None): if request.method == "GET": if ts_id is not None: ts = Timeseries.objects.get(id=ts_id) - if ts.user != request.user and ts.open_source is False: - raise PermissionDenied return JsonResponse({"values": ts.get_values}) @@ -1957,6 +1858,7 @@ def asset_cops_create_or_update( @json_view @login_required @require_http_methods(["GET"]) +@user_is_owner def view_mvs_data_input(request, scen_id=0, testing=False): if scen_id == 0: return JsonResponse( @@ -1966,13 +1868,6 @@ def view_mvs_data_input(request, scen_id=0, testing=False): ) # Load scenario scenario = Scenario.objects.get(id=scen_id) - - if scenario.project.user != request.user: - logger.warning( - f"Unauthorized user tried to access scenario with db id = {scen_id}." - ) - raise PermissionDenied - try: data_clean = format_scenario_for_mvs(scenario, testing) except Exception as e: @@ -1999,6 +1894,7 @@ def test_mvs_data_input(request, scen_id=0): # @json_view @login_required @require_http_methods(["GET", "POST"]) +@user_has_edit_rights def request_mvs_simulation(request, scen_id=0): if scen_id == 0: answer = JsonResponse( @@ -2073,6 +1969,7 @@ def request_mvs_simulation(request, scen_id=0): @login_required @require_http_methods(["GET", "POST"]) +@user_has_edit_rights def request_ezp_simulation(request, scen_id=0): if scen_id == 0: answer = JsonResponse( @@ -2200,19 +2097,12 @@ def fetch_sensitivity_analysis_results(request, sa_id): @login_required @require_http_methods(["GET"]) +@user_is_owner def simulation_cancel(request, scen_id): scenario = get_object_or_404(Scenario, id=scen_id) - if scenario.project.user == request.user: - qs = Simulation.objects.filter(scenario=scen_id) - if qs.exists(): - scenario.simulation.delete() - else: - messages.error( - request, - _( - "You do not have the permission to reset a simulation on a project shared with you" - ), - ) + qs = Simulation.objects.filter(scenario=scen_id) + if qs.exists(): + scenario.simulation.delete() return HttpResponseRedirect( reverse("scenario_review", args=[scenario.project.id, scen_id]) diff --git a/app/templates/error_403.html b/app/templates/error_403.html index 028bbec9..efb2b967 100644 --- a/app/templates/error_403.html +++ b/app/templates/error_403.html @@ -1,70 +1,50 @@ - - +{% extends 'base.html' %} +{% load i18n %} - - - - - - DSO Toolbox - - - - - - - - - - - - - - - - - - +{% block start_body_scripts %} - -




-
-
-
-
-
-

-

403

-
- - -

{% translate "Error! You don't have permission to access the requested resource." %}

-
- - - {% translate "Go to Home Page" %}

- -


+{% endblock %} + +{% block content %} +
+
+
+
+

{% translate "403" %}

+
+ {% translate "Back" %} +
-
-
+ +
+
+
+ + +
+
+

{% translate "Error! You don't have permission to access the requested resource." %}

+
+
+
+

{{ reason }}

+
+
+
- - +{% endblock content %} diff --git a/app/templates/project/project_search.html b/app/templates/project/project_search.html index a52d3089..9e1407b3 100644 --- a/app/templates/project/project_search.html +++ b/app/templates/project/project_search.html @@ -122,6 +122,8 @@

{% translate project.name %}

{% translate "Duplicate" %} + {% endif %} + {% if project.id|has_viewer_read_rights:request.user %}
{% csrf_token %} @@ -131,7 +133,7 @@

{% translate project.name %}

{% translate "Delete" %}
-
+
diff --git a/app/users/models.py b/app/users/models.py index edfe292c..4e21f0bf 100644 --- a/app/users/models.py +++ b/app/users/models.py @@ -2,6 +2,9 @@ class CustomUser(AbstractUser): + def is_owner(self, project): + return project.user == self + def has_edit_rights(self, project): return (project.user.email == self.email) or ( project.viewers.filter(user__email=self.email, share_rights="edit").exists()