Meine Dokumentationen

Diese Seite ist nicht editierbar. Sie können den Quelltext sehen, jedoch nicht verändern. Kontaktieren Sie den Administrator, wenn Sie glauben, dass hier ein Fehler vorliegt.
====== Update Script für lokale Server unter Linux ======
===== Vorbereitung =====
Script erstellen (/usr/local/bin/Pruefe-installationen.py)
<code>
#!/usr/bin/env python3
import os
import time
import datetime
import paramiko
import requests
from concurrent.futures import ThreadPoolExecutor, as_completed

# =========================
# Optional: .env / env-file laden (ohne python-dotenv)
# - systemd EnvironmentFile gewinnt (os.environ bereits gesetzt)
# - nur fehlende Variablen werden aus Dateien ergänzt
# =========================
def load_env_file(path: str) -> bool:
    try:
        with open(path, "r") as f:
            for line in f:
                line = line.strip()
                if not line or line.startswith("#") or "=" not in line:
                    continue
                k, v = line.split("=", 1)
                k = k.strip()
                v = v.strip().strip('"').strip("'")
                os.environ.setdefault(k, v)
        return True
    except FileNotFoundError:
        return False

# zuerst systemd-typische Datei, dann lokale .env neben Script
load_env_file("/etc/update-checker.env")
load_env_file("/usr/local/bin/.env")

# =========================
# KONFIGURATION (ENV)
# =========================
TG_BOT_TOKEN = os.getenv("TG_BOT_TOKEN", "").strip()
TG_CHAT_ID = os.getenv("TG_CHAT_ID", "").strip()
TG_THREAD_ID = os.getenv("TG_THREAD_ID", "").strip()
TG_THREAD_ID = int(TG_THREAD_ID) if TG_THREAD_ID else None

SSH_KEY_PATH = os.getenv("SSH_KEY_PATH", "/root/.ssh/id_ed25519")
LOG_FILE = os.getenv("LOG_FILE", "/var/log/update_checker.log")

AUTO_UPDATE = os.getenv("AUTO_UPDATE", "true").lower() in ("1", "true", "yes", "on")
ONLY_REPORT_CHANGES = os.getenv("ONLY_REPORT_CHANGES", "false").lower() in ("1", "true", "yes", "on")
MAX_WORKERS = int(os.getenv("MAX_WORKERS", "5"))

SERVERS = [
    {"host": "172.20.50.52", "user": "root"},
    {"host": "172.20.60.1", "user": "root"},
    {"host": "172.20.70.2", "user": "root"},
    {"host": "172.20.70.3", "user": "root"},
    {"host": "172.20.70.4", "user": "root"},
    {"host": "172.20.70.5", "user": "root"},
    {"host": "172.20.90.1", "user": "root"},
    {"host": "172.20.90.2", "user": "root"},
    {"host": "172.20.90.3", "user": "root"},
    {"host": "172.20.90.4", "user": "root"},
    {"host": "172.20.90.6", "user": "root"},
    {"host": "172.20.90.7", "user": "root"},
    {"host": "172.20.90.8", "user": "root"},
    {"host": "172.20.90.9", "user": "root"},
    {"host": "172.20.90.12", "user": "root"},
    {"host": "172.20.90.13", "user": "root"},
    {"host": "172.20.90.14", "user": "root"},
    {"host": "172.20.90.15", "user": "root"},
    {"host": "172.20.90.17", "user": "root"},
    {"host": "172.20.90.18", "user": "root"},
]

# =========================
# SSH Helper
# =========================
def ssh_run(ssh: paramiko.SSHClient, cmd: str, timeout: int = 1200):
    stdin, stdout, stderr = ssh.exec_command(cmd, timeout=timeout)
    exit_code = stdout.channel.recv_exit_status()
    out = stdout.read().decode(errors="ignore").strip()
    err = stderr.read().decode(errors="ignore").strip()
    return exit_code, out, err

def detect_pkg_manager(ssh):
    # apt bevorzugen, dann dnf, dann yum
    _, out, _ = ssh_run(ssh, "command -v apt-get || true; command -v dnf || true; command -v yum || true", timeout=30)
    joined = "\n".join([l.strip() for l in out.splitlines() if l.strip()])
    if "apt-get" in joined:
        return "apt"
    if "dnf" in joined:
        return "dnf"
    if "yum" in joined:
        return "yum"
    return None

