tegwick
022cf4b727
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a06ed7-828d-7ca0-a8d4-0c3e5a0c4102
89 lines
4.2 KiB
Python
89 lines
4.2 KiB
Python
#!/usr/bin/env python3
|
|
"""Non-mutating live signing acceptance. Prints metadata and booleans only."""
|
|
import argparse
|
|
import base64
|
|
import hashlib
|
|
import hmac
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
import urllib.request
|
|
|
|
|
|
def kube(args, stdin=None):
|
|
r = subprocess.run(['kubectl', '-n', 'state-hub', *args], input=stdin,
|
|
capture_output=True, timeout=60)
|
|
if r.returncode:
|
|
raise RuntimeError('kubernetes_check_failed')
|
|
return r.stdout
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('--save-token', type=Path)
|
|
parser.add_argument('--predecessor-file', type=Path)
|
|
args = parser.parse_args()
|
|
secret = json.loads(kube(['get', 'secret', 'state-hub-rename-preflight', '-o', 'json']))
|
|
key = base64.b64decode(secret['data']['REPOSITORY_RENAME_PREFLIGHT_SECRET'])
|
|
pods = json.loads(kube(['get', 'pods', '-l', 'app=state-hub', '-o', 'json']))['items']
|
|
deployment = json.loads(kube(['get', 'deployment', 'state-hub', '-o', 'json']))
|
|
replicas = deployment['spec']['replicas']
|
|
assert replicas > 0 and len(pods) == replicas
|
|
health = json.load(urllib.request.urlopen('http://127.0.0.1:8000/state/health', timeout=20))
|
|
assert health['status'] == 'ok' and health['db'] == 'connected'
|
|
assert health['instance_role'] == 'primary' and health['instance_label'] == 'railiance01'
|
|
# Fixture is read-only and is the existing consuming migration's source ID.
|
|
req = urllib.request.Request(
|
|
'http://127.0.0.1:8000/repos/fda8ad85-a7d7-4055-8f21-902a533e59df/rename/preflight',
|
|
data=json.dumps({'new_slug': 'access-engine'}).encode(), method='POST',
|
|
headers={'Content-Type': 'application/json', 'X-StateHub-Component': 'state-hub.signing-acceptance'})
|
|
with urllib.request.urlopen(req, timeout=60) as response:
|
|
report = json.load(response)
|
|
assert report['safe_to_apply'] and report['preflight_token']
|
|
token = report['preflight_token']
|
|
encoded, sig = token.split('.')
|
|
assert hmac.compare_digest(hmac.new(key, encoded.encode(), hashlib.sha256).digest(),
|
|
base64.urlsafe_b64decode(sig + '=' * (-len(sig) % 4)))
|
|
predecessor = args.predecessor_file.read_text().strip() if args.predecessor_file else None
|
|
results = []
|
|
for pod in pods:
|
|
assert not pod['metadata'].get('deletionTimestamp')
|
|
assert any(c['type'] == 'Ready' and c['status'] == 'True' for c in pod['status']['conditions'])
|
|
# Only tokens, never the key, travel to the pod; no mutation endpoint runs.
|
|
payload = json.dumps({'token': token, 'predecessor': predecessor})
|
|
script = '''import json
|
|
from api.services.repository_rename import _verify_preflight_token, RenamePreconditionFailed
|
|
payload = json.loads(%r)
|
|
_verify_preflight_token(payload['token'])
|
|
if payload['predecessor']:
|
|
try:
|
|
_verify_preflight_token(payload['predecessor'])
|
|
except RenamePreconditionFailed as error:
|
|
if str(error) != 'Invalid repository rename preflight token':
|
|
raise RuntimeError('predecessor_rejection_was_not_signature_invalidation')
|
|
else:
|
|
raise RuntimeError('predecessor_not_invalidated')
|
|
print('verified')
|
|
''' % payload
|
|
assert kube(['exec', '-i', pod['metadata']['name'], '-c', 'state-hub', '--', 'python', '-'],
|
|
script.encode()).strip() == b'verified'
|
|
results.append({'pod_uid': pod['metadata']['uid'], 'accepted_key_version': True,
|
|
'predecessor_rejected': predecessor is not None})
|
|
if args.save_token:
|
|
fd = os.open(args.save_token, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
|
|
with os.fdopen(fd, 'w') as out:
|
|
out.write(token)
|
|
print(json.dumps({'schema': 'state-hub.signing-runtime-acceptance.v1',
|
|
'health': 'passed', 'signed_preflight': True, 'replicas': replicas,
|
|
'secret_resource_version': secret['metadata']['resourceVersion'],
|
|
'pods': results, 'repository_mutations': 0}))
|
|
|
|
|
|
if __name__ == '__main__':
|
|
try:
|
|
main()
|
|
except Exception:
|
|
print('{"status":"failed","error":"runtime_acceptance_failed"}')
|
|
sys.exit(1)
|