"""sshpull-collector: holt Logs per SSH von den Hosts und schreibt sie nach /remote-logs//.log, wo der Wazuh-Manager sie per liest. Quellen pro Host (hosts.yml): journal journalctl -o json -f; Fortsetzung nach Verbindungsabbruch per Cursor file tail -F auf eine Datei; Fortsetzung per Inode + Byte-Offset Optional: agent_tunnel ssh -R 1514/1515 -> Wazuh-Agent auf dem Host meldet sich an 127.0.0.1 Unterbefehle: run (Standard) Dauerbetrieb keyscan Host-Keys holen und nach /ssh/known_hosts schreiben test Verbindung + Rechte auf allen Hosts pruefen """ import asyncio import json import logging import os import re import shlex import signal import sys import time sys.path.insert(0, "/app") import sshpull_hosts # noqa: E402 HOSTS_FILE = os.environ.get("HOSTS_FILE", "/sshpull/hosts.yml") SSH_DIR = "/ssh" KNOWN_HOSTS = f"{SSH_DIR}/known_hosts" SSH_CONFIG = "/run/sshpull_ssh_config" LOG_DIR = os.environ.get("LOG_DIR", "/remote-logs") STATE_DIR = os.environ.get("STATE_DIR", "/state") MAX_LOG_BYTES = int(os.environ.get("MAX_LOG_MB", "50")) * 1024 * 1024 MANAGER = os.environ.get("MANAGER_HOST", "wazuh.manager") MONTHS = "Jan Feb Mar Apr May Jun Jul Aug Sep Oct Nov Dec".split() CURSOR_RE = re.compile(r"^[A-Za-z0-9=;]+$") log = logging.getLogger("sshpull") def syslog_ts(epoch): t = time.localtime(epoch) return f"{MONTHS[t.tm_mon - 1]} {t.tm_mday:2d} {t.tm_hour:02d}:{t.tm_min:02d}:{t.tm_sec:02d}" def ssh_cmd(host, *extra): return ["ssh", "-F", SSH_CONFIG, *extra, host] class LogWriter: """Haengt Zeilen an eine Datei an und rotiert bei MAX_LOG_BYTES (eine .1-Datei).""" def __init__(self, path): self.path = path os.makedirs(os.path.dirname(path), exist_ok=True) self.fh = open(path, "a", encoding="utf-8", errors="replace") def write(self, line): self.fh.write(line.replace("\n", " ").replace("\r", " ") + "\n") self.fh.flush() if self.fh.tell() > MAX_LOG_BYTES: self.fh.close() os.replace(self.path, self.path + ".1") self.fh = open(self.path, "a", encoding="utf-8", errors="replace") class Status: """Meldet Verbindungswechsel (up/down) als JSON nach /remote-logs/_status/status.log. Nur Zustandswechsel werden geschrieben, damit Reconnect-Schleifen keine Alarmflut erzeugen.""" def __init__(self): self.writer = None self.last = {} def open(self): # Datei sofort anlegen: der Wazuh-Logcollector oeffnet spaeter auftauchende # Dateien am Ende und wuerde die ersten Meldungen verpassen. self.writer = LogWriter(f"{LOG_DIR}/_status/status.log") def set(self, host, check, status, message=""): key = (host, check) if self.last.get(key, "up") == status: # Startzustand gilt als "up" self.last[key] = status return self.last[key] = status if self.writer is None: self.writer = LogWriter(f"{LOG_DIR}/_status/status.log") self.writer.write(json.dumps({"sshpull_status": { "host": host, "check": check, "status": status, "message": message[:500]}})) log.info("%s/%s: Status %s", host, check, status) STATUS = Status() async def watch_up(proc, host, check, delay=15): """Laeuft die Verbindung nach Sekunden noch, gilt sie als aufgebaut.""" await asyncio.sleep(delay) if proc.returncode is None: STATUS.set(host, check, "up") class State: def __init__(self, host, name): self.path = os.path.join(STATE_DIR, host, f"{name}.json") os.makedirs(os.path.dirname(self.path), exist_ok=True) try: with open(self.path, encoding="utf-8") as fh: self.data = json.load(fh) except (OSError, ValueError): self.data = {} self._saved = 0.0 def save(self, force=False): now = time.monotonic() if not force and now - self._saved < 5: return tmp = self.path + ".tmp" with open(tmp, "w", encoding="utf-8") as fh: json.dump(self.data, fh) os.replace(tmp, self.path) self._saved = now class Backoff: def __init__(self): self.delay = 5 async def wait(self, ran_for): if ran_for > 120: self.delay = 5 await asyncio.sleep(self.delay) self.delay = min(self.delay * 2, 300) async def drain_stderr(stream, prefix, on_line=None, last=None): async for raw in stream: line = raw.decode("utf-8", "replace").rstrip() if line: log.warning("%s: %s", prefix, line) if last is not None: last[:] = [line] if on_line: on_line(line) # -------------------------------------------------------------------------- # journal # -------------------------------------------------------------------------- def journal_line(host, entry): msg = entry.get("MESSAGE", "") if isinstance(msg, list): # nicht-UTF-8-Nachrichten kommen als Byte-Array msg = bytes(b & 0xFF for b in msg if isinstance(b, int)).decode("utf-8", "replace") elif msg is None: msg = "" ident = entry.get("SYSLOG_IDENTIFIER") or entry.get("_COMM") or "unknown" pid = entry.get("SYSLOG_PID") or entry.get("_PID") try: ts = int(entry.get("__REALTIME_TIMESTAMP", 0)) / 1e6 except (TypeError, ValueError): ts = time.time() tag = f"{ident}[{pid}]" if pid and ident != "kernel" else ident return f"{syslog_ts(ts)} {host} {tag}: {msg}" def journal_script(src, sudo, cursor): args = ["journalctl", "--no-pager", "-o", "json", "-f"] for unit in src.get("units", []): args += ["-u", unit] if src.get("priority"): args += ["-p", src["priority"]] if cursor: args += [f"--after-cursor={cursor}"] else: args += ["-n", "0"] prefix = "sudo -n " if sudo else "" return f"exec {prefix}{shlex.join(args)}\n" async def run_journal(h, src, stop): name = h["name"] writer = LogWriter(f"{LOG_DIR}/{name}/{src['name']}.log") state = State(name, src["name"]) backoff = Backoff() tag = f"{name}/{src['name']}" while not stop.is_set(): cursor = state.data.get("cursor") if cursor and not CURSOR_RE.match(cursor): cursor = None started = time.monotonic() got_lines = 0 proc = await asyncio.create_subprocess_exec( *ssh_cmd(name), "sh", "-s", stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, limit=1024 * 1024) proc.stdin.write(journal_script(src, h["sudo"], cursor).encode()) await proc.stdin.drain() proc.stdin.close() log.info("%s: verbunden (%s)", tag, "ab Cursor" if cursor else "ab jetzt") last_err = [] err_task = asyncio.create_task(drain_stderr(proc.stderr, tag, last=last_err)) up_task = asyncio.create_task(watch_up(proc, name, src["name"])) stop_task = asyncio.create_task(stop.wait()) try: while True: read = asyncio.create_task(proc.stdout.readline()) done, _ = await asyncio.wait({read, stop_task}, return_when=asyncio.FIRST_COMPLETED) if read not in done: read.cancel() break raw = read.result() if not raw: break try: entry = json.loads(raw) except ValueError: continue writer.write(journal_line(name, entry)) got_lines += 1 if got_lines == 1: STATUS.set(name, src["name"], "up") if entry.get("__CURSOR"): state.data["cursor"] = entry["__CURSOR"] state.save() except ValueError: # Zeile > limit log.error("%s: ueberlange Zeile, Verbindung wird neu aufgebaut", tag) finally: stop_task.cancel() state.save(force=True) if proc.returncode is None: proc.terminate() rc = await proc.wait() await err_task up_task.cancel() if stop.is_set(): break STATUS.set(name, src["name"], "down", last_err[0] if last_err else f"ssh beendet (rc={rc})") ran = time.monotonic() - started # Cursor existiert nicht mehr (Journal rotiert) -> ab jetzt neu beginnen if cursor and got_lines == 0 and ran < 10 and rc not in (0, 255): log.warning("%s: Cursor ungueltig, starte ohne Cursor", tag) state.data.pop("cursor", None) state.save(force=True) log.warning("%s: Verbindung beendet (rc=%s), neuer Versuch", tag, rc) await backoff.wait(ran) # -------------------------------------------------------------------------- # file # -------------------------------------------------------------------------- def file_script(src, sudo, ino, offset): path = shlex.quote(src["path"]) tail = "sudo -n tail" if sudo else "tail" stat = "sudo -n stat" if sudo else "stat" return f"""f={path} ino=$({stat} -c %i "$f" 2>/dev/null); size=$({stat} -c %s "$f" 2>/dev/null) if [ -n "$ino" ] && [ "$ino" = {shlex.quote(str(ino or ''))} ] && [ "$size" -ge {int(offset or 0)} ]; then echo "@@SSHPULL $ino {int(offset or 0)}" exec {tail} -F -c +{int(offset or 0) + 1} "$f" else echo "@@SSHPULL ${{ino:-none}} ${{size:-0}}" exec {tail} -F -n 0 "$f" fi """ async def run_file(h, src, stop): name = h["name"] writer = LogWriter(f"{LOG_DIR}/{name}/{src['name']}.log") state = State(name, src["name"]) backoff = Backoff() tag = f"{name}/{src['name']}" def on_stderr(line): # tail -F meldet Rotation/Truncate -> Offset ab hier nicht mehr eindeutig if "has been replaced" in line or "truncated" in line or "has appeared" in line: state.data["ino"] = None while not stop.is_set(): started = time.monotonic() proc = await asyncio.create_subprocess_exec( *ssh_cmd(name), "sh", "-s", stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, limit=1024 * 1024) proc.stdin.write(file_script(src, h["sudo"], state.data.get("ino"), state.data.get("offset")).encode()) await proc.stdin.drain() proc.stdin.close() last_err = [] err_task = asyncio.create_task(drain_stderr(proc.stderr, tag, on_stderr, last_err)) up_task = asyncio.create_task(watch_up(proc, name, src["name"])) stop_task = asyncio.create_task(stop.wait()) header = True try: while True: read = asyncio.create_task(proc.stdout.readline()) done, _ = await asyncio.wait({read, stop_task}, return_when=asyncio.FIRST_COMPLETED) if read not in done: read.cancel() break raw = read.result() if not raw or not raw.endswith(b"\n"): break # EOF; unvollstaendige Zeile wird beim naechsten Mal neu gelesen if header: header = False parts = raw.decode(errors="replace").split() if len(parts) == 3 and parts[0] == "@@SSHPULL": state.data["ino"] = None if parts[1] == "none" else parts[1] state.data["offset"] = int(parts[2]) if parts[2].isdigit() else 0 log.info("%s: verbunden (inode %s, offset %s)", tag, parts[1], parts[2]) continue writer.write(raw.decode("utf-8", "replace").rstrip("\n")) if state.data.get("ino"): state.data["offset"] = state.data.get("offset", 0) + len(raw) state.save() except ValueError: log.error("%s: ueberlange Zeile, Verbindung wird neu aufgebaut", tag) state.data["ino"] = None finally: stop_task.cancel() state.save(force=True) if proc.returncode is None: proc.terminate() rc = await proc.wait() await err_task up_task.cancel() if stop.is_set(): break STATUS.set(name, src["name"], "down", last_err[0] if last_err else f"ssh beendet (rc={rc})") log.warning("%s: Verbindung beendet (rc=%s), neuer Versuch", tag, rc) await backoff.wait(time.monotonic() - started) # -------------------------------------------------------------------------- # Agent-Tunnel # -------------------------------------------------------------------------- async def run_tunnel(h, stop): name = h["name"] fwd = [] for port in h["tunnel_ports"]: fwd += ["-R", f"127.0.0.1:{port}:{MANAGER}:{port}"] backoff = Backoff() while not stop.is_set(): started = time.monotonic() proc = await asyncio.create_subprocess_exec( *ssh_cmd(name, "-N", "-o", "ExitOnForwardFailure=yes", *fwd), stdin=asyncio.subprocess.DEVNULL, stdout=asyncio.subprocess.DEVNULL, stderr=asyncio.subprocess.PIPE) log.info("%s/tunnel: aufgebaut (Ports %s)", name, h["tunnel_ports"]) last_err = [] err_task = asyncio.create_task(drain_stderr(proc.stderr, f"{name}/tunnel", last=last_err)) up_task = asyncio.create_task(watch_up(proc, name, "tunnel")) wait_task = asyncio.create_task(proc.wait()) stop_task = asyncio.create_task(stop.wait()) await asyncio.wait({wait_task, stop_task}, return_when=asyncio.FIRST_COMPLETED) stop_task.cancel() if proc.returncode is None: proc.terminate() rc = await wait_task await err_task up_task.cancel() if stop.is_set(): break STATUS.set(name, "tunnel", "down", last_err[0] if last_err else f"ssh beendet (rc={rc})") log.warning("%s/tunnel: beendet (rc=%s), neuer Versuch", name, rc) await backoff.wait(time.monotonic() - started) # -------------------------------------------------------------------------- def prepare(cfg): with open(SSH_CONFIG, "w", encoding="utf-8") as fh: fh.write(sshpull_hosts.render_ssh_config(cfg, SSH_DIR, KNOWN_HOSTS)) os.chmod(SSH_CONFIG, 0o600) if not os.path.exists(KNOWN_HOSTS): open(KNOWN_HOSTS, "a").close() async def cmd_run(cfg): stop = asyncio.Event() loop = asyncio.get_running_loop() for sig in (signal.SIGTERM, signal.SIGINT): loop.add_signal_handler(sig, stop.set) STATUS.open() tasks = [] for h in cfg["hosts"]: for src in h["logs"]: fn = run_journal if src["type"] == "journal" else run_file tasks.append(asyncio.create_task(fn(h, src, stop))) if h["agent_tunnel"]: tasks.append(asyncio.create_task(run_tunnel(h, stop))) if not tasks: log.warning("keine Hosts/Quellen konfiguriert - warte") log.info("%d Host(s), %d Stream(s) gestartet", len(cfg["hosts"]), len(tasks)) await stop.wait() await asyncio.gather(*tasks, return_exceptions=True) log.info("beendet") async def cmd_keyscan(cfg): existing = open(KNOWN_HOSTS, encoding="utf-8").read() if os.path.exists(KNOWN_HOSTS) else "" added = [] for h in cfg["hosts"]: key = h["address"] if h["port"] == 22 else f"[{h['address']}]:{h['port']}" if any(line.split(" ", 1)[0].split(",")[0] == key for line in existing.splitlines()): print(f"{h['name']:20} bereits in known_hosts ({key})") continue proc = await asyncio.create_subprocess_exec( "ssh-keyscan", "-T", "10", "-p", str(h["port"]), h["address"], stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.DEVNULL) out, _ = await proc.communicate() lines = [ln for ln in out.decode().splitlines() if ln and not ln.startswith("#")] if not lines: print(f"{h['name']:20} FEHLER: keine Host-Keys von {h['address']}:{h['port']}") continue added += lines fp = await asyncio.create_subprocess_exec( "ssh-keygen", "-lf", "-", stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.PIPE) fps, _ = await fp.communicate("\n".join(lines).encode()) print(f"{h['name']:20} hinzugefuegt:") for ln in fps.decode().splitlines(): print(f"{'':20} {ln}") if added: with open(KNOWN_HOSTS, "a", encoding="utf-8") as fh: fh.write("\n".join(added) + "\n") print("\nBitte die Fingerprints mit den Hosts vergleichen " "(dort: ssh-keygen -lf /etc/ssh/ssh_host_ed25519_key.pub)") async def cmd_test(cfg): ok = True for h in cfg["hosts"]: sudo = "sudo -n " if h["sudo"] else "" checks = ["echo user=$(id -un) groups=$(id -Gn)"] if any(s["type"] == "journal" for s in h["logs"]): checks.append(f"{sudo}journalctl -n 1 -q -o cat >/dev/null 2>&1 && echo journal=ok " f"|| echo journal=FEHLER") checks.append(f"{sudo}journalctl -n 0 2>&1 | grep -qi 'not seeing messages' " f"&& echo journal=WARNUNG:nur-eigene-Logs-sichtbar-Gruppe-systemd-journal-fehlt " f"|| true") for s in h["logs"]: if s["type"] == "file": p = shlex.quote(s["path"]) checks.append(f"{sudo}test -r {p} && echo {s['name']}=ok || echo {s['name']}=FEHLER") if h["agent_tunnel"]: checks.append("(command -v ss >/dev/null && ss -ltn | grep -qE '127.0.0.1:(1514|1515) ') " "&& echo tunnelport=BELEGT || echo tunnelport=frei") proc = await asyncio.create_subprocess_exec( *ssh_cmd(h["name"]), "sh", "-s", stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE) try: out, err = await asyncio.wait_for(proc.communicate("\n".join(checks).encode()), 60) except asyncio.TimeoutError: proc.kill() await proc.wait() out, err = b"", b"Timeout" res = out.decode().split() good = proc.returncode == 0 and not any("FEHLER" in r for r in res) ok &= good print(f"{'OK ' if good else 'ERR'} {h['name']:20} {' '.join(res) or err.decode().strip()}") return 0 if ok else 1 def main(): logging.basicConfig(level=os.environ.get("LOG_LEVEL", "INFO"), format="%(asctime)s %(levelname)-7s %(message)s", stream=sys.stdout) cmd = sys.argv[1] if len(sys.argv) > 1 else "run" try: cfg = sshpull_hosts.load(HOSTS_FILE) except (sshpull_hosts.ConfigError, OSError, ValueError) as exc: log.error("hosts.yml: %s", exc) return 1 prepare(cfg) if cmd == "run": asyncio.run(cmd_run(cfg)) return 0 if cmd == "keyscan": asyncio.run(cmd_keyscan(cfg)) return 0 if cmd == "test": return asyncio.run(cmd_test(cfg)) print(__doc__) return 2 if __name__ == "__main__": sys.exit(main())