informed-decision/tests/test_browser_auth.py

"""Signed synthetic issuer fixtures: these are not proof of a native human login."""

import base64
import copy
import hashlib
import io
import json
import time
from urllib.parse import parse_qs, urlencode, urlsplit

from cryptography.hazmat.primitives.asymmetric import rsa
import jwt
import pytest

from informed_decision.http_transport import JSONTransport, TransportError
from informed_decision.oidc import CALLBACK, CLIENT_ID, KeyCapeLogin, LoginError, ORIGIN
from informed_decision.provenance import Route
from informed_decision.web import App, FLOW_COOKIE, SESSION_COOKIE

ISSUER = "https://keycape.test"


@pytest.fixture(scope="module")
def signing_key():
    return rsa.generate_private_key(public_exponent=65537, key_size=2048)


class IssuerFixture:
    def __init__(self, key):
        self.key = key
        self.params = None
        self.calls = []
        self.change = lambda tokens, claims: None
        self.now = int(time.time())

    def request(self, method, url, *, headers=None, body=None):
        self.calls.append((method, url, headers, body))
        if url == ISSUER + "/jwks":
            jwk = json.loads(jwt.algorithms.RSAAlgorithm.to_jwk(self.key.public_key()))
            return 200, {"keys": [{**jwk, "kid": "key-1", "alg": "RS256", "use": "sig"}]}
        assert method == "POST" and url == ISSUER + "/token"
        params = {k: v[0] for k, v in parse_qs(body.decode()).items()}
        assert params["client_id"] == CLIENT_ID and params["redirect_uri"] == CALLBACK
        assert params["grant_type"] == "authorization_code" and "client_secret" not in params
        challenge = base64.urlsafe_b64encode(hashlib.sha256(params["code_verifier"].encode()).digest()).rstrip(b"=").decode()
        assert challenge == self.params["code_challenge"]
        claims = {"iss": ISSUER, "sub": "human-fixture", "iat": self.now, "exp": self.now + 900,
                  "tenant": "tenant:platform", "tenant_source": "registration", "principal_type": "human",
                  "roles": [], "assurance": {"at": self.now - 120, "level": "aal2", "mfa": True,
                      "methods": ["pwd", "otp"], "source": "key-cape"}}
        payloads = {"id_token": {**copy.deepcopy(claims), "aud": CLIENT_ID, "nonce": self.params["nonce"]},
                    "access_token": {**copy.deepcopy(claims), "aud": "approval-engine",
                                     "scope": "openid approval:read approval:approve"}}
        tokens = {"token_type": "Bearer"}
        self.change(tokens, payloads)
        for name, value in payloads.items():
            tokens.setdefault(name, jwt.encode(value, self.key, algorithm="RS256", headers={"kid": "key-1"}))
        return 200, tokens


@pytest.fixture
def login(signing_key):
    transport = IssuerFixture(signing_key)
    login = KeyCapeLogin(ISSUER, transport=transport)
    return login, transport


def begin(pair):
    login, issuer = pair
    url, browser = login.start()
    issuer.params = {k: v[0] for k, v in parse_qs(urlsplit(url).query).items()}
    return issuer.params["state"], browser


def finish(pair):
    state, browser = begin(pair)
    return pair[0].finish(state, browser, "synthetic-code")


def call(app, path="/", method="GET", cookie="", query="", body=b"", origin=None, content_type="application/x-www-form-urlencoded"):
    env = {"PATH_INFO": path, "REQUEST_METHOD": method, "HTTP_COOKIE": cookie,
           "QUERY_STRING": query, "wsgi.input": io.BytesIO(body), "CONTENT_LENGTH": str(len(body)),
           "CONTENT_TYPE": content_type}
    if origin is not None:
        env["HTTP_ORIGIN"] = origin
    response = {}
    def start(status, headers):
        response.update(status=int(status.split()[0]), headers=headers)
    response["body"] = b"".join(app(env, start)).decode()
    return response


def test_pkce_nonce_scope_and_imported_human_session(login):
    sid = finish(login)
    session = login[0].session(sid)
    assert session.subject == "human-fixture"
    assert session.principal_type.route is Route.AUTHENTICATION
    assert session.tenant.route is Route.REGISTRATION
    assert session.assurance["at"] == login[1].now - 120  # Authentication, never mint time.
    assert session.access_token not in repr(session)
    assert login[1].params["scope"] == "openid approval:read approval:approve"
    assert login[1].params["code_challenge_method"] == "S256"
    assert login[1].params["nonce"] != login[1].params["state"]


