ping-mee-self-service-platform/portal/tests/test_idp.py

"""The OIDC adapter against a stand-in provider: real signed tokens, canned HTTP answers."""

import json
import time
from urllib.parse import parse_qs, urlsplit

import pytest
import requests
from joserfc import jwt
from joserfc.jwk import KeySet, OctKey, RSAKey

from portal.adapters.idp import IdpUnavailable, LoginRejected, OidcIdp
from portal.config.env import OidcConfig

ISSUER = "https://idp.example.test/application/o/portal/"
CLIENT_ID = "portal-client"
SECRET = "client-secret-value-0123456789abcdef"
BASE = "https://idp.example.test/application/o"
DISCOVERY_URL = ISSUER + ".well-known/openid-configuration"
JWKS_URL = f"{BASE}/jwks/"
TOKEN_URL = f"{BASE}/token/"
CONFIG = OidcConfig(ISSUER, CLIENT_ID, SECRET, "https://portal.example.test")

KEY = RSAKey.generate_key(2048, parameters={"kid": "k1", "use": "sig"})
OTHER_KEY = RSAKey.generate_key(2048, parameters={"kid": "k1", "use": "sig"})
ROTATED_KEY = RSAKey.generate_key(2048, parameters={"kid": "k2", "use": "sig"})


class Provider(requests.adapters.BaseAdapter):
    """Answers requests from a table of address -> (status, json) and records them."""

    def __init__(self):
        super().__init__()
        self.calls = []
        self.error = None
        self.routes = {
            DISCOVERY_URL: (
                200,
                {
                    "issuer": ISSUER,
                    "authorization_endpoint": f"{BASE}/authorize/",
                    "token_endpoint": TOKEN_URL,
                    "jwks_uri": JWKS_URL,
                },
            ),
            JWKS_URL: (200, KeySet([KEY]).as_dict(private=False)),
            TOKEN_URL: (200, {"access_token": "at", "token_type": "Bearer", "id_token": ""}),
        }

    def send(self, request, **kwargs):
        self.calls.append(request)
        if self.error:
            raise self.error
        status, body = self.routes[request.url.split("?")[0]]
        response = requests.Response()
        response.status_code = status
        response._content = json.dumps(body).encode()
        response.headers["Content-Type"] = "application/json"
        response.url = request.url
        response.request = request
        response.encoding = "utf-8"
        return response

    def close(self):
        pass

    def count(self, url):
        return sum(1 for r in self.calls if r.url.split("?")[0] == url)


@pytest.fixture
def provider():
    return Provider()


@pytest.fixture
def idp(provider):
    instance = OidcIdp(CONFIG)

    def factory():
        session = instance._default_session()
        session.mount("https://", provider)
        return session

    instance._factory = factory
    return instance


def mint(flow, *, key=KEY, alg="RS256", header=None, **overrides):
    claims = {
        "iss": ISSUER,
        "aud": CLIENT_ID,
        "sub": "user-1",
        "exp": int(time.time()) + 300,
        "iat": int(time.time()),
        "nonce": flow.nonce,
        "preferred_username": "alice",
    }
    claims.update(overrides)
    claims = {k: v for k, v in claims.items() if v is not None}
    return jwt.encode({"alg": alg, **(header or {"kid": "k1"})}, claims, key)


def sign_in(idp, provider, flow, token, **params):
    provider.routes[TOKEN_URL] = (
        200,
        {"access_token": "at", "token_type": "Bearer", "id_token": token},
    )
    return idp.complete({"state": flow.state, "code": "the-code", **params}, flow)


def test_begin_builds_a_pkce_authorization_request(idp):
    url, flow = idp.begin()
    parts = urlsplit(url)
    query = {k: v[0] for k, v in parse_qs(parts.query).items()}
    assert f"{parts.scheme}://{parts.netloc}{parts.path}" == f"{BASE}/authorize/"
    assert query["response_type"] == "code"
    assert query["client_id"] == CLIENT_ID
    assert query["redirect_uri"] == "https://portal.example.test/auth/callback"
    assert query["scope"] == "openid profile email"
    assert query["code_challenge_method"] == "S256"
    assert query["state"] == flow.state and query["nonce"] == flow.nonce
    assert "code_challenge" in query
    assert flow.code_verifier not in url  # only its hash is sent


def test_each_flow_has_fresh_random_values(idp):
    _, first = idp.begin()
    _, second = idp.begin()
    assert len({first.state, second.state, first.nonce, second.nonce}) == 4


def test_discovery_is_fetched_once(idp, provider):
    idp.begin()
    idp.begin()
    assert provider.count(DISCOVERY_URL) == 1


def test_a_valid_sign_in_returns_the_identity(idp, provider):
    _, flow = idp.begin()
    identity = sign_in(idp, provider, flow, mint(flow))
    assert (identity.username, identity.subject) == ("alice", "user-1")


