496 lines
19 KiB
Python
496 lines
19 KiB
Python
"""sshpull-collector: holt Logs per SSH von den Hosts und schreibt sie nach
|
|
/remote-logs/<host>/<quelle>.log, wo der Wazuh-Manager sie per <localfile> 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 <delay> 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())
|