feat: accept KeyCape approval clients with tested recovery

Assistant: codex
Assistant-Model: gpt-5.6-luna
Assistant-Session: 01a07ff8-19d0-7820-b4d0-1353833cb7fc

sso-mfa/k8s/keycape/exercise-approval-clients.py

@@ -0,0 +1,92 @@
+import base64, importlib.util, json, os, secrets, socket, subprocess, tempfile, time
+from pathlib import Path
+from unittest.mock import patch
+from cryptography.hazmat.primitives import serialization
+from cryptography.hazmat.primitives.asymmetric import rsa
+import yaml
+import http.server, ssl, threading, urllib.request, urllib.error, datetime, ipaddress
+from cryptography import x509
+from cryptography.x509.oid import NameOID
+import argparse
+parser=argparse.ArgumentParser(description='Exercise the pinned image and verifier over loopback HTTPS using synthetic credentials only.')
+parser.add_argument('--receipt', required=True, type=Path)
+args=parser.parse_args()
+spec=importlib.util.spec_from_file_location('rollout',Path(__file__).with_name('approval-clients-rollout.py'))
+m=importlib.util.module_from_spec(spec); spec.loader.exec_module(m)
+m.verify_artifact()
+def call(args):
+    p=subprocess.run(args,capture_output=True,timeout=45)
+    if p.returncode: raise RuntimeError('contained_docker_command_failed')
+    return p
+sock=socket.socket(); sock.bind(('127.0.0.1',0)); port=sock.getsockname()[1]; sock.close()
+class Proxy(http.server.BaseHTTPRequestHandler):
+    def log_message(self, *args): pass
+    def handle_proxy(self):
+        body = self.rfile.read(int(self.headers.get('Content-Length', 0))) if self.command == 'POST' else None
+        req = urllib.request.Request(f'http://127.0.0.1:{port}'+self.path, data=body,
+              headers={k:self.headers[k] for k in ['Content-Type','Authorization'] if k in self.headers})
+        try:
+            response=urllib.request.urlopen(req,timeout=20)
+        except urllib.error.HTTPError as error: response=error
+        with response:
+            self.send_response(response.code)
+            self.send_header('Content-Type', response.headers.get('Content-Type','application/json'))
+            self.end_headers(); self.wfile.write(response.read())
+    do_GET=handle_proxy
+    do_POST=handle_proxy
+server=http.server.ThreadingHTTPServer(('127.0.0.1',0),Proxy)
+https_port=server.server_address[1]
+name='keycape-approval-exercise-'+secrets.token_hex(5)
+with tempfile.TemporaryDirectory(prefix='keycape-private-exercise-') as temp:
+    root=Path(temp)
+    key=rsa.generate_private_key(public_exponent=65537,key_size=2048)
+    (root/'key.pem').write_bytes(key.private_bytes(serialization.Encoding.PEM,serialization.PrivateFormat.PKCS8,serialization.NoEncryption()))
+    os.chmod(root/'key.pem',0o600)
+    name_attr=x509.Name([x509.NameAttribute(NameOID.COMMON_NAME,'KeyCape local exercise')])
+    cert=(x509.CertificateBuilder().subject_name(name_attr).issuer_name(name_attr).public_key(key.public_key())
+          .serial_number(x509.random_serial_number()).not_valid_before(datetime.datetime.now(datetime.timezone.utc)-datetime.timedelta(minutes=1))
+          .not_valid_after(datetime.datetime.now(datetime.timezone.utc)+datetime.timedelta(hours=1))
+          .add_extension(x509.SubjectAlternativeName([x509.IPAddress(ipaddress.ip_address('127.0.0.1'))]),False)
+          .add_extension(x509.BasicConstraints(ca=True,path_length=None),True))
+    from cryptography.hazmat.primitives import hashes
+    (root/'cert.pem').write_bytes(cert.sign(key,hashes.SHA256()).public_bytes(serialization.Encoding.PEM))
+    context=ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER); context.load_cert_chain(root/'cert.pem',root/'key.pem')
+    server.socket=context.wrap_socket(server.socket,server_side=True)
+    threading.Thread(target=server.serve_forever,daemon=True).start()
+    config=yaml.safe_load((m.KEYCAPE/'config/dev-config.yaml').read_text())
+    config.update(issuer=f'https://127.0.0.1:{https_port}')
+    config['authelia']['issuer']='https://auth.coulomb.social'
+    config['clients'].extend(m.registrations())
+    (root/'config.yaml').write_text(yaml.safe_dump(config,sort_keys=False))
+    os.chmod(root/'config.yaml',0o600)
+    values={c['clientId']:secrets.token_urlsafe(48) for c in m.registrations()}
+    (root/'env').write_text('KEYCAPE_CONFIG=/etc/keycape/config.yaml\n'+''.join(c['secretRef'].removeprefix('env:')+'='+values[c['clientId']]+'\n' for c in m.registrations()))
+    os.chmod(root/'env',0o600)
+    created=False
+    try:
+        call(['docker','run','-d','--rm','--name',name,'--user',str(os.getuid()),'--publish',f'127.0.0.1:{port}:8080',
+              '--env-file',str(root/'env'),'--mount','type=bind,source='+temp+',target=/etc/keycape,readonly',m.IMAGE])
+        created=True
+        with patch.object(m,'ISSUER',config['issuer']), patch.dict(os.environ,{'SSL_CERT_FILE':str(root/'cert.pem')}):
+            for _ in range(30):
+                try:
+                    status,_=m.http('/.well-known/openid-configuration')
+                    if status==200: break
+                except Exception: pass
+                time.sleep(.5)
+            else: raise RuntimeError('scratch_keycape_not_ready')
+            def fake_get(kube,kind,name):
+                index=m.SECRET_NAMES.index(name)
+                return {'data': {'client-secret':base64.b64encode(values[m.IDS[index]].encode()).decode()}}
+            receipt={}
+            with patch.object(m,'get',side_effect=fake_get):
+                m.acceptance([],receipt)
+            receipt.update(target='disposable pinned KeyCape image; synthetic keys and clients only',
+                           pinned_native_cli_verification='passed against loopback HTTPS with synthetic credentials',cleanup_complete=False)
+    finally:
+        if created: call(['docker','stop','--time=5',name])
+        server.shutdown(); server.server_close()
+    receipt['cleanup_complete']=True
+    fd=os.open(args.receipt,os.O_WRONLY|os.O_CREAT|os.O_EXCL,0o600)
+    with os.fdopen(fd,'w') as out: out.write(json.dumps(receipt,indent=2)+'\n')
+print('Pinned-image scratch acceptance passed; synthetic credentials removed.')