@pytest.mark.parametrize("where,key,value", [
    ("access_token", "aud", CLIENT_ID), ("id_token", "aud", "approval-engine"),
    ("access_token", "aud", ["approval-engine", CLIENT_ID]),
    ("access_token", "iss", "https://attacker.test"),
    ("id_token", "nonce", "wrong"), ("id_token", "nonce", None),
    ("access_token", "sub", "another-human"), ("access_token", "sub", ""),
    ("both", "principal_type", "service"), ("both", "principal_type", "agent"),
    ("both", "tenant", "tenant:coulomb"), ("both", "tenant_source", None),
    ("both", "tenant_source", "default"), ("both", "tenant_source", "invented"),
    ("both", "tenant_source", ["registration"]),
    ("access_token", "scope", "openid approval:read approval:approve approval:consume"),
    ("access_token", "scope", "openid approval:approve"), ("access_token", "scope", ["openid"]),
    ("access_token", "roles", None), ("access_token", "roles", [1]),
    ("both", "assurance", {}),
    ("access_token", "exp", 1), ("id_token", "exp", 1),
    ("access_token", "iat", 9999999999), ("access_token", "exp", "9999999999"),
])
def test_invalid_identity_refused(login, where, key, value):
    def change(tokens, claims):
        for name in claims if where == "both" else [where]:
            claims[name][key] = value
    login[1].change = change
    with pytest.raises(LoginError):
        finish(login)
    assert not login[0]._sessions


@pytest.mark.parametrize("key,value", [
    ("mfa", False), ("mfa", 1), ("level", "aal1"), ("methods", ["pwd"]),
    ("source", "self"), ("at", 0), ("at", "123"), ("at", 9999999999),
])
def test_mfa_profile_required(login, key, value):
    def change(tokens, claims):
        for payload in claims.values():
            payload["assurance"][key] = value
    login[1].change = change
    with pytest.raises(LoginError, match="MFA"):
        finish(login)


def test_directory_provenance_preserved(login):
    def change(tokens, claims):
        for payload in claims.values():
            payload["tenant_source"] = "directory"
    login[1].change = change
    assert login[0].session(finish(login)).tenant.route is Route.DIRECTORY


@pytest.mark.parametrize("kind", ["wrong-browser", "missing-browser", "unknown-state", "expired", "missing-code"])
def test_callback_fails_before_exchange(login, kind):
    state, browser = begin(login)
    code = "code"
    if kind == "wrong-browser": browser = "other"
    if kind == "missing-browser": browser = ""
    if kind == "unknown-state": state = "unknown"
    if kind == "expired": login[0].clock = lambda: time.time() + 301
    if kind == "missing-code": code = ""
    with pytest.raises(LoginError): login[0].finish(state, browser, code)
    assert not login[1].calls


def test_callback_single_use_even_after_refusal(login):
    state, browser = begin(login)
    login[0].finish(state, browser, "code")
    count = len(login[1].calls)
    with pytest.raises(LoginError): login[0].finish(state, browser, "code")
    assert len(login[1].calls) == count


@pytest.mark.parametrize("value", ["broken", "", None])
def test_bad_token_never_becomes_session(login, value):
    login[1].change = lambda tokens, claims: tokens.update(access_token=value)
    with pytest.raises(LoginError): finish(login)


def test_wrong_signature_refused(login):
    other = rsa.generate_private_key(public_exponent=65537, key_size=2048)
    def change(tokens, claims):
        tokens["access_token"] = jwt.encode(claims["access_token"], other, algorithm="RS256", headers={"kid": "key-1"})
    login[1].change = change
    with pytest.raises(LoginError): finish(login)


@pytest.mark.parametrize("algorithm,kid", [("HS256", "key-1"), ("RS256", "unknown-key")])
def test_algorithm_confusion_and_unknown_key_refused(login, algorithm, kid):
    def change(tokens, claims):
        key = b"synthetic-key-material-for-this-test-only" if algorithm == "HS256" else login[1].key
        tokens["access_token"] = jwt.encode(claims["access_token"], key, algorithm=algorithm, headers={"kid": kid})
    login[1].change = change
    with pytest.raises(LoginError): finish(login)


@pytest.mark.parametrize("failure", ["token", "jwks", "expired-jwks-cache"])
def test_dependency_outage_cannot_authenticate_from_stale_keys(login, failure):
    if failure == "expired-jwks-cache":
        finish(login)
        login[0]._keys_until = 0
    state, browser = begin(login)
    original = login[1].request
    def request(method, url, **kwargs):
        if url.endswith("/token" if failure == "token" else "/jwks"):
            raise TransportError("secret-sentinel")
        return original(method, url, **kwargs)
    login[1].request = request
    count = len(login[0]._sessions)
    with pytest.raises(LoginError) as error: login[0].finish(state, browser, "code")
    assert "secret-sentinel" not in str(error.value) and len(login[0]._sessions) == count


def test_browser_subject_is_html_escaped(login):
    def change(tokens, claims):
        for payload in claims.values(): payload["sub"] = '<script>alert("fixture")</script>'
    login[1].change = change
    sid = finish(login)
    body = call(App(login[0]), cookie=f"{SESSION_COOKIE}={sid}")["body"]
    assert "<script>" not in body and "&lt;script&gt;" in body


