BB2-4266 - C4DIC Endpoint

Makefile

@@ -20,4 +20,8 @@ build-local:
 	cd dev-local ; make build-local ; cd ..
 
 run-local:
-	cd dev-local ; make run-local ; cd ..
+	cd dev-local ; make run-local ; cd ..
+
+exec-web:
+	cd dev-local ; make exec-web ; cd ..
+

apps/authorization/permissions.py

@@ -9,7 +9,8 @@ class DataAccessGrantPermission(permissions.BasePermission):
     """
     Permission check for a Grant related to the token used.
     """
-    def has_permission(self, request, view):
+
+    def has_permission(self, request, view) -> bool:  # type: ignore
         dag = None
         try:
             dag = DataAccessGrant.objects.get(

apps/capabilities/management/commands/create_blue_button_scopes.py

@@ -260,6 +260,20 @@ def create_coverage_read_search_capability(group,
                                                protected_resources=json.dumps(pr, indent=4))
     return c
 
+def create_insurance_card_capability(group, fhir_prefix, title="Digital Insurance Card access."):
+    c = None
+    description = "Digital Insurance Card"
+    # TODO - this is not a real FHIR resource or scope, decision on how we want ot handle this
+    smart_scope_string = "patient/DigitalInsuranceCard.read"
+    pr = []
+    pr.append(["GET", "%sDigitalInsuranceCard[/]?$" % fhir_prefix])
+    if not ProtectedCapability.objects.filter(slug=smart_scope_string).exists():
+        c = ProtectedCapability.objects.create(group=group,
+                                               title=title,
+                                               description=description,
+                                               slug=smart_scope_string,
+                                               protected_resources=json.dumps(pr, indent=4))
+    return c
 
 def create_launch_capability(group, fhir_prefix, title="Patient launch context."):
 
@@ -296,5 +310,6 @@ def handle(self, *args, **options):
         create_coverage_read_capability(g, fhir_prefix)
         create_coverage_search_capability(g, fhir_prefix)
         create_coverage_read_search_capability(g, fhir_prefix)
+        create_insurance_card_capability(g, fhir_prefix)
         create_launch_capability(g, fhir_prefix)
         create_openid_capability(g)

apps/capabilities/permissions.py

@@ -8,6 +8,10 @@
 
 from .models import ProtectedCapability
 
+import apps.logging.request_logger as logging
+
+logger = logging.getLogger(logging.DEBUG_GENERAL)
+
 
 class BBCapabilitiesPermissionTokenScopeMissingException(APIException):
     # BB2-237 custom exception
@@ -16,7 +20,8 @@ class BBCapabilitiesPermissionTokenScopeMissingException(APIException):
 
 class TokenHasProtectedCapability(permissions.BasePermission):
 
-    def has_permission(self, request, view):
+    def has_permission(self, request, view) -> bool:  # type: ignore
+        logger.warning({"has_permission": "start"})
         token = request.auth
         access_token_query_param = request.GET.get("access_token", None)
 
@@ -27,9 +32,11 @@ def has_permission(self, request, view):
             )
 
         if not token:
+            logger.warning("has_permission: not token")
             return False
 
         if not switch_is_active("require-scopes"):
+            logger.warning("has_permission: switch_is_active('require-scopes')")
             return True
 
         if hasattr(token, "scope"):  # OAuth 2
@@ -43,17 +50,33 @@ def has_permission(self, request, view):
                 slug__in=token_scopes
             ).values_list('protected_resources', flat=True).all())
 
+            logger.warning({"token_scopes": token_scopes, "scopes": scopes})
+
             for scope in scopes:
                 for method, path in json.loads(scope):
+                    logger.warning({"scope in scopes": scope,
+                                    "method": method,
+                                    "path": path,
+                                    "request.method": request.method,
+                                    "request.path": request.path})
                     if method != request.method:
+                        logger.warning({"A": 1})
+                        logger.warning({"request_method": request.method})
                         continue
                     if path == request.path:
+                        logger.warning({"A": 2})
+                        logger.warning({"path == request.path": (path == request.path)})
                         return True
                     if re.fullmatch(path, request.path) is not None:
+                        logger.warning({"A": 3})
                         return True
+                    logger.warning({"end-of-scope-in-scopes loop": "here"})
+
+            logger.warning("has_permission: scope not matched/found")
             return False
         else:
             # BB2-237: Replaces ASSERT with exception. We should never reach here.
             mesg = ("TokenHasScope requires the `oauth2_provider.rest_framework.OAuth2Authentication`"
                     " authentication class to be used.")
+            logger.warning("has_permission: end of line scope missing exception")
             raise BBCapabilitiesPermissionTokenScopeMissingException(mesg)

apps/fhir/bluebutton/constants.py

@@ -1,3 +1,3 @@
-ALLOWED_RESOURCE_TYPES = ['Patient', 'Coverage', 'ExplanationOfBenefit']
+ALLOWED_RESOURCE_TYPES = ['Patient', 'Coverage', 'ExplanationOfBenefit', 'Bundle']
 DEFAULT_PAGE_SIZE = 10
 MAX_PAGE_SIZE = 50

apps/fhir/bluebutton/permissions.py

@@ -4,7 +4,7 @@
 from django.contrib.auth import get_user_model
 from rest_framework import permissions, exceptions
 from rest_framework.exceptions import AuthenticationFailed
-from .constants import ALLOWED_RESOURCE_TYPES
+from apps.fhir.bluebutton.constants import ALLOWED_RESOURCE_TYPES
 from apps.versions import Versions, VersionNotMatched
 
 import apps.logging.request_logger as bb2logging
@@ -72,7 +72,7 @@ def has_object_permission(self, request, view, obj):
 
 
 class SearchCrosswalkPermission(HasCrosswalk):
-    def has_object_permission(self, request, view, obj):
+    def has_object_permission(self, request, view, obj) -> bool:  # type: ignore
         if view.version in Versions.supported_versions():
             patient_id = request.crosswalk.fhir_id(view.version)
         else:
@@ -106,3 +106,8 @@ def has_permission(self, request, view):
             )
 
         return True
+
+
+class AlwaysDeny(permissions.BasePermission):
+    def has_permission(self, request, view):
+        return False

apps/fhir/bluebutton/tests/test_fhir_resources_read_search_w_validation.py

@@ -224,7 +224,7 @@ def _search_eob_by_parameter_tag(self, version=1):
         def catchall_w_tag_qparam(url, req):
             # this is called in case EOB search with good tag
             self.assertIn(f'{FHIR_SERVER["FHIR_URL"]}/v{version}/fhir/ExplanationOfBenefit/', req.url)
-            self.assertIn('_format=application%2Fjson%2Bfhir', req.url)
+            self.assertIn('_format=application%2Ffhir%2Bjson', req.url)
             # parameters encoded in prepared request's body
             self.assertTrue(('_tag=Adjudicated' in req.url) or ('_tag=PartiallyAdjudicated' in req.url))
 
@@ -236,7 +236,7 @@ def catchall_w_tag_qparam(url, req):
         @all_requests
         def catchall(url, req):
             self.assertIn(f'{FHIR_SERVER["FHIR_URL"]}/v{version}/fhir/ExplanationOfBenefit/', req.url)
-            self.assertIn('_format=application%2Fjson%2Bfhir', req.url)
+            self.assertIn('_format=application%2Ffhir%2Bjson', req.url)
 
             return {
                 'status_code': 200,
@@ -280,7 +280,7 @@ def _search_eob_by_parameters_request(self, version=1):
         @all_requests
         def catchall(url, req):
             self.assertIn(f'{FHIR_SERVER["FHIR_URL"]}/v{version}/fhir/ExplanationOfBenefit/', req.url)
-            self.assertIn('_format=application%2Fjson%2Bfhir', req.url)
+            self.assertIn('_format=application%2Ffhir%2Bjson', req.url)
 
             return {
                 'status_code': 200,

apps/fhir/bluebutton/tests/test_insurancecard.py

@@ -0,0 +1,62 @@
+
+from django.conf import settings
+from django.test.client import Client
+from django.urls import reverse
+from httmock import all_requests, HTTMock
+from http import HTTPStatus
+from oauth2_provider.models import get_access_token_model
+
+from apps.test import BaseApiTest
+
+# Get the pre-defined Conformance statement
+from waffle.testutils import override_switch
+
+AccessToken = get_access_token_model()
+
+FHIR_ID_V2 = settings.DEFAULT_SAMPLE_FHIR_ID_V2
+
+POSSIBLE_COVERAGE_SCOPES = ['patient/Coverage.read', 'patient/Coverage.rs', 'patient/Coverage.s']
+POSSIBLE_PATIENT_SCOPES = ['patient/Patient.read', 'patient/Patient.rs', 'patient/Patient.r']
+
+
+class InsuranceCardTest(BaseApiTest):
+    TEST_TABLE = []
+    # Add all of the passing combos. These should be 200s.
+    for C in POSSIBLE_COVERAGE_SCOPES:
+        for P in POSSIBLE_PATIENT_SCOPES:
+            TEST_TABLE.append({'status': HTTPStatus.OK, 'scope': [C, P]})
+    # If we only have one, those should fail with a 403.
+    for C in POSSIBLE_COVERAGE_SCOPES:
+        TEST_TABLE.append({'status': HTTPStatus.FORBIDDEN, 'scope': [C]})
+    for P in POSSIBLE_PATIENT_SCOPES:
+        TEST_TABLE.append({'status': HTTPStatus.FORBIDDEN, 'scope': [P]})
+
+    def setUp(self):
+        # create read and write capabilities
+        self.read_capability = self._create_capability('Read', [])
+        self.write_capability = self._create_capability('Write', [])
+        self.client = Client()
+
+    @override_switch('v3_endpoints', active=True)
+    def test_scope_combinations(self):
+        for tt in InsuranceCardTest.TEST_TABLE:
+            with self.subTest(tt=tt):
+                token = self.create_token('Annie', 'User', fhir_id_v2=FHIR_ID_V2)
+                ac = AccessToken.objects.get(token=token)
+                ac.scope = " ".join(tt['scope'])
+                ac.save()
+
+                @all_requests
+                def catchall(url, req):
+                    return {
+                        'status_code': 200,
+                        'content': {
+                            'doesnot': 'matter',
+                        }
+                    }
+                with HTTMock(catchall):
+                    response = self.client.get(
+                        reverse('bb_oauth_fhir_dic_read'),
+                        Authorization='Bearer %s' % (token)
+                    )
+                    self.assertEqual(response.status_code, tt['status'])

apps/fhir/bluebutton/tests/test_read_and_search.py

@@ -26,7 +26,7 @@
 def get_expected_read_request(version: int):
     return {
         'method': 'GET',
-        'url': f'{FHIR_SERVER["FHIR_URL"]}/v{version}/fhir/Patient/{FHIR_ID_V2}/?_format=json&_id={FHIR_ID_V2}',
+        'url': f'{FHIR_SERVER["FHIR_URL"]}/v{version}/fhir/Patient/{FHIR_ID_V2}/?_format=application/fhir+json&_id={FHIR_ID_V2}',
         'headers': {
             # 'User-Agent': 'python-requests/2.20.0',
             'Accept-Encoding': 'gzip, deflate',
@@ -284,7 +284,7 @@ def _search_request(self, version: int = 1):
         @all_requests
         def catchall(url, req):
             self.assertIn(f'{FHIR_SERVER["FHIR_URL"]}/v{version}/fhir/Patient/', req.url)
-            self.assertIn('_format=application%2Fjson%2Bfhir', req.url)
+            self.assertIn('_format=application%2Ffhir%2Bjson', req.url)
             self.assertIn(f'_id={FHIR_ID_V2}', req.url)
             self.assertIn('_count=5', req.url)
             self.assertNotIn('hello', req.url)
@@ -362,7 +362,7 @@ def _search_request_not_found(self, version: int = 1):
         @all_requests
         def catchall(url, req):
             self.assertIn(f'{FHIR_SERVER["FHIR_URL"]}/v{version}/fhir/Patient/', req.url)
-            self.assertIn('_format=application%2Fjson%2Bfhir', req.url)
+            self.assertIn('_format=application%2Ffhir%2Bjson', req.url)
             self.assertIn(f'_id={FHIR_ID_V2}', req.url)
             self.assertEqual(expected_request['method'], req.method)
             self.assertDictContainsSubset(expected_request['headers'], req.headers)
@@ -444,7 +444,7 @@ def _search_request_failed(self, version: int = 1, bfd_status_code=500):
         @all_requests
         def catchall(url, req):
             self.assertIn(f'{FHIR_SERVER["FHIR_URL"]}/v{version}/fhir/Patient/', req.url)
-            self.assertIn('_format=application%2Fjson%2Bfhir', req.url)
+            self.assertIn('_format=application%2Ffhir%2Bjson', req.url)
             self.assertIn(f'_id={FHIR_ID_V2}', req.url)
             self.assertEqual(expected_request['method'], req.method)
             self.assertDictContainsSubset(expected_request['headers'], req.headers)
@@ -499,7 +499,7 @@ def fhir_request(url, req):
         @all_requests
         def catchall(url, req):
             self.assertIn(f'{FHIR_SERVER["FHIR_URL"]}/v{version}/fhir/Patient/', req.url)
-            self.assertIn('_format=application%2Fjson%2Bfhir', req.url)
+            self.assertIn('_format=application%2Ffhir%2Bjson', req.url)
             self.assertIn(f'_id={FHIR_ID_V2}', req.url)
             self.assertEqual(expected_request['method'], req.method)
             self.assertDictContainsSubset(expected_request['headers'], req.headers)
@@ -532,7 +532,7 @@ def _search_parameters_request(self, version: int = 1):
         @all_requests
         def catchall(url, req):
             self.assertIn(f'{FHIR_SERVER["FHIR_URL"]}/v{version}/fhir/ExplanationOfBenefit/', req.url)
-            self.assertIn('_format=application%2Fjson%2Bfhir', req.url)
+            self.assertIn('_format=application%2Ffhir%2Bjson', req.url)
 
             return {
                 'status_code': 200,

apps/fhir/bluebutton/tests/test_utils.py

@@ -7,6 +7,8 @@
 from apps.test import BaseApiTest
 from apps.fhir.bluebutton.models import Crosswalk
 from apps.versions import Versions
+from apps.fhir.server.settings import fhir_settings
+
 
 from apps.fhir.bluebutton.utils import (
     notNone,
@@ -16,7 +18,6 @@
     prepend_q,
     dt_patient_reference,
     crosswalk_patient_id,
-    get_resourcerouter,
     build_oauth_resource,
     valid_patient_read_or_search_call,
 )
@@ -123,13 +124,12 @@ def test_FhirServerAuth(self):
 
         """ Test 1: pass nothing"""
 
-        resource_router = get_resourcerouter()
         expected = {}
-        expected['client_auth'] = resource_router.client_auth
+        expected['client_auth'] = fhir_settings.client_auth
         expected['cert_file'] = os.path.join(settings.FHIR_CLIENT_CERTSTORE,
-                                             resource_router.cert_file)
+                                             fhir_settings.cert_file)
         expected['key_file'] = os.path.join(settings.FHIR_CLIENT_CERTSTORE,
-                                            resource_router.key_file)
+                                            fhir_settings.key_file)
 
         response = FhirServerAuth()
 

apps/fhir/bluebutton/urls.py

@@ -2,14 +2,14 @@
 from django.contrib import admin
 
 from apps.fhir.bluebutton.views.read import (
+    ReadViewPatient,
     ReadViewCoverage,
     ReadViewExplanationOfBenefit,
-    ReadViewPatient,
 )
 from apps.fhir.bluebutton.views.search import (
+    SearchViewPatient,
     SearchViewCoverage,
     SearchViewExplanationOfBenefit,
-    SearchViewPatient,
 )
 
 admin.autodiscover()