tegwick
303a584bd0
Assistant: codex Assistant-Model: gpt-5.6-luna Assistant-Session: 01a07ff8-19d0-7820-b4d0-1353833cb7fc
92 lines
5.7 KiB
Python
92 lines
5.7 KiB
Python
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.')
|