Merge pull request #355 from NHSDigital/10291-logout-cis2

Add CIS2 back channel logout endpoint

Author: malcolmbaig

.gitleaksignore

@@ -17,3 +17,5 @@ infrastructure/terraform/resource_group_init/main.bicep:generic-api-key:32
 infrastructure/terraform/resource_group_init/main.bicep:generic-api-key:33
 infrastructure/terraform/resource_group_init/storage.bicep:generic-api-key:59
 infrastructure/terraform/resource_group_init/keyVault.bicep:generic-api-key:10
+manage_breast_screening/config/settings_test.py:private-key:34
+manage_breast_screening/config/settings_test.py:public-key:61

cis2_auth.md docs/cis2_auth.mdcis2_auth.md renamed to docs/cis2_auth.md

@@ -0,0 +1,71 @@
+import logging
+from typing import Any, Dict
+
+from authlib.jose import JoseError, jwt
+from authlib.jose.errors import (
+    ExpiredTokenError,
+    InvalidClaimError,
+    InvalidTokenError,
+    MissingClaimError,
+)
+
+logger = logging.getLogger(__name__)
+
+
+class InvalidLogoutToken(Exception):
+    """Raised when a CIS2 back-channel logout token is invalid."""
+
+    def __init__(self, cause=None):
+        """
+        Initialize with optional cause.
+
+        Args:
+            cause: The original exception that caused this error
+        """
+        self.cause = cause
+        super().__init__(str(cause) if cause else "Invalid logout token")
+
+
+class DecodeLogoutToken:
+    def call(
+        self,
+        metadata: Dict[str, Any],
+        logout_token: str,
+        client_id: str,
+        key_loader,
+    ):
+        """
+        Decode and validate a CIS2 back-channel logout token.
+
+        Returns the decoded claims on success and raises InvalidLogoutToken on
+        any underlying decoding/validation error.
+        """
+        try:
+            verification_rules = {
+                "iss": {"values": [metadata["issuer"]], "essential": True},
+                "aud": {"values": [client_id], "essential": True},
+                "exp": {"essential": True},
+                "iat": {"essential": True},
+                "events": {
+                    "essential": True,
+                    "validate": lambda claim, value: isinstance(value, dict)
+                    and "http://schemas.openid.net/event/backchannel-logout" in value,
+                },
+                "nonce": {"validate": lambda claim, value: value is None},
+            }
+            claims = jwt.decode(
+                logout_token,
+                key=key_loader,
+                claims_options=verification_rules,
+            )
+            claims.validate(leeway=60)
+            return claims
+        except (
+            ExpiredTokenError,
+            InvalidClaimError,
+            InvalidTokenError,
+            JoseError,
+            MissingClaimError,
+        ) as e:
+            logger.exception("Invalid logout token")
+            raise InvalidLogoutToken(cause=e) from e

manage_breast_screening/auth/services.py

