PPHA-475: Add basic NHS Login backend

Author: Themitchell

.env.example

@@ -11,3 +11,7 @@ DATABASE_PASSWORD=password
 POSTGRES_DB=lung_cancer_screening
 POSTGRES_USER=lung_cancer_screening
 POSTGRES_PASSWORD=password
+
+OIDC_RP_CLIENT_PRIVATE_KEY="MYSUPERSECRETPRIVATEKEY"
+OIDC_RP_CLIENT_ID="lcrc"
+OIDC_OP_FQDN="https://example.com"

lung_cancer_screening/questions/auth.py

@@ -0,0 +1,127 @@
+"""
+Custom OIDC authentication backend for NHS Login with private key JWT.
+"""
+import logging
+import time
+import requests
+import jwt
+from django.conf import settings
+from django.contrib.auth import get_user_model
+
+from mozilla_django_oidc.auth import OIDCAuthenticationBackend
+from cryptography.hazmat.primitives import serialization
+
+logger = logging.getLogger(__name__)
+
+
+class NHSLoginOIDCBackend(OIDCAuthenticationBackend):
+    """
+    Custom OIDC authentication backend that uses private key JWT
+    for client authentication instead of client secret.
+    Uses NHS number as the username field.
+    """
+
+    def filter_users_by_claims(self, claims):
+        """Find users by NHS number from OIDC claims."""
+        User = get_user_model()
+
+        nhs_number = claims.get('nhs_number')
+        if not nhs_number:
+            return User.objects.none()
+
+        return User.objects.filter(nhs_number=nhs_number)
+
+    def create_user(self, claims):
+        User = get_user_model()
+
+        nhs_number = claims.get('nhs_number')
+        if not nhs_number:
+            raise ValueError("Missing 'nhs_number' claim in OIDC token")
+        return User.objects.create_user(nhs_number=nhs_number)
+
+    def update_user(self, user, _claims):
+        return user
+
+    def _create_client_assertion(self):
+        private_key_pem = settings.OIDC_RP_CLIENT_PRIVATE_KEY
+        if not private_key_pem:
+            return None
+
+        try:
+            private_key = serialization.load_pem_private_key(
+                private_key_pem.encode('utf-8'),
+                password=None,
+            )
+        except Exception as e:
+            raise ValueError(f"Failed to load private key: {e}") from e
+
+        token_endpoint = settings.OIDC_OP_TOKEN_ENDPOINT
+        client_id = settings.OIDC_RP_CLIENT_ID
+
+        now = int(time.time())
+        claims = {
+            'iss': client_id,
+            'sub': client_id,
+            'aud': token_endpoint,
+            'jti': f"{client_id}-{now}",
+            'iat': now,
+            'exp': now + 300,
+        }
+
+        headers = {'alg': settings.OIDC_RP_SIGN_ALGO}
+
+        assertion = jwt.encode(
+            claims, private_key, algorithm=settings.OIDC_RP_SIGN_ALGO, headers=headers
+        )
+
+        if isinstance(assertion, bytes):
+            return assertion.decode('utf-8')
+        return assertion
+
+    def get_token(self, token_payload):
+        token_endpoint = settings.OIDC_OP_TOKEN_ENDPOINT
+        redirect_uri = token_payload.get('redirect_uri')
+        authorization_code = token_payload.get('code')
+
+        if not redirect_uri:
+            redirect_uri = settings.OIDC_RP_REDIRECT_URI
+
+        client_assertion = self._create_client_assertion()
+        if not client_assertion:
+            return super().get_token(token_payload)
+
+        token_payload_updated = {
+            'grant_type': 'authorization_code',
+            'code': authorization_code,
+            'redirect_uri': redirect_uri,
+            'client_assertion': client_assertion,
+            'client_assertion_type': (
+                'urn:ietf:params:oauth:client-assertion-type:jwt-bearer'
+            ),
+        }
+
+        response = requests.post(
+            token_endpoint,
+            data=token_payload_updated,
+            headers={'Content-Type': 'application/x-www-form-urlencoded'},
+            verify=True,
+            timeout=30,
+        )
+
+        if not response.ok:
+            error_detail = response.text
+            try:
+                error_detail = response.json()
+            except Exception:
+                pass
+            logger.error(
+                "Token request failed: %s - %s",
+                response.status_code,
+                error_detail
+            )
+            raise ValueError(
+                f"Token request failed: {response.status_code} - "
+                f"{error_detail}"
+            )
+
+        return response.json()

