APRENDER. PRATICAR. EVOLUIR.Delimitar o incidente e a confiança nos sinais
Durante um incidente, a ausência de alertas admite várias explicações. A atividade pode não ter ocorrido, a recolha pode ter falhado, uma regra pode ter sido desativada ou a entrega pode estar atrasada. Antes de concluir, identifica o caminho entre a atividade e a decisão: API server, política de auditoria, transporte, plugin, regras, saída e destino. Para cada fronteira, regista o que observaste e o que ainda depende de confirmação. Um processo saudável não prova que todos estes passos funcionam. Num caso bancário fictício, APS acompanha o fecho de fundos enquanto segurança investiga uma alteração nas regras do Falco. O responsável de negócio precisa de saber se pode continuar o processamento e qual o risco da lacuna. Prepara uma linha temporal com a última configuração conhecida, a alteração, a primeira observação inesperada e os intervalos de retenção disponíveis. Não atribuas o comportamento a uma pessoa só porque uma conta aparece num registo. A identidade autenticada é uma pista a relacionar com acesso, contexto e outras evidências. Define perguntas concretas para a reunião: que operações deveriam gerar sinal, quais foram observadas na origem e quais chegaram ao destino? Que exceção estava autorizada, para que conta, alvo e período? Qual a decisão de continuidade até ao próximo checkpoint? O caso é didático e não representa procedimentos internos da BNP Paribas. O objetivo é praticar a coordenação entre operação, segurança e negócio sem transformar informação incompleta numa garantia.
Rever o conjunto de regras realmente aplicado
Uma regra correta num ficheiro pode não participar na avaliação pretendida. Começa por identificar a versão do Falco, os plugins carregados, os ficheiros efetivos e a sua ordem. Um override que acrescenta uma condição depende de uma definição anterior. Replace exige atenção particular: substituir a condição pode perder os limites de namespace, recurso ou verbo que o autor anterior tinha estabelecido. Revê a expressão completa resultante, incluindo parênteses, em vez de aprovar apenas a linha alterada. Com rule_matching: first, uma regra geral pode corresponder antes de uma regra específica. Aumentar a prioridade desta última não altera a ordem. All permite mais correspondências, mas pode aumentar volume e fazer várias regras descreverem o mesmo evento. Distingue eventos, alertas e incidentes no relatório. O limiar global priority também tem efeito próprio: priority: error não inclui uma regra WARNING. Log_level: debug aumenta detalhe operacional, sem alargar o conjunto de regras selecionado pelo limiar. Uma exceção de manutenção deve refletir a combinação autorizada. Se ops-a pode atuar apenas no namespace de ensaio, excluir toda a conta permite suprimir atividades fora desse âmbito. Prepara exemplos que diferem num só campo: mesma conta noutro namespace, outra conta no alvo autorizado e outro verbo sobre o mesmo recurso. Esses controlos tornam visíveis limites que um único exemplo positivo não demonstra. Regista também as opções de ativação e desativação da linha de comando, cuja ordem participa no resultado efetivo.
Separar validação, reprodução e operação real
Validar a sintaxe e as dependências de regras é necessário, mas não demonstra que uma entrada produza o resultado esperado. Um dry run não processa eventos para provar essa correspondência. Além disso, skip-if-unknown-filter: true pode permitir carregar uma regra sem a executar quando contém um campo desconhecido. Confirma os campos do plugin efetivamente instalado, observa uma entrada que deve corresponder e inclui entradas que devem ser excluídas. Ausência de alerta num caso negativo não distingue uma regra correta de uma regra inativa. Para o exercício proposto, usa apenas eventos sintéticos originais no formato audit.k8s.io/v1. Dá a cada pedido um auditID distinto e inclui estágio, identidade, verbo, alvo e resultado quando aplicável. O plugin k8saudit pode ler um caminho local sem esquema até ao fim do ficheiro; file:// tem comportamento de acompanhamento contínuo e não deve ser confundido com essa entrada finita. Guarda a versão do plugin, a configuração, as regras e um hash da entrada para repetir a comparação. Prevê testes para ordem de regras, exceção delimitada, override e severidade mínima. Compara os identificadores observados com os esperados. O laboratório executou Falco 0.45.0 com k8saudit 0.18.0 em duas execuções finais de 41 verificações cada, usando exatamente o código apresentado. Confirmou correspondências, exclusões e falhas esperadas; uma execução inicial revelou que o teste de versão precisava de interpretar JSON. Continua a ser necessário avaliar separadamente o transporte do API server ao destino real, captura de syscalls e contenção no cluster. Mantém essas fronteiras explícitas no relatório de prontidão. A imagem e o plugin estão fixados por digest e checksum. O ensaio corre sem rede, como utilizador sem privilégios, com filesystem só de leitura e apenas os ficheiros sintéticos montados. Remove os seus containers e temporários no fim. A biblioteca e a imagem precisam de estar disponíveis antes de executar o comando descrito no código.
Investigar perdas e atrasos entre componentes
A cadeia de recolha tem filas e buffers diferentes. Em modo batch, a auditoria do API server exporta eventos de forma assíncrona. Uma pausa do coletor pode fazer crescer a fila antes de qualquer regra Falco ver os dados. Num exemplo de dimensionamento, 120 pedidos por segundo com dois estágios auditados produzem 1200 eventos em cinco segundos. Esta conta ignora picos, outros eventos, tamanhos e recuperação da fila; serve para levantar requisitos, não para certificar capacidade. No Falco, a fila entre o motor e os canais de saída é distinta do buffering dentro de um canal. Uma capacidade limitada pode descartar eventos quando esgota; uma fila sem limite também cria risco de memória. Desativar buffering pode reduzir atraso de flush no stdout, com custo operacional, mas não recupera eventos já perdidos a montante. Investiga métricas do produtor, saúde da entrega, recursos do sensor e persistência no destino. O contador apiserver_audit_error_total ajuda a observar perdas de exportação, sem substituir uma investigação do período afetado. A partir do Falco 0.45, a apresentação de bytes inválidos e caracteres de controlo também merece revisão dos parsers. As condições avaliam a entrada original, enquanto a saída pode representar caracteres de forma diferente. Evita reconstruir bytes originais a partir de texto normalizado quando a investigação exige fidelidade. No exercício, usa dados fictícios para testar parsing e preserva o ficheiro de entrada. Documenta a diferença entre atraso, perda e transformação de representação, porque cada uma exige uma resposta diferente.
Correlacionar o pedido com o seu resultado
Um pedido auditado pode gerar eventos em vários estágios. RequestReceived mostra receção, não conclusão bem-sucedida. Se o resultado posterior tem código 403, o registo inicial não deve ser apresentado como prova de acesso autorizado. Conserva a relação entre auditID e estágios. Um coletor que retém apenas o primeiro evento de cada identificador pode eliminar precisamente a informação de resultado necessária à investigação. Deduplicação precisa de preservar diferenças relevantes, não apenas reduzir linhas. Num control plane com vários API servers, uma consulta através do balanceador pode não atingir a instância com política desatualizada. Confirma adoção da política e origem dos registos por instância. Regista também o cluster ao agregar dados. Um teste bem-sucedido num caminho não cobre automaticamente outro com configuração diferente. Se o destino tem uma lacuna, consulta as fontes ainda disponíveis e explicita os limites da reconstrução; não inventes um resultado de sucesso para pedidos sem confirmação. Como aplicação guiada, prepara uma tabela com pedido, estágio, alvo, resultado observado, origem e incerteza. Usa três situações fictícias: receção seguida de 403, conclusão bem-sucedida e apenas receção durante uma falha de exportação. Explica o que cada linha permite concluir e o que não permite. Leva essa distinção ao handover: «tentativa observada» e «operação concluída» implicam decisões diferentes. Relaciona a leitura com controlo de acesso, retenção de evidência e análise do impacto no serviço.
Demonstrar o efeito da contenção
Uma ação aceite pela API pode ter um efeito mais limitado do que a equipa imagina. Rollout pause impede progressão do rollout, mas não interrompe os processos das réplicas existentes. A eliminação forçada de um Pod também não espera confirmação de terminação pelo kubelet. Se o nó está inacessível, o objeto pode desaparecer enquanto a execução continua. Antes de escrever «isolado» no registo, define a fronteira pretendida e que observação a demonstrará. Para NetworkPolicy, considera todas as permissões aplicáveis. Uma política de quarentena sem regras egress não anula uma permissão concedida por outra política que seleciona o mesmo Pod. Novas ligações bloqueadas também não demonstram que uma sessão anterior terminou: o tratamento de ligações existentes depende da implementação. O comportamento de hostNetwork precisa de confirmação própria no plugin. A aceitação do objeto não fornece, por si só, o instante em que o dataplane adotou a alteração. Num exercício de decisão, o Pod de reconciliação deixa de abrir novas ligações, mas mantém uma sessão suspeita com um destino externo. A resposta deve tratar esse caminho residual e preservar os acessos necessários à investigação. Identifica a autoridade para uma ação com impacto no serviço, a consequência esperada e a observação posterior. Evita alternar mudanças de política sem hipótese clara, pois isso dificulta reconstruir a sequência. Estes limites são estudados na documentação; a reprodução local de eventos Falco não executa nem demonstra contenção de rede num cluster.
Recuperar sem perder dados nem vestígios
Uma imagem aprovada não torna automaticamente seguros os dados montados. Se o incidente alterou configuração num volume persistente, substituir o Pod e reutilizar esse volume pode recuperar o mesmo comportamento. Identifica a origem de cada configuração, os dados de negócio afetados e o estado a partir do qual é aceitável recuperar. Preserva o necessário para investigação e reconciliação antes de uma ação destrutiva. Reiniciar não desfaz pagamentos já concluídos. O emptyDir tem outra vida útil: conserva dados durante crashes de containers no mesmo Pod, mas a remoção do Pod perde esse conteúdo. Se contém os únicos ficheiros úteis ao diagnóstico e existe uma janela autorizada, recolhe-os antes de substituir o Pod. Regista a sua relação com o Pod, container e momento da recolha. A urgência da contenção pode limitar o que é possível preservar; explicita a decisão e a informação perdida, em vez de prometer reconstrução posterior. Para investigar uma imagem distroless, um ephemeral container pode fornecer ferramentas sem as incorporar na imagem da aplicação. Avalia imagem, acesso, permissões e visibilidade de processos. O mecanismo não fornece reinício automático e a entrada não pode ser alterada ou removida depois de adicionada. Regista a intervenção, mesmo quando o processo de diagnóstico termina. Como exercício, distingue três decisões: recolher vestígios, terminar execução suspeita e restabelecer processamento consistente. Cada uma tem critérios próprios e pode exigir coordenação com equipas diferentes.
Preparar o handover e os critérios de recuperação
O handover deve permitir que a equipa seguinte continue a partir de factos. Entrega a versão das regras e plugins, o intervalo investigado, os pedidos relevantes, as perdas conhecidas, a contenção observada e os caminhos ainda por testar. Separa ações concluídas de decisões autorizadas mas pendentes de execução. Para cada incerteza, indica impacto possível, responsável e próximo checkpoint. Inclui também a alteração necessária para recuperar a configuração anterior quando uma mudança de deteção falha. No caso de pagamentos, um processo saudável e uma imagem aprovada não autorizam reenvio indiscriminado de um lote. Confirma o estado transacional com a equipa responsável, delimita os itens repetíveis e os controlos contra duplicação. Se os dados não permitem uma decisão antes da janela, apresenta uma contingência avaliada ou a suspensão do lote com impacto comunicado. Esta é uma aplicação do raciocínio de recuperação ao contexto profissional, não uma regra operacional de uma instituição específica. Conclui o exercício entregando uma matriz com cinco colunas: afirmação, evidência, limite, decisão e responsável. Exemplos de afirmações são «a regra voltou a corresponder», «o destino recebeu o alerta», «a sessão suspeita terminou» e «o lote está reconciliado». Exige evidência diferente para cada uma. O resumo da aula é simples: restaurar o sensor, conter a execução e recuperar o serviço são trabalhos relacionados, mas cada conclusão precisa da sua própria observação. Relaciona-os com auditoria, imutabilidade, gestão de incidentes, mudanças e continuidade.
#!/usr/bin/env python3
"""Original synthetic Kubernetes audit replay. No cluster, host sensors or real credentials.
Requires Docker, the pinned Falco image and the checksum-verified ARM64 plugin.
Usage: python3 run.py --plugin /path/libk8saudit.so --report /path/evidence.json
"""
import argparse,datetime,hashlib,json,pathlib,shutil,subprocess,tempfile,uuid
IMAGE='falcosecurity/falco@sha256:788f1129c542171813083d4afc61b16730a47dde8c23d9c39370acef996349b6'
PLUGIN_SHA='a01cf425b9c345b6908196fe828d363531a3f1261e6346e4e8f4c25a281eccd3'
BASE='ka.target.resource=secrets and ka.verb=get and ka.stage=ResponseComplete'
def sha(data):return hashlib.sha256(data).hexdigest()
def event(n,user='ops-a',namespace='funds',resource='secrets',verb='get',stage='ResponseComplete',code=200):
return dict(apiVersion='audit.k8s.io/v1',kind='Event',level='Metadata',auditID=f'dr-{n:02}',stage=stage,verb=verb,user={'username':user},objectRef={'resource':resource,'namespace':namespace,'name':'synthetic-only'},responseStatus={'code':code},requestReceivedTimestamp='2026-10-07T08:00:00.000000Z',stageTimestamp='2026-10-07T08:00:00.000001Z')
def rule(name,condition=BASE,**extra):
return dict(rule=name,desc='Original dr.pt synthetic event exercise',condition=condition,output='DR request=%ka.auditid user=%ka.user.name namespace=%ka.target.namespace code=%ka.response.code',priority='WARNING',source='k8s_audit',**extra)
def main():
ap=argparse.ArgumentParser();ap.add_argument('--plugin',type=pathlib.Path,required=True);ap.add_argument('--report',type=pathlib.Path,required=True);a=ap.parse_args()
assert sha(a.plugin.read_bytes())==PLUGIN_SHA,'Unexpected plugin bytes'
assert not a.report.exists(),'Report must be a new path'
a.report.parent.mkdir(parents=True,exist_ok=True)
work=pathlib.Path(tempfile.mkdtemp(prefix='dr-cks-falco-'));work.chmod(0o755)
shutil.copyfile(a.plugin,work/'libk8saudit.so');(work/'libk8saudit.so').chmod(0o644)
name='dr-cks-falco-'+uuid.uuid4().hex[:12];observations=[];runs=[]
report=dict(startedAt=datetime.datetime.now(datetime.timezone.utc).isoformat(),scriptSha256=sha(pathlib.Path(__file__).read_bytes()),image=IMAGE,pluginSha256=PLUGIN_SHA,syntheticOnly=True,actualFalcoReplay=True,liveCluster=False,syscallCapture=False,productionWebhook=False,CNIContainment=False,fullPracticalMock=False,observations=observations,runs=runs)
events=[event(1),event(2,namespace='test'),event(3,user='ops-b'),event(4,resource='configmaps'),event(5,verb='list'),event(6,stage='RequestReceived'),event(7,code=403),event(8,user='ops-a\nFORGED-LINE')]
(work/'events.jsonl').write_text(''.join(json.dumps(x)+'\n' for x in events));report['inputSha256']=sha((work/'events.jsonl').read_bytes())
def check(label,actual,expected):
observations.append(dict(name=label,observed=actual,expected=expected,passed=actual==expected));assert actual==expected,(label,actual,expected)
def run(label,rule_sets,extra=None,argv=None,expected=None,failure=False):
filenames=[]
for i,rules in enumerate(rule_sets):
f=work/f'rules-{i}.yaml';f.write_text(json.dumps(rules,indent=2));filenames.append('/lab/'+f.name)
config=dict(engine={'kind':'nodriver'},plugins_hostinfo=False,plugins=[dict(name='k8saudit',library_path='/lab/libk8saudit.so',init_config=json.dumps({'maxEventSize':262144,'useAsync':False}),open_params='/lab/events.jsonl')],load_plugins=['k8saudit'],rules_files=filenames,json_output=True,buffered_outputs=False,stdout_output={'enabled':True},syslog_output={'enabled':False},webserver={'enabled':False},rule_matching='first',priority='debug')
if extra:config.update(extra)
(work/'falco.yaml').write_text(json.dumps(config,indent=2))
cmd=['docker','run','--rm','--name',name,'--network','none','--read-only','--cap-drop','ALL','--security-opt','no-new-privileges','--user','65534:65534','--memory','256m','--pids-limit','64','--tmpfs','/tmp:rw,noexec,nosuid,size=16m','-v',str(work)+':/lab:ro','--entrypoint','/usr/bin/falco',IMAGE,'-c','/lab/falco.yaml','--enable-source','k8s_audit',*(argv or [])]
r=subprocess.run(cmd,capture_output=True,text=True,timeout=30)
alerts=[]
for line in r.stdout.splitlines():
try:d=json.loads(line)
except ValueError:continue
if isinstance(d,dict) and d.get('source')=='k8s_audit' and 'rule' in d and 'output_fields' in d:alerts.append(d)
facts=sorted([[x['rule'],x['output_fields']['ka.auditid']] for x in alerts])
runs.append(dict(name=label,exitCode=r.returncode,config=config,rules=rule_sets,arguments=argv or [],stdout=r.stdout,stderr=r.stderr,alerts=alerts))
check(label+'-exit',r.returncode!=0 if failure else r.returncode,True if failure else 0)
if expected is not None:check(label+'-matches',facts,sorted(expected))
return alerts,r
try:
generic=rule('DR General');specific=rule('DR Specific',BASE+' and ka.user.name=ops-a')
ids=[1,2,3,7,8];pairs=lambda rule_name,ids:[[rule_name,f'dr-{n:02}'] for n in ids]
run('first-match',[[generic,specific]],expected=pairs('DR General',ids))
run('all-match',[[generic,specific]],extra={'rule_matching':'all'},expected=pairs('DR General',ids)+pairs('DR Specific',[1,2,7]))
run('reordered-first',[[specific,generic]],expected=pairs('DR Specific',[1,2,7])+pairs('DR General',[3,8]))
ex=rule('DR Exception',exceptions=[{'name':'maintenance','fields':['ka.user.name','ka.target.namespace'],'comps':['=','='],'values':[['ops-a','test']]}])
run('tuple-exception',[[ex]],expected=pairs('DR Exception',[1,3,7,8]))
override={'rule':'DR General','condition':'and ka.target.namespace=funds','override':{'condition':'append'}}
run('append-after',[[generic],[override]],expected=pairs('DR General',[1,3,7,8]))
_,bad=run('append-before',[[override],[generic]],expected=[],failure=True)
check('append-before-error-identifies-rule','DR General' in bad.stderr+bad.stdout,True)
replace={'rule':'DR General','condition':'ka.user.name=ops-a','override':{'condition':'replace'}}
run('replace-loses-bounds',[[generic],[replace]],expected=pairs('DR General',[1,2,4,5,6,7]))
run('disable-then-enable',[[generic]],argv=['-o','rules[].disable.rule=*','-o','rules[].enable.rule=DR General'],expected=pairs('DR General',ids))
run('enable-then-disable',[[generic]],argv=['-o','rules[].enable.rule=DR General','-o','rules[].disable.rule=*'],expected=[])
run('severity-excludes',[[generic]],extra={'priority':'error','log_level':'debug'},expected=[])
run('severity-includes',[[generic]],extra={'priority':'warning','log_level':'error'},expected=pairs('DR General',ids))
unknown=rule('DR Unknown','ka.nonexistent_dr_field=example')
_,bad=run('unknown-strict',[[generic,unknown]],failure=True,expected=[])
check('unknown-strict-error-identifies-field','ka.nonexistent_dr_field' in bad.stderr+bad.stdout,True)
unknown['skip-if-unknown-filter']=True
run('unknown-skipped',[[unknown,generic]],expected=pairs('DR General',ids))
run('dry-run',[[generic]],argv=['--dry-run'],expected=[])
allowed=rule('DR Successful',BASE+' and ka.response.code="200"')
run('response-success',[[allowed]],expected=pairs('DR Successful',[1,2,3,8]))
alerts,_=run('json-control-character',[[generic]],expected=pairs('DR General',ids))
changed=next(x for x in alerts if x['output_fields']['ka.auditid']=='dr-08')
check('json-user-roundtrip',changed['output_fields']['ka.user.name'],'ops-a\nFORGED-LINE')
check('single-alert-for-control-character',sum(x['output_fields']['ka.auditid']=='dr-08' for x in alerts),1)
_,version=run('version',[[generic]],argv=['--version'])
report['version']=json.loads(version.stdout);check('falco-version',report['version']['falco_version'],'0.45.0')
_,plugin=run('plugin-info',[[generic]],argv=['--plugin-info','k8saudit']);report['pluginOutput']=plugin.stdout+plugin.stderr;check('plugin-version','0.18.0' in report['pluginOutput'],True)
check('input-unchanged',sha((work/'events.jsonl').read_bytes()),report['inputSha256'])
report['passed']=True
except BaseException as exc:
report['passed']=False;report['error']=str(exc);raise
finally:
subprocess.run(['docker','rm','-f',name],capture_output=True,timeout=15)
r=subprocess.run(['docker','ps','-a','--filter','name=^/'+name+'$','--format','{{.Names}}'],capture_output=True,text=True,timeout=15)
shutil.rmtree(work)
report['cleanup']=dict(ownedContainersRemoved=r.returncode==0 and not r.stdout.strip(),temporaryMaterialRemoved=not work.exists(),realCredentialsUsed=False,publicRegistryWrites=False)
report['finishedAt']=datetime.datetime.now(datetime.timezone.utc).isoformat();a.report.write_text(json.dumps(report,ensure_ascii=False,indent=2)+'\n')
print(json.dumps({'passed':report['passed'],'observations':len(observations),'report':str(a.report)}))
if __name__=='__main__':main()
O sensor está saudável, mas um override alargou a exceção e ocultou acessos fora do namespace autorizado.
Armadilhas comuns
Confundir ausência de alertas com ausência de atividade, pausa de rollout com contenção e saúde do processo com integridade dos dados.
Tópicos relacionados: Auditoria e observabilidade · Imutabilidade em execução · Resposta a incidentes · Recuperação e reconciliação
Cada conclusão precisa de evidência da fronteira que afirma controlar.
Referência: CKS domains and exam details · Kubernetes v1.35; current six-domain CKS outline