coulomb-social/tests/test_oidc_userinfo_binding.py

"""Exercise callback identity binding with real signed tokens and synthetic HTTP."""

import time

import httpx
import pytest
from authlib.jose import JsonWebKey, jwt
from django.urls import reverse

from coulomb_social.apps.identity import oidc, views
from coulomb_social.apps.members.models import Member

ISSUER = "https://issuer.example.test"
CLIENT = "synthetic-coulomb-social"


@pytest.fixture(scope="module")
def signing_key():
    return JsonWebKey.generate_key("RSA", 2048, is_private=True)


@pytest.fixture
def callback(client, settings, monkeypatch, signing_key):
    settings.OIDC_ENABLED = True
    settings.OIDC_ISSUER = ISSUER
    settings.OIDC_CLIENT_ID = CLIENT
    settings.OIDC_REDIRECT_URI = "https://app.example.test/auth/callback/"
    settings.USER_ENGINE_BASE_URL = ""
    settings.USER_ENGINE_PROXY_SECRET = ""
    settings.FLEX_AUTH_BASE_URL = ""

    def invoke(userinfo, *, token_overrides=None, signed_overrides=None):
        claims = {"iss": ISSUER, "aud": CLIENT, "sub": "signed-member",
                  "exp": int(time.time()) + 120, "iat": int(time.time()),
                  "tenant": "tenant:synthetic", "roles": ["user"], "groups": [],
                  "acr": "aal1", "assurance": {"mfa": False},
                  "name": "Token Profile", "email": "member@example.test"}
        claims.update(signed_overrides or {})
        token = {"id_token": jwt.encode({"alg": "RS256", "kid": "synthetic"},
                                        claims, signing_key).decode(),
                 "access_token": "synthetic-access-token"}
        token.update(token_overrides or {})
        monkeypatch.setattr(oidc, "exchange_code", lambda *args, **kwargs: token)

        def get(url, **kwargs):
            responses = {
                ISSUER + "/.well-known/openid-configuration": {
                    "jwks_uri": ISSUER + "/jwks", "userinfo_endpoint": ISSUER + "/userinfo"},
                ISSUER + "/jwks": {"keys": [{**signing_key.as_dict(is_private=False),
                                             "kid": "synthetic"}]},
                ISSUER + "/userinfo": userinfo,
            }
            assert url in responses, "Only synthetic identity endpoints permitted"
            return httpx.Response(200, json=responses[url], request=httpx.Request("GET", url))

        monkeypatch.setattr(httpx, "get", get)
        session = client.session
        session[views.SESSION_OIDC_STATE] = "synthetic-state"
        session[views.SESSION_OIDC_VERIFIER] = "synthetic-verifier"
        session.save()
        return client.get(reverse("identity:callback") + "?code=synthetic&state=synthetic-state")

    return invoke


@pytest.mark.django_db
@pytest.mark.usecase("CSOC-UC-009")
@pytest.mark.parametrize("userinfo", [
    {"sub": "another-member", "email": "member@example.test"},
    {"name": "Profile without subject"},
    {"sub": 123},
])
def test_callback_refuses_userinfo_without_matching_signed_subject(callback, client, userinfo):
    assert callback(userinfo).status_code == 302
    assert "_auth_user_id" not in client.session
    assert Member.objects.count() == 0
    assert views.SESSION_OIDC_STATE not in client.session


@pytest.mark.django_db
@pytest.mark.usecase("CSOC-UC-009")
@pytest.mark.parametrize("unsigned", [None, "", 123])
def test_callback_requires_verified_id_token_even_with_userinfo(callback, client, unsigned):
    assert callback({"sub": "signed-member"}, token_overrides={
        "id_token": unsigned, "userinfo": {"sub": "token-response-member"}}).status_code == 302
    assert "_auth_user_id" not in client.session
    assert Member.objects.count() == 0


@pytest.mark.django_db
@pytest.mark.usecase("CSOC-UC-009")
def test_matching_userinfo_enriches_profile_without_replacing_signed_authority(callback, client):
    assert callback({"sub": "signed-member", "name": "Current Profile",
                     "email": "current@example.test", "iss": "https://other.example.test",
                     "aud": "other-client", "tenant": "tenant:other",
                     "roles": ["admin"], "groups": ["admins"], "acr": "aal2",
                     "assurance": {"mfa": True}, "azp": "other-client"}).status_code == 302
    member = Member.objects.get(subject="signed-member")
    assert member.issuer == ISSUER
    assert member.tenant_id == "tenant:synthetic"
    assert member.display_name == "Current Profile"
    assert member.email == "current@example.test"
    assert str(client.session["_auth_user_id"]) == str(member.user_id)
    assert not member.user.has_usable_password()
    # Check the actual downstream claims rather than treating login alone as
    # proof that an overlay did not change the asserted authority.
    assert client.session["identity_roles"] == ["user"]
    assert client.session["identity_groups"] == []
    assert client.session["identity_assurance"] == {"mfa": False}


@pytest.mark.django_db
@pytest.mark.usecase("CSOC-UC-009")
def test_unavailable_userinfo_keeps_verified_token_login(callback, client):
    assert callback({}).status_code == 302
    assert Member.objects.get(subject="signed-member").issuer == ISSUER
    assert "_auth_user_id" in client.session


@pytest.mark.django_db
@pytest.mark.usecase("CSOC-UC-009")
def test_userinfo_cannot_satisfy_requested_step_up(callback, client):
    session = client.session
    session[views.SESSION_OIDC_REQUIRED_ACR] = "aal2"
    session.save()
    response = callback({"sub": "signed-member", "acr": "aal2", "assurance": {"mfa": True}})
    assert response.status_code == 400
    assert Member.objects.count() == 0
    assert "_auth_user_id" not in client.session


@pytest.mark.django_db
@pytest.mark.usecase("CSOC-UC-009")
@pytest.mark.parametrize("overrides", [
    {"iss": "https://other.example.test"}, {"aud": "other-client"},
    {"exp": 1}, {"sub": ""},
])
def test_valid_userinfo_cannot_rescue_invalid_signed_token(callback, client, overrides):
    assert callback({"sub": "signed-member"}, signed_overrides=overrides).status_code == 302
    assert Member.objects.count() == 0
    assert "_auth_user_id" not in client.session