4b34c239bc
CCR-2026-0026/0027 are approved for the create-only informed-decision client. Live KeyCape registration and CAS=0 custody stay in the contained helper; no sitting POST. Assistant: grok Assistant-Session: 01a0a23b-3bf0-7341-b4e5-9dc05f72573a
101 lines
6.2 KiB
Python
101 lines
6.2 KiB
Python
"""Silent attended provisioning of the informed-decision sitting requester."""
|
|
import copy, json, os, secrets, sys, time
|
|
from pathlib import Path
|
|
from datetime import datetime, timezone
|
|
ROOT = Path('/home/worsch/railiance-platform')
|
|
sys.path.insert(0, str(ROOT / 'scripts'))
|
|
from state_hub_preflight_lane import command, bao, data, assert_cluster, LaneError
|
|
from keycape_approval_custody import require, read_optional, BOUNDARY
|
|
import importlib.util
|
|
KUBE = ['kubectl', '--kubeconfig', '/home/worsch/.kube/config-railiance01']
|
|
KV = 'platform/data/workloads/informed-decision/sitting-requester'
|
|
META = 'platform/metadata/workloads/informed-decision/sitting-requester'
|
|
CLIENT = json.loads(Path('/home/worsch/key-cape/docs/sitting-requester-registration.json').read_text())
|
|
RECEIPT = ROOT / 'docs/evidence/2026-09-15-sitting-requester-provision.json'
|
|
CCRS = ('CCR-2026-0026', 'CCR-2026-0027')
|
|
ENV_NAME = 'KEYCAPE_INFORMED_DECISION_SITTING_REQUESTER_CLIENT_SECRET'
|
|
ESO = ROOT / 'argocd/platform-addons/openbao-secretstore/sitting-requester.yaml'
|
|
|
|
|
|
def load(name, path):
|
|
spec = importlib.util.spec_from_file_location(name, path)
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
rollout = load('requester_rollout', '/home/worsch/net-kingdom/sso-mfa/k8s/keycape/approval-clients-rollout.py')
|
|
rollout.IDS = ('informed-decision-sitting-requester',)
|
|
|
|
|
|
def contract():
|
|
require(CLIENT['clientId'] == 'informed-decision-sitting-requester', 'requester_contract_drift')
|
|
require(CLIENT['allowedScopes'] == ['approval:create'], 'requester_contract_drift')
|
|
require(CLIENT['serviceSubject'] == 'informed-decision', 'requester_contract_drift')
|
|
require(CLIENT['audience'] == 'approval-engine', 'requester_contract_drift')
|
|
require(CLIENT['tenant'] == 'tenant:platform', 'requester_contract_drift')
|
|
require('approval:consume' not in CLIENT['allowedScopes'], 'consume_scope_forbidden')
|
|
require('approval:approve' not in CLIENT['allowedScopes'], 'approve_scope_forbidden')
|
|
|
|
|
|
def run(receipt):
|
|
require(Path.home().parent.name == '.warden-attended-login' and not os.getenv('BAO_TOKEN') and not os.getenv('VAULT_TOKEN'), 'attended_envelope_required')
|
|
assert_cluster(KUBE)
|
|
identity = data(bao(['token', 'lookup', '-format=json']))['data']
|
|
require('platform-admin' in identity['policies'] and 'root' not in identity['policies'], 'attended_operator_required')
|
|
contract()
|
|
require(read_optional(META) is None, 'existing_sitting_requester_custody_requires_reconciliation')
|
|
for ref in CCRS:
|
|
command(['python3', str(ROOT / 'scripts/credential-change.py'), 'applier-dry-run', ref, '--json'])
|
|
secret = rollout.get(KUBE, 'secret', 'keycape-config')
|
|
deployment = rollout.get(KUBE, 'deployment', 'keycape')
|
|
config = rollout.replacement(secret, [CLIENT])
|
|
spec = copy.deepcopy(deployment['spec'])
|
|
require(spec.get('replicas', 1) == 1 and spec['strategy']['type'] == 'Recreate', 'single_recreate_keycape_required')
|
|
container = next(c for c in spec['template']['spec']['containers'] if c['name'] == 'keycape')
|
|
image = container['image']
|
|
require('@sha256:' in image, 'pinned_keycape_image_required')
|
|
require(not any(e['name'] == ENV_NAME for e in container.setdefault('env', [])), 'existing_sitting_requester_env')
|
|
container['env'].append({'name': ENV_NAME, 'valueFrom': {'secretKeyRef': {'name': 'keycape-informed-decision-sitting-requester-client', 'key': 'client-secret'}}})
|
|
rollout.patch_object(KUBE, 'secret', secret, '/data/config.yaml', config, dry=True)
|
|
rollout.patch_object(KUBE, 'deployment', deployment, '/spec', spec, dry=True)
|
|
receipt['phase'] = 'metadata_preflight_passed'
|
|
boundary = read_optional(BOUNDARY)
|
|
require(boundary is not None, 'agent_boundary_required')
|
|
additions = ''.join('path "' + p + '" { capabilities = ["deny"] }\n' for p in (KV, META))
|
|
require(all('"' + p + '"' not in boundary['policy'] for p in (KV, META)), 'boundary_path_requires_reconciliation')
|
|
bao(['write', BOUNDARY, '-'], payload={'policy': boundary['policy'] + '\n' + additions})
|
|
require(read_optional(BOUNDARY)['policy'] == boundary['policy'] + '\n' + additions, 'boundary_readback_failed')
|
|
for ref in CCRS:
|
|
command(['python3', str(ROOT / 'scripts/credential-change.py'), 'applier-apply', ref, '--actor', 'operator via attended sitting-requester custody session', '--confirm', 'DELEGATED APPLY ' + ref, '--quiet'])
|
|
receipt['phase'] = 'reader_roles_applied'
|
|
result = data(bao(['write', '-format=json', KV, '-'], payload={'options': {'cas': 0}, 'data': {'CLIENT_SECRET': secrets.token_urlsafe(48)}}))
|
|
require(result['data']['version'] == 1, 'initial_version_mismatch')
|
|
receipt.update(phase='custody_seeded', kv_version=1, request_id=result.get('request_id'))
|
|
RECEIPT.write_text(json.dumps(receipt, indent=2) + '\n')
|
|
command(KUBE + ['apply', '-f', str(ESO)])
|
|
deadline = time.monotonic() + 90
|
|
while time.monotonic() < deadline:
|
|
obj = rollout.get(KUBE, 'externalsecret', 'keycape-informed-decision-sitting-requester-client')
|
|
if any(c.get('type') == 'Ready' and c.get('status') == 'True' for c in obj.get('status', {}).get('conditions', [])):
|
|
break
|
|
time.sleep(2)
|
|
else:
|
|
raise LaneError('verifier_sync_failed')
|
|
receipt['phase'] = 'verifier_synced'
|
|
rollout.patch_object(KUBE, 'secret', secret, '/data/config.yaml', config)
|
|
after = rollout.get(KUBE, 'secret', 'keycape-config')
|
|
require(after['data'] == dict(secret['data'], **{'config.yaml': config}), 'unrelated_config_or_key_changed')
|
|
rollout.patch_object(KUBE, 'deployment', deployment, '/spec', spec)
|
|
receipt['keycape'] = rollout.ready(KUBE, image)
|
|
receipt.update(status='applied', phase='keycape_ready', image_unchanged=image, unrelated_config_preserved=True, verifier_secret_exported=False, sitting_post=False)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
receipt = {'observed_at': datetime.now(timezone.utc).isoformat(), 'status': 'failed', 'phase': 'preflight', 'ccrs': list(CCRS), 'credential_values_emitted': False, 'approval_mutations': False, 'sitting_post': False}
|
|
try:
|
|
run(receipt)
|
|
except Exception:
|
|
raise SystemExit(1) from None
|
|
finally:
|
|
RECEIPT.write_text(json.dumps(receipt, indent=2) + '\n')
|