LEARN. PRACTICE. GROW.Prepare disposable certificates
Save the complete code below as run.py and run python3 run.py --output evidence.json. Set DR_OPENSSL_BIN if the OpenSSL executable is elsewhere. Recorded execution used Python 3.13.1, OpenSSL CLI 3.6.1 and the OpenSSL 3.6.1 library on macOS. It creates temporary CAs and EC keys, restricts file permissions and deletes the material afterwards. Connections use only IPv4 loopback and TLS 1.3. One- or two-day certificate lifetimes serve the experiment without defining a production issuance policy.
Draw both directions
Draw two arrows before analyzing the matrix. The client verifies the server certificate using server-ca and the name api.fund.test. The server verifies client certificates using client-old-ca, client-new-ca or both. Updating one store does not automatically correct the opposite direction. The wrong-server-trust group deliberately gives the client the wrong CA and fails before the operation. The wrong-reference group retains trust but fails name verification. In a fictional incident, identify who verified which certificate under which policy before distributing additional CAs.
Predict migration across three states
Build a table with three rows: old trust, overlap and new trust. Columns represent old and new clients. In the first state only the old client receives ALLOW; during overlap both do; in the final state only the new client does. This is a planned migration in which both CAs remain approved. The table does not justify continued trust in a compromised CA. For a fictional batch service, add overnight consumers and verifier instances to the inventory. An interactive pilot does not establish that every consumer stopped depending on the old CA.
Test rejection beyond the issuer
The server requires a client certificate. A client without one is rejected, and a certificate issued by an approved CA but restricted to serverAuth also fails in the client role. Record phase, verifier error and application-read counter. A trusted CA does not remove the other validation conditions. Do not change verify_mode or purpose merely to make a negative case pass. Rejection is that case’s expected result. In the acceptance matrix, preserving correct rejection matters as much as successful operation for authorized consumers.
Separate identity from permission
The teaching application reads the verified certificate’s DNS SAN and authorizes only batch.fund.test. Two certificates with different keys and issuers retain that identifier and receive ALLOW during overlap. Another trusted certificate identifies unmapped.fund.test: the handshake finishes, but the response is DENY. This mapping is an original fixture rule, not a universal mechanism imposed by TLS. During a fictional renewal, check whether the profile changed identity and whether that change was intended. Correction needs the identity-contract or authorization owner, rather than indiscriminately widening trust.
Read outcomes from both sides
In recorded negative cases, client wrap_socket returned before the client observed server rejection during subsequent I/O. The server did not complete client validation or read the operation. Therefore, do not use only the client’s initial return as proof of accepted mTLS. Compare both sides’ logs and the required application response. Error codes and message wording can vary by runtime; retain the version and observed meaning. The fixture pins TLS 1.3 to make scope explicit and does not measure compatibility with every actual client.
"""Original bounded TLS 1.3/mTLS rollover lab, disposable PKI and IPv4 loopback.
DR_OPENSSL_BIN=/path/to/openssl python3 run.py --output evidence.json
No system trust changes, production keys, external traffic or key logging.
"""
import argparse,hashlib,json,os,pathlib,platform,shutil,socket,ssl,subprocess,tempfile,threading
from datetime import datetime,timezone
def run:
openssl=os.environ.get('DR_OPENSSL_BIN','/opt/homebrew/bin/openssl');checks={};details={};commands=[];tracked=[];workers=[]
def check(name,values,ok):
checks[name]={'passed':bool(ok),**values}
if not ok:raise AssertionError(name+': '+json.dumps(values))
with tempfile.TemporaryDirectory(prefix='dr-tls-rollover-') as td:
work=pathlib.Path(td);os.chmod(work,0o700);(work/'req.cnf').write_text('[req]\ndistinguished_name=dn\n[dn]\n')
def cmd(args):
p=subprocess.run([openssl,*args],cwd=work,capture_output=True,timeout=20);commands.append({'args':args,'exitCode':p.returncode})
if p.returncode:raise RuntimeError(p.stderr.decode(errors='replace'))
return p.stdout
version=cmd(['version']).decode.strip
def key(name):
cmd(['genpkey','-algorithm','EC','-pkeyopt','ec_paramgen_curve:P-256','-out',name+'.key']);os.chmod(work/(name+'.key'),0o600)
for n in ['server-ca','client-old-ca','client-new-ca']:
key(n);cmd(['req','-new','-x509','-config','req.cnf','-key',n+'.key','-out',n+'.pem','-subj','/CN=DR synthetic '+n,'-days','2','-addext','basicConstraints=critical,CA:TRUE,pathlen:0','-addext','keyUsage=critical,keyCertSign,cRLSign','-addext','subjectKeyIdentifier=hash'])
def issue(name,ca,serial,eku,dns):
key(name);cmd(['req','-new','-config','req.cnf','-key',name+'.key','-out',name+'.csr','-subj','/CN=DR synthetic '+name])
(work/(name+'.ext')).write_text('basicConstraints=critical,CA:FALSE\nkeyUsage=critical,digitalSignature\nextendedKeyUsage='+eku+'\nsubjectAltName=DNS:'+dns+'\nsubjectKeyIdentifier=hash\nauthorityKeyIdentifier=keyid,issuer\n')
cmd(['x509','-req','-in',name+'.csr','-CA',ca+'.pem','-CAkey',ca+'.key','-set_serial',str(serial),'-days','1','-extfile',name+'.ext','-out',name+'.pem'])
issue('server-old','server-ca',101,'serverAuth','api.fund.test');issue('server-new','server-ca',102,'serverAuth','api.fund.test')
issue('client-old','client-old-ca',201,'clientAuth','batch.fund.test');issue('client-new','client-new-ca',301,'clientAuth','batch.fund.test')
issue('client-renamed','client-new-ca',302,'clientAuth','unmapped.fund.test');issue('client-wrong-purpose','client-new-ca',303,'serverAuth','batch.fund.test')
(work/'overlap.pem').write_bytes((work/'client-old-ca.pem').read_bytes+(work/'client-new-ca.pem').read_bytes)
def tune(ctx):
ctx.minimum_version=ssl.TLSVersion.TLSv1_3;ctx.maximum_version=ssl.TLSVersion.TLSv1_3;ctx.verify_flags|=ssl.VERIFY_X509_STRICT
return ctx
def server(trust='overlap',cert='server-old',keyname=None):
ctx=tune(ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER));ctx.verify_mode=ssl.CERT_REQUIRED;ctx.num_tickets=0
ctx.load_cert_chain(work/(cert+'.pem'),work/((keyname or cert)+'.key'));ctx.load_verify_locations(work/(trust+'.pem'));return ctx
def client(cert='client-new',trust='server-ca'):
ctx=tune(ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT));ctx.hostname_checks_common_name=False;ctx.load_verify_locations(work/(trust+'.pem'))
if cert:ctx.load_cert_chain(work/(cert+'.pem'),work/(cert+'.key'))
return ctx
def fingerprint(cert):return hashlib.sha256(cmd(['x509','-in',cert+'.pem','-outform','DER'])).hexdigest
cert_hashes={n:fingerprint(n) for n in ['server-old','server-new']};details['certificateHashes']=cert_hashes
def line(conn):
data=bytearray
while len(data)<64:
b=conn.recv(1)
if not b:raise EOFError('fixture response ended')
data.extend(b)
if b==b'\n':return bytes(data)
raise ValueError('fixture line too long')
def listener:
s=socket.socket;tracked.append(s);s.settimeout(3);s.bind(('127.0.0.1',0));s.listen(2);return s
def exchange(label,serverctx,clientctx,hostname='api.fund.test'):
report={'serverHandshake':False,'clientHandshake':False,'serverApplicationReads':0,'response':None};listen=listener;report['endpoint']=listen.getsockname
def serve:
raw=None
try:
raw,_=listen.accept;tracked.append(raw);raw.settimeout(3)
with serverctx.wrap_socket(raw,server_side=True) as secure:
tracked.append(secure);report['serverHandshake']=True;report['serverTLS']=secure.version;report['clientNames']=[v for k,v in secure.getpeercert.get('subjectAltName',[]) if k=='DNS'];request=line(secure);report['serverApplicationReads']+=1
response=b'ALLOW\n' if request==b'OP\n' and report['clientNames']==['batch.fund.test'] else b'DENY\n'secure.sendall(response)
except (ssl.SSLError,OSError,EOFError) as error:report['serverError']={'type':type(error).__name__,'reason':getattr(error,'reason',None),'verifyCode':getattr(error,'verify_code',None),'message':str(error)}
finally:
if raw:raw.close
worker=threading.Thread(target=serve,daemon=True);workers.append(worker);worker.start
try:
with socket.socket as raw:
tracked.append(raw);raw.settimeout(3);raw.connect(listen.getsockname)
with clientctx.wrap_socket(raw,server_hostname=hostname) as secure:
tracked.append(secure);report['clientHandshake']=True;report['clientTLS']=secure.version;report['serverCertificateSHA256']=hashlib.sha256(secure.getpeercert(binary_form=True)).hexdigest;report['sessionReused']=secure.session_reused;secure.sendall(b'OP\n');report['response']=line(secure).decode.strip
except (ssl.SSLError,OSError,EOFError) as error:report['clientError']={'type':type(error).__name__,'reason':getattr(error,'reason',None),'verifyCode':getattr(error,'verify_code',None),'message':str(error)}
finally:
worker.join(timeout=4);listen.close
if worker.is_alive:raise RuntimeError('fixture worker did not finish')
details[label]=report;return report
def allowed(x):return x['serverHandshake'] and x['response']=='ALLOW'
old=server('client-old-ca');overlap=server;new=server('client-new-ca')
a=exchange('old-trust-old-client',old,client('client-old'));b=exchange('old-trust-new-client',old,client('client-new'))
check('old-client-trust-does-not-admit-new-issuer',{'oldAllowed':allowed(a),'newServerHandshake':b['serverHandshake'],'newApplicationReads':b['serverApplicationReads']},allowed(a) and not b['serverHandshake'] and b['serverApplicationReads']==0)
a=exchange('overlap-old-client',overlap,client('client-old'));b=exchange('overlap-new-client',overlap,client('client-new'))
check('approved-overlap-admits-both-client-issuers',{'oldAllowed':allowed(a),'newAllowed':allowed(b),'sameApplicationIdentity':a.get('clientNames')==b.get('clientNames')},allowed(a) and allowed(b) and a.get('clientNames')==b.get('clientNames'))
a=exchange('new-trust-old-client',new,client('client-old'));b=exchange('new-trust-new-client',new,client('client-new'))
check('new-only-trust-rejects-old-client-issuer',{'oldServerHandshake':a['serverHandshake'],'oldApplicationReads':a['serverApplicationReads'],'newAllowed':allowed(b)},not a['serverHandshake'] and a['serverApplicationReads']==0 and allowed(b))
a=exchange('no-client-certificate',overlap,client(None))
check('required-client-certificate-is-enforced',{'serverHandshake':a['serverHandshake'],'applicationReads':a['serverApplicationReads'],'serverReason':a.get('serverError',{}).get('reason')},not a['serverHandshake'] and a['serverApplicationReads']==0 and a.get('serverError',{}).get('reason')=='PEER_DID_NOT_RETURN_A_CERTIFICATE')
a=exchange('wrong-client-purpose',new,client('client-wrong-purpose'))
check('trusted-issuer-does-not-replace-client-purpose',{'serverHandshake':a['serverHandshake'],'applicationReads':a['serverApplicationReads'],'verifyCode':a.get('serverError',{}).get('verifyCode')},not a['serverHandshake'] and a['serverApplicationReads']==0 and a.get('serverError',{}).get('verifyCode')==26)
a=exchange('trusted-unmapped-identity',new,client('client-renamed'))
check('authenticated-client-can-be-denied-by-application',{'serverHandshake':a['serverHandshake'],'clientNames':a.get('clientNames'),'applicationReads':a['serverApplicationReads'],'response':a['response']},a['serverHandshake'] and a.get('clientNames')==['unmapped.fund.test'] and a['response']=='DENY')
a=exchange('wrong-server-trust',new,client('client-new','client-new-ca'))
check('client-and-server-trust-directions-are-independent',{'clientHandshake':a['clientHandshake'],'applicationReads':a['serverApplicationReads'],'clientErrorType':a.get('clientError',{}).get('type')},not a['clientHandshake'] and a['serverApplicationReads']==0 and a.get('clientError',{}).get('type')=='SSLCertVerificationError')
a=exchange('wrong-reference-name',new,client,'other.fund.test')
check('server-name-verification-remains-enabled-during-rollover',{'clientHandshake':a['clientHandshake'],'applicationReads':a['serverApplicationReads'],'verifyCode':a.get('clientError',{}).get('verifyCode')},not a['clientHandshake'] and a['serverApplicationReads']==0 and a.get('clientError',{}).get('verifyCode')==62)
for ext in ['pem','key']:shutil.copyfile(work/('server-old.'+ext),work/('deployed.'+ext))
os.chmod(work/'deployed.key',0o600);loaded=server(cert='deployed')
for ext in ['pem','key']:shutil.copyfile(work/('server-new.'+ext),work/('deployed.'+ext))
a=exchange('old-context-after-file-replacement',loaded,client);fresh=server(cert='deployed');b=exchange('fresh-context-after-file-replacement',fresh,client)
check('replacing-certificate-files-does-not-reload-existing-context',{'diskMatchesNewCertificate':fingerprint('deployed')==cert_hashes['server-new'],'oldContextServesOld':a.get('serverCertificateSHA256')==cert_hashes['server-old'],'freshContextServesNew':b.get('serverCertificateSHA256')==cert_hashes['server-new'],'bothAllowed':allowed(a) and allowed(b)},fingerprint('deployed')==cert_hashes['server-new'] and a.get('serverCertificateSHA256')==cert_hashes['server-old'] and b.get('serverCertificateSHA256')==cert_hashes['server-new'] and allowed(a) and allowed(b))
try:server(cert='server-new',keyname='server-old')
except ssl.SSLError as error:reason=error.reason
else:raise AssertionError('mismatched certificate/key loaded')
check('mismatched-key-is-rejected-before-serving',{'reason':reason,'listenerCreatedForMismatch':False},reason=='KEY_VALUES_MISMATCH')
shutil.copyfile(work/'client-old-ca.pem',work/'deployed-trust.pem');loaded=server(trust='deployed-trust');shutil.copyfile(work/'client-new-ca.pem',work/'deployed-trust.pem')
a=exchange('old-context-after-trust-file-replacement',loaded,client('client-old'));fresh=server(trust='deployed-trust');b=exchange('fresh-context-old-client',fresh,client('client-old'));c=exchange('fresh-context-new-client',fresh,client)
check('replacing-trust-file-does-not-retire-loaded-anchor',{'oldContextStillAllowsOld':allowed(a),'freshContextRejectsOld':not b['serverHandshake'],'freshContextAllowsNew':allowed(c)},allowed(a) and not b['serverHandshake'] and allowed(c))
listen=listener;state={}
def established_server:
raw=None
try:
raw,_=listen.accept;tracked.append(raw);raw.settimeout(3)
with overlap.wrap_socket(raw,server_side=True) as secure:
tracked.append(secure);state['peerNames']=[v for k,v in secure.getpeercert.get('subjectAltName',[]) if k=='DNS']
for _ in range(2):assert line(secure)==b'OP\n'secure.sendall(b'EXISTING-OK\n')
except Exception as error:state['error']=str(error)
finally:
if raw:raw.close
worker=threading.Thread(target=established_server,daemon=True);workers.append(worker);worker.start
try:
with socket.socket as raw:
tracked.append(raw);raw.settimeout(3);raw.connect(listen.getsockname)
with client('client-old').wrap_socket(raw,server_hostname='api.fund.test') as secure:
tracked.append(secure);secure.sendall(b'OP\n');first=line(secure);probe=exchange('new-policy-probe-while-old-connection-open',new,client('client-old'));secure.sendall(b'OP\n');second=line(secure)
finally:
worker.join(timeout=4);listen.close
if worker.is_alive:raise RuntimeError('persistent fixture did not finish')
details['establishedConnection']=state
check('new-context-policy-does-not-terminate-an-existing-connection',{'firstResponse':first.decode.strip,'newProbeRejected':not probe['serverHandshake'],'secondResponse':second.decode.strip,'oldConnectionReauthenticated':False,'separateProbeListener':True},first==second==b'EXISTING-OK\n' and not probe['serverHandshake'] and 'error' not in state)
result={'executedAt':datetime.now(timezone.utc).isoformat,'pythonVersion':platform.python_version,'platform':platform.platform,'opensslCLI':version,'opensslLibrary':ssl.OPENSSL_VERSION,'checks':checks,'details':details,'commands':commands,'passed':sum(x['passed'] for x in checks.values),'failed':sum(not x['passed'] for x in checks.values),'allSocketsClosed':all(s.fileno==-1 for s in tracked),'allWorkersStopped':all(not t.is_alive for t in workers),'temporaryMaterialDeleted':not pathlib.Path(td).exists,'scriptSha256':hashlib.sha256(pathlib.Path(__file__).read_bytes).hexdigest,'scope':'Actual TLS 1.3 full handshakes on synthetic IPv4 loopback endpoints with disposable EC certificates. Explicit client/server trust and SAN-based teaching authorization; no system trust changes or secret output. Approved-CA overlap only, not a compromised-CA recovery policy. No CRL/OCSP, session resumption, proxy, Kubernetes deployment, external traffic, production workload or human workshop. Contexts are recreated rather than mutated after use. The established-connection test probes new policy on a separate listener; it does not perform a production hot reload.'}
assert result['allSocketsClosed'] and result['allWorkersStopped'] and result['temporaryMaterialDeleted'];return result
if __name__=='__main__':
p=argparse.ArgumentParser;p.add_argument('--output',required=True);args=p.parse_args;result=run;pathlib.Path(args.output).write_text(json.dumps(result,indent=2)+'\n');print(json.dumps({k:result[k] for k in ['passed','failed','allSocketsClosed','allWorkersStopped','temporaryMaterialDeleted']}))
Exercise: fill the old/overlap/new matrix for two clients, then add a missing certificate, wrong purpose and trusted identity without permission. Predict handshake and response separately.
Common pitfalls
Swapping trust directions, validating only one pilot, turning negative cases into success or confusing authentication with operation authorization.
Related topics: Negotiation, mTLS, and the application · Renewal and served certificates · Diagnosis with explicit criteria
The matrix should establish who is accepted and rejected in each phase while keeping trust, identity and permission separate.
Reference: Python ssl: TLS contexts and verification · BigSavant TLS/certificates 2026-09; selected TLS 1.2/1.3, RFC 9525 identity and OpenSSL 3.5 diagnostics