def test_no_session_survives_expiry_logout_or_process_restart(login):
    sid = finish(login)
    login[0].logout(sid)
    assert login[0].session(sid) is None
    sid = finish(login)
    assert KeyCapeLogin(ISSUER).session(sid) is None
    login[0].clock = lambda: time.time() + 901
    assert login[0].session(sid) is None


def test_pending_and_session_capacity_fail_closed(login):
    login[0].capacity = 1
    sid = finish(login)
    assert login[0].session(sid)
    with pytest.raises(LoginError, match="session capacity"): finish(login)
    login[0].start()
    with pytest.raises(LoginError, match="login capacity"): login[0].start()


def test_callback_browser_cookie_rotation_and_token_secrecy(login):
    app = App(login[0])
    old_sid = finish(login)
    response = call(app, "/auth/start")
    headers = dict(response["headers"])
    login[1].params = {k: v[0] for k, v in parse_qs(urlsplit(headers["Location"]).query).items()}
    cookie = headers["Set-Cookie"].split(";")[0]
    response = call(app, "/auth/callback", cookie=cookie + f"; {SESSION_COOKIE}={old_sid}",
                    query=urlencode({"code": "code", "state": login[1].params["state"]}))
    assert response["status"] == 303
    assert dict(response["headers"])["Location"] == "/"
    assert login[0].session(old_sid) is None
    cookies = [v for k, v in response["headers"] if k == "Set-Cookie"]
    assert all("Secure; HttpOnly; SameSite=Lax" in v and "Path=/" in v for v in cookies)
    assert any(v.startswith(FLOW_COOKIE + "=;") for v in cookies)
    new_cookie = next(v.split(";")[0] for v in cookies if v.startswith(SESSION_COOKIE))
    response = call(app, cookie=new_cookie)
    session = login[0].session(new_cookie.split("=")[1])
    assert "Signed in as human-fixture" in response["body"]
    assert session.access_token not in str(response)
    assert dict(response["headers"])["Cache-Control"] == "no-store"


def test_issuer_error_and_duplicate_params_do_not_leak_or_exchange(login):
    state, browser = begin(login)
    response = call(App(login[0]), "/auth/callback", cookie=f"{FLOW_COOKIE}={browser}",
                    query=urlencode({"state": state, "error": "denied", "error_description": "secret-sentinel"}))
    assert response["status"] == 401 and "secret-sentinel" not in str(response)
    assert not login[1].calls
    response = call(App(login[0]), "/auth/callback", query="code=x&code=y")
    assert response["status"] == 400


@pytest.mark.parametrize("origin,csrf,expected", [(ORIGIN, "correct", 303), (ORIGIN, "wrong", 403),
    (None, "correct", 403), ("null", "correct", 403), ("https://attacker.test", "correct", 403)])
def test_logout_requires_origin_and_session_csrf(login, origin, csrf, expected):
    sid = finish(login)
    token = login[0].session(sid).csrf if csrf == "correct" else "wrong"
    response = call(App(login[0]), "/auth/logout", method="POST", cookie=f"{SESSION_COOKIE}={sid}",
                    body=urlencode({"csrf": token}).encode(), origin=origin)
    assert response["status"] == expected
    assert bool(login[0].session(sid)) == (expected != 303)


@pytest.mark.parametrize("path,policy", [("/", "same-origin"), ("/review", "same-origin"),
    ("/auth/callback", "no-referrer"), ("/presentations/pres-fixture/packet/0", "no-referrer")])
def test_browser_referrer_policy_preserves_form_origin_and_protects_auth_urls(login, path, policy):
    assert dict(call(App(login[0]), path)["headers"])["Referrer-Policy"] == policy


@pytest.mark.parametrize("path,method,expected", [("/healthz", "GET", 200), ("/readyz", "GET", 503),
    ("/approvals/approval-fixture", "GET", 404), ("/approvals/approval-fixture/accept", "POST", 404),
    ("/v1/approvals/approval-fixture/entries", "POST", 404), ("/auth/logout", "GET", 404)])
def test_authentication_does_not_enable_memo_or_binding_routes(login, path, method, expected):
    sid = finish(login)
    assert call(App(login[0]), path, method, cookie=f"{SESSION_COOKIE}={sid}")["status"] == expected


@pytest.mark.parametrize("issuer", ["http://keycape.test", "https://user:pass@keycape.test", "https://keycape.test/path",
                                     "https://keycape.test?redirect=bad", "https://keycape.test\n"])
def test_issuer_endpoint_must_be_fixed_https_origin(issuer):
    with pytest.raises(ValueError): KeyCapeLogin(issuer)


def test_transport_refuses_cleartext_before_network():
    with pytest.raises(TransportError): JSONTransport().request("POST", "http://attacker.test/token", body=b"secret")