WIP [#912] Harden access control on case-details/documents

Author: Bart van der Schoor

src/open_inwoner/accounts/urls.py

@@ -87,7 +87,7 @@
     path("themes/", MyCategoriesView.as_view(), name="my_themes"),
     path("cases/", CasesListView.as_view(), name="my_cases"),
     path(
-        "cases/document/<str:object_id>/",
+        "cases/<str:object_id>/document/<str:info_id>/",
         CasesDocumentDownloadView.as_view(),
         name="case_document_download",
     ),

src/open_inwoner/accounts/views/cases.py

@@ -17,6 +17,7 @@
 from zgw_consumers.concurrent import parallel
 
 from open_inwoner.openzaak.cases import (
+    fetch_case_roles_for_bsn,
     fetch_case_types,
     fetch_cases,
     fetch_single_case,
@@ -26,6 +27,7 @@
     InformatieObject,
     create_document_content_stream,
     fetch_case_information_objects,
+    fetch_case_information_objects_for_case_and_info,
     fetch_single_information_object,
     fetch_single_information_object_uuid,
 )
@@ -174,6 +176,10 @@ def get_context_data(self, **kwargs):
         case = fetch_single_case(case_uuid)
 
         if case:
+            # check if we have a role this case
+            if not fetch_case_roles_for_bsn(case.url, self.request.user.bsn):
+                raise PermissionDenied()
+
             documents = self.get_case_document_files(case)
 
             statuses = fetch_status_history(case.url)
@@ -243,7 +249,8 @@ def get_case_document_files(self, case) -> List[SimpleFile]:
                     url=reverse(
                         "accounts:case_document_download",
                         kwargs={
-                            "object_id": info_obj.uuid,
+                            "object_id": case.uuid,
+                            "info_id": info_obj.uuid,
                         },
                     ),
                 )
@@ -273,12 +280,27 @@ def handle_no_permission(self):
         return super().handle_no_permission()
 
     def get(self, *args, **kwargs):
-        info_object_uuid = kwargs["object_id"]
+        case_uuid = kwargs["object_id"]
+        case = fetch_single_case(case_uuid)
+        if not case:
+            raise Http404
+
+        # check if we have a role this case
+        if not fetch_case_roles_for_bsn(case.url, self.request.user.bsn):
+            raise PermissionDenied()
 
+        info_object_uuid = kwargs["info_id"]
         info_object = fetch_single_information_object_uuid(info_object_uuid)
         if not info_object:
             raise Http404
 
+        # check if this info_object belongs to this case
+        if not fetch_case_information_objects_for_case_and_info(
+            case.url, info_object.url
+        ):
+            raise PermissionDenied()
+
+        # check if this info_object should be visible
         config = OpenZaakConfig.get_solo()
         if not filter_info_object_visibility(
             info_object, config.document_max_confidentiality

src/open_inwoner/openzaak/cases.py

@@ -7,7 +7,7 @@
 from zds_client import ClientError
 from zgw_consumers.api_models.base import factory
 from zgw_consumers.api_models.catalogi import ZGWModel
-from zgw_consumers.api_models.zaken import Zaak
+from zgw_consumers.api_models.zaken import Rol, Zaak
 from zgw_consumers.service import get_paginated_results
 
 from .clients import build_client
@@ -152,3 +152,31 @@ def fetch_single_case_type(case_type_url: str) -> Optional[ZaakType]:
     case_type = factory(ZaakType, response)
 
     return case_type
+
+
+def fetch_case_roles_for_bsn(case_url: str, bsn: str) -> List[Rol]:
+    client = build_client("zaak")
+
+    if client is None:
+        return []
+
+    try:
+        response = client.list(
+            "rol",
+            request_kwargs={
+                "params": {
+                    "zaak": case_url,
+                    "betrokkeneIdentificatie__natuurlijkPersoon__inpBsn": bsn,
+                }
+            },
+        )
+    except RequestException as e:
+        logger.exception("exception while making request", exc_info=e)
+        return []
+    except ClientError as e:
+        logger.exception("exception while making request", exc_info=e)
+        return []
+
+    roles = factory(Rol, response["results"])
+
+    return roles

src/open_inwoner/openzaak/info_objects.py

@@ -79,6 +79,36 @@ def fetch_case_information_objects(case_url: str) -> List[ZaakInformatieObject]:
     return case_info_objects
 
 
+def fetch_case_information_objects_for_case_and_info(
+    case_url: str, info_object_url: str
+) -> List[ZaakInformatieObject]:
+    client = build_client("zaak")
+
+    if client is None:
+        return []
+
+    try:
+        response = client.list(
+            "zaakinformatieobject",
+            request_kwargs={
+                "params": {
+                    "zaak": case_url,
+                    "informatieobject": info_object_url,
+                },
+            },
+        )
+    except RequestException as e:
+        logger.exception("exception while making request", exc_info=e)
+        return []
+    except ClientError as e:
+        logger.exception("exception while making request", exc_info=e)
+        return []
+
+    case_info_objects = factory(ZaakInformatieObject, response)
+
+    return case_info_objects
+
+
 def fetch_single_information_object(info_object_url: str) -> Optional[InformatieObject]:
     client = build_client("document")
 

src/open_inwoner/openzaak/tests/test_documents.py

@@ -11,9 +11,11 @@
 from open_inwoner.accounts.tests.factories import UserFactory
 
 from ...accounts.views.cases import SimpleFile
+from ...utils.test import paginated_response
 from ..models import OpenZaakConfig
 from .factories import ServiceFactory
 
+ZAKEN_ROOT = "https://zaken.nl/api/v1/"
 CATALOGI_ROOT = "https://catalogi.nl/api/v1/"
 DOCUMENTEN_ROOT = "https://documenten.nl/api/v1/"
 
@@ -40,6 +42,24 @@ def setUpTestData(self):
         )
         self.config.save()
 
