Setting the file. One moment.
Test X402 Policy · Agents Pay · aws/agent-toolkit-for-aws · Skills Docs
Repo No. 14 · Agents Pay
↖ Back to the coverEnd User Computing Skills
Messaging And Streaming Skills
Migration And Modernization Skills
Networking And Content Delivery Skills
Security And Identity Skills
Web And Mobile Development
132 skills · 818 min
Back to the top of the page 70
Creating Amazon Aurora Db Cluster With Instances
93
Routing Traffic With Route53 And CloudFront
Resilience Program Design
Creating API Gateway Stage
This file
Number 21.9
Position 9 of 14
Type Python
Size 57 KB
Lines 1,430 scripts/ test_x402_policy.py
Python · 1,430 lines · 57 KB
sys
17 import tempfile
18 import unittest
19 from unittest import mock
20 from pathlib import Path
21
22 import x402_policy as pol
23
24 USDC_BASE_SEPOLIA = "0x036CbD53842c5426634e7929541eC2318f3dCF7e"
25 MERCHANT = "0x1111111111111111111111111111111111111111"
26 ATTACKER = "0x2222222222222222222222222222222222222222"
27 ORIGIN = "https://sandbox.node4all.com"
28
29 BASE_POLICY = {
30 "max_per_payment_usd" : "0.50" ,
31 "allowed_networks" : [ "eip155:84532" ],
32 "allowed_assets" : { "eip155:84532" : [ USDC_BASE_SEPOLIA ]},
33 "allowed_recipients" : [ MERCHANT ],
34 "allowed_origins" : [ ORIGIN ],
35 "allowed_schemes" : [ "exact" ],
36 }
37
38
39 def challenge (url: str | None = None , ** overrides) -> dict :
40 """A well-formed x402 v1 challenge for $0.10 to the approved merchant."""
41 accept = {
42 "scheme" : "exact" ,
43 "network" : "eip155:84532" ,
44 "asset" : USDC_BASE_SEPOLIA ,
45 "payTo" : MERCHANT ,
46 "amount" : "100000" , # 0.10 USDC at 6 decimals
47 "extra" : { "nonce" : "abc123" },
48 }
49 accept.update(overrides)
50 resource_url = url or f " { ORIGIN } /v1/x402-test"
51 return { "x402Version" : 1 , "resource" : { "url" : resource_url}, "accepts" : [accept]}
52
53
54 class PolicyFileTests ( unittest . TestCase ):
55 """Local payment configuration requires restrictive file protections."""
56
57 def setUp (self):
58 self .tmp = tempfile.TemporaryDirectory()
59 self .path = Path( self .tmp.name) / "policy.json"
60 self .path.write_text(json.dumps( BASE_POLICY ))
61 self .path.chmod( 0o 600 )
62
63 def tearDown (self):
64 self .tmp.cleanup()
65
66 def test_loads_when_mode_is_600 (self):
67 self .assertEqual(pol.load_config( self .path)[ "max_per_payment_usd" ], "0.50" )
68
69 def test_rejects_group_or_world_readable_policy (self):
70 for bad_mode in ( 0o 640 , 0o 604 , 0o 666 , 0o 660 ):
71 self .path.chmod(bad_mode)
72 with self .assertRaises(pol.PolicyError) as ctx:
73 pol.load_config( self .path)
74 self .assertIn( "expected 0600" , str (ctx.exception))
75
76 def test_rejects_symlink_policy (self):
77 link = Path( self .tmp.name) / "link.json"
78 link.symlink_to( self .path)
79 with self .assertRaises(pol.PolicyError) as ctx:
80 pol.load_config(link)
81 self .assertIn( "symlink" , str (ctx.exception))
82
83 def test_missing_policy_denies_rather_than_defaulting_open (self):
84 with self .assertRaises(pol.PolicyError):
85 pol.load_config(Path( self .tmp.name) / "absent.json" )
86
87 def test_rejects_nonpositive_or_missing_cap (self):
88 for bad in ({}, { "max_per_payment_usd" : "0" }, { "max_per_payment_usd" : "-1" }):
89 self .path.write_text(json.dumps(bad))
90 self .path.chmod( 0o 600 )
91 with self .assertRaises(pol.PolicyError):
92 pol.load_config( self .path)
93
94
95 class ConfigPrecedenceTests ( unittest . TestCase ):
96 """The merged config must not weaken anything, and should strengthen one thing.
97
98 Resource identifiers and the policy live in one file. The session ID is a
99 spending credential, so the file must WIN over the environment: otherwise an
100 agent able to set a variable could point the runtime at a larger-budget session
101 and the 0600 file would be decorative.
102 """
103
104 def setUp (self):
105 self .tmp = tempfile.TemporaryDirectory()
106 self .path = Path( self .tmp.name) / "config.json"
107 self ._saved = {k: os.environ.get(k) for k in pol. RESOURCE_ENV .values()}
108
109 def tearDown (self):
110 for k, v in self ._saved.items():
111 if v is None :
112 os.environ.pop(k, None )
113 else :
114 os.environ[k] = v
115 self .tmp.cleanup()
116
117 def _write (self, resources: dict ) -> dict :
118 self .path.write_text(json.dumps({ "resources" : resources, "policy" : BASE_POLICY }))
119 self .path.chmod( 0o 600 )
120 return pol.load_config( self .path)
121
122 def test_config_file_beats_environment (self):
123 os.environ[ "PAYMENT_SESSION_ID" ] = "ps-ATTACKER-HUGE"
124 cfg = self ._write({ "payment_session_id" : "ps-APPROVED-SMALL" })
125 self .assertEqual(pol.resolve_resource(cfg, "payment_session_id" ), "ps-APPROVED-SMALL" )
126
127 def test_runtime_ignores_environment_selected_policy_file (self):
128 trusted = Path( self .tmp.name) / "trusted.json"
129 hostile = Path( self .tmp.name) / "hostile.json"
130 trusted.write_text(json.dumps({ "resources" : {}, "policy" : BASE_POLICY }))
131 trusted.chmod( 0o 600 )
132 hostile_policy = dict ( BASE_POLICY )
133 hostile_policy[ "max_per_payment_usd" ] = "99"
134 hostile_policy.pop( "allowed_recipients" )
135 hostile_policy[ "allow_any_recipient" ] = True
136 hostile.write_text(json.dumps({ "resources" : {}, "policy" : hostile_policy}))
137 hostile.chmod( 0o 600 )
138
139 with (
140 mock.patch.object(pol, "runtime_config_path" , return_value = trusted),
141 mock.patch.dict(
142 os.environ,
143 {
144 "AGENTS_PAY_CONFIG" : str (hostile),
145 "X402_POLICY_FILE" : str (hostile),
146 },
147 ),
148 ):
149 cfg = pol.load_config()
150
151 self .assertEqual(cfg[ "max_per_payment_usd" ], BASE_POLICY [ "max_per_payment_usd" ])
152 self .assertEqual(cfg[ "allowed_recipients" ], BASE_POLICY [ "allowed_recipients" ])
153 self .assertNotIn( "allow_any_recipient" , cfg)
154
155 def test_runtime_config_path_ignores_home_environment (self):
156 import pwd
157
158 expected = Path(pwd.getpwuid(os.getuid()).pw_dir) / ".agents-pay" / "config.json"
159 with mock.patch.dict(os.environ, { "HOME" : self .tmp.name}):
160 self .assertEqual(pol.runtime_config_path(), expected)
161
162 def test_environment_is_the_fallback_when_the_file_is_silent (self):
163 """Containers and Lambda inject identifiers; that path must still work."""
164 os.environ[ "PAYMENT_SESSION_ID" ] = "ps-FROM-ENV"
165 cfg = self ._write({})
166 self .assertEqual(pol.resolve_resource(cfg, "payment_session_id" ), "ps-FROM-ENV" )
167
168 def test_missing_everywhere_resolves_to_none (self):
169 os.environ.pop( "PAYMENT_SESSION_ID" , None )
170 cfg = self ._write({})
171 self .assertIsNone(pol.resolve_resource(cfg, "payment_session_id" ))
172
173 def test_policy_section_is_read_from_the_nested_shape (self):
174 cfg = self ._write({ "user_id" : "alice" })
175 self .assertEqual(cfg[ "max_per_payment_usd" ], "0.50" )
176 self .assertEqual(pol.resolve_resource(cfg, "user_id" ), "alice" )
177
178 def test_flat_legacy_policy_file_still_loads (self):
179 """An existing policy.json must keep working, not fail open or fail loudly."""
180 self .path.write_text(json.dumps( BASE_POLICY ))
181 self .path.chmod( 0o 600 )
182 cfg = pol.load_config( self .path)
183 self .assertEqual(cfg[ "max_per_payment_usd" ], "0.50" )
184
185 def test_earlier_cap_key_is_still_honoured (self):
186 legacy = dict ( BASE_POLICY )
187 legacy[ "max_amount_usd" ] = legacy.pop( "max_per_payment_usd" )
188 self .path.write_text(json.dumps(legacy))
189 self .path.chmod( 0o 600 )
190 cfg = pol.load_config( self .path)
191 self .assertEqual(pol.per_payment_cap(cfg), "0.50" )
192
193 def test_config_file_permissions_remain_enforced_with_resource_identifiers (self):
194 """The config remains protected after resource identifiers are added."""
195 self ._write({ "payment_session_id" : "ps-1" })
196 for bad in ( 0o 640 , 0o 666 ):
197 self .path.chmod(bad)
198 with self .assertRaises(pol.PolicyError):
199 pol.load_config( self .path)
200
201 def test_missing_per_payment_cap_refuses_rather_than_paying_unbounded (self):
202 no_cap = {k: v for k, v in BASE_POLICY .items() if k != "max_per_payment_usd" }
203 with self .assertRaises(pol.PolicyError) as ctx:
204 pol.select_accept_entry(challenge(), no_cap)
205 self .assertIn( "per-payment ceiling" , str (ctx.exception))
206
207
208 class RecipientAndValueTests ( unittest . TestCase ):
209 """Untrusted payment challenges cannot control recipient or value."""
210
211 def test_amount_above_ceiling_is_refused (self):
212 # 5.00 USDC against a 0.50 cap.
213 with self .assertRaises(pol.PolicyError):
214 pol.select_accept_entry(challenge( amount = "5000000" ), BASE_POLICY )
215
216 def test_amount_at_ceiling_is_allowed (self):
217 entry = pol.select_accept_entry(challenge( amount = "500000" ), BASE_POLICY )
218 self .assertEqual(entry[ "payTo" ], MERCHANT )
219
220 def test_zero_and_negative_amounts_are_refused (self):
221 for bad in ( "0" , "-100000" ):
222 with self .assertRaises(pol.PolicyError):
223 pol.select_accept_entry(challenge( amount = bad), BASE_POLICY )
224
225 def test_unapproved_network_is_refused (self):
226 with self .assertRaises(pol.PolicyError):
227 pol.select_accept_entry(challenge( network = "eip155:1" ), BASE_POLICY )
228
229 def test_wrong_asset_contract_is_refused (self):
230 with self .assertRaises(pol.PolicyError):
231 pol.select_accept_entry(challenge( asset = ATTACKER ), BASE_POLICY )
232
233 def test_unapproved_scheme_is_refused (self):
234 with self .assertRaises(pol.PolicyError):
235 pol.select_accept_entry(challenge( scheme = "upto" ), BASE_POLICY )
236
237 def test_explicitly_empty_scheme_list_refuses_every_payment (self):
238 policy = dict ( BASE_POLICY )
239 policy[ "allowed_schemes" ] = []
240 with self .assertRaises(pol.PolicyError) as ctx:
241 pol.select_accept_entry(challenge(), policy)
242 self .assertIn( "allows no schemes" , str (ctx.exception))
243
244 def test_absent_scheme_list_uses_the_exact_scheme_default (self):
245 policy = dict ( BASE_POLICY )
246 policy.pop( "allowed_schemes" )
247 self .assertEqual(pol.select_accept_entry(challenge(), policy)[ "scheme" ], "exact" )
248
249 def test_null_amount_uses_max_amount_required (self):
250 entry = pol.select_accept_entry(
251 challenge( amount = None , maxAmountRequired = "100000" ),
252 BASE_POLICY ,
253 )
254 self .assertEqual(entry[ "maxAmountRequired" ], "100000" )
255
256 def test_conflicting_amount_fields_are_refused (self):
257 for amount, maximum in (( "1" , "50000000" ), ( "50000000" , "1" )):
258 with self .subTest( amount = amount, maxAmountRequired = maximum):
259 with self .assertRaises(pol.PolicyError):
260 pol.select_accept_entry(
261 challenge( amount = amount, maxAmountRequired = maximum),
262 BASE_POLICY ,
263 )
264
265 def test_equal_amount_fields_are_allowed (self):
266 entry = pol.select_accept_entry(
267 challenge( amount = "100000" , maxAmountRequired = "100000" ),
268 BASE_POLICY ,
269 )
270 self .assertEqual(entry[ "amount" ], entry[ "maxAmountRequired" ])
271
272 def test_checksum_capitalization_still_matches (self):
273 entry = pol.select_accept_entry(challenge( asset = USDC_BASE_SEPOLIA .lower()), BASE_POLICY )
274 self .assertIsNotNone(entry)
275
276 def test_does_not_blindly_take_first_accepts_entry (self):
277 """A compliant later entry must win over a hostile first entry."""
278 ch = challenge()
279 hostile = dict (ch[ "accepts" ][ 0 ])
280 hostile[ "amount" ] = "5000000" # $5.00, over the $0.50 ceiling
281 ch[ "accepts" ] = [hostile, ch[ "accepts" ][ 0 ]]
282 self .assertEqual(pol.select_accept_entry(ch, BASE_POLICY )[ "amount" ], "100000" )
283
284 def test_missing_required_fields_are_refused (self):
285 for field in ( "scheme" , "network" , "asset" , "payTo" ):
286 ch = challenge()
287 del ch[ "accepts" ][ 0 ][field]
288 with self .assertRaises(pol.PolicyError):
289 pol.select_accept_entry(ch, BASE_POLICY )
290
291 def test_refusal_message_does_not_echo_challenge_values (self):
292 """A uniform refusal stops a publisher probing the policy field by field."""
293 with self .assertRaises(pol.PolicyError) as ctx:
294 pol.select_accept_entry(challenge( asset = ATTACKER , amount = "5000000" ), BASE_POLICY )
295 message = str (ctx.exception)
296 self .assertNotIn( ATTACKER , message)
297 self .assertNotIn( "5000000" , message)
298
299
300 class SingleTenantUserIdTests ( unittest . TestCase ):
301 """The payer identity is generated once at init and read from the config after.
302
303 One installation = one payer. Requiring --user-id at every step invited a
304 mismatch between create-instrument and new-session, which yields a session that
305 cannot spend the instrument — a confusing failure with no clear error.
306 """
307
308 @ classmethod
309 def setUpClass (cls):
310 import importlib.util
311
312 spec = importlib.util.spec_from_file_location(
313 "agents_pay_admin" , Path( __file__ ).resolve().parent / "agents_pay_admin.py"
314 )
315 cls .admin = importlib.util.module_from_spec(spec)
316 spec.loader.exec_module( cls .admin)
317
318 def setUp (self):
319 self .tmp = tempfile.TemporaryDirectory()
320 self .path = Path( self .tmp.name) / "config.json"
321 self ._saved = os.environ.pop( "PAYMENT_USER_ID" , None )
322
323 def tearDown (self):
324 if self ._saved is not None :
325 os.environ[ "PAYMENT_USER_ID" ] = self ._saved
326 self .tmp.cleanup()
327
328 def test_generated_id_is_opaque (self):
329 """It reaches the payments API and the wallet, so it must not leak host or login."""
330 generated = self .admin.generate_user_id()
331 self .assertTrue(generated.startswith( "agents-pay-" ))
332 self .assertGreater( len (generated), len ( "agents-pay-" ) + 8 )
333 for leak in (os.environ.get( "USER" , " \0 " ), os.uname().nodename):
334 self .assertNotIn(leak, generated)
335
336 def test_generated_ids_differ_between_installations (self):
337 self .assertNotEqual( self .admin.generate_user_id(), self .admin.generate_user_id())
338
339 def test_resolves_from_the_config (self):
340 self .path.write_text(json.dumps({ "resources" : { "user_id" : "agents-pay-abc" }, "policy" : {}}))
341 self .assertEqual( self .admin.resolve_user_id( None , self .path), "agents-pay-abc" )
342
343 def test_explicit_flag_wins (self):
344 self .path.write_text(json.dumps({ "resources" : { "user_id" : "agents-pay-abc" }, "policy" : {}}))
345 self .assertEqual( self .admin.resolve_user_id( "override" , self .path), "override" )
346
347 def test_environment_is_the_last_resort (self):
348 os.environ[ "PAYMENT_USER_ID" ] = "from-env"
349 self .assertEqual( self .admin.resolve_user_id( None , self .path), "from-env" )
350
351 def test_absent_everywhere_returns_none (self):
352 self .assertIsNone( self .admin.resolve_user_id( None , self .path))
353
354 def test_admin_calls_are_attributed_to_the_skill (self):
355 import inspect
356
357 self .assertEqual( self .admin. AGENT_NAME , "aws-agents-pay" )
358 for command in ( self .admin.cmd_create_instrument, self .admin.cmd_new_session):
359 self .assertIn( "agent_name=AGENT_NAME" , inspect.getsource(command))
360
361 def test_runtime_calls_are_attributed_to_the_skill (self):
362 import inspect
363 import x402_fetch
364
365 self .assertEqual(x402_fetch. AGENT_NAME , "openclaw-aws-agents-pay" )
366 source = inspect.getsource(x402_fetch)
367 self .assertEqual(source.count( "agent_name=AGENT_NAME" ), 3 )
368
369 def test_admin_config_path_preserves_operator_override (self):
370 env_path = str ( self .path.with_name( "from-env.json" ))
371 explicit_path = str ( self .path.with_name( "from-flag.json" ))
372 with mock.patch.dict(os.environ, { "AGENTS_PAY_CONFIG" : env_path}):
373 self .assertEqual( self .admin.admin_config_path( None ), Path(env_path))
374 self .assertEqual(
375 self .admin.admin_config_path(explicit_path),
376 Path(explicit_path),
377 )
378
379
380 class RecipientValidationTests ( unittest . TestCase ):
381 """Recipient mode is explicit, exclusive, and fail-closed by default."""
382
383 def test_unknown_recipient_is_refused (self):
384 with self .assertRaises(pol.PolicyError):
385 pol.select_accept_entry(challenge( payTo = ATTACKER ), BASE_POLICY )
386
387 def test_known_recipient_is_accepted (self):
388 entry = pol.select_accept_entry(challenge( payTo = MERCHANT ), BASE_POLICY )
389 self .assertEqual(entry[ "payTo" ], MERCHANT )
390
391 def test_missing_recipient_allowlist_denies_by_default (self):
392 policy = dict ( BASE_POLICY )
393 policy.pop( "allowed_recipients" )
394 with self .assertRaises(pol.PolicyError) as ctx:
395 pol.select_accept_entry(challenge( payTo = MERCHANT ), policy)
396 self .assertIn( "allows no recipients" , str (ctx.exception))
397
398 def test_recipient_match_is_case_insensitive (self):
399 policy = dict ( BASE_POLICY )
400 policy[ "allowed_recipients" ] = [ MERCHANT .upper()]
401 entry = pol.select_accept_entry(challenge( payTo = MERCHANT .lower()), policy)
402 self .assertEqual(entry[ "payTo" ], MERCHANT .lower())
403
404 def test_allow_any_recipient_accepts_an_unlisted_payee (self):
405 policy = dict ( BASE_POLICY )
406 policy.pop( "allowed_recipients" )
407 policy[ "allow_any_recipient" ] = True
408 entry = pol.select_accept_entry(challenge( payTo = ATTACKER ), policy)
409 self .assertEqual(entry[ "payTo" ], ATTACKER )
410
411 def test_recipient_modes_are_mutually_exclusive (self):
412 for value in ( True , False ):
413 with self .subTest( allow_any_recipient = value):
414 policy = dict ( BASE_POLICY )
415 policy[ "allow_any_recipient" ] = value
416 with self .assertRaises(pol.PolicyError) as ctx:
417 pol.select_accept_entry(challenge(), policy)
418 self .assertIn( "mutually exclusive" , str (ctx.exception))
419
420 def test_allow_any_recipient_requires_a_boolean (self):
421 policy = dict ( BASE_POLICY )
422 policy.pop( "allowed_recipients" )
423 policy[ "allow_any_recipient" ] = "true"
424 with self .assertRaises(pol.PolicyError) as ctx:
425 pol.select_accept_entry(challenge(), policy)
426 self .assertIn( "must be a boolean" , str (ctx.exception))
427
428 def test_allow_any_recipient_keeps_other_policy_checks (self):
429 policy = dict ( BASE_POLICY )
430 policy.pop( "allowed_recipients" )
431 policy[ "allow_any_recipient" ] = True
432 for kwargs in (
433 { "network" : "eip155:1" },
434 { "asset" : ATTACKER },
435 { "scheme" : "upto" },
436 { "amount" : "500001" },
437 ):
438 with self .subTest( ** kwargs):
439 with self .assertRaises(pol.PolicyError):
440 pol.select_accept_entry(challenge( payTo = ATTACKER , ** kwargs), policy)
441
442 def test_network_asset_and_scheme_are_still_validated_for_known_recipient (self):
443 for kwargs in ({ "network" : "eip155:1" }, { "asset" : ATTACKER }, { "scheme" : "upto" }):
444 with self .subTest( ** kwargs):
445 with self .assertRaises(pol.PolicyError):
446 pol.select_accept_entry(challenge( payTo = MERCHANT , ** kwargs), BASE_POLICY )
447
448
449 class AdminRecipientModeTests ( unittest . TestCase ):
450 """The human-facing CLI writes exactly one recipient authorization mode."""
451
452 @ classmethod
453 def setUpClass (cls):
454 import importlib.util
455
456 spec = importlib.util.spec_from_file_location(
457 "agents_pay_admin_recipient_tests" ,
458 Path( __file__ ).resolve().parent / "agents_pay_admin.py" ,
459 )
460 cls .admin = importlib.util.module_from_spec(spec)
461 spec.loader.exec_module( cls .admin)
462
463 def setUp (self):
464 self .tmp = tempfile.TemporaryDirectory()
465 self .path = Path( self .tmp.name) / "config.json"
466
467 def tearDown (self):
468 self .tmp.cleanup()
469
470 def _run (self, * arguments):
471 from contextlib import redirect_stderr, redirect_stdout
472 from io import StringIO
473
474 argv = [
475 "agents_pay_admin.py" ,
476 "init-config" ,
477 "--path" ,
478 str ( self .path),
479 * arguments,
480 ]
481 with (
482 mock.patch.object(sys, "argv" , argv),
483 redirect_stdout(StringIO()),
484 redirect_stderr(StringIO()),
485 ):
486 return self .admin.main()
487
488 def test_recipient_flag_writes_an_allowlist (self):
489 self .assertEqual( self ._run( "--recipient" , MERCHANT ), 0 )
490 policy = json.loads( self .path.read_text())[ "policy" ]
491 self .assertEqual(policy[ "allowed_recipients" ], [ MERCHANT ])
492 self .assertNotIn( "allow_any_recipient" , policy)
493
494 def test_allow_any_recipient_writes_explicit_mode (self):
495 self .assertEqual( self ._run( "--allow-any-recipient" ), 0 )
496 policy = json.loads( self .path.read_text())[ "policy" ]
497 self .assertIs(policy[ "allow_any_recipient" ], True )
498 self .assertNotIn( "allowed_recipients" , policy)
499
500 def test_recipient_modes_cannot_be_combined (self):
501 with self .assertRaises( SystemExit ) as ctx:
502 self ._run( "--recipient" , MERCHANT , "--allow-any-recipient" )
503 self .assertEqual(ctx.exception.code, 2 )
504 self .assertFalse( self .path.exists())
505
506 def test_one_recipient_mode_is_required (self):
507 with self .assertRaises( SystemExit ) as ctx:
508 self ._run()
509 self .assertEqual(ctx.exception.code, 2 )
510 self .assertFalse( self .path.exists())
511
512
513 class OptionalOriginTests ( unittest . TestCase ):
514 """Origin allowlisting is optional; baseline URL protections remain required."""
515
516 def _policy (self, origins = None ):
517 p = dict ( BASE_POLICY )
518 if origins is None :
519 p.pop( "allowed_origins" , None )
520 else :
521 p[ "allowed_origins" ] = origins
522 return p
523
524 def test_no_origin_list_allows_any_public_https_site (self):
525 decision = pol.authorize_payment( "https://example.com/paid" , challenge( url = "https://example.com/paid" ), self ._policy())
526 self .assertEqual(decision[ "origin" ], "https://example.com" )
527
528 def test_an_origin_list_still_pins_when_provided (self):
529 with self .assertRaises(pol.PolicyError) as ctx:
530 pol.authorize_payment( "https://example.com/paid" , challenge( url = "https://example.com/paid" ), self ._policy([ ORIGIN ]))
531 self .assertIn( "allowed_origins" , str (ctx.exception))
532
533 def test_mandatory_ssrf_controls_apply_even_with_no_origin_list (self):
534 """Dropping the preference must not drop the requirements."""
535 for url in ( "http://example.com/x" , "https://user:pw@example.com/x" ):
536 with self .subTest( url = url):
537 with self .assertRaises(pol.PolicyError):
538 pol.authorize_payment(url, challenge(), self ._policy())
539 for addr in ( "169.254.169.254" , "127.0.0.1" , "100.64.0.1" ):
540 with self .subTest( addr = addr):
541 with self .assertRaises(pol.PolicyError):
542 pol.assert_public_ip(addr)
543
544
545 class SsrfTests ( unittest . TestCase ):
546 """Arbitrary URL fetching must not enable server-side request forgery."""
547
548 def test_non_https_schemes_are_refused (self):
549 for url in ( "http://example.com/x" , "file:///etc/passwd" , "gopher://h/1" ):
550 with self .assertRaises(pol.PolicyError):
551 pol.assert_public_https_url(url)
552
553 def test_embedded_credentials_are_refused (self):
554 with self .assertRaises(pol.PolicyError):
555 pol.assert_public_https_url( "https://user:pw@example.com/x" )
556
557 def test_internal_addresses_are_refused (self):
558 internal = [
559 "127.0.0.1" , # loopback
560 "10.0.0.1" , # RFC1918
561 "192.168.1.1" , # RFC1918
562 "172.16.0.1" , # RFC1918
563 "169.254.169.254" , # cloud metadata
564 "100.64.0.1" , # CGNAT — not caught by is_private
565 "224.0.0.1" , # multicast — not caught by is_private
566 "0.0.0.0" , # unspecified
567 "::1" , # v6 loopback
568 "fd00::1" , # v6 unique-local
569 "fe80::1" , # v6 link-local
570 "::ffff:127.0.0.1" , # v4-mapped loopback
571 ]
572 for addr in internal:
573 with self .subTest( addr = addr):
574 with self .assertRaises(pol.PolicyError):
575 pol.assert_public_ip(addr)
576
577 def test_public_addresses_are_allowed (self):
578 for addr in ( "93.184.216.34" , "1.1.1.1" , "2606:4700:4700::1111" ):
579 with self .subTest( addr = addr):
580 pol.assert_public_ip(addr) # must not raise
581
582 def test_origin_must_be_allowlisted (self):
583 """A resolvable but unapproved origin is refused by the allowlist.
584
585 Uses example.com because it actually resolves — otherwise the DNS check
586 fires first and we would not be testing the allowlist at all.
587 """
588 with self .assertRaises(pol.PolicyError) as ctx:
589 pol.authorize_payment( "https://example.com/x" , challenge(), BASE_POLICY )
590 self .assertIn( "allowed_origins" , str (ctx.exception))
591
592
593 class IdempotencyTests ( unittest . TestCase ):
594 """Payment retries use stable idempotency keys."""
595
596 def setUp (self):
597 self ._saved = os.environ.get( "PAYMENT_SESSION_ID" )
598 os.environ[ "PAYMENT_SESSION_ID" ] = "sess-1"
599
600 def tearDown (self):
601 if self ._saved is None :
602 os.environ.pop( "PAYMENT_SESSION_ID" , None )
603 else :
604 os.environ[ "PAYMENT_SESSION_ID" ] = self ._saved
605
606 def _token (self, url = f " { ORIGIN } /v1/x402-test" , ** overrides):
607 ch = challenge( ** overrides)
608 return pol.derive_client_token(url, ch[ "accepts" ][ 0 ], ch)
609
610 def test_same_purchase_yields_same_token (self):
611 """This is what makes a retry replay one authorization instead of two."""
612 self .assertEqual( self ._token(), self ._token())
613
614 def test_token_survives_a_rotating_publisher_nonce (self):
615 """The retry path re-fetches the 402, so the nonce may be fresh each time.
616
617 Mixing the nonce into the token would give every attempt a different
618 token — converting the retry this function protects into a second real
619 payment, and letting a hostile publisher force double charges by
620 rotating nonces. Two separately fetched challenges for the same purchase
621 must derive the SAME token.
622 """
623 first = self ._token( extra = { "nonce" : "server-nonce-1" })
624 second = self ._token( extra = { "nonce" : "server-nonce-2" })
625 self .assertEqual(first, second)
626
627 def test_token_ignores_an_absent_nonce (self):
628 ch = challenge()
629 del ch[ "accepts" ][ 0 ][ "extra" ]
630 without = pol.derive_client_token( f " { ORIGIN } /v1/x402-test" , ch[ "accepts" ][ 0 ], ch)
631 self .assertEqual(without, self ._token())
632
633 def test_explicit_purchase_id_distinguishes_deliberate_repeat_buys (self):
634 ch = challenge()
635 base = pol.derive_client_token( f " { ORIGIN } /v1/x402-test" , ch[ "accepts" ][ 0 ], ch)
636 first = pol.derive_client_token( f " { ORIGIN } /v1/x402-test" , ch[ "accepts" ][ 0 ], ch, "order-1" )
637 second = pol.derive_client_token( f " { ORIGIN } /v1/x402-test" , ch[ "accepts" ][ 0 ], ch, "order-2" )
638 self .assertNotEqual(first, second)
639 self .assertNotEqual(base, first)
640
641 def test_token_is_stable_across_process_restart (self):
642 """Derived, not random — so a restart mid-purchase cannot double-charge."""
643 expected = self ._token()
644 for _ in range ( 5 ):
645 self .assertEqual( self ._token(), expected)
646
647 def test_different_amount_yields_different_token (self):
648 self .assertNotEqual( self ._token(), self ._token( amount = "200000" ))
649
650 def test_conflicting_amount_aliases_cannot_change_the_signed_value_or_token (self):
651 ch = challenge( amount = "1" , maxAmountRequired = "50000000" )
652 with self .assertRaises(pol.PolicyError):
653 pol.derive_client_token(
654 f " { ORIGIN } /v1/x402-test" ,
655 ch[ "accepts" ][ 0 ],
656 ch,
657 )
658
659 def test_different_recipient_yields_different_token (self):
660 self .assertNotEqual( self ._token(), self ._token( payTo = ATTACKER ))
661
662 def test_different_resource_yields_different_token (self):
663 self .assertNotEqual( self ._token(), self ._token( url = f " { ORIGIN } /v1/other" ))
664
665 def test_different_session_yields_different_token (self):
666 first = self ._token()
667 os.environ[ "PAYMENT_SESSION_ID" ] = "sess-2"
668 self .assertNotEqual(first, self ._token())
669
670 def test_authorization_token_uses_config_session_over_environment (self):
671 """The protected config session is the spend boundary, so it keys retries."""
672 policy = dict ( BASE_POLICY )
673 policy[ "_resources" ] = { "payment_session_id" : "sess-approved-small" }
674 os.environ[ "PAYMENT_SESSION_ID" ] = "sess-attacker-huge"
675 ch = challenge()
676
677 decision = pol.authorize_payment( f " { ORIGIN } /v1/x402-test" , ch, policy)
678 expected = pol.derive_client_token(
679 f " { ORIGIN } /v1/x402-test" ,
680 ch[ "accepts" ][ 0 ],
681 ch,
682 session_id = "sess-approved-small" ,
683 )
684 env_based = pol.derive_client_token(
685 f " { ORIGIN } /v1/x402-test" ,
686 ch[ "accepts" ][ 0 ],
687 ch,
688 session_id = "sess-attacker-huge" ,
689 )
690
691 self .assertEqual(decision[ "client_token" ], expected)
692 self .assertNotEqual(decision[ "client_token" ], env_based)
693
694 def test_derive_client_token_with_policy_param_prefers_config_over_env (self):
695 """A caller that resolves session_id itself via `policy=` must not
696 invert resolve_resource()'s documented config-file-first precedence.
697
698 Before this fix, omitting `session_id` fell through to a bare
699 `os.environ.get("PAYMENT_SESSION_ID", "")` unconditionally — bypassing
700 config.json entirely for any caller using this path. Passing `policy=`
701 now correctly prefers the config file, matching resolve_resource().
702 """
703 policy = dict ( BASE_POLICY )
704 policy[ "_resources" ] = { "payment_session_id" : "sess-config-value" }
705 saved = os.environ.get( "PAYMENT_SESSION_ID" )
706 os.environ[ "PAYMENT_SESSION_ID" ] = "sess-env-value"
707 try :
708 ch = challenge()
709 via_policy = pol.derive_client_token(
710 f " { ORIGIN } /v1/x402-test" , ch[ "accepts" ][ 0 ], ch, policy = policy
711 )
712 via_explicit_session = pol.derive_client_token(
713 f " { ORIGIN } /v1/x402-test" ,
714 ch[ "accepts" ][ 0 ],
715 ch,
716 session_id = "sess-config-value" ,
717 )
718 self .assertEqual(
719 via_policy,
720 via_explicit_session,
721 "policy= must resolve the session via resolve_resource() (config-first),"
722 " matching what an explicit config-derived session_id would produce" ,
723 )
724
725 # Explicit session_id still wins over policy= when both are given.
726 via_both = pol.derive_client_token(
727 f " { ORIGIN } /v1/x402-test" ,
728 ch[ "accepts" ][ 0 ],
729 ch,
730 session_id = "sess-explicit-wins" ,
731 policy = policy,
732 )
733 self .assertNotEqual(via_both, via_policy)
734 finally :
735 if saved is None :
736 os.environ.pop( "PAYMENT_SESSION_ID" , None )
737 else :
738 os.environ[ "PAYMENT_SESSION_ID" ] = saved
739
740 def test_derive_client_token_without_session_id_or_policy_still_uses_env (self):
741 """Backward compatibility: pre-existing callers that pass neither
742 `session_id` nor `policy` keep the original bare env-var behavior
743 (the path this file's own `_token()` helper and several tests above
744 rely on).
745 """
746 saved = os.environ.get( "PAYMENT_SESSION_ID" )
747 os.environ[ "PAYMENT_SESSION_ID" ] = "sess-bare-env"
748 try :
749 ch = challenge()
750 token = pol.derive_client_token( f " { ORIGIN } /v1/x402-test" , ch[ "accepts" ][ 0 ], ch)
751 expected = pol.derive_client_token(
752 f " { ORIGIN } /v1/x402-test" ,
753 ch[ "accepts" ][ 0 ],
754 ch,
755 session_id = "sess-bare-env" ,
756 )
757 self .assertEqual(token, expected)
758 finally :
759 if saved is None :
760 os.environ.pop( "PAYMENT_SESSION_ID" , None )
761 else :
762 os.environ[ "PAYMENT_SESSION_ID" ] = saved
763
764
765 class PolicyDirectoryTests ( unittest . TestCase ):
766 """The directory matters as much as the file.
767
768 Write access to the containing directory lets another principal rename a
769 wider policy into place, which checks on the original inode cannot detect.
770 """
771
772 def setUp (self):
773 self .tmp = tempfile.TemporaryDirectory()
774 self .dir = Path( self .tmp.name) / "cfg"
775 self .dir.mkdir( mode =0o 700 )
776 self .path = self .dir / "policy.json"
777 self .path.write_text(json.dumps( BASE_POLICY ))
778 self .path.chmod( 0o 600 )
779
780 def tearDown (self):
781 self .dir.chmod( 0o 700 ) # so cleanup can remove it
782 self .tmp.cleanup()
783
784 def test_loads_from_a_0700_directory (self):
785 self .assertTrue(pol.load_config( self .path))
786
787 def test_refuses_a_group_or_world_writable_directory (self):
788 for bad_mode in ( 0o 777 , 0o 722 , 0o 770 ):
789 self .dir.chmod(bad_mode)
790 with self .subTest( mode = oct (bad_mode)):
791 with self .assertRaises(pol.PolicyError) as ctx:
792 pol.load_config( self .path)
793 self .assertIn( "writable" , str (ctx.exception))
794
795
796 class SignerInputTests ( unittest . TestCase ):
797 """The signer receives only the policy-approved challenge.
798
799 Validating a challenge and then forwarding the publisher's raw response to
800 the signer would mean approving one document and signing another. These
801 assert on what the signer receives, not only on what the gate returns.
802 """
803
804 @ classmethod
805 def setUpClass (cls):
806 try :
807 import x402_fetch # noqa: F401
808 except ImportError as e: # pragma: no cover
809 raise unittest.SkipTest( f "x402_fetch unavailable: { e } " )
810
811 def test_signer_receives_only_the_vetted_entry (self):
812 """A hostile accepts[0] must never reach the signer."""
813 hostile = {
814 "scheme" : "exact" , "network" : "eip155:1" , "asset" : ATTACKER ,
815 "payTo" : ATTACKER , "amount" : "50000000" ,
816 }
817 compliant = {
818 "scheme" : "exact" , "network" : "eip155:84532" , "asset" : USDC_BASE_SEPOLIA ,
819 "payTo" : MERCHANT , "amount" : "100000" ,
820 }
821 ch = { "x402Version" : 2 , "accepts" : [hostile, compliant]}
822
823 decision = pol.authorize_payment( f " { ORIGIN } /v1/x402-test" , ch, BASE_POLICY )
824 # This mirrors exactly what x402_fetch hands to generate_payment_header.
825 forwarded = json.loads(
826 json.dumps({ "x402Version" : decision[ "x402_version" ], "accepts" : [decision[ "accept" ]]})
827 )
828
829 self .assertEqual( len (forwarded[ "accepts" ]), 1 , "signer must see exactly one option" )
830 self .assertEqual(forwarded[ "accepts" ][ 0 ], compliant)
831 blob = json.dumps(forwarded)
832 self .assertNotIn( ATTACKER , blob)
833 self .assertNotIn( "eip155:1 \" " , blob)
834 self .assertNotIn( "50000000" , blob)
835
836 def test_signer_receives_only_the_canonical_validated_amount (self):
837 for version, expected_key, removed_key in (
838 ( 1 , "maxAmountRequired" , "amount" ),
839 ( 2 , "amount" , "maxAmountRequired" ),
840 ):
841 with self .subTest( version = version):
842 ch = challenge( amount = "100000" , maxAmountRequired = "100000" )
843 ch[ "x402Version" ] = version
844 decision = pol.authorize_payment(
845 f " { ORIGIN } /v1/x402-test" ,
846 ch,
847 BASE_POLICY ,
848 )
849 accepted = decision[ "accept" ]
850 self .assertEqual(accepted[expected_key], "100000" )
851 self .assertNotIn(removed_key, accepted)
852 self .assertEqual(decision[ "amount_base_units" ], "100000" )
853
854 def test_runtime_hands_the_signer_only_the_canonical_amount (self):
855 import types
856
857 import httpx
858 import x402_fetch as fetch
859
860 url = f " { ORIGIN } /v1/x402-test"
861 ch = challenge( amount = "100000" , maxAmountRequired = "100000" )
862 ch[ "x402Version" ] = 2
863 policy = {
864 ** BASE_POLICY ,
865 "_resources" : {
866 "payment_manager_arn" : "arn:aws:bedrock-agentcore:us-west-2:111111111111:payment-manager/test" ,
867 "payment_instrument_id" : "instrument-1" ,
868 "payment_session_id" : "session-1" ,
869 "user_id" : "user-1" ,
870 },
871 }
872 captured: list[ dict ] = []
873
874 class PaymentManager :
875 def __init__ (self, ** _kwargs):
876 pass
877
878 def generate_payment_header (self, ** kwargs):
879 captured.append(json.loads(kwargs[ "payment_required_request" ][ "body" ]))
880 return { "PAYMENT-SIGNATURE" : "proof" }
881
882 payments_module = types.ModuleType( "bedrock_agentcore.payments" )
883 payments_module.PaymentManager = PaymentManager
884 agentcore_module = types.ModuleType( "bedrock_agentcore" )
885 agentcore_module.payments = payments_module
886 challenge_response = httpx.Response(
887 402 ,
888 json = ch,
889 request = httpx.Request( "GET" , url),
890 )
891 challenge_response.read_body = challenge_response.content
892 paid_response = httpx.Response(
893 200 ,
894 content = b "paid" ,
895 headers = { "content-type" : "text/plain" },
896 request = httpx.Request( "GET" , url),
897 )
898 paid_response.read_body = paid_response.content
899 responses = [challenge_response, paid_response]
900
901 with (
902 mock.patch.object(fetch.pol, "load_config" , return_value = policy),
903 mock.patch.object(fetch.pol, "assert_public_https_url" , return_value = ORIGIN ),
904 mock.patch.object(fetch, "_get" , side_effect = responses),
905 mock.patch.dict(
906 sys.modules,
907 {
908 "bedrock_agentcore" : agentcore_module,
909 "bedrock_agentcore.payments" : payments_module,
910 },
911 ),
912 ):
913 result = json.loads(fetch.x402_fetch(url))
914
915 self .assertTrue(result[ "paid" ], result)
916 self .assertEqual( len (captured), 1 )
917 accepted = captured[ 0 ][ "accepts" ][ 0 ]
918 self .assertEqual(accepted[ "amount" ], "100000" )
919 self .assertNotIn( "maxAmountRequired" , accepted)
920
921 def test_fetch_does_not_forward_publisher_headers_or_body (self):
922 """The raw 402 must not be passed through to the signer."""
923 import inspect
924
925 import x402_fetch as fetch
926
927 src = inspect.getsource(fetch.x402_fetch)
928 self .assertIn( "vetted_challenge" , src)
929 self .assertIn( '"body": vetted_challenge' , src)
930 # The publisher's own headers/body must not reach the signer call.
931 self .assertNotIn( "dict(response.headers)" , src)
932 self .assertNotIn( '"body": _body_text(response)' , src)
933
934 def test_compressed_responses_are_refused (self):
935 """A decompression bomb must not be able to blow past the size cap."""
936 import inspect
937
938 import x402_fetch as fetch
939
940 src = inspect.getsource(fetch._get)
941 self .assertIn( '"Accept-Encoding": "identity"' , src)
942 # The cap must be checked BEFORE the buffer grows.
943 self .assertIn( "if len(body) + len(chunk) > MAX_BODY_BYTES" , src)
944
945 def test_body_limit_env_var_cannot_disable_the_cap (self):
946 import x402_fetch as fetch
947
948 self .assertGreaterEqual(fetch. MAX_BODY_BYTES , 1024 )
949 for bad in ( "0" , "-1" , "abc" , "" ):
950 with self .subTest( value = bad):
951 with mock.patch.dict(os.environ, { "X402_BODY_LIMIT_BYTES" : bad}):
952 self .assertEqual(
953 fetch._bounded_env( "X402_BODY_LIMIT_BYTES" , 256.0 , 1.0 , 1000.0 ),
954 256.0 ,
955 )
956
957
958 class TransientSettlementTests ( unittest . TestCase ):
959 """Base Sepolia settlement can be intermittently transient.
960
961 The proof is valid but the paid retry still returns 402. Without a retry that
962 surfaces as a failed fetch for a payment the user already made. Safe only because
963 the same derived client_token is replayed, so ProcessPayment stays idempotent.
964 """
965
966 @ classmethod
967 def setUpClass (cls):
968 try :
969 import x402_fetch # noqa: F401
970 except ImportError as e: # pragma: no cover
971 raise unittest.SkipTest( f "x402_fetch unavailable: { e } " )
972
973 def test_retry_cap_is_bounded (self):
974 """A bad env value must not produce an unbounded payment loop."""
975 import x402_fetch as fetch
976
977 self .assertGreaterEqual(fetch. MAX_PAYMENT_ATTEMPTS , 1 )
978 self .assertLessEqual(fetch. MAX_PAYMENT_ATTEMPTS , 10 )
979 for bad in ( "0" , "-1" , "abc" , "99999" ):
980 with self .subTest( value = bad):
981 with mock.patch.dict(os.environ, { "X402_MAX_PAYMENT_ATTEMPTS" : bad}):
982 self .assertEqual(
983 fetch._bounded_env( "X402_MAX_PAYMENT_ATTEMPTS" , 5 , 1 , 10 ),
984 5 ,
985 )
986
987 def test_the_same_token_is_reused_across_attempts (self):
988 """This is what makes the retry safe rather than a double-charge."""
989 import inspect
990
991 import x402_fetch as fetch
992
993 src = inspect.getsource(fetch.x402_fetch)
994 # The decision (which derives the token) is made once, BEFORE the attempt
995 # loop, so every attempt replays the same authorization.
996 self .assertLess(src.index( "pol.authorize_payment" ), src.index( "for attempt in range" ))
997 self .assertEqual(src.count( "pol.authorize_payment" ), 1 )
998
999 def test_exhausted_retries_report_no_double_charge (self):
1000 import inspect
1001
1002 import x402_fetch as fetch
1003
1004 src = inspect.getsource(fetch.x402_fetch)
1005 self .assertIn( "no double charge" , src)
1006 self .assertIn( '"paid": False' , src)
1007
1008
1009 class SafeMethodTests ( unittest . TestCase ):
1010 """Method support without opening an exfiltration channel.
1011
1012 The fetch tool takes any method. Paid retrieval needs GET/HEAD; a body-bearing verb
1013 would let the agent push agent-chosen data to an arbitrary origin, which the
1014 policy gate does not validate because it checks the URL, not a body.
1015 """
1016
1017 @ classmethod
1018 def setUpClass (cls):
1019 try :
1020 import x402_fetch # noqa: F401
1021 except ImportError as e: # pragma: no cover
1022 raise unittest.SkipTest( f "x402_fetch unavailable: { e } " )
1023
1024 def test_body_bearing_methods_are_refused (self):
1025 import x402_fetch as fetch
1026
1027 for method in ( "POST" , "PUT" , "PATCH" , "DELETE" ):
1028 with self .subTest( method = method):
1029 with self .assertRaises(fetch.PaymentBlocked):
1030 fetch._get( "https://example.com/x" , method = method)
1031
1032 def test_get_and_head_are_allowed (self):
1033 import x402_fetch as fetch
1034
1035 self .assertEqual(fetch. _ALLOWED_METHODS , ( "GET" , "HEAD" ))
1036
1037 def test_method_is_case_insensitive (self):
1038 import x402_fetch as fetch
1039
1040 with self .assertRaises(fetch.PaymentBlocked):
1041 fetch._get( "https://example.com/x" , method = "post" )
1042
1043
1044 class HarnessCliTests ( unittest . TestCase ):
1045 """The CLI is the interface a harness actually uses, so its contract is tested.
1046
1047 Claude Code, Codex, Cursor, Kiro and OpenClaw run shell commands — they do not
1048 import Python and build an agent object. Exit codes matter because a harness
1049 branches on them without parsing JSON.
1050 """
1051
1052 @ classmethod
1053 def setUpClass (cls):
1054 cls .cli = Path( __file__ ).resolve().parent / "x402_fetch_cli.py"
1055 if not cls .cli.exists(): # pragma: no cover
1056 raise unittest.SkipTest( "CLI not present" )
1057
1058 def setUp (self):
1059 self .tmp = tempfile.TemporaryDirectory()
1060 self .cfg = Path( self .tmp.name) / "config.json"
1061 self .cfg.write_text(json.dumps({ "resources" : {}, "policy" : BASE_POLICY }))
1062 self .cfg.chmod( 0o 600 )
1063 self .env = dict (os.environ)
1064 for k in pol. RESOURCE_ENV .values():
1065 self .env.pop(k, None )
1066
1067 def tearDown (self):
1068 self .tmp.cleanup()
1069
1070 def _run (self, * args):
1071 import subprocess
1072
1073 scripts = self .cli.parent
1074 bootstrap = """
1075 import runpy
1076 import sys
1077 from pathlib import Path
1078
1079 scripts = Path(sys.argv.pop(1))
1080 config = Path(sys.argv.pop(1))
1081 cli = scripts / "x402_fetch_cli.py"
1082 sys.path.insert(0, str(scripts))
1083 import x402_policy
1084 x402_policy.runtime_config_path = lambda: config
1085 sys.argv = [str(cli), *sys.argv[1:]]
1086 runpy.run_path(str(cli), run_name="__main__")
1087 """
1088 return subprocess.run(
1089 [
1090 sys.executable,
1091 "-c" ,
1092 bootstrap,
1093 str (scripts),
1094 str ( self .cfg),
1095 * args,
1096 ],
1097 capture_output = True , text = True , env = self .env, timeout = 120 ,
1098 )
1099
1100 def test_refusal_exits_2_not_1 (self):
1101 """A refusal is a decision, not a fault — a harness must be able to tell."""
1102 r = self ._run( "https://169.254.169.254/latest/meta-data/" )
1103 self .assertEqual(r.returncode, 2 )
1104 self .assertTrue(json.loads(r.stdout)[ "refused" ])
1105
1106 def test_status_reports_unusable_without_a_session (self):
1107 r = self ._run( "--status" )
1108 self .assertEqual(r.returncode, 2 )
1109 self .assertFalse(json.loads(r.stdout)[ "usable" ])
1110
1111 def test_output_is_a_single_json_object_on_stdout (self):
1112 """Harnesses parse stdout; diagnostics must not contaminate it."""
1113 r = self ._run( "https://169.254.169.254/" )
1114 json.loads(r.stdout) # raises if not exactly one JSON document
1115
1116 def test_no_url_and_no_flag_is_a_usage_error (self):
1117 self .assertEqual( self ._run().returncode, 1 )
1118
1119 def test_needs_no_framework_import (self):
1120 """The CLI must not depend on Strands, LangGraph, or any agent framework."""
1121 src = self .cli.read_text()
1122 for framework in ( "strands" , "langgraph" , "langchain" , "from agents import" , "crewai" ):
1123 self .assertNotIn(framework, src.lower())
1124
1125
1126 class BrowserHandleTests ( unittest . TestCase ):
1127 """Browser payments use an opaque, single-purpose handle."""
1128
1129 @ classmethod
1130 def setUpClass (cls):
1131 try :
1132 import x402_fetch # noqa: F401
1133 except ImportError as e: # pragma: no cover
1134 raise unittest.SkipTest( f "x402_fetch unavailable: { e } " )
1135
1136 def setUp (self):
1137 import time
1138
1139 import x402_fetch as fetch
1140
1141 self .fetch = fetch
1142 fetch. _PROOF_VAULT .clear()
1143 self .handle = "x402h_unittest"
1144 fetch. _PROOF_VAULT [ self .handle] = {
1145 "header" : { "PAYMENT-SIGNATURE" : "PROOF_BYTES_THAT_MUST_NOT_LEAK" },
1146 "origin" : ORIGIN ,
1147 "path" : "/v1/x402-test" ,
1148 "expires_at" : time.monotonic() + 90 ,
1149 }
1150
1151 def tearDown (self):
1152 self .fetch. _PROOF_VAULT .clear()
1153
1154 def test_handle_is_bound_to_the_origin (self):
1155 with self .assertRaises( self .fetch.PaymentBlocked):
1156 self .fetch.attach_browser_payment( self .handle, "https://evil.example.com/v1/x402-test" )
1157
1158 def test_handle_is_bound_to_the_resource_path (self):
1159 with self .assertRaises( self .fetch.PaymentBlocked):
1160 self .fetch.attach_browser_payment( self .handle, f " { ORIGIN } /some/other/path" )
1161
1162 def test_handle_redeems_once_then_is_consumed (self):
1163 header = self .fetch.attach_browser_payment( self .handle, f " { ORIGIN } /v1/x402-test" )
1164 self .assertIn( "PAYMENT-SIGNATURE" , header)
1165 with self .assertRaises( self .fetch.PaymentBlocked):
1166 self .fetch.attach_browser_payment( self .handle, f " { ORIGIN } /v1/x402-test" )
1167
1168 def test_expired_handle_is_refused (self):
1169 import time
1170
1171 self .fetch. _PROOF_VAULT [ self .handle][ "expires_at" ] = time.monotonic() - 1
1172 with self .assertRaises( self .fetch.PaymentBlocked):
1173 self .fetch.attach_browser_payment( self .handle, f " { ORIGIN } /v1/x402-test" )
1174
1175 def test_unknown_handle_is_refused (self):
1176 with self .assertRaises( self .fetch.PaymentBlocked):
1177 self .fetch.attach_browser_payment( "x402h_nope" , f " { ORIGIN } /v1/x402-test" )
1178
1179 def test_handle_is_not_derived_from_the_proof (self):
1180 """A handle must carry no information about the proof it references."""
1181 self .assertTrue( self .handle.startswith( "x402h_" ))
1182 self .assertNotIn( "PROOF" , self .handle.upper())
1183
1184 def test_model_facing_output_never_contains_the_proof (self):
1185 """prepare_browser_payment returns a handle + receipt, never proof bytes."""
1186 import inspect
1187
1188 src = inspect.getsource( self .fetch.prepare_browser_payment)
1189 # The proof goes into the vault, and only the handle is serialized out.
1190 self .assertIn( '"header": payment_header, # stays here; never returned' , src)
1191 self .assertIn( '"handle": handle' , src)
1192 self .assertNotIn( '"header": payment_header}' , src)
1193 self .assertNotIn( "json.dumps(payment_header" , src)
1194
1195
1196 class SessionStatusTests ( unittest . TestCase ):
1197 """Feature parity with the plugin's `get_payment_session_status` (read-only)."""
1198
1199 @ classmethod
1200 def setUpClass (cls):
1201 try :
1202 import x402_fetch # noqa: F401
1203 except ImportError as e: # pragma: no cover
1204 raise unittest.SkipTest( f "x402_fetch unavailable: { e } " )
1205
1206 def test_reports_unusable_and_names_the_human_step (self):
1207 import x402_fetch as fetch
1208
1209 saved = os.environ.pop( "PAYMENT_SESSION_ID" , None )
1210 try :
1211 result = json.loads(fetch.payment_session_status())
1212 self .assertFalse(result[ "usable" ])
1213 # Must not imply the agent can fix it itself.
1214 self .assertIn( "operator" , result[ "next_step" ].lower())
1215 finally :
1216 if saved is not None :
1217 os.environ[ "PAYMENT_SESSION_ID" ] = saved
1218
1219 def test_status_is_read_only (self):
1220 """It must not be able to create or extend a session."""
1221 import inspect
1222
1223 import x402_fetch as fetch
1224
1225 src = inspect.getsource(fetch.payment_session_status)
1226 for mutator in ( "create_payment_session" , "CreatePaymentSession" , "update_" , "delete_" ):
1227 self .assertNotIn(mutator, src)
1228
1229 def test_region_is_not_forced_to_a_hardcoded_default (self):
1230 """region_name must come from resolve_resource(), never a bare `or "us-west-2"`.
1231
1232 A hardcoded fallback here would silently override a region resolved from
1233 config.json, the environment, or boto3's own session/profile resolution
1234 whenever none of those apply — forcing a real payment manager lookup
1235 (which lives in the operator's actual deployment region) against the
1236 wrong AWS region and producing a confusing manager-not-found error.
1237 """
1238 import inspect
1239
1240 import x402_fetch as fetch
1241
1242 for fn in (fetch.payment_session_status, fetch.prepare_browser_payment, fetch.x402_fetch):
1243 src = inspect.getsource(fn)
1244 self .assertNotIn(
1245 'or "us-west-2"' ,
1246 src,
1247 f ' { fn. __name__ } must not hardcode a region fallback; let boto3 resolve it' ,
1248 )
1249
1250 def test_unknown_sdk_status_is_not_reported_as_usable (self):
1251 import types
1252
1253 import x402_fetch as fetch
1254
1255 class PaymentManager :
1256 def __init__ (self, ** _kwargs):
1257 pass
1258
1259 def get_payment_session (self, ** _kwargs):
1260 return { "state" : "MAYBE_ACTIVE" }
1261
1262 payments_module = types.ModuleType( "bedrock_agentcore.payments" )
1263 payments_module.PaymentManager = PaymentManager
1264 agentcore_module = types.ModuleType( "bedrock_agentcore" )
1265 agentcore_module.payments = payments_module
1266 policy = {
1267 "_resources" : {
1268 "payment_session_id" : "session-1" ,
1269 "payment_manager_arn" : "arn:aws:bedrock-agentcore:region:account:payment-manager/pm-1" ,
1270 }
1271 }
1272
1273 with mock.patch.object(fetch.pol, "load_config" , return_value = policy):
1274 with mock.patch.dict(
1275 sys.modules,
1276 {
1277 "bedrock_agentcore" : agentcore_module,
1278 "bedrock_agentcore.payments" : payments_module,
1279 },
1280 ):
1281 result = json.loads(fetch.payment_session_status())
1282
1283 self .assertFalse(result[ "usable" ])
1284 self .assertEqual(result[ "status" ], "unknown" )
1285
1286
1287 class DnsPinningTests ( unittest . TestCase ):
1288 """DNS rebinding must not reopen the SSRF window.
1289
1290 These live here rather than in a separate file so the whole security surface
1291 runs in one command. They import x402_fetch, which needs httpx.
1292 """
1293
1294 @ classmethod
1295 def setUpClass (cls):
1296 try :
1297 import x402_fetch # noqa: F401
1298 except ImportError as e: # pragma: no cover - environment without httpx
1299 raise unittest.SkipTest( f "x402_fetch unavailable: { e } " )
1300
1301 def test_pin_is_not_implemented_by_patching_a_global (self):
1302 """Patching socket.getaddrinfo is racy: a concurrent fetch can unpin."""
1303 import socket
1304
1305 import x402_fetch as fetch
1306
1307 before = socket.getaddrinfo
1308 transport = fetch._PinnedResolverTransport( "93.184.216.34" , verify = fetch._ssl_context())
1309 self .assertIs(socket.getaddrinfo, before, "must not mutate socket.getaddrinfo" )
1310 self .assertEqual( type (transport._pool._network_backend). __name__ , "_PinnedBackend" )
1311
1312 def test_connect_ignores_the_requested_host (self):
1313 """A rebind between resolve and connect cannot change the destination."""
1314 import inspect
1315
1316 import x402_fetch as fetch
1317
1318 transport = fetch._PinnedResolverTransport( "93.184.216.34" , verify = fetch._ssl_context())
1319 src = inspect.getsource(transport._pool._network_backend.connect_tcp)
1320 self .assertIn( "self._pinned, port" , src)
1321
1322 def test_internal_pin_is_refused_at_connect_time (self):
1323 import x402_fetch as fetch
1324
1325 transport = fetch._PinnedResolverTransport( "169.254.169.254" , verify = fetch._ssl_context())
1326 with self .assertRaises(pol.PolicyError):
1327 transport._pool._network_backend.connect_tcp( "anything.test" , 443 )
1328
1329 def test_tls_context_never_disables_verification (self):
1330 import ssl
1331
1332 import x402_fetch as fetch
1333
1334 ctx = fetch._ssl_context()
1335 self .assertTrue(ctx.check_hostname)
1336 self .assertEqual(ctx.verify_mode, ssl. CERT_REQUIRED )
1337
1338
1339 class AuthorizeTests ( unittest . TestCase ):
1340 """End-to-end gate behavior."""
1341
1342 def test_approved_payment_returns_full_decision (self):
1343 decision = pol.authorize_payment( f " { ORIGIN } /v1/x402-test" , challenge(), BASE_POLICY )
1344 self .assertEqual(decision[ "amount_usd" ], "0.1" )
1345 self .assertEqual(decision[ "accept" ][ "payTo" ], MERCHANT )
1346 self .assertEqual(decision[ "origin" ], ORIGIN )
1347 self .assertEqual(decision[ "x402_version" ], 1 )
1348 self .assertEqual( len (decision[ "client_token" ]), 64 )
1349 # x402 v2 requires resource in the signed payload for URL binding
1350 self .assertEqual(decision[ "resource" ], { "url" : f " { ORIGIN } /v1/x402-test" })
1351
1352 def test_resource_none_when_challenge_omits_it (self):
1353 ch = challenge()
1354 del ch[ "resource" ]
1355 decision = pol.authorize_payment( f " { ORIGIN } /v1/x402-test" , ch, BASE_POLICY )
1356 self .assertIsNone(decision[ "resource" ])
1357
1358 def test_resource_url_mismatch_is_refused (self):
1359 ch = challenge()
1360 ch[ "resource" ] = { "url" : "https://attacker.example/evil" }
1361 with self .assertRaises(pol.PolicyError):
1362 pol.authorize_payment( f " { ORIGIN } /v1/x402-test" , ch, BASE_POLICY )
1363
1364 def test_resource_extra_fields_are_stripped (self):
1365 ch = challenge()
1366 ch[ "resource" ] = { "url" : f " { ORIGIN } /v1/x402-test" , "injected" : "payload" , "nested" : { "x" : 1 }}
1367 decision = pol.authorize_payment( f " { ORIGIN } /v1/x402-test" , ch, BASE_POLICY )
1368 self .assertEqual(decision[ "resource" ], { "url" : f " { ORIGIN } /v1/x402-test" })
1369
1370 def test_resource_non_dict_is_ignored (self):
1371 ch = challenge()
1372 ch[ "resource" ] = "not-a-dict"
1373 decision = pol.authorize_payment( f " { ORIGIN } /v1/x402-test" , ch, BASE_POLICY )
1374 self .assertIsNone(decision[ "resource" ])
1375
1376 def test_resource_non_string_url_is_ignored (self):
1377 ch = challenge()
1378 ch[ "resource" ] = { "url" : 12345 }
1379 decision = pol.authorize_payment( f " { ORIGIN } /v1/x402-test" , ch, BASE_POLICY )
1380 self .assertIsNone(decision[ "resource" ])
1381
1382 def test_non_dict_challenge_is_refused (self):
1383 for bad in ( "[]" , None , 5 , []):
1384 with self .assertRaises(pol.PolicyError):
1385 pol.authorize_payment( f " { ORIGIN } /x" , bad, BASE_POLICY )
1386
1387 def test_base_unit_conversion (self):
1388 self .assertEqual(pol.base_units_to_usd( "1000000" ), pol.Decimal( "1" ))
1389 self .assertEqual(pol.base_units_to_usd( "1" ), pol.Decimal( "0.000001" ))
1390
1391 def test_fractional_base_units_are_refused (self):
1392 with self .assertRaises(pol.PolicyError):
1393 pol.base_units_to_usd( "100.5" )
1394
1395 def test_only_canonical_integer_amounts_are_accepted (self):
1396 """A publisher must not express an amount in an exotic encoding.
1397
1398 Decimal() alone parses all of these. They are rejected so an amount can
1399 never read as one value to the code and another to a human reading logs.
1400 """
1401 for bad in (
1402 "1E+7" , # scientific notation
1403 "Infinity" , # not finite
1404 "NaN" , # not a number
1405 "1_000_000" , # underscore separators
1406 "500000.0" , # trailing fraction
1407 " 500000 " , # surrounding whitespace
1408 "+500000" , # explicit sign
1409 "0x7A120" , # hex
1410 "" , # empty
1411 "-500000" , # negative
1412 ):
1413 with self .subTest( amount = bad):
1414 with self .assertRaises(pol.PolicyError):
1415 pol.base_units_to_usd(bad)
1416
1417 def test_canonical_amounts_still_work (self):
1418 self .assertEqual(pol.base_units_to_usd( "500000" ), pol.Decimal( "0.5" ))
1419 self .assertEqual(pol.base_units_to_usd( 500000 ), pol.Decimal( "0.5" ))
1420
1421 def test_exotic_amount_in_a_challenge_is_refused_not_paid (self):
1422 """The gate must refuse such a challenge outright, not coerce the value."""
1423 for bad in ( "1E+7" , "Infinity" , "500000.0" ):
1424 with self .subTest( amount = bad):
1425 with self .assertRaises(pol.PolicyError):
1426 pol.select_accept_entry(challenge( amount = bad), BASE_POLICY )
1427
1428
1429 if __name__ == "__main__" :
1430 unittest.main( verbosity = 2 )