From 51892a02c0a2b37f942733004fa998f4aa3e16c3 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Sun, 18 Oct 2026 23:09:07 -0400 Subject: [PATCH] feat(syslog): a syslog receiver: UDP and TCP, senders offered before they're kept, each source's lines in a ring (#5999) - plugins/syslog: RFC 5424 and RFC 3164 parsing, RFC 6587 TCP framing. - Lines are queued as they arrive and stored every 5 s through a Holding, so a storage stall delays them rather than losing them. - A new sender is an offered source, counted and not kept, until an admin adds it on the Syslog page (name, host); it can be ignored instead. - The listener follows the plugin's settings (port, UDP, TCP), rebinding within seconds of a save; a port it can't open is said on the page. - Hourly ring per source: age window and byte cap from Settings. Co-Authored-By: Claude Opus 5.5 --- plugins/syslog/__init__.py | 40 +++ plugins/syslog/migrations/__init__.py | 0 .../syslog/migrations/versions/__init__.py | 0 .../versions/syslog_001_sources_lines.py | 56 ++++ plugins/syslog/models.py | 67 +++++ plugins/syslog/parse.py | 161 +++++++++++ plugins/syslog/plugin.yaml | 20 ++ plugins/syslog/receiver.py | 260 ++++++++++++++++++ plugins/syslog/retention.py | 59 ++++ plugins/syslog/routes.py | 182 ++++++++++++ plugins/syslog/templates/syslog/edit.html | 48 ++++ plugins/syslog/templates/syslog/index.html | 114 ++++++++ steward/core/settings.py | 1 + steward/settings/routes.py | 2 +- tests/integration/test_syslog.py | 100 +++++++ tests/plugins/syslog/__init__.py | 0 tests/plugins/syslog/test_parse.py | 61 ++++ tests/plugins/syslog/test_receiver.py | 96 +++++++ tests/plugins/syslog/test_syslog_page.py | 73 +++++ 19 files changed, 1339 insertions(+), 1 deletion(-) create mode 100644 plugins/syslog/__init__.py create mode 100644 plugins/syslog/migrations/__init__.py create mode 100644 plugins/syslog/migrations/versions/__init__.py create mode 100644 plugins/syslog/migrations/versions/syslog_001_sources_lines.py create mode 100644 plugins/syslog/models.py create mode 100644 plugins/syslog/parse.py create mode 100644 plugins/syslog/plugin.yaml create mode 100644 plugins/syslog/receiver.py create mode 100644 plugins/syslog/retention.py create mode 100644 plugins/syslog/routes.py create mode 100644 plugins/syslog/templates/syslog/edit.html create mode 100644 plugins/syslog/templates/syslog/index.html create mode 100644 tests/integration/test_syslog.py create mode 100644 tests/plugins/syslog/__init__.py create mode 100644 tests/plugins/syslog/test_parse.py create mode 100644 tests/plugins/syslog/test_receiver.py create mode 100644 tests/plugins/syslog/test_syslog_page.py diff --git a/plugins/syslog/__init__.py b/plugins/syslog/__init__.py new file mode 100644 index 0000000..2fa05e8 --- /dev/null +++ b/plugins/syslog/__init__.py @@ -0,0 +1,40 @@ +# plugins/syslog/__init__.py +from __future__ import annotations +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from quart import Quart + +_app: "Quart | None" = None + + +def setup(app: "Quart") -> None: + global _app + _app = app + from .models import SyslogLine, SyslogSource # noqa: F401 — registers the tables + from .receiver import close + + # The listener is bound by the receive task, so it follows the settings + # and starts on a live enable too; it is closed with the app. + app.after_serving(close) + + +def get_scheduled_tasks() -> list: + from .receiver import make_receive_task + from .retention import make_prune_task + return [make_receive_task(_app), make_prune_task(_app)] + + +def get_blueprint(): + from .routes import syslog_bp + return syslog_bp + + +def get_nav() -> list[dict]: + return [{"label": "Syslog", "href": "/plugins/syslog/"}] + + +def get_audit_entities() -> dict: + """Audit rows about a source name it and link to the Syslog page.""" + from .models import SyslogSource + return {"syslog_source": (SyslogSource, "/plugins/syslog/")} diff --git a/plugins/syslog/migrations/__init__.py b/plugins/syslog/migrations/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/plugins/syslog/migrations/versions/__init__.py b/plugins/syslog/migrations/versions/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/plugins/syslog/migrations/versions/syslog_001_sources_lines.py b/plugins/syslog/migrations/versions/syslog_001_sources_lines.py new file mode 100644 index 0000000..ae28472 --- /dev/null +++ b/plugins/syslog/migrations/versions/syslog_001_sources_lines.py @@ -0,0 +1,56 @@ +"""Syslog: the devices that send it, and the lines kept from those added + +A sender is a row whether or not it has been added: one not yet added is +counted and offered, never stored (milestone 532, #5999). + +Revision ID: syslog_001_sources_lines +Revises: (none — a branch off core via depends_on) +Create Date: 2026-10-19 +""" +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + +revision: str = "syslog_001_sources_lines" +down_revision: Union[str, None] = None +branch_labels: Union[str, Sequence[str], None] = "syslog" +depends_on: Union[str, Sequence[str], None] = ("0004_app_settings",) + + +def upgrade() -> None: + op.create_table( + "syslog_sources", + sa.Column("id", sa.String(36), primary_key=True), + sa.Column("address", sa.String(64), nullable=False, unique=True), + sa.Column("state", sa.String(10), nullable=False, server_default="offered"), + sa.Column("name", sa.String(100), nullable=False, server_default=""), + sa.Column("host_id", sa.String(36), sa.ForeignKey("hosts.id", ondelete="SET NULL"), + nullable=True), + sa.Column("hostname_seen", sa.String(255), nullable=False, server_default=""), + sa.Column("first_seen_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("last_seen_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("lines_total", sa.BigInteger, nullable=False, server_default="0"), + sa.Column("last_severity", sa.SmallInteger, nullable=True), + sa.Column("last_line", sa.Text, nullable=False, server_default=""), + ) + op.create_table( + "syslog_lines", + sa.Column("id", sa.String(36), primary_key=True), + sa.Column("source_id", sa.String(36), + sa.ForeignKey("syslog_sources.id", ondelete="CASCADE"), nullable=False), + sa.Column("received_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("sent_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("facility", sa.SmallInteger, nullable=False), + sa.Column("severity", sa.SmallInteger, nullable=False), + sa.Column("hostname", sa.String(255), nullable=False, server_default=""), + sa.Column("app", sa.String(128), nullable=False, server_default=""), + sa.Column("message", sa.Text, nullable=False, server_default=""), + ) + op.create_index("ix_syslog_lines_source_time", "syslog_lines", ["source_id", "received_at"]) + op.create_index("ix_syslog_lines_received", "syslog_lines", ["received_at"]) + + +def downgrade() -> None: + op.drop_table("syslog_lines") + op.drop_table("syslog_sources") diff --git a/plugins/syslog/models.py b/plugins/syslog/models.py new file mode 100644 index 0000000..d5723ca --- /dev/null +++ b/plugins/syslog/models.py @@ -0,0 +1,67 @@ +# plugins/syslog/models.py +from __future__ import annotations + +from datetime import datetime, timezone + +from sqlalchemy import BigInteger, DateTime, ForeignKey, Index, SmallInteger, String, Text +from sqlalchemy.orm import Mapped, mapped_column + +from steward.core.ids import time_ordered_id +from steward.models.base import Base + + +class SyslogSource(Base): + """A device that sends syslog to Steward, one row per sending address. + + A sender Steward hasn't been told about is a row too, `state="offered"`: + its lines are counted and its newest kept as a sample, so the page can + offer it ("1,204 lines from 192.168.0.7: add as a source?"), but none is + stored. Adding it makes it `"accepted"` and its lines are kept from then + on; `"ignored"` stops the offer and keeps counting. + """ + __tablename__ = "syslog_sources" + + id: Mapped[str] = mapped_column(String(36), primary_key=True, default=time_ordered_id) + address: Mapped[str] = mapped_column(String(64), nullable=False, unique=True) + state: Mapped[str] = mapped_column(String(10), nullable=False, default="offered") + name: Mapped[str] = mapped_column(String(100), nullable=False, default="") + host_id: Mapped[str | None] = mapped_column( + String(36), ForeignKey("hosts.id", ondelete="SET NULL"), nullable=True) + # Steward's name for the device, from what the lines say they come from. + hostname_seen: Mapped[str] = mapped_column(String(255), nullable=False, default="") + first_seen_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, default=lambda: datetime.now(timezone.utc)) + last_seen_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, default=lambda: datetime.now(timezone.utc)) + lines_total: Mapped[int] = mapped_column(BigInteger, nullable=False, default=0) + last_severity: Mapped[int | None] = mapped_column(SmallInteger, nullable=True) + last_line: Mapped[str] = mapped_column(Text, nullable=False, default="") + + +class SyslogLine(Base): + """One line an accepted source sent. A state table, not a time series + (System 29): bounded by a per-source size and age ring, like docker_logs. + `received_at` is when Steward got it, the time to trust; `sent_at` is the + device's own, kept only when the line says its zone (RFC 5424).""" + __tablename__ = "syslog_lines" + + id: Mapped[str] = mapped_column(String(36), primary_key=True, default=time_ordered_id) + source_id: Mapped[str] = mapped_column( + String(36), ForeignKey("syslog_sources.id", ondelete="CASCADE"), nullable=False) + received_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) + sent_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + facility: Mapped[int] = mapped_column(SmallInteger, nullable=False) + severity: Mapped[int] = mapped_column(SmallInteger, nullable=False) + hostname: Mapped[str] = mapped_column(String(255), nullable=False, default="") + app: Mapped[str] = mapped_column(String(128), nullable=False, default="") + message: Mapped[str] = mapped_column(Text, nullable=False, default="") + + # A source's lines newest first, and the ring's age cut; the twin-index + # idiom of docker_logs. + __table_args__ = ( + Index("ix_syslog_lines_source_time", "source_id", "received_at"), + Index("ix_syslog_lines_received", "received_at"), + ) + + +STATES = ("offered", "accepted", "ignored") diff --git a/plugins/syslog/parse.py b/plugins/syslog/parse.py new file mode 100644 index 0000000..b9d6b21 --- /dev/null +++ b/plugins/syslog/parse.py @@ -0,0 +1,161 @@ +# plugins/syslog/parse.py +"""One syslog message read into its parts: RFC 5424 (the IETF format) and +RFC 3164 (the older BSD format), and whatever a device sends that is neither. + +Pure: no I/O and no clock, so every format a device has been seen to send +is a unit test. + +A message that doesn't parse is still kept, whole, as its own text: a line +the device sent is never dropped for its shape. +""" +from __future__ import annotations + +import re +from dataclasses import dataclass +from datetime import datetime + +SEVERITIES = ("emerg", "alert", "crit", "err", "warning", "notice", "info", "debug") +FACILITIES = ("kern", "user", "mail", "daemon", "auth", "syslog", "lpr", "news", "uucp", "cron", + "authpriv", "ftp", "ntp", "audit", "alert", "clock", + "local0", "local1", "local2", "local3", "local4", "local5", "local6", "local7") + +# The longest message kept; a device that sends more is cut here. +MAX_MESSAGE = 8192 +# A message with no is user.notice (RFC 3164 §4.3.3). +_DEFAULT_PRI = 13 + +_PRI = re.compile(r"<(\d{1,3})>") +# 1 TIMESTAMP HOSTNAME APP-NAME PROCID MSGID [SD] MSG +_RFC5424 = re.compile(r"(\d{1,2}) (\S+) (\S+) (\S+) (\S+) (\S+) ?(.*)", re.S) +# Mmm dd hh:mm:ss (the day space-padded) +_BSD_TIME = re.compile(r"([A-Z][a-z]{2}) {1,2}(\d{1,2}) (\d{2}):(\d{2}):(\d{2}) ") +# TAG[pid]: or TAG: at the start of a BSD message +_TAG = re.compile(r"([^\s:\[\]]{1,128})(?:\[[^\]]*\])?: ?") +_MONTHS = ("Jan", "Feb", "Mar", "Apr", "May", "Jun", "Jul", "Aug", "Sep", "Oct", "Nov", "Dec") + + +@dataclass +class Message: + facility: int + severity: int + sent_at: datetime | None # only when the message says its own zone (RFC 5424) + hostname: str + app: str + text: str + + +def parse(raw: bytes | str) -> Message: + """Read one message. A BSD timestamp has no year and no zone, so it is + read past and never kept as the sent time: the time Steward received the + line is the one to trust.""" + if isinstance(raw, bytes): + raw = raw.decode("utf-8", errors="replace") + raw = raw.lstrip("\ufeff").rstrip("\r\n\x00 ") + pri, rest = _DEFAULT_PRI, raw + m = _PRI.match(raw) + if m and int(m.group(1)) <= 191: + pri, rest = int(m.group(1)), raw[m.end():] + facility, severity = divmod(pri, 8) + + m = _RFC5424.match(rest) + if m and m.group(1) == "1": + return _rfc5424(facility, severity, m) + return _bsd(facility, severity, rest) + + +def _nil(value: str) -> str: + return "" if value == "-" else value + + +def _rfc5424(facility: int, severity: int, m: re.Match) -> Message: + _, stamp, host, app, _procid, _msgid, tail = m.groups() + text = _strip_structured_data(tail) + return Message(facility, severity, _iso(stamp), _nil(host)[:255], _nil(app)[:128], + text.lstrip("\ufeff")[:MAX_MESSAGE]) + + +def _strip_structured_data(tail: str) -> str: + """Drop RFC 5424's STRUCTURED-DATA ("-" or one or more [id k="v"] + elements) from in front of the message.""" + if tail.startswith("- ") or tail == "-": + return tail[2:] + i = 0 + while i < len(tail) and tail[i] == "[": + quoted, j = False, i + 1 + while j < len(tail): + c = tail[j] + if c == "\\": + j += 2 + continue + if c == '"': + quoted = not quoted + elif c == "]" and not quoted: + break + j += 1 + i = j + 1 + return tail[i:].lstrip(" ") + + +def _iso(stamp: str) -> datetime | None: + if stamp == "-": + return None + try: + at = datetime.fromisoformat(stamp.replace("Z", "+00:00")) + except ValueError: + return None + return at if at.tzinfo else None + + +def _bsd(facility: int, severity: int, rest: str) -> Message: + hostname = "" + m = _BSD_TIME.match(rest) + if m and m.group(1) in _MONTHS: + rest = rest[m.end():] + # The HOSTNAME follows the time, unless the sender left it out and the + # tag comes straight after (some devices do). + word, _, after = rest.partition(" ") + if word and not word.endswith(":") and "[" not in word: + hostname, rest = word[:255], after + app = "" + t = _TAG.match(rest) + if t: + app, rest = t.group(1)[:128], rest[t.end():] + return Message(facility, severity, None, hostname, app, rest[:MAX_MESSAGE]) + + +def severity_name(level: int) -> str: + return SEVERITIES[level] if 0 <= level < len(SEVERITIES) else str(level) + + +def facility_name(code: int) -> str: + return FACILITIES[code] if 0 <= code < len(FACILITIES) else str(code) + + +def frames(buffer: bytearray) -> list[bytes]: + """Take every complete message out of a TCP stream's buffer (RFC 6587): + octet-counted (" ") or newline-ended. What is left is a message + still arriving. A count past MAX_MESSAGE × 2 is not a count: the stream is + read as newline-ended from there.""" + out: list[bytes] = [] + while buffer: + if buffer[:1].isdigit(): + space = buffer.find(b" ", 0, 8) + if space > 0 and buffer[:space].isdigit() and int(buffer[:space]) <= MAX_MESSAGE * 2: + size = int(buffer[:space]) + end = space + 1 + size + if len(buffer) < end: + break + out.append(bytes(buffer[space + 1:end])) + del buffer[:end] + continue + nl = buffer.find(b"\n") + if nl < 0: + if len(buffer) > MAX_MESSAGE * 2: # no end in sight: take what's there + out.append(bytes(buffer)) + buffer.clear() + break + line = bytes(buffer[:nl]).rstrip(b"\r\x00") + del buffer[:nl + 1] + if line: + out.append(line) + return out diff --git a/plugins/syslog/plugin.yaml b/plugins/syslog/plugin.yaml new file mode 100644 index 0000000..5886d17 --- /dev/null +++ b/plugins/syslog/plugin.yaml @@ -0,0 +1,20 @@ +# plugins/syslog/plugin.yaml +name: syslog +version: "1.0.0" +description: "Receives syslog from NASes, routers and other devices, and keeps the lines of the sources you add." +author: "Steward" +license: "MIT" +min_app_version: "0.1.0" +tags: + - syslog + - logs + +# Sources are added from the Syslog page (/plugins/syslog/), where a device +# that starts sending is offered. Steward runs as an unprivileged user, so +# the port is above 1024: point devices at it, or map 514 to it. +config: + port: 5514 + udp: true + tcp: true + retention_days: 14 + max_bytes_per_source: 50000000 diff --git a/plugins/syslog/receiver.py b/plugins/syslog/receiver.py new file mode 100644 index 0000000..3e1315c --- /dev/null +++ b/plugins/syslog/receiver.py @@ -0,0 +1,260 @@ +# plugins/syslog/receiver.py +"""The syslog listener: UDP and TCP on one port, every line taken as it +arrives and stored a few seconds later. + +A line is never written from inside the socket's callback. It is queued with +the time it arrived and the address it came from, and the receive task, every +few seconds, reads what is queued into a round and hands it to a Holding +(steward/core/holding.py). The NAS that stalls the storage under Postgres is +one of the devices sending, so its lines about the stall must wait out the +stall in memory, not be lost to it. + +The listener follows the plugin's settings: the receive task binds it, and +binds it again when the port or a protocol is changed, so a save in Settings +takes effect within seconds with no restart (rule 25). A port that can't be +bound is said on the Syslog page, and tried again every minute. +""" +from __future__ import annotations + +import asyncio +import logging +from collections import deque +from dataclasses import dataclass +from datetime import datetime, timedelta, timezone +from typing import TYPE_CHECKING + +from sqlalchemy import func, select + +from steward.core.scheduler import ScheduledTask + +from .parse import MAX_MESSAGE, Message, frames, parse + +if TYPE_CHECKING: + from quart import Quart + +logger = logging.getLogger(__name__) + +# Lines received and not yet read into a round. Past this the oldest go, and +# the page says how many: a flood is bounded in memory. +MAX_QUEUED = 20_000 +# The most lines one round takes; the rest wait for the next. +ROUND_LINES = 2_000 +# Rounds held while the database doesn't take them: half an hour at the +# five-second receive. +HELD_ROUNDS = 360 +RECEIVE_S = 5 +# How long a receive waits for its own store before the next one may start. +STORE_WAIT_S = 4.0 +# How long after a failed bind it is tried again. +RETRY_BIND = timedelta(minutes=1) +# Senders known at most, added or offered: a new one past them is counted, +# not kept, so a spray of forged UDP sources can't fill the table. +MAX_SENDERS = 200 +# TCP connections at once, and how long one may say nothing before it is +# closed (a sender reconnects when it next has a line). +MAX_CONNECTIONS = 64 +TCP_IDLE_S = 3600.0 +# The sample of a source's newest line kept on its row. +SAMPLE_CHARS = 500 + + +@dataclass +class Listener: + """What the listener is bound to, and what it has taken since it started. + The page reads it as it is; nothing here is stored.""" + port: int | None = None + udp: bool = False + tcp: bool = False + error: str | None = None + tried_at: datetime | None = None + bound_at: datetime | None = None + received: int = 0 + dropped: int = 0 # queued lines lost to MAX_QUEUED + unoffered: int = 0 # lines from senders past MAX_SENDERS + connections: int = 0 + _udp_transport: object = None + _tcp_server: object = None + + @property + def listening(self) -> bool: + return self._udp_transport is not None or self._tcp_server is not None + + +LISTENER = Listener() +# (received_at, address, raw) +QUEUE: deque[tuple[datetime, str, bytes]] = deque() + + +def take(address: str, raw: bytes, now: datetime | None = None) -> None: + """Queue one received message.""" + if not raw: + return + LISTENER.received += 1 + QUEUE.append((now or datetime.now(timezone.utc), address, raw[:MAX_MESSAGE * 2])) + while len(QUEUE) > MAX_QUEUED: + QUEUE.popleft() + LISTENER.dropped += 1 + + +class _Udp(asyncio.DatagramProtocol): + def datagram_received(self, data: bytes, addr) -> None: + take(addr[0], data) + + +async def _tcp_client(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None: + peer = writer.get_extra_info("peername") + address = peer[0] if peer else "?" + if LISTENER.connections >= MAX_CONNECTIONS: + writer.close() + return + LISTENER.connections += 1 + buffer = bytearray() + try: + while chunk := await asyncio.wait_for(reader.read(65536), TCP_IDLE_S): + buffer += chunk + for message in frames(buffer): + take(address, message) + for message in frames(buffer + b"\n"): # the last line, unended + take(address, message) + except (ConnectionError, asyncio.IncompleteReadError, TimeoutError): + pass + finally: + LISTENER.connections -= 1 + writer.close() + + +def settings(app: "Quart") -> tuple[int, bool, bool]: + cfg = (app.config.get("PLUGINS") or {}).get("syslog") or {} + return int(cfg.get("port", 5514)), bool(cfg.get("udp", True)), bool(cfg.get("tcp", True)) + + +async def close() -> None: + if LISTENER._udp_transport is not None: + LISTENER._udp_transport.close() + if LISTENER._tcp_server is not None: + LISTENER._tcp_server.close() + await LISTENER._tcp_server.wait_closed() + LISTENER._udp_transport = LISTENER._tcp_server = None + LISTENER.bound_at = None + + +async def ensure_listening(app: "Quart", now: datetime | None = None) -> None: + """Bind the listener as the settings say, when it isn't already: at + start, after a change of port or protocol, and a minute after a failure.""" + now = now or datetime.now(timezone.utc) + port, udp, tcp = settings(app) + same = (port, udp, tcp) == (LISTENER.port, LISTENER.udp, LISTENER.tcp) + if same and LISTENER.error is None and (LISTENER.listening or not (udp or tcp)): + return + if same and LISTENER.error and LISTENER.tried_at and now - LISTENER.tried_at < RETRY_BIND: + return + await close() + LISTENER.port, LISTENER.udp, LISTENER.tcp = port, udp, tcp + LISTENER.error, LISTENER.tried_at = None, now + loop = asyncio.get_running_loop() + failed = [] + if udp: + try: + LISTENER._udp_transport, _ = await loop.create_datagram_endpoint( + _Udp, local_addr=("0.0.0.0", port)) + except OSError as exc: + failed.append(f"UDP: {exc.strerror or exc}") + if tcp: + try: + LISTENER._tcp_server = await asyncio.start_server(_tcp_client, "0.0.0.0", port) + except OSError as exc: + failed.append(f"TCP: {exc.strerror or exc}") + if failed: + LISTENER.error = f"Port {port} couldn't be opened ({'; '.join(failed)})." + logger.warning("Syslog: %s", LISTENER.error) + if LISTENER.listening: + LISTENER.bound_at = now + bound = (("UDP", LISTENER._udp_transport), ("TCP", LISTENER._tcp_server)) + logger.info("Syslog: listening on %s port %d", " and ".join(p for p, on in bound if on), port) + + +@dataclass +class Received: + at: datetime + address: str + message: Message + + +def read_round(limit: int = ROUND_LINES) -> list[Received]: + """Take up to `limit` queued lines, oldest first, each read into its parts.""" + out = [] + while QUEUE and len(out) < limit: + at, address, raw = QUEUE.popleft() + out.append(Received(at, address, parse(raw))) + return out + + +def make_receive_task(app: "Quart") -> ScheduledTask: + async def receive() -> None: + await _receive(app) + + return ScheduledTask(name="syslog_receive", coro_factory=receive, + interval_seconds=RECEIVE_S, run_on_startup=True) + + +async def _receive(app: "Quart") -> None: + from steward.core.holding import holding + + await ensure_listening(app) + held = holding(app, "Syslog", store=lambda rounds: _store(app, rounds), limit=HELD_ROUNDS) + lines = read_round() + if lines: + held.hold(lines) + await held.flush(STORE_WAIT_S) + + +async def _store(app: "Quart", rounds: list[list[Received]]) -> None: + """Write held rounds, oldest first, in one transaction: each sender's + count, newest line and when it was last heard, and the lines of the + sources that have been added. A sender not seen before becomes an + offered source; its lines are counted and not kept.""" + async with app.db_sessionmaker() as session: + async with session.begin(): + await store_lines(session, [line for lines in rounds for line in lines]) + + +async def store_lines(session, lines: list[Received]) -> None: + from .models import SyslogLine, SyslogSource + + if not lines: + return + by_address: dict[str, list[Received]] = {} + for line in lines: + by_address.setdefault(line.address, []).append(line) + sources = {s.address: s for s in (await session.execute( + select(SyslogSource).where(SyslogSource.address.in_(list(by_address))))).scalars()} + new = [a for a in by_address if a not in sources] + if new: + known = (await session.execute(select(func.count()).select_from(SyslogSource))).scalar_one() + for address in new: + if known >= MAX_SENDERS: + LISTENER.unoffered += len(by_address.pop(address)) + continue + first = by_address[address][0] + source = SyslogSource(address=address, first_seen_at=first.at, last_seen_at=first.at) + session.add(source) + sources[address] = source + known += 1 + await session.flush() + + for address, received in by_address.items(): + source = sources[address] + newest = received[-1] + source.lines_total = (source.lines_total or 0) + len(received) + source.last_seen_at = max(source.last_seen_at, newest.at) if source.last_seen_at else newest.at + source.last_severity = newest.message.severity + source.last_line = newest.message.text[:SAMPLE_CHARS] + named = next((r.message.hostname for r in reversed(received) if r.message.hostname), "") + if named: + source.hostname_seen = named[:255] + if source.state == "accepted": + session.add_all([SyslogLine( + source_id=source.id, received_at=r.at, sent_at=r.message.sent_at, + facility=r.message.facility, severity=r.message.severity, + hostname=r.message.hostname, app=r.message.app, message=r.message.text, + ) for r in received]) diff --git a/plugins/syslog/retention.py b/plugins/syslog/retention.py new file mode 100644 index 0000000..c7cc21a --- /dev/null +++ b/plugins/syslog/retention.py @@ -0,0 +1,59 @@ +# plugins/syslog/retention.py +"""The syslog_lines ring: each source's lines kept for the age window and +within its byte cap, whichever is less, the same shape as the docker_logs +ring (plugins/docker/retention.py). Run hourly by the plugin's own task, with +both bounds read from its settings each run (rule 25).""" +from __future__ import annotations + +import logging +from datetime import datetime, timedelta, timezone +from typing import TYPE_CHECKING + +from sqlalchemy import delete, func, select + +from steward.core.scheduler import ScheduledTask + +if TYPE_CHECKING: + from quart import Quart + +logger = logging.getLogger(__name__) + + +async def prune_syslog_lines(session, *, retention_days: int, max_bytes_per_source: int, + now: datetime | None = None) -> dict: + """Drop lines past the age window, then keep only the newest ~cap bytes + per source. Only sources at or over the cap are ranked. Runs inside the + caller's open transaction.""" + from .models import SyslogLine + + now = now or datetime.now(timezone.utc) + res = await session.execute( + delete(SyslogLine).where(SyslogLine.received_at < now - timedelta(days=retention_days))) + age_pruned = res.rowcount or 0 + + size = func.length(SyslogLine.message) + over = (select(SyslogLine.source_id).group_by(SyslogLine.source_id) + .having(func.sum(size) >= max_bytes_per_source).subquery()) + # Newest first, a line goes once the lines strictly newer than it already + # fill the cap: the newest line always stays, however long. + running = func.sum(size).over(partition_by=SyslogLine.source_id, + order_by=[SyslogLine.received_at.desc(), SyslogLine.id.desc()]) + ranked = (select(SyslogLine.id, (running - size).label("prefix_excl")) + .join(over, SyslogLine.source_id == over.c.source_id).subquery()) + res = await session.execute(delete(SyslogLine).where(SyslogLine.id.in_( + select(ranked.c.id).where(ranked.c.prefix_excl >= max_bytes_per_source)))) + return {"age_pruned": age_pruned, "size_pruned": res.rowcount or 0} + + +def make_prune_task(app: "Quart") -> ScheduledTask: + async def prune() -> None: + cfg = (app.config.get("PLUGINS") or {}).get("syslog") or {} + async with app.db_sessionmaker() as session: + async with session.begin(): + counts = await prune_syslog_lines( + session, retention_days=int(cfg.get("retention_days", 14)), + max_bytes_per_source=int(cfg.get("max_bytes_per_source", 50_000_000))) + if any(counts.values()): + logger.info("Syslog retention: %s", counts) + + return ScheduledTask(name="syslog_prune", coro_factory=prune, interval_seconds=3600) diff --git a/plugins/syslog/routes.py b/plugins/syslog/routes.py new file mode 100644 index 0000000..360e3cd --- /dev/null +++ b/plugins/syslog/routes.py @@ -0,0 +1,182 @@ +# plugins/syslog/routes.py +"""The Syslog page: whether the listener is open, the sources whose lines are +kept, and the senders offered to be added. Every GET reads Steward's own rows +and the listener's state in memory; nothing here contacts a device (System +178's ruling). Adding, ignoring, editing and removing a source are an admin's, +and audited.""" +from __future__ import annotations + +from datetime import datetime, timezone + +from quart import Blueprint, abort, current_app, redirect, render_template, request, session +from sqlalchemy import select + +from steward.auth.middleware import require_role +from steward.core.audit import log_audit +from steward.models.users import UserRole + +from .models import SyslogSource +from .parse import severity_name +from .receiver import LISTENER, MAX_SENDERS, settings + +syslog_bp = Blueprint("syslog", __name__, template_folder="templates") + +INDEX = "/plugins/syslog/" + + +def _actor() -> tuple[str | None, str]: + return session.get("user_id"), session.get("username", "") + + +async def _hosts(db) -> list: + from steward.models.hosts import Host + return list((await db.execute(select(Host.id, Host.name, Host.address).order_by(Host.name))).all()) + + +def listener_view(port: int, udp: bool, tcp: bool) -> dict: + """What the page says about the listener: open on which protocols, or + why not.""" + protocols = [p for p, on in (("UDP", udp), ("TCP", tcp)) if on] + return {"port": port, "protocols": protocols, "listening": LISTENER.listening, + "error": LISTENER.error, "bound_at": LISTENER.bound_at, + "received": LISTENER.received, "dropped": LISTENER.dropped, + "unoffered": LISTENER.unoffered, "connections": LISTENER.connections, + "senders_full": False, "max_senders": MAX_SENDERS} + + +def source_view(s: SyslogSource, hosts: dict) -> dict: + host = hosts.get(s.host_id) if s.host_id else None + return {"id": s.id, "address": s.address, "state": s.state, + "name": s.name or s.hostname_seen or s.address, + "hostname_seen": s.hostname_seen, "host": host, + "first_seen_at": s.first_seen_at, "last_seen_at": s.last_seen_at, + "lines_total": s.lines_total or 0, + "severity": severity_name(s.last_severity) if s.last_severity is not None else None, + "level": s.last_severity, "last_line": s.last_line} + + +def suggested_host(source: SyslogSource, hosts: list) -> str | None: + """The host whose address is the sender's, if one is.""" + return next((h.id for h in hosts if h.address == source.address), None) + + +@syslog_bp.get("/") +@require_role(UserRole.viewer) +async def index(): + async with current_app.db_sessionmaker() as db: + rows = list((await db.execute( + select(SyslogSource).order_by(SyslogSource.last_seen_at.desc()))).scalars()) + hosts = await _hosts(db) + by_id = {h.id: h for h in hosts} + views = {state: [] for state in ("accepted", "offered", "ignored")} + for row in rows: + v = source_view(row, by_id) + if row.state == "offered": + v["suggested_host"] = suggested_host(row, hosts) + views.setdefault(row.state, []).append(v) + views["accepted"].sort(key=lambda v: v["name"].lower()) + listener = listener_view(*settings(current_app)) + listener["senders_full"] = len(rows) >= MAX_SENDERS + return await render_template("syslog/index.html", listener=listener, + sources=views["accepted"], offered=views["offered"], + ignored=views["ignored"], hosts=hosts, + now=datetime.now(timezone.utc)) + + +async def _source_or_404(db, source_id: str) -> SyslogSource: + source = await db.get(SyslogSource, source_id) + if source is None: + abort(404) + return source + + +async def _host_id(db, value: str | None) -> str | None: + from steward.models.hosts import Host + if not value: + return None + return value if await db.get(Host, value) is not None else None + + +@syslog_bp.post("/sources//add") +@require_role(UserRole.admin) +async def add_source(source_id: str): + """Keep an offered sender's lines from now on, under the name given.""" + form = await request.form + async with current_app.db_sessionmaker() as db: + async with db.begin(): + source = await _source_or_404(db, source_id) + source.name = (form.get("name") or "").strip()[:100] or source.hostname_seen or source.address + source.host_id = await _host_id(db, form.get("host_id")) + source.state = "accepted" + detail = {"name": source.name, "address": source.address} + await log_audit(current_app, *_actor(), "syslog.source_added", entity_type="syslog_source", + entity_id=source_id, detail=detail) + return redirect(INDEX) + + +@syslog_bp.post("/sources//ignore") +@require_role(UserRole.admin) +async def ignore_source(source_id: str): + """Stop offering a sender: it is still counted, never kept.""" + return await _set_state(source_id, "ignored", "syslog.source_ignored") + + +@syslog_bp.post("/sources//offer") +@require_role(UserRole.admin) +async def offer_source(source_id: str): + """Offer an ignored sender again.""" + return await _set_state(source_id, "offered", "syslog.source_offered") + + +async def _set_state(source_id: str, state: str, action: str): + async with current_app.db_sessionmaker() as db: + async with db.begin(): + source = await _source_or_404(db, source_id) + source.state = state + detail = {"address": source.address} + await log_audit(current_app, *_actor(), action, entity_type="syslog_source", + entity_id=source_id, detail=detail) + return redirect(INDEX) + + +@syslog_bp.get("/sources//edit") +@require_role(UserRole.admin) +async def edit_source(source_id: str): + async with current_app.db_sessionmaker() as db: + source = await _source_or_404(db, source_id) + hosts = await _hosts(db) + return await render_template("syslog/edit.html", source=source, hosts=hosts, + label=source.name or source.hostname_seen or source.address) + + +@syslog_bp.post("/sources//edit") +@require_role(UserRole.admin) +async def save_source(source_id: str): + form = await request.form + async with current_app.db_sessionmaker() as db: + async with db.begin(): + source = await _source_or_404(db, source_id) + old_name = source.name + source.name = (form.get("name") or "").strip()[:100] or source.hostname_seen or source.address + source.host_id = await _host_id(db, form.get("host_id")) + detail = {"name": source.name} + if old_name != source.name: + detail["old_name"] = old_name + await log_audit(current_app, *_actor(), "syslog.source_updated", entity_type="syslog_source", + entity_id=source_id, detail=detail) + return redirect(INDEX) + + +@syslog_bp.post("/sources//delete") +@require_role(UserRole.admin) +async def delete_source(source_id: str): + """Remove a source and the lines kept from it. A device still sending is + offered again.""" + async with current_app.db_sessionmaker() as db: + async with db.begin(): + source = await _source_or_404(db, source_id) + detail = {"name": source.name, "address": source.address} + await db.delete(source) + await log_audit(current_app, *_actor(), "syslog.source_removed", entity_type="syslog_source", + entity_id=source_id, detail=detail) + return redirect(INDEX) diff --git a/plugins/syslog/templates/syslog/edit.html b/plugins/syslog/templates/syslog/edit.html new file mode 100644 index 0000000..2621683 --- /dev/null +++ b/plugins/syslog/templates/syslog/edit.html @@ -0,0 +1,48 @@ +{# plugins/syslog/templates/syslog/edit.html — a source's name and host, and + Remove, which deletes the lines kept from it (#5999). #} +{% extends "base.html" %} +{% from "_macros.html" import crumbs, page_header %} +{% block title %}Edit {{ label }} — Steward{% endblock %} +{% block breadcrumb %}{{ crumbs([("Syslog", "/plugins/syslog/"), (label, ""), ("Edit", "")]) }}{% endblock %} +{% block content %} +{{ page_header("Edit " ~ label) }} + +
+
{{ csrf_field() }} +
+

Source

+
+
+ + +
+
+ Address + {{ source.address }} + Lines are matched to a source by the address they come from. +
+
+ + + Which host's page shows this source's lines. +
+
+
+
+ + Cancel +
+
+ +
+

Remove

+

Deletes the source and every line kept from it. If the device is still sending, it is offered again.

+
{{ csrf_field() }} + +
+
+
+{% endblock %} diff --git a/plugins/syslog/templates/syslog/index.html b/plugins/syslog/templates/syslog/index.html new file mode 100644 index 0000000..0e8e4ce --- /dev/null +++ b/plugins/syslog/templates/syslog/index.html @@ -0,0 +1,114 @@ +{# plugins/syslog/templates/syslog/index.html — whether Steward is listening, + the sources whose lines are kept, and the devices sending that haven't been + added: counted, never stored, until an admin adds one (#5999). #} +{% extends "base.html" %} +{% from "_macros.html" import page_header, status_badge, empty_state %} +{% block title %}Syslog — Steward{% endblock %} +{% macro tone(level) -%}{% if level is not none and level <= 3 %}tone-crit{% elif level == 4 %}tone-warn{% endif %}{%- endmacro %} +{% block content %} +{% set admin = session.user_role == 'admin' %} +{{ page_header("Syslog", (sources|length ~ " source" ~ ("s" if sources|length != 1 else "")) if sources else none) }} + +
+
+ {% if listener.error %}{{ status_badge('down', 'Not listening') }} + {% elif listener.listening %}{{ status_badge('up', 'Listening') }} + {% elif not listener.protocols %}{{ status_badge('inactive', 'Off') }} + {% else %}{{ status_badge('pending', 'Starting') }}{% endif %} +

Listener

+ {% if listener.protocols %}{{ listener.protocols|join(" and ") }} port {{ listener.port }}{% endif %} + {% if admin %}Settings{% endif %} +
+ {% if listener.error %} +

{{ listener.error }}

+

Tried again every minute. Another program may hold the port, or it is below 1024, which Steward can't open: choose another in Settings.

+ {% elif not listener.protocols %} +

UDP and TCP are both off in Settings, so nothing is received.

+ {% else %} +

+ Point a device's remote syslog at this server, {{ listener.protocols|join(" or ") }} port {{ listener.port }}. + {%- if listener.bound_at %} Open since {{ listener.bound_at|ago }}: {{ "{:,}".format(listener.received) }} line{{ "s" if listener.received != 1 }} received{% if listener.connections %}, {{ listener.connections }} TCP sender{{ "s" if listener.connections != 1 }} connected{% endif %}.{% endif %} +

+ {% endif %} + {% if listener.dropped %}

{{ "{:,}".format(listener.dropped) }} line{{ "s" if listener.dropped != 1 }} lost: they arrived faster than they could be read.

{% endif %} + {% if listener.senders_full %}

Steward knows {{ listener.max_senders }} senders, the most it keeps; a new one is counted, not offered{% if listener.unoffered %} ({{ "{:,}".format(listener.unoffered) }} line{{ "s" if listener.unoffered != 1 }} so far){% endif %}. Remove ones you don't need.

{% endif %} +
+ +{% if offered %} +
+

Sending, not added

+

These devices are sending syslog. Their lines are counted and not kept until you add them.

+ + {% if admin %}{% endif %} + + {% for s in offered %} + + + + + + {% if admin %} + + {% endif %} + + {% endfor %} + +
SenderLinesLast heardNewest line
{{ s.address }}{% if s.hostname_seen %}
calls itself {{ s.hostname_seen }}
{% endif %}		{{ "{:,}".format(s.lines_total) }}{{ s.last_seen_at|ago }}{{ s.severity or "" }} {{ s.last_line or "—" }} +
{{ csrf_field() }} + + + +
+
{{ csrf_field() }} + +
+
+
+{% endif %} + +
+

Sources

+ {% if not sources %} + {{ empty_state("No sources yet. Point a NAS, router or other device at this server's syslog port: it shows above as sending, and Add keeps its lines.") }} + {% else %} + + {% if admin %}{% endif %} + + {% for s in sources %} + + + + + + + {% if admin %}{% endif %} + + {% endfor %} + +
SourceHostLinesLast heardNewest line
{{ s.name }}
{{ s.address }}
{% if s.host %}{{ s.host.name }}{% else %}—{% endif %}{{ "{:,}".format(s.lines_total) }}{{ s.last_seen_at|ago }}{{ s.severity or "" }} {{ s.last_line or "—" }}Edit
+

Lines counts every line since the device was first heard, added or not.

+ {% endif %} +
+ +{% if ignored %} +
+ Ignored ({{ ignored|length }}) + + {% if admin %}{% endif %} + + {% for s in ignored %} + + + + + {% if admin %}{% endif %} + + {% endfor %} + +
SenderLinesLast heard
{{ s.address }}{% if s.hostname_seen %}
calls itself {{ s.hostname_seen }}
{% endif %}		{{ "{:,}".format(s.lines_total) }}{{ s.last_seen_at|ago }}
{{ csrf_field() }}
+
+{% endif %} +{% endblock %} diff --git a/steward/core/settings.py b/steward/core/settings.py index c89bff5..165d560 100644 --- a/steward/core/settings.py +++ b/steward/core/settings.py @@ -162,6 +162,7 @@ DEFAULTS: dict[str, Any] = { "plugin.host_agent": {"enabled": True}, "plugin.snmp": {"enabled": True}, "plugin.prometheus": {"enabled": True}, + "plugin.syslog": {"enabled": True}, # OIDC single-sign-on "oidc.enabled": False, "oidc.discovery_url": "", diff --git a/steward/settings/routes.py b/steward/settings/routes.py index 221a8be..9aae29f 100644 --- a/steward/settings/routes.py +++ b/steward/settings/routes.py @@ -975,7 +975,7 @@ def _build_plugin_cfg_from_form(plugin: dict, form) -> dict: # of a host), not discrete vendor integrations. Presentation-only split — they # still load through the normal plugin mechanism. A plugin.yaml may set # kind: capability|integration to override. -CAPABILITY_PLUGINS = {"host_agent", "snmp", "docker"} +CAPABILITY_PLUGINS = {"host_agent", "snmp", "docker", "syslog"} @settings_bp.get("/plugins/") diff --git a/tests/integration/test_syslog.py b/tests/integration/test_syslog.py new file mode 100644 index 0000000..95ebe7d --- /dev/null +++ b/tests/integration/test_syslog.py @@ -0,0 +1,100 @@ +"""Integration: syslog lines stored against Postgres (#5999). A new sender is +offered and its lines counted, not kept; an added source's lines are kept; +the ring keeps each source within its age window and byte cap.""" +from __future__ import annotations + +import asyncio +import os +from datetime import datetime, timedelta, timezone + +import pytest + +pytestmark = pytest.mark.integration + + +@pytest.fixture +def app(): + if not os.environ.get("STEWARD_DATABASE_URL"): + pytest.skip("needs Postgres") + from steward.app import create_app + return create_app(testing=False) + + +def _received(address, text, at): + from plugins.syslog.parse import parse + from plugins.syslog.receiver import Received + return Received(at, address, parse(f"<11>Oct 19 01:12:03 synnas kernel: {text}")) + + +def test_an_unknown_sender_is_offered_and_an_added_one_kept(app): + from sqlalchemy import delete, func, select + + from plugins.syslog.models import SyslogLine, SyslogSource + from plugins.syslog.receiver import store_lines + + now = datetime(2026, 10, 19, 3, 0, tzinfo=timezone.utc) + + async def go(): + async with app.db_sessionmaker() as s: + async with s.begin(): + await s.execute(delete(SyslogSource)) + await store_lines(s, [_received("10.9.0.1", "first", now), + _received("10.9.0.1", "second", now + timedelta(seconds=1))]) + async with app.db_sessionmaker() as s: + src = (await s.execute(select(SyslogSource))).scalar_one() + assert (src.state, src.lines_total, src.hostname_seen) == ("offered", 2, "synnas") + assert src.last_line == "second" and src.last_severity == 3 + assert (await s.execute(select(func.count()).select_from(SyslogLine))).scalar_one() == 0 + src_id = src.id + async with app.db_sessionmaker() as s: + async with s.begin(): + (await s.get(SyslogSource, src_id)).state = "accepted" + async with app.db_sessionmaker() as s: + async with s.begin(): + await store_lines(s, [_received("10.9.0.1", "kept", now + timedelta(seconds=2))]) + async with app.db_sessionmaker() as s: + lines = (await s.execute(select(SyslogLine))).scalars().all() + assert [(l.app, l.hostname, l.message, l.severity) for l in lines] == [("kernel", "synnas", "kept", 3)] + assert (await s.get(SyslogSource, src_id)).lines_total == 3 + async with s.begin(): + await s.execute(delete(SyslogSource)) + + asyncio.run(go()) + + +def test_the_ring_keeps_each_source_within_its_window_and_cap(app): + from sqlalchemy import delete, select + + from plugins.syslog.models import SyslogLine, SyslogSource + from plugins.syslog.retention import prune_syslog_lines + + now = datetime(2026, 10, 19, 3, 0, tzinfo=timezone.utc) + line40 = "x" * 40 # cap 100 keeps 3 (prefixes 0, 40, 80) + + async def go(): + async with app.db_sessionmaker() as s: + async with s.begin(): + await s.execute(delete(SyslogSource)) + a = SyslogSource(address="10.9.0.2", state="accepted", first_seen_at=now, last_seen_at=now) + b = SyslogSource(address="10.9.0.3", state="accepted", first_seen_at=now, last_seen_at=now) + s.add_all([a, b]) + await s.flush() + + def line(src, minutes_ago, text): + return SyslogLine(source_id=src.id, received_at=now - timedelta(minutes=minutes_ago), + facility=1, severity=6, message=text) + + s.add(line(a, 60 * 24 * 30, "old")) + s.add_all([line(a, 6 - i, line40) for i in range(1, 6)]) + s.add_all([line(b, 2, line40), line(b, 1, line40)]) + a_id, b_id = a.id, b.id + async with app.db_sessionmaker() as s: + async with s.begin(): + counts = await prune_syslog_lines(s, retention_days=14, max_bytes_per_source=100, now=now) + kept = (await s.execute(select(SyslogLine.source_id))).scalars().all() + assert counts == {"age_pruned": 1, "size_pruned": 2} + assert kept.count(a_id) == 3 and kept.count(b_id) == 2 + async with s.begin(): + await s.execute(delete(SyslogSource)) + + asyncio.run(go()) diff --git a/tests/plugins/syslog/__init__.py b/tests/plugins/syslog/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/plugins/syslog/test_parse.py b/tests/plugins/syslog/test_parse.py new file mode 100644 index 0000000..e4d9749 --- /dev/null +++ b/tests/plugins/syslog/test_parse.py @@ -0,0 +1,61 @@ +"""Syslog messages read into their parts (#5999): RFC 5424, RFC 3164 with and +without a hostname, a message with no PRI, and TCP framing (RFC 6587).""" +from datetime import datetime, timezone + +from plugins.syslog.parse import MAX_MESSAGE, facility_name, frames, parse, severity_name + + +def test_rfc5424_with_structured_data_and_a_zone(): + m = parse(b'<165>1 2026-10-19T01:12:03.5+00:00 synnas kernel 1234 ID47 ' + b'[exampleSDID@32473 iut="3" eventSource="App\\"lication"] nfsd: server not responding') + assert (facility_name(m.facility), severity_name(m.severity)) == ("local4", "notice") + assert m.sent_at == datetime(2026, 10, 19, 1, 12, 3, 500000, tzinfo=timezone.utc) + assert (m.hostname, m.app, m.text) == ("synnas", "kernel", "nfsd: server not responding") + + +def test_rfc5424_nil_fields_and_no_zone_keep_no_sent_time(): + m = parse("<14>1 2026-10-19T01:12:03 - - - - - plain text") + assert m.sent_at is None and (m.hostname, m.app, m.text) == ("", "", "plain text") + + +def test_bsd_with_hostname_and_tag(): + m = parse(b"<38>Oct 9 22:14:15 adm-nas sshd[4121]: Accepted publickey for admin") + assert (m.facility, m.severity) == (4, 6) + assert m.sent_at is None # no year, no zone: never trusted + assert (m.hostname, m.app, m.text) == ("adm-nas", "sshd", "Accepted publickey for admin") + + +def test_bsd_without_hostname_reads_the_tag(): + m = parse("<30>Oct 19 01:12:03 rsyncd[881]: connect from 192.168.0.4") + assert (m.hostname, m.app, m.text) == ("", "rsyncd", "connect from 192.168.0.4") + + +def test_no_pri_is_user_notice_and_an_unknown_shape_is_kept_whole(): + m = parse(b"something the device made up\n") + assert (m.facility, m.severity) == (1, 5) + assert m.text == "something the device made up" + assert parse(b"<999>not a pri").text == "<999>not a pri" + + +def test_a_long_message_is_cut_and_bad_bytes_kept_readable(): + m = parse(b"<13>" + b"x" * (MAX_MESSAGE + 50)) + assert len(m.text) == MAX_MESSAGE + assert "�" in parse(b"<13>caf\xe9").text + + +def test_frames_octet_counted_and_newline_ended_with_a_partial_left(): + buf = bytearray(b"11 <13>hello 1<14>second line\n<15>third") + assert frames(buf) == [b"<13>hello 1", b"<14>second line"] + assert buf == bytearray(b"<15>third") + buf += b" part\n" + assert frames(buf) == [b"<15>third part"] and not buf + + +def test_frames_waits_for_the_rest_of_a_counted_message(): + buf = bytearray(b"20 <13>only part") + assert frames(buf) == [] and buf == bytearray(b"20 <13>only part") + + +def test_a_count_too_large_is_read_as_text(): + buf = bytearray(b"9999999 is not a count\n") + assert frames(buf) == [b"9999999 is not a count"] diff --git a/tests/plugins/syslog/test_receiver.py b/tests/plugins/syslog/test_receiver.py new file mode 100644 index 0000000..9eb3f23 --- /dev/null +++ b/tests/plugins/syslog/test_receiver.py @@ -0,0 +1,96 @@ +"""The syslog listener (#5999): lines queued as they arrive, bounded, read in +rounds; the listener bound as the settings say, and a port it can't open +said, not raised.""" +import asyncio +import socket +from datetime import datetime, timedelta, timezone +from types import SimpleNamespace + +import pytest + +from plugins.syslog import receiver + + +@pytest.fixture(autouse=True) +def fresh(monkeypatch): + monkeypatch.setattr(receiver, "LISTENER", receiver.Listener()) + monkeypatch.setattr(receiver, "QUEUE", receiver.deque()) + + +def test_the_queue_is_bounded_and_says_what_it_lost(monkeypatch): + monkeypatch.setattr(receiver, "MAX_QUEUED", 3) + for i in range(5): + receiver.take("10.0.0.5", f"<13>line {i}".encode()) + assert receiver.LISTENER.received == 5 and receiver.LISTENER.dropped == 2 + assert [r.message.text for r in receiver.read_round()] == ["line 2", "line 3", "line 4"] + + +def test_a_round_takes_at_most_its_share_and_leaves_the_rest(): + for i in range(5): + receiver.take("10.0.0.5", f"<13>line {i}".encode()) + first = receiver.read_round(limit=2) + assert [r.message.text for r in first] == ["line 0", "line 1"] and len(receiver.QUEUE) == 3 + assert first[0].address == "10.0.0.5" + + +def _app(**cfg): + return SimpleNamespace(config={"PLUGINS": {"syslog": cfg}}) + + +def _free_port() -> int: + with socket.socket() as s: + s.bind(("0.0.0.0", 0)) + return s.getsockname()[1] + + +def test_it_listens_on_udp_and_tcp_and_takes_what_arrives(): + async def go(): + port = _free_port() + await receiver.ensure_listening(_app(port=port, udp=True, tcp=True)) + assert receiver.LISTENER.listening and receiver.LISTENER.error is None + udp = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + udp.sendto(b"<13>over udp", ("127.0.0.1", port)) + udp.close() + reader, writer = await asyncio.wait_for(asyncio.open_connection("127.0.0.1", port), 5) + writer.write(b"12 <13>over tcp<14>and a line\n") + await writer.drain() + writer.close() + for _ in range(50): + if len(receiver.QUEUE) >= 3: + break + await asyncio.sleep(0.05) + texts = sorted(r.message.text for r in receiver.read_round()) + await receiver.close() + return texts + + assert asyncio.run(go()) == ["and a line", "over tcp", "over udp"] + + +def test_a_port_it_cannot_open_is_said_and_tried_again_later(): + async def go(): + with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as held: + held.bind(("0.0.0.0", 0)) + port = held.getsockname()[1] + now = datetime(2026, 10, 19, 3, 0, tzinfo=timezone.utc) + await receiver.ensure_listening(_app(port=port, udp=True, tcp=False), now) + said = receiver.LISTENER.error + tried = receiver.LISTENER.tried_at + await receiver.ensure_listening(_app(port=port, udp=True, tcp=False), now + timedelta(seconds=10)) + assert receiver.LISTENER.tried_at == tried # not retried within the minute + return said + + said = asyncio.run(go()) + assert said and said.startswith("Port ") and "UDP" in said + + +def test_a_change_of_port_moves_the_listener(): + async def go(): + a, b = _free_port(), _free_port() + await receiver.ensure_listening(_app(port=a, udp=True, tcp=False)) + await receiver.ensure_listening(_app(port=b, udp=True, tcp=False)) + port = receiver.LISTENER._udp_transport.get_extra_info("sockname")[1] + await receiver.close() + return port, b + + port, b = asyncio.run(go()) + assert port == b diff --git a/tests/plugins/syslog/test_syslog_page.py b/tests/plugins/syslog/test_syslog_page.py new file mode 100644 index 0000000..71df839 --- /dev/null +++ b/tests/plugins/syslog/test_syslog_page.py @@ -0,0 +1,73 @@ +"""The Syslog page (#5999): whether the listener is open or why not, the +sources kept, and the senders offered to an admin to add or ignore.""" +from datetime import datetime, timedelta, timezone +from types import SimpleNamespace as NS + +from tests._templates import render_plugin_template +from tests.test_style_tokens import hex_colours + + +def _listener(**over): + base = {"port": 5514, "protocols": ["UDP", "TCP"], "listening": True, "error": None, + "bound_at": datetime.now(timezone.utc) - timedelta(hours=2), "received": 1204, + "dropped": 0, "unoffered": 0, "connections": 1, "senders_full": False, "max_senders": 200} + return {**base, **over} + + +def _source(name, state, level=6, line="nfsd: server not responding", host=None): + now = datetime.now(timezone.utc) + return {"id": f"id-{name}", "address": "192.168.0.7", "state": state, "name": name, + "hostname_seen": "synnas", "host": host, "first_seen_at": now - timedelta(days=1), + "last_seen_at": now - timedelta(minutes=3), "lines_total": 1204, + "severity": "err" if level == 3 else "info", "level": level, "last_line": line, + "suggested_host": "h1"} + + +async def _render(app, role, **ctx): + from quart import session + ctx = {"listener": _listener(), "sources": [], "offered": [], "ignored": [], + "hosts": [NS(id="h1", name="SynNAS", address="192.168.0.7")], + "now": datetime.now(timezone.utc), **ctx} + async with app.test_request_context("/plugins/syslog/"): + session["user_role"] = role + page = await render_plugin_template(app, "syslog", "syslog/index.html", **ctx) + return " ".join(page.split()) + + +async def test_an_offered_sender_can_be_added_or_ignored_by_an_admin(app): + page = await _render(app, "admin", offered=[_source("synnas", "offered", level=3)]) + assert "Sending, not added" in page and "calls itself synnas" in page + assert 'action="/plugins/syslog/sources/id-synnas/add"' in page + assert 'action="/plugins/syslog/sources/id-synnas/ignore"' in page + assert '' in page # the host at its address + assert 'class="tone-crit">err nfsd: server not responding' in page + assert "UDP and TCP port 5514" in page and "1,204 lines received" in page + main = page[page.index("Syslog' in page + assert 'action="/plugins/syslog/sources/s1/delete"' in page and "data-confirm=" in page