APRENDER. PRATICAR. EVOLUIR.Definir a pergunta de investigação
Às 18:42, durante o fecho de um fundo fictício, surge um alerta de shell numa réplica de reconciliação. Existe uma janela de manutenção, mas ainda não há correspondência entre o alerta e a intervenção autorizada. APS precisa de manter o processamento enquanto segurança esclarece a atividade. Começa por formular perguntas observáveis: que instância executou o processo, que identidade iniciou a operação, que ficheiros mudaram e que período tem dados disponíveis? Regista hipóteses separadamente dos factos recolhidos. Um ticket aprovado não autoriza qualquer ação na mesma data. Compara ator, âmbito, horário e ação esperada. Um pedido negado pode ser apenas uma tentativa de diagnóstico mal configurada ou parte de uma sequência que merece escalada. Mantém alternativas abertas até existir evidência suficiente. Define quem decide contenção, quem valida continuidade e quando é necessário envolver o responsável de negócio. Esta aula liga pedidos à API, estado do container e qualidade da observação. A prática executa operações reais num cluster descartável, com valores sintéticos. O cenário empresarial, os prazos e as decisões são originais e não representam procedimentos internos da BNP Paribas. O objetivo é produzir uma conclusão defensável para a próxima decisão operacional, explicando o que foi observado, o que se infere e o que continua desconhecido.
Construir cobertura de auditoria verificável
Parte da operação que queres observar e verifica a política efetiva. Uma seleção de pods não cobre automaticamente pods/log. Quando um analista procura quem consultou logs, precisa de confirmar o subrecurso e o backend de auditoria; os Events usados para diagnosticar scheduling não oferecem a mesma evidência. Se uma regra anterior já corresponde ao pedido, uma regra posterior mais detalhada pode nunca ser aplicada. Revê a ordem com exemplos concretos. Na prática, Secrets ficam em Metadata e ConfigMaps sintéticos em RequestResponse. Assim, é possível comparar o pedido de alteração com o resultado sem copiar valores de credenciais para a evidência publicada. Em sistemas reais, avalia também respostas de list: um pedido sem corpo pode produzir uma coleção com dados sensíveis. Retenção curta reduz a duração da exposição, mas não elimina a criação desnecessária de cópias. Regista a união de omitStages global e por regra antes de explicar um estágio ausente. Um watch pode gerar diferentes estágios com o mesmo auditID; não contes automaticamente cada linha como uma ação distinta. Verifica rotação, recolha e pesquisa de um evento conhecido de ponta a ponta. Um ficheiro local preenchido não prova que o coletor conservou os ficheiros rodados nem que o analista os consegue consultar no intervalo do incidente.
Correlacionar pedidos sem inventar identidade
O laboratório faz um pedido usando impersonação autorizada pelo administrador do cluster de treino. A autorização da operação impersonada é recusada. O evento preserva user e impersonatedUser, permitindo distinguir o cliente autenticado da identidade usada para a decisão de acesso. Não transforma uma ServiceAccount numa pessoa. Para atribuição humana, seriam necessárias fontes adicionais da sessão e do sistema de identidade, com confiança e limites explícitos. User-Agent ajuda a procurar padrões, mas é informação fornecida pelo cliente. Uma string com kubectl não atesta o binário nem o dispositivo. Da mesma forma, sourceIPs exige conhecimento dos proxies e da origem dos cabeçalhos; um endereço encaminhado pode não ter a confiança do último salto observado. Conserva o registo original antes de normalizar campos para pesquisa e documenta a origem das informações acrescentadas. Não forces todos os pedidos para um esquema de objeto com metadata. Um JSON Patch pode conter um array de operações. Distingue requestObject, responseObject e annotations do próprio evento. Ao construir a linha temporal, separa instante do pedido, estágio e ingestão no coletor. Se houver atraso ou desvio de relógio, apresenta essa incerteza em vez de ordenar causalmente ações apenas pela chegada ao dashboard. A sequência deve continuar compreensível quando outra equipa revê os dados.
Distinguir imagem, filesystem e instância
Um digest identifica conteúdo de imagem, mas não descreve todo o estado gravável de uma execução. O exercício altera /tmp/dr-state e observa que a referência no Pod permanece igual. Depois compara um container com readOnlyRootFilesystem:true: a escrita em /tmp devolve EROFS, enquanto /scratch, montado num emptyDir, continua gravável. Os resultados identificam fronteiras concretas de proteção e não uma contradição da configuração. Quando uma aplicação necessita de escrita, localiza o caminho e o seu ciclo de vida. Distingue cache descartável, configuração aprovada e dados persistentes. Um chmod não transforma uma montagem só de leitura numa montagem gravável. Um volume deve ter apenas as permissões e o âmbito necessários, com responsabilidade definida sobre limpeza, capacidade e recuperação. Verifica também se a aplicação pode carregar código ou configuração a partir dessa área. Para investigar uma alteração, conserva ficheiro, caminho, montagem, UID do Pod, identidade do container e tempo. O nome do Pod pode reaparecer depois de uma recriação; o laboratório demonstra o novo UID. PIDs também precisam de contexto do namespace de processos. Um hash calculado depois do alerta identifica os bytes recolhidos, sem demonstrar que estão limpos. Compara com uma origem aprovada e conserva os dados necessários à análise, segundo o procedimento autorizado de recolha.
Desenhar uma deteção com âmbito explícito
Antes de escrever uma regra Falco, identifica a origem do evento. Campos de pedidos Kubernetes pertencem ao plugin k8s_audit; uma regra de syscalls não ganha esses campos só porque inclui ka.verb. A instalação do sensor, do plugin e do caminho de entrega precisa de validação própria. A prática desta aula não instala Falco: as regras são analisadas a partir da documentação primária, sem alegar deteção executada no motor. Escreve a intenção e exemplos antes da condição. Para A and (B or C), testa uma entrada que só satisfaz C e falha A, para detetar um agrupamento demasiado amplo. Uma exceção para rotate-agent em /var/app/checkpoints deve conservar a relação entre processo e destino. Excluir apenas o nome do processo também pode silenciar escritas fora do caminho autorizado. Testa uma ação permitida e uma ação semelhante que deve continuar a alertar. Um campo ausente não é necessariamente a string <NA>; confirma a semântica de extração na versão do plugin e usa exists quando a intenção é presença. Priority comunica gravidade, sem demonstrar ordem de avaliação ou contenção. Se a saída alimentar automação, valida separadamente consumidor, autorização, condição e resultado. O relatório deve dizer qual destes passos foi realmente exercitado e quais continuam apenas desenhados.
Medir a confiança na observação
Um dashboard sem alertas pode coincidir com uma falha de captura. Num pico de processamento, regista o crescimento de scap.n_drops e investiga carga, buffers, seleção de eventos e regras. Retirar regras pode reduzir trabalho e simultaneamente retirar cobertura. Compara o efeito com o requisito de deteção, sem aceitar apenas uma melhoria do indicador técnico. Se um contador cumulativo reinicia com o processo, não interpretes a diferença negativa como recuperação de eventos perdidos. Separa os ciclos de vida e calcula aumentos em cada um. Conserva o instante de restart e a janela sem observação suficiente. Distingue também perda na captura, perda na fila de saída e atraso no coletor. Cada fronteira pode exigir um teste diferente; um Pod Running não prova que a mensagem chegou ao destino pesquisável. O mesmo cuidado aplica-se à auditoria da API. Um evento pods/exec em Metadata não contém uma transcrição integral de stdout. Na prática, um marcador é enviado por stdin, impresso no cliente e procurado no audit log, onde não aparece. Isso demonstra uma limitação daquele canal e daquela configuração. Não demonstra ausência de execução nem substitui recolha de processos, ficheiros ou outra telemetria autorizada. Inclui estas limitações no handover para evitar que uma equipa seguinte trate silêncio como prova.
Executar a prática e interpretar os resultados
Usa um cluster kind dedicado com auditoria configurada, Docker, Python 3 e kubectl. O material do laboratório inclui audit-policy.json e instruções de configuração. Importa previamente uma imagem Python e regista o digest da plataforma realmente disponível. O script usa imagePullPolicy:Never e recusa contextos sem o prefixo kind-dr-cks-runtime-. Não o apliques a um ambiente partilhado: cria ConfigMaps, Secrets sintéticos e Pods num namespace que deve estar ausente. Guarda o código como run.py e executa python3 run.py --kubeconfig CAMINHO --context CONTEXTO --node NO-CONTROL-PLANE --image REFERENCIA@sha256:DIGEST --output RESULTADO-NOVO.json. Confirma os argumentos no README do laboratório. O script recusa sobrescrever resultados, remove o namespace no fim e conserva eventos da execução. A remoção do cluster é uma etapa separada, depois de guardar a evidência necessária. Os ensaios usam servidor Kubernetes 1.37.0 e cliente 1.37.1; a página oficial CKS indica runtime 1.35. Esta diferença está registada e não é apresentada como validação do runtime do exame. Interpreta as 14 observações pelo seu âmbito: autorização negada, ficheiros, montagem, UID e eventos de auditoria. Não há sensor Falco, captura de syscalls, contenção de rede, memória forense ou simulado prático completo nesta execução.
Decidir contenção e entregar a investigação
No segundo caso, um ficheiro num volume de reconciliação mudou e o sensor perdeu eventos no mesmo intervalo. O digest aprovado e o root filesystem só de leitura não resolvem a legitimidade dessa alteração. Existe uma réplica alternativa validada e o plano autoriza contenção dirigida. Coordena preservação de contexto com a medida aprovada, confirma continuidade e regista o período de observação incompleta. O laboratório não executa essa contenção; o caso avalia a decisão. Entrega a investigação com uma linha temporal curta, identificadores, fontes, hashes dos ficheiros recolhidos e ações executadas. Separa factos, hipóteses, decisões e próximos passos. Regista quem autoriza uma mudança, quem confirma capacidade e quando rever a medida. Evita reiniciar automaticamente antes da recolha só para obter um estado verde. Se a continuidade estiver em risco, aplica os critérios aprovados e documenta o efeito sobre evidência ainda não recolhida. Resumo: determina cobertura antes de interpretar ausência, conserva identidades distintas, correlaciona instâncias e avalia filesystem e volumes separadamente. Regras de deteção precisam de testes positivos e negativos; métricas precisam de contexto temporal. Liga esta aula a RBAC, admission, hardening Linux, cadeia de fornecimento e gestão de incidentes. A preparação CKS continua a exigir treino prático mais amplo e revisão especializada, além desta avaliação interna de decisões.
#!/usr/bin/env python3
"""Real API-audit and filesystem exercise for a dedicated kind training cluster.
Requires the supplied audit policy and an already imported Python image digest.
Only synthetic ConfigMaps/Secrets and owned Pods are created. No Falco sensor,
network containment, forensic memory capture or production identity is used.
"""
import argparse, base64, datetime, hashlib, json, pathlib, subprocess, time, uuid
p = argparse.ArgumentParser(description=__doc__)
p.add_argument('--kubeconfig', required=True)
p.add_argument('--context', required=True)
p.add_argument('--node', required=True)
p.add_argument('--image', required=True)
p.add_argument('--output', required=True)
p.add_argument('--kubectl', default='kubectl', help='Compatible kubectl executable')
a = p.parse_args()
if not a.context.startswith('kind-dr-cks-runtime-'):
raise SystemExit('Use a dedicated kind-dr-cks-runtime-* context.')
if a.node != a.context.removeprefix('kind-') + '-control-plane':
raise SystemExit('Node must belong to the named dedicated cluster.')
if '@sha256:' not in a.image:
raise SystemExit('Provide an already imported image by digest; this lab never pulls.')
out = pathlib.Path(a.output).resolve()
if out.exists():
raise SystemExit('Output already exists; choose a new report path.')
ns = 'dr-cks-runtime-lab'
prefix = 'trial-' + uuid.uuid4().hex[:8]
observations = []
def run(cmd, body=None, ok=True, timeout=40):
r = subprocess.run(cmd, input=body, text=True, capture_output=True, timeout=timeout)
if ok and r.returncode:
raise RuntimeError(f'{cmd[0]} failed: {r.stderr[-1500:]}')
return r
def k(*args, body=None, ok=True, timeout=40):
return run([a.kubectl, '--kubeconfig', a.kubeconfig, '--context', a.context, *args], body, ok, timeout)
def create(obj):
return json.loads(k('create', '-f', '-', '-o', 'json', body=json.dumps(obj)).stdout)
def observe(name, passed, **values):
observations.append(dict(name=name, passed=bool(passed), observed=values))
if not passed:
raise AssertionError(name + ': ' + str(values))
def pod(name, readonly):
return dict(apiVersion='v1', kind='Pod', metadata=dict(name=name, namespace=ns),
spec=dict(automountServiceAccountToken=False, restartPolicy='Never',
securityContext=dict(runAsNonRoot=True, runAsUser=10001, runAsGroup=10001,
fsGroup=10001, seccompProfile=dict(type='RuntimeDefault')),
containers=[dict(name='worker', image=a.image, imagePullPolicy='Never',
command=['python3', '-c', 'import time; time.sleep(900)'],
securityContext=dict(allowPrivilegeEscalation=False,
readOnlyRootFilesystem=readonly, capabilities=dict(drop=['ALL'])),
volumeMounts=[dict(name='scratch', mountPath='/scratch')])],
volumes=[dict(name='scratch', emptyDir={})]))
def execute(name, script):
return k('-n', ns, 'exec', name, '-c', 'worker', '--', 'python3', '-c', script)
report = dict(startedAt=datetime.datetime.now(datetime.timezone.utc).isoformat(),
scriptSha256=hashlib.sha256(pathlib.Path(__file__).read_bytes()).hexdigest(),
version=json.loads(k('version', '-o', 'json').stdout), context=a.context,
imageReference=a.image, namespace=ns, trial=prefix, observations=observations,
examVersion='1.35', fullPracticalMock=False, independentVerification=False,
scope='Actual isolated API audit events, authorization denial and container filesystem writes. No Falco engine, syscall capture, network isolation, memory forensics or production incident.')
client_minor = int(report['version']['clientVersion']['minor'].rstrip('+'))
server_minor = int(report['version']['serverVersion']['minor'].rstrip('+'))
if abs(client_minor - server_minor) > 1:
raise SystemExit('Use kubectl within one minor version of the API server; set --kubectl explicitly.')
existing = k('get', 'namespace', ns, ok=False)
if existing.returncode == 0 or 'NotFound' not in existing.stderr:
raise SystemExit('Namespace exists or absence is unconfirmed; refusing to reuse it.')
owned = False
try:
create(dict(apiVersion='v1', kind='Namespace', metadata=dict(name=ns)))
owned = True
cm = prefix + '-settings'
secret = prefix + '-credential'
marker = 'synthetic-only-' + uuid.uuid4().hex
config = create(dict(apiVersion='v1', kind='ConfigMap', metadata=dict(name=cm, namespace=ns), data=dict(mode='approved')))
create(dict(apiVersion='v1', kind='Secret', metadata=dict(name=secret, namespace=ns), stringData=dict(token=marker)))
k('-n', ns, 'get', 'secret', secret, '-o', 'json')
k('-n', ns, 'patch', 'configmap', cm, '--type=json', '-p', '[{"op":"replace","path":"/data/mode","value":"changed"}]')
impersonation = 'system:serviceaccount:' + ns + ':unprivileged-reader'
denied = k('-n', ns, 'get', 'configmap', cm, '--as', impersonation, ok=False)
observe('unauthorized-read-denied', denied.returncode != 0 and 'Forbidden' in denied.stderr,
clientExitCode=denied.returncode, impersonatedIdentity=impersonation)
mutable = prefix + '-mutable'
immutable = prefix + '-readonly'
initial = create(pod(mutable, False))
create(pod(immutable, True))
k('-n', ns, 'wait', '--for=condition=Ready', 'pod/' + mutable, 'pod/' + immutable, '--timeout=90s', timeout=100)
write = "import pathlib,hashlib,json; p=pathlib.Path('/tmp/dr-state'); p.write_text('approved'); before=hashlib.sha256(p.read_bytes()).hexdigest(); p.write_text('changed'); after=hashlib.sha256(p.read_bytes()).hexdigest(); print(json.dumps({'before':before,'after':after,'changed':before!=after}))"
change = json.loads(execute(mutable, write).stdout)
observe('mutable-root-file-changed', change['changed'], **change)
checked = json.loads(k('-n', ns, 'get', 'pod', mutable, '-o', 'json').stdout)
observe('image-reference-unchanged-after-write', checked['spec']['containers'][0]['image'] == a.image,
declaredImage=checked['spec']['containers'][0]['image'], uid=checked['metadata']['uid'],
imageID=checked['status']['containerStatuses'][0]['imageID'])
readonly = "import pathlib,json,errno\ntry:\n pathlib.Path('/tmp/dr-state').write_text('changed'); result={'denied':False}\nexcept OSError as e:\n result={'denied':e.errno==errno.EROFS,'errno':e.errno}\nprint(json.dumps(result))"
result = json.loads(execute(immutable, readonly).stdout)
observe('readonly-root-write-denied', result['denied'], **result)
scratch = "import pathlib,json; p=pathlib.Path('/scratch/dr-state'); p.write_text('synthetic'); print(json.dumps({'writable':p.read_text()=='synthetic'}))"
result = json.loads(execute(immutable, scratch).stdout)
observe('explicit-volume-still-writable', result['writable'], **result)
# Prove that the API records an exec request, not a transcript of its stdout.
stdout_marker = 'stdout-only-' + uuid.uuid4().hex
result = k('-n', ns, 'exec', '-i', mutable, '-c', 'worker', '--', 'python3', '-c',
'import sys; print(sys.stdin.read())', body=stdout_marker)
observe('exec-returned-synthetic-output', stdout_marker in result.stdout, outputMatched=True)
# Watch uses an API timeout and generates stages that can share an auditID.
uri = '/api/v1/namespaces/' + ns + '/configmaps?watch=true&timeoutSeconds=2'
k('get', '--raw', uri, timeout=10)
old_uid = checked['metadata']['uid']
k('-n', ns, 'delete', 'pod', mutable, '--wait=true', '--timeout=45s', timeout=50)
recreated = create(pod(mutable, False))
observe('recreated-name-has-new-uid', recreated['metadata']['uid'] != old_uid,
objectName=mutable, previousUID=old_uid, currentUID=recreated['metadata']['uid'])
# Read only the owned control-plane log, then select this trial namespace.
events = []
for _ in range(30):
raw = run(['docker', 'exec', a.node, 'cat', '/var/log/dr-audit/audit.log']).stdout
candidates = [json.loads(line) for line in raw.splitlines() if line.strip()]
events = [e for e in candidates if e.get('objectRef', {}).get('namespace') == ns
and e.get('requestReceivedTimestamp', '') >= report['startedAt'].replace('+00:00', 'Z')]
if any(e.get('verb') == 'watch' and e.get('stage') == 'ResponseComplete' for e in events):
break
time.sleep(0.2)
def select(resource, verb=None, name=None):
return [e for e in events if e.get('objectRef', {}).get('resource') == resource
and (verb is None or e.get('verb') == verb)
and (name is None or e.get('objectRef', {}).get('name') == name)]
created = select('configmaps', 'create', cm)
observe('configmap-request-response-recorded', any(e.get('level') == 'RequestResponse'
and e.get('requestObject', {}).get('data', {}).get('mode') == 'approved'
and e.get('responseObject', {}).get('metadata', {}).get('uid') == config['metadata']['uid']
and e.get('responseStatus', {}).get('code') == 201 for e in created), events=len(created))
patches = select('configmaps', 'patch', cm)
observe('json-patch-array-recorded', any(isinstance(e.get('requestObject'), list)
and e.get('responseObject', {}).get('data', {}).get('mode') == 'changed'
and e.get('responseStatus', {}).get('code') == 200 for e in patches), events=len(patches))
secret_events = select('secrets', name=secret)
observe('secret-bodies-omitted', len(secret_events) >= 2 and marker not in raw
and base64.b64encode(marker.encode()).decode() not in raw and all(e['level'] == 'Metadata'
and 'requestObject' not in e and 'responseObject' not in e for e in secret_events),
events=len(secret_events), secretMarkerAbsent=marker not in raw,
encodedSecretMarkerAbsent=base64.b64encode(marker.encode()).decode() not in raw)
denied_events = [e for e in select('configmaps', 'get', cm) if e.get('responseStatus', {}).get('code') == 403]
observe('impersonation-attribution-retained', any(e.get('impersonatedUser', {}).get('username') == impersonation
and e.get('user', {}).get('username') != impersonation for e in denied_events),
attribution=[dict(user=e.get('user', {}).get('username'),
impersonatedUser=e.get('impersonatedUser', {}).get('username'),
auditID=e.get('auditID'), code=e.get('responseStatus', {}).get('code')) for e in denied_events])
watches = select('configmaps', 'watch')
grouped = {}
for e in watches:
grouped.setdefault(e['auditID'], []).append(e['stage'])
observe('watch-stages-share-request-id', any('ResponseStarted' in stages and 'ResponseComplete' in stages
for stages in grouped.values()), requestStages=grouped)
observe('requestreceived-omitted-by-policy', bool(events) and all(e['stage'] != 'RequestReceived' for e in events),
retainedEvents=len(events))
exec_events = [e for e in select('pods') if e.get('objectRef', {}).get('subresource') == 'exec']
observe('exec-metadata-without-stdout-transcript', bool(exec_events) and stdout_marker not in raw,
execAuditEvents=len(exec_events), stdoutMarkerAbsent=stdout_marker not in raw)
report['events'] = events
report['auditExportSha256'] = hashlib.sha256(json.dumps(events, sort_keys=True).encode()).hexdigest()
report['passed'] = all(o['passed'] for o in observations)
finally:
if owned:
k('delete', 'namespace', ns, '--wait=true', '--timeout=90s', timeout=100)
absent = k('get', 'namespace', ns, ok=False)
report['cleanup'] = dict(namespaceRemoved=absent.returncode != 0 and 'NotFound' in absent.stderr,
realCredentialsUsed=False, clusterDeletionSeparate=True)
report['finishedAt'] = datetime.datetime.now(datetime.timezone.utc).isoformat()
out.write_text(json.dumps(report, indent=2) + '\n')
print(json.dumps(dict(passed=report['passed'], observations=len(observations), report=str(out))))
Um ficheiro num volume muda apesar do digest aprovado; a equipa preserva o contexto, reconhece eventos perdidos e decide contenção dirigida.
Armadilhas comuns
Confundir imagem com estado do volume, nome com UID, User-Agent com identidade ou ausência de alertas com cobertura completa.
Tópicos relacionados: RBAC e identidades · Hardening e volumes · Resposta a incidentes
Uma investigação útil preserva contexto, explica a confiança em cada fonte e suporta a próxima decisão operacional.
Referência: CKS certification and domains · Kubernetes v1.35; current six-domain CKS outline