OpenConceptLab/ocl_online#224 | Access-control fixes for content in repos the requester can't view

core/bundles/models.py

@@ -11,8 +11,10 @@
 
 
 class Bundle:
-    def __init__(self, root, repo_version, params=None, verbose=False, requested_url=None):  # pylint: disable=too-many-arguments
+    def __init__(  # pylint: disable=too-many-arguments
+            self, root, repo_version, params=None, verbose=False, requested_url=None, user=None):
         params = params or {}
+        self.user = user
         self.repo_version = repo_version
         self.params = params
         self.verbose = verbose
@@ -96,7 +98,14 @@ def set_cascade_direction(self):
             self.reverse = self.params[CASCADE_DIRECTION_PARAM] in TRUTHY
 
     def set_omit_if_exists_in(self):
-        self.omit_if_exists_in = self.params.get(OMIT_IF_EXISTS_IN, None) or None
+        """A repo the user can't view is ignored, as if the expression didn't resolve."""
+        omit_if_exists_in = self.params.get(OMIT_IF_EXISTS_IN, None) or None
+        if omit_if_exists_in:
+            from core.common.models import ConceptContainerModel
+            repo, _ = ConceptContainerModel.resolve_expression_to_version(omit_if_exists_in)
+            if get(repo, 'id') and not repo.has_view_access(self.user):
+                omit_if_exists_in = None
+        self.omit_if_exists_in = omit_if_exists_in
 
     def set_include_self(self):
         if INCLUDE_SELF in self.params:
@@ -183,8 +192,9 @@ def cascade_as_hierarchy(self):
             self.root, context={'request': {'query_params': self.params}}).data
 
     def set_entries(self):
-        self.entries += self.get_concept_serializer()(self.concepts, many=True).data
-        self.entries += self.get_mapping_serializer()(self.mappings, many=True).data
+        context = {'user': self.user}
+        self.entries += self.get_concept_serializer()(self.concepts, many=True, context=context).data
+        self.entries += self.get_mapping_serializer()(self.mappings, many=True, context=context).data
 
     def get_mapping_serializer(self):
         from core.mappings.models import Mapping
@@ -210,7 +220,8 @@ def clone(  # pylint: disable=too-many-arguments
             params=parameters,
             verbose=is_verbose,
             repo_version=clone_from_source,
-            requested_url=requested_url
+            requested_url=requested_url,
+            user=user
         )
         bundle.set_cascade_parameters()
         _parameters = {}

core/code_systems/tests.py

@@ -7,6 +7,7 @@
 from core.code_systems.converter import CodeSystemConverter
 from core.code_systems.serializers import CodeSystemDetailSerializer
 from core.code_systems.views import CodeSystemLookupNotFoundError
+from core.common.constants import ACCESS_TYPE_NONE
 from core.common.tests import OCLTestCase
 from core.concepts.models import Concept
 from core.concepts.tests.factories import ConceptFactory
@@ -259,6 +260,67 @@ def test_lookup_for_code_system_with_new_canonical(self):
                 {'name': 'version', 'valueString': self.org_source_v2.version},
                 {'name': 'display', 'valueString': self.concept_1.display_name}]}))
 
+    @patch('core.sources.models.index_source_concepts', Mock(__name__='index_source_concepts'))
+    @patch('core.sources.models.index_source_mappings', Mock(__name__='index_source_mappings'))
+    def test_private_code_system_answers_as_missing_for_non_viewers(self):  # pylint: disable=too-many-locals
+        private_org = OrganizationFactory()
+        private_source = OrganizationSourceFactory(
+            organization=private_org, canonical_url='/private/url', public_access=ACCESS_TYPE_NONE)
+        concept = ConceptFactory(parent=private_source, names=1, names__name='private_concept_name')
+        Source.persist_new_version(OrganizationSourceFactory.build(
+            version='v1', mnemonic=private_source.mnemonic, organization=private_org,
+            public_access=ACCESS_TYPE_NONE), private_source.created_by)
+        member = UserProfileFactory()
+        private_org.members.add(member)
+        outsider = {'HTTP_AUTHORIZATION': 'Token ' + self.user_token}
+        validate_url = f'/fhir/CodeSystem/$validate-code/?url=/private/url&code={concept.mnemonic}'
+        lookup_url = '/fhir/CodeSystem/$lookup/?system={}&code={}'
+        not_found = self.client.get(lookup_url.format('/unknown/url', concept.mnemonic))
+        self.assertEqual(not_found.status_code, 400)
+
+        public_concept = ConceptFactory()
+        selector = (f'&org={public_concept.parent.organization.mnemonic}&source={public_concept.parent.mnemonic}'
+                    f'&concept={public_concept.mnemonic}')
+        for headers in [{}, outsider]:
+            # without a code, a code system the requester can't view still answers as one that doesn't exist
+            for system_url in ['/private/url', '/unknown/url']:
+                response = self.client.get(f'/fhir/CodeSystem/$validate-code/?url={system_url}{selector}', **headers)
+                self.assertFalse(response.data['parameter'][0]['valueBoolean'], (system_url, headers))
+            for url in [validate_url, f'{validate_url}&display={concept.display_name}']:
+                response = self.client.get(url, **headers)
+                self.assertEqual(response.status_code, 200, (url, headers))
+                self.assertFalse(response.data['parameter'][0]['valueBoolean'], (url, headers))
+            for code in [concept.mnemonic, 'missing']:
+                response = self.client.get(lookup_url.format('/private/url', code), **headers)
+                self.assertEqual(response.status_code, 400, (code, headers))
+                self.assertEqual(response.content, not_found.content, (code, headers))
+
+        member_headers = {'HTTP_AUTHORIZATION': 'Token ' + member.get_token()}
+        response = self.client.get(validate_url, **member_headers)
+        self.assertTrue(response.data['parameter'][0]['valueBoolean'])
+        response = self.client.get(lookup_url.format('/private/url', concept.mnemonic), **member_headers)
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(response.data['parameter'][2]['valueString'], concept.display_name)
+
+    @patch('core.sources.models.index_source_concepts', Mock(__name__='index_source_concepts'))
+    @patch('core.sources.models.index_source_mappings', Mock(__name__='index_source_mappings'))
+    def test_private_code_system_does_not_hide_public_one_with_same_url(self):
+        private_source = UserSourceFactory(user=self.user, public_access=ACCESS_TYPE_NONE, canonical_url='/shared/url')
+        Source.persist_new_version(UserSourceFactory.build(
+            version='v1', mnemonic=private_source.mnemonic, user=self.user), private_source.created_by)
+        public_source = OrganizationSourceFactory(canonical_url='/shared/url')
+        concept = ConceptFactory(parent=public_source)
+        Source.persist_new_version(OrganizationSourceFactory.build(
+            version='v1', mnemonic=public_source.mnemonic, organization=public_source.organization),
+            public_source.created_by)
+
+        response = self.client.get(f'/fhir/CodeSystem/$lookup/?system=/shared/url&code={concept.mnemonic}')
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(response.data['parameter'][0]['valueString'], public_source.mnemonic)
+
+        response = self.client.get(f'/fhir/CodeSystem/$validate-code/?url=/shared/url&code={concept.mnemonic}')
+        self.assertTrue(response.data['parameter'][0]['valueBoolean'])
+
     @patch('core.sources.models.index_source_concepts', Mock(__name__='index_source_concepts'))
     @patch('core.sources.models.index_source_mappings', Mock(__name__='index_source_mappings'))
     def test_post_code_system_without_concepts(self):

