LEARN. PRACTICE. GROW.1. Define collection outcomes
An event passes through stages: generation, sending, receipt, parsing and retention in a queryable destination. Successful sendall does not prove every stage. The lab uses three synthetic messages and loopback-only sockets; lists it calls stored are in memory rather than durable storage. Run it with Python 3.13 and compare outcomes with the manifest without targeting a real SIEM. RFC 6587 has Historic status and describes legacy framing. The plaintext TCP exercise teaches message boundaries; it is not a recommendation to send production logs without protection. TLS and destination identity belong to a separate transport design.
2. Delimit bytes, not network calls
The first text has 64 characters but occupies 68 UTF-8 bytes, including its BOM marker. The octet-counting prefix measures payload bytes rather than characters or the prefix itself. The collector reads at most seven bytes per recv, accumulates fragments and emits a message only when the announced length arrives. Three messages may arrive fragmented or together; a recv call does not define a message. The second contains an internal newline. Splitting solely at newline would create another boundary, while length preserves content. This implementation does not parse or validate the full RFC 5424 schema: it tests byte framing.
3. Make failures visible
The parser rejects zero length, nondecimal prefixes, excessive headers and frames above its instructional 1024-byte bound. That value is not a universal syslog maximum. An incomplete body at EOF is marked truncated without inventing a complete message. In a second TCP connection, the collector receives and parses an event but deliberately omits retention. With UDP, all three datagrams arrive and only two enter the retained list. No network packet loss was induced: omission occurs after receipt. The result shows why send counts do not replace destination reconciliation or a persistence policy.
4. Correlate origin and time
For investigation, retain event time, receipt time, declared origin and transport identity when available. A TLS-authenticated relay may forward events from other hosts; its certificate alone does not prove that HOSTNAME was emitted by the named host. Unsynchronized clocks make timestamp ordering uncertain. The timeQuality element can communicate limitations but must reflect actual state rather than invented precision. Use identifiers and flow context to distinguish retransmissions, beginning/end events and separate occurrences. Do not remove everything with identical text: two legitimate denials may have exactly the same message.
5. Accept the pipeline using measurable criteria
In a SIEM integration, combine an identified synthetic event with destination search, field interpretation and availability delay. Measure queues, rejections, lag and free space, including an authorized interruption and recovery behavior. Capacity planning must reflect volume and retention, and sensitive data should be limited. In FTD connection events, the end usually adds session information; the beginning may matter for denial or timely detection. A long session without an end event does not yet prove broken collection. The lab does not validate Splunk, TLS syslog, throughput or disk. Summary: confirm destination outcomes and make gaps between transport and evidence explicit.
#!/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.socketas 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_alivefor 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
The TCP collector reconstructs three messages; another connection receives one and retains zero. UDP receives three, but deliberate omission leaves two in the list.
Common pitfalls
recv as a message; characters as bytes; sending as persistence; relay TLS as authenticated origin of every payload; silence as absence of events.
Related topics: Syslog · SIEM and correlation · Clocks and NTP
A telemetry pipeline needs correct framing, context and confirmation of the retained outcome.
Reference: Transmission of Syslog Messages over TCP · 350-701 SCOR v2.0, effective 2026-08-27; core component of CCNP Security