def get_hostname(ssh):
    _, out, _ = ssh_run(ssh, "hostname -f 2>/dev/null || hostname 2>/dev/null || echo unknown", timeout=30)
    return out.strip() if out.strip() else "unknown"

# =========================
# Core check/update
# =========================
def check_and_update(server):
    host = server["host"]
    user = server["user"]
    start = time.time()

    try:
        ssh = paramiko.SSHClient()
        ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        ssh.connect(host, username=user, key_filename=SSH_KEY_PATH, timeout=15)

        hostname = get_hostname(ssh)
        pm = detect_pkg_manager(ssh)
        if not pm:
            ssh.close()
            return {
                "host": host,
                "name": hostname,
                "status": "error",
                "msg": "❌ Paketmanager nicht erkannt",
                "reboot": False,
                "duration": time.time() - start,
            }

        reboot_needed = False
        updates_available = False
        updated = False

        if pm == "apt":
            # Update-Check via Simulation (verlässlicher als 'apt list --upgradable')
            rc, out, err = ssh_run(
                ssh,
                "DEBIAN_FRONTEND=noninteractive apt-get -s upgrade | awk '/^Inst /{c++} END{print c+0}'",
                timeout=120
            )
            if rc != 0:
                raise RuntimeError(err or out or "apt-get -s upgrade fehlgeschlagen")
            count = int(out.strip() or "0")
            updates_available = count > 0

            if updates_available and AUTO_UPDATE:
                rc, out, err = ssh_run(
                    ssh,
                    "DEBIAN_FRONTEND=noninteractive apt-get update && DEBIAN_FRONTEND=noninteractive apt-get -y upgrade",
                    timeout=2400
                )
                if rc != 0:
                    raise RuntimeError(err or "apt upgrade fehlgeschlagen")
                updated = True

                rc, out, _ = ssh_run(ssh, "test -f /var/run/reboot-required && echo yes || echo no", timeout=30)
                reboot_needed = (out.strip() == "yes")

        elif pm in ("dnf", "yum"):
            # check-update Exit 100 => Updates vorhanden, 0 => keine, sonst Fehler
            check_cmd = "dnf -q check-update"
            if pm == "yum":
                check_cmd = "yum -q check-update"

            rc, out, err = ssh_run(ssh, f"{check_cmd} >/dev/null 2>&1; echo $?", timeout=180)
            code_line = out.strip().splitlines()[-1] if out.strip() else "1"
            code = int(code_line)

            if code == 100:
                updates_available = True
            elif code == 0:
                updates_available = False
            else:
                raise RuntimeError(err or f"{pm} check-update Fehler (Exit {code})")

            if updates_available and AUTO_UPDATE:
                up_cmd = "dnf -y upgrade"
                if pm == "yum":
                    up_cmd = "yum -y update"

                rc, out, err = ssh_run(ssh, up_cmd, timeout=3600)
                if rc != 0:
                    raise RuntimeError(err or f"{pm} upgrade fehlgeschlagen")
                updated = True

                rc, out, _ = ssh_run(
                    ssh,
                    "command -v needs-restarting >/dev/null 2>&1 && needs-restarting -r; echo $?",
                    timeout=120
                )
                reboot_needed = ("reboot" in out.lower()) or out.strip().endswith("1")

        ssh.close()

        if updates_available and updated:
            msg = "⬆️ Updates installiert"
            status = "updated"
        elif updates_available and not AUTO_UPDATE:
            msg = "⬆️ Updates verfügbar"
            status = "updates"
        else:
            msg = "✅ Keine Updates"
            status = "ok"

        return {
            "host": host,
            "name": hostname,
            "status": status,
            "msg": msg,
            "reboot": reboot_needed,
            "duration": time.time() - start,
        }

    except Exception as e:
        return {
            "host": host,
            "name": "unknown",
            "status": "error",
            "msg": f"❌ Fehler: {e}",
            "reboot": False,
            "duration": time.time() - start,
        }

# =========================
# Telegram formatting (2 lines per host + runtime in line 1)
# =========================
def format_telegram(results):
    ts = datetime.datetime.now().strftime("%d.%m.%Y %H:%M")
    header = f"*Update-Status* ({ts})\n"

    # Optional: nur Änderungen/Probleme melden
    if ONLY_REPORT_CHANGES:
        results = [r for r in results if r["status"] in ("updated", "updates", "error") or r.get("reboot")]

    if not results:
        return header + "\nAlles aktuell ✅"

    lines = [header]
    for r in sorted(results, key=lambda x: x["host"]):
        host = r["host"]
        name = r["name"]
        dur = f"{r['duration']:.1f}s"

        # Zeile 1: IP + Host + Laufzeit
        lines.append(f"🖥️ `{host}` ({name}) [{dur}]")

        # Zeile 2: Status
        lines.append(f"   {r['msg']}")

        # optional dritte Zeile: reboot
        if r.get("reboot"):
            lines.append("   ⚠️ Neustart erforderlich")

        lines.append("")  # Leerzeile

    return "\n".join(lines).rstrip()

def send_telegram(message: str):
    if not TG_BOT_TOKEN or not TG_CHAT_ID:
        raise RuntimeError("TG_BOT_TOKEN oder TG_CHAT_ID fehlt (ENV prüfen).")

    url = f"https://api.telegram.org/bot{TG_BOT_TOKEN}/sendMessage"
    data = {
        "chat_id": TG_CHAT_ID,
        "text": message,
        "parse_mode": "Markdown",
        "disable_web_page_preview": True,
    }
    if TG_THREAD_ID is not None:
        data["message_thread_id"] = TG_THREAD_ID

    r = requests.post(url, data=data, timeout=20)
    r.raise_for_status()

def log_results(results):
    try:
        with open(LOG_FILE, "a") as f:
            f.write(f"\n[{datetime.datetime.now()}]\n")
            for r in sorted(results, key=lambda x: x["host"]):
                line = f"{r['host']} ({r['name']}): {r['msg']}"
                if r.get("reboot"):
                    line += " [REBOOT]"
                line += f" ({r['duration']:.1f}s)\n"
                f.write(line)
    except Exception:
        pass  # Logging darf nicht killen

def main():
    results = []
    with ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor:
        futures = [executor.submit(check_and_update, s) for s in SERVERS]
        for fut in as_completed(futures):
            results.append(fut.result())

    msg = format_telegram(results)
    send_telegram(msg)
    log_results(results)

if __name__ == "__main__":
    main()
</code>

Lege dann eine .env-Datei in /etc/update-checker.env an
<code>
TG_BOT_TOKEN=DEIN_NEUER_TOKEN
TG_CHAT_ID=-1002414259924
TG_THREAD_ID=21266

SSH_KEY_PATH=/root/.ssh/id_ed25519
LOG_FILE=/var/log/update_checker.log

AUTO_UPDATE=true
ONLY_REPORT_CHANGES=false
MAX_WORKERS=5
</code>

Danach folgenden Befehl eingeben:
<code>
install -m 750 /usr/local/bin/pruefe-installation.py /usr/local/bin/update-checker.py
</code>

==== Anlegen des systemd Services ====
Service Datei anlegen:
<code>
nano /etc/systemd/system/update-checker.service
</code>
Inhalt:
<code>
[Unit]
Description=Update Checker (SSH) + Telegram
After=network-online.target
Wants=network-online.target

[Service]
Type=oneshot
User=root
EnvironmentFile=/etc/update-checker.env
ExecStart=/usr/bin/python3 /usr/local/bin/update-checker.py

# optional: Sicherheit/Isolation
NoNewPrivileges=true
PrivateTmp=true

[Install]
WantedBy=multi-user.target
</code>

==== systemd Timer anlegen ====
Datei anlegen:
<code>
nano /etc/systemd/system/update-checker.timer
</code>
Inhalt:
<code>
[Unit]
Description=Run Update Checker daily

[Timer]
OnCalendar=*-*-* 06:30:00
Persistent=true
Unit=update-checker.service

[Install]
WantedBy=timers.target
</code>

==== systemd neu laden & Timer aktivieren ====
<code>
systemctl daemon-reload
systemctl enable --now update-checker.timer
</code>

==== Testen (Manuell) ====
<code>
systemctl start update-checker.service
systemctl status update-checker.service --no-pager
</code>
Meine Dokumentationen

Benutzer-Werkzeuge

Webseiten-Werkzeuge

Seiten-Werkzeuge
