test: add non-mutating live signing rotation acceptance probe

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a06ed7-828d-7ca0-a8d4-0c3e5a0c4102

scripts/verify_preflight_signing_runtime.py

@@ -0,0 +1,87 @@
+#!/usr/bin/env python3
+"""Non-mutating live signing acceptance. Prints metadata and booleans only."""
+import argparse
+import base64
+from datetime import datetime, timedelta, timezone
+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))
+    # 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:
+        pass
+    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)