OpenCyber Academy
Tecnólogo · Capítulo 29 · Português
Código do curso: OCA-092Capítulo 29: OCA-092-C029

Capítulo 29 — Construção do trabalho de conclusão

Ensino gratuito e aberto para quem deseja proteger pessoas e tecnologias. Não é preciso criar uma conta para estudar ou consultar as respostas.

Mapa da aula

Aula 29 publicada. Mapa, exemplo, EX085.

flowchart TD
  N0["29.1 O que o trabalho de conclusão precisa demonstra"]
  N1["29.2 A construção"]
  N2["29.3 Ler o registro de evidências"]
  N3["29.4 Um defeito que só aparece na combinação"]
  N4["29.5 Construir o trabalho de conclusão real"]
  N5["Síntese do capítulo"]
  N0 --> N1 --> N2 --> N3 --> N4 --> N5
  N5 --> EX["Exemplo resolvido"]
  EX --> Q["Exercício EX085"]

Desenho lógico

flowchart LR
  N0["29.1 O que o trabalho de conclusão precisa demonstra"]
  N1["29.2 A construção"]
  N2["29.3 Ler o registro de evidências"]
  N3["29.4 Um defeito que só aparece na combinação"]
  N4["29.5 Construir o trabalho de conclusão real"]
  N5["Síntese do capítulo"]
  N0 --> N1 --> N2 --> N3 --> N4 --> N5 --> Q["EX085"]
[29.1 O que o trabalho de conclusão precisa demonstra]
    |
    v
[29.2 A construção]
    |
    v
[29.3 Ler o registro de evidências]
    |
    v
[29.4 Um defeito que só aparece na combinação]
    |
    v
[29.5 Construir o trabalho de conclusão real]
    |
    v
[Síntese do capítulo]
    |
    v
[Exemplo resolvido] --> [Exercício EX085]

Objetivos de aprendizagem e conexão

Depois deste capítulo, você deve ser capaz de:

Este é o resultado prático do curso: implementar um serviço de ponta a ponta com controles monitorados. O Laboratório 29 combina, num único arquivo, a selagem nas estações do Capítulo 15, a validação do Capítulo 3, um historiador num modelo relacional (Capítulo 4), o monitoramento com uma regra de ausência (Capítulo 19) e um registro de evidências com um digest (Capítulos 6 e 20). Ele é pequeno o bastante para ser lido por inteiro; o projeto de conclusão pede que você construa a mesma forma em escala maior.

29.1 O que o trabalho de conclusão precisa demonstrar

  1. O caminho dos dados funciona: as leituras de todas as estações chegam ao historiador, validadas e autenticadas.
  2. Os controles funcionam: as falhas injetadas — um replay, uma mensagem forjada, um valor fora da faixa e um silêncio de estação — são, cada uma, rejeitadas, colocadas em quarentena ou alertadas.
  3. O monitoramento funciona: são gerados alertas para cada falha injetada e para nada mais.
  4. As evidências existem: um registro legível por máquina da execução, com contagens que fecham e um digest, para que mudanças posteriores nele sejam detectáveis.
  5. Os limites estão declarados: o que a execução não mostra.

29.2 A construção

O Laboratório 29 gera duas horas de leituras sintéticas de três estações, com a PS-09 silenciosa do minuto 50 ao 84. Ele injeta quatro testes: um replay de uma mensagem anterior da PS-04, uma mensagem da PS-07 selada com a chave errada e uma leitura de cloro fora da faixa, de 7,5 mg/L, da PS-07, que ocupa o lugar do minuto 20, além do silêncio da PS-09. O relay verifica o MAC, rejeita replays, coloca em quarentena os valores fora da faixa e armazena o resto; o monitoramento acrescenta uma regra de silêncio; o registro de evidências resume a execução.

Laboratório 29 (salve como lab29.py, execute com python3 lab29.py):

# Lab 29 - capstone: an end-to-end monitored telemetry service in one file (offline, standard library only)
# Synthetic stations -> authenticated relay -> validated historian (sqlite3) -> monitoring -> evidence record.
import hashlib, hmac, json, random, sqlite3
random.seed(29)
KEY = b"lantern-lab-key-not-real-0001"
LIMITS = {"LEVEL_M": (0.0, 12.0), "CL2_MGL": (0.0, 5.0)}
def seal(m, key=KEY):
    body = json.dumps(m, sort_keys=True, separators=(",", ":"))
    return body + "." + hmac.new(key, body.encode(), hashlib.sha256).hexdigest()
