Setting the file. One moment.
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
def validated_amount_units
— line 380
This file
Number 21.12
Position 12 of 14
Type Python
Size 26 KB
Lines 629 scripts/ x402_policy.py
Python · 629 lines · 26 KB
The runtime resolves ~/.agents-pay/config.json from the operating-system account,
17 not HOME or a config-path environment variable. Administrative callers and tests
18 may pass a path explicitly. The file must be regular, owned by the current user,
19 mode 0600, in a directory that is not group/world-writable, with no symlinks (see
20 load_config). One file, two sections:
21
22 {
23 "resources": { # what to pay WITH
24 "payment_manager_arn": "arn:aws:bedrock-agentcore:...:payment-manager/pm-1",
25 "payment_instrument_id": "pi-...",
26 "payment_session_id": "ps-...", # a spending credential
27 "user_id": "alice",
28 "region": "us-west-2"
29 },
30 "policy": { # what may be paid
31 "max_per_payment_usd": "0.05", # PER-PAYMENT ceiling, required
32 "allowed_networks": ["eip155:84532"],
33 "allowed_assets": { # network -> exact asset contract(s)
34 "eip155:84532": ["0x036CbD53842c5426634e7929541eC2318f3dCF7e"]
35 },
36 "allowed_recipients": ["0x1111111111111111111111111111111111111111"],
37 # Or, instead of allowed_recipients:
38 # "allow_any_recipient": true,
39 "allowed_origins": ["https://sandbox.node4all.com"],
40 "allowed_schemes": ["exact"]
41 }
42 }
43
44 Absent keys deny rather than allow. There is no implicit wildcard.
45
46 Recipient validation
47 --------------------
48 By default, the payee (`payTo`) must appear in `allowed_recipients`, an
49 operator-approved allowlist in the config file. Unknown recipients are refused
50 before signing. An operator may instead set `allow_any_recipient` to the literal
51 boolean `true`. The two modes are mutually exclusive, and the runtime rejects a
52 policy that enables both.
53
54 Allowing any recipient means a publisher controls the beneficiary. Network,
55 asset, scheme, origin/resource, per-payment, and cumulative session limits still
56 apply, but recipient allowlisting no longer protects against a malicious payee.
57
58 Two ceilings, not one
59 ---------------------
60 `max_per_payment_usd` is NOT a duplicate of the session budget:
61
62 * the session budget is CUMULATIVE — total spend before a human must re-approve;
63 * `max_per_payment_usd` is PER TRANSACTION.
64
65 With only the session budget, one hostile challenge for the full remaining balance
66 drains it in a single payment. A trusted positive maximum for each payment keeps
67 both bounds meaningful.
68
69 Scope / non-goals
70 -----------------
71 This module decides IF a payment may proceed and derives the idempotency key.
72 It never holds provider credentials, never signs, never performs network I/O.
73 Fetching and settlement live in x402_fetch.py.
74 """
75
76 from __future__ import annotations
77
78 import hashlib
79 import ipaddress
80 import json
81 import os
82 import socket
83 import stat
84 from decimal import Decimal, InvalidOperation
85 from pathlib import Path
86 from urllib.parse import urlparse
87
88 # USDC has 6 decimals on every chain AgentCore Payments supports. x402 quotes
89 # amounts as integer base units, so 500000 base units == 0.50 USDC.
90 USDC_DECIMALS = 6
91
92 # Fields we require on an accepts[] entry before it is eligible for payment.
93 _REQUIRED_ACCEPT_FIELDS = ( "scheme" , "network" , "asset" , "payTo" )
94
95
96 class PolicyError ( Exception ):
97 """A payment was refused, or the policy itself is unusable.
98
99 Carries no challenge values, so the message is safe to surface to a model.
100 """
101
102
103 def _fail (reason: str ) -> "PolicyError" :
104 return PolicyError(reason)
105
106
107 def runtime_config_path () -> Path:
108 """Return the config path bound to the current OS account.
109
110 Path.home(), HOME, and config-path environment variables are caller-controlled
111 in shell-capable harnesses. The runtime therefore asks the OS account database
112 for the home directory and fails closed where that trusted lookup is unavailable.
113 """
114 try :
115 import pwd
116 except ImportError :
117 raise _fail(
118 "Cannot resolve the runtime payment config from the OS account on this "
119 "platform. Payments are refused."
120 ) from None
121
122 try :
123 home = Path(pwd.getpwuid(os.getuid()).pw_dir)
124 except KeyError :
125 raise _fail(
126 "The current uid has no OS account home directory. Payments are refused."
127 ) from None
128 if not home.is_absolute():
129 raise _fail( "The OS account home directory is not absolute. Payments are refused." )
130 return home / ".agents-pay" / "config.json"
131
132
133 def load_config (path: str | os.PathLike[ str ] | None = None ) -> dict :
134 """Load and validate the operator config: resource identifiers + payment policy.
135
136 One file, two sections. `resources` holds the identifiers the runtime needs
137 (manager ARN, instrument, session, user); `policy` holds the limits. They live
138 together because they are written at the same moment by the same human, and
139 splitting them made the operator hand-copy identifiers between steps.
140
141 Merging them also buys a real control, not just convenience: the session ID is
142 a spending credential, and the sanctioned runtime cannot select a replacement
143 policy file through HOME or a config-path environment variable. Resource values
144 in the file also win over environment fallbacks — see resolve_resource().
145
146 The file is a security control, so one that anyone else can write is treated as
147 no config at all. lstat (not stat) rejects symlinks rather than following them.
148 These checks do not protect against arbitrary code already running as the file
149 owner; that requires OS, container, or IAM isolation around the signer.
150 """
151 p = Path(path) if path is not None else runtime_config_path()
152
153 try :
154 st = p.lstat()
155 except FileNotFoundError :
156 raise _fail(
157 f "No payment config at { p } . Payments are refused until an operator "
158 "creates one (agents_pay_admin.py init-config). This is intentional: "
159 "there is no permissive default."
160 ) from None
161
162 if stat.S_ISLNK(st.st_mode):
163 raise _fail( f "Payment config { p } is a symlink; refusing to follow it." )
164 if not stat.S_ISREG(st.st_mode):
165 raise _fail( f "Payment config { p } is not a regular file." )
166 if st.st_uid != os.getuid():
167 raise _fail( f "Payment config { p } is not owned by the current user." )
168 # Reject any group/other bit: a config others can rewrite is not a control.
169 if st.st_mode & 0o 077 :
170 raise _fail(
171 f "Payment config { p } has mode { stat.filemode(st.st_mode) } ; "
172 "expected 0600. Run: chmod 600 " + str (p)
173 )
174
175 # The containing directory matters as much as the file. Write access to the
176 # directory lets another principal rename a wider config into place, which
177 # no amount of checking on the old inode would detect.
178 try :
179 dir_st = p.parent.lstat()
180 except OSError :
181 raise _fail( f "Cannot stat the directory containing { p } ." ) from None
182 if dir_st.st_uid != os.getuid():
183 raise _fail( f "Directory { p.parent } is not owned by the current user." )
184 if dir_st.st_mode & 0o 022 :
185 raise _fail(
186 f "Directory { p.parent } is group/world-writable "
187 f "( { stat.filemode(dir_st.st_mode) } ); another user could replace the "
188 "config file. Run: chmod 700 " + str (p.parent)
189 )
190
191 try :
192 raw = json.loads(p.read_text())
193 except json.JSONDecodeError as e:
194 raise _fail( f "Payment config { p } is not valid JSON: { e } " ) from None
195 if not isinstance (raw, dict ):
196 raise _fail( f "Payment config { p } must be a JSON object." )
197
198 # A flat file (policy keys at the top level) is still accepted, so an existing
199 # policy.json keeps working rather than failing open or failing loudly.
200 policy = raw.get( "policy" ) if isinstance (raw.get( "policy" ), dict ) else raw
201 resources = raw.get( "resources" ) if isinstance (raw.get( "resources" ), dict ) else {}
202
203 if "max_per_payment_usd" in policy:
204 cap_key = "max_per_payment_usd"
205 elif "max_amount_usd" in policy: # earlier name, still honoured
206 cap_key = "max_amount_usd"
207 else :
208 raise _fail(
209 "Payment policy is missing 'max_per_payment_usd'. This is the PER-PAYMENT "
210 "ceiling and is not the same thing as the session budget: the session "
211 "budget is cumulative, so without a per-payment cap a single hostile "
212 "challenge can drain the whole budget in one transaction."
213 )
214 try :
215 cap = Decimal( str (policy[cap_key]))
216 except (InvalidOperation, ValueError ):
217 raise _fail( f "Payment policy ' { cap_key } ' is not a decimal number." ) from None
218 policy[ "max_per_payment_usd" ] = policy[cap_key] # normalize for callers
219 policy[ "_resources" ] = resources # carried for resolve_resource()
220 if cap <= 0 :
221 raise _fail( f "Payment policy ' { cap_key } ' must be greater than zero." )
222
223 return policy
224
225
226 #: config `resources` key -> the environment variable that may substitute for it.
227 RESOURCE_ENV = {
228 "payment_manager_arn" : "PAYMENT_MANAGER_ARN" ,
229 "payment_instrument_id" : "PAYMENT_INSTRUMENT_ID" ,
230 "payment_session_id" : "PAYMENT_SESSION_ID" ,
231 "user_id" : "PAYMENT_USER_ID" ,
232 "region" : "AWS_REGION" ,
233 }
234
235
236 def resolve_resource (policy: dict , key: str ) -> str | None :
237 """Resolve one resource identifier: **config file first**, environment second.
238
239 The precedence is the security-relevant part, and it is deliberately the
240 opposite of the usual "env overrides file" convention.
241
242 The session ID is a spending credential: it names the budget being drawn down.
243 If the environment could override the file, an agent able to set a variable
244 could point the runtime at some other session with a larger budget, and the
245 0600 file would be decorative. So the file wins wherever it speaks.
246
247 The environment remains the fallback for deployments with no writable home —
248 containers, Lambda — where identifiers arrive by injection. That is a real
249 need, but it is the weaker mode: anything that can set the environment can
250 choose the session.
251 """
252 resources = policy.get( "_resources" ) or {}
253 from_file = resources.get(key)
254 if from_file:
255 return str (from_file)
256 env_name = RESOURCE_ENV .get(key)
257 return os.environ.get(env_name) if env_name else None
258
259 def per_payment_cap (policy: dict ) -> str :
260 """The per-payment USD ceiling, under either the current or the earlier key.
261
262 load_config() normalizes this, but callers may hand a policy dict straight in
263 (tests, embedded use), so resolve it here too rather than assuming. Missing is
264 a refusal, never an unbounded payment.
265 """
266 for key in ( "max_per_payment_usd" , "max_amount_usd" ):
267 if key in policy:
268 return str (policy[key])
269 raise _fail(
270 "Payment policy has no per-payment ceiling ('max_per_payment_usd'). "
271 "Refusing to pay: the session budget is cumulative and does not bound a "
272 "single transaction."
273 )
274
275
276 def _policy_list (policy: dict , key: str ) -> list[ str ]:
277 """Read a list-valued policy key. Missing or malformed means deny (empty)."""
278 value = policy.get(key)
279 if value is None :
280 return []
281 if not isinstance (value, list ):
282 raise _fail( f "Payment policy ' { key } ' must be a list." )
283 return [ str (v) for v in value]
284
285
286 def _canonical_origin (url: str ) -> str :
287 """scheme://host[:port], lowercased, with the default HTTPS port dropped."""
288 u = urlparse(url)
289 host = (u.hostname or "" ).lower()
290 if u.port and u.port != 443 :
291 return f " { u.scheme } :// { host } : { u.port } "
292 return f " { u.scheme } :// { host } "
293
294
295 def assert_public_https_url (url: str ) -> str :
296 """Require HTTPS and a publicly routable destination; return the origin.
297
298 Resolves every address the hostname maps to and rejects the request if ANY
299 of them is internal. Checking only the first answer would let a host with
300 one public and one private address through.
301
302 This is a pre-flight check. It does not by itself defeat DNS rebinding,
303 because the OS resolves again when the socket is opened — x402_fetch.py
304 closes that gap by pinning the connection to a vetted address.
305 """
306 u = urlparse(url)
307 if u.scheme != "https" :
308 raise _fail( "Only https:// payment URLs are allowed." )
309 if not u.hostname:
310 raise _fail( "Payment URL has no host." )
311 if u.username or u.password:
312 raise _fail( "Payment URL must not contain embedded credentials." )
313
314 try :
315 infos = socket.getaddrinfo(u.hostname, u.port or 443 , proto = socket. IPPROTO_TCP )
316 except socket.gaierror:
317 raise _fail( "Payment URL hostname does not resolve." ) from None
318
319 for info in infos:
320 assert_public_ip(info[ 4 ][ 0 ])
321
322 return _canonical_origin(url)
323
324
325 def assert_public_ip (addr: str ) -> None :
326 """Reject loopback, private, link-local, and other non-public ranges.
327
328 Deliberately broader than is_private/is_loopback/is_link_local, which alone
329 still admit multicast and the CGNAT 100.64/10 shared-address space.
330 """
331 try :
332 ip = ipaddress.ip_address(addr)
333 except ValueError :
334 raise _fail( "Payment URL resolved to an unparseable address." ) from None
335
336 # Unwrap ::ffff:127.0.0.1 style addresses so v4 rules apply to them.
337 mapped = getattr (ip, "ipv4_mapped" , None )
338 if mapped is not None :
339 ip = mapped
340
341 if (
342 ip.is_private
343 or ip.is_loopback
344 or ip.is_link_local
345 or ip.is_multicast
346 or ip.is_reserved
347 or ip.is_unspecified
348 ):
349 raise _fail( "Payment URL resolves to a non-public address." )
350
351 # 100.64.0.0/10 (CGNAT) is not flagged by any is_* property but is not
352 # publicly routable, and 169.254.169.254 lives behind it in some networks.
353 if ip.version == 4 and ip in ipaddress.ip_network( "100.64.0.0/10" ):
354 raise _fail( "Payment URL resolves to a non-public address." )
355
356
357 def base_units_to_usd (amount: str | int , decimals: int = USDC_DECIMALS ) -> Decimal:
358 """Convert an x402 integer base-unit amount to a USD Decimal.
359
360 Accepts ONLY a canonical non-negative integer literal: digits, nothing else.
361 Decimal() would otherwise happily parse "1E+7", "Infinity", "500000.0",
362 " 500000 ", and "+500000". None of those breach the ceiling on their own
363 (the comparison still holds), but a publisher should not get to express an
364 amount in a form that a human reviewing logs would misread. An exotic
365 encoding therefore fails closed rather than relying on downstream
366 arithmetic to save us.
367 """
368 text = amount if isinstance (amount, str ) else str (amount)
369 if not isinstance (amount, int ) and not text.isdigit():
370 raise _fail( "Payment challenge amount must be a plain integer in base units." )
371 try :
372 raw = Decimal(text)
373 except InvalidOperation:
374 raise _fail( "Payment challenge amount is not a number." ) from None
375 if raw <= 0 :
376 raise _fail( "Payment challenge amount must be positive." )
377 return raw / (Decimal( 10 ) ** decimals)
378
379
380 def validated_amount_units (entry: dict ) -> str :
381 """Resolve one canonical amount and reject conflicting aliases.
382
383 x402 v1 commonly uses maxAmountRequired while v2 uses amount. Supporting both
384 is safe only when they cannot describe different transactions.
385 """
386 amount = entry.get( "amount" )
387 maximum = entry.get( "maxAmountRequired" )
388 if amount is not None and maximum is not None and str (amount) != str (maximum):
389 raise _fail( "Payment challenge contains conflicting amount fields." )
390 resolved = amount if amount is not None else maximum
391 if resolved is None :
392 raise _fail( "Payment challenge has no amount." )
393 text = str (resolved)
394 base_units_to_usd(text)
395 return text
396
397
398 def select_accept_entry (challenge: dict , policy: dict ) -> dict :
399 """Return the first accepts[] entry that satisfies policy, else raise.
400
401 Each entry must earn selection by passing every configured check. The publisher
402 cannot choose the network, asset, recipient, or amount unilaterally.
403 """
404 accepts = challenge.get( "accepts" )
405 if not isinstance (accepts, list ) or not accepts:
406 raise _fail( "Payment challenge has no accepts entries." )
407
408 allowed_networks = _policy_list(policy, "allowed_networks" )
409 allowed_schemes = _policy_list(policy, "allowed_schemes" )
410 if "allowed_schemes" not in policy:
411 allowed_schemes = [ "exact" ]
412 allow_any_recipient = policy.get( "allow_any_recipient" , False )
413 if not isinstance (allow_any_recipient, bool ):
414 raise _fail( "Payment policy 'allow_any_recipient' must be a boolean." )
415 if (
416 "allow_any_recipient" in policy
417 and "allowed_recipients" in policy
418 ):
419 raise _fail(
420 "Payment policy 'allow_any_recipient' and 'allowed_recipients' "
421 "are mutually exclusive."
422 )
423 allowed_recipients = [r.lower() for r in _policy_list(policy, "allowed_recipients" )]
424 allowed_assets = policy.get( "allowed_assets" ) or {}
425 if not isinstance (allowed_assets, dict ):
426 raise _fail( "Payment policy 'allowed_assets' must be an object." )
427 cap_usd = Decimal( str (per_payment_cap(policy)))
428
429 if not allowed_networks:
430 raise _fail( "Payment policy allows no networks." )
431 if not allowed_schemes:
432 raise _fail( "Payment policy allows no schemes." )
433 if not allow_any_recipient and not allowed_recipients:
434 raise _fail( "Payment policy allows no recipients." )
435
436 for entry in accepts:
437 if not isinstance (entry, dict ):
438 continue
439 if any (entry.get(f) in ( None , "" ) for f in _REQUIRED_ACCEPT_FIELDS ):
440 continue
441 if str (entry[ "scheme" ]) not in allowed_schemes:
442 continue
443 network = str (entry[ "network" ])
444 if network not in allowed_networks:
445 continue
446 # Asset contracts are compared case-insensitively: EVM addresses are
447 # hex and often differ only by EIP-55 checksum capitalization.
448 permitted_assets = [a.lower() for a in allowed_assets.get(network, [])]
449 if str (entry[ "asset" ]).lower() not in permitted_assets:
450 continue
451 if (
452 not allow_any_recipient
453 and str (entry[ "payTo" ]).lower() not in allowed_recipients
454 ):
455 continue
456 try :
457 amount_units = validated_amount_units(entry)
458 if base_units_to_usd(amount_units) > cap_usd:
459 continue
460 except PolicyError:
461 continue
462 return entry
463
464 # Uniform refusal: naming the failed field would let a publisher probe the
465 # policy by iterating challenges until the message changed.
466 raise _fail(
467 "No payment option in this challenge satisfies the configured policy "
468 "(scheme, network, asset, recipient, and amount ceiling are all enforced). "
469 "Refusing to pay."
470 )
471
472
473 def _validated_resource (challenge: dict , url: str ) -> dict | None :
474 """Validate and sanitize the challenge's resource object before signing.
475
476 The x402 v2 spec requires the signed header to echo the resource for URL
477 binding. However, the publisher controls the challenge, so we must verify
478 that resource.url matches the URL we actually requested. This prevents a
479 hostile publisher from binding the signature to a different resource.
480
481 Comparison uses origin+path only: query parameters appended by the
482 publisher (e.g. tracking params) do not change which server gets paid.
483
484 Returns a sanitized dict with only the `url` field, or None if the
485 challenge has no resource. Raises PolicyError on mismatch.
486 """
487 resource = challenge.get( "resource" )
488 if resource is None :
489 return None
490 if not isinstance (resource, dict ):
491 return None
492 resource_url = resource.get( "url" )
493 if not isinstance (resource_url, str ):
494 return None
495 # Bind: resource.url origin+path must match the requested origin+path.
496 requested = urlparse(url)
497 challenged = urlparse(resource_url)
498 req_base = f " { requested.scheme } :// { requested.netloc }{ requested.path } " .rstrip( "/" )
499 ch_base = f " { challenged.scheme } :// { challenged.netloc }{ challenged.path } " .rstrip( "/" )
500 if req_base != ch_base:
501 raise _fail( "Challenge resource.url does not match the requested URL." )
502 # Forward only the url field — no arbitrary publisher-controlled keys.
503 return { "url" : resource_url}
504
505
506 def authorize_payment (
507 url: str ,
508 challenge: dict ,
509 policy: dict | None = None ,
510 purchase_id: str | None = None ,
511 ) -> dict :
512 """Full gate: validate destination and challenge, return an approved decision.
513
514 Returns the vetted accepts entry plus the derived amount, origin, and a
515 stable idempotency key. Raises PolicyError on any refusal.
516
517 `purchase_id` distinguishes deliberate repeat purchases of the same resource;
518 see derive_client_token.
519 """
520 policy = policy if policy is not None else load_config()
521
522 origin = assert_public_https_url(url)
523
524 # Origin allowlisting is optional. HTTPS, address vetting, redirect refusal,
525 # rebinding protection, timeouts, and byte limits are always enforced. An
526 # operator with a known merchant set can pin it with allowed_origins.
527 allowed_origins = _policy_list(policy, "allowed_origins" )
528 if allowed_origins and origin.lower() not in [o.lower().rstrip( "/" ) for o in allowed_origins]:
529 raise _fail( f "Origin { origin } is not in the configured allowed_origins." )
530
531 if not isinstance (challenge, dict ):
532 raise _fail( "Payment challenge is not a JSON object." )
533
534 entry = select_accept_entry(challenge, policy)
535 amount_units = validated_amount_units(entry)
536 x402_version = int (challenge.get( "x402Version" ) or challenge.get( "version" ) or 1 )
537
538 # Sign exactly the amount the gate validated. v1 uses maxAmountRequired and
539 # v2 uses amount; the non-canonical alias is removed so downstream code cannot
540 # apply different precedence from the policy gate.
541 vetted_entry = dict (entry)
542 vetted_entry.pop( "amount" , None )
543 vetted_entry.pop( "maxAmountRequired" , None )
544 if x402_version == 1 :
545 vetted_entry[ "maxAmountRequired" ] = amount_units
546 else :
547 vetted_entry[ "amount" ] = amount_units
548
549 return {
550 "accept" : vetted_entry,
551 "origin" : origin,
552 "resource" : _validated_resource(challenge, url),
553 "amount_base_units" : str (amount_units),
554 "amount_usd" : str (base_units_to_usd(amount_units)),
555 "x402_version" : x402_version,
556 "client_token" : derive_client_token(
557 url,
558 vetted_entry,
559 challenge,
560 purchase_id,
561 session_id = resolve_resource(policy, "payment_session_id" ) or "" ,
562 ),
563 }
564
565
566 def derive_client_token (
567 url: str ,
568 accept: dict ,
569 challenge: dict ,
570 purchase_id: str | None = None ,
571 * ,
572 session_id: str | None = None ,
573 policy: dict | None = None ,
574 ) -> str :
575 """Derive a stable idempotency token for one logical purchase.
576
577 Part of the same fix batch as the region-resolution changes in x402_fetch.py
578 (see that module's docstring): the `policy=` parameter below closes a second
579 instance of the "config.json has the value, code reads the environment
580 instead" pattern found in this batch, this time for the session ID used as
581 idempotency-token material rather than for region.
582
583 Same (session, resource, network, asset, recipient, amount) always yields the
584 same token, so a retry after a lost response replays the SAME authorization
585 instead of creating a second irreversible payment. Derived rather than random
586 precisely so it survives a process restart.
587
588 The publisher's nonce is deliberately NOT part of the material. A retry
589 re-fetches the 402 and many servers issue a fresh nonce each time, so mixing
590 it in would produce a different token per attempt — turning the retry this
591 function exists to protect into a second real payment, and handing a hostile
592 publisher a way to force double charges by rotating nonces.
593
594 The trade-off: two intentional purchases of the same resource, for the same
595 amount, in the same session collapse to one token, so the second would be
596 suppressed as a replay. Pass an explicit `purchase_id` (an order number, a
597 turn counter — anything the caller controls) to distinguish deliberate
598 repeat buys. Suppressing a duplicate charge is the safer default when the
599 caller has not said otherwise.
600
601 session_id resolution: an explicit `session_id` always wins. If neither
602 `session_id` nor `policy` is given, this falls back to a raw environment
603 read for backward compatibility with existing callers (tests, embedded use)
604 that predate the `policy` parameter. A caller that DOES pass `policy` gets
605 resolve_resource()'s documented config-file-first precedence instead — the
606 correct behavior for any new caller resolving the session itself rather than
607 passing an explicit session_id (the production call site in
608 authorize_payment() already always passes session_id explicitly and is
609 unaffected either way).
610 """
611 if session_id is not None :
612 session = session_id
613 elif policy is not None :
614 session = resolve_resource(policy, "payment_session_id" ) or ""
615 else :
616 session = os.environ.get( "PAYMENT_SESSION_ID" , "" )
617 material = " \x1f " .join( # unit separator: cannot appear in these values
618 [
619 session,
620 _canonical_origin(url),
621 urlparse(url).path or "/" ,
622 str (accept.get( "network" , "" )),
623 str (accept.get( "asset" , "" )).lower(),
624 str (accept.get( "payTo" , "" )).lower(),
625 validated_amount_units(accept),
626 purchase_id or "" ,
627 ]
628 )
629 return hashlib.sha256(material.encode( "utf-8" )).hexdigest()