Merge pull request #32 from NHSDigital/AMB-804-test-apikey-protected-endpoints-logs

AMB-804 add apikey protected endpoint

Author: artronics

azure/templates/run-integration-tests.yml

@@ -20,6 +20,7 @@ steps:
     export STATUS_ENDPOINT_API_KEY="$(status-endpoint-api-key)"
     export APIGEE_API_TOKEN="$(secret.AccessToken)"
     export OAUTH_PROXY="oauth2"
+    export APP_CLIENT_ID="Too5BdPayTQACdw1AJK1rD4nKUD0Ag7J"
     export OAUTH_BASE_URI="https://$(APIGEE_ENVIRONMENT).api.service.nhs.uk"
     export ACCESS_TOKEN_HASH_SECRET="$(ACCESS_TOKEN_SECRET)"
     ./test_env/bin/pytest --reruns 2 --reruns-delay 1 -v --junitxml=test-report.xml

proxies/live/apiproxy/policies/VerifyApiKey.Apikey.xml

@@ -0,0 +1,6 @@
+<?xml version="1.0" encoding="UTF-8" standalone="yes"?>
+<VerifyAPIKey async="false" continueOnError="false" enabled="true" name="VerifyApiKey.Apikey">
+    <DisplayName>VerifyApiKey.Apikey</DisplayName>
+    <Properties/>
+    <APIKey ref="request.header.apikey"/>
+</VerifyAPIKey>

proxies/live/apiproxy/proxies/default.xml

@@ -63,6 +63,14 @@
       </Response>
       <Condition>(proxy.pathsuffix MatchesPath "/splunk-test") and ((request.verb = "GET") or (request.verb = "HEAD"))</Condition>
     </Flow>
+    <Flow name="ApiKeyProtectedEndpoint">
+      <Request>
+        <Step>
+          <Name>VerifyApiKey.Apikey</Name>
+        </Step>
+      </Request>
+      <Condition>(proxy.pathsuffix MatchesPath "/apikey-protected") and ((request.verb = "GET") or (request.verb = "HEAD"))</Condition>
+    </Flow>
     <Flow name="OpenAccessEndpoint">
       <Condition>(proxy.pathsuffix MatchesPath "/open-access") and ((request.verb = "GET") or (request.verb = "HEAD"))</Condition>
     </Flow>
@@ -97,6 +105,9 @@
   <RouteRule name="NoRouteOpenAccess">
     <Condition>(proxy.pathsuffix MatchesPath "/open-access") and ((request.verb = "GET") or (request.verb = "HEAD"))</Condition>
   </RouteRule>
+  <RouteRule name="NoRouteApiKeyProtected">
+    <Condition>(proxy.pathsuffix MatchesPath "/apikey-protected") and ((request.verb = "GET") or (request.verb = "HEAD"))</Condition>
+  </RouteRule>
   <RouteRule name="shared-flow-testing-target">
     <TargetEndpoint>shared-flow-testing-target</TargetEndpoint>
   </RouteRule>

tests/configuration/config.py

@@ -13,3 +13,4 @@
 SERVICE_BASE_PATH = ENV['service_base_path']
 SERVICE_NAME = ENV['service_name']
 ACCESS_TOKEN_HASH_SECRET = environ.get("ACCESS_TOKEN_HASH_SECRET")
+APP_CLIENT_ID = environ.get("APP_CLIENT_ID")

tests/splunk_logging_schema.json

