Setting the file. One moment.
Stripe CLI · Using Stripe CLI · bitwarden/ai-plugins · Skills Docs
Back to the top of the page def build_get_clock_argv
— line 267
This file
Number 65.2
Position 2 of 6
Type Python
Size 19 KB
Lines 478 scripts/ stripe_cli.py
Python · 478 lines · 19 KB
15 typed into `stripe login --interactive` or `stripe config --set
16 test_mode_api_key` is used without --live. So before every call check_key
17 resolves the key the CLI would use (STRIPE_API_KEY, else the active profile's
18 test-mode key in the CLI config, per Stripe CLI v1.35 GetAPIKey() in
19 pkg/config/profile.go), refuses anything that is not a test key, and pins the
20 checked key for the call. Pinning goes through STRIPE_API_KEY itself, because
21 the CLI reads that variable ahead of every other source, including --api-key.
22 A config the wrapper cannot verify is refused, not passed through.
23
24 Three operations, all read-only except the single permitted test-clock advance:
25 stripe_cli.py read --path /v1/<resource> [--param k=v ...]
26 stripe_cli.py preview-invoice --subscription <subscription_id>
27 stripe_cli.py advance-clock --clock <clock_id> --days <n>
28
29 preview-invoice is a POST to /v1/invoices/create_preview, the only form of
30 Stripe's upcoming-invoice preview the API still accepts (GET
31 /v1/invoices/upcoming is deprecated), but it creates no Stripe object, so it is
32 a read. advance-clock is the single permitted write: advancing an
33 ALREADY-ATTACHED test clock. Everything else that creates, updates, or deletes
34 Stripe state is out of scope and unreachable through this script.
35
36 Exit codes: 0 ok; 1 the Stripe CLI failed, or preview-invoice got a Stripe API
37 error; 2 usage error; 20 disallowed path, malformed test clock id, or malformed
38 subscription id; 21 the key the CLI would use is live, missing, or cannot be
39 verified as a test key.
40 """
41 import argparse
42 import functools
43 import json
44 import os
45 import re
46 import subprocess
47 import sys
48 import time
49
50 EXIT_OK = 0
51 EXIT_CLI = 1
52 EXIT_USAGE = 2
53 EXIT_PATH = 20
54 EXIT_KEY = 21
55
56 LIVE_KEY_PREFIXES = ( "sk_live_" , "rk_live_" )
57 TEST_KEY_PREFIXES = ( "sk_test_" , "rk_test_" )
58 # Stripe CLI v1.35 test-mode key fields, in the order GetAPIKey() consults them:
59 # the legacy secret_key and api_key aliases win whenever they are present.
60 KEY_FIELDS = ( "secret_key" , "api_key" , "test_mode_api_key" )
61 # Stripe test clock ids are 'clock_' followed by an alphanumeric token. Anything
62 # else is rejected before it can be interpolated into a request path.
63 CLOCK_ID_PATTERN = re.compile( r "clock_ [ A-Za-z0-9 ] + " )
64 # Stripe subscription ids are 'sub_' followed by an alphanumeric token. This also
65 # rejects schedule ids (sub_sched_...), which create_preview takes under a
66 # different parameter.
67 SUBSCRIPTION_ID_PATTERN = re.compile( r "sub_ [ A-Za-z0-9 ] + " )
68 # A read path is /v1/ followed by one or more non-empty segments of letters,
69 # digits, and underscores, which covers every Stripe v1 resource and object id.
70 READ_PATH_PATTERN = re.compile( r "/v1 ( / [ A-Za-z0-9_ ] + ) + " )
71 CLOCK_POLL_DELAY = 2
72 CLOCK_POLL_LIMIT = 60
73 SECONDS_PER_DAY = 86400
74 # A single advance polls up to CLOCK_POLL_LIMIT * CLOCK_POLL_DELAY seconds per
75 # simulated day, so a batch larger than this cannot finish inside the Bash
76 # tool's 600000 ms (10 minute) ceiling before it is killed mid-advance. Reject
77 # the oversized batch up front rather than let it die partway in a state that
78 # is hard to diagnose. See SKILL.md, "The one permitted write".
79 MAX_ADVANCE_DAYS = 4
80
81
82 class GuardError ( Exception ):
83 """A request rejected by policy, carrying its documented exit code."""
84
85 def __init__ (self, code, message):
86 super (). __init__ (message)
87 self .code = code
88 self .message = message
89
90
91 def _key_error (message):
92 return GuardError( EXIT_KEY , message)
93
94
95 def _fold_case (table, where):
96 """Lower-case a TOML table's keys the way viper does, refusing collisions."""
97 folded = {}
98 for name, value in table.items():
99 lowered = name.lower()
100 if lowered in folded:
101 raise _key_error(
102 f " { where } has keys that differ only by case (' { lowered } '); the "
103 "Stripe CLI would merge them, so the key cannot be verified."
104 )
105 folded[lowered] = value
106 return folded
107
108
109 def config_path (env):
110 """The config file the Stripe CLI reads (pkg/config/config.go)."""
111 base = env.get( "XDG_CONFIG_HOME" )
112 if not base:
113 home = env.get( "HOME" ) or env.get( "USERPROFILE" )
114 if not home:
115 raise _key_error(
116 "cannot locate the Stripe CLI config: neither XDG_CONFIG_HOME, "
117 "HOME, nor USERPROFILE is set."
118 )
119 base = os.path.join(home, ".config" )
120 return os.path.join(base, "stripe" , "config.toml" )
121
122
123 def resolve_test_key (env):
124 """Return (key, source) for the key the Stripe CLI would use in test mode.
125
126 key is None when nothing is configured; source then says what was missing.
127 Every variable is read from env, and an empty value counts as unset.
128 """
129 key = (env.get( "STRIPE_API_KEY" ) or "" ).strip()
130 if key:
131 return key, "STRIPE_API_KEY"
132
133 path = config_path(env)
134 try :
135 import tomllib
136 except ImportError :
137 raise _key_error(
138 "Python 3.11+ is required to verify the Stripe CLI config; set "
139 "STRIPE_API_KEY to a test key or use a newer python3."
140 )
141 try :
142 with open (path, "rb" ) as handle:
143 data = tomllib.load(handle)
144 except FileNotFoundError :
145 return None , f "no Stripe CLI config at { path } "
146 except ( OSError , UnicodeDecodeError , tomllib.TOMLDecodeError) as err:
147 raise _key_error(
148 f "could not read the Stripe CLI config at { path } to verify the key "
149 f "is a test key: { err } "
150 )
151
152 top = _fold_case(data, path)
153 profile = env.get( "STRIPE_PROJECT_NAME" ) or top.get( "project-name" ) or "default"
154 if not isinstance (profile, str ):
155 raise _key_error( f "project-name in { path } is not a string." )
156 profile = profile.lower()
157
158 table = None
159 nested = top.get( "profiles" )
160 if isinstance (nested, dict ):
161 table = _fold_case(nested, f "[profiles] in { path } " ).get(profile)
162 if table is None :
163 table = top.get(profile)
164 if table is None :
165 return None , f "no ` { profile } ` profile in { path } "
166 if not isinstance (table, dict ):
167 raise _key_error( f "` { profile } ` in { path } is not a table." )
168
169 fields = _fold_case(table, f "the ` { profile } ` profile in { path } " )
170 for name in KEY_FIELDS :
171 if name in fields:
172 value = fields[name]
173 source = f "the ` { profile } ` profile's ` { name } ` in { path } "
174 if not isinstance (value, str ):
175 raise _key_error( f " { source } is not a string." )
176 return value.strip() or None , source
177 return None , f "the ` { profile } ` profile in { path } has no test-mode key"
178
179
180 def check_key (env):
181 """Return the test-mode key the Stripe CLI would use, or refuse with exit 21.
182
183 The CLI validates only a key's sk_/rk_ prefix, never its mode, so a live key
184 in STRIPE_API_KEY or in the profile's test-mode slot would reach live data.
185 This allows only sk_test_/rk_test_ keys, and fails closed on a missing or
186 unverifiable config. The caller pins the returned key via STRIPE_API_KEY,
187 so the key checked here is the key the CLI uses.
188 """
189 key, source = resolve_test_key(env)
190 if not key:
191 raise _key_error(
192 f "no Stripe test-mode key found ( { source } ); run 'stripe login' (or "
193 "set STRIPE_API_KEY to an sk_test_ key)."
194 )
195 if key.startswith( LIVE_KEY_PREFIXES ):
196 raise _key_error(
197 f " { source } holds a LIVE key. Every command would run against live "
198 "data. Restore a test-mode key (for example, 'stripe login') and retry."
199 )
200 if not key.startswith( TEST_KEY_PREFIXES ):
201 raise _key_error(
202 f " { source } is not a test-mode key (expected sk_test_ or rk_test_)."
203 )
204 return key
205
206
207 def check_path (path):
208 """Reject anything that is not a bare /v1/ resource path.
209
210 Like the clock and subscription id guards, this allows only the documented
211 shape rather than denying known-bad characters. That shuts out whitespace,
212 flag-like segments, `..` traversal (raw or percent-encoded as %2e), and a
213 `?` query string in one rule; query parameters go through --param.
214 """
215 if not READ_PATH_PATTERN .fullmatch(path or "" ):
216 raise GuardError(
217 EXIT_PATH ,
218 "--path must be a /v1/ resource path made of letters, digits, "
219 "underscores, and single slashes (pass parameters with --param), "
220 f "got: { path !r} " ,
221 )
222
223
224 def check_clock_id (clock_id):
225 """Reject anything that is not a bare Stripe test clock id.
226
227 Like check_path, this is an allowlist. The clock id is interpolated
228 straight into /v1/test_helpers/test_clocks/<id>[/advance], which is the
229 only Stripe write this wrapper permits. Flag injection is not
230 reachable, since argv is a list and no shell is involved, but a value like
231 'clock_1/../../customers' would traverse to another resource, and the id
232 ultimately comes from plan content that can originate in untrusted Jira
233 text. Allowing only the documented id shape closes that.
234 """
235 if not CLOCK_ID_PATTERN .fullmatch(clock_id or "" ):
236 raise GuardError(
237 EXIT_PATH ,
238 "--clock must be a Stripe test clock id of the form "
239 f "clock_<alphanumeric>, got: { clock_id !r} " ,
240 )
241
242
243 def check_subscription_id (subscription_id):
244 """Reject anything that is not a bare Stripe subscription id.
245
246 The id is sent as the value of a -d parameter rather than in the path, so
247 traversal is not the risk; but it comes from plan content that can originate
248 in untrusted Jira text, and allowing only the documented id shape keeps
249 anything else out of the request.
250 """
251 if not SUBSCRIPTION_ID_PATTERN .fullmatch(subscription_id or "" ):
252 raise GuardError(
253 EXIT_PATH ,
254 "--subscription must be a Stripe subscription id of the form "
255 f "sub_<alphanumeric>, got: { subscription_id !r} " ,
256 )
257
258
259 def build_read_argv (path, params):
260 """argv for a read. Built from scratch, so no caller flag is ever forwarded."""
261 argv = [ "stripe" , "get" , path]
262 for param in params or []:
263 argv.extend([ "-d" , param])
264 return argv
265
266
267 def build_get_clock_argv (clock_id):
268 return [ "stripe" , "get" , f "/v1/test_helpers/test_clocks/ { clock_id } " ]
269
270
271 def build_advance_argv (clock_id, frozen_time):
272 return [
273 "stripe" ,
274 "post" ,
275 f "/v1/test_helpers/test_clocks/ { clock_id } /advance" ,
276 "-d" ,
277 f "frozen_time= { frozen_time } " ,
278 ]
279
280
281 def build_preview_argv (subscription_id):
282 return [
283 "stripe" ,
284 "post" ,
285 "/v1/invoices/create_preview" ,
286 "-d" ,
287 f "subscription= { subscription_id } " ,
288 ]
289
290
291 def run_cli (argv, key):
292 """Execute the Stripe CLI with key pinned and return stdout.
293
294 Raises GuardError on failure.
295 """
296 try :
297 completed = subprocess.run(
298 argv,
299 capture_output = True ,
300 text = True ,
301 check = False ,
302 env = { ** os.environ, "STRIPE_API_KEY" : key},
303 )
304 except FileNotFoundError :
305 raise GuardError(
306 EXIT_CLI ,
307 "the 'stripe' CLI is not installed or not on PATH. Install it and run "
308 "'stripe login'; do not improvise another data source." ,
309 )
310 if completed.returncode != 0 :
311 raise GuardError(
312 EXIT_CLI ,
313 f "stripe CLI failed ( { completed.returncode } ): { completed.stderr.strip() } " ,
314 )
315 return completed.stdout
316
317
318 def _run_json (argv, run):
319 """Run the CLI, parse its JSON, and surface an API-level error as GuardError.
320
321 The Stripe CLI returns exit 0 with a top-level {"error": {...}} body for
322 API-level failures (a bad id, a rejected advance), so run_cli does not raise
323 on them. Left unchecked on an internal clock call, a rejected advance would
324 masquerade as success: the loop would see the clock still 'ready' at its old
325 frozen_time and report an advance that never happened. Detect the error body
326 here and raise EXIT_CLI so the failure is loud. preview-invoice uses it too,
327 so a rejected preview is never printed as if it were an invoice. The read
328 command keeps its pass-through behavior on purpose (see SKILL.md,
329 "Interpreting responses").
330 """
331 body = run(argv)
332 try :
333 data = json.loads(body)
334 except ValueError as err:
335 raise GuardError( EXIT_CLI , f "unparseable Stripe response: { err } " )
336 if isinstance (data, dict ) and isinstance (data.get( "error" ), dict ):
337 error = data[ "error" ]
338 detail = error.get( "message" ) or error.get( "type" ) or "unknown error"
339 raise GuardError( EXIT_CLI , f "Stripe API error: { detail } " )
340 return data
341
342
343 def _clock (clock_id, run):
344 return _run_json(build_get_clock_argv(clock_id), run)
345
346
347 def preview_invoice (subscription_id, run):
348 """Return Stripe's preview of the subscription's next invoice.
349
350 The preview includes a pending subscription schedule phase change, so a
351 schedule-managed subscription needs no separate schedule parameter.
352 """
353 check_subscription_id(subscription_id)
354 return _run_json(build_preview_argv(subscription_id), run)
355
356
357 def _advanced_to (clock, frozen):
358 """True once the clock is 'ready' with its frozen_time at or past frozen."""
359 if clock.get( "status" ) != "ready" :
360 return False
361 try :
362 return int (clock.get( "frozen_time" )) >= frozen
363 except ( TypeError , ValueError ):
364 return False
365
366
367 def advance_clock (clock_id, days, run, sleep):
368 """Advance an already-attached test clock one day at a time.
369
370 One day per step is deliberate: Stripe's smart retry policy fires a payment
371 retry per simulated day, which is what drives a subscription to unpaid after
372 eight failures. Each step waits for status to return to 'ready' before the
373 next, because Stripe rejects an advance on a clock that is still advancing.
374 A step is done only once the clock is 'ready' AND its frozen_time has
375 reached the requested time: a clock read just after the POST can still
376 report 'ready' at its old frozen_time, and gating on status alone would
377 send the next day's advance while this one is still being applied.
378 """
379 check_clock_id(clock_id)
380 response = _clock(clock_id, run)
381 try :
382 frozen = int (response[ "frozen_time" ])
383 except ( KeyError , TypeError , ValueError ):
384 # A response missing, nulling, or non-numerically typing frozen_time
385 # would otherwise escape as a traceback: main() only catches GuardError,
386 # so the operator would lose the documented exit code.
387 raise GuardError(
388 EXIT_CLI ,
389 f "test clock { clock_id } response has no usable 'frozen_time': "
390 f " { response !r} " ,
391 )
392 completed = 0
393 for _ in range (days):
394 frozen += SECONDS_PER_DAY
395 try :
396 _run_json(build_advance_argv(clock_id, frozen), run)
397 for _attempt in range ( CLOCK_POLL_LIMIT ):
398 if _advanced_to(_clock(clock_id, run), frozen):
399 break
400 sleep( CLOCK_POLL_DELAY )
401 else :
402 raise GuardError(
403 EXIT_CLI ,
404 f "test clock { clock_id } did not return to 'ready' at "
405 f "frozen_time= { frozen } after "
406 f " { CLOCK_POLL_LIMIT * CLOCK_POLL_DELAY } s" ,
407 )
408 except GuardError as err:
409 # A multi-day advance can run long enough to be interrupted (see
410 # the Bash-timeout note in SKILL.md). Annotate whatever failed with
411 # how far it got, so the caller can re-read frozen_time and resume
412 # instead of reissuing the full day count, and preserve the
413 # original exit code.
414 raise GuardError(
415 err.code,
416 f " { err.message } (advanced { completed } of { days } day(s) before "
417 f "failing; last attempted frozen_time= { frozen } )" ,
418 ) from err
419 completed += 1
420 return frozen
421
422
423 def main (argv, env):
424 parser = argparse.ArgumentParser(
425 prog = "stripe_cli.py" ,
426 description = "Policy-guarded Stripe CLI access (test mode only)." ,
427 )
428 sub = parser.add_subparsers( dest = "command" , required = True )
429
430 read = sub.add_parser( "read" , help = "GET a /v1/ resource" )
431 read.add_argument( "--path" , required = True )
432 read.add_argument( "--param" , action = "append" , default = [])
433
434 preview_parser = sub.add_parser(
435 "preview-invoice" , help = "preview a subscription's next invoice"
436 )
437 preview_parser.add_argument( "--subscription" , required = True )
438
439 advance = sub.add_parser( "advance-clock" , help = "advance an attached test clock" )
440 advance.add_argument( "--clock" , required = True )
441 advance.add_argument( "--days" , type = int , required = True )
442
443 try :
444 args = parser.parse_args(argv)
445 except SystemExit as exc:
446 return exc.code if isinstance (exc.code, int ) else EXIT_USAGE
447
448 try :
449 run = functools.partial(run_cli, key = check_key(env))
450 if args.command == "read" :
451 check_path(args.path)
452 sys.stdout.write(run(build_read_argv(args.path, args.param)))
453 elif args.command == "preview-invoice" :
454 preview = preview_invoice(args.subscription, run)
455 print (json.dumps(preview, indent = 2 ))
456 else :
457 check_clock_id(args.clock)
458 if args.days < 1 :
459 raise GuardError( EXIT_USAGE , "--days must be at least 1" )
460 if args.days > MAX_ADVANCE_DAYS :
461 raise GuardError(
462 EXIT_USAGE ,
463 f "--days must be at most { MAX_ADVANCE_DAYS } : a single call "
464 "cannot advance more simulated days than fit inside the Bash "
465 "tool's 10-minute ceiling before it is killed mid-advance. "
466 f "Split it into batches of at most { MAX_ADVANCE_DAYS } days, "
467 "re-reading the clock status between them." ,
468 )
469 frozen = advance_clock(args.clock, args.days, run, time.sleep)
470 print ( f "test clock { args.clock } advanced to frozen_time= { frozen } " )
471 except GuardError as err:
472 print ( f "ERROR: { err.message } " , file = sys.stderr)
473 return err.code
474 return EXIT_OK
475
476
477 if __name__ == "__main__" :
478 sys.exit(main(sys.argv[ 1 :], os.environ))