# 1. stations emit one reading per minute for 120 minutes; PS-09 goes silent 50-84; one injected test of each fault
wire, now0 = [], 1_790_000_000
for st, tag, base in (("PS-04", "LEVEL_M", 3.4), ("PS-07", "CL2_MGL", 0.6), ("PS-09", "LEVEL_M", 5.1)):
    for i in range(120):
        if st == "PS-09" and 50 <= i < 85: continue
        if st == "PS-07" and i == 20: continue  # the injected out-of-range test takes this slot
        wire.append((now0 + 60*i, seal({"st": st, "tag": tag, "v": round(base + random.gauss(0, 0.05), 3), "ts": now0 + 60*i, "seq": i})))
wire.append((now0 + 600, wire[3][1]))                                                         # TEST: replay
wire.append((now0 + 900, seal({"st": "PS-07", "tag": "CL2_MGL", "v": 0.6, "ts": now0 + 900, "seq": 999}, key=b"x")))  # TEST: forged
wire.append((now0 + 1200, seal({"st": "PS-07", "tag": "CL2_MGL", "v": 7.5, "ts": now0 + 1200, "seq": 20})))         # TEST: out of range
wire.sort(key=lambda w: w[0])
# 2. relay + historian
db = sqlite3.connect(":memory:")
db.execute("CREATE TABLE reading(st TEXT, tag TEXT, v REAL, ts INTEGER, seq INTEGER, PRIMARY KEY(st, seq))")
events, last = [], {}
for t, w in wire:
    body, mac = w.rsplit(".", 1)
    if not hmac.compare_digest(hmac.new(KEY, body.encode(), hashlib.sha256).hexdigest(), mac):
        events.append((t, "mac_reject", "?")); continue
    m = json.loads(body)
    if m["seq"] <= last.get(m["st"], -1):
        events.append((t, "replay_reject", m["st"])); continue
    lo, hi = LIMITS[m["tag"]]
    last[m["st"]] = m["seq"]
    if not lo <= m["v"] <= hi:
        events.append((t, "range_quarantine", m["st"])); continue
    db.execute("INSERT INTO reading VALUES (?,?,?,?,?)", (m["st"], m["tag"], m["v"], m["ts"], m["seq"]))
# 3. monitoring: security events and silence > 15 min
alerts = [(t, kind, st) for t, kind, st in events]
for st, in db.execute("SELECT DISTINCT st FROM reading"):
    ts = [r[0] for r in db.execute("SELECT ts FROM reading WHERE st=? ORDER BY ts", (st,))]
    alerts += [(a + 900, "silence", st) for a, b in zip(ts, ts[1:]) if b - a > 900]
# 4. evidence record
stored = db.execute("SELECT COUNT(*) FROM reading").fetchone()[0]
record = {"period_utc_s": [now0, now0 + 7140], "messages_on_wire": len(wire), "stored": stored,
          "alerts": sorted({(k, s) for _, k, s in alerts}), "injected_tests": 4,
          "injected_tests_detected": len({k for _, k, _ in alerts} & {"replay_reject", "mac_reject", "range_quarantine", "silence"}),
          "alert_count": len(alerts)}
blob = json.dumps(record, sort_keys=True).encode()
print(json.dumps(record, sort_keys=True))
print("evidence sha256:", hashlib.sha256(blob).hexdigest()[:24], "...")
print("PS-07 mean chlorine stored: %.3f mg/L" % db.execute("SELECT AVG(v) FROM reading WHERE st='PS-07'").fetchone()[0])

Saída esperada (Python 3.13.5):

{"alert_count": 4, "alerts": [["mac_reject", "?"], ["range_quarantine", "PS-07"], ["replay_reject", "PS-04"], ["silence", "PS-09"]], "injected_tests": 4, "injected_tests_detected": 4, "messages_on_wire": 327, "period_utc_s": [1790000000, 1790007140], "stored": 324}
evidence sha256: aa40e3905d178699f1460e1c ...
PS-07 mean chlorine stored: 0.606 mg/L

