APRENDER. PRATICAR. EVOLUIR.1. Definir o resultado da recolha
Um evento tem várias etapas: ser gerado, enviado, recebido, interpretado e conservado num destino consultável. Um sendall bem-sucedido não prova todas essas etapas. O laboratório usa três mensagens sintéticas e sockets apenas em loopback; as listas que chama stored estão em memória e não são armazenamento durável. Executa-o com Python 3.13 e compara os resultados com o manifesto, sem apontar para um SIEM real. O RFC 6587 tem estatuto Historic e descreve framing legado. O exercício TCP simples ensina limites de mensagens; não é uma recomendação para enviar logs de produção sem proteção. TLS e identidade do destino pertencem a um desenho de transporte próprio.
2. Delimitar bytes, não chamadas de rede
O primeiro texto tem 64 caracteres mas ocupa 68 bytes UTF-8, incluindo o marcador BOM. O prefixo de octet counting mede os bytes do payload, não os caracteres nem o próprio prefixo. O coletor lê no máximo sete bytes por recv, acumula os fragmentos e só entrega uma mensagem quando recebe o comprimento anunciado. Três mensagens podem chegar em fragmentos ou juntas; uma chamada recv não define uma mensagem. A segunda contém uma quebra de linha interna. Separar exclusivamente por newline produziria outra fronteira, enquanto o comprimento preserva o conteúdo. Esta implementação não interpreta nem valida todo o esquema RFC 5424: testa o enquadramento dos bytes.
3. Tornar falhas visíveis
O parser recusa comprimento zero, prefixo não decimal, cabeçalho excessivo e frames acima do limite didático de 1024 bytes. Esse valor não é um máximo universal do syslog. Um corpo incompleto no EOF é marcado como truncado, sem inventar uma mensagem completa. Numa segunda ligação TCP, o coletor recebe e interpreta o evento, mas omite deliberadamente a sua conservação. Em UDP, os três datagramas chegam e só dois entram na lista conservada. Não se provocou perda de pacotes na rede: a omissão ocorre depois da receção. O resultado mostra por que contagens de envio não substituem reconciliação no destino nem uma política de persistência.
4. Correlacionar origem e tempo
Para investigação, conserva hora do evento, hora de receção, origem declarada e identidade do transporte quando disponível. Um relay autenticado por TLS pode encaminhar eventos de outros hosts; o certificado do relay não prova por si que o campo HOSTNAME foi emitido pelo host indicado. Relógios sem sincronização tornam a ordenação por timestamp incerta. O elemento timeQuality pode comunicar limitações, mas deve refletir o estado real, não uma precisão inventada. Usa identificadores e contexto do fluxo para distinguir retransmissões, eventos de início e fim e acontecimentos diferentes. Não elimines tudo o que tem o mesmo texto: duas recusas legítimas podem ter exatamente a mesma mensagem.
5. Aceitar a cadeia com critérios mensuráveis
Numa integração com SIEM, combina um evento sintético identificado com pesquisa no destino, interpretação dos campos e tempo de disponibilização. Mede filas, rejeições, atraso e espaço disponível, incluindo uma interrupção autorizada e o comportamento de recuperação. A reserva de capacidade deve refletir volume e retenção, e os dados sensíveis devem ser limitados. Nos eventos de ligação FTD, o fim costuma acrescentar informação da sessão; o início pode ser relevante para recusa ou deteção rápida. Uma sessão longa sem evento final ainda não é prova de recolha avariada. O laboratório não valida Splunk, TLS syslog, throughput ou disco. Resumo: confirma o resultado no destino e torna explícitas as lacunas entre transporte e evidência.
#!/usr/bin/env python3
"""Original bounded loopback framing lab; not a production syslog collector."""
import hashlib,json,platform,socket,threading
from pathlib import Path
MAX_FRAME=1024 # local teaching bound, not a protocol maximum
class Frames:
def __init__(self):self.pending=b'';self.messages=[]
def feed(self,data):
self.pending+=data
while self.pending:
cut=self.pending.find(b' ')
if cut<0:
if len(self.pending)>9:raise ValueError('header-too-long')
break
prefix=self.pending[:cut]
if not prefix or prefix[:1]==b'0' or any(c<48 or c>57 for c in prefix):raise ValueError('invalid-length')
if len(prefix)>9:raise ValueError('header-too-long')
length=int(prefix)
if length>MAX_FRAME:raise ValueError('frame-too-large')
if len(self.pending)<cut+1+length:break
self.messages.append(self.pending[cut+1:cut+1+length]);self.pending=self.pending[cut+1+length:]
def finish(self):
if self.pending:raise ValueError('truncated-frame')
return self.messages
def frame(message):return str(len(message)).encode('ascii')+b' '+message
def tcp_exchange(payloads,discard=False):
# No files or production addresses: listener bound to ephemeral loopback only.
state={'stored':[],'received':[],'errors':[],'readSizes':[]}
with socket.socket()as listener:
listener.bind(('127.0.0.1',0));listener.listen(1);listener.settimeout(4)
def receive():
try:
with listener.accept()[0]as conn:
conn.settimeout(4);parser=Frames()
while True:
part=conn.recv(7)
if not part:break
state['readSizes'].append(len(part));parser.feed(part)
state['received']=parser.finish()
if not discard:state['stored']=list(state['received'])
# stored means an in-memory teaching list, never durable storage.
except Exception as exc:state['errors'].append(type(exc).__name__+': '+str(exc))
thread=threading.Thread(target=receive);thread.start()
try:
with socket.create_connection(listener.getsockname(),timeout=4)as sender:
for part in payloads:sender.sendall(part)
sender.shutdown(socket.SHUT_WR)
finally:thread.join(5)
if thread.is_alive():raise RuntimeError('listener did not finish')
if state['errors']:raise RuntimeError(state['errors'])
return state
def main():
checks=[]
def check(label,condition):
if not condition:raise AssertionError(label)
checks.append(label)
def rejected(label,payload,expected,finish=False):
try:
parser=Frames();parser.feed(payload)
if finish:parser.finish()
except ValueError as e:check(label,str(e)==expected)
else:raise AssertionError(label+' accepted')
text=['<134>1 2026-10-07T10:00:00Z lab.test dr 1 EV01 - \ufeffação=permitida', '<132>1 2026-10-07T10:00:01Z lab.test dr 1 EV02 - \ufefflinha um\nlinha dois', '<131>1 2026-10-07T10:00:02Z lab.test dr 1 EV03 - \ufeffação=recusada']
messages=[s.encode('utf-8')for s in text];stream=b''.join(map(frame,messages))
check('UTF-8 byte count differs from character count',len(messages[0])>len(text[0]))
check('frame prefix counts payload bytes',int(frame(messages[0]).split(b' ',1)[0])==len(messages[0]))
parser=Frames()
for value in stream:parser.feed(bytes([value]))
check('byte-at-a-time parser reconstructs three messages',parser.finish()==messages)
parser=Frames();parser.feed(stream)
check('coalesced frames reconstruct three messages',parser.finish()==messages)
check('embedded newline stays inside second message',b'\n'in parser.messages[1]and len(parser.messages)==3)
check('line splitting would create an extra boundary',len(messages[1].split(b'\n'))==2)
for label,payload,error,finish in [('zero length rejected',b'0 x','invalid-length',False),('leading zero rejected',b'01 x','invalid-length',False),('negative length rejected',b'-1 x','invalid-length',False),('nondecimal length rejected',b'x hi','invalid-length',False),('oversize rejected before body',b'1025 ','frame-too-large',False),('header length bounded',b'1111111111','header-too-long',False),('incomplete body rejected at EOF',b'5 abc','truncated-frame',True),('incomplete prefix rejected at EOF',b'12','truncated-frame',True)]:rejected(label,payload,error,finish)
parser=Frames();parser.feed(frame(b'x'*1024));check('configured maximum accepted',len(parser.finish()[0])==1024)
parser=Frames();parser.feed(b'');check('empty stream has no events',parser.finish()==[])
live=tcp_exchange([stream[:1],stream[1:5],stream[5:]])
check('real TCP collector reconstructs all messages',live['received']==messages)
check('collector in-memory list matches messages',live['stored']==messages)
check('TCP receiver uses bounded reads',all(0<n<=7 for n in live['readSizes']))
check('TCP message spans multiple recv calls',len(live['readSizes'])>3)
check('UTF-8 retained after TCP framing',live['stored'][0].decode('utf-8')==text[0])
check('embedded newline retained after TCP framing',live['stored'][1]==messages[1])
discarded=tcp_exchange([frame(messages[0])],discard=True)
check('sendall completed and receiver parsed discard case',discarded['received']==[messages[0]])
check('parsed event can be intentionally unretained',discarded['stored']==[])
with socket.socket(socket.AF_INET,socket.SOCK_DGRAM)as receiver, socket.socket(socket.AF_INET,socket.SOCK_DGRAM)as sender:
receiver.bind(('127.0.0.1',0));receiver.settimeout(3);observed=[];kept=[];sent=[]
for index,message in enumerate(messages):
sent.append(sender.sendto(message,receiver.getsockname()));data,_=receiver.recvfrom(2048);observed.append(data)
if index!=1:kept.append(data) # Deliberate collector omission, not network loss.
check('UDP sends return full datagram lengths',sent==list(map(len,messages)))
check('three UDP datagrams actually received',observed==messages)
check('collector deliberately retains only two UDP events',kept==[messages[0],messages[2]])
check('UDP omission is visible against expected event set',messages[1]not in kept)
check('no receiver threads remain',not any(t.name!='MainThread'and t.is_alive()for t in threading.enumerate()))
evidence={'passed':len(checks),'failed':0,'checks':checks,'python':platform.python_version(),'scriptSHA256':hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),'actualLoopbackTCP':True,'actualLoopbackUDP':True,'actualCiscoDevice':False,'actualIPsecTunnel':False,'tlsUsedInThisLab':False,'durableStorageTested':False,'networkPacketLossInjected':False,'collectorOmissionIntentional':True,'productionSyslogConformanceClaimed':False,'observations':{'messageByteLengths':list(map(len,messages)),'messageCharacterLengths':list(map(len,text)),'tcpReconstructed':len(live['received']),'tcpRetainedInMemory':len(live['stored']),'tcpDiscardCaseReceived':len(discarded['received']),'tcpDiscardCaseRetained':len(discarded['stored']),'udpReceived':len(observed),'udpRetainedInMemory':len(kept),'udpMissingMessageIds':['EV02'],'localMaxFrameBytes':MAX_FRAME},'cleanup':'All loopback sockets closed and receiver threads joined.'}
print(json.dumps(evidence,ensure_ascii=False,indent=2))
if __name__=='__main__':main()
O coletor TCP reconstrói três mensagens; outra ligação recebe uma e conserva zero. UDP recebe três, mas a omissão deliberada deixa duas na lista.
Armadilhas comuns
recv como mensagem; caracteres como bytes; envio como persistência; TLS do relay como origem autenticada de todo o payload; silêncio como ausência de eventos.
Tópicos relacionados: Syslog · SIEM e correlação · Relógios e NTP
A cadeia de telemetria precisa de framing correto, contexto e confirmação do resultado conservado.
Referência: Transmission of Syslog Messages over TCP · 350-701 SCOR v2.0, effective 2026-08-27; core component of CCNP Security