[GPCAPIM-285] JWT creation

gateway-api/pyproject.toml

@@ -6,13 +6,14 @@ authors = [
     {name = "Your Name", email = "you@example.com"}
 ]
 readme = "README.md"
-requires-python = ">3.13,<4.0.0"
+requires-python = ">=3.14,<4.0.0"
 
 [tool.poetry.dependencies]
 clinical-data-common = { git = "https://github.com/NHSDigital/clinical-data-common.git", tag = "v0.1.0" }
 flask = "^3.1.2"
 types-flask = "^1.1.6"
 requests = "^2.32.5"
+pyjwt = "^2.11.0"
 
 [tool.poetry]
 packages = [{include = "gateway_api", from = "src"},

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

@@ -0,0 +1,5 @@
+from .device import Device
+from .jwt import JWT
+from .practitioner import Practitioner
+
+__all__ = ["JWT", "Device", "Practitioner"]

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

@@ -0,0 +1,29 @@
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True, kw_only=True)
+class Device:
+    system: str
+    value: str
+    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}"
+        }}
+        """
+        return outstr.strip()
+
+    def __str__(self) -> str:
+        return self.json

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

@@ -0,0 +1,77 @@
+from dataclasses import dataclass, field
+from datetime import UTC, datetime
+from time import time
+from typing import Any
+
+import jwt as pyjwt
+
+
+@dataclass(frozen=True, kw_only=True)
+class JWT:
+    issuer: str
+    subject: str
+    audience: str
+    requesting_device: str
+    requesting_organization: str
+    requesting_practitioner: str
+
+    # Time fields
+    issued_at: int = field(default_factory=lambda: int(time()))
+    expiration: int = field(default_factory=lambda: int(time()) + 300)
+
+    # These are here for future proofing but are not expected ever to be changed
+    algorithm: str | None = None
+    type: str = "JWT"
+    reason_for_request: str = "directcare"
+    requested_scope: str = "patient/*.read"
+
+    @property
+    def issue_time(self) -> str:
+        return datetime.fromtimestamp(self.issued_at, tz=UTC).isoformat()
+
+    @property
+    def exp_time(self) -> str:
+        return datetime.fromtimestamp(self.expiration, tz=UTC).isoformat()
+
+    def encode(self) -> str:
+        return pyjwt.encode(
+            self.payload(),
+            key=None,
+            algorithm=self.algorithm,
+            headers={"typ": self.type},
+        )
+
+    @staticmethod
+    def decode(token: str) -> "JWT":
+        token_dict = pyjwt.decode(
+            token,
+            options={"verify_signature": False},  # NOSONAR S5659 (not signed)
+        )
+
+        return JWT(
+            issuer=token_dict["iss"],
+            subject=token_dict["sub"],
+            audience=token_dict["aud"],
+            expiration=token_dict["exp"],
+            issued_at=token_dict["iat"],
+            requesting_device=token_dict["requesting_device"],
+            requesting_organization=token_dict["requesting_organization"],
+            requesting_practitioner=token_dict["requesting_practitioner"],
+        )
+
+    def payload(self) -> dict[str, Any]:
+        return {
+            "iss": self.issuer,
+            "sub": self.subject,
+            "aud": self.audience,
+            "exp": self.expiration,
+            "iat": self.issued_at,
+            "requesting_device": self.requesting_device,
+            "requesting_organization": self.requesting_organization,
+            "requesting_practitioner": self.requesting_practitioner,
+            "reason_for_request": self.reason_for_request,
+            "requested_scope": self.requested_scope,
+        }
+
+    def __str__(self) -> str:
+        return self.encode()

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

@@ -0,0 +1,49 @@
+from dataclasses import dataclass
+
+
+@dataclass(kw_only=True)
+class Practitioner:
+    id: str
+    sds_userid: str
+    role_profile_id: str
+    userid_url: str
+    userid_value: str
+    family_name: str
+    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"
+
+        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}
+        }}
+        """
+        return outstr.strip()
+
+    def __str__(self) -> str:
+        return self.json

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

@@ -0,0 +1,62 @@
+"""
+Unit tests for :mod:`gateway_api.clinical_jwt.device`.
+"""
+
+from json import loads
+
+from gateway_api.clinical_jwt import Device
+
+
+def test_device_creation_with_all_required_fields() -> None:
+    """
+    Test that a Device instance can be created with all required fields.
+    """
+    device = Device(
+        system="https://consumersupplier.com/Id/device-identifier",
+        value="CONS-APP-4",
+        model="Consumer product name",
+        version="5.3.0",
+    )
+
+    assert device.system == "https://consumersupplier.com/Id/device-identifier"
+    assert device.value == "CONS-APP-4"
+    assert device.model == "Consumer product name"
+    assert device.version == "5.3.0"
+
+
+def test_device_json_property_returns_valid_json_structure() -> None:
+    """
+    Test that the json property returns a valid JSON structure for requesting_device.
+    """
+    input_device = Device(
+        system="https://consumersupplier.com/Id/device-identifier",
+        value="CONS-APP-4",
+        model="Consumer product name",
+        version="5.3.0",
+    )
+
+    json_output = input_device.json
+    jdict = loads(json_output)
+
+    output_device = Device(
+        system=jdict["identifier"][0]["system"],
+        value=jdict["identifier"][0]["value"],
+        model=jdict["model"],
+        version=jdict["version"],
+    )
+
+    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