APRENDER. PRATICAR. EVOLUIR.Desenhar o fluxo antes de escolher o controlo
Num cenário fictício de fundos, um reconciler chama o serviço ledger durante uma janela de processamento. Há requisitos diferentes: o conteúdo deve estar protegido no enlace entre nós, apenas os clientes previstos devem chegar ao serviço e a aplicação deve autorizar cada operação. Desenha origem, destino, namespace, nó, porto e nome utilizado. Indica onde existe tradução do Service, resolução DNS e transporte entre nós. Esta descrição permite escolher observações que correspondem ao requisito. Uma resposta HTTP 200 mostra que aquela chamada recebeu uma resposta; não demonstra, isoladamente, confidencialidade do enlace nem autorização funcional correta. Da mesma forma, um timeout não prova que uma política bloqueou a ligação: pode haver um servidor parado ou uma rota em falta. Começa por estabelecer uma chamada positiva com um marcador sintético conhecido. Depois altera uma fronteira de cada vez e mantém uma chamada de controlo para interpretar a diferença. A prática usa Cilium com WireGuard e NetworkPolicy. O túnel protege o tráfego abrangido entre nós; a política restringe ligações segundo seletores, direções e portos. Este laboratório não instala Istio nem certificados TLS individuais de aplicações. O caso bancário é original e didático, sem representar procedimentos internos da BNP Paribas. A utilidade para APS está em separar a observação técnica da decisão de continuidade e da responsabilidade por cada controlo.
Escolher a topologia e o ponto de observação
A encriptação WireGuard do Cilium estabelece peers entre os agentes dos nós. Um teste com dois Pods no mesmo nó não atravessa o enlace entre nós e não demonstra a propriedade exigida nesse enlace. No exercício, o cliente fica no control plane e o servidor no worker. Os nomes e endereços observados são guardados no relatório. Esta disposição serve o ensaio; não é uma recomendação de colocação de aplicações num control plane de produção. O ponto de captura também muda o significado do resultado. Em cilium_wg0 é possível observar o pacote interno, incluindo HTTP. Isso não equivale a encontrar o mesmo texto na representação enviada pela interface externa. No laboratório, um observador no namespace de rede do nó conta apenas pacotes IPv4 UDP 51871 entre os dois endereços de nós e verifica a presença do marcador sintético. Não guarda payloads. Confirma que houve pacotes; uma captura vazia não é uma demonstração de confidencialidade. Relaciona o resultado com os peers e com chamadas efetivamente concluídas durante a janela. A ausência de um marcador numa captura limitada não certifica todos os fluxos do cluster nem a implementação criptográfica. O objetivo é produzir evidência observável do caminho ensaiado e explicar os seus limites. Tráfego no mesmo nó, hostNetwork, fornecedores externos e outros modos de encaminhamento precisam de avaliação própria.
Preparar e executar o ambiente descartável
O script associado precisa de Python 3, Docker, kind 0.33.0, kubectl 1.35.8, Helm 3.19.0 e do chart oficial Cilium 1.20.2 indicado no código. Recebe os caminhos em --kind, --kubectl, --helm, --chart e um novo --report. Cria o seu próprio cluster com dois nós e kubeconfig privado. O CNI padrão do kind está desativado para que a execução das políticas pertença ao Cilium. Os nós podem ficar temporariamente NotReady antes de o CNI estar operacional. As imagens são obtidas por digest e carregadas nos nós com aliases exclusivos do ensaio, para reduzir dependência de downloads durante o arranque. O relatório relaciona a origem verificada, o alias e o identificador da imagem. O observador usa hostNetwork e NET_RAW apenas dentro do cluster descartável. Não monta diretórios pessoais nem utiliza credenciais de produção. O código apaga o cluster, os aliases próprios e os ficheiros temporários no fim; os conteúdos de imagem em cache podem continuar disponíveis. Prevê os resultados antes de executar: os três clientes ligam inicialmente; uma política de ingress sem permissões bloqueia-os; a exceção delimitada permite apenas o reconciler previsto. Duas execuções finais passaram 42 observações cada, incluindo 12 chamadas repetidas durante a captura; não são 42 tarefas práticas distintas. Os primeiros ensaios observaram o caminho encriptado e as diferenças de seleção, mas também identificaram um detalhe do erro DNS e uma instalação que não ficou pronta dentro do prazo. O runner validado destina-se a Docker ARM64; a exportação das imagens seleciona linux/arm64. Outro sistema precisa de adaptar e validar essa dependência, sem presumir que esta evidência abrange todas as arquiteturas.
Demonstrar quem pode ligar e quem deve falhar
A matriz do ensaio contém três clientes: role=reconciler em dr-funds, outra role no mesmo namespace e role=reconciler em dr-vendor. Esta combinação distingue uma seleção correta de uma regra que aceita todas as labels iguais em qualquer namespace. A chamada positiva de cada cliente é registada antes da restrição, evitando atribuir a uma política uma indisponibilidade que já existia. Para permitir apenas a combinação prevista, namespaceSelector e podSelector aparecem na mesma entrada from. O ensaio divide-os depois em entradas separadas e observa o alargamento ao cliente com outra role dentro do namespace autorizado. O podSelector isolado continua relativo ao namespace da política; não admite automaticamente o cliente com a mesma label em dr-vendor. A diferença deve ser explicada com a estrutura da regra e com a matriz de chamadas, sem depender apenas da aparência do YAML. O script volta à interseção antes de continuar. Usa ligações novas para cada observação e espera pelo resultado configurado, porque aceitação pela API não fornece o instante exato de adoção no dataplane. Estes testes não demonstram terminação de sessões antigas. Num incidente, essa limitação é relevante: bloquear novas ligações e encerrar uma sessão já estabelecida são afirmações diferentes e precisam de observações próprias.
Diagnosticar as duas direções e a dependência DNS
O servidor permitir ingress não basta se a origem estiver isolada para egress sem a permissão necessária. O laboratório introduz essa condição no reconciler e observa a falha por IP. Acrescenta depois apenas a saída para o servidor e porto 8080. A leitura por IP volta a funcionar, mas o nome do Service continua dependente de resolução. Esta sequência separa problemas de alcance da aplicação e problemas de DNS. A permissão seguinte inclui os Pods CoreDNS selecionados no namespace kube-system, com UDP e TCP 53. O teste volta a utilizar ledger.dr-funds.svc.cluster.local e exige o marcador esperado, não apenas ausência de erro no comando. A localização de DNS é específica desta topologia. Num cluster com NodeLocal DNSCache ou outro desenho, confirma o destino efetivo antes de reutilizar seletores; copiar uma regra sem essa verificação pode manter a falha. O primeiro ensaio devolveu gaierror com o texto «Try again», enquanto a verificação esperava outras palavras. A correção usa o tipo estruturado do erro de resolução, preservando a exigência de falha DNS. Esta é uma lição útil para APS: mensagens variam entre bibliotecas e sistemas, mas a classificação do erro deve continuar ligada à camada que falhou. Não respondas automaticamente a um erro de resolução com alargamento de permissões de negócio ou retirada de toda a segmentação.
Rever permissões acumuladas e particularidades do plugin
Depois de restaurar a chamada autorizada por nome, o ensaio acrescenta uma segunda política que permite ingress sem restrição de origem. O cliente de dr-vendor volta a ligar. A política delimitada ainda existe, mas já não descreve sozinha a superfície permitida. Ao retirar a permissão ampla, o cliente externo ao âmbito volta a falhar e o reconciler continua funcional. Regista estes dois resultados antes de encerrar a exceção temporária. A regra de união de permissões aplica-se às NetworkPolicy normais usadas neste ensaio. Não a transformes numa afirmação universal sobre todos os formatos e níveis de política que o Cilium suporta. A documentação distingue políticas Kubernetes, CiliumNetworkPolicy e políticas de âmbito global com semânticas próprias. Quando existem várias, inventaria o conjunto efetivo antes de atribuir uma ligação inesperada a uma única regra. Outro exemplo é ipBlock. A documentação Cilium 1.20.2 indica que, por omissão, não corresponde a IPs internos de Pods ou nós; policyCIDRMatchMode permite alterar esse comportamento. O laboratório usa seletores, não testa essa opção. O exercício conceptual pede que expliques por que alargar um CIDR não substitui compreender a classificação do destino. A versão do plugin, as opções e a topologia devem acompanhar qualquer recomendação transportada de outro ambiente.
Investigar os limites do túnel sem alterar controlos ao acaso
Se o fluxo entre nós falha depois de uma mudança, começa pelo caminho que foi alterado. O endpoint WireGuard padrão usa UDP 51871. Compara alcance entre nós, chave pública remota, IPs associados ao peer e estado do novo endpoint. Um Pod recriado pode ter outro IP; o sucesso de um Pod antigo não demonstra atualização do estado do novo. Mantém esta hipótese separada de falhas de autorização da aplicação. A encriptação adicional de tráfego de nó tem outro âmbito e inclui uma exclusão padrão dos nós control plane para evitar uma dependência de arranque: alcançar a API pode ser necessário para atualizar a chave pública. Esta prática não ativa esse modo. Também não testa CNI chaining, alterações de MTU ou fornecedores externos. Esses casos são discutidos para delimitar decisões e apontar evidência adicional, sem os apresentar como executados. Num cenário de degradação após encapsulamento, compara chamadas pequenas e maiores e investiga MTU e fragmentação antes de alterar RBAC sem ligação ao sintoma. Num relatório sobre fornecedor externo, desenha o segmento de saída e o controlo que o protege. A chave WireGuard do nó não é um certificado TLS individual de cada Pod. Quando o requisito é identidade de workload ou autorização de uma operação, identifica e valida o mecanismo correspondente.
Preparar a decisão de retoma e o handover
O caso de fundos termina com uma decisão baseada numa matriz, não apenas numa linha verde de health check. Entrega as chamadas autorizadas, as recusadas, os endereços e nós observados, as políticas efetivas e o resultado por nome de Service. Identifica o efeito da exceção ampla e confirma a sua retirada. Se a aplicação exige reconciliação de resultados anteriores, essa validação pertence ao fluxo de negócio e não é substituída pelo marcador HTTP sintético. O handover deve conservar versão do cluster e do plugin, origem e digest das imagens, hash do script, janela de observação e limitações. Inclui resultados dos ensaios falhados e o motivo da correção, sem classificar um timeout de instalação como prova de falha da política. A repetição do laboratório demonstra apenas o comportamento no âmbito descrito; não é um simulado prático completo nem revisão especializada independente. Resumo: começa com uma chamada de controlo, escolhe um caminho entre nós para testar o túnel, observa o ponto de captura correto e avalia seletores, egress, ingress e DNS. Confirma simultaneamente o que deve funcionar e o que deve continuar bloqueado. Relaciona esta prática com observabilidade, mudanças, identidade de workloads, resposta a incidentes e continuidade. Nas perguntas, justifica a próxima ação a partir do sintoma e explica por que as outras ações não demonstram o requisito apresentado.
#!/usr/bin/env python3
"""Original CKS network exercise: disposable two-node kind, Cilium and synthetic HTTP.
Requires Docker,kind0.33.0,kubectl1.35.8,Helm3.19.0 and the official Cilium1.20.2 chart.
Usage: python3 run.py --kind PATH --kubectl PATH --helm PATH --chart PATH --report NEW.json
This validated runner targets an ARM64 Docker engine. Uses its own cluster and kubeconfig. Never applies resources to an existing cluster.
"""
import argparse,datetime,hashlib,json,os,pathlib,shutil,subprocess,tempfile,time,uuid
NODE='kindest/node:v1.35.8@sha256:07b2536e30b803ed61d1677a79df6115f798ce64c80f9e22f6ed45afd09323c0'
PYTHON='python:3.13-alpine@sha256:2d9aefe2fef018a7eb2c13064c89c71929800fd2e5dccdbf52ea5da5bb8d929a'
CILIUM='quay.io/cilium/cilium:v1.20.2@sha256:2939231d0d3e3ebddcd80fffa168b7ddcc78fdf0dc864d1c8c126ff523c54f01'
OPERATOR='quay.io/cilium/operator-generic:v1.20.2@sha256:64d8798350e8569b8e7622563fed6e44dce2625f311e4651b774816516c744fc'
CHART_SHA='b2afd87b7f75f875f92a14559f14f59b7babbb479d968e3fd625a20bf30ec20e'
MARKER='DR_SYNTHETIC_FUNDS_NETWORK_20261007'
SERVER="""from http.server import BaseHTTPRequestHandler,HTTPServer
class H(BaseHTTPRequestHandler):
def do_GET(self):
body=b'DR_SYNTHETIC_FUNDS_NETWORK_20261007';self.send_response(200);self.end_headers();self.wfile.write(body)
def log_message(self,*args):pass
HTTPServer(('0.0.0.0',8080),H).serve_forever()
"""
CLIENT="""import json,sys,urllib.request
try:
with urllib.request.urlopen(sys.argv[1],timeout=2) as r:print(json.dumps(dict(ok=r.status==200,body=r.read().decode())))
except Exception as e:print(json.dumps(dict(ok=False,error=type(e).__name__,detail=str(e),reasonType=type(getattr(e,'reason',None)).__name__,reasonErrno=getattr(getattr(e,'reason',None),'errno',None))))
"""
CAPTURE="""import socket,struct,time,json,pathlib,sys
s=socket.socket(socket.AF_PACKET,socket.SOCK_RAW,socket.htons(3));s.bind(('eth0',0));s.settimeout(.25)
peers=set(sys.argv[1:3]);packets=0;marker=False;start=time.monotonic();pathlib.Path('/tmp/ready').write_text('ready')
while time.monotonic()-start<18:
try:p=s.recv(65535)
except TimeoutError:continue
if len(p)<42 or p[12:14]!=b'\\x08\\x00' or p[23]!=17:continue
ihl=(p[14]&15)*4;src=socket.inet_ntoa(p[26:30]);dst=socket.inet_ntoa(p[30:34]);off=14+ihl
if {src,dst}!=peers or len(p)<off+8:continue
ports=struct.unpack('!HH',p[off:off+4])
if 51871 not in ports:continue
packets+=1;marker=marker or b'DR_SYNTHETIC_FUNDS_NETWORK_20261007' in p
print(json.dumps(dict(interface='eth0',wireguardPackets=packets,syntheticMarkerObserved=marker,scope='Only IPv4 UDP51871 packets between the two owned node IPs during the observation window. No payload retained.')))
"""
def main():
ap=argparse.ArgumentParser()
for n in ['kind','kubectl','helm','chart','report']:ap.add_argument('--'+n,required=True)
a=ap.parse_args();out=pathlib.Path(a.report);assert not out.exists();assert hashlib.sha256(pathlib.Path(a.chart).read_bytes()).hexdigest()==CHART_SHA
os.umask(0o077);work=pathlib.Path(tempfile.mkdtemp(prefix='dr-cks-network-'));kc=work/'kubeconfig';name='dr-cks-network-'+uuid.uuid4().hex[:10];control=name+'-control-plane';worker=name+'-worker';ns='dr-funds';foreign='dr-vendor';started=False;observations=[];probes=[];ownedAliases=[];appImage=name+'-python:lab'
env={**os.environ,'HELM_CACHE_HOME':str(work/'helm-cache'),'HELM_CONFIG_HOME':str(work/'helm-config'),'HELM_DATA_HOME':str(work/'helm-data')}
report=dict(startedAt=datetime.datetime.now(datetime.timezone.utc).isoformat(),cluster=name,scriptSha256=hashlib.sha256(pathlib.Path(__file__).read_bytes()).hexdigest(),chartSha256=CHART_SHA,nodeImage=NODE,applicationImage=PYTHON,observations=observations,probes=probes,actualCluster=True,actualCNIEnforcement=True,actualCrossNodeTraffic=True,syntheticOnly=True,applicationMTLS=False,serviceMesh=False,fullPracticalMock=False,independentVerification=False)
def run(cmd,data=None,timeout=40,check=True):
r=subprocess.run(cmd,input=data,text=True,capture_output=True,timeout=timeout,env=env)
if check and r.returncode:raise RuntimeError(str(cmd[:3])+' failed: '+r.stderr[-1000:])
return r
def k(*args,obj=None,timeout=40,check=True):return run([a.kubectl,'--kubeconfig',str(kc),'--context','kind-'+name,'--cache-dir',str(work/'cache'),'--request-timeout=20s',*args],json.dumps(obj) if obj is not None else None,timeout,check)
def apply(o):k('apply','-f','-',obj=o)
def observe(n,v,e):observations.append(dict(name=n,observed=v,expected=e,passed=v==e));print(n,flush=True);assert v==e,(n,v,e)
def wait(fn,limit=90):
end=time.monotonic()+limit
while time.monotonic()<end:
try:
v=fn()
if v:return v
except (RuntimeError,subprocess.TimeoutExpired):pass
time.sleep(2)
raise AssertionError('Timed out waiting for observed state')
def pod(n,namespace,node,labels,command,host=False):
c=dict(name='app',image=appImage,imagePullPolicy='IfNotPresent',command=command,resources={'requests':{'cpu':'20m','memory':'32Mi'},'limits':{'memory':'128Mi'}})
if host:c['securityContext']={'capabilities':{'add':['NET_RAW']}}
return dict(apiVersion='v1',kind='Pod',metadata=dict(name=n,namespace=namespace,labels=labels),spec=dict(nodeName=node,automountServiceAccountToken=False,hostNetwork=host,dnsPolicy='ClusterFirstWithHostNet' if host else 'ClusterFirst',containers=[c],tolerations=[{'operator':'Exists'}]))
def request(podname='allowed',namespace=ns,url=None):
d=json.loads(k('-n',namespace,'exec',podname,'--','python','-c',CLIENT,url or target).stdout);probes.append(dict(pod=podname,namespace=namespace,url=url or target,**d));return d
def outcome(label,podname='allowed',namespace=ns,ok=True,url=None):
def check():
d=request(podname,namespace,url)
if ok:return d.get('ok') and d.get('body')==MARKER
return d.get('ok') is False and 'timed out' in d.get('detail','').lower()
wait(check);observe(label,True,True)
def policy(n,namespace,selector,types,**rules):return dict(apiVersion='networking.k8s.io/v1',kind='NetworkPolicy',metadata=dict(name=n,namespace=namespace),spec=dict(podSelector={'matchLabels':selector},policyTypes=types,**rules))
try:
config=dict(kind='Cluster',apiVersion='kind.x-k8s.io/v1alpha4',networking=dict(apiServerAddress='127.0.0.1',disableDefaultCNI=True),nodes=[dict(role='control-plane'),dict(role='worker')]);(work/'kind.json').write_text(json.dumps(config));started=True;print('creating-owned-cluster '+name,flush=True)
run([a.kind,'create','cluster','--name',name,'--image',NODE,'--config',str(work/'kind.json'),'--kubeconfig',str(kc),'--wait','120s'],timeout=240)
versions=json.loads(k('version','-o','json').stdout);report['versions']=versions;observe('server-version',versions['serverVersion']['gitVersion'],'v1.35.8');observe('client-version',versions['clientVersion']['gitVersion'],'v1.35.8')
helmversion=run([a.helm,'version','--short']).stdout.strip();report['helmVersion']=helmversion;observe('helm-version',helmversion.startswith('v3.19.0'),True)
print('preloading-pinned-images',flush=True)
observe('docker-architecture',run(['docker','info','--format','{{.Architecture}}']).stdout.strip(),'aarch64')
report['preloadedImages']=[]
for reference,alias in [(CILIUM,name+'-cilium:lab'),(OPERATOR,name+'-operator:lab'),(PYTHON,appImage)]:
run(['docker','pull',reference],timeout=300);info=json.loads(run(['docker','image','inspect',reference]).stdout)[0]
run(['docker','tag',reference,alias]);ownedAliases.append(alias);report['preloadedImages'].append(dict(source=reference,alias=alias,imageID=info['Id'],repositoryDigests=info['RepoDigests']))
archive=work/'images.tar';run(['docker','image','save','--platform','linux/arm64','-o',str(archive),*ownedAliases],timeout=180)
run([a.kind,'load','image-archive',str(archive),'--name',name],timeout=240);archive.unlink()
print('installing-cilium',flush=True)
values={'image.override':name+'-cilium:lab','operator.image.override':name+'-operator:lab','ipam.mode':'kubernetes','operator.replicas':'1','kubeProxyReplacement':'false','encryption.enabled':'true','encryption.type':'wireguard','hubble.enabled':'false','envoy.enabled':'false','l7Proxy':'false'}
cmd=[a.helm,'install','cilium',a.chart,'--namespace','kube-system','--kubeconfig',str(kc),'--kube-context','kind-'+name,'--wait','--timeout','5m']
for key,value in values.items():cmd+=['--set',key+'='+value]
run(cmd,timeout=330);k('-n','kube-system','rollout','status','ds/cilium','--timeout=180s',timeout=200);k('wait','--for=condition=Ready','nodes','--all','--timeout=120s',timeout=140)
agents=json.loads(k('-n','kube-system','get','pods','-l','k8s-app=cilium','-o','json').stdout)['items'];observe('two-cilium-agents',len(agents),2);report['agentImages']=[x['status']['containerStatuses'] for x in agents]
for agent in agents:
text=k('-n','kube-system','exec',agent['metadata']['name'],'--','cilium-dbg','status').stdout
observe('wireguard-enabled-'+agent['spec']['nodeName'],bool('Encryption:' in text and 'Wireguard' in text and 'Peers: 1' in text),True)
for namespace in [ns,foreign]:apply(dict(apiVersion='v1',kind='Namespace',metadata=dict(name=namespace,labels={'dr-team':'funds' if namespace==ns else 'vendor'})))
apply(pod('server',ns,worker,{'app':'ledger'},['python','-u','-c',SERVER]));apply(pod('allowed',ns,control,{'role':'reconciler'},['python','-c','import time;time.sleep(3600)']));apply(pod('wrong-role',ns,control,{'role':'other'},['python','-c','import time;time.sleep(3600)']));apply(pod('foreign-client',foreign,control,{'role':'reconciler'},['python','-c','import time;time.sleep(3600)']))
apply(dict(apiVersion='v1',kind='Service',metadata=dict(name='ledger',namespace=ns),spec=dict(selector={'app':'ledger'},ports=[dict(port=8080,targetPort=8080)])))
for namespace in [ns,foreign]:k('-n',namespace,'wait','--for=condition=Ready','pods','--all','--timeout=150s',timeout=170)
server=json.loads(k('-n',ns,'get','pod','server','-o','json').stdout);client=json.loads(k('-n',ns,'get','pod','allowed','-o','json').stdout);observe('distinct-source-destination-nodes',client['spec']['nodeName']!=server['spec']['nodeName'],True)
target='http://'+server['status']['podIP']+':8080/';service='http://ledger.dr-funds.svc.cluster.local:8080/';report['topology']={x:{'node':p['spec']['nodeName'],'podIP':p['status']['podIP'],'nodeIP':p['status']['hostIP']} for x,p in [('client',client),('server',server)]}
outcome('baseline-allowed');outcome('baseline-wrong-role','wrong-role');outcome('baseline-foreign','foreign-client',foreign);outcome('baseline-service-dns',url=service)
capture=pod('wire-observer',ns,control,{'app':'observer'},['python','-c',CAPTURE,client['status']['hostIP'],server['status']['hostIP']],True);capture['spec']['restartPolicy']='Never';apply(capture)
wait(lambda:k('-n',ns,'exec','wire-observer','--','test','-e','/tmp/ready',check=False).returncode==0)
for _ in range(12):outcome('captured-request-'+str(_));time.sleep(.15)
# Keep the capture Pod alive long enough to retrieve its synthetic summary.
wait(lambda:json.loads(k('-n',ns,'get','pod','wire-observer','-o','json').stdout)['status']['phase']=='Succeeded',limit=40)
# Terminated containers expose logs; the capture program also prints the summary below.
cap=json.loads(k('-n',ns,'logs','wire-observer').stdout);report['capture']=cap;observe('wireguard-packets-observed',cap['wireguardPackets']>0,True);observe('no-marker-in-selected-wire-packets',cap['syntheticMarkerObserved'],False)
apply(policy('default-deny',ns,{'app':'ledger'},['Ingress'],ingress=[]));outcome('deny-blocks-allowed',ok=False);outcome('deny-blocks-wrong-role','wrong-role',ok=False);outcome('deny-blocks-foreign','foreign-client',foreign,False)
peer={'namespaceSelector':{'matchLabels':{'dr-team':'funds'}},'podSelector':{'matchLabels':{'role':'reconciler'}}};ingress=[{'from':[peer],'ports':[{'protocol':'TCP','port':8080}]}]
apply(policy('ledger-access',ns,{'app':'ledger'},['Ingress'],ingress=ingress));outcome('intersection-allows-client');outcome('intersection-blocks-wrong-role','wrong-role',ok=False);outcome('intersection-blocks-foreign','foreign-client',foreign,False)
loose=[{'from':[{'namespaceSelector':peer['namespaceSelector']},{'podSelector':peer['podSelector']}],'ports':[{'protocol':'TCP','port':8080}]}]
# A bare podSelector is scoped to this policy's namespace. The first OR branch still admits wrong-role there.
apply(policy('ledger-access',ns,{'app':'ledger'},['Ingress'],ingress=loose));outcome('or-broadens-same-namespace','wrong-role');outcome('bare-pod-selector-not-cross-namespace','foreign-client',foreign,False)
apply(policy('ledger-access',ns,{'app':'ledger'},['Ingress'],ingress=ingress));outcome('intersection-restored','wrong-role',ok=False)
apply(policy('client-egress',ns,{'role':'reconciler'},['Egress'],egress=[]));outcome('egress-deny-overrides-ingress-allow',ok=False)
serverpeer={'namespaceSelector':{'matchLabels':{'dr-team':'funds'}},'podSelector':{'matchLabels':{'app':'ledger'}}};toapp={'to':[serverpeer],'ports':[{'protocol':'TCP','port':8080}]}
apply(policy('client-egress',ns,{'role':'reconciler'},['Egress'],egress=[toapp]));outcome('server-egress-restores-ip')
d=request(url=service);observe('dns-still-unavailable',not d['ok'] and d.get('reasonType')=='gaierror',True)
dns={'to':[{'namespaceSelector':{'matchLabels':{'kubernetes.io/metadata.name':'kube-system'}},'podSelector':{'matchLabels':{'k8s-app':'kube-dns'}}}],'ports':[{'protocol':'UDP','port':53},{'protocol':'TCP','port':53}]}
apply(policy('client-egress',ns,{'role':'reconciler'},['Egress'],egress=[toapp,dns]));outcome('dns-and-service-restored',url=service)
apply(policy('temporary-broad-allow',ns,{'app':'ledger'},['Ingress'],ingress=[{}]));outcome('additive-policy-admits-foreign','foreign-client',foreign)
k('-n',ns,'delete','networkpolicy','temporary-broad-allow');outcome('remove-broad-policy-restores-boundary','foreign-client',foreign,False);outcome('final-authorized-service',url=service)
report['passed']=True
except BaseException as e:
report['passed']=False;report['error']=str(e)
if kc.exists():
report['diagnostics']=k('-n','kube-system','get','pods','-o','wide',check=False).stdout
report['events']=k('-n','kube-system','get','events','--sort-by=.lastTimestamp',check=False).stdout[-6000:]
raise
finally:
if started:run([a.kind,'delete','cluster','--name',name,'--kubeconfig',str(kc)],timeout=90,check=False)
remaining=run(['docker','ps','-a','--filter','label=io.x-k8s.kind.cluster='+name,'--format','{{.Names}}'],check=False);shutil.rmtree(work)
aliasesRemoved=True
for alias in ownedAliases:
result=run(['docker','image','rm',alias],check=False);aliasesRemoved=aliasesRemoved and result.returncode==0
report['cleanup']=dict(ownedImageAliasesRemoved=aliasesRemoved,ownedClusterRemoved=remaining.returncode==0 and not remaining.stdout.strip(),temporaryMaterialRemoved=not work.exists(),defaultKubeconfigModified=False,realCredentialsUsed=False);report['finishedAt']=datetime.datetime.now(datetime.timezone.utc).isoformat();out.parent.mkdir(parents=True,exist_ok=True);out.write_text(json.dumps(report,indent=2)+'\n')
print(json.dumps(dict(passed=report['passed'],observations=len(observations),report=str(out))))
if __name__=='__main__':main()
A chamada por IP recupera após permitir egress para o servidor, mas o job continua a falhar até recuperar a resolução DNS.
Armadilhas comuns
Usar uma captura vazia como prova de encriptação, confundir seletor local com todos os namespaces e esquecer a dependência DNS.
Tópicos relacionados: NetworkPolicy e DNS · Encriptação em trânsito · Identidade de workloads · Mudanças e incidentes
Demonstra o caminho protegido e a fronteira de acesso com observações distintas e verificáveis.
Referência: CKS domains and exam details · Kubernetes v1.35; current six-domain CKS outline