[GPCAPIM-287] Orange box

.github/workflows/preview-env.yml

@@ -507,20 +507,20 @@ jobs:
       # ---------- Security scanning ----------
       - name: Trivy IaC scan
         if: github.event.action != 'closed'
-        uses: nhs-england-tools/trivy-action/iac-scan@3456c1657a37d500027fd782e6b08911725392da
+        uses: nhs-england-tools/trivy-action/iac-scan@289984b2f03034233a347d6dbadecd5ca9ea9634
         with:
           scan-ref: infrastructure/environments/preview
           artifact-name: trivy-iac-scan-${{ steps.meta.outputs.branch_name }}
 
       - name: Trivy filesystem scan
         if: github.event.action != 'closed'
-        uses: nhs-england-tools/trivy-action/image-scan@3456c1657a37d500027fd782e6b08911725392da
+        uses: nhs-england-tools/trivy-action/image-scan@289984b2f03034233a347d6dbadecd5ca9ea9634
         with:
           image-ref: ${{steps.meta.outputs.ecr_url}}:${{steps.meta.outputs.branch_name}}
           artifact-name: trivy-scan-${{ steps.meta.outputs.branch_name }}
 
       - name: Generate SBOM
-        uses: nhs-england-tools/trivy-action/sbom-scan@3456c1657a37d500027fd782e6b08911725392da
+        uses: nhs-england-tools/trivy-action/sbom-scan@289984b2f03034233a347d6dbadecd5ca9ea9634
         if: github.event.action != 'closed'
         with:
           image-ref: ${{steps.meta.outputs.ecr_url}}:${{steps.meta.outputs.branch_name}}

Makefile

@@ -54,11 +54,25 @@ publish: # Publish the project artefact @Pipeline
 	# TODO: Implement the artefact publishing step
 
 deploy: clean build # Deploy the project artefact to the target environment @Pipeline
-	@if [[ -n "$${IN_BUILD_CONTAINER}" ]]; then \
+	# Build up list of environment variables to pass to the container
+	@ENVIRONMENT_STRING="" ; \
+	if [[ -n "$${STUB_PROVIDER}" ]]; then \
+		ENVIRONMENT_STRING="$${ENVIRONMENT_STRING} -e STUB_PROVIDER=$${STUB_PROVIDER}" ; \
+	fi ; \
+	if [[ -n "$${STUB_PDS}" ]]; then \
+		ENVIRONMENT_STRING="$${ENVIRONMENT_STRING} -e STUB_PDS=$${STUB_PDS}" ; \
+	fi ; \
+	if [[ -n "$${STUB_SDS}" ]]; then \
+		ENVIRONMENT_STRING="$${ENVIRONMENT_STRING} -e STUB_SDS=$${STUB_SDS}" ; \
+	fi ; \
+	if [[ -n "$${CDG_DEBUG}" ]]; then \
+		ENVIRONMENT_STRING="$${ENVIRONMENT_STRING} -e CDG_DEBUG=$${CDG_DEBUG}" ; \
+	fi ; \
+	if [[ -n "$${IN_BUILD_CONTAINER}" ]]; then \
 		echo "Starting using local docker network ..." ; \
-		$(docker) run --platform linux/amd64 --name gateway-api -p 5000:8080 --network gateway-local -d ${IMAGE_NAME} ; \
+		$(docker) run --platform linux/amd64 --name gateway-api -p 5000:8080 --network gateway-local $${ENVIRONMENT_STRING} -d ${IMAGE_NAME} ; \
 	else \
-		$(docker) run --platform linux/amd64 --name gateway-api -p 5000:8080 -d ${IMAGE_NAME} ; \
+		$(docker) run --platform linux/amd64 --name gateway-api -p 5000:8080 $${ENVIRONMENT_STRING} -d ${IMAGE_NAME} ; \
 	fi
 
 clean:: stop # Clean-up project resources (main) @Operations

gateway-api/src/fhir/constants.py

