[#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

@@ -9,11 +9,13 @@
 
 from open_inwoner.accounts.choices import LoginTypeChoices
 from open_inwoner.accounts.tests.factories import UserFactory
+from open_inwoner.accounts.views.cases import SimpleFile
+from open_inwoner.utils.test import paginated_response
 
-from ...accounts.views.cases import SimpleFile
 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/"
 
@@ -31,6 +33,12 @@ def setUpTestData(self):
             login_type=LoginTypeChoices.digid, bsn="900222086", email="[email protected]"
         )
         self.config = OpenZaakConfig.get_solo()
+        self.zaak_service = ServiceFactory(api_root=ZAKEN_ROOT, api_type=APITypes.zrc)
+        self.config.zaak_service = self.zaak_service
+        self.catalogi_service = ServiceFactory(
+            api_root=CATALOGI_ROOT, api_type=APITypes.ztc
+        )
+        self.config.catalogi_service = self.catalogi_service
         self.document_service = ServiceFactory(
             api_root=DOCUMENTEN_ROOT, api_type=APITypes.drc
         )
@@ -40,6 +48,35 @@ 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.zaak_informatie_object = generate_oas_component(
+            "zrc",
+            "schemas/ZaakInformatieObject",
+            url=f"{ZAKEN_ROOT}zaakinformatieobjecten/e55153aa-ad2c-4a07-ae75-15add57d6",
+            informatieobject=f"{DOCUMENTEN_ROOT}enkelvoudiginformatieobjecten/014c38fe-b010-4412-881c-3000032fb812",
+            zaak=f"{ZAKEN_ROOT}zaken/d8bbdeb7-770f-4ca9-b1ea-77b4730bf67d",
+            aard_relatie_weergave="some content",
+            titel="",
+            beschrijving="",
+            registratiedatum="2021-01-12",
+        )
+        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,13 +97,33 @@ 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"],
                 },
             ),
         )
 
-    def _setUpMocks(self, m):
+    def _setUpOASMocks(self, m):
+        mock_service_oas_get(m, ZAKEN_ROOT, "zrc")
+        mock_service_oas_get(m, CATALOGI_ROOT, "ztc")
         mock_service_oas_get(m, DOCUMENTEN_ROOT, "drc")
+
+    def _setUpAccessMocks(self, m):
+        # the minimal mocks needed to be able to access the information object
+        self._setUpOASMocks(m)
+        m.get(self.zaak["url"], json=self.zaak)
+        m.get(
+            f"{ZAKEN_ROOT}rollen?zaak={self.zaak['url']}&betrokkeneIdentificatie__natuurlijkPersoon__inpBsn={self.user.bsn}",
+            json=paginated_response([self.role]),
+        )
+        m.get(
+            f"{ZAKEN_ROOT}zaakinformatieobjecten?zaak={self.zaak['url']}&informatieobject={self.informatie_object['url']}",
+            # note the real API doesn't return a paginated_response here
+            json=[self.zaak_informatie_object],
+        )
+
+    def _setUpMocks(self, m):
+        self._setUpAccessMocks(m)
         m.get(self.informatie_object["url"], json=self.informatie_object)
         m.get(self.informatie_object["inhoud"], content=self.informatie_object_content)
 
@@ -75,7 +132,8 @@ def test_document_content_is_retrieved_when_user_logged_in_via_digid(self, m):
         url = reverse(
             "accounts:case_document_download",
             kwargs={
-                "object_id": self.informatie_object["uuid"],
+                "object_id": self.zaak["uuid"],
+                "info_id": self.informatie_object["uuid"],
             },
         )
         response = self.app.get(url, user=self.user)
@@ -94,7 +152,8 @@ def test_document_content_is_retrieved_when_user_logged_in_via_digid(self, m):
         )
 
     def test_document_content_with_bad_status_is_http_403(self, m):
-        mock_service_oas_get(m, DOCUMENTEN_ROOT, "drc")
+        self._setUpAccessMocks(m)
+
         info_object = generate_oas_component(
             "drc",
             "schemas/EnkelvoudigInformatieObject",
@@ -109,13 +168,15 @@ def test_document_content_with_bad_status_is_http_403(self, m):
         url = reverse(
             "accounts:case_document_download",
             kwargs={
-                "object_id": info_object["uuid"],
+                "object_id": self.zaak["uuid"],
+                "info_id": info_object["uuid"],
             },
         )
         self.app.get(url, user=self.user, status=403)
 
     def test_document_content_with_bad_confidentiality_is_http_403(self, m):
-        mock_service_oas_get(m, DOCUMENTEN_ROOT, "drc")
+        self._setUpAccessMocks(m)
+
         info_object = generate_oas_component(
             "drc",
             "schemas/EnkelvoudigInformatieObject",
@@ -130,7 +191,8 @@ def test_document_content_with_bad_confidentiality_is_http_403(self, m):
         url = reverse(
             "accounts:case_document_download",
             kwargs={
-                "object_id": info_object["uuid"],
+                "object_id": self.zaak["uuid"],
+                "info_id": info_object["uuid"],
             },
         )
         self.app.get(url, user=self.user, status=403)
@@ -156,27 +218,58 @@ def test_anonymous_user_has_no_access_to_download_page(self, m):
             f"{reverse('login')}?next={self.informatie_object_file.url}",
         )
 
+    def test_no_data_is_retrieved_when_case_object_http_404(self, m):
+        self._setUpOASMocks(m)
+        m.get(self.zaak["url"], status_code=404)
+
+        self.app.get(self.informatie_object_file.url, user=self.user, status=404)
+
+    def test_no_data_is_retrieved_when_no_related_roles_are_found_for_user_bsn(self, m):
+        self._setUpOASMocks(m)
+        m.get(self.zaak["url"], json=self.zaak)
+        m.get(
+            f"{ZAKEN_ROOT}rollen?zaak={self.zaak['url']}&betrokkeneIdentificatie__natuurlijkPersoon__inpBsn={self.user.bsn}",
+            # no roles found
+            json=paginated_response([]),
+        )
+        self.app.get(self.informatie_object_file.url, user=self.user, status=403)
+
+    def test_no_data_is_retrieved_when_no_matching_case_info_object_is_found(self, m):
+        self._setUpOASMocks(m)
+        m.get(self.zaak["url"], json=self.zaak)
+        m.get(
+            f"{ZAKEN_ROOT}rollen?zaak={self.zaak['url']}&betrokkeneIdentificatie__natuurlijkPersoon__inpBsn={self.user.bsn}",
+            json=paginated_response([self.role]),
+        )
+        m.get(self.informatie_object["url"], json=self.informatie_object)
+        m.get(
+            f"{ZAKEN_ROOT}zaakinformatieobjecten?zaak={self.zaak['url']}&informatieobject={self.informatie_object['url']}",
+            # no case info objects found
+            json=[],
+        )
+        self.app.get(self.informatie_object_file.url, user=self.user, status=403)
+
     def test_no_data_is_retrieved_when_info_object_http_404(self, m):
-        mock_service_oas_get(m, DOCUMENTEN_ROOT, "drc")
+        self._setUpAccessMocks(m)
         m.get(self.informatie_object["url"], status_code=404)
 
         self.app.get(self.informatie_object_file.url, user=self.user, status=404)
 
     def test_no_data_is_retrieved_when_info_object_http_500(self, m):
-        mock_service_oas_get(m, DOCUMENTEN_ROOT, "drc")
+        self._setUpAccessMocks(m)
         m.get(self.informatie_object["url"], status_code=500)
 
         self.app.get(self.informatie_object_file.url, user=self.user, status=404)
 
     def test_no_data_is_retrieved_when_document_download_data_http_404(self, m):
-        mock_service_oas_get(m, DOCUMENTEN_ROOT, "drc")
+        self._setUpAccessMocks(m)
         m.get(self.informatie_object["url"], json=self.informatie_object)
         m.get(self.informatie_object["inhoud"], status_code=404)
 
         self.app.get(self.informatie_object_file.url, user=self.user, status=404)
 
     def test_no_data_is_retrieved_when_document_download_data_http_500(self, m):
-        mock_service_oas_get(m, DOCUMENTEN_ROOT, "drc")
+        self._setUpAccessMocks(m)
         m.get(self.informatie_object["url"], json=self.informatie_object)
         m.get(self.informatie_object["inhoud"], status_code=500)