29.3 Ler o registro de evidências

Confira se os números fecham. Mensagens na rede: 120 + 119 + 85 + 3 injetadas = 327. Armazenadas: 324 (todas as leituras genuínas). Não armazenadas: 3 (o replay, a falsificação e a leitura fora da faixa). Alertas: quatro, um de cada tipo, e quatro de quatro testes injetados detectados. O digest permite que qualquer pessoa confirme depois que o registro não foi editado; num serviço real, o registro seria guardado fora do host e o digest escrito no ticket da mudança ou do teste.

Se a contagem de armazenadas fosse 325, uma leitura genuína ficaria sem explicação.

29.4 Um defeito que só aparece na combinação

Mude o número de sequência da mensagem fora da faixa injetada de 20 para 1000 — como poderia fazer um gateway de estação defeituoso depois de uma reinicialização — e execute de novo. O resultado muda drasticamente: as leituras armazenadas caem de 324 para 225 e os alertas sobem de 4 para 103. Noventa e nove leituras genuínas da PS-07 são rejeitadas como replays. A causa é que o relay atualiza o seu último número de sequência visto depois da autenticação, mas antes da verificação de faixa: a mensagem fora da faixa é autêntica, então o seu número de sequência 1000 é registrado, e toda mensagem genuína posterior (sequência 21 a 119) parece um replay.

Cada componente estava "correto" pelos testes do seu próprio capítulo. O Capítulo 3 alertou contra atualizar o estado de replay antes de todas as outras verificações passarem; o Capítulo 15 colocou a atualização depois da autenticação. A combinação mostra o buraco restante: uma mensagem autêntica, mas implausível pode envenenar o estado de replay. As correções incluem: atualizar o estado apenas na aceitação ou na quarentena de uma mensagem plausível na sequência (rejeitar saltos de sequência maiores que um limite, digamos 1.000 ou um dia de mensagens, a não ser num modo explícito de ressincronização); alertar sobre saltos grandes; e dar aos operadores um procedimento de ressincronização revisado. É exatamente por isso que o trabalho de conclusão testa o serviço integrado, e não apenas as suas partes.

29.5 Construir o trabalho de conclusão real

A versão de projeto do trabalho de conclusão, feita em equipe (Capítulo 25), constrói a mesma forma com componentes reais num laboratório: três máquinas virtuais de relay, um banco de dados do historiador, encaminhamento de logs para um coletor, o normalizador, três regras de monitoramento mais o automonitoramento, configuração como código com uma verificação de deriva, um backup e uma restauração cronometrada, runbooks e um pacote de passagem. Use apenas equipamentos de laboratório ou estações simuladas. O entregável é o registro de evidências do serviço: para cada requisito de "O que o trabalho de conclusão precisa demonstrar" acima, o teste realizado, a sua data, a saída bruta, o resultado contra uma regra de afirmação declarada de antemão (Capítulo 23) e os itens em aberto.

29.6 Erros comuns e limites

O resumo da Equipe A no Capítulo 30 comete o quinto: não declara limite algum.

Uma falsificação injetada construída com a chave real da estação não testaria nada; a equipe precisa projetar testes que possam falhar.

Exercícios — capítulo 29

EX085 — Reproduza o defeito. Objetivo: encontrar e explicar um defeito de integração. Entradas: Laboratório 29. Restrições: mude apenas o número de sequência do teste fora da faixa. Etapas: (1) defina-o como 1000 e execute; (2) relate a contagem de armazenadas, a contagem de alertas e quais mensagens foram perdidas; (3) explique a causa em termos da ordem das verificações. Evidência esperada: saída e explicação. Rubrica: 3 = armazenadas 225, alert_count 103, 99 leituras genuínas da PS-07 rejeitadas como replays, explicação de que o estado foi atualizado por uma mensagem autêntica, mas fora da faixa; 2 = números certos, explicação rasa; 1 = não executado; 0 = nada.

