@@ -197,6 +197,115 @@ async def test_aauth_without_keyid_param(monkeypatch: pytest.MonkeyPatch) -> Non
197197 assert sig is not None and sig .scheme == "aauth" and sig .sub == "agent-42"
198198
199199
200+ def _issuer_pair (kid : str = "iss-1" , alg : str = "EdDSA" ):
201+ priv = Ed25519PrivateKey .generate ()
202+ jwk = {"kty" : "OKP" , "crv" : "Ed25519" , "kid" : kid , "alg" : alg ,
203+ "x" : b64url (priv .public_key ().public_bytes_raw ())}
204+ return priv , jwk
205+
206+
207+ def _mint (issuer_priv , * , typ : str , alg : str , claims : dict ) -> str :
208+ from regent_httpsig .verify import _register_fully_specified_algs
209+
210+ _register_fully_specified_algs ()
211+ return pyjwt .encode (claims , issuer_priv , algorithm = alg ,
212+ headers = {"typ" : typ , "kid" : "iss-1" })
213+
214+
215+ class TestFullySpecifiedAlgs :
216+ """AAuth -11 / RFC 9864: Ed25519 accepted; polymorphic EdDSA gated by config."""
217+
218+ async def _roundtrip (self , alg : str , config : HttpsigConfig ,
219+ monkeypatch : pytest .MonkeyPatch ):
220+ issuer_priv , issuer_jwk = _issuer_pair (alg = alg )
221+ agent = EgressSigner (seed = generate_seed (), signature_agent = "https://issuer.example" )
222+ now = int (time .time ())
223+ token = _mint (issuer_priv , typ = "aa-agent+jwt" , alg = alg , claims = {
224+ "iss" : "https://issuer.example" , "sub" : "a-1" , "iat" : now , "exp" : now + 600 ,
225+ "dwk" : "aauth-agent.json" ,
226+ "cnf" : {"jwk" : {** agent .public_jwk , "alg" : "Ed25519" }},
227+ })
228+ url = "https://api.example/v1/x"
229+ headers = agent .sign ("POST" , url , {"Host" : "api.example" })
230+ headers ["Signature-Key" ] = f'sig1=jwt;jwt="{ token } "'
231+ verifier = HttpsigVerifier (config )
232+ monkeypatch .setattr (verifier , "_fetch_json" , _mock_fetch ({
233+ "https://issuer.example/.well-known/aauth-agent.json" :
234+ {"jwks_uri" : "https://issuer.example/j" },
235+ "https://issuer.example/j" : {"keys" : [issuer_jwk ]},
236+ }))
237+ return await verifier .verify ("POST" , url , headers )
238+
239+ async def test_ed25519_fully_specified_verifies (self , monkeypatch ) -> None :
240+ sig = await self ._roundtrip ("Ed25519" , HttpsigConfig (), monkeypatch )
241+ assert sig is not None and sig .scheme == "aauth"
242+
243+ async def test_eddsa_accepted_in_transition_mode (self , monkeypatch ) -> None :
244+ sig = await self ._roundtrip ("EdDSA" , HttpsigConfig (), monkeypatch )
245+ assert sig is not None # default: -10 ecosystem still accepted
246+
247+ async def test_eddsa_rejected_in_strict_mode (self , monkeypatch ) -> None :
248+ strict = HttpsigConfig (require_fully_specified_algs = True )
249+ assert await self ._roundtrip ("EdDSA" , strict , monkeypatch ) is None
250+
251+ async def test_ed25519_verifies_in_strict_mode (self , monkeypatch ) -> None :
252+ strict = HttpsigConfig (require_fully_specified_algs = True )
253+ sig = await self ._roundtrip ("Ed25519" , strict , monkeypatch )
254+ assert sig is not None
255+
256+
257+ class TestPersonTokens :
258+ """AAuth -11 person tokens: PS-issued, per-resource aud, cnf-bound, ≤1h."""
259+
260+ def _headers (self , * , aud : str , lifetime : int = 600 ):
261+ ps_priv , ps_jwk = _issuer_pair ()
262+ agent = EgressSigner (seed = generate_seed (), signature_agent = "https://ps.example" )
263+ now = int (time .time ())
264+ token = _mint (ps_priv , typ = "aa-person+jwt" , alg = "Ed25519" , claims = {
265+ "iss" : "https://ps.example" , "sub" : "directed-sub-1" , "aud" : aud ,
266+ "iat" : now , "exp" : now + lifetime , "dwk" : "aauth-person.json" ,
267+ "jti" : "pt-1" , "cnf" : {"jwk" : {** agent .public_jwk , "alg" : "Ed25519" }},
268+ })
269+ url = "https://api.example/v1/x"
270+ headers = agent .sign ("POST" , url , {"Host" : "api.example" })
271+ headers ["Signature-Key" ] = f'sig1=jwt;jwt="{ token } "'
272+ return url , headers , ps_jwk
273+
274+ def _verifier (self , ps_jwk , monkeypatch , ** cfg ):
275+ verifier = HttpsigVerifier (HttpsigConfig (** cfg ))
276+ monkeypatch .setattr (verifier , "_fetch_json" , _mock_fetch ({
277+ "https://ps.example/.well-known/aauth-person.json" :
278+ {"jwks_uri" : "https://ps.example/j" },
279+ "https://ps.example/j" : {"keys" : [ps_jwk ]},
280+ }))
281+ return verifier
282+
283+ async def test_person_token_roundtrip (self , monkeypatch ) -> None :
284+ url , headers , ps_jwk = self ._headers (aud = "https://api.example" )
285+ v = self ._verifier (ps_jwk , monkeypatch , resource_url = "https://api.example" )
286+ sig = await v .verify ("POST" , url , headers )
287+ assert sig is not None
288+ assert sig .scheme == "aauth-person"
289+ assert sig .agent == "https://ps.example" # the PS, not the agent operator
290+ assert sig .sub == "directed-sub-1"
291+ assert sig .claims .get ("jti" ) == "pt-1"
292+
293+ async def test_person_token_wrong_audience_rejected (self , monkeypatch ) -> None :
294+ url , headers , ps_jwk = self ._headers (aud = "https://OTHER.example" )
295+ v = self ._verifier (ps_jwk , monkeypatch , resource_url = "https://api.example" )
296+ assert await v .verify ("POST" , url , headers ) is None
297+
298+ async def test_person_token_disabled_without_resource_url (self , monkeypatch ) -> None :
299+ url , headers , ps_jwk = self ._headers (aud = "https://api.example" )
300+ v = self ._verifier (ps_jwk , monkeypatch ) # no resource_url → path disabled
301+ assert await v .verify ("POST" , url , headers ) is None
302+
303+ async def test_person_token_overlong_lifetime_rejected (self , monkeypatch ) -> None :
304+ url , headers , ps_jwk = self ._headers (aud = "https://api.example" , lifetime = 7200 )
305+ v = self ._verifier (ps_jwk , monkeypatch , resource_url = "https://api.example" )
306+ assert await v .verify ("POST" , url , headers ) is None
307+
308+
200309async def test_cache_is_per_instance () -> None :
201310 a , b = HttpsigVerifier (), HttpsigVerifier ()
202311 a ._cache_put ("https://x.example/doc" , {"keys" : []}, ttl = 60 )
0 commit comments