+        self.zaak = generate_oas_component(
+            "zrc",
+            "schemas/Zaak",
+            uuid="d8bbdeb7-770f-4ca9-b1ea-77b4730bf67d",
+            url=f"{ZAKEN_ROOT}zaken/d8bbdeb7-770f-4ca9-b1ea-77b4730bf67d",
+            zaaktype=f"{CATALOGI_ROOT}zaaktypen/53340e34-7581-4b04-884f",
+            identificatie="ZAAK-2022-0000000024",
+            omschrijving="Zaak naar aanleiding van ingezonden formulier",
+            startdatum="2022-01-02",
+            einddatum=None,
+            status=f"{ZAKEN_ROOT}statussen/3da89990-c7fc-476a-ad13-c9023450083c",
+        )
+        self.role = generate_oas_component(
+            "zrc",
+            "schemas/Rol",
+            url=f"{ZAKEN_ROOT}rollen/f33153aa-ad2c-4a07-ae75-15add5891",
+            betrokkene_identificatie="foo",
+        )
         self.informatie_object_content = "my document content".encode("utf8")
         self.informatie_object = generate_oas_component(
             "drc",
@@ -60,7 +80,8 @@ def setUpTestData(self):
             url=reverse(
                 "accounts:case_document_download",
                 kwargs={
-                    "object_id": self.informatie_object["uuid"],
+                    "object_id": self.zaak["uuid"],
+                    "info_id": self.informatie_object["uuid"],
                 },
             ),
         )
@@ -69,6 +90,10 @@ def _setUpMocks(self, m):
         mock_service_oas_get(m, DOCUMENTEN_ROOT, "drc")
         m.get(self.informatie_object["url"], json=self.informatie_object)
         m.get(self.informatie_object["inhoud"], content=self.informatie_object_content)
+        m.get(
+            f"{ZAKEN_ROOT}rollen?zaak={self.zaak['url']}&betrokkeneIdentificatie__natuurlijkPersoon__inpBsn={self.user.bsn}",
+            json=paginated_response([self.role]),
+        )
 
     def test_document_content_is_retrieved_when_user_logged_in_via_digid(self, m):
         self._setUpMocks(m)

src/open_inwoner/openzaak/tests/test_statuses.py

@@ -6,6 +6,7 @@
 
 import requests_mock
 from django_webtest import WebTest
+from furl import furl
 from zgw_consumers.api_models.base import factory
 from zgw_consumers.api_models.catalogi import StatusType
 from zgw_consumers.api_models.constants import VertrouwelijkheidsAanduidingen
@@ -58,6 +59,7 @@ def setUpTestData(self):
         self.zaak = generate_oas_component(
             "zrc",
             "schemas/Zaak",
+            uuid="d8bbdeb7-770f-4ca9-b1ea-77b4730bf67d",
             url=f"{ZAKEN_ROOT}zaken/d8bbdeb7-770f-4ca9-b1ea-77b4730bf67d",
             zaaktype=f"{CATALOGI_ROOT}zaaktypen/53340e34-7581-4b04-884f",
             identificatie="ZAAK-2022-0000000024",
@@ -126,6 +128,12 @@ def setUpTestData(self):
             volgnummer=2,
             is_eindstatus=False,
         )
+        self.role = generate_oas_component(
+            "zrc",
+            "schemas/Rol",
+            url=f"{ZAKEN_ROOT}rollen/f33153aa-ad2c-4a07-ae75-15add5891",
+            betrokkene_identificatie="foo",
+        )
         self.zaak_informatie_object = generate_oas_component(
             "zrc",
             "schemas/ZaakInformatieObject",
@@ -185,7 +193,8 @@ def setUpTestData(self):
             url=reverse(
                 "accounts:case_document_download",
                 kwargs={
-                    "object_id": self.informatie_object["uuid"],
+                    "object_id": self.zaak["uuid"],
+                    "info_id": self.informatie_object["uuid"],
                 },
             ),
         )
@@ -206,6 +215,10 @@ def _setUpMocks(self, m):
             f"{ZAKEN_ROOT}statussen?zaak={self.zaak['url']}",
             json=paginated_response([self.status1, self.status2]),
         )
+        m.get(
+            f"{ZAKEN_ROOT}rollen?zaak={self.zaak['url']}&betrokkeneIdentificatie__natuurlijkPersoon__inpBsn={self.user.bsn}",
+            json=paginated_response([self.role]),
+        )
         m.get(f"{CATALOGI_ROOT}zaaktypen/53340e34-7581-4b04-884f", json=self.zaaktype)
         m.get(
             f"{CATALOGI_ROOT}statustypen?zaaktype={self.zaaktype['url']}",
@@ -324,6 +337,9 @@ def test_anonymous_user_has_no_access_to_status_page(self, m):
             f"{reverse('login')}?next={reverse('accounts:case_status', kwargs={'object_id': 'd8bbdeb7-770f-4ca9-b1ea-77b4730bf67d'})}",
         )
 
+    def test_access_with_role(self, m):
+        self.fail("implement me")
+
     def test_no_data_is_retrieved_when_http_404(self, m):
         mock_service_oas_get(m, ZAKEN_ROOT, "zrc")
         mock_service_oas_get(m, CATALOGI_ROOT, "ztc")