APRENDER. PRATICAR. EVOLUIR.1. Recuperar o serviço sem perder os controlos
Uma equipa APS fictícia está a preparar o fecho de fundos. Durante uma rotação planeada de certificados, os clientes novos funcionam numa réplica e falham noutra. Ao mesmo tempo, a versão anterior da aplicação deixou de poder criar Pods sob a política atual. O gestor técnico precisa de separar problemas de confiança, adoção da configuração e capacidade de recuperação. Estes exemplos não representam procedimentos internos da BNP Paribas. Começa por uma matriz simples: consumidor, destino, geração do servidor, identidade apresentada, resultado TLS e resultado da operação. Um erro de verificação do certificado, um HTTP 403 e um Pod que nem arrancou pertencem a etapas diferentes. Recolhe o resultado no ponto certo e atribui a investigação à equipa que controla esse ponto. Não uses um estado agregado verde para substituir a observação do fluxo que está a falhar. Define também o que tem de continuar verdadeiro durante a recuperação: clientes previstos conseguem operar, identidades não autorizadas continuam recusadas, dados não são repetidos e a equipa RUN consegue explicar qual a versão em serviço. Uma rotação planeada pode permitir sobreposição aprovada de confiança; compromisso da CA exige outra análise de contenção. Regista essa diferença antes de escolher a contingência. A prioridade operacional não elimina o requisito de segurança, mas ajuda a decidir a sequência e o ponto de escalamento.
2. Preparar uma versão de recuperação admissível
Uma recuperação Kubernetes tem de passar pelos controlos em vigor no momento em que os Pods são criados. Guardar o ReplicaSet anterior não garante que o seu template ainda seja admissível. No exercício de decisão, a release anterior usa privilégios agora proibidos. A equipa deve preparar uma imagem e um template compatíveis, validar dependências e testar a criação antes da janela de mudança. Inspeciona o campo recusado. Os controlos de Pod Security abrangem também init containers e ephemeral containers; acrescentar uma ferramenta de diagnóstico privilegiada pode ser recusado mesmo quando a aplicação original era compatível. Um acesso de emergência precisa de âmbito e aprovação definidos. Isentar uma RuntimeClass tem outro efeito relevante: os pedidos abrangidos deixam de passar pelos comportamentos enforce, audit e warn. A falta de avisos nessa situação não é uma avaliação positiva. Restricted também não é uma lista completa dos requisitos da organização. Por exemplo, a aceitação nessa política não demonstra que o root filesystem esteja read-only. Verifica esse requisito separadamente, incluindo onde a aplicação precisa de escrever. Para o caso de reconciliação, acrescenta um checkpoint funcional: qual foi o último movimento confirmado e como retomar sem duplicação? O plano de rollback deve conter configuração compatível, dependências disponíveis, testes e responsável pela decisão. A prática desta aula executa Pods Restricted, mas não reproduz os cenários de admissão e rollback descritos nesta secção.
3. Distinguir entrega, adoção e recuperação de Secrets
Desenha o percurso do material: objeto Secret, projeção no volume, leitura pelo processo e estado carregado. Cada seta é uma etapa a verificar. Nesta prática, o servidor Python lê o bundle de CAs quando cria o SSLContext e não implementa reload. A atualização posterior do volume pode estar correta enquanto o servidor continua a usar a confiança anterior. O hash do ficheiro e o resultado de uma ligação nova respondem a perguntas diferentes. Para pares de chave e certificado, prepara material coerente e valida a carga antes da promoção. Atualizar dois objetos num comando não cria uma transação de adoção na aplicação. Uma estratégia com versões identificadas e substituição controlada permite associar o consumidor ao conjunto pretendido. Um Secret obrigatório em falta deve continuar a impedir um arranque que, sem ele, seria inseguro; optional não acrescenta autenticação à aplicação. A recuperação de cifragem em repouso tem dependências próprias. Antes de promover uma chave de escrita, os API servers relevantes precisam de conseguir ler os dados dessa geração. Um rollback para uma configuração que só conhece a chave antiga pode falhar ao ler objetos novos. Além do cluster atual, considera backups retidos e as chaves de que dependem. A evidência necessária inclui capacidade de restauro, não apenas escrita atual. Esta aula não executa cifragem de etcd, KMS ou restauro de backups; usa estas situações para treinar decisões de recuperação.
4. Preservar a fronteira de isolamento na contingência
Imagina um serviço que executa código de um parceiro num runtime sandbox aprovado. O pool dedicado fica indisponível. Retirar runtimeClassName para arrancar no runtime normal altera uma propriedade de segurança da arquitetura. A decisão deve considerar capacidade alternativa compatível ou uma exceção de risco explícita; não deve ficar escondida num patch de recuperação. Na estimativa de capacidade, inclui overhead de RuntimeClass além dos requests dos containers. Um pool que cabe exatamente nos requests pode não ser suficiente para o workload real. Confirma também o handler instalado e a elegibilidade dos nodes. O objetivo do ensaio de contingência é demonstrar que a configuração aprovada consegue efetivamente executar e cumprir o fluxo, dentro da capacidade disponível. O isolamento de dados e rede precisa de verificações próprias. Um PersistentVolume retido pode sobreviver ao namespace e continuar a conter dados do tenant anterior. A sua reutilização exige tratamento de retenção e sanitização adequado. Uma NetworkPolicy aceite pela API também não demonstra enforcement se o plugin não a implementar. Na passagem ao ambiente de contingência, testa tráfego legítimo e proibido e identifica o componente que aplica o controlo. A prática abaixo usa a rede normal de um cluster kind e não instala sandbox nem um motor de NetworkPolicy. Ter mTLS na aplicação não fornece prova desses mecanismos. Mantém esses ensaios como trabalho prático adicional.
5. Executar mTLS entre Pods e separar autorização
A prática usa um cluster descartável, certificados sintéticos de curta duração e uma imagem Python local fixada por digest. O script exige contexto e kubeconfig dedicados, confirma que o API server é local e cria um namespace aleatório. Não usa credenciais reais. O exame oficial indica Kubernetes 1.35; a execução usa server 1.37.0 e kubectl 1.37.1, com Pod Security Restricted fixado em v1.35. Estas versões não são apresentadas como equivalentes em todos os comportamentos. O servidor escuta em 8443, exige certificado de cliente e valida a cadeia contra o seu bundle. O cliente valida a CA do servidor e o nome esperado, mesmo ligando ao IP do Pod. O certificado do servidor tem DNS SAN e finalidade serverAuth; os certificados de cliente têm URI SAN e clientAuth. São emitidos localmente para o exercício, sem ligação automática a ServiceAccounts ou a um emissor SPIFFE real. Depois do TLS, a aplicação permite apenas a URI sintética de reconciler. O cliente autorizado recebe 200 e reporter, apesar de certificado válido, recebe 403. Sem certificado ou com emissor desconhecido, o fluxo falha antes da resposta HTTP. Nome esperado errado ou confiança errada no servidor produzem falha de verificação no cliente. Antes de executar, prevê estes resultados numa tabela. Usa as respostas para distinguir transporte, autenticação e autorização, mantendo os limites deste programa simples.
6. Observar sobreposição e retirada de confiança
A primeira geração do servidor aceita a CA antiga de clientes. O cliente novo é recusado. O script acrescenta a CA nova ao Secret e espera até observar o hash atualizado no volume do servidor antigo. Mesmo assim, esse processo continua a recusar o cliente novo. A entrega ocorreu; o SSLContext carregado não mudou. A medição do atraso de projeção pertence a esta execução e não define um SLA. O script cria então uma geração de sobreposição que carrega o bundle atualizado. Tanto o cliente antigo como o novo recebem 200. Depois, o Secret passa a conter apenas a CA nova. Uma terceira geração lê esse conjunto e recusa o cliente antigo, mantendo o novo funcional. Entretanto, os servidores anteriores ainda conseguem aceitar clientes antigos: cada um conserva o seu próprio contexto. A prática remove essas duas gerações e volta a testar aceitação do novo e recusa do antigo. As 20 verificações cobrem novas ligações TLS e pedidos curtos. Não há teste de revogação OCSP, CRL, terminação de sessões persistentes ou rotação automática de uma service mesh. Também não há medição de tráfego de produção nem garantia de ausência de perda durante rollout. A experiência permite formular critérios melhores: identificar todos os verificadores ativos, observar adoção e retirar gerações antigas de forma controlada. Na operação real, acrescenta drenagem, observabilidade, checkpoints e testes dos percursos efetivamente servidos.
7. Aplicar os conceitos à política de uma mesh
As perguntas de Istio desta aula são baseadas na documentação primária, sem instalação de Istio no laboratório. PeerAuthentication trata requisitos de mTLS de entrada. AuthorizationPolicy define decisões de acesso. Uma identidade autenticada não implica permissão para qualquer operação, tal como reporter recebeu 403 no programa sintético. A analogia ajuda a pensar, mas não demonstra que o dataplane Istio foi configurado ou testado. No review de uma alteração, considera a semântica completa. Um DENY correspondente prevalece sobre ALLOW; acrescentar outro ALLOW não corrige essa recusa. Uma política ALLOW sem regras e uma com uma regra vazia têm efeitos diferentes. Em DENY, atributos HTTP ausentes podem fazer corresponder tráfego TCP, pelo que protocolo e âmbito exigem atenção. O campo de principal usa identidade do certificado do peer, não um header arbitrário preenchido pelo cliente. A migração entre sidecar e ambient também precisa de validação própria: a documentação atual não suporta DISABLE em ambient. Não copies uma exceção só porque o recurso tem o mesmo nome. Regista versão instalada, modo, namespace, seletor, alvo e resultado esperado antes do teste. Para cada regra, prepara um cliente autorizado e outro que deve ser recusado. Se os resultados divergem da intenção, preserva o controlo enquanto identificas a regra e o âmbito responsáveis. Só promove a alteração depois de os testes explicarem o acesso efetivo.
8. Fechar a mudança com evidência útil ao RUN
O entregável do gestor técnico é uma decisão suportada e repetível. Para o caso de fecho, apresenta a matriz por consumidor e geração, a janela de sobreposição aprovada, os testes positivos e negativos, o ponto de retoma funcional e os responsáveis. Se a CA antiga estivesse comprometida, a sobreposição planeada não poderia ser reutilizada automaticamente como contenção. Essa alteração de contexto exige nova decisão de risco. Depois do laboratório, confirma a remoção do namespace, dos ficheiros de chaves, do cluster descartável e do kubeconfig dedicado. O relatório conserva versões, digest da imagem, hash do script, resultados e limitações, sem guardar chaves privadas. Compara as duas execuções finais e explica qualquer diferença de tempo sem transformar uma observação num prazo garantido. O sucesso do script deve ser acompanhável pelas verificações, não depender de uma mensagem PASS isolada. Resume a aprendizagem numa sequência: identificar o requisito, localizar o verificador, observar a configuração entregue, confirmar a adoção, testar acessos e recuperar de forma controlada. Os temas relacionados são admissão, Secrets, identidade, autorização, isolamento, capacidade e recuperação. As perguntas treinam decisões originais; não são questões reais do exame. A avaliação interna não atribui certificação oficial. A prática adicional de mesh, CNI, sandbox e simulados completos, bem como revisão especializada independente, continua necessária para uma preparação madura.
"""Original disposable-cluster lab: application mTLS, authorization and trust rotation."""
import argparse,datetime,hashlib,json,os,shutil,subprocess,tempfile,time,uuid
from pathlib import Path
p=argparse.ArgumentParser();p.add_argument('--kubectl',required=True);p.add_argument('--kubeconfig',required=True);p.add_argument('--cluster',required=True);p.add_argument('--image',required=True);p.add_argument('--openssl',required=True);p.add_argument('--output',required=True);a=p.parse_args()
assert a.cluster.startswith('dr-cks-workload-response-') and '@sha256:' in a.image
assert Path(a.kubeconfig).is_absolute();out=Path(a.output);assert not out.exists()
os.umask(0o077);tmp=Path(tempfile.mkdtemp(prefix='dr-cks-mtls-'));ns='dr-mtls-'+uuid.uuid4().hex[:8];records=[];created=False
base=[a.kubectl,'--kubeconfig',a.kubeconfig,'--context','kind-'+a.cluster,'--cache-dir',str(tmp/'cache'),'--request-timeout=15s']
def run(cmd,data=None):
r=subprocess.run(cmd,input=data,text=True,capture_output=True,timeout=60)
if r.returncode:raise RuntimeError('Command failed: '+str(cmd[:3])+': '+r.stderr[:180])
return r.stdout
def k(*args,obj=None):return run(base+list(args),json.dumps(obj) if obj is not None else None)
def create(obj):return k('-n',ns,'create','-f','-',obj=obj)
def get(kind,name):return json.loads(k('-n',ns,'get',kind,name,'-o','json'))
def check(name,value,expected):
assert value==expected,(name,value,expected);records.append(dict(name=name,observed=value,passed=True));print(name,flush=True)
def poll(fn,seconds=150):
deadline=time.monotonic()+seconds
while time.monotonic()<deadline:
v=fn()
if v:return v
time.sleep(1)
raise AssertionError('Deadline exceeded')
def op(*args):return run([a.openssl,*map(str,args)])
def ca(name):
op('req','-x509','-newkey','rsa:2048','-nodes','-days','2','-subj','/CN='+name,'-keyout',tmp/(name+'.key'),'-out',tmp/(name+'.crt'),'-addext','basicConstraints=critical,CA:TRUE','-addext','keyUsage=critical,keyCertSign,cRLSign')
def leaf(name,issuer,san,purpose):
op('req','-new','-newkey','rsa:2048','-nodes','-subj','/CN='+name,'-keyout',tmp/(name+'.key'),'-out',tmp/(name+'.csr'))
ext=tmp/(name+'.ext');ext.write_text('basicConstraints=critical,CA:FALSE\nkeyUsage=critical,digitalSignature,keyEncipherment\nextendedKeyUsage='+purpose+'\nsubjectAltName='+san+'\nsubjectKeyIdentifier=hash\nauthorityKeyIdentifier=keyid,issuer\n')
op('x509','-req','-in',tmp/(name+'.csr'),'-CA',tmp/(issuer+'.crt'),'-CAkey',tmp/(issuer+'.key'),'-CAcreateserial','-days','1','-extfile',ext,'-out',tmp/(name+'.crt'))
def secret(name,files):create(dict(apiVersion='v1',kind='Secret',metadata={'name':name},stringData={key:(tmp/value).read_text() for key,value in files.items()}))
server=r'''import ssl,socket,json,sys
ctx=ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER);ctx.minimum_version=ssl.TLSVersion.TLSv1_2
ctx.load_cert_chain('/identity/tls.crt','/identity/tls.key');ctx.load_verify_locations('/trust/ca.crt');ctx.verify_mode=ssl.CERT_REQUIRED
s=socket.socket();s.setsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR,1);s.bind(('0.0.0.0',8443));s.listen(16)
print('ready',flush=True)
while True:
raw,_=s.accept();raw.settimeout(4)
try:
with ctx.wrap_socket(raw,server_side=True) as conn:
cert=conn.getpeercert();allowed=('URI','spiffe://dr.test/ns/funds/sa/reconciler') in cert.get('subjectAltName',())
conn.recv(4096);status=200 if allowed else 403
conn.sendall(('HTTP/1.1 '+str(status)+' Result\r\nContent-Length: 0\r\nConnection: close\r\n\r\n').encode())
except (ssl.SSLError,OSError):raw.close()
'''
client=r'''import ssl,socket,json,sys
host,identity,trust,hostname=sys.argv[1:];ctx=ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT);ctx.minimum_version=ssl.TLSVersion.TLSv1_2
ctx.load_verify_locations('/credentials/'+trust+'.crt')
if identity!='none':ctx.load_cert_chain('/credentials/'+identity+'.crt','/credentials/'+identity+'.key')
try:
with socket.create_connection((host,8443),timeout=4) as raw:
with ctx.wrap_socket(raw,server_hostname=hostname) as s:
s.sendall(b'GET /reconcile HTTP/1.1\r\nHost: funds\r\n\r\n');response=s.recv(4096)
print(json.dumps({'status':int(response.split()[1]),'tls':s.version(),'peerVerified':True}))
except ssl.SSLCertVerificationError as e:print(json.dumps({'error':'verification','verifyCode':e.verify_code}))
except ssl.SSLError as e:print(json.dumps({'error':'tls','reason':e.reason}))
'''
def pod(name,serving=False):
mounts=[dict(name='code',mountPath='/code',readOnly=True)];vols=[dict(name='code',configMap={'name':'code'})]
if serving:
for key,secretName in [('identity','server'),('trust','client-trust')]:
mounts.append(dict(name=key,mountPath='/'+key,readOnly=True));vols.append(dict(name=key,secret={'secretName':secretName,'defaultMode':0o440}))
else:
mounts.append(dict(name='credentials',mountPath='/credentials',readOnly=True));vols.append(dict(name='credentials',secret={'secretName':'clients','defaultMode':0o440}))
return dict(apiVersion='v1',kind='Pod',metadata={'name':name,'labels':{'app':name}},spec=dict(automountServiceAccountToken=False,terminationGracePeriodSeconds=1,securityContext=dict(runAsNonRoot=True,runAsUser=1000,runAsGroup=1000,fsGroup=1000,seccompProfile={'type':'RuntimeDefault'}),volumes=vols,containers=[dict(name='worker',image=a.image,imagePullPolicy='Never',command=['python','/code/server.py'] if serving else ['python','-c','import time;time.sleep(1800)'],volumeMounts=mounts,securityContext=dict(allowPrivilegeEscalation=False,readOnlyRootFilesystem=True,capabilities={'drop':['ALL']}),resources=dict(requests={'cpu':'10m','memory':'24Mi'},limits={'cpu':'250m','memory':'64Mi'}))]))
def ready(name):poll(lambda:get('pod',name).get('status',{}).get('containerStatuses',[{}])[0].get('ready',False))
def request(target,identity,trust='server-ca',hostname=None):
return json.loads(k('-n',ns,'exec','client','--','python','/code/client.py',get('pod',target)['status']['podIP'],identity,trust,hostname or dns))
def status(target,identity):return request(target,identity).get('status')
def denial(target,identity,trust='server-ca',hostname=None):return request(target,identity,trust,hostname).get('error')
def gen(name):
create(pod(name,True));ready(name);poll(lambda:'ready' in k('-n',ns,'logs',name))
def trust_update(names):
data=''.join((tmp/(n+'.crt')).read_text() for n in names)
k('-n',ns,'patch','secret','client-trust','--type=merge','-p',json.dumps({'stringData':{'ca.crt':data}}))
expected=hashlib.sha256(data.encode()).hexdigest()
code="import hashlib;from pathlib import Path;print(hashlib.sha256(Path('/trust/ca.crt').read_bytes()).hexdigest())"
start=time.monotonic();poll(lambda:k('-n',ns,'exec','server-old','--','python','-c',code).strip()==expected)
return dict(projected=True,delaySeconds=round(time.monotonic()-start,2))
try:
cfg=json.loads(k('config','view','--minify','--raw','-o','json'))['clusters'][0]['cluster'];assert cfg['server'].startswith('https://127.0.0.1:') and not cfg.get('insecure-skip-tls-verify');cfg=None
versions=json.loads(k('version','-o','json'));assert versions['serverVersion']['gitVersion']=='v1.37.0' and versions['clientVersion']['gitVersion']=='v1.37.1'
k('create','namespace',ns);created=True;k('label','namespace',ns,'pod-security.kubernetes.io/enforce=restricted','pod-security.kubernetes.io/enforce-version=v1.35')
dns='funds.'+ns+'.svc';[ca(n) for n in ['server-ca','old-ca','new-ca','rogue-ca']]
leaf('server','server-ca','DNS:'+dns,'serverAuth')
for name,issuer,identity in [('old','old-ca','reconciler'),('new','new-ca','reconciler'),('unauthorized','old-ca','reporter'),('rogue','rogue-ca','reconciler')]:leaf(name,issuer,'URI:spiffe://dr.test/ns/funds/sa/'+identity,'clientAuth')
secret('server',{'tls.crt':'server.crt','tls.key':'server.key'});secret('client-trust',{'ca.crt':'old-ca.crt'})
files={n+suffix:n+suffix for n in ['old','new','unauthorized','rogue'] for suffix in ['.crt','.key']};files.update({'server-ca.crt':'server-ca.crt','rogue-ca.crt':'rogue-ca.crt'});secret('clients',files)
create(dict(apiVersion='v1',kind='ConfigMap',metadata={'name':'code'},data={'server.py':server,'client.py':client}));create(pod('client'));ready('client');gen('server-old')
check('restricted-pods-running',all(get('pod',n)['spec']['securityContext']['runAsUser']==1000 for n in ['client','server-old']),True)
runtime=json.loads(k('-n',ns,'exec','client','--','python','-c',"import sys,ssl,platform,json;print(json.dumps({'python':platform.python_version(),'openssl':ssl.OPENSSL_VERSION,'kernel':platform.release(),'architecture':platform.machine()}))"));images={n:get('pod',n)['status']['containerStatuses'][0]['imageID'] for n in ['client','server-old']}
check('old-authorized-client-accepted',status('server-old','old'),200)
check('missing-client-certificate-rejected',denial('server-old','none'),'tls')
check('untrusted-client-certificate-rejected',denial('server-old','rogue'),'tls')
check('trusted-unauthorized-client-denied',status('server-old','unauthorized'),403)
check('wrong-server-hostname-rejected',denial('server-old','old',hostname='different.dr.test'),'verification')
check('wrong-server-trust-rejected',denial('server-old','old',trust='rogue-ca'),'verification')
check('new-client-before-overlap-rejected',denial('server-old','new'),'tls')
overlap=trust_update(['old-ca','new-ca']);check('overlap-trust-projected',overlap['projected'],True)
check('loaded-context-still-rejects-new-client',denial('server-old','new'),'tls')
gen('server-overlap');check('replacement-context-accepts-old-client',status('server-overlap','old'),200);check('replacement-context-accepts-new-client',status('server-overlap','new'),200)
retirement=trust_update(['new-ca']);check('new-only-trust-projected',retirement['projected'],True)
check('old-generation-still-accepts-old-client',status('server-old','old'),200)
gen('server-new');check('new-generation-rejects-old-client',denial('server-new','old'),'tls');check('new-generation-accepts-new-client',status('server-new','new'),200)
check('overlap-generation-still-accepts-old-client',status('server-overlap','old'),200)
for n in ['server-old','server-overlap']:k('-n',ns,'delete','pod',n,'--wait=true','--timeout=40s')
remaining=json.loads(k('-n',ns,'get','pods','-o','json'))['items'];check('old-generations-removed',sorted(x['metadata']['name'] for x in remaining),['client','server-new'])
check('accepted-flow-survives-retirement',status('server-new','new'),200)
check('retired-client-still-denied',denial('server-new','old'),'tls')
assert len(records)==20
finally:
if created:k('delete','namespace',ns,'--wait=true','--timeout=45s')
shutil.rmtree(tmp)
report=dict(executedAt=datetime.datetime.now(datetime.timezone.utc).isoformat(),scriptSha256=hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),versions=versions,runtime=runtime,issuerOpenSSL=op('version').strip(),images=images,examVersion='1.35',policyVersion='v1.35',observations=records,projectionDelays={'overlap':overlap,'retirement':retirement},passed=True,cleanup=dict(namespaceRemoved=True,privateMaterialRemoved=not tmp.exists(),realCredentialsUsed=False),scope='Actual application mTLS between Pods using synthetic Secrets, hostname and trust failures, authorization403, trust overlap and replacement server generations. No mesh,CNI policy,sandbox,encryption-at-rest,revocation service,active-session termination,production recovery or full practical mock executed.')
out.write_text(json.dumps(report,indent=2)+'\n');print('PASS',len(records))
O bundle montado já inclui a CA nova, mas o servidor antigo continua a recusá-la; uma geração nova aceita ambos os clientes durante sobreposição.
Armadilhas comuns
Confundir projeção com adoção, autenticação com autorização, uma réplica com todo o serviço ou mTLS na aplicação com uma mesh testada.
Tópicos relacionados: Pod Security e recuperação · Secrets e rotação de chaves · mTLS e autorização · Isolamento e handover
A recuperação só está demonstrada quando os consumidores previstos funcionam e os limites de acesso continuam observáveis nas gerações relevantes.
Referência: CKS domains and exam details · Kubernetes v1.35; current six-domain CKS outline