"""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"] = ''
login[1].change = change
sid = finish(login)
body = call(App(login[0]), cookie=f"{SESSION_COOKIE}={sid}")["body"]
assert "