tegwick
03f18d8258
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a10780-6d41-7952-9718-dfbc1f41236f
105 lines
4 KiB
Python
105 lines
4 KiB
Python
import base64
|
|
import json
|
|
import time
|
|
from urllib.parse import parse_qs, unquote_plus
|
|
|
|
from cryptography.hazmat.primitives.asymmetric import rsa
|
|
import jwt
|
|
import pytest
|
|
|
|
from informed_decision.requester import RequesterError, RequesterTokenProvider, CLIENT_ID
|
|
|
|
ISSUER = "https://identity.example.test"
|
|
|
|
|
|
@pytest.fixture
|
|
def lane(tmp_path):
|
|
private = rsa.generate_private_key(public_exponent=65537, key_size=2048)
|
|
jwk = json.loads(jwt.algorithms.RSAAlgorithm.to_jwk(private.public_key()))
|
|
jwk.update(kid="test", alg="RS256", use="sig")
|
|
now = int(time.time())
|
|
claims = dict(iss=ISSUER, aud="approval-engine", sub="informed-decision", iat=now,
|
|
exp=now + 900, tenant="tenant:platform", principal_type="service",
|
|
roles=[CLIENT_ID], scope="approval:create")
|
|
secret = tmp_path / "client-secret"
|
|
secret.write_text("synthetic-first")
|
|
class Transport:
|
|
seen = []
|
|
def request(self, method, url, *, headers=None, body=None):
|
|
if method == "GET":
|
|
assert url == ISSUER + "/jwks"
|
|
return 200, {"keys": [jwk]}
|
|
assert url == ISSUER + "/token"
|
|
fields = parse_qs(body.decode())
|
|
assert set(fields) == {"grant_type", "scope"}
|
|
scheme, encoded = headers["Authorization"].split(" ", 1)
|
|
assert scheme == "Basic"
|
|
client_id, credential = base64.b64decode(encoded, validate=True).decode().split(":", 1)
|
|
assert unquote_plus(client_id) == CLIENT_ID
|
|
fields["decoded_credential"] = unquote_plus(credential)
|
|
self.seen.append(fields)
|
|
return 200, {"token_type": "Bearer", "access_token": jwt.encode(
|
|
claims, private, algorithm="RS256", headers={"kid": "test"})}
|
|
transport = Transport()
|
|
return RequesterTokenProvider(ISSUER, secret, transport=transport), secret, claims, transport, jwk
|
|
|
|
|
|
def test_exchange_reopens_projected_rotation(lane, tmp_path):
|
|
provider, secret, _, transport, _ = lane
|
|
provider()
|
|
replacement = tmp_path / "replacement"
|
|
replacement.write_text("synthetic-second")
|
|
replacement.replace(secret)
|
|
provider()
|
|
assert [call["decoded_credential"] for call in transport.seen] == ["synthetic-first", "synthetic-second"]
|
|
assert all(call["scope"] == ["approval:create"] and call["grant_type"] == ["client_credentials"]
|
|
for call in transport.seen)
|
|
secret.unlink()
|
|
with pytest.raises(RequesterError, match="^requester exchange refused$"):
|
|
provider()
|
|
assert len(transport.seen) == 2
|
|
|
|
|
|
@pytest.mark.parametrize("field,value", [
|
|
("aud", "other"), ("sub", "other"), ("tenant", "other"),
|
|
("principal_type", "human"), ("roles", []),
|
|
("scope", "approval:create approval:approve"), ("scope", "approval:read"),
|
|
("scope", "approval:consume"), ("iss", "https://wrong.example.test"),
|
|
("exp", 1), ("iat", True),
|
|
])
|
|
def test_refuses_wrong_profile(lane, field, value):
|
|
provider, _, claims, _, _ = lane
|
|
claims[field] = value
|
|
with pytest.raises(RequesterError, match="^requester exchange refused$"):
|
|
provider()
|
|
|
|
|
|
def test_refuses_long_lifetime(lane):
|
|
provider, _, claims, _, _ = lane
|
|
claims["exp"] = claims["iat"] + 901
|
|
with pytest.raises(RequesterError):
|
|
provider()
|
|
|
|
|
|
def test_refuses_signature_mismatch(lane):
|
|
provider, _, _, _, jwk = lane
|
|
other = rsa.generate_private_key(public_exponent=65537, key_size=2048)
|
|
jwk.update(json.loads(jwt.algorithms.RSAAlgorithm.to_jwk(other.public_key())))
|
|
with pytest.raises(RequesterError):
|
|
provider()
|
|
|
|
|
|
@pytest.mark.parametrize("secret", ["", "synthetic\n", "x" * 4097])
|
|
def test_invalid_custody_never_exchanges(lane, secret):
|
|
provider, path, _, transport, _ = lane
|
|
path.write_text(secret)
|
|
with pytest.raises(RequesterError):
|
|
provider()
|
|
assert transport.seen == []
|
|
|
|
|
|
def test_basic_auth_form_encodes_special_characters(lane):
|
|
provider, path, _, transport, _ = lane
|
|
path.write_text("synthetic+%:/=&")
|
|
provider()
|
|
assert transport.seen[-1]["decoded_credential"] == "synthetic+%:/=&"
|