ae27a42b18
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a06ecb-456a-71c2-b41e-0755d336e883
92 lines
4.6 KiB
Python
92 lines
4.6 KiB
Python
#!/usr/bin/env python3
|
|
"""Attended API-only signing acceptance; values stay in captured memory."""
|
|
import argparse
|
|
import base64
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
|
|
from state_hub_preflight_lane import CLUSTER_UID, FIELD
|
|
|
|
POD_PROBE = '''
|
|
import hmac,json,sys,urllib.request
|
|
from api.config import settings
|
|
from api.services.repository_rename import _verify_preflight_token, RenamePreconditionFailed
|
|
x=json.load(sys.stdin)
|
|
with urllib.request.urlopen('http://127.0.0.1:8000/state/health',timeout=20) as r:
|
|
health=json.load(r)
|
|
req=urllib.request.Request('http://127.0.0.1:8000/repos/fda8ad85-a7d7-4055-8f21-902a533e59df/rename/preflight',data=b'{"new_slug":"access-engine"}',headers={'Content-Type':'application/json'},method='POST')
|
|
with urllib.request.urlopen(req,timeout=60) as r: report=json.load(r)
|
|
token=report.get('preflight_token')
|
|
valid=False
|
|
if token:
|
|
_verify_preflight_token(token)
|
|
valid=True
|
|
old_rejected=None
|
|
if x.get('old_token'):
|
|
try: _verify_preflight_token(x['old_token'])
|
|
except RenamePreconditionFailed as e: old_rejected=str(e)=='Invalid repository rename preflight token'
|
|
else: old_rejected=False
|
|
print(json.dumps({'key_matches':hmac.compare_digest(settings.repository_rename_preflight_secret or '',x['key']), 'signed_preflight_valid':valid, 'old_token_rejected':old_rejected, 'blockers':[b.get('code') for b in report['blockers']], 'token':token, 'health':health.get('instance_role')=='primary' and health.get('instance_label')=='railiance01'}))
|
|
'''
|
|
|
|
|
|
def main():
|
|
p = argparse.ArgumentParser(description=__doc__)
|
|
p.add_argument('--kubeconfig', required=True)
|
|
p.add_argument('--receipt', required=True)
|
|
p.add_argument('--fixture', required=True)
|
|
p.add_argument('--after-rotation', action='store_true')
|
|
args = p.parse_args()
|
|
k = ['kubectl', '--kubeconfig', args.kubeconfig]
|
|
def command(argv, payload=None):
|
|
r = subprocess.run(argv, input=payload, capture_output=True, timeout=90)
|
|
if r.returncode:
|
|
raise RuntimeError('captured_command_failed')
|
|
return json.loads(r.stdout)
|
|
receipt = {'schema': 'platform.statehub-preflight-api-acceptance.v1', 'status': 'failed'}
|
|
fd = os.open(args.receipt, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
|
|
try:
|
|
assert command(k + ['get', 'namespace', 'kube-system', '-o', 'json'])['metadata']['uid'] == CLUSTER_UID
|
|
def get(kind, name):
|
|
return command(k + ['-n', 'state-hub', 'get', kind, name, '-o', 'json'])
|
|
secret = get('secret', 'state-hub-rename-preflight')
|
|
key = base64.b64decode(secret['data'][FIELD]).decode()
|
|
assert len(key) == 64
|
|
assert FIELD not in get('secret', 'state-hub-env').get('data', {})
|
|
dep = get('deployment', 'state-hub')
|
|
refs = [e for c in dep['spec']['template']['spec']['containers'] for e in c.get('env', []) if e['name'] == FIELD]
|
|
assert refs == [{'name': FIELD, 'valueFrom': {'secretKeyRef': {'name': 'state-hub-rename-preflight', 'key': FIELD, 'optional': False}}}]
|
|
mcp = get('deployment', 'state-hub-mcp')
|
|
assert 'state-hub-rename-preflight' not in json.dumps(mcp['spec']['template']['spec'])
|
|
pods = command(k + ['-n', 'state-hub', 'get', 'pods', '-l', 'app=state-hub', '-o', 'json'])['items']
|
|
assert len(pods) == dep['spec']['replicas'] and pods
|
|
old = Path(args.fixture).read_text() if args.after_rotation else None
|
|
results = []
|
|
token = None
|
|
for pod in pods:
|
|
assert not pod['metadata'].get('deletionTimestamp')
|
|
result = command(k + ['-n', 'state-hub', 'exec', '-i', pod['metadata']['name'], '-c', 'state-hub', '--', 'python', '-c', POD_PROBE], json.dumps({'key': key, 'old_token': old}).encode())
|
|
token = result.pop('token')
|
|
assert result['key_matches'] and result['signed_preflight_valid'] and result['health']
|
|
if args.after_rotation:
|
|
assert result['old_token_rejected']
|
|
results.append(result)
|
|
if not args.after_rotation:
|
|
fixture_fd = os.open(args.fixture, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
|
|
with os.fdopen(fixture_fd, 'w') as out:
|
|
out.write(token)
|
|
receipt.update(status='passed', api_only=True, replicas=len(pods), secret_resource_version=secret['metadata']['resourceVersion'], results=results)
|
|
except Exception:
|
|
receipt['error'] = 'acceptance_failed_details_suppressed'
|
|
finally:
|
|
with os.fdopen(fd, 'w') as out:
|
|
json.dump(receipt, out, indent=2)
|
|
out.write('\n')
|
|
return 0 if receipt['status'] == 'passed' else 1
|
|
|
|
|
|
if __name__ == '__main__':
|
|
sys.exit(main())
|