[CDAPI-85]: Initial introduction of APIM Authenticator class

Initial Introduction of the ApimAuthenticator class, handling authentication with the API Management platform utilising Signed JWT application restricted access.
This commit also includes the creation of a `SessionManager` class handling the creation of a `request.Session` object with appropriate default configuration.

Author: nhsd-jack-wainwright

pathology-api/poetry.lock

@@ -9,7 +9,8 @@ readme = "README.md"
 requires-python = ">3.13,<4.0.0"
 dependencies = [
     "aws-lambda-powertools (>=3.24.0,<4.0.0)",
-    "pydantic (>=2.12.5,<3.0.0)"
+    "pydantic (>=2.12.5,<3.0.0)",
+    "pyjwt[crypto] (>=2.11.0,<3.0.0)"
 ]
 
 [tool.poetry]

pathology-api/pyproject.toml

@@ -0,0 +1,115 @@
+import uuid
+from collections.abc import Callable
+from datetime import datetime, timedelta, timezone
+from typing import Any, Concatenate, TypedDict
+
+import jwt
+import requests
+
+from pathology_api.http import SessionManager
+
+
+class ApimAuthenticationException(Exception):
+    pass
+
+
+# Type alias describing the expected signature for use with the `Authenticator.auth`
+# decorator.
+# Any function that takes a `requests.Session` as its first argument, followed by any
+# number of additional arguments, and returns any type of value.
+type AuthenticatedMethod[**P] = Callable[Concatenate[requests.Session, P], Any]
+
+
+class ApimAuthenticator:
+    class __AccessToken(TypedDict):
+        value: str
+        expiry: datetime
+
+    def __init__(
+        self,
+        private_key: str,
+        key_id: str,
+        api_key: str,
+        token_validity_threshold: timedelta,
+        token_endpoint: str,
+        session_manager: SessionManager,
+    ):
+        self._private_key = private_key
+        self._key_id = key_id
+        self._api_key = api_key
+        self._token_validity_threshold = token_validity_threshold
+        self._token_endpoint = token_endpoint
+        self._session_manager = session_manager
+
+        self.__access_token: ApimAuthenticator.__AccessToken | None = None
+
+    def auth[**P](
+        self, func: AuthenticatedMethod[P]
+    ) -> Callable[[AuthenticatedMethod[P]], AuthenticatedMethod[P]]:
+        """
+        Decorate a given function with APIM authentication. This authentication will be
+        provided via a `requests.Session` object.
+        """
+
+        def wrapper(*args: Any, **kwargs: Any) -> Any:
+            # If there isn't an access token yet, or the token will expire within the
+            # token validity threshold, reauthenticate.
+            if (
+                self.__access_token is None
+                or self.__access_token["expiry"] - datetime.now(tz=timezone.utc)
+                < self._token_validity_threshold
+            ):
+                self.__access_token = self._authenticate()
+
+            with self._session_manager.open_session() as session:
+                session.headers.update(
+                    {"Authorization": f"Bearer {self.__access_token['value']}"}
+                )
+                return func(session, *args, **kwargs)
+
+        return wrapper
+
+    def _create_client_assertion(self) -> str:
+        claims = {
+            "sub": self._api_key,
+            "iss": self._api_key,
+            "jti": str(uuid.uuid4()),
+            "aud": self._token_endpoint,
+            "exp": int(
+                (datetime.now(tz=timezone.utc) + timedelta(seconds=30)).timestamp()
+            ),
+        }
+
+        return jwt.encode(
+            claims,
+            self._private_key,
+            algorithm="RS512",
+            headers={"kid": self._key_id},
+        )
+
+    def _authenticate(self) -> __AccessToken:
+        with self._session_manager.open_session() as session:
+            response = session.post(
+                self._token_endpoint,
+                data={
+                    "grant_type": "client_credentials",
+                    "client_assertion_type": "urn:ietf:params:oauth"
+                    ":client-assertion-type:jwt-bearer",
+                    "client_assertion": self._create_client_assertion(),
+                },
+            )
+
+            if response.status_code != 200:
+                raise ApimAuthenticationException(
+                    f"Failed to authenticate with APIM. "
+                    f"Status code: {response.status_code}"
+                    f", Response: {response.text}"
+                )
+
+            response_data = response.json()
+
+            return {
+                "value": response_data["access_token"],
+                "expiry": datetime.now(tz=timezone.utc)
+                + timedelta(seconds=response_data["expires_in"]),
+            }

pathology-api/src/pathology_api/apim.py

@@ -0,0 +1,53 @@
+from datetime import timedelta
+from typing import Any, TypedDict
+
+import requests
+from requests.adapters import HTTPAdapter
+
+
+class ClientCertificate(TypedDict):
+    certificate_path: str
+    key_path: str
+
+
+class SessionManager:
+    class _Adapter(HTTPAdapter):
+        """
+        HTTPAdapter to apply default configuration to apply to all created
+        `request.Session` objects.
+        """
+
+        def __init__(self, timeout: float):
+            self._timeout = timeout
+            super().__init__()
+
+        def send(
+            self,
+            request: requests.PreparedRequest,
+            *args: Any,
+            **kwargs: Any,
+        ) -> requests.Response:
+            kwargs["timeout"] = self._timeout
+            return super().send(request, *args, **kwargs)
+
+    def __init__(
+        self,
+        client_timeout: timedelta,
+        client_certificate: ClientCertificate | None = None,
+    ):
+        self._client_adapter = self._Adapter(timeout=client_timeout.total_seconds())
+        self._client_certificate = client_certificate
+
+    def open_session(self) -> requests.Session:
+        session = requests.Session()
+
+        if self._client_certificate is not None:
+            session.cert = (
+                self._client_certificate["certificate_path"],
+                self._client_certificate["key_path"],
+            )
+
+        session.mount("https://", self._client_adapter)
+        session.mount("http://", self._client_adapter)
+
+        return session