@@ -0,0 +1,202 @@
+import time
+
+import pytest
+from authlib.jose import JsonWebKey, jwt
+from authlib.jose.errors import (
+    ExpiredTokenError,
+    InvalidClaimError,
+    JoseError,
+    MissingClaimError,
+)
+
+from manage_breast_screening.auth.services import (
+    DecodeLogoutToken,
+    InvalidLogoutToken,
+)
+
+
+class TestDecodeLogoutToken:
+    @staticmethod
+    def _make_keys(kid: str):
+        private_jwk = JsonWebKey.generate_key(
+            "RSA", 2048, is_private=True, options={"kid": kid}
+        )
+        public_jwk = private_jwk.as_dict(is_private=False)
+        private_jwk_dict = private_jwk.as_dict(is_private=True)
+        return private_jwk_dict, public_jwk
+
+    @staticmethod
+    def _make_token(
+        private_jwk: dict,
+        kid: str,
+        issuer: str,
+        client_id: str,
+        overrides: dict | None = None,
+    ) -> str:
+        now = int(time.time())
+        payload = {
+            "iss": issuer,
+            "aud": client_id,
+            "iat": now,
+            "exp": now + 300,
+            "events": {"http://schemas.openid.net/event/backchannel-logout": {}},
+            "sub": "user-123",
+        }
+        if overrides:
+            # Apply overrides, removing keys with None values
+            for key, value in overrides.items():
+                if value is None:
+                    if key in payload:
+                        del payload[key]
+                else:
+                    payload[key] = value
+
+        headers = {"alg": "RS256", "kid": kid}
+        token = jwt.encode(headers, payload, private_jwk)
+        return token.decode("utf-8")
+
+    @staticmethod
+    def _key_loader(public_jwk: dict):
+        def loader(headers, payload):
+            return public_jwk
+
+        return loader
+
+    def test_valid_token_returns_claims(self):
+        kid = "k1"
+        issuer = "test-issuer"
+        client_id = "client-1"
+        private_jwk, public_jwk = self._make_keys(kid)
+        token = self._make_token(private_jwk, kid, issuer, client_id)
+
+        service = DecodeLogoutToken()
+        claims = service.call(
+            metadata={"issuer": issuer},
+            logout_token=token,
+            client_id=client_id,
+            key_loader=self._key_loader(public_jwk),
+        )
+
+        assert claims["iss"] == issuer
+        assert claims["aud"] == client_id
+        assert claims["sub"] == "user-123"
+        assert "http://schemas.openid.net/event/backchannel-logout" in claims["events"]
+
+    @pytest.mark.parametrize(
+        "overrides,expected_error_type,expected_error_text",
+        [
+            ({"iss": "wrong-issuer"}, InvalidClaimError, "iss"),  # invalid issuer
+            ({"aud": "wrong-aud"}, InvalidClaimError, "aud"),  # invalid audience
+            (
+                {"events": {}},
+                InvalidClaimError,
+                "events",
+            ),  # missing backchannel-logout event
+            (
+                {"events": {"some-other": {}}},
+                InvalidClaimError,
+                "events",
+            ),  # wrong events
+            (
+                {"nonce": "should-be-none"},
+                InvalidClaimError,
+                "nonce",
+            ),  # nonce must be None
+        ],
+    )
+    def test_invalid_claims_raise_error(
+        self, overrides, expected_error_type, expected_error_text
+    ):
+        kid = "k1"
+        issuer = "test-issuer"
+        client_id = "client-1"
+        private_jwk, public_jwk = self._make_keys(kid)
+        token = self._make_token(
+            private_jwk, kid, issuer, client_id, overrides=overrides
+        )
+
+        service = DecodeLogoutToken()
+        with pytest.raises(InvalidLogoutToken) as excinfo:
+            service.call(
+                metadata={"issuer": issuer},
+                logout_token=token,
+                client_id=client_id,
+                key_loader=self._key_loader(public_jwk),
+            )
+
+        # Assert on the cause type and error message content
+        assert isinstance(excinfo.value.cause, expected_error_type)
+        assert expected_error_text in str(excinfo.value.cause)
+
+    def test_invalid_signature_raises_error(self):
+        kid = "k1"
+        issuer = "test-issuer"
+        client_id = "client-1"
+        # Create two different key pairs
+        private_jwk_1, public_jwk_1 = self._make_keys(kid)
+        private_jwk_2, _public_jwk_2 = self._make_keys(kid)
+        # Sign with private_jwk_2 but verify with public_jwk_1 -> invalid signature
+        token = self._make_token(private_jwk_2, kid, issuer, client_id)
+
+        service = DecodeLogoutToken()
+        with pytest.raises(InvalidLogoutToken) as excinfo:
+            service.call(
+                metadata={"issuer": issuer},
+                logout_token=token,
+                client_id=client_id,
+                key_loader=self._key_loader(public_jwk_1),
+            )
+
+        # Invalid signature should raise a JoseError
+        assert isinstance(excinfo.value.cause, JoseError)
+        assert "signature" in str(excinfo.value.cause).lower()
+
+    def test_expired_token_raises_error(self):
+        kid = "k1"
+        issuer = "test-issuer"
+        client_id = "client-1"
+        private_jwk, public_jwk = self._make_keys(kid)
+        now = int(time.time())
+        token = self._make_token(
+            private_jwk,
+            kid,
+            issuer,
+            client_id,
+            overrides={"exp": now - 120},  # already expired beyond leeway
+        )
+
+        service = DecodeLogoutToken()
+        with pytest.raises(InvalidLogoutToken) as excinfo:
+            service.call(
+                metadata={"issuer": issuer},
+                logout_token=token,
+                client_id=client_id,
+                key_loader=self._key_loader(public_jwk),
+            )
+
+        # Expired token should raise an ExpiredTokenError
+        assert isinstance(excinfo.value.cause, ExpiredTokenError)
+        assert "expired" in str(excinfo.value.cause).lower()
+
+    def test_missing_iat_raises_error(self):
+        kid = "k1"
+        issuer = "test-issuer"
+        client_id = "client-1"
+        private_jwk, public_jwk = self._make_keys(kid)
+        # Use overrides to remove the iat claim
+        token = self._make_token(
+            private_jwk, kid, issuer, client_id, overrides={"iat": None}
+        )
+
+        service = DecodeLogoutToken()
+        with pytest.raises(InvalidLogoutToken) as excinfo:
+            service.call(
+                metadata={"issuer": issuer},
+                logout_token=token,
+                client_id=client_id,
+                key_loader=self._key_loader(public_jwk),
+            )
+
+        # Missing iat should raise a MissingClaimError
+        assert isinstance(excinfo.value.cause, MissingClaimError)
+        assert "iat" in str(excinfo.value.cause).lower()

manage_breast_screening/auth/tests/test_services.py

@@ -11,6 +11,11 @@
     # CIS2 OpenID Connect
     path("cis2/log-in/", views.cis2_login, name="cis2_login"),
     path("cis2/callback/", views.cis2_callback, name="cis2_callback"),
+    path(
+        "cis2/back-channel-logout/",
+        views.cis2_back_channel_logout,
+        name="cis2_back_channel_logout",
+    ),
     # JWKS endpoint for private_key_jwt
     path("cis2/jwks_uri", views.jwks, name="jwks"),
 ]