def test_the_code_is_exchanged_with_the_verifier_and_client_authentication(idp, provider):
    _, flow = idp.begin()
    sign_in(idp, provider, flow, mint(flow))
    request = [r for r in provider.calls if r.url == TOKEN_URL][0]
    form = parse_qs(request.body)
    assert form["grant_type"] == ["authorization_code"]
    assert form["code"] == ["the-code"]
    assert form["code_verifier"] == [flow.code_verifier]
    assert form["redirect_uri"] == ["https://portal.example.test/auth/callback"]
    assert request.headers["Authorization"].startswith("Basic ")


def test_an_audience_list_that_contains_the_client_is_accepted(idp, provider):
    _, flow = idp.begin()
    assert sign_in(idp, provider, flow, mint(flow, aud=[CLIENT_ID, "other"])).username == "alice"


@pytest.mark.parametrize(
    "params", [{"state": "wrong"}, {"state": ""}, {"code": ""}, {"error": "access_denied"}]
)
def test_a_bad_callback_is_rejected_before_any_token_request(idp, provider, params):
    _, flow = idp.begin()
    with pytest.raises(LoginRejected):
        idp.complete({"state": flow.state, "code": "c", **params}, flow)
    assert provider.count(TOKEN_URL) == 0


def test_a_refused_code_is_rejected(idp, provider):
    _, flow = idp.begin()
    provider.routes[TOKEN_URL] = (400, {"error": "invalid_grant"})
    with pytest.raises(LoginRejected):
        idp.complete({"state": flow.state, "code": "c"}, flow)


def test_an_unreachable_token_endpoint_is_an_outage(idp, provider):
    _, flow = idp.begin()
    provider.error = requests.ConnectionError("down")
    with pytest.raises(IdpUnavailable):
        idp.complete({"state": flow.state, "code": "c"}, flow)


def test_a_missing_id_token_is_rejected(idp, provider):
    _, flow = idp.begin()
    provider.routes[TOKEN_URL] = (200, {"access_token": "at", "token_type": "Bearer"})
    with pytest.raises(LoginRejected):
        idp.complete({"state": flow.state, "code": "c"}, flow)


@pytest.mark.parametrize(
    "overrides",
    [
        {"nonce": "somebody-elses"},
        {"nonce": None},
        {"iss": "https://evil.example.test/"},
        {"aud": "another-client"},
        {"exp": int(time.time()) - 3600},
        {"exp": None},
        {"sub": None},
        {"preferred_username": None},
        {"preferred_username": "   "},
        {"preferred_username": 42},
    ],
)
def test_invalid_claims_are_rejected(idp, provider, overrides):
    _, flow = idp.begin()
    with pytest.raises(LoginRejected):
        sign_in(idp, provider, flow, mint(flow, **overrides))


def test_a_token_signed_by_another_key_is_rejected(idp, provider):
    _, flow = idp.begin()
    with pytest.raises(LoginRejected):
        sign_in(idp, provider, flow, mint(flow, key=OTHER_KEY))


def test_a_token_signed_with_the_client_secret_is_rejected(idp, provider):
    _, flow = idp.begin()
    forged = mint(flow, key=OctKey.import_key(SECRET), alg="HS256", header={})
    with pytest.raises(LoginRejected):
        sign_in(idp, provider, flow, forged)


def test_an_unsigned_token_is_rejected(idp, provider):
    _, flow = idp.begin()
    unsigned = "eyJhbGciOiJub25lIn0.eyJzdWIiOiJ4In0."
    with pytest.raises(LoginRejected):
        sign_in(idp, provider, flow, unsigned)


def test_a_rotated_key_is_picked_up_with_one_refetch(idp, provider):
    _, flow = idp.begin()
    sign_in(idp, provider, flow, mint(flow))  # caches the key set with k1 only
    provider.routes[JWKS_URL] = (200, KeySet([KEY, ROTATED_KEY]).as_dict(private=False))
    _, flow = idp.begin()
    token = mint(flow, key=ROTATED_KEY, header={"kid": "k2"})
    assert sign_in(idp, provider, flow, token).username == "alice"
    assert provider.count(JWKS_URL) == 2


def test_an_unknown_key_is_rejected_after_one_refetch(idp, provider):
    _, flow = idp.begin()
    token = mint(flow, key=ROTATED_KEY, header={"kid": "k2"})
    with pytest.raises(LoginRejected):
        sign_in(idp, provider, flow, token)
    assert provider.count(JWKS_URL) == 2


def test_a_different_issuer_in_discovery_is_an_outage(idp, provider):
    provider.routes[DISCOVERY_URL] = (
        200,
        {**provider.routes[DISCOVERY_URL][1], "issuer": "https://evil.example.test/"},
    )
    with pytest.raises(IdpUnavailable):
        idp.begin()


def test_a_discovery_document_with_a_plain_http_endpoint_is_an_outage(idp, provider):
    document = {**provider.routes[DISCOVERY_URL][1], "token_endpoint": "http://x.test/token"}
    provider.routes[DISCOVERY_URL] = (200, document)
    with pytest.raises(IdpUnavailable):
        idp.begin()


def test_an_unreachable_provider_is_an_outage(idp, provider):
    provider.error = requests.ConnectionError("down")
    with pytest.raises(IdpUnavailable):
        idp.begin()