Add rotating create-only requester provider and owner review

Assistant: codex
Assistant-Model: gpt-6.1-sol
Assistant-Session: 01a1046b-5629-7f52-badd-f718eadb6042

tests/test_requester.py

@@ -0,0 +1,90 @@
+import json
+import time
+from urllib.parse import parse_qs
+
+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"
+            self.seen.append(parse_qs(body.decode()))
+            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["client_secret"] 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 == []