EX086 — Corrija. Objetivo: corrigir o defeito sem abrir um buraco de replay. Entradas: Laboratório 29 com sequência 1000. Restrições: apenas a biblioteca padrão; os replays ainda precisam ser rejeitados. Etapas: (1) implemente um salto máximo de sequência de 1.000 (rejeitando saltos maiores com um evento distinto seq_jump_reject) ou outra correção defensável; (2) execute com a sequência 1000 e com a sequência 20; (3) mostre que o teste de replay continua rejeitado nos dois casos. Evidência esperada: código e as duas saídas. Rubrica: 3 = o salto de 981 (19→1000) fica dentro de 1.000 — então o aluno precisa notar que um limite de 1.000 não pega este caso e escolher um limite mais apertado ou baseado no tempo (por exemplo, salto não maior que os minutos decorridos mais uma margem), com as 324 armazenadas restauradas e o replay ainda rejeitado; 2 = a correção funciona para este caso, mas falta o raciocínio sobre o limite; 1 = remove a proteção contra replay; 0 = nada.

EX087 — Revisão do registro de evidências. Objetivo: conferir um registro como avaliador. Entradas: a saída do Laboratório 29 com a sequência 20. Restrições: verificações apenas a partir do registro e do código do laboratório. Etapas: (1) verifique se as contagens fecham; (2) recalcule o digest a partir do JSON impresso e compare os seus primeiros 24 caracteres; (3) liste três limites que o registro deveria declarar. Evidência esperada: cálculos e lista. Rubrica: 3 = 327 = 324 + 3 confirmado; digest recalculado aplicando o hash à string JSON exata; limites como dados sintéticos, sem dispositivos de campo nem operadora, execução única, sem adversário adaptativo, sem carga em escala; 2 = duas partes; 1 = uma; 0 = nada.

Síntese do capítulo

O trabalho de conclusão integra o curso num único serviço monitorado e o comprova com testes injetados e um registro de evidências conferível. As contagens precisam fechar, o registro precisa estar protegido por um digest e guardado fora do host, e os seus limites precisam estar declarados. Os testes de integração encontram defeitos que os testes de componentes perdem: aqui, uma mensagem autêntica, mas implausível, envenenou o estado de replay, um buraco visível apenas quando a validação, a autenticação e a ordenação se encontraram.

Respostas comentadas

EX085 — Reproduza o defeito. Com a sequência do teste fora da faixa definida como 1000: "stored": 225, "alert_count": 103, e os alertas passam a incluir ["replay_reject", "PS-07"]. Perdidas: 99 leituras genuínas da PS-07 (sequência 21–119). Causa: o relay registra o último número de sequência depois da autenticação, mas antes da verificação de faixa; a mensagem autêntica fora da faixa define a última sequência da PS-07 como 1000, então toda mensagem genuína posterior tem um número menor e é rejeitada como replay.

EX086 — Corrija. Um limite de 1.000 não ajuda aqui: o salto de 19 para 1000 é de 981. Um limite baseado no tempo funciona: rejeitar (evento seq_jump_reject) quando o salto exceder os minutos decorridos desde a última mensagem aceita da estação mais uma margem de 5. Implementado no Laboratório 29 (acompanhando last_t por estação, ao lado de last), a execução com a sequência 1000 dá "stored": 324, "alert_count": 4, com seq_jump_reject para a PS-07 no lugar da quarentena por faixa, e o replay da PS-04 continua rejeitado; com a sequência 20, a saída não muda em relação à original (324 armazenadas, quatro alertas). O retorno da PS-09 depois do seu silêncio de 35 minutos é aceito, porque o seu salto de 36 fica dentro de 36 + 5. Observe que injected_tests_detected informa 3 com a sequência 1000, porque o conjunto de detecção ainda não inclui seq_jump_reject — acrescente-o.

EX087 — Revisão do registro de evidências. Contagens: mensagens na rede 327 = 120 (PS-04) + 119 (PS-07, menos o lugar ocupado pelo teste) + 85 (PS-09) + 3 mensagens injetadas; armazenadas 324 = 327 − 3 (replay rejeitado, falsificação rejeitada, fora da faixa em quarentena). Digest: aplicar o hash exatamente à linha JSON impressa (sha256 dos bytes sem a quebra de linha final) dá um digest que começa com aa40e3905d178699f1460e1c, igual ao impresso. Limites a declarar: estações e valores sintéticos; uma única execução de duas horas; sem dispositivos de campo reais, operadora ou carga em escala; sem adversário adaptativo; tratamento de chaves simplificado (chave no código-fonte).