← CKS: segurança Kubernetes em produção
20 / 22 · 120 MIN

Recuperar serviços e validar controlos Linux

Distingue modos, descritores, threads e políticas carregadas para conter e recuperar serviços com evidência.

1. Definir o estado que a mudança deve produzir

Uma equipa APS prepara o fecho de reconciliação quando recebe uma recomendação de hardening. O ticket pede retirar um daemon desnecessário, limitar acesso aos ficheiros do batch e aplicar um filtro de syscalls. São três alterações diferentes, com evidências diferentes. Um comando concluído não basta para afirmar que o resultado pretendido existe. Este caso bancário é fictício e não representa procedimentos internos da BNP Paribas. Começa por identificar o componente, o dono, os consumidores, a versão e o impacto da recuperação. Um pacote pode ter sido atualizado enquanto um processo ainda usa o executável anterior. Uma unidade pode estar parada enquanto outra a ativa por socket. Um ficheiro pode ter outro modo enquanto um processo mantém um descritor aberto. Uma thread pode estar filtrada enquanto outra continua com a política anterior. Cada situação exige uma observação no nível onde o controlo atua. Constrói uma tabela com estado anterior, alteração, prova positiva e prova negativa. A prova positiva confirma o fluxo legítimo, como concluir uma reconciliação de teste. A negativa confirma a restrição, como uma abertura ou syscall recusada. Acrescenta os limites: filesystem, arquitetura, thread, node e mecanismo de ativação. Antes de promover, pede a outro operador que explique o que cada resultado prova. Se só conseguir repetir o comando executado, a evidência de aceitação ainda precisa de trabalho.

2. Permissões, propriedade e referências abertas

A prática usa apenas ficheiros sintéticos num tmpfs local. Abre um ficheiro para leitura e escrita, retira os bits de acesso e tenta abri-lo outra vez. A abertura nova recebe EACCES. O descritor anterior continua a ler e a escrever. Não se trata de um bypass criado pelo script: são duas operações feitas através de referências e verificações distintas. A conclusão é limitada ao filesystem e aos controlos realmente exercitados. O processo é proprietário do ficheiro e não tem capabilities efetivas, permitidas ou bounding. Mesmo assim consegue repor o modo 0600. Essa observação impede uma resposta simplista a compromisso: retirar capabilities não retira automaticamente a autoridade do proprietário sobre o modo. Se o atacante controla a conta proprietária, uma contenção baseada apenas em chmod pode ser revertida por essa conta. Avalia a identidade, os processos existentes e a propriedade antes de escolher a medida. Depois, o script substitui o pathname por outro ficheiro. Uma leitura nova obtém o estado novo; o descritor antigo mantém o inode antigo, agora sem links. Compara stat do caminho com fstat do descritor para perceber a diferença. Não assumes corrupção apenas porque os leitores discordam. Também não exportes a conclusão para NFS: o servidor remoto pode verificar acesso em operações posteriores. O laboratório não executa NFS, AppArmor ou SELinux e não demonstra o comportamento desses mecanismos.

3. Verificar a cobertura por thread

A segunda parte cria uma worker antes de instalar o filtro seccomp. A principal e a worker executam a mesma syscall direta getppid. Depois instala-se na principal um filtro adicional que devolve EPERM para essa syscall, usando flags=0. A principal passa a receber a recusa; a worker anterior continua a obter o resultado normal. Uma thread nova criada pela principal herda o filtro da criadora e recebe a recusa. Este desenho separa três situações que uma única leitura de /proc pode esconder: a thread que instalou o controlo, uma thread que já existia e uma thread criada depois. O script consulta o estado por TID e executa probes sincronizados por filas, evitando atribuir a uma worker um resultado observado noutra. Não se altera o perfil builtin do Docker; os filtros novos acrescentam restrições à política existente. A seguir, uma instalação com TSYNC sincroniza as threads neste caso compatível. A worker passa a receber EPERM e ambas mostram dois filtros adicionais. A prática não constrói uma árvore divergente que faça TSYNC falhar. Esse caso é tratado nas perguntas com base na documentação: sem a variante TSYNC_ESRCH, um retorno positivo pode identificar a thread que impediu a sincronização. Não uses a regra “diferente de -1 é sucesso”. Regista retorno, errno quando aplicável e comportamento de cada thread antes de declarar cobertura.

4. Executar num ambiente descartável

