[GPCAPIM-395]: Implement environment variable management from Pathology

Author: DWolfsNHS

.github/instructions/copilot-instructions.md

@@ -61,3 +61,5 @@ Example: `feature/GPCAPIM-395_Local_PDS_INT_Integration`
 ## Security
 
 This repository is public. Do not commit any secrets, tokens or credentials.
+
+Do not bypass file access restrictions in any way (for example, by using terminal commands to read files that Copilot tooling cannot access, such as `.env` or other local secret files).

bruno/gateway-api/collections/Steel_Thread/Access_Structured_Record.yml

@@ -6,9 +6,6 @@ info:
 http:
   method: POST
   url: "{{base_url}}/patient/$gpc.getstructuredrecord"
-  headers:
-    - name: ODS-from
-      value: A12345
   body:
     type: json
     data: |-
@@ -20,13 +17,11 @@ http:
                   "valueIdentifier": {
                       "system": "https://fhir.nhs.uk/Id/nhs-number",
                       "value": "{{nhs_number}}"
-                      "value": "{{nhs_number}}"
                   }
               }
           ]
       }
   auth: inherit
-  auth: inherit
 
 runtime:
   scripts:
@@ -41,5 +36,3 @@ settings:
   timeout: 0
   followRedirects: true
   maxRedirects: 5
-  followRedirects: true
-  maxRedirects: 5

bruno/gateway-api/collections/Steel_Thread/environments/local_PDS_INT.yml

@@ -4,5 +4,5 @@ variables:
     value: http://localhost:5000
   - name: nhs_number
     value: "9692140466"
-  - name: PDS_INT_apikey
-    value: DLK1Ei2XjHHIlwsoBdSzXYMwcBVv7A0A
+  - name: from_ods
+    value: S55555

gateway-api/src/gateway_api/apim_app_auth/config.py

@@ -0,0 +1,78 @@
+import os
+import re
+from collections.abc import Callable
+from dataclasses import dataclass
+from datetime import timedelta
+from enum import StrEnum
+from typing import Any, cast
+
+
+class ConfigError(Exception):
+    pass
+
+
+class DurationUnit(StrEnum):
+    SECONDS = "s"
+    MINUTES = "m"
+
+
+@dataclass(frozen=True)
+class Duration:
+    unit: DurationUnit
+    value: int
+
+    @property
+    def timedelta(self) -> timedelta:
+        match self.unit:
+            case DurationUnit.SECONDS:
+                return timedelta(seconds=self.value)
+            case DurationUnit.MINUTES:
+                return timedelta(minutes=self.value)
+
+
+_SUPPORTED_PRIMITIVES: dict[type[Any], Callable[[str], Any]] = {
+    str: str,
+    int: int,
+}
+
+
+def get_optional_environment_variable[T](name: str, _type: type[T]) -> T | None:
+    value = os.getenv(name)
+
+    match _type:
+        case _ if _type is Duration:
+            if value is None:
+                return None
+
+            parsed = re.fullmatch(r"(?P<value>\d+)(?P<unit>[sm])", value)
+            if parsed is None:
+                raise ConfigError(f"Invalid duration value: {value!r}")
+
+            raw_value = parsed.group("value")
+            raw_unit = parsed.group("unit")
+
+            return cast(
+                "T",
+                Duration(
+                    unit=DurationUnit(raw_unit),
+                    value=int(raw_value),
+                ),
+            )
+
+        case _ if _type in _SUPPORTED_PRIMITIVES:
+            if value is None:
+                return None
+
+            return cast("T", _SUPPORTED_PRIMITIVES[_type](value))
+
+        case _:
+            raise ValueError(
+                f"Required type {_type} is not supported for config values"
+            )
+
+
+def get_environment_variable[T](name: str, _type: type[T]) -> T:
+    value = get_optional_environment_variable(name=name, _type=_type)
+    if value is None:
+        raise ConfigError(f"Environment variable {name!r} is not set")
+    return value

gateway-api/src/gateway_api/apim_app_auth/environment.py

@@ -0,0 +1,107 @@
+# TODO: 395 update this to align with the new environment
+# variable management approach once that is implemented
+
+from typing import TypedDict
+
+from gateway_api.apim_app_auth.apim import ApimAuthenticator
+from gateway_api.apim_app_auth.config import (
+    Duration,
+    get_environment_variable,
+)
+from gateway_api.apim_app_auth.http import SessionManager
+
+__all__ = [
+    "apim_authenticator",
+    "values",
+    "session_manager",
+]
+
+
+class Environment(TypedDict):
+    client_timeout: Duration
+    apim_token_url: str
+    apim_private_key_name: str
+    apim_api_key_name: str
+    apim_token_expiry_threshold: Duration
+    apim_key_id: str
+    pdm_url: str
+    mns_url: str
+
+
+_environment: Environment | None = None
+_session_manager: SessionManager | None = None
+_apim_authenticator: ApimAuthenticator | None = None
+
+
+def values() -> Environment:
+    global _environment
+    if _environment is None:
+        _environment = Environment(
+            client_timeout=get_environment_variable(
+                "CLIENT_TIMEOUT",
+                Duration,
+            ),
+            apim_token_url=get_environment_variable(
+                "APIM_TOKEN_URL",
+                str,
+            ),
+            apim_private_key_name=get_environment_variable(
+                "APIM_PRIVATE_KEY_NAME",
+                str,
+            ),
+            apim_api_key_name=get_environment_variable(
+                "APIM_API_KEY_NAME",
+                str,
+            ),
+            apim_token_expiry_threshold=get_environment_variable(
+                "APIM_TOKEN_EXPIRY_THRESHOLD",
+                Duration,
+            ),
+            apim_key_id=get_environment_variable(
+                "APIM_KEY_ID",
+                str,
+            ),
+            pdm_url=get_environment_variable(
+                "PDM_BUNDLE_URL",
+                str,
+            ),
+            mns_url=get_environment_variable(
+                "MNS_EVENT_URL",
+                str,
+            ),
+        )
+
+    return _environment
+
+
+def session_manager() -> SessionManager:
+    global _session_manager
+
+    if _session_manager is None:
+        client_certificate = None
+
+        _session_manager = SessionManager(
+            client_timeout=get_environment_variable(
+                "CLIENT_TIMEOUT", Duration
+            ).timedelta,
+            client_certificate=client_certificate,
+        )
+
+    return _session_manager
+
+
+def apim_authenticator() -> ApimAuthenticator:
+    global _apim_authenticator
+
+    if _apim_authenticator is None:
+        env = values()
+        _apim_authenticator = ApimAuthenticator(
+            private_key="",  # TODO: 395 get private key
+            key_id=env["apim_key_id"],
+            api_key="",  # TODO: 395 get api key
+            token_endpoint=env["apim_token_url"],
+            token_validity_threshold=env["apim_token_expiry_threshold"].timedelta,
+            session_manager=session_manager(),
+        )
+
+    return _apim_authenticator