APRENDER. PRATICAR. EVOLUIR.Preparar certificados descartáveis
Guarda o código completo abaixo como run.py e executa python3 run.py --output evidence.json. Define DR_OPENSSL_BIN se o executável OpenSSL estiver noutro caminho. O ensaio registado usou Python 3.13.1, OpenSSL CLI 3.6.1 e biblioteca OpenSSL 3.6.1 em macOS. Cria CAs e chaves EC temporárias, restringe permissões dos ficheiros e elimina o material no fim. As ligações usam apenas IPv4 loopback e TLS 1.3. Os prazos de validade de um ou dois dias servem o ensaio, sem definir uma política de emissão para produção.
Desenhar as duas direções
Desenha duas setas antes de analisar a matriz. O cliente verifica o certificado do servidor usando server-ca e o nome api.fund.test. O servidor verifica os certificados dos clientes usando client-old-ca, client-new-ca ou ambas. Atualizar um store não corrige automaticamente a direção oposta. O grupo wrong-server-trust usa deliberadamente no cliente a CA errada e falha antes da operação. O grupo de referência errada preserva a confiança mas falha no nome. Num incidente fictício, identifica quem verificou qual certificado e com que política antes de distribuir mais CAs.
Prever a migração em três estados
Constrói uma tabela com três linhas: confiança antiga, sobreposição e confiança nova. As colunas representam clientes antigos e novos. No primeiro estado só o antigo recebe ALLOW; na sobreposição ambos recebem; no estado final só o novo recebe. Esta é uma migração planeada em que as duas CAs permanecem aprovadas. A tabela não justifica continuar a confiar numa CA comprometida. Para um batch fictício, acrescenta consumidores noturnos e instâncias de verificação ao inventário. Um piloto interativo não demonstra que todos os consumidores deixaram de depender da CA antiga.
Testar recusas além do emissor
O servidor exige certificado de cliente. Um cliente sem certificado é rejeitado, e um certificado emitido por uma CA aprovada mas limitado a serverAuth também falha no papel de cliente. Regista a fase, o erro do verificador e o contador de leituras da aplicação. Uma CA confiada não elimina as restantes condições de validação. Não alteres verify_mode nem o propósito apenas para fazer um caso negativo passar. O resultado esperado desse caso é uma recusa. Na matriz de aceitação, a preservação das recusas corretas tem tanta importância como o sucesso dos consumidores autorizados.
Separar identidade de permissão
A aplicação didática lê o SAN DNS do certificado verificado e autoriza apenas batch.fund.test. Dois certificados com chaves e emissores diferentes mantêm esse identificador e recebem ALLOW durante a sobreposição. Outro certificado confiado identifica unmapped.fund.test: o handshake termina, mas a resposta é DENY. Este mapeamento é uma regra original do ensaio, não um mecanismo universal imposto por TLS. Numa renovação fictícia, confirma se o perfil alterou a identidade e se essa alteração era pretendida. A correção exige o responsável pelo contrato de identidade ou pela autorização, não alargar confiança indiscriminadamente.
Ler os resultados dos dois lados
Nos casos negativos registados, wrap_socket regressou no cliente antes de este observar a recusa do servidor durante o I/O seguinte. O servidor não completou a validação do cliente e não leu a operação. Por isso, não uses apenas o retorno inicial do cliente como prova de mTLS aceite. Compara logs dos dois lados e a resposta aplicacional exigida. Os códigos de erro e a forma das mensagens podem variar por runtime; conserva a versão e o significado observado. O ensaio fixa TLS 1.3 para tornar o âmbito explícito e não mede compatibilidade com todos os clientes reais.
"""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']}))
Exercício: preenche a matriz antiga/sobreposição/nova para dois clientes, depois acrescenta cliente sem certificado, propósito errado e identidade confiada sem permissão. Prevê handshake e resposta separadamente.
Armadilhas comuns
Trocar as direções de confiança, validar apenas um piloto, transformar casos negativos em sucesso ou confundir autenticação com autorização da operação.
Tópicos relacionados: Negociação, mTLS e aplicação · Renovação e certificado servido · Diagnóstico com critérios explícitos
A matriz deve provar quem é aceite e quem é rejeitado em cada fase, mantendo separadas confiança, identidade e permissão.
Referência: Python ssl: TLS contexts and verification · DR TLS/certificates 2026-09; selected TLS 1.2/1.3, RFC 9525 identity and OpenSSL 3.5 diagnostics