Hacker-Software/docker-wazuh
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

first commit

Browse Source
This commit is contained in:
duffyduck
2026-10-07 10:47:30 +02:00
commit 117bb582de
18 changed files with 2060 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
collector/Dockerfile
+13
View File
@@ -0,0 +1,13 @@
FROM python:3.12-slim
RUN apt-get update \
&& apt-get install -y --no-install-recommends openssh-client tzdata \
&& rm -rf /var/lib/apt/lists/* \
&& pip install --no-cache-dir pyyaml==6.0.2
WORKDIR /app
COPY common/sshpull_hosts.py collector/collector.py /app/
ENV PYTHONUNBUFFERED=1
ENTRYPOINT ["python", "/app/collector.py"]
CMD ["run"]
collector/collector.py
+495
View File
@@ -0,0 +1,495 @@
"""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())