#!/usr/bin/env python3 """Test client for the RFC 5424 syslog server (UDP only). Service discovery: broadcast an empty datagram to 255.255.255.255:8888, the server's reply source address is the syslog server IP. Then log messages go to :514 (RFC 5424 format, no structured data). Examples: python syslog_client.py "hello world" python syslog_client.py -s 4 -f 1 --app myapp "disk almost full" python syslog_client.py --server 192.168.0.10 "skip discovery" python syslog_client.py --oversize 2000 "too long on purpose" python syslog_client.py --raw "garbage, not RFC 5424" """ import argparse import os import socket import sys from datetime import datetime, timezone DISCOVERY_PORT = 8888 SYSLOG_PORT = 514 BROADCAST = "255.255.255.255" def discover(timeout, retries, discovery_port, broadcast=BROADCAST): sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1) sock.settimeout(timeout) try: for attempt in range(1, retries + 1): print(f"[discovery] broadcast {attempt}/{retries} -> {broadcast}:{discovery_port}") sock.sendto(b"", (broadcast, discovery_port)) try: _, (ip, port) = sock.recvfrom(1024) print(f"[discovery] reply from {ip}:{port}") return ip except socket.timeout: continue finally: sock.close() return None def build_message(facility, severity, app, procid, msgid, text, hostname): pri = facility * 8 + severity # RFC 5424 timestamp: full-date "T" full-time, UTC with Z ts = datetime.now(timezone.utc).isoformat(timespec="milliseconds").replace("+00:00", "Z") # HEADER SP STRUCTURED-DATA SP MSG ; SD is NILVALUE "-" return f"<{pri}>1 {ts} {hostname} {app} {procid} {msgid} - {text}" def run_demo(server, port, hostname, procid): bm = lambda fac, sev, app, msgid, text, host=hostname: build_message( fac, sev, app, procid, msgid, text, host).encode("utf-8") ts = datetime.now(timezone.utc).isoformat(timespec="milliseconds").replace("+00:00", "Z") cases = [ ("valid: user.info", bm(1, 6, "demo", "ID1", "hello from demo client")), ("valid: kernel.emerg", bm(0, 0, "kernel", "BOOT", "system on fire")), ("valid: local7.debug", bm(23, 7, "myapp", "-", "debug details")), ("valid: nil fields", f"<14>1 {ts} - - - - - only nil header".encode()), ("valid: UTF-8 text with BOM", f"<14>1 {ts} {hostname} demo {procid} U1 - ".encode() + b"\xef\xbb\xbf" + "Grüße, 世界".encode("utf-8")), ("valid: no message body", f"<14>1 {ts} {hostname} demo {procid} E1 -".encode()), ("valid: ~480 bytes", bm(1, 6, "demo", "LEN", "A" * 400)), ("INVALID: too long (3000 B)", bm(1, 6, "demo", "LEN", "B" * 3000)), ("INVALID: APP-NAME 49 chars", bm(1, 6, "a" * 49, "ID1", "app name too long")), ("INVALID: MSGID 33 chars", bm(1, 6, "demo", "m" * 33, "msgid too long")), ("INVALID: no PRI", f"1 {ts} {hostname} demo {procid} ID1 - no pri".encode()), ("INVALID: PRI out of range <192>", f"<192>1 {ts} {hostname} demo {procid} ID1 - bad pri".encode()), ("INVALID: garbage", b"this is not syslog"), ("INVALID: empty datagram", b""), ] sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) try: for label, data in cases: sock.sendto(data, (server, port)) print(f"[demo] {label:34} {len(data):5} B {data[:60]!r}") finally: sock.close() return 0 def main(): p = argparse.ArgumentParser(description="Syslog (RFC 5424, UDP) test client") p.add_argument("message", nargs="?", help="log message text (not needed with --demo)") p.add_argument("--demo", action="store_true", help="discover server, then send a set of example packets (valid and invalid)") p.add_argument("-s", "--severity", type=int, default=6, choices=range(8), help="0=emerg .. 7=debug (default 6=info)") p.add_argument("-f", "--facility", type=int, default=1, choices=range(24), help="0..23 (default 1=user)") p.add_argument("--app", default="testclient", help="APP-NAME") p.add_argument("--msgid", default="-", help="MSGID") p.add_argument("--hostname", default=socket.gethostname().replace(" ", "_") or "-") p.add_argument("-n", "--count", type=int, default=1, help="send message N times") p.add_argument("--server", help="server IP, skips auto-discovery") p.add_argument("--port", type=int, default=SYSLOG_PORT, help="syslog port (default 514)") p.add_argument("--discovery-port", type=int, default=DISCOVERY_PORT) p.add_argument("--broadcast", default=BROADCAST, metavar="ADDR", help="discovery target (default 255.255.255.255; use 127.255.255.255 " "or 127.0.0.1 to stay on localhost)") p.add_argument("--timeout", type=float, default=2.0, help="discovery wait per try (s)") p.add_argument("--retries", type=int, default=3) p.add_argument("--oversize", type=int, metavar="BYTES", help="pad message to this many bytes total (test server length check)") p.add_argument("--raw", action="store_true", help="send MESSAGE verbatim, no RFC 5424 header (test server validation)") args = p.parse_args() server = args.server or discover(args.timeout, args.retries, args.discovery_port, args.broadcast) if not server: print("[error] no server found", file=sys.stderr) return 1 procid = str(os.getpid()) if args.demo: return run_demo(server, args.port, args.hostname, procid) if not args.message: p.error("message required unless --demo") if args.raw: data = args.message.encode("utf-8") else: data = build_message(args.facility, args.severity, args.app, procid, args.msgid, args.message, args.hostname).encode("utf-8") if args.oversize and len(data) < args.oversize: data += b"X" * (args.oversize - len(data)) sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) try: for i in range(args.count): try: sock.sendto(data, (server, args.port)) except OSError as e: print(f"[error] send failed ({len(data)} bytes): {e}", file=sys.stderr) return 1 print(f"[send] {i + 1}/{args.count} {len(data)} bytes -> {server}:{args.port}") finally: sock.close() return 0 if __name__ == "__main__": sys.exit(main())