APRENDER. PRATICAR. EVOLUIR.1. Definir o que foi contido
Às 17:20, a equipa de segurança encontra uma credencial do dashboard num artefacto de diagnóstico. O fecho de fundos ocorre às 17:30. A equipa APS retira uma RoleBinding e obtém 403 num novo pedido. Essa evidência é útil, mas a frase “a conta está bloqueada” ainda é imprecisa. É necessário indicar a credencial testada, a operação, o namespace, o objeto, o endpoint e o momento. O caso é fictício e não descreve procedimentos internos de um banco. Constrói uma matriz com quatro perguntas: o servidor reconhece a identidade; permite a operação nova; continua a entregar dados numa ligação anterior; existem cópias fora da API? Cada linha tem prova própria. Neste módulo, TokenReview positivo responde à primeira pergunta, GET ou watch novo à segunda e um marcador criado depois da alteração à terceira. A quarta exige investigação do cliente e dos destinos de exportação, fora do laboratório. Antes de alterar, regista o estado mínimo necessário para explicar o incidente, sem guardar o token no ticket. Combina um teste permitido com outro que deve falhar. Depois da alteração, repete ambos com a identidade correta. Um administrador que consegue ler o objeto demonstra disponibilidade para si, não a ausência de acesso do atacante. Fecha a comunicação com uma conclusão delimitada: “novos GET desta credencial ao objeto foram recusados às 17:23; o stream antigo ainda exige contenção”.
2. Retirar concessões sem as repor por acidente
Uma alteração RBAC deve considerar quem gere o objeto. No exemplo do dashboard, a RoleBinding removida reaparece e o audit log identifica um controlador GitOps. Repetir a eliminação deixa uma corrida entre operador e reconciliador. Corrige a origem declarativa, coordena a reconciliação e verifica o objeto efetivo. Documenta a exceção temporária para que a próxima sincronização não restaure uma concessão comprometida. Existe um mecanismo diferente nos papéis padrão do Kubernetes: o API server pode repor regras e subjects em falta no arranque. Antes de alterar objetos geridos pelo control plane, identifica a finalidade e as dependências. Uma anotação que desative auto-reconciliação não é uma receita universal de hardening. Um papel usado por componentes essenciais pode precisar de direitos que parecem amplos quando lidos sem contexto. Testa o comportamento pretendido e a continuidade dos componentes numa mudança aprovada. No fecho de fundos, a recuperação não exige devolver o papel original à conta exposta. Define a leitura de estado concreta, atribui-a a uma identidade limpa e limita o âmbito. Regista responsável, prazo e prova de remoção posterior. Se for necessário alargar temporariamente o acesso, explicita a necessidade e quem aceita o risco. Uma concessão “só durante o incidente” sem prazo verificável tende a tornar-se permanente. Confirma ainda se outra binding ou outro autorizador continua a permitir a operação que querias bloquear.
3. Escolher a unidade de revogação
A prática cria dois tokens para a mesma ServiceAccount: um ligado a um Secret de controlo e outro sem ligação a objeto. O Secret contém apenas texto sintético; não é o local onde guardamos o token. Ao eliminar esse objeto, o token ligado passa a ser rejeitado, enquanto o outro continua utilizável quando as permissões regressam. O nome da conta igual não torna iguais as dependências de todas as suas credenciais. A seguir, recriamos o Secret com o mesmo nome. O token antigo continua rejeitado, porque o novo objeto tem outro UID. Emitir um token ligado à instância nova permite voltar a testar o acesso legítimo. No final, eliminar a própria ServiceAccount afeta ambos os tipos de token. A escolha entre invalidar uma ligação e retirar toda a identidade depende do inventário e do impacto. A equipa deve saber que processos partilham a conta antes de usar uma medida ampla. Desativar automount reduz a montagem automática em Pods; não elimina uma credencial anteriormente emitida nem impede um emissor autorizado de pedir outra. A montagem do Pod tem regras próprias e não foi executada nesta prática. Também não medimos a tolerância temporal associada a objetos pendentes de eliminação com finalizers. A documentação trata essa situação separadamente. Nos testes, aguarda-se a propagação e verifica-se cada credencial individualmente, com limite de tempo, em vez de anunciar revogação instantânea universal.
4. Preparar e executar a prática isolada
Guarda o código desta aula como run.py. Precisas de Python 3, kind, Docker e kubectl compatível. Usa um cluster descartável exclusivo com nome iniciado por dr-cks-identity-response- e um kubeconfig separado, de permissões 0600. A execução de referência usou kind 0.33.0, node image kindest/node:v1.37.0@sha256:a1ed56cfb0e7b93589bdf97c8cd566405a265939e3620fc4f5de89adff580ae5 e kubectl 1.37.1. O exame oficial indica runtime 1.35; esta diferença é explícita e o script recusa versões diferentes das verificadas. Não aponta para o contexto habitual da tua estação. Cria o cluster com kind create cluster --name dr-cks-identity-response-estudo --image IMAGEM --kubeconfig /caminho/absoluto/kubeconfig, substituindo IMAGEM pelo identificador acima. Passa depois a python3 run.py os argumentos --kubectl /caminho/absoluto/kubectl, --kubeconfig /caminho/absoluto/kubeconfig, --cluster dr-cks-identity-response-estudo e --output /caminho/novo/resultado.json. O ficheiro de resultado não pode existir. Confirma as versões antes de executar e não alteres as asserções só para obter um resultado verde. O script cria um namespace aleatório, uma conta, um Secret, um ConfigMap e RBAC limitado. Os tokens ficam na memória e em stdin, sem argumentos de comando com credenciais. As respostas completas de TokenReview não são publicadas. O finally remove o namespace e o cache privado. Quando terminares, elimina apenas o cluster dedicado com kind delete cluster --name dr-cks-identity-response-estudo e remove o seu kubeconfig. Se houver falha, inspeciona a causa e a limpeza antes de repetir com um novo ficheiro de resultados.
5. Interpretar o stream e a sequência de resultados
Antes de executar, prevê os resultados em papel. Primeiro, um token autentica mas não consegue ler. Depois da binding, a leitura funciona. Abre-se um watch a partir da resourceVersion observada e publica-se um marcador inicial. Após retirar RBAC, GET e watch novos falham; um segundo marcador, criado depois dessas recusas, chega pelo stream já aberto. Eliminar o Secret ligado faz novos pedidos devolverem 401, mas um terceiro marcador ainda chega pelo mesmo stream. Esta sequência distingue um evento novo de dados antigos que já estavam num buffer. O código não se limita a verificar que o processo cliente existe. Observa a entrega do valor sintético esperado depois de cada mudança. O watch tem timeoutSeconds limitado; o leitor termina antes de se fechar o respetivo objeto de resposta. Não executámos uma terminação forçada no servidor nem demonstrámos um procedimento universal para proxies, balanceadores ou versões diferentes. O relatório contém 23 observações, versões, hash do script e tempos dos probes de propagação. Compara o hash com o código que executaste antes de associar evidência à aula. Se obtiveres um resultado diferente, regista configuração, versão e ordem exata; não o escondas. Num ambiente real, a resposta pode exigir terminar ligações por mecanismos suportados ou aplicar contenção mais ampla autorizada. A aceitação precisa de provar o fim da entrega e de verificar que o fluxo legítimo de recuperação continua disponível. Nas duas execuções de referência, o probe que aguardou 401 após eliminar o Secret demorou cerca de 9,8 segundos. É uma medição deste ambiente, não um SLA nem uma regra universal de cache ou revogação. O limite de 15 segundos do script é um limite do ensaio.
6. Guardar evidência sem voltar a expor a credencial
TokenReview recebe uma credencial e devolve um veredicto; a resposta pode repetir spec.token. Publicar o JSON completo num ticket, log ou relatório recria o problema que estás a resolver. O laboratório retém apenas o booleano de autenticação. Num procedimento operacional, acrescenta identificadores não secretos aprovados, hora, audience testada e conclusão, com acesso e retenção adequados. Não uses descodificadores públicos de JWT para investigar tokens reais. Há duas perguntas diferentes ao analisar um JWT: os bytes assinados são válidos para este destinatário e a ligação ao objeto ainda é válida agora? A verificação offline pode responder à primeira, mas não consulta automaticamente o estado atual do cluster. TokenReview permite consultar o API server. Se um serviço depende desse estado para aceitar chamadas, define audience, tratamento de indisponibilidade e duração de cache. Um resultado antigo em cache não equivale a uma verificação efetuada neste instante. Esta prática não implementa um verificador criptográfico offline. Ler ou descodificar claims, por si só, não valida a assinatura. Também não permite recuperar revisões antigas com kubectl get: TokenReview não é armazenado como um recurso normal consultável. Para o exercício, entrega uma tabela de resultados sem tokens, identifica as verificações negativas e explica o que cada uma prova. Uma captura que omite identidade, operação ou momento pode ser visualmente convincente e ainda assim insuficiente para fechar o incidente.
7. Tratar autenticação externa e exposição da API
O mecanismo exercitado usa ServiceAccounts locais ao cluster. Não generalizes os resultados para o login de um utilizador humano. Num modelo OIDC com JWT validado localmente, terminar a sessão no IdP não informa obrigatoriamente o API server em cada pedido. Um token já emitido pode continuar a conter um grupo retirado do diretório. Antes de afirmar revogação, confirma o modelo de integração, a validade restante e o resultado de pedidos com a credencial visada. O laboratório não executa IdP nem prova comportamento de um fornecedor específico. Também distingue uma rota de saúde de uma rota de dados. No cluster descartável, /livez sem credencial responde 200 e a leitura anónima do ConfigMap recebe 403. Não concluímos daí que todos os clusters devam ter a mesma exposição. Em v1.35, AuthenticationConfiguration permite delimitar os caminhos que aceitam autenticação anónima. É necessário combinar esse desenho com autorização, necessidades do balanceador e testes negativos de outras rotas. Imagina uma alteração que fecha o acesso anónimo indiscriminadamente e retira todos os API servers do balanceador. A intenção de segurança não elimina o impacto operacional. Define primeiro que probe é usado, por que origem, com que identidade e por que caminho. Testa a mudança num âmbito controlado e prepara recuperação aprovada. O relatório de aceitação deve incluir a proteção dos recursos e a saúde do caminho de gestão, porque perder esse caminho também pode impedir a resposta a incidentes.
8. Atualizar, recuperar e entregar ao RUN
A resposta a vulnerabilidades exige confirmar componente afetado, versão, condições de exploração e correção indicada na fonte oficial. Uma lista de versões, isoladamente, não demonstra exposição nem eficácia da mudança. Associa cada ação ao risco concreto e evita inventar um CVE para justificar o exercício. Os cenários desta aula usam transições hipotéticas; não executámos upgrade de Kubernetes. Planeia a ordem suportada, incluindo todos os API servers que cada consumidor pode alcançar. Num estado HA com 1.34 e 1.35, um kubelet 1.35 seria mais recente do que o servidor 1.34 ainda alcançável. A tolerância a componentes mais antigos não autoriza inverter essa relação. Não saltes minors do API server e verifica restrições adicionais da ferramenta ou fornecedor. Webhooks e clientes fazem parte da aceitação, além do número mostrado por kubectl version. Entrega ao RUN uma matriz com estado anterior, mudança, resultado esperado, resultado observado, dono e limitação. Inclui credenciais antigas, credencial limpa de recuperação, GET novo, watch novo e stream anterior. Acrescenta o destino de recuperação aprovado, prazo da exceção e confirmação do negócio. A avaliação final deve pedir decisões fundamentadas: que evidência falta, que ação cabe à equipa e o que não foi provado? Resume o princípio: conter uma identidade requer observar credenciais, direitos, ligações e dados já obtidos. Os quatro controlos relacionados não são intercambiáveis e cada um precisa de critérios de encerramento.
"""Original CKS lab: permission withdrawal, bound-token invalidation and watches.
Use only a dedicated kind cluster. Tokens stay in process memory and stdin;
the report contains status codes and synthetic object observations, not tokens.
"""
import argparse
import base64
import datetime
import hashlib
import json
import queue
import shutil
import ssl
import subprocess
import tempfile
import threading
import time
import urllib.error
import urllib.request
import uuid
from pathlib import Path
p = argparse.ArgumentParser()
for name in ('kubectl', 'kubeconfig', 'cluster', 'output'):
p.add_argument('--' + name, required=True)
a = p.parse_args()
assert a.cluster.startswith('dr-cks-identity-response-')
assert Path(a.kubeconfig).is_absolute()
out = Path(a.output)
assert not out.exists()
private = Path(tempfile.mkdtemp(prefix='dr-cks-identity-response-private-'))
ns = 'dr-response-' + uuid.uuid4().hex[:8]
base = [a.kubectl, '--kubeconfig', a.kubeconfig, '--context', 'kind-' + a.cluster,
'--cache-dir', str(private / 'cache'), '--request-timeout=15s']
tokens, records, streams, settlements = {}, [], [], []
created = False
def k(*args, obj=None):
result = subprocess.run(base + list(args), text=True, capture_output=True,
input=json.dumps(obj) if obj is not None else None, timeout=30)
if result.returncode:
# Do not echo input or full output from a token-bearing API operation.
raise RuntimeError('Admin operation failed: ' + str(args[:3]))
return result.stdout
def resource(kind, name, **fields):
group = 'rbac.authorization.k8s.io/v1' if kind in ('Role', 'RoleBinding') else 'v1'
return dict(apiVersion=group, kind=kind, metadata=dict(name=name), **fields)
def create(obj):
return json.loads(k('-n', ns, 'create', '-f', '-', '-o', 'json', obj=obj))
def binding():
return resource('RoleBinding', 'read-status',
roleRef=dict(apiGroup='rbac.authorization.k8s.io', kind='Role', name='read-status'),
subjects=[dict(kind='ServiceAccount', name='reporter', namespace=ns)])
def token(name, secret=None):
args = ['-n', ns, 'create', 'token', 'reporter', '--duration=10m']
if secret:
args += ['--bound-object-kind=Secret', '--bound-object-name=token-anchor',
'--bound-object-uid=' + secret]
tokens[name] = k(*args).strip()
def request(who, path):
headers = {'Authorization': 'Bearer ' + tokens[who]} if who else {}
req = urllib.request.Request(server + path, headers=headers)
try:
with urllib.request.urlopen(req, context=ctx, timeout=15) as response:
return response.status, response.read()
except urllib.error.HTTPError as error:
return error.code, error.read()
def record(name, observed, expected):
passed = observed == expected
records.append(dict(name=name, observed=observed, expected=expected, passed=passed))
print(name, 'PASS' if passed else 'FAIL', flush=True)
assert passed, name
def check(name, who, path, expected):
status, _ = request(who, path)
record(name, status, expected)
def settle(who, path, expected):
started = time.monotonic()
deadline = started + 15
while time.monotonic() < deadline:
if request(who, path)[0] == expected:
settlements.append(dict(credentialLabel=who, expected=expected, elapsedSeconds=round(time.monotonic()-started, 3)))
return
time.sleep(0.2)
raise AssertionError('Propagation deadline exceeded')
def review(name, who, expected):
result = json.loads(k('create', '-f', '-', '-o', 'json', obj={
'apiVersion': 'authentication.k8s.io/v1', 'kind': 'TokenReview',
'spec': {'token': tokens[who]}}))
# The response echoes spec.token. Retain only the authentication verdict.
record(name, result.get('status', {}).get('authenticated', False), expected)
def start_watch(who, resource_version):
events = queue.Queue()
path = watch_path + '&resourceVersion=' + resource_version
req = urllib.request.Request(server + path, headers={'Authorization': 'Bearer ' + tokens[who]})
stream = urllib.request.urlopen(req, context=ctx, timeout=30)
streams.append(stream)
def reader():
try:
for line in stream:
event = json.loads(line)
obj = event.get('object', {})
if event.get('type') == 'MODIFIED':
events.put(obj.get('data', {}).get('state'))
except (OSError, ValueError):
pass
thread = threading.Thread(target=reader, daemon=True)
thread.start()
return events, thread
def update_state(state):
k('-n', ns, 'patch', 'configmap', 'closing-status', '--type=merge',
'-p', json.dumps({'data': {'state': state}}))
try:
cfg = json.loads(k('config', 'view', '--minify', '--raw', '-o', 'json'))
cluster = cfg['clusters'][0]['cluster']
server = cluster['server']
assert server.startswith('https://127.0.0.1:')
assert not cluster.get('insecure-skip-tls-verify')
ctx = ssl.create_default_context(cadata=base64.b64decode(cluster['certificate-authority-data']).decode())
cfg = None
versions = json.loads(k('version', '-o', 'json'))
assert versions['serverVersion']['gitVersion'] == 'v1.37.0'
assert versions['clientVersion']['gitVersion'] == 'v1.37.1'
k('create', 'namespace', ns)
created = True
create(resource('ServiceAccount', 'reporter', automountServiceAccountToken=False))
anchor = create(resource('Secret', 'token-anchor', type='Opaque', stringData={'purpose': 'synthetic-revocation-anchor'}))
first_uid = anchor['metadata']['uid']
token('bound', first_uid)
token('unbound')
create(resource('ConfigMap', 'closing-status', data={'state': 'initial'}))
core = '/api/v1/namespaces/' + ns
target = core + '/configmaps/closing-status'
watch_path = core + '/configmaps?watch=true&timeoutSeconds=25&fieldSelector=metadata.name%3Dclosing-status'
review('explicit-token-valid-with-automount-disabled', 'bound', True)
check('valid-identity-without-permission', 'bound', target, 403)
create(resource('Role', 'read-status', rules=[dict(apiGroups=[''], resources=['configmaps'],
resourceNames=['closing-status'], verbs=['get', 'watch'])]))
create(binding())
settle('bound', target, 200)
check('bound-token-permitted', 'bound', target, 200)
check('unbound-token-same-permission', 'unbound', target, 200)
rv = json.loads(request('bound', target)[1])['metadata']['resourceVersion']
events, thread = start_watch('bound', rv)
update_state('before-revocation')
record('watch-receives-before-revocation', events.get(timeout=10), 'before-revocation')
k('-n', ns, 'delete', 'rolebinding', 'read-status')
settle('bound', target, 403)
check('new-get-denied-after-rbac-withdrawal', 'bound', target, 403)
check('new-watch-denied-after-rbac-withdrawal', 'bound', watch_path, 403)
review('identity-still-valid-after-rbac-withdrawal', 'bound', True)
update_state('after-rbac-withdrawal')
record('existing-watch-receives-after-rbac-withdrawal', events.get(timeout=10), 'after-rbac-withdrawal')
k('-n', ns, 'delete', 'secret', 'token-anchor')
settle('bound', target, 401)
check('new-request-rejects-deleted-anchor', 'bound', target, 401)
review('review-rejects-deleted-anchor', 'bound', False)
update_state('after-anchor-deletion')
record('existing-watch-receives-after-anchor-deletion', events.get(timeout=10), 'after-anchor-deletion')
# Let the bounded watch finish before closing its response object.
# This does not exercise an emergency server-side forced disconnect.
thread.join(timeout=30)
for stream in streams:
stream.close()
record('watch-reader-ended', thread.is_alive(), False)
anchor = create(resource('Secret', 'token-anchor', type='Opaque', stringData={'purpose': 'new-synthetic-anchor'}))
record('recreated-anchor-has-new-uid', anchor['metadata']['uid'] != first_uid, True)
create(binding())
settle('unbound', target, 200)
check('unbound-token-survives-unrelated-anchor-deletion', 'unbound', target, 200)
check('same-name-does-not-restore-old-bound-token', 'bound', target, 401)
review('review-rejects-old-uid-after-recreation', 'bound', False)
token('replacement', anchor['metadata']['uid'])
review('replacement-token-authenticated', 'replacement', True)
check('replacement-token-permitted', 'replacement', target, 200)
check('anonymous-health-path', None, '/livez', 200)
check('anonymous-workload-read-denied', None, target, 403)
k('-n', ns, 'delete', 'serviceaccount', 'reporter')
settle('unbound', target, 401)
check('account-deletion-rejects-unbound-token', 'unbound', target, 401)
settle('replacement', target, 401)
check('account-deletion-rejects-replacement-token', 'replacement', target, 401)
assert len(records) == 23
finally:
for stream in streams:
stream.close()
tokens.clear()
if created:
k('delete', 'namespace', ns, '--wait=true', '--timeout=60s')
shutil.rmtree(private)
report = dict(executedAt=datetime.datetime.now(datetime.timezone.utc).isoformat(),
scriptSha256=hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),
versions=versions, examVersion='1.35', passed=all(r['passed'] for r in records),
observations=records, propagationChecks=settlements,
cleanup=dict(namespaceRemoved=True, privateCacheRemoved=not private.exists(),
tokensRecorded=False, realCredentialsUsed=False, watchReaderEnded=not thread.is_alive()),
scope='Actual HTTPS requests, TokenReview, Secret-bound TokenRequest, RBAC withdrawal and an existing watch on disposable kind server1.37.0/client1.37.1. Exam runtime1.35 differs. No OIDC provider, offline cryptographic JWT validation, finalizer grace timing, workload execution, server-side forced disconnect or upgrade was exercised.')
out.write_text(json.dumps(report, indent=2) + '\n')
print('PASS', len(records))
Um watch antigo recebe um marcador novo após a RoleBinding ser retirada; um novo watch recebe 403.
Armadilhas comuns
Confundir 403 com token inválido, automount com revogação, nome com UID e recusa nova com encerramento de streams.
Tópicos relacionados: RBAC e reconciliação · TokenRequest e TokenReview · Gestão de incidentes e continuidade · Atualizações e compatibilidade
Mede separadamente a identidade, a permissão, o pedido em curso e a exposição de dados já obtidos.
Referência: CKS certification and domains · Kubernetes v1.35; current six-domain CKS outline