admin / Syanpse-Vanguard

"""Synapse-Vanguard Linux agent.

Collectors (each on its own thread, appending to the shared disk buffer):
  • journald  — `journalctl -o json -f` follow stream
  • syslog    — tail of traditional syslog files
  • auditd    — tail of /var/log/audit/audit.log

Package as a standalone executable with:
  pyinstaller --onefile --name sv-agent-linux \
      --collect-submodules agents agents/linux/sv_agent_linux.py
"""
from __future__ import annotations

import json
import logging
import subprocess
import threading
import time
from datetime import datetime, timezone

from agents.common.buffer import DiskBuffer
from agents.common.config import AgentConfig
from agents.common.sender import Sender

log = logging.getLogger("sv.agent.linux")


def _emit(buffer: DiskBuffer, source_type: str, host: str, message: str,
          fields: dict, ts: datetime | None = None) -> None:
    buffer.append({
        "source_type": source_type,
        "host": host,
        "ts": (ts or datetime.now(timezone.utc)).isoformat(),
        "message": message,
        "fields": fields,
    })


def collect_journald(cfg: AgentConfig, buffer: DiskBuffer) -> None:
    cmd = ["journalctl", "-o", "json", "-f", "--no-pager"]
    while True:
        try:
            proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, text=True)
            for line in proc.stdout:                       # type: ignore[union-attr]
                line = line.strip()
                if not line:
                    continue
                try:
                    rec = json.loads(line)
                except json.JSONDecodeError:
                    continue
                ts = _journald_ts(rec.get("__REALTIME_TIMESTAMP"))
                _emit(buffer, "journald", cfg.hostname,
                      rec.get("MESSAGE", ""), rec, ts)
        except FileNotFoundError:
            log.warning("journalctl not found; journald collector disabled")
            return
        except Exception as exc:                            # restart the follow on error
            log.warning("journald collector error, retrying: %s", exc)
            time.sleep(3)


def tail_file(cfg: AgentConfig, buffer: DiskBuffer, path: str, source_type: str) -> None:
    """Simple tail -F: follow appends, reopen on truncation/rotation."""
    while True:
        try:
            with open(path, "r", encoding="utf-8", errors="replace") as fh:
                fh.seek(0, 2)                               # start at EOF
                while True:
                    line = fh.readline()
                    if not line:
                        time.sleep(0.5)
                        # detect rotation: file shrank
                        if fh.tell() > _size(path):
                            break
                        continue
                    _emit(buffer, source_type, cfg.hostname, line.rstrip("\n"),
                          {"path": path})
        except FileNotFoundError:
            time.sleep(5)
        except Exception as exc:
            log.warning("tail %s error: %s", path, exc)
            time.sleep(3)


def main() -> None:
    logging.basicConfig(level=logging.INFO,
                        format="%(asctime)s %(name)s %(levelname)s %(message)s")
    cfg = AgentConfig.load()
    cfg.platform = "linux"
    if not cfg.channels:
        cfg.channels = ["journald", "syslog", "auditd"]

    buffer = DiskBuffer(cfg.buffer_path)
    threads = [threading.Thread(target=collect_journald, args=(cfg, buffer), daemon=True)]
    for p in cfg.syslog_paths:
        threads.append(threading.Thread(target=tail_file,
                                        args=(cfg, buffer, p, "syslog"), daemon=True))
    threads.append(threading.Thread(target=tail_file,
                                    args=(cfg, buffer, cfg.auditd_path, "auditd"),
                                    daemon=True))
    for t in threads:
        t.start()

    Sender(cfg, buffer).run_forever()


def _journald_ts(us) -> datetime | None:
    try:
        return datetime.fromtimestamp(int(us) / 1_000_000, tz=timezone.utc)
    except (TypeError, ValueError):
        return None


def _size(path: str) -> int:
    import os
    try:
        return os.path.getsize(path)
    except OSError:
        return 0


if __name__ == "__main__":
    main()