@@ -0,0 +1,15 @@
+from enum import StrEnum
+
+
+class FHIRSystem(StrEnum):
+    """
+    Enum for FHIR identifier systems used in the clinical data gateway.
+    """
+
+    NHS_NUMBER = "https://fhir.nhs.uk/Id/nhs-number"
+    ODS_CODE = "https://fhir.nhs.uk/Id/ods-organization-code"
+    SDS_USER_ID = "https://fhir.nhs.uk/Id/sds-user-id"
+    SDS_ROLE_PROFILE_ID = "https://fhir.nhs.uk/Id/sds-role-profile-id"
+    NHS_SERVICE_INTERACTION_ID = "https://fhir.nhs.uk/Id/nhsServiceInteractionId"
+    NHS_MHS_PARTY_KEY = "https://fhir.nhs.uk/Id/nhsMhsPartyKey"
+    NHS_SPINE_ASID = "https://fhir.nhs.uk/Id/nhsSpineASID"

gateway-api/src/gateway_api/clinical_jwt/__init__.py

@@ -1,5 +1,7 @@
 from .device import Device
 from .jwt import JWT
+from .organization import Organization
 from .practitioner import Practitioner
+from .validator import JWTValidator
 
-__all__ = ["JWT", "Device", "Practitioner"]
+__all__ = ["JWT", "Device", "Organization", "Practitioner", "JWTValidator"]

gateway-api/src/gateway_api/clinical_jwt/device.py

@@ -1,4 +1,5 @@
 from dataclasses import dataclass
+from typing import Any
 
 
 @dataclass(frozen=True, kw_only=True)
@@ -8,22 +9,13 @@ class Device:
     model: str
     version: str
 
-    @property
-    def json(self) -> str:
-        outstr = f"""
-        {{
-        "resourceType": "Device",
-        "identifier": [
-            {{
-                "system": "{self.system}",
-                "value": "{self.value}"
-            }}
-        ],
-        "model": "{self.model}",
-        "version": "{self.version}"
-        }}
+    def to_dict(self) -> dict[str, Any]:
         """
-        return outstr.strip()
-
-    def __str__(self) -> str:
-        return self.json
+        Return the Device as a dictionary suitable for JWT payload.
+        """
+        return {
+            "resourceType": "Device",
+            "identifier": [{"system": self.system, "value": self.value}],
+            "model": self.model,
+            "version": self.version,
+        }

gateway-api/src/gateway_api/clinical_jwt/jwt.py

@@ -11,20 +11,24 @@ class JWT:
     issuer: str
     subject: str
     audience: str
-    requesting_device: str
-    requesting_organization: str
-    requesting_practitioner: str
+    requesting_device: dict[str, Any]
+    requesting_organization: dict[str, Any]
+    requesting_practitioner: dict[str, Any]
 
     # Time fields
     issued_at: int = field(default_factory=lambda: int(time()))
-    expiration: int = field(default_factory=lambda: int(time()) + 300)
+    expiration: int = 0
 
     # These are here for future proofing but are not expected ever to be changed
-    algorithm: str | None = None
+    algorithm: str = "none"
     type: str = "JWT"
     reason_for_request: str = "directcare"
     requested_scope: str = "patient/*.read"
 
+    def __post_init__(self) -> None:
+        if self.expiration == 0:
+            object.__setattr__(self, "expiration", self.issued_at + 300)
+
     @property
     def issue_time(self) -> str:
         return datetime.fromtimestamp(self.issued_at, tz=UTC).isoformat()

gateway-api/src/gateway_api/clinical_jwt/organization.py

@@ -0,0 +1,25 @@
+from dataclasses import dataclass
+from typing import Any
+
+from fhir.constants import FHIRSystem
+
+
+@dataclass(frozen=True, kw_only=True)
+class Organization:
+    ods_code: str
+    name: str
+
+    def to_dict(self) -> dict[str, Any]:
+        """
+        Return the Organization as a dictionary suitable for JWT payload.
+        """
+        return {
+            "resourceType": "Organization",
+            "identifier": [
+                {
+                    "system": FHIRSystem.ODS_CODE,
+                    "value": self.ods_code,
+                }
+            ],
+            "name": self.name,
+        }

gateway-api/src/gateway_api/clinical_jwt/practitioner.py

@@ -1,7 +1,10 @@
 from dataclasses import dataclass
+from typing import Any
 
+from fhir.constants import FHIRSystem
 
-@dataclass(kw_only=True)
+
+@dataclass(frozen=True, kw_only=True)
 class Practitioner:
     id: str
     sds_userid: str
@@ -12,38 +15,29 @@ class Practitioner:
     given_name: str | None = None
     prefix: str | None = None
 
-    def __post_init__(self) -> None:
-        given = "" if self.given_name is None else f',"given":["{self.given_name}"]'
-        prefix = "" if self.prefix is None else f',"prefix":["{self.prefix}"]'
-        self._name_str = f'[{{"family": "{self.family_name}"{given}{prefix}}}]'
-
-    @property
-    def json(self) -> str:
-        user_id_system = "https://fhir.nhs.uk/Id/sds-user-id"
-        role_id_system = "https://fhir.nhs.uk/Id/sds-role-profile-id"
+    def _build_name(self) -> list[dict[str, Any]]:
+        """Build the name array with proper structure for JWT."""
+        name_dict: dict[str, Any] = {"family": self.family_name}
+        if self.given_name is not None:
+            name_dict["given"] = [self.given_name]
+        if self.prefix is not None:
+            name_dict["prefix"] = [self.prefix]
+        return [name_dict]
 
-        outstr = f"""
-        {{
-        "resourceType": "Practitioner",
-        "id": "{self.id}",
-        "identifier": [
-        {{
-            "system": "{user_id_system}",
-            "value": "{self.sds_userid}"
-        }},
-        {{
-            "system": "{role_id_system}",
-            "value": "{self.role_profile_id}"
-        }},
-        {{
-            "system": "{self.userid_url}",
-            "value": "{self.userid_value}"
-        }}
-        ],
-        "name": {self._name_str}
-        }}
+    def to_dict(self) -> dict[str, Any]:
+        """
+        Return the Practitioner as a dictionary suitable for JWT payload.
         """
-        return outstr.strip()
+        user_id_system = FHIRSystem.SDS_USER_ID
+        role_id_system = FHIRSystem.SDS_ROLE_PROFILE_ID
 
-    def __str__(self) -> str:
-        return self.json
+        return {
+            "resourceType": "Practitioner",
+            "id": self.id,
+            "identifier": [
+                {"system": user_id_system, "value": self.sds_userid},
+                {"system": role_id_system, "value": self.role_profile_id},
+                {"system": self.userid_url, "value": self.userid_value},
+            ],
+            "name": self._build_name(),
+        }

gateway-api/src/gateway_api/clinical_jwt/test_device.py

@@ -2,8 +2,6 @@
 Unit tests for :mod:`gateway_api.clinical_jwt.device`.
 """
 
-from json import loads
-
 from gateway_api.clinical_jwt import Device
 
 
@@ -35,8 +33,7 @@ def test_device_json_property_returns_valid_json_structure() -> None:
         version="5.3.0",
     )
 
-    json_output = input_device.json
-    jdict = loads(json_output)
+    jdict = input_device.to_dict()
 
     output_device = Device(
         system=jdict["identifier"][0]["system"],
@@ -46,17 +43,3 @@ def test_device_json_property_returns_valid_json_structure() -> None:
     )
 
     assert input_device == output_device
-
-
-def test_device_str_returns_json() -> None:
-    """
-    Test that __str__ returns the same value as the json property.
-    """
-    device = Device(
-        system="https://test.com/device",
-        value="TEST-001",
-        model="Test Model",
-        version="1.0.0",
-    )
-
-    assert str(device) == device.json