c6dc4286e2
Assistant: codex Assistant-Model: gpt-5.6-luna Assistant-Session: 01a07ff8-19d0-7820-b4d0-1353833cb7fc
366 lines
22 KiB
Python
366 lines
22 KiB
Python
#!/usr/bin/env python3
|
|
"""Silent, attended first provision and verification of CCR-2026-0017/0018."""
|
|
from __future__ import annotations
|
|
import argparse
|
|
import hashlib
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import secrets
|
|
import time
|
|
from datetime import datetime, timezone
|
|
import yaml
|
|
|
|
from state_hub_preflight_lane import ROOT, LaneError, assert_cluster, bao, command, data, revoke
|
|
|
|
LANES = (
|
|
('CCR-2026-0017', 'secrets-engine', 'approval-client', 'keycape-secrets-engine-approval'),
|
|
('CCR-2026-0018', 'approval-engine', 'operator-client', 'keycape-approval-engine-operator'),
|
|
)
|
|
BOUNDARY = 'sys/policies/acl/agent-high-risk-boundary'
|
|
|
|
|
|
def require(condition, reason):
|
|
if not condition:
|
|
raise LaneError(reason)
|
|
|
|
|
|
def contracts():
|
|
spec = importlib.util.spec_from_file_location('credential_change', ROOT / 'scripts/credential-change.py')
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
result = []
|
|
for ref, workload, purpose, suffix in LANES:
|
|
path = module.resolve_ccr(ref)
|
|
ccr, errors, _ = module.validate_ccr(path)
|
|
require(not errors and ccr['status'] in {'approved', 'applied', 'verified', 'active'}, 'approved_ccrs_required')
|
|
for role in ('platform-operator', 'key-cape-owner'):
|
|
require(any(x.get('decision') == 'approved' and role in x.get('reviewer', '')
|
|
and 'explicit session approval' in x.get('reviewer', '')
|
|
for x in ccr['review']['comments']), 'both_named_reviews_required')
|
|
expected_path = f'platform/workloads/{workload}/{purpose}'
|
|
policy = 'workload-kv-read-' + suffix
|
|
role = 'external-secrets-' + suffix
|
|
expected_role = {'bound_service_account_names': ['external-secrets'],
|
|
'bound_service_account_namespaces': ['external-secrets'],
|
|
'policies': policy, 'ttl': '15m'}
|
|
require(ccr['openbao']['kv_path'] == expected_path and ccr['openbao']['fields'] == ['CLIENT_SECRET']
|
|
and ccr['openbao']['policy_name'] == policy
|
|
and ccr['openbao']['auth']['role'] == role
|
|
and ccr['openbao']['auth']['mount'] == 'kubernetes'
|
|
and ccr['openbao']['auth']['bound_claims_confirmed'] is True
|
|
and module.auth_payload(ccr) == expected_role, 'exact_approved_contract_required')
|
|
hcl = module.generated_policy_hcl(ccr)
|
|
require((ROOT / ccr['openbao']['policy_file']).read_text() == hcl, 'policy_source_drift')
|
|
result.append({'ccr': ref, 'kv': expected_path.replace('platform/', 'platform/data/', 1),
|
|
'metadata': expected_path.replace('platform/', 'platform/metadata/', 1),
|
|
'policy': policy, 'hcl': hcl, 'role': role, 'role_payload': expected_role,
|
|
'store': 'openbao-' + suffix, 'secret': suffix + '-client',
|
|
'source_sha256': hashlib.sha256(path.read_bytes()).hexdigest()})
|
|
return result
|
|
|
|
|
|
|
|
def resume_provision(lanes, receipt, prior_path):
|
|
prior = json.loads(Path(prior_path).read_text())
|
|
require(prior.get('status') == 'failed' and prior.get('verifier_delivery_disabled_custody_versions_retained')
|
|
and prior.get('keycape', {}).get('compatible_pair_restored'), 'verified_partial_activation_receipt_required')
|
|
old_rows = prior['lanes']
|
|
require(len(old_rows) == len(lanes), 'partial_receipt_lane_mismatch')
|
|
for lane, old in zip(lanes, old_rows):
|
|
require(old['ccr'] == lane['ccr'] and (old.get('custody_seeded') or old.get('existing_version_reused')) and old.get('kv_version') == 1,
|
|
'initial_version_provenance_required')
|
|
metadata = read_optional(lane['metadata'])
|
|
require(metadata is not None and metadata['current_version'] == 1
|
|
and not metadata['versions']['1'].get('destroyed') and not metadata['versions']['1'].get('deletion_time'),
|
|
'initial_custody_version_changed')
|
|
role = read_optional('auth/kubernetes/role/' + lane['role'])
|
|
require(role is not None and role.get('token_policies') == []
|
|
and role_matches(dict(role, token_policies=[lane['policy']]), lane), 'disabled_role_drift')
|
|
require(read_optional('sys/policies/acl/' + lane['policy'])['policy'] == lane['hcl'], 'resume_policy_drift')
|
|
require(all(re.search(r'path\s+"' + re.escape(path) + r'"\s*\{\s*capabilities\s*=\s*\["deny"\]\s*\}', read_optional(BOUNDARY)['policy'])
|
|
for path in (lane['kv'], lane['metadata'])), 'resume_boundary_drift')
|
|
# Only reattach the same reviewed reader roles. No KV write, import or rotation.
|
|
for lane, old in zip(lanes, old_rows):
|
|
bao(['write', 'auth/kubernetes/role/' + lane['role'], '-'], payload=lane['role_payload'])
|
|
require(role_matches(read_optional('auth/kubernetes/role/' + lane['role']), lane), 'resumed_role_readback_failed')
|
|
receipt['lanes'].append({'ccr': lane['ccr'], 'source_sha256': lane['source_sha256'],
|
|
'custody_seeded': False, 'existing_version_reused': True, 'kv_version': 1,
|
|
'initial_request_id': old.get('initial_request_id', old.get('request_id')), 'policy_applied': False, 'role_applied': True})
|
|
receipt['custody_versions_unchanged'] = True
|
|
|
|
|
|
def read_optional(path):
|
|
result = bao(['read', '-format=json', path], allow_failure=True)
|
|
if result.returncode == 0:
|
|
return data(result)['data']
|
|
require(b'404' in result.stderr or b'No value found' in result.stderr, 'absence_not_proven')
|
|
return None
|
|
|
|
|
|
def role_matches(actual, lane):
|
|
return (actual.get('bound_service_account_names') == ['external-secrets']
|
|
and actual.get('bound_service_account_namespaces') == ['external-secrets']
|
|
and not actual.get('bound_service_account_namespace_selector')
|
|
and actual.get('token_policies') == [lane['policy']]
|
|
and actual.get('token_ttl') == 900 and not actual.get('token_period')
|
|
and not actual.get('audience') and not actual.get('token_no_default_policy')
|
|
and not actual.get('token_bound_cidrs') and not actual.get('token_num_uses'))
|
|
|
|
|
|
def provision(lanes, receipt):
|
|
# Complete all drift/absence checks before the first write.
|
|
for lane in lanes:
|
|
require(read_optional(lane['metadata']) is None, 'existing_custody_requires_resume_or_rotation_review')
|
|
old_policy = read_optional('sys/policies/acl/' + lane['policy'])
|
|
require(old_policy is None or old_policy['policy'] == lane['hcl'], 'policy_drift')
|
|
old_role = read_optional('auth/kubernetes/role/' + lane['role'])
|
|
require(old_role is None or role_matches(old_role, lane), 'role_drift')
|
|
# The new values must not become readable through a generic coding-agent grant.
|
|
boundary = read_optional(BOUNDARY)
|
|
require(boundary is not None, 'coding_agent_boundary_required')
|
|
current = boundary['policy']
|
|
additions = ''
|
|
for lane in lanes:
|
|
for path in (lane['kv'], lane['metadata']):
|
|
if '"' + path + '"' not in current:
|
|
additions += 'path "' + path + '" { capabilities = ["deny"] }\n'
|
|
else:
|
|
require(re.search(r'path\s+"' + re.escape(path) + r'"\s*\{\s*capabilities\s*=\s*\["deny"\]\s*\}', current), 'boundary_path_drift')
|
|
if additions:
|
|
require(read_optional(BOUNDARY)['policy'] == current, 'boundary_revision_changed')
|
|
bao(['write', BOUNDARY, '-'], payload={'policy': current + '\n' + additions})
|
|
require(read_optional(BOUNDARY)['policy'] == current + '\n' + additions, 'boundary_readback_failed')
|
|
receipt['coding_agent_boundary_extended_only_to_new_paths'] = bool(additions)
|
|
for lane in lanes:
|
|
row = {'ccr': lane['ccr'], 'source_sha256': lane['source_sha256'], 'policy_applied': False,
|
|
'role_applied': False, 'custody_seeded': False}
|
|
receipt['lanes'].append(row)
|
|
bao(['write', 'sys/policies/acl/' + lane['policy'], '-'], payload={'policy': lane['hcl']})
|
|
row['policy_applied'] = True
|
|
bao(['write', 'auth/kubernetes/role/' + lane['role'], '-'], payload=lane['role_payload'])
|
|
row['role_applied'] = True
|
|
require(read_optional('sys/policies/acl/' + lane['policy'])['policy'] == lane['hcl'], 'policy_readback_failed')
|
|
require(role_matches(read_optional('auth/kubernetes/role/' + lane['role']), lane), 'role_readback_failed')
|
|
for lane, row in zip(lanes, receipt['lanes']):
|
|
result = data(bao(['write', '-format=json', lane['kv'], '-'],
|
|
payload={'options': {'cas': 0}, 'data': {'CLIENT_SECRET': secrets.token_urlsafe(48)}}))
|
|
require(result['data']['version'] == 1, 'unexpected_initial_version')
|
|
row.update(custody_seeded=True, kv_version=1, request_id=result.get('request_id'))
|
|
receipt['initial_values_generation'] = 'independent CSPRNG 48-byte values; memory-to-OpenBao stdin only; CAS=0'
|
|
|
|
|
|
def require_denied(result):
|
|
require(result.returncode != 0 and b'403' in result.stderr and b'permission denied' in result.stderr.lower(), 'denial_not_proven')
|
|
|
|
|
|
def verify_native(kube, lanes, receipt):
|
|
for lane, row in zip(lanes, receipt['lanes']):
|
|
jwt = command(kube + ['-n', 'external-secrets', 'create', 'token', 'external-secrets', '--duration=10m']).stdout.decode().strip()
|
|
auth = data(bao(['write', '-format=json', 'auth/kubernetes/login', '-'], payload={'role': lane['role'], 'jwt': jwt}))['auth']
|
|
token = auth['client_token']
|
|
try:
|
|
require(set(auth['token_policies']) == {lane['policy'], 'default'} and 0 < auth['lease_duration'] <= 900, 'issued_role_or_ttl_mismatch')
|
|
native = data(bao(['read', '-format=json', lane['kv']], token=token))['data']
|
|
require(set(native['data']) == {'CLIENT_SECRET'} and len(native['data']['CLIENT_SECRET']) == 64, 'custody_shape_mismatch')
|
|
sibling = next(x['kv'] for x in lanes if x['ccr'] != lane['ccr'])
|
|
require_denied(bao(['read', '-format=json', sibling], token=token, allow_failure=True))
|
|
require_denied(bao(['list', '-format=json', 'platform/metadata/workloads'], token=token, allow_failure=True))
|
|
row.update(native_reader_verified=True, cross_path_denied=True, parent_listing_denied=True, auth_ttl=auth['lease_duration'])
|
|
finally:
|
|
revoke(token)
|
|
require_denied(bao(['token', 'lookup', '-format=json'], token=token, allow_failure=True))
|
|
row['reader_revocation_verified'] = True
|
|
bad_jwt = command(kube + ['-n', 'external-secrets', 'create', 'token', 'default', '--duration=10m']).stdout.decode().strip()
|
|
bad = bao(['write', '-format=json', 'auth/kubernetes/login', '-'], payload={'role': lane['role'], 'jwt': bad_jwt}, allow_failure=True)
|
|
if bad.returncode == 0:
|
|
revoke(data(bad)['auth']['client_token'])
|
|
raise LaneError('wrong_service_account_authenticated')
|
|
require(b'403' in bad.stderr and b'service account' in bad.stderr.lower(), 'wrong_service_account_denial_inconclusive')
|
|
row['wrong_service_account_denied'] = True
|
|
boundary_child = data(bao(['token', 'create', '-format=json', '-policy=' + lane['policy'], '-policy=agent-high-risk-boundary', '-no-default-policy', '-ttl=60s']))['auth']['client_token']
|
|
try:
|
|
for path in (lane['kv'], lane['metadata']):
|
|
require_denied(bao(['read', '-format=json', path], token=boundary_child, allow_failure=True))
|
|
row['coding_agent_deny_wins'] = True
|
|
finally:
|
|
revoke(boundary_child)
|
|
|
|
|
|
def deliver(kube, lanes, receipt):
|
|
base = ROOT / 'argocd/platform-addons/openbao-secretstore'
|
|
command(kube + ['apply', '-f', str(base / 'openbao-keycape-approval-clients.clustersecretstore.yaml')])
|
|
for lane in lanes:
|
|
command(kube + ['wait', '--for=condition=Ready', 'clustersecretstore/' + lane['store'], '--timeout=45s'])
|
|
command(kube + ['apply', '-f', str(base / 'keycape-approval-clients.externalsecrets.yaml')])
|
|
for lane, row in zip(lanes, receipt['lanes']):
|
|
command(kube + ['-n', 'sso', 'wait', '--for=condition=Ready', 'externalsecret/' + lane['secret'], '--timeout=45s'])
|
|
es = data(command(kube + ['-n', 'sso', 'get', 'externalsecret', lane['secret'], '-o', 'json']))
|
|
secret = data(command(kube + ['-n', 'sso', 'get', 'secret', lane['secret'], '-o', 'json']))
|
|
import base64
|
|
native = data(bao(['read', '-format=json', lane['kv']]))['data']
|
|
require(set(secret['data']) == {'client-secret'} and base64.b64decode(secret['data']['client-secret']).decode() == native['data']['CLIENT_SECRET'], 'eso_delivery_mismatch')
|
|
require(any(x['uid'] == es['metadata']['uid'] for x in secret['metadata'].get('ownerReferences', [])), 'eso_does_not_own_secret')
|
|
row.update(store_ready=True, external_secret_ready=True, delivery_matches=True,
|
|
secret_uid=secret['metadata']['uid'], secret_resource_version=secret['metadata']['resourceVersion'])
|
|
|
|
|
|
|
|
def verify_namespace_boundaries(kube, lanes, receipt):
|
|
# A unique, exclusively created namespace isolates the negative probes.
|
|
ns = 'keycape-custody-check-' + secrets.token_hex(5)
|
|
created = data(command(kube + ['create', 'namespace', ns, '-o', 'json']))
|
|
try:
|
|
command(kube + ['-n', ns, 'create', 'serviceaccount', 'external-secrets'])
|
|
jwt = command(kube + ['-n', ns, 'create', 'token', 'external-secrets', '--duration=10m']).stdout.decode().strip()
|
|
for lane, row in zip(lanes, receipt['lanes']):
|
|
bad = bao(['write', '-format=json', 'auth/kubernetes/login', '-'],
|
|
payload={'role': lane['role'], 'jwt': jwt}, allow_failure=True)
|
|
if bad.returncode == 0:
|
|
revoke(data(bad)['auth']['client_token'])
|
|
raise LaneError('wrong_namespace_authenticated')
|
|
require(b'403' in bad.stderr and b'namespace' in bad.stderr.lower(), 'wrong_namespace_denial_inconclusive')
|
|
row['wrong_namespace_denied'] = True
|
|
probe = {'apiVersion': 'external-secrets.io/v1', 'kind': 'ExternalSecret',
|
|
'metadata': {'name': lane['secret'], 'namespace': ns},
|
|
'spec': {'refreshInterval': '15s',
|
|
'secretStoreRef': {'kind': 'ClusterSecretStore', 'name': lane['store']},
|
|
'target': {'name': lane['secret'], 'creationPolicy': 'Owner'},
|
|
'data': [{'secretKey': 'client-secret', 'remoteRef': {
|
|
'key': lane['kv'].removeprefix('platform/data/'), 'property': 'CLIENT_SECRET'}}]}}
|
|
command(kube + ['create', '-f', '-'], payload=probe)
|
|
denied = False
|
|
for _ in range(20):
|
|
result = data(command(kube + ['-n', ns, 'get', 'externalsecret', lane['secret'], '-o', 'json']))
|
|
conditions = result.get('status', {}).get('conditions', [])
|
|
denied = any(c.get('type') == 'Ready' and c.get('status') == 'False'
|
|
and c.get('reason') == 'SecretSyncedError' for c in conditions)
|
|
if denied:
|
|
events = data(command(kube + ['-n', ns, 'get', 'events',
|
|
'--field-selector=involvedObject.uid=' + result['metadata']['uid'], '-o', 'json']))
|
|
denied = any('not allowed' in e.get('message', '').lower()
|
|
and 'namespace' in e.get('message', '').lower() for e in events['items'])
|
|
if denied:
|
|
break
|
|
time.sleep(2)
|
|
require(denied, 'outside_namespace_store_denial_inconclusive')
|
|
absent = command(kube + ['-n', ns, 'get', 'secret', lane['secret'], '-o', 'json'], allow_failure=True)
|
|
require(absent.returncode != 0 and b'NotFound' in absent.stderr, 'outside_namespace_secret_not_absent')
|
|
row['outside_namespace_store_denied'] = True
|
|
finally:
|
|
command(kube + ['delete', '--raw=/api/v1/namespaces/' + ns, '-f', '/dev/stdin'],
|
|
payload={'apiVersion': 'v1', 'kind': 'DeleteOptions',
|
|
'preconditions': {'uid': created['metadata']['uid']}})
|
|
receipt['namespace_probe_cleanup_requested'] = True
|
|
|
|
|
|
|
|
def load_rollout():
|
|
path = Path('/home/worsch/net-kingdom/sso-mfa/k8s/keycape/approval-clients-rollout.py')
|
|
spec = importlib.util.spec_from_file_location('keycape_rollout', path)
|
|
module = importlib.util.module_from_spec(spec); spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
def deactivate_delivery(kube, lanes, receipt):
|
|
# Config/image restoration runs in the owner helper before this call.
|
|
for lane in lanes:
|
|
existing = command(kube + ['-n', 'sso', 'get', 'externalsecret', lane['secret'], '-o', 'json'], allow_failure=True)
|
|
if existing.returncode == 0:
|
|
obj = data(existing)
|
|
require(obj['spec']['secretStoreRef'] == {'kind': 'ClusterSecretStore', 'name': lane['store']}, 'rollback_external_secret_drift')
|
|
command(kube + ['delete', '--raw=/apis/external-secrets.io/v1/namespaces/sso/externalsecrets/' + lane['secret'], '-f', '/dev/stdin'],
|
|
payload={'apiVersion': 'v1', 'kind': 'DeleteOptions', 'preconditions': {'uid': obj['metadata']['uid']}})
|
|
else:
|
|
require(b'NotFound' in existing.stderr, 'rollback_external_secret_absence_unproven')
|
|
role = read_optional('auth/kubernetes/role/' + lane['role'])
|
|
if role is not None:
|
|
require(role_matches(role, lane), 'rollback_auth_role_drift')
|
|
payload = dict(lane['role_payload'], policies=[])
|
|
bao(['write', 'auth/kubernetes/role/' + lane['role'], '-'], payload=payload)
|
|
receipt['verifier_delivery_disabled_custody_versions_retained'] = True
|
|
|
|
|
|
def run(args, receipt):
|
|
require(Path.home().parent.name == '.warden-attended-login' and not os.environ.get('BAO_TOKEN') and not os.environ.get('VAULT_TOKEN'), 'attended_warden_envelope_required')
|
|
lanes = contracts()
|
|
kube = ['kubectl', '--kubeconfig', args.kubeconfig]
|
|
assert_cluster(kube)
|
|
identity = data(bao(['token', 'lookup', '-format=json']))['data']
|
|
require('platform-admin' in identity['policies'] and 'root' not in identity['policies'], 'attended_platform_admin_required')
|
|
env = os.environ.copy(); env['KUBECONFIG'] = args.kubeconfig
|
|
check = data(command(['python3', '-B', '/home/worsch/net-kingdom/sso-mfa/k8s/keycape/openbao-client-config.py', 'issuer-check-live'], env=env))
|
|
require(check['issuer_matches'] and check['verified_issuer'] == 'https://auth.coulomb.social', 'verified_issuer_pin_required')
|
|
receipt['issuer_pin_revision'] = check['before']['resource_version']
|
|
rollout = load_rollout() if args.action in {'activate', 'resume-activate'} else None
|
|
if rollout:
|
|
rollout.verify_artifact()
|
|
# Exercise config construction and API admission before any custody write.
|
|
secret = rollout.get(kube, 'secret', 'keycape-config')
|
|
dep = rollout.get(kube, 'deployment', 'keycape')
|
|
config = rollout.replacement(secret, rollout.registrations())
|
|
candidate = rollout.candidate_spec(dep)
|
|
rollout.patch_object(kube, 'secret', secret, '/data/config.yaml', config, dry=True)
|
|
rollout.patch_object(kube, 'deployment', dep, '/spec', candidate, dry=True)
|
|
require(args.recovery and not Path(args.recovery).exists(), 'unique_recovery_path_required')
|
|
for lane in lanes:
|
|
for kind in ('secret', 'externalsecret'):
|
|
existing = command(kube + ['-n', 'sso', 'get', kind, lane['secret'], '-o', 'json'], allow_failure=True)
|
|
require(existing.returncode != 0 and b'NotFound' in existing.stderr, 'existing_delivery_requires_reconciliation')
|
|
try:
|
|
receipt['phase'] = 'provision'
|
|
if args.action in {'provision', 'activate'}:
|
|
provision(lanes, receipt)
|
|
elif args.action == 'resume-activate':
|
|
require(args.prior_receipt, 'prior_receipt_required')
|
|
resume_provision(lanes, receipt, args.prior_receipt)
|
|
else:
|
|
receipt['lanes'] = [{'ccr': lane['ccr']} for lane in lanes]
|
|
receipt['phase'] = 'native_verification'
|
|
verify_native(kube, lanes, receipt)
|
|
receipt['phase'] = 'eso_delivery'
|
|
deliver(kube, lanes, receipt)
|
|
receipt['phase'] = 'namespace_boundaries'
|
|
verify_namespace_boundaries(kube, lanes, receipt)
|
|
receipt['status'] = 'custody_and_eso_verified_pending_keycape_rollout'
|
|
if rollout:
|
|
receipt['phase'] = 'keycape_rollout'
|
|
receipt['keycape'] = {'existing_human_login_before': True}
|
|
rollout.rollout(kube, receipt['keycape'], Path(args.recovery))
|
|
receipt['status'] = 'custody_and_service_acceptance_passed_pending_fresh_human_login'
|
|
receipt['keycape_rollout_completed'] = True
|
|
receipt['phase'] = 'awaiting_fresh_human_login'
|
|
except Exception:
|
|
receipt['status'] = 'failed'
|
|
if rollout and (not receipt.get('keycape') or receipt['keycape'].get('compatible_pair_restored')):
|
|
deactivate_delivery(kube, lanes, receipt)
|
|
raise
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('action', choices=['provision', 'verify', 'activate', 'resume-activate'])
|
|
parser.add_argument('--kubeconfig', required=True)
|
|
parser.add_argument('--receipt', required=True)
|
|
parser.add_argument('--recovery')
|
|
parser.add_argument('--prior-receipt')
|
|
args = parser.parse_args()
|
|
receipt = {'schema': 'platform.keycape-approval-custody.v1', 'status': 'failed', 'lanes': [],
|
|
'started_at': datetime.now(timezone.utc).isoformat(), 'credential_values_emitted': False,
|
|
'client_side_read_admitted': False, 'keycape_rollout_completed': False}
|
|
fd = os.open(args.receipt, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
|
|
try:
|
|
run(args, receipt)
|
|
except Exception as exc:
|
|
receipt['error'] = str(exc) if isinstance(exc, LaneError) else 'contained_operation_failed'
|
|
finally:
|
|
receipt['finished_at'] = datetime.now(timezone.utc).isoformat()
|
|
with os.fdopen(fd, 'w') as out:
|
|
json.dump(receipt, out, indent=2); out.write('\n')
|
|
return 0 if receipt['status'] != 'failed' else 1
|
|
|
|
|
|
if __name__ == '__main__':
|
|
raise SystemExit(main())
|