@@ -120,7 +120,7 @@
         },
         "client_id": {
           "type": "string",
-          "pattern": "(^[0-9A-Za-z]{32}$)|^$"
+          "pattern": "(^[0-9A-Za-z]{32}$)|Not provided"
         },
         "application_id": {
           "type": "string",

tests/test_splunk_logging.py

@@ -7,11 +7,12 @@
 import requests
 from jsonschema import validate
 
-from .configuration.config import SERVICE_BASE_PATH, ENVIRONMENT, ACCESS_TOKEN_HASH_SECRET
+from .configuration.config import SERVICE_BASE_PATH, ENVIRONMENT, ACCESS_TOKEN_HASH_SECRET, APP_CLIENT_ID
 
 
 class TestSplunkLogging:
-    url = f"https://{ENVIRONMENT}.api.service.nhs.uk/{SERVICE_BASE_PATH}/splunk-test"
+    oauth_protected_url = f"https://{ENVIRONMENT}.api.service.nhs.uk/{SERVICE_BASE_PATH}/splunk-test"
+    apikey_protected_url = f"https://{ENVIRONMENT}.api.service.nhs.uk/{SERVICE_BASE_PATH}/apikey-protected"
     open_access_url = f"https://{ENVIRONMENT}.api.service.nhs.uk/{SERVICE_BASE_PATH}/open-access"
 
     @staticmethod
@@ -37,7 +38,7 @@ async def test_splunk_auth_with_client_credentials(self, get_token_client_creden
         # When
         await debug.start_trace()
         requests.get(
-            url=self.url,
+            url=self.oauth_protected_url,
             headers={"Authorization": f"Bearer {token}"},
         )
         payload = await self._get_payload_from_splunk(debug)
@@ -65,7 +66,7 @@ async def test_splunk_auth_with_authorization_code(self, get_token, debug):
         # When
         await debug.start_trace()
         requests.get(
-            url=self.url,
+            url=self.oauth_protected_url,
             headers={"Authorization": f"Bearer {token}"},
         )
         payload = await self._get_payload_from_splunk(debug)
@@ -93,7 +94,7 @@ async def test_splunk_auth_with_cis2_token_exchange(self, get_token_cis2_token_e
         # When
         await debug.start_trace()
         requests.get(
-            url=self.url,
+            url=self.oauth_protected_url,
             headers={"Authorization": f"Bearer {token}"},
         )
         payload = await self._get_payload_from_splunk(debug)
@@ -121,7 +122,7 @@ async def test_splunk_auth_with_nhs_login_token_exchange(self, get_token_nhs_log
         # When
         await debug.start_trace()
         requests.get(
-            url=self.url,
+            url=self.oauth_protected_url,
             headers={"Authorization": f"Bearer {token}"},
         )
         payload = await self._get_payload_from_splunk(debug)
@@ -139,6 +140,162 @@ async def test_splunk_auth_with_nhs_login_token_exchange(self, get_token_nhs_log
         auth_user = auth["user"]
         assert auth_user["user_id"] == "900000000001"
 
+    @pytest.mark.splunk
+    @pytest.mark.asyncio
+    async def test_splunk_auth_with_invalid_token(self, debug):
+        # Given
+        token = "invalid token"
+        expected_hashed_token = ""
+
+        # When
+        await debug.start_trace()
+        requests.get(
+            url=self.oauth_protected_url,
+            headers={"Authorization": f"Bearer {token}"},
+        )
+        payload = await self._get_payload_from_splunk(debug)
+
+        # Then
+        auth = payload["auth"]
+        assert auth["access_token_hash"] == expected_hashed_token
+
+        auth_meta = auth["meta"]
+        assert auth_meta["auth_type"] == "app"
+        assert auth_meta["grant_type"] == ""
+        assert auth_meta["level"] == "Level0"
+        assert auth_meta["provider"] == "apim"
+
+        auth_user = auth["user"]
+        assert auth_user["user_id"] == ""
+
+        meta = payload["meta"]
+        assert meta["client_id"] == "Not provided"
+        assert meta["application"] == "unknown"
+        assert meta["product"] == ""
+
+    @pytest.mark.splunk
+    @pytest.mark.asyncio
+    async def test_splunk_auth_with_expired_token(self, debug):
+        # Given
+        token = "zRygtc34R2pwxbiUktLsMJWX0iJW"
+        expected_hashed_token = ""
+
+        # When
+        await debug.start_trace()
+        requests.get(
+            url=self.oauth_protected_url,
+            headers={"Authorization": f"Bearer {token}"},
+        )
+        payload = await self._get_payload_from_splunk(debug)
+
+        # Then
+        auth = payload["auth"]
+        assert auth["access_token_hash"] == expected_hashed_token
+
+        auth_meta = auth["meta"]
+        assert auth_meta["auth_type"] == "app"
+        assert auth_meta["grant_type"] == ""
+        assert auth_meta["level"] == "Level0"
+        assert auth_meta["provider"] == "apim"
+
+        auth_user = auth["user"]
+        assert auth_user["user_id"] == ""
+
+        meta = payload["meta"]
+        assert meta["client_id"] == "Not provided"
+        assert meta["application"] == "unknown"
+        assert meta["product"] == ""
+
+    @pytest.mark.splunk
+    @pytest.mark.asyncio
+    @pytest.mark.debug
+    async def test_splunk_auth_with_apikey(self, debug):
+        # Given
+        apikey = APP_CLIENT_ID
+
+        # When
+        await debug.start_trace()
+        requests.get(
+            url=self.apikey_protected_url,
+            headers={"apikey": apikey},
+        )
+        payload = await self._get_payload_from_splunk(debug)
+
+        # Then
+        auth = payload["auth"]
+        assert auth["access_token_hash"] == ""
+
+        auth_meta = auth["meta"]
+        assert auth_meta["auth_type"] == "app"
+        assert auth_meta["grant_type"] == ""
+        assert auth_meta["level"] == "Level0"
+        assert auth_meta["provider"] == "apim"
+
+        auth_user = auth["user"]
+        assert auth_user["user_id"] == ""
+
+        meta = payload["meta"]
+        assert meta["client_id"] == apikey
+
+    @pytest.mark.splunk
+    @pytest.mark.asyncio
+    async def test_splunk_auth_with_invalid_apikey(self, debug):
+        # Given
+        apikey = "invalid api key"
+
+        # When
+        await debug.start_trace()
+        requests.get(
+            url=self.apikey_protected_url,
+            headers={"apikey": apikey},
+        )
+        payload = await self._get_payload_from_splunk(debug)
+
+        # Then
+        auth = payload["auth"]
+        assert auth["access_token_hash"] == ""
+
+        auth_meta = auth["meta"]
+        assert auth_meta["auth_type"] == "app"
+        assert auth_meta["grant_type"] == ""
+        assert auth_meta["level"] == "Level0"
+        assert auth_meta["provider"] == "apim"
+
+        auth_user = auth["user"]
+        assert auth_user["user_id"] == ""
+
+        meta = payload["meta"]
+        assert meta["client_id"] == "Not provided"
+        assert meta["application"] == "unknown"
+        assert meta["product"] == ""
+
+    @pytest.mark.splunk
+    @pytest.mark.asyncio
+    async def test_splunk_auth_open_access(self, debug):
+        # When
+        await debug.start_trace()
+        requests.get(
+            url=self.open_access_url,
+        )
+        payload = await self._get_payload_from_splunk(debug)
+
+        # Then
+        auth = payload["auth"]
+        assert auth["access_token_hash"] == ""
+
+        auth_meta = auth["meta"]
+        assert auth_meta["auth_type"] == "app"
+        assert auth_meta["grant_type"] == ""
+        assert auth_meta["level"] == "Level0"
+        assert auth_meta["provider"] == "apim"
+
+        auth_user = auth["user"]
+        assert auth_user["user_id"] == ""
+
+        meta = payload["meta"]
+        assert meta["client_id"] == "Not provided"
+        assert meta["application"] == "unknown"
+
     @pytest.mark.splunk
     @pytest.mark.asyncio
     async def test_splunk_payload_schema(self, get_token, debug):
@@ -148,7 +305,7 @@ async def test_splunk_payload_schema(self, get_token, debug):
         # When
         await debug.start_trace()
         requests.get(
-            url=self.url,
+            url=self.oauth_protected_url,
             headers={"Authorization": f"Bearer {token}"},
         )
         payload = await self._get_payload_from_splunk(debug)