Guarda o código da aula como run.py e executa python3 run.py --output resultado.json numa estação com Docker Linux ARM64. O ficheiro de saída ainda não pode existir. A imagem python@sha256:2d9aefe2fef018a7eb2c13064c89c71929800fd2e5dccdbf52ea5da5bb8d929a tem de estar disponível localmente: o script usa --pull=never. As execuções de referência usaram Docker 29.1.3, Python 3.13.16 no contentor e Linux 6.12.54-linuxkit. O script não suporta x86; números de syscalls e ABI não são portáveis por simples cópia. O contentor tem nome aleatório próprio, rede desativada, UID/GID 1000, capabilities removidas, no-new-privileges e seccomp builtin. A raiz é de leitura, /tmp é um tmpfs limitado e o único bind mount contém este script, também de leitura. Há limites de memória, CPU e processos. Não montes diretórios de produção para adaptar este exercício. Não é necessário executar o contentor como privileged nem alterar perfis de segurança do host. Antes de correr, prevê os resultados de cada fase: nova abertura, descritor antigo, alteração pelo proprietário, substituição de inode e cobertura de threads. Depois compara com o relatório JSON. O script remove apenas o contentor que criou e os ficheiros temporários desaparecem com a prática. Se a execução falhar por arquitetura, runtime ou syscall, conserva o erro para diagnóstico e mantém a restrição que motivou a falha; não retires controlos só para obter PASS.

5. Ler os resultados e escolher recuperação

As duas execuções finais registaram 24 verificações com o mesmo hash de código. Entre elas estão o modo efetivo do processo, ausência de capabilities, leitura e escrita pelo descritor antigo, recusa de nova abertura e adoção do inode novo apenas na leitura pelo pathname. A segunda metade demonstra o filtro local à thread, herança pela thread nova e TSYNC bem-sucedido para a worker existente. Uma syscall diferente continua a funcionar, fornecendo um controlo positivo. Estas verificações não medem desempenho nem validam uma aplicação bancária real. O getppid foi escolhido como operação sintética observável, sem executar uma exploração. O facto de uma operação funcionar não prova que todo o perfil está correto; uma aplicação pode usar outras syscalls apenas no fecho de ficheiros, numa recuperação ou sob carga. Constrói testes funcionais a partir do fluxo concreto e inclui caminhos raros que o serviço precisa de manter. Se um filtro instalado bloquear uma operação necessária, acrescentar ALLOW não elimina uma recusa de um filtro anterior. Uma recuperação pode exigir uma instância de substituição criada com o perfil aprovado pelo runtime ou supervisor. Um exec no processo já filtrado não é, por si, essa recuperação, porque os filtros podem ser herdados. Define de onde nasce a instância limpa, como transfere trabalho e como se confirma que a política efetiva é a aprovada.

6. Reduzir serviços sem perder o caminho de gestão

Ao retirar um daemon de diagnóstico, identifica também sockets, timers e outros mecanismos que o podem ativar. Uma unidade service parada não prova que o respetivo listener deixou de existir. A documentação de systemd distingue o serviço da unidade socket que recebe tráfego e pode ativá-lo. A ação correta depende das dependências e da função autorizada: não desligues todas as sockets de um host apenas para simplificar o inventário. Se alterares uma unidade, distingue a configuração em disco, a configuração conhecida pelo gestor e o estado do processo. Daemon-reload recarrega a configuração do gestor; não equivale a reiniciar o serviço nem a aplicar automaticamente todos os efeitos ao processo já em execução. Da mesma forma, um pacote novo pode deixar um processo associado ao executável anterior, visível através de /proc/PID/exe. Define a observação que comprova a nova geração efetiva. No batch, uma substituição deve respeitar checkpoints, dependências e prevenção de duplicação. Parar o processo sem confirmar o último item reconciliado pode criar um incidente funcional enquanto tentas resolver o de segurança. Coordena a janela com APS e negócio, reserva um caminho de gestão testado e define quem pode autorizar a recuperação. Esta aula não executa systemd nem atualizações de pacotes: estes são cenários documentados para aplicar o raciocínio de estado efetivo observado na prática Linux.

7. Validar alcance de rede e adoção de perfis