core/code_systems/views.py

@@ -1,7 +1,7 @@
 import logging
 
 from django.db.models import F
-from rest_framework.exceptions import ValidationError, NotAuthenticated
+from rest_framework.exceptions import ValidationError, NotAuthenticated, PermissionDenied
 
 from core.bundles.serializers import FHIRBundleSerializer
 from core.code_systems.serializers import CodeSystemDetailSerializer, \
@@ -18,6 +18,16 @@
 logger = logging.getLogger('oclapi')
 
 
+def get_viewable_repo(repos, user):
+    """The first of these repo versions the user can view, skipping the rest as if they didn't exist. Its HEAD,
+    which governs access to its concepts, must be viewable too."""
+    for repo in repos.order_by('id'):
+        head = repo.head
+        if head and repo.has_view_access(user) and head.has_view_access(user):
+            return repo
+    return None
+
+
 class CodeSystemListView(SourceListView):
     serializer_class = CodeSystemDetailSerializer
 
@@ -96,14 +106,22 @@ def get_queryset(self):
                 else:
                     source = source.filter(is_latest_version=True).exclude(version=HEAD)
 
+                source = get_viewable_repo(source, self.request.user)
                 if source:
-                    queryset = queryset.filter(sources=source.first(), mnemonic=code)
+                    queryset = queryset.filter(sources=source, mnemonic=code)
                     if not queryset.exists():
                         raise CodeSystemLookupNotFoundError(code)
                     return queryset
 
         raise CodeSystemLookupNotFoundError()
 
+    def get_object(self, queryset=None):
+        """A code the requester can't view gives the same answer as a code system that doesn't exist."""
+        try:
+            return super().get_object(queryset)
+        except (NotAuthenticated, PermissionDenied) as ex:
+            raise CodeSystemLookupNotFoundError() from ex
+
     def get_serializer(self, instance=None):  # pylint: disable=arguments-differ
         if instance:
             return ParametersSerializer.from_concept(instance)
@@ -147,19 +165,19 @@ def get_queryset(self):
                 url = IdentifierSerializer.convert_fhir_url_to_ocl_uri(url, 'sources')
                 source = Source.objects.filter(uri=url)
 
-            if not source:
-                return queryset.none()
-
             if version:
                 source = source.filter(version=version)
             else:
                 source = source.filter(is_latest_version=True).exclude(version=HEAD)
+            source = get_viewable_repo(source, self.request.user)
+            if not source:
+                return queryset.none()
 
         if code:
             if not source:
                 return queryset.none()
 
-            queryset = queryset.filter(sources=source.first(), mnemonic=code)
+            queryset = queryset.filter(sources=source, mnemonic=code)
 
         if display and queryset:
             instance = queryset.first()
@@ -187,7 +205,7 @@ def get_object(self, queryset=None):
         if instance:
             try:
                 self.check_object_permissions(self.request, instance)
-            except NotAuthenticated:
+            except (NotAuthenticated, PermissionDenied):
                 return {
                     'parameter': [
                         {'name': 'result', 'valueBoolean': False}

core/collections/models.py

@@ -877,6 +877,12 @@ def get_concept_cascade_params(self, system_version=None):
                          'reverse', 'omit_if_exists_in']:
                 if attr in self.cascade:
                     cascade_params[attr] = get(self.cascade, attr)
+            omit_if_exists_in = cascade_params.get('omit_if_exists_in')
+            if omit_if_exists_in:
+                # a repo the reference's creator can't view is ignored, as if the expression didn't resolve
+                repo, _ = ConceptContainerModel.resolve_expression_to_version(omit_if_exists_in)
+                if get(repo, 'id') and not repo.has_view_access(self.created_by):
+                    cascade_params.pop('omit_if_exists_in')
             map_types = get(self.cascade, 'map_types', None)
             exclude_map_types = get(self.cascade, 'exclude_map_types', None)
             return_map_types = get(self.cascade, 'return_map_types', None)