lung_cancer_screening/questions/migrations/0019_user.py

@@ -0,0 +1,28 @@
+# Generated by Django 5.2.9 on 2025-12-03 16:51
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('questions', '0018_responseset_respiratory_conditions'),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='User',
+            fields=[
+                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('password', models.CharField(max_length=128, verbose_name='password')),
+                ('last_login', models.DateTimeField(blank=True, null=True, verbose_name='last login')),
+                ('nhs_number', models.CharField(max_length=10, unique=True)),
+                ('created_at', models.DateTimeField(auto_now_add=True)),
+                ('updated_at', models.DateTimeField(auto_now=True)),
+            ],
+            options={
+                'verbose_name': 'user',
+                'verbose_name_plural': 'users',
+            },
+        )
+    ]

lung_cancer_screening/questions/models/__init__.py

@@ -1,2 +1,3 @@
 from .participant import Participant # noqa: F401
 from .response_set import ResponseSet # noqa: F401
+from .user import User # noqa: F401

lung_cancer_screening/questions/models/user.py

@@ -0,0 +1,36 @@
+from django.contrib.auth.models import (
+    AbstractBaseUser,
+    BaseUserManager
+)
+from django.db import models
+
+
+class UserManager(BaseUserManager):
+
+    def create_user(self, nhs_number, **extra_fields):
+        if not nhs_number:
+            raise ValueError('The NHS number must be set')
+        user = self.model(nhs_number=nhs_number, **extra_fields)
+        # Set an unusable password since AbstractBaseUser requires it
+        user.set_unusable_password()
+        user.save(using=self._db)
+        return user
+
+
+class User(AbstractBaseUser):
+    nhs_number = models.CharField(max_length=10, unique=True)
+    created_at = models.DateTimeField(auto_now_add=True)
+    updated_at = models.DateTimeField(auto_now=True)
+
+    objects = UserManager()
+
+    USERNAME_FIELD = 'nhs_number'
+    REQUIRED_FIELDS = []
+
+    def save(self, *args, **kwargs):
+        self.full_clean()  # Validate before saving
+        super().save(*args, **kwargs)
+
+    class Meta:
+        verbose_name = 'user'
+        verbose_name_plural = 'users'

lung_cancer_screening/questions/tests/unit/models/test_user.py

@@ -0,0 +1,51 @@
+from django.test import TestCase
+from datetime import datetime
+from django.core.exceptions import ValidationError
+
+from ....models.user import User
+
+
+class TestUser(TestCase):
+    def setUp(self):
+        self.nhs_number = "1234567890"
+        self.user = User.objects.create_user(self.nhs_number)
+
+    def test_has_nhs_number_as_a_string(self):
+        self.assertIsInstance(
+            self.user.nhs_number,
+            str
+        )
+
+    def test_has_created_at_as_a_datetime(self):
+        self.assertIsInstance(
+            self.user.created_at,
+            datetime
+        )
+
+    def test_has_updated_at_as_a_datetime(self):
+        self.assertIsInstance(
+            self.user.updated_at,
+            datetime
+        )
+
+    def test_nhs_number_has_a_max_length_of_10(self):
+        with self.assertRaises(ValidationError) as context:
+            User.objects.create_user("1"*11)
+
+        self.assertIn(
+            "Ensure this value has at most 10 characters (it has 11).",
+            context.exception.messages
+        )
+
+    def test_raises_a_value_error_if_nhs_number_is_null(self):
+        with self.assertRaises(ValueError):
+            User.objects.create_user(None)
+
+    def test_raises_a_validation_error_if_nhs_number_is_duplicate(self):
+        with self.assertRaises(ValidationError) as context:
+            User.objects.create_user(self.nhs_number)
+
+        self.assertIn(
+            "User with this Nhs number already exists.",
+            context.exception.messages
+        )