Uma alteração SSH deve ser testada com origem, utilizador, endereço de destino e família de rede identificados. Uma recusa IPv4 não prova recusa IPv6 se ambas as famílias continuam configuradas. Com regras Match, uma validação de sintaxe não basta para explicar a configuração selecionada para a ligação. O modo sshd -T com parâmetros -C ajuda a inspecionar essa seleção, mas continua a ser necessário testar o daemon efetivo e preservar um acesso de recuperação aprovado. Uma opção de protocolo também tem âmbito limitado. Desativar AllowTcpForwarding restringe o forwarding fornecido pelo SSH; não transforma automaticamente uma shell permitida num ambiente sem possibilidades de encaminhamento. Se o requisito é isolamento, avalia ferramentas disponíveis, destinos alcançáveis e controlos adicionais. Não confundas autenticação recusada por falta de chave com uma recusa de rede que pretendias demonstrar. Para AppArmor, distingue ficheiro distribuído, perfil carregado, modo e associação ao processo. A documentação explica que alterações ao ficheiro precisam de reload para ter efeito. Um perfil em complain não demonstra que a operação observada foi impedida; o teste negativo tem de mostrar o resultado pretendido. Abrange os nodes onde o workload pode executar. Neste módulo não foram executados SSH, AppArmor, firewall ou Kubernetes. As perguntas sobre esses mecanismos são aplicações originais das fontes, com limites explícitos da prática realizada.

8. Entregar decisões e evidência ao RUN

No caso final, a principal do batch está filtrada e uma worker antiga continua a executar a syscall. A primeira decisão é reconhecer a cobertura desigual, preservar o estado necessário e medir as threads relevantes. Se TSYNC for a solução suportada, confirma o retorno e os resultados. Uma falha de sincronização não pode ser transformada em sucesso pelo código que só verifica -1. Se a mudança compromete o fluxo legítimo, prepara a instância de recuperação aprovada. O gestor técnico coordena quem controla o processo, quem valida a política e quem confirma a integridade funcional. Define o ponto de decisão antes do fecho e o impacto de esperar, substituir ou ativar a contingência. Usa os checkpoints para evitar repetição de movimentos; não prometas continuidade só porque um processo novo arrancou. O negócio deve confirmar o resultado de reconciliação e APS deve confirmar observabilidade, acesso e suporte operacional. O handover inclui versões, hash do código ou perfil, identidade efetiva, matriz de threads, referências de ficheiros, resultados esperados e observados, limpeza e limitações. Mantém as perguntas abertas separadas das provas concluídas. Resume os conceitos relacionados: referências de objetos, composição de controlos, herança, reconciliação e mudança segura. A avaliação interna deste percurso treina decisões; não substitui o exame prático oficial nem prova competência completa apenas por obter uma pontuação. A revisão especializada independente continua necessária.

"""Original DR practice: local file references and seccomp thread synchronization.

Host entrypoint creates only an owned, restricted, disposable Linux container.
"""
import sys


def probe():
    import ctypes
    import errno
    import json
    import os
    import platform
    import queue
    import tempfile
    import threading
    from pathlib import Path

    assert platform.system() == 'Linux' and platform.machine() == 'aarch64'
    libc = ctypes.CDLL(None, use_errno=True)
    libc.syscall.restype = ctypes.c_long
    records = []

    def check(name, observed, expected):
        assert observed == expected, (name, observed, expected)
        records.append(dict(name=name, observed=observed, expected=expected, passed=True))

    def status(tid):
        raw = dict(line.split(':', 1) for line in Path('/proc/self/task/' + str(tid) + '/status').read_text().splitlines() if ':' in line)
        return {k: raw[k].strip() for k in ['Uid', 'Gid', 'CapEff', 'CapPrm', 'CapBnd', 'NoNewPrivs', 'Seccomp', 'Seccomp_filters']}

    def syscall(number):
        ctypes.set_errno(0)
        result = libc.syscall(ctypes.c_long(number))
        return dict(result=result, errno=ctypes.get_errno())

    main_tid = threading.get_native_id()
    initial = status(main_tid)
    check('restricted-process', [os.getuid(), os.getgid(), initial['NoNewPrivs'], initial['Seccomp']], [1000, 1000, '1', '2'])
    check('capability-sets-empty', all(int(initial[k], 16) == 0 for k in ['CapEff', 'CapPrm', 'CapBnd']), True)
    # Local tmpfs only; do not infer NFS or mandatory-access-control behavior.
    with tempfile.TemporaryDirectory(prefix='dr-system-response-') as directory:
        target = Path(directory) / 'state'
        target.write_bytes(b'old-state')
        fd = os.open(target, os.O_RDWR)
        try:
            check('initial-descriptor-read', os.pread(fd, 9, 0).decode(), 'old-state')
            target.chmod(0)
            try:
                new_fd = os.open(target, os.O_RDONLY)
                os.close(new_fd)
                open_errno = 0
            except OSError as error:
                open_errno = error.errno
            check('new-open-denied-after-chmod', open_errno, errno.EACCES)
            check('existing-descriptor-still-reads', os.pread(fd, 9, 0).decode(), 'old-state')
            os.pwrite(fd, b'old-write', 0)
            check('existing-descriptor-still-writes', os.pread(fd, 9, 0).decode(), 'old-write')
            target.chmod(0o600)
            check('owner-restores-mode-without-capabilities', target.stat().st_mode & 0o777, 0o600)
            check('fresh-open-after-owner-restores-mode', target.read_text(), 'old-write')
            replacement = Path(directory) / 'replacement'
            replacement.write_bytes(b'new-state')
            os.replace(replacement, target)
            check('replacement-changes-path-inode', os.fstat(fd).st_ino != target.stat().st_ino, True)
            check('old-descriptor-retains-old-inode-data', os.pread(fd, 9, 0).decode(), 'old-write')
            check('fresh-path-read-uses-replacement', target.read_text(), 'new-state')
            check('old-inode-unlinked-but-open', os.fstat(fd).st_nlink, 0)
        finally:
            os.close(fd)

    check('synthetic-files-removed', not Path(directory).exists(), True)
    # Native aarch64 ABI only: getpid=172, getppid=173, seccomp=277.
    baseline_parent = syscall(173)
    check('main-getppid-initially-allowed', baseline_parent['result'] >= 0 and baseline_parent['errno'] == 0, True)
    inbox, replies = queue.Queue(), queue.Queue()

    def worker():
        replies.put(dict(tid=threading.get_native_id(), query=syscall(173)))
        while True:
            message = inbox.get()
            if message == 'stop':
                return
            replies.put(dict(query=syscall(173), status=status(threading.get_native_id())))

    thread = threading.Thread(target=worker)
    thread.start()
    worker_start = replies.get(timeout=5)
    check('existing-worker-initially-allowed', worker_start['query'], baseline_parent)
    worker_initial = status(worker_start['tid'])

    class Filter(ctypes.Structure):
        _fields_ = [('code', ctypes.c_ushort), ('jt', ctypes.c_ubyte), ('jf', ctypes.c_ubyte), ('k', ctypes.c_uint)]

    class Program(ctypes.Structure):
        _fields_ = [('len', ctypes.c_ushort), ('filter', ctypes.POINTER(Filter))]

    def install(flags):
        # Reject unexpected ABI, deny getppid, otherwise retain preceding filters.
        rows = [(0x20, 0, 0, 4), (0x15, 1, 0, 0xc00000b7), (0x06, 0, 0, 0x80000000),
                (0x20, 0, 0, 0), (0x15, 0, 1, 173), (0x06, 0, 0, 0x00050000 | errno.EPERM),
                (0x06, 0, 0, 0x7fff0000)]
        filters = (Filter * len(rows))(*(Filter(*row) for row in rows))
        program = Program(len(rows), filters)
        ctypes.set_errno(0)
        result = libc.syscall(ctypes.c_long(277), ctypes.c_uint(1), ctypes.c_uint(flags), ctypes.byref(program))
        return dict(result=result, errno=ctypes.get_errno())

    denied = dict(result=-1, errno=errno.EPERM)
    try:
        check('thread-local-filter-installed', install(0), dict(result=0, errno=0))
        check('main-getppid-now-denied', syscall(173), denied)
        inbox.put('probe')
        still_allowed = replies.get(timeout=5)
        check('existing-worker-not-synchronized-by-flags-zero', still_allowed['query'], baseline_parent)
        inherited = queue.Queue()
        child_thread = threading.Thread(target=lambda: inherited.put(syscall(173)))
        child_thread.start()
        child_thread.join(timeout=5)
        assert not child_thread.is_alive()
        check('new-thread-inherits-creator-filter', inherited.get(timeout=5), denied)
        # The worker tree remains an ancestor of the caller tree, so TSYNC can succeed.
        check('tsync-filter-installed', install(1), dict(result=0, errno=0))
        inbox.put('probe')
        synchronized = replies.get(timeout=5)
        check('existing-worker-denied-after-tsync', synchronized['query'], denied)
        after = status(main_tid)
        check('both-threads-have-two-additional-filters', [int(after['Seccomp_filters']) - int(initial['Seccomp_filters']),
              int(synchronized['status']['Seccomp_filters']) - int(worker_initial['Seccomp_filters'])], [2, 2])
        check('unrelated-getpid-still-allowed', syscall(172)['result'] == os.getpid(), True)
    finally:
        inbox.put('stop')
        thread.join(timeout=5)
        assert not thread.is_alive()
    check('worker-thread-ended', thread.is_alive(), False)
    print(json.dumps(dict(python=platform.python_version(),kernel=platform.release(),architecture=platform.machine(),
                          observations=records,initialStatus=initial,workerInitialStatus=worker_initial)))


def run():
    import argparse
    import datetime
    import hashlib
    import json
    import subprocess
    import uuid
    from pathlib import Path
    parser = argparse.ArgumentParser()
    parser.add_argument('--output', required=True)
    args = parser.parse_args()
    out = Path(args.output)
    assert not out.exists()
    image = 'python@sha256:2d9aefe2fef018a7eb2c13064c89c71929800fd2e5dccdbf52ea5da5bb8d929a'
    name = 'dr-cks-system-response-' + uuid.uuid4().hex[:12]

    def docker(*arguments, timeout=30):
        result = subprocess.run(['docker', *arguments], capture_output=True, text=True, timeout=timeout)
        if result.returncode:
            raise RuntimeError(result.stderr[:1500])
        return result.stdout

    info = json.loads(docker('info', '--format', '{{json .}}'))
    assert info['Architecture'] == 'aarch64'
    assert any('seccomp' in option for option in info['SecurityOptions'])
    try:
        result = docker('run','--rm','--pull=never','--name',name,'--label','dr.lesson=cks-system-response',
                        '--network=none','--read-only','--user','1000:1000','--cap-drop=ALL',
                        '--security-opt=no-new-privileges=true','--security-opt=seccomp=builtin',
                        '--pids-limit=32','--memory=64m','--cpus=0.5',
                        '--tmpfs','/tmp:rw,noexec,nosuid,nodev,size=1m,mode=1777',
                        '--mount','type=bind,src='+str(Path(__file__).resolve())+',dst=/lesson/run.py,readonly',
                        '--entrypoint','python',image,'-B','/lesson/run.py','--inside',timeout=45)
        evidence = json.loads(result)
        assert all(row['passed'] for row in evidence['observations'])
    finally:
        if docker('ps','-aq','--filter','name=^/'+name+'$').strip():
            docker('rm','-f',name)
        assert not docker('ps','-aq','--filter','name=^/'+name+'$').strip()
    report = dict(executedAt=datetime.datetime.now(datetime.timezone.utc).isoformat(),
                  scriptSha256=hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),
                  dockerVersion=info['ServerVersion'],imageReference=image,container=name,passed=True,execution=evidence,
                  cleanup=dict(ownedContainerRemoved=True,hostSecurityChanged=False),
                  scope='Actual local tmpfs DAC and descriptor observations, aarch64 thread-local seccomp and successful TSYNC on compatible filter trees. No NFS, AppArmor, SELinux, systemd, SSH, firewall, Kubernetes, divergent-tree TSYNC failure, production recovery or full practical mock executed.')
    out.write_text(json.dumps(report,indent=2)+'\n')
    print('PASS',len(evidence['observations']),evidence['python'],evidence['kernel'])


if __name__ == '__main__':
    if '--inside' in sys.argv:
        probe()
    else:
        run()
NA PRÁTICA

A principal recebe EPERM e uma worker antiga continua a executar getppid até uma sincronização TSYNC bem-sucedida.

Armadilhas comuns

Confundir modo com revogação de descritores, ficheiro com política carregada, thread principal com todas as workers e novo PID com política limpa.

Tópicos relacionados: Permissões e propriedade Linux · Seccomp e TSYNC · AppArmor e adoção de perfis · Recuperação e handover

Leva esta ideia contigo

A aceitação deve observar o objeto, a referência e a thread onde o controlo realmente atua.

Criar conta

Referência: CKS domains and exam details · Kubernetes v1.35; current six-domain CKS outline

Kubernetes® e CKS são marcas comerciais ou marcas registadas de The Linux Foundation. A dr.pt é uma plataforma de preparação independente e não está afiliada, associada, patrocinada, autorizada nem aprovada por The Linux Foundation. Os conteúdos e as perguntas são originais, não são perguntas oficiais de exame, e concluir os nossos testes não atribui nem garante qualquer certificação. Os nomes são usados apenas para identificar o tema. Todas as outras marcas pertencem aos respetivos titulares.