Python fuer Systemadministratoren: Infrastrukturaufgaben Automatisieren

Jeder Systemadministrator erreicht irgendwann den Punkt, an dem Bash-Skripte unhandlich werden. Eine Festplattenauslastungswarnung, die als Fuenf-Zeilen-Skript begann, umfasst jetzt dreihundert Zeilen, behandelt Randfaelle schlecht und versagt stillschweigend auf einer anderen Distribution. Python schliesst die Luecke zwischen schnellen Shell-Einzeilern und vollwertigen Konfigurationsmanagement-Tools wie Ansible. Es bietet ordentliche Fehlerbehandlung, umfangreiche Datenstrukturen, Tausende von Bibliotheken und Code, der auch Monate nach dem Schreiben noch lesbar ist.

Dieser Artikel fuehrt durch die wesentlichen Python-Faehigkeiten, die ein Linux-Systemadministrator benoetigt: Dateioperationen, Subprozess-Verwaltung, SSH-Automatisierung, REST-API-Integration, Log-Analyse, Systemueberwachung und Benachrichtigungen. Jeder Abschnitt enthaelt produktionsreifen Code, den Sie an Ihre Umgebung anpassen koennen.

---

Voraussetzungen

Bevor Sie beginnen, stellen Sie sicher, dass Sie Folgendes haben:

• Python 3.10+ — pruefen Sie mit python3 --version
• pip — der Python-Paketinstaller (normalerweise mit Python mitgeliefert)
• venv — das integrierte Modul fuer virtuelle Umgebungen
• Einen Linux-Server (Ubuntu 22.04/24.04, Debian 12 oder RHEL 9 empfohlen)
• Grundlegende Vertrautheit mit der Linux-Kommandozeile und Bash
• Ein SSH-Schluesselpaar, konfiguriert fuer den Zugriff auf Remote-Server (siehe SSH-Haertung)

# Python-Installation pruefen
python3 --version
# Python 3.12.3

# pip pruefen
pip3 --version
# pip 24.0 from /usr/lib/python3/dist-packages/pip (python 3.12)

---

Eine Python-Umgebung Einrichten

Installieren Sie Python-Pakete niemals global auf einem Produktionsserver. Verwenden Sie virtuelle Umgebungen, um Abhaengigkeiten fuer jedes Projekt zu isolieren.

Eine virtuelle Umgebung erstellen

# Projektverzeichnis erstellen
mkdir -p ~/sysadmin-scripts
cd ~/sysadmin-scripts

# Virtuelle Umgebung erstellen
python3 -m venv venv

# Aktivieren
source venv/bin/activate

# Ihr Prompt aendert sich und zeigt (venv) an
(venv) user@server:~/sysadmin-scripts$

Wichtige Bibliotheken installieren

# Alle Bibliotheken installieren, die wir in diesem Artikel verwenden
pip install paramiko requests psutil python-dotenv jinja2

# Abhaengigkeiten fuer Reproduzierbarkeit einfrieren
pip freeze > requirements.txt

Projektstruktur

Ein gut organisiertes Sysadmin-Skript-Projekt sieht folgendermassen aus:

sysadmin-scripts/
├── venv/
├── requirements.txt
├── .env                  # Geheimnisse (API-Schluessel, Passwoerter)
├── config.yaml           # Serverlisten, Schwellenwerte
├── scripts/
│   ├── disk_monitor.py
│   ├── log_parser.py
│   ├── remote_exec.py
│   └── backup_rotation.py
└── logs/
    └── automation.log

---

Datei- und Verzeichnisoperationen

Das pathlib-Modul von Python (seit Python 3.4 in der Standardbibliothek) bietet eine objektorientierte Schnittstelle fuer Dateisystempfade. In Kombination mit shutil deckt es praktisch jede Dateioperation ab, die ein Systemadministrator benoetigt.

Arbeiten mit pathlib

from pathlib import Path
import shutil
from datetime import datetime, timedelta

# Pfade erstellen (funktioniert auf jedem Betriebssystem)
log_dir = Path("/var/log/myapp")
backup_dir = Path("/backup/logs")

# Verzeichnisse erstellen (parents=True wirkt wie mkdir -p)
backup_dir.mkdir(parents=True, exist_ok=True)

# Alle .log-Dateien auflisten
for log_file in log_dir.glob("*.log"):
    print(f"{log_file.name}  {log_file.stat().st_size / 1024:.1f} KB")

# Dateien rekursiv suchen
for conf in Path("/etc").rglob("*.conf"):
    print(conf)

Automatisiertes Log-Rotationsskript

#!/usr/bin/env python3
"""Log-Dateien aelter als N Tage rotieren und komprimieren."""

import gzip
import shutil
from pathlib import Path
from datetime import datetime, timedelta

LOG_DIR = Path("/var/log/myapp")
ARCHIVE_DIR = Path("/var/log/myapp/archive")
MAX_AGE_DAYS = 7

ARCHIVE_DIR.mkdir(parents=True, exist_ok=True)

cutoff = datetime.now() - timedelta(days=MAX_AGE_DAYS)

for log_file in LOG_DIR.glob("*.log"):
    mtime = datetime.fromtimestamp(log_file.stat().st_mtime)
    if mtime < cutoff:
        # Datei komprimieren
        gz_path = ARCHIVE_DIR / f"{log_file.name}.gz"
        with open(log_file, "rb") as f_in:
            with gzip.open(gz_path, "wb") as f_out:
                shutil.copyfileobj(f_in, f_out)
        log_file.unlink()
        print(f"Archiviert: {log_file.name} -> {gz_path.name}")

Backup-Skript mit Zeitstempel-Verzeichnissen

#!/usr/bin/env python3
"""Backups mit Zeitstempel erstellen und Backups aelter als 30 Tage entfernen."""

import shutil
from pathlib import Path
from datetime import datetime, timedelta

SOURCE = Path("/etc/nginx")
BACKUP_ROOT = Path("/backup/nginx")
RETENTION_DAYS = 30

# Backup mit Zeitstempel erstellen
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
backup_path = BACKUP_ROOT / timestamp

shutil.copytree(SOURCE, backup_path)
print(f"Backup erstellt: {backup_path}")

# Alte Backups bereinigen
cutoff = datetime.now() - timedelta(days=RETENTION_DAYS)
for old_backup in BACKUP_ROOT.iterdir():
    if old_backup.is_dir():
        mtime = datetime.fromtimestamp(old_backup.stat().st_mtime)
        if mtime < cutoff:
            shutil.rmtree(old_backup)
            print(f"Altes Backup entfernt: {old_backup.name}")

---

Systembefehle mit subprocess Ausfuehren

Das subprocess-Modul ermoeglicht es, Shell-Befehle aus Python heraus mit voller Kontrolle ueber Eingabe, Ausgabe und Fehlerbehandlung auszufuehren. Verwenden Sie immer subprocess.run() anstelle des aelteren os.system().

Grundlegende Befehlsausfuehrung

import subprocess

# Befehl ausfuehren und Ausgabe erfassen
result = subprocess.run(
    ["df", "-h", "/"],
    capture_output=True,
    text=True,
    timeout=30
)

print(result.stdout)
print(f"Rueckgabecode: {result.returncode}")

# Auf Fehler pruefen
if result.returncode != 0:
    print(f"Fehler: {result.stderr}")

Mehrere Befehle sicher ausfuehren

import subprocess
import sys

def run_cmd(cmd, description=""):
    """Einen Befehl ausfuehren und Fehler behandeln."""
    try:
        result = subprocess.run(
            cmd,
            capture_output=True,
            text=True,
            timeout=60,
            check=True  # Wirft CalledProcessError bei Exit-Code ungleich Null
        )
        print(f"[OK] {description}")
        return result.stdout
    except subprocess.CalledProcessError as e:
        print(f"[FEHLER] {description}: {e.stderr.strip()}")
        sys.exit(1)
    except subprocess.TimeoutExpired:
        print(f"[TIMEOUT] {description}: Befehl hat 60s ueberschritten")
        sys.exit(1)

# Beispiel: Systemaktualisierungssequenz
run_cmd(["sudo", "apt", "update"], "Paketlisten aktualisieren")
run_cmd(["sudo", "apt", "upgrade", "-y"], "Pakete aktualisieren")
run_cmd(["sudo", "apt", "autoremove", "-y"], "Nicht verwendete Pakete entfernen")

Befehlsausgabe analysieren

import subprocess

def get_listening_ports():
    """Eine Liste der Ports mit lauschenden Diensten zurueckgeben."""
    result = subprocess.run(
        ["ss", "-tlnp"],
        capture_output=True,
        text=True,
        check=True
    )
    ports = []
    for line in result.stdout.strip().split("\n")[1:]:  # Kopfzeile ueberspringen
        parts = line.split()
        if len(parts) >= 4:
            addr = parts[3]
            port = addr.rsplit(":", 1)[-1]
            ports.append(port)
    return sorted(set(ports))

print("Lauschende Ports:", get_listening_ports())

---

SSH-Automatisierung mit Paramiko

Paramiko ist die Standard-Python-Bibliothek fuer SSH2-Verbindungen. Sie ermoeglicht die Ausfuehrung von Befehlen auf Remote-Servern, die Uebertragung von Dateien per SFTP und die Verwaltung von SSH-Schluesseln — alles aus Python heraus.

Paramiko installieren

pip install paramiko

Remote-Befehle ausfuehren

import paramiko

def ssh_exec(hostname, username, command, key_path="~/.ssh/id_ed25519"):
    """Einen Befehl auf einem Remote-Server per SSH ausfuehren."""
    client = paramiko.SSHClient()
    client.set_missing_host_key_policy(paramiko.AutoAddPolicy())

    try:
        client.connect(
            hostname=hostname,
            username=username,
            key_filename=str(Path(key_path).expanduser()),
            timeout=10
        )
        stdin, stdout, stderr = client.exec_command(command, timeout=30)
        exit_code = stdout.channel.recv_exit_status()

        return {
            "stdout": stdout.read().decode().strip(),
            "stderr": stderr.read().decode().strip(),
            "exit_code": exit_code
        }
    finally:
        client.close()

# Anwendungsbeispiel
result = ssh_exec("web01.example.com", "deploy", "uptime")
print(result["stdout"])

Befehle auf mehreren Servern ausfuehren

import paramiko
from pathlib import Path
from concurrent.futures import ThreadPoolExecutor, as_completed

SERVERS = [
    {"host": "web01.example.com", "user": "deploy"},
    {"host": "web02.example.com", "user": "deploy"},
    {"host": "db01.example.com",  "user": "deploy"},
]

def check_uptime(server):
    """Betriebszeit auf einem einzelnen Server pruefen."""
    client = paramiko.SSHClient()
    client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    try:
        client.connect(
            hostname=server["host"],
            username=server["user"],
            key_filename=str(Path("~/.ssh/id_ed25519").expanduser()),
            timeout=10
        )
        _, stdout, _ = client.exec_command("uptime -p")
        return server["host"], stdout.read().decode().strip()
    except Exception as e:
        return server["host"], f"FEHLER: {e}"
    finally:
        client.close()

# Parallel mit ThreadPoolExecutor ausfuehren
with ThreadPoolExecutor(max_workers=5) as executor:
    futures = {executor.submit(check_uptime, s): s for s in SERVERS}
    for future in as_completed(futures):
        host, uptime = future.result()
        print(f"{host}: {uptime}")

Dateiuebertragung per SFTP

import paramiko
from pathlib import Path

def upload_file(hostname, username, local_path, remote_path):
    """Eine Datei per SFTP auf einen Remote-Server hochladen."""
    transport = paramiko.Transport((hostname, 22))
    key = paramiko.Ed25519Key.from_private_key_file(
        str(Path("~/.ssh/id_ed25519").expanduser())
    )
    transport.connect(username=username, pkey=key)
    sftp = paramiko.SFTPClient.from_transport(transport)

    try:
        sftp.put(str(local_path), str(remote_path))
        print(f"Hochgeladen {local_path} -> {hostname}:{remote_path}")
    finally:
        sftp.close()
        transport.close()

# Konfigurationsdatei hochladen
upload_file(
    "web01.example.com",
    "deploy",
    Path("./configs/nginx.conf"),
    "/tmp/nginx.conf"
)

---

Arbeiten mit REST-APIs

Die requests-Bibliothek macht HTTP-Aufrufe unkompliziert. Systemadministratoren verwenden APIs, um mit Ueberwachungsplattformen, Cloud-Anbietern, DNS-Diensten und internen Tools zu interagieren.

GET-Anfragen — Daten abrufen

import requests

# Serverstatus von einer Ueberwachungs-API pruefen
response = requests.get(
    "https://api.example.com/v1/servers",
    headers={"Authorization": "Bearer YOUR_API_KEY"},
    timeout=10
)

if response.status_code == 200:
    servers = response.json()
    for server in servers:
        print(f"{server['name']}: {server['status']}")
else:
    print(f"API-Fehler: {response.status_code} {response.text}")

POST-Anfragen — Daten senden

import requests
import json

# DNS-Eintrag per API erstellen
payload = {
    "type": "A",
    "name": "web03.example.com",
    "content": "203.0.113.50",
    "ttl": 300
}

response = requests.post(
    "https://api.cloudflare.com/client/v4/zones/ZONE_ID/dns_records",
    headers={
        "Authorization": "Bearer YOUR_CF_TOKEN",
        "Content-Type": "application/json"
    },
    json=payload,
    timeout=15
)

result = response.json()
if result.get("success"):
    print(f"DNS-Eintrag erstellt: {result['result']['id']}")
else:
    print(f"Fehler: {result.get('errors')}")

Robuster API-Client mit Wiederholungslogik

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

def create_api_session(base_url, api_key, retries=3):
    """Eine requests-Sitzung mit Wiederholungslogik erstellen."""
    session = requests.Session()
    session.headers.update({
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json"
    })

    retry_strategy = Retry(
        total=retries,
        backoff_factor=1,
        status_forcelist=[429, 500, 502, 503, 504]
    )
    adapter = HTTPAdapter(max_retries=retry_strategy)
    session.mount("https://", adapter)
    session.mount("http://", adapter)

    return session

# Verwendung
api = create_api_session("https://api.example.com", "YOUR_KEY")
response = api.get("https://api.example.com/v1/health", timeout=10)
print(response.json())

---

Log-Analyse und -Auswertung

Systemadministratoren verbringen einen erheblichen Teil ihrer Zeit mit der Analyse von Log-Dateien. Python erleichtert das Parsen, Filtern und Aggregieren von Log-Daten.

Syslog mit regulaeren Ausdruecken analysieren

#!/usr/bin/env python3
"""Syslog analysieren und fehlgeschlagene SSH-Anmeldeversuche melden."""

import re
from collections import Counter
from pathlib import Path

LOG_FILE = Path("/var/log/auth.log")
PATTERN = re.compile(
    r"(\w+\s+\d+\s+[\d:]+)\s+\S+\s+sshd\[\d+\]:\s+"
    r"Failed password for (?:invalid user )?(\S+) from (\S+)"
)

ip_counter = Counter()
user_counter = Counter()

with open(LOG_FILE) as f:
    for line in f:
        match = PATTERN.search(line)
        if match:
            timestamp, user, ip = match.groups()
            ip_counter[ip] += 1
            user_counter[user] += 1

print("Top 10 angreifende IPs:")
for ip, count in ip_counter.most_common(10):
    print(f"  {ip:20s} {count:5d} Versuche")

print("\nTop 10 angegriffene Benutzernamen:")
for user, count in user_counter.most_common(10):
    print(f"  {user:20s} {count:5d} Versuche")

Strukturierte Logs (JSON) analysieren

#!/usr/bin/env python3
"""Anwendungslogs im JSON-Format analysieren."""

import json
from collections import Counter
from pathlib import Path
from datetime import datetime

LOG_FILE = Path("/var/log/myapp/access.json")

status_codes = Counter()
slow_requests = []

with open(LOG_FILE) as f:
    for line in f:
        try:
            entry = json.loads(line)
            status_codes[entry["status"]] += 1
            if entry.get("response_time", 0) > 2.0:
                slow_requests.append(entry)
        except json.JSONDecodeError:
            continue

print("Verteilung der Statuscodes:")
for code, count in sorted(status_codes.items()):
    print(f"  {code}: {count}")

print(f"\nLangsame Anfragen (>2s): {len(slow_requests)}")
for req in slow_requests[:5]:
    print(f"  {req['method']} {req['path']} - {req['response_time']:.2f}s")

Echtzeit-Log-Ueberwachung

#!/usr/bin/env python3
"""Eine Log-Datei verfolgen und bei Fehlermustern warnen."""

import time
from pathlib import Path

LOG_FILE = Path("/var/log/myapp/error.log")
ALERT_PATTERNS = ["CRITICAL", "OOM", "segfault", "disk full"]

def tail_file(filepath, interval=1.0):
    """Neue Zeilen ausgeben, sobald sie an eine Datei angehaengt werden."""
    with open(filepath) as f:
        f.seek(0, 2)  # Zum Dateiende springen
        while True:
            line = f.readline()
            if line:
                yield line.strip()
            else:
                time.sleep(interval)

print(f"Verfolge {LOG_FILE}... (Strg+C zum Beenden)")
for line in tail_file(LOG_FILE):
    for pattern in ALERT_PATTERNS:
        if pattern in line:
            print(f"[WARNUNG] {line}")
            # Hier koennte eine Benachrichtigung gesendet werden
            break

---

Skripte zur Systemueberwachung

Die psutil-Bibliothek bietet plattformuebergreifenden Zugriff auf Systemmetriken — CPU, Arbeitsspeicher, Festplatte, Netzwerk und Prozessinformationen. Sie bildet die Grundlage fuer den Aufbau individueller Ueberwachungstools.

psutil installieren

pip install psutil

Umfassende Systemzustandspruefung

#!/usr/bin/env python3
"""Umfassendes Skript zur Systemzustandspruefung."""

import psutil
import socket
from datetime import datetime

def bytes_to_human(n):
    """Bytes in lesbares Format umwandeln."""
    for unit in ["B", "KB", "MB", "GB", "TB"]:
        if n < 1024:
            return f"{n:.1f} {unit}"
        n /= 1024
    return f"{n:.1f} PB"

def check_system():
    """Alle Systempruefungen durchfuehren und Ergebnisse zurueckgeben."""
    results = {}
    hostname = socket.gethostname()

    # CPU
    cpu_percent = psutil.cpu_percent(interval=1)
    cpu_count = psutil.cpu_count()
    load_avg = psutil.getloadavg()
    results["cpu"] = {
        "usage_percent": cpu_percent,
        "cores": cpu_count,
        "load_avg_1m": load_avg[0],
        "load_avg_5m": load_avg[1],
        "load_avg_15m": load_avg[2],
    }

    # Arbeitsspeicher
    mem = psutil.virtual_memory()
    swap = psutil.swap_memory()
    results["memory"] = {
        "total": bytes_to_human(mem.total),
        "used": bytes_to_human(mem.used),
        "available": bytes_to_human(mem.available),
        "percent": mem.percent,
        "swap_percent": swap.percent,
    }

    # Festplatte
    results["disks"] = []
    for partition in psutil.disk_partitions():
        try:
            usage = psutil.disk_usage(partition.mountpoint)
            results["disks"].append({
                "mount": partition.mountpoint,
                "device": partition.device,
                "total": bytes_to_human(usage.total),
                "used_percent": usage.percent,
            })
        except PermissionError:
            continue

    # Netzwerk
    net = psutil.net_io_counters()
    results["network"] = {
        "bytes_sent": bytes_to_human(net.bytes_sent),
        "bytes_recv": bytes_to_human(net.bytes_recv),
    }

    return hostname, results

hostname, health = check_system()
print(f"=== Systemzustandsbericht: {hostname} ===")
print(f"Zeitstempel: {datetime.now().isoformat()}\n")

print(f"CPU: {health['cpu']['usage_percent']}% "
      f"({health['cpu']['cores']} Kerne, "
      f"Last: {health['cpu']['load_avg_1m']:.2f})")

print(f"Arbeitsspeicher: {health['memory']['percent']}% "
      f"({health['memory']['used']} / {health['memory']['total']}) "
      f"Swap: {health['memory']['swap_percent']}%")

for disk in health["disks"]:
    status = "WARNUNG" if disk["used_percent"] > 85 else "OK"
    print(f"Festplatte {disk['mount']}: {disk['used_percent']}% "
          f"von {disk['total']} [{status}]")

print(f"Netzwerk: gesendet={health['network']['bytes_sent']}, "
      f"empfangen={health['network']['bytes_recv']}")

Festplattenauslastungswarnungen mit Schwellenwerten

#!/usr/bin/env python3
"""Festplattenauslastung ueberwachen und warnen, wenn Schwellenwerte ueberschritten werden."""

import psutil

THRESHOLDS = {
    "/": 85,
    "/var": 80,
    "/home": 90,
    "/tmp": 75,
}

alerts = []

for partition in psutil.disk_partitions():
    mount = partition.mountpoint
    if mount in THRESHOLDS:
        usage = psutil.disk_usage(mount)
        if usage.percent > THRESHOLDS[mount]:
            alerts.append({
                "mount": mount,
                "usage": usage.percent,
                "threshold": THRESHOLDS[mount],
                "free_gb": usage.free / (1024 ** 3),
            })

if alerts:
    print("FESTPLATTENAUSLASTUNGS-WARNUNGEN:")
    for alert in alerts:
        print(f"  {alert['mount']}: {alert['usage']}% belegt "
              f"(Schwellenwert: {alert['threshold']}%, "
              f"frei: {alert['free_gb']:.1f} GB)")
else:
    print("Alle Festplattenpartitionen innerhalb der normalen Schwellenwerte.")

Prozessueberwachung

#!/usr/bin/env python3
"""Bestimmte Prozesse ueberwachen und warnen, wenn sie nicht mehr laufen."""

import psutil

REQUIRED_PROCESSES = ["nginx", "postgresql", "redis-server", "sshd"]

running = {p.name() for p in psutil.process_iter(["name"])}

for proc_name in REQUIRED_PROCESSES:
    if proc_name in running:
        print(f"  [LAEUFT] {proc_name}")
    else:
        print(f"  [GESTOPPT] {proc_name} -- AKTION ERFORDERLICH")

---

Benachrichtigungen Senden

Automatisierung ist nur nuetzlich, wenn Sie erfahren, wenn etwas schiefgeht. Python kann Warnmeldungen per E-Mail, Slack oder jeden Webhook-basierten Dienst senden.

E-Mail-Benachrichtigungen senden

import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart

def send_email(subject, body, to_addr, from_addr, smtp_server, smtp_port=587,
               username=None, password=None):
    """Eine E-Mail-Benachrichtigung senden."""
    msg = MIMEMultipart()
    msg["From"] = from_addr
    msg["To"] = to_addr
    msg["Subject"] = subject
    msg.attach(MIMEText(body, "plain"))

    with smtplib.SMTP(smtp_server, smtp_port) as server:
        server.starttls()
        if username and password:
            server.login(username, password)
        server.send_message(msg)

# Anwendungsbeispiel
send_email(
    subject="[WARNUNG] Festplattenauslastung ueber 90% auf web01",
    body="Partition /var bei 92% Auslastung. Freier Speicherplatz: 3,2 GB.",
    to_addr="admin@example.com",
    from_addr="warnungen@example.com",
    smtp_server="smtp.example.com",
    username="warnungen@example.com",
    password="app-spezifisches-passwort"
)

Slack-Webhook-Benachrichtigungen

import requests
import json

def send_slack_alert(webhook_url, message, severity="warning"):
    """Eine Warnung per Webhook an einen Slack-Kanal senden."""
    colors = {
        "info": "#36a64f",
        "warning": "#ff9900",
        "critical": "#ff0000",
    }

    payload = {
        "attachments": [{
            "color": colors.get(severity, "#36a64f"),
            "title": f"Serverwarnung ({severity.upper()})",
            "text": message,
            "footer": "SysAdmin-Automatisierung",
        }]
    }

    response = requests.post(
        webhook_url,
        json=payload,
        timeout=10
    )
    return response.status_code == 200

# Anwendungsbeispiel
send_slack_alert(
    webhook_url="https://hooks.slack.com/services/T00/B00/XXXX",
    message="Festplattenauslastung auf /var hat 90% auf web01.example.com ueberschritten",
    severity="critical"
)

Ueberwachung mit Benachrichtigungen kombinieren

#!/usr/bin/env python3
"""Systemressourcen ueberwachen und Warnungen senden, wenn Schwellenwerte ueberschritten werden."""

import psutil
import requests
import socket
from datetime import datetime

SLACK_WEBHOOK = "https://hooks.slack.com/services/T00/B00/XXXX"
HOSTNAME = socket.gethostname()
CPU_THRESHOLD = 90
MEMORY_THRESHOLD = 85
DISK_THRESHOLD = 85

def alert(message, severity="warning"):
    """Warnung an Slack senden."""
    payload = {
        "text": f"*[{severity.upper()}]* `{HOSTNAME}` - {message}\n"
                f"_Zeitstempel: {datetime.now().isoformat()}_"
    }
    requests.post(SLACK_WEBHOOK, json=payload, timeout=10)

# CPU pruefen
cpu = psutil.cpu_percent(interval=2)
if cpu > CPU_THRESHOLD:
    alert(f"CPU-Auslastung bei {cpu}% (Schwellenwert: {CPU_THRESHOLD}%)", "critical")

# Arbeitsspeicher pruefen
mem = psutil.virtual_memory()
if mem.percent > MEMORY_THRESHOLD:
    alert(f"Speicherauslastung bei {mem.percent}% (Schwellenwert: {MEMORY_THRESHOLD}%)", "critical")

# Festplatten pruefen
for part in psutil.disk_partitions():
    try:
        usage = psutil.disk_usage(part.mountpoint)
        if usage.percent > DISK_THRESHOLD:
            alert(f"Festplatte {part.mountpoint} bei {usage.percent}% "
                  f"(Schwellenwert: {DISK_THRESHOLD}%)", "warning")
    except PermissionError:
        continue

---

Python-Skripte Zeitgesteuert Ausfuehren

Cron verwenden

Der einfachste Weg, Python-Skripte unter Linux zeitgesteuert auszufuehren, ist mit cron.

# Crontab bearbeiten
crontab -e

# Festplattenmonitor alle 15 Minuten ausfuehren
*/15 * * * * /home/deploy/sysadmin-scripts/venv/bin/python /home/deploy/sysadmin-scripts/scripts/disk_monitor.py >> /var/log/disk_monitor.log 2>&1

# Vollstaendige Zustandspruefung stuendlich ausfuehren
0 * * * * /home/deploy/sysadmin-scripts/venv/bin/python /home/deploy/sysadmin-scripts/scripts/health_check.py >> /var/log/health_check.log 2>&1

# Log-Analyse taeglich um 6 Uhr ausfuehren
0 6 * * * /home/deploy/sysadmin-scripts/venv/bin/python /home/deploy/sysadmin-scripts/scripts/log_parser.py >> /var/log/log_analysis.log 2>&1

Verwenden Sie immer den vollstaendigen Pfad zur Python-Binary innerhalb Ihrer virtuellen Umgebung. Dies stellt sicher, dass das Skript die richtige Python-Version verwendet und Zugriff auf die installierten Pakete hat.

systemd-Timer verwenden

Fuer erweiterte Zeitplanung mit Protokollierungsintegration und Abhaengigkeitsverwaltung verwenden Sie systemd-Timer.

# /etc/systemd/system/health-check.service
[Unit]
Description=System Health Check
After=network.target

[Service]
Type=oneshot
User=deploy
ExecStart=/home/deploy/sysadmin-scripts/venv/bin/python /home/deploy/sysadmin-scripts/scripts/health_check.py
StandardOutput=journal
StandardError=journal

# /etc/systemd/system/health-check.timer
[Unit]
Description=Run health check every 15 minutes

[Timer]
OnBootSec=5min
OnUnitActiveSec=15min
Persistent=true

[Install]
WantedBy=timers.target

# Timer aktivieren und starten
sudo systemctl daemon-reload
sudo systemctl enable health-check.timer
sudo systemctl start health-check.timer

# Timer-Status pruefen
systemctl list-timers --all | grep health

---

Referenztabelle Nuetzlicher Bibliotheken

Bibliothek	Verwendungszweck	Installation	Dokumentation
pathlib	Datei-/Verzeichnispfad-Operationen	Integriert	docs.python.org/3/library/pathlib
subprocess	Systembefehle ausfuehren	Integriert	docs.python.org/3/library/subprocess
shutil	Dateioperationen auf hoher Ebene (Kopieren, Verschieben, Archivieren)	Integriert	docs.python.org/3/library/shutil
os	Low-Level-Betriebssystemschnittstelle, Umgebungsvariablen	Integriert	docs.python.org/3/library/os
paramiko	SSH2-Protokoll, Remote-Befehlsausfuehrung, SFTP	pip install paramiko	paramiko.org
requests	HTTP-Client fuer REST-APIs	pip install requests	requests.readthedocs.io
psutil	Systemueberwachung (CPU, Arbeitsspeicher, Festplatte, Netzwerk)	pip install psutil	psutil.readthedocs.io
python-dotenv	Umgebungsvariablen aus .env-Dateien laden	pip install python-dotenv	pypi.org/project/python-dotenv
jinja2	Template-Engine zur Generierung von Konfigurationsdateien	pip install jinja2	jinja.palletsprojects.com
pyyaml	YAML-Konfigurationsdateien lesen und schreiben	pip install pyyaml	pyyaml.org
fabric	SSH-Automatisierung auf hoher Ebene (basierend auf Paramiko)	pip install fabric	fabfile.org
schedule	Benutzerfreundliche Aufgabenplanung	pip install schedule	schedule.readthedocs.io
logging	Strukturierte Anwendungsprotokollierung	Integriert	docs.python.org/3/library/logging

---

Fehlerbehebung

ModuleNotFoundError bei Ausfuehrung ueber cron

Cron verwendet eine minimale Umgebung. Wenn Ihr Skript mit ModuleNotFoundError fehlschlaegt, verwenden Sie wahrscheinlich nicht den vollstaendigen Pfad zur Python-Binary der virtuellen Umgebung.

# Falsch -- verwendet System-Python, dem Ihre Pakete fehlen
* * * * * python3 /home/deploy/scripts/monitor.py

# Richtig -- verwendet das Python der virtuellen Umgebung
* * * * * /home/deploy/sysadmin-scripts/venv/bin/python /home/deploy/scripts/monitor.py

Verbindung von Paramiko abgelehnt

Wenn Paramiko einen ConnectionRefusedError wirft, ueberpruefen Sie, ob SSH auf dem Zielsystem laeuft und der Port korrekt ist.

# Benutzerdefinierten Port angeben, falls Ihr Server einen nicht standardmaessigen SSH-Port verwendet
client.connect(hostname="web01.example.com", port=2222, username="deploy",
               key_filename="/home/deploy/.ssh/id_ed25519")

Zugriff verweigert bei Dateioperationen

Wenn Skripte als Nicht-Root-Benutzer ausgefuehrt werden, haben sie moeglicherweise keinen Zugriff auf Dateien unter /var/log oder /etc. Fuehren Sie das Skript entweder mit sudo aus oder passen Sie die Dateiberechtigungen an.

# Lesezugriff auf auth.log fuer den Benutzer deploy gewaehren
sudo usermod -aG adm deploy

# Oder ACLs fuer feinkoernige Steuerung verwenden
sudo setfacl -m u:deploy:r /var/log/auth.log

psutil gibt 0,0% CPU beim ersten Aufruf zurueck

psutil.cpu_percent() gibt beim ersten Aufruf 0,0 zurueck, da zwei Messungen zur Berechnung der Auslastung benoetigt werden. Uebergeben Sie immer ein Intervall oder rufen Sie die Funktion zweimal auf.

# Richtig: Intervall fuer eine blockierende Messung uebergeben
cpu = psutil.cpu_percent(interval=1)

# Oder zweimal mit Verzoegerung aufrufen
psutil.cpu_percent()
import time; time.sleep(1)
cpu = psutil.cpu_percent()

Timeout- oder SSL-Fehler bei requests

Beim Verbinden mit APIs hinter Unternehmens-Proxys oder mit selbstsignierten Zertifikaten koennen SSL-Fehler auftreten.

# SSL-Verifizierung deaktivieren (nur Entwicklung -- niemals in Produktion)
response = requests.get("https://internal-api.local", verify=False)

# Oder ein benutzerdefiniertes CA-Bundle angeben
response = requests.get("https://api.example.com", verify="/etc/ssl/custom-ca.pem")

---

Zusammenfassung

Python ist das Schweizer Taschenmesser der Systemadministration. In dieser Anleitung haben Sie gelernt:

• Isolierte Python-Umgebungen mit venv fuer Sysadmin-Projekte einzurichten
• Dateien und Verzeichnisse programmatisch mit pathlib und shutil zu verwalten
• Systembefehle mit subprocess auszufuehren und zu analysieren
• SSH-Verbindungen und Dateiuebertragungen mit paramiko zu automatisieren
• Mit REST-APIs ueber requests mit ordentlicher Wiederholungslogik zu interagieren
• Log-Dateien mit regulaeren Ausdruecken und JSON-Verarbeitung zu analysieren
• Ueberwachungsskripte mit psutil fuer CPU-, Arbeitsspeicher-, Festplatten- und Netzwerkmetriken zu erstellen
• Benachrichtigungen per E-Mail und Slack-Webhooks zu senden
• Skripte zuverlaessig mit cron und systemd-Timern zu planen

Fangen Sie klein an: Waehlen Sie eine sich wiederholende Aufgabe, die Sie woechentlich durchfuehren (Log-Rotation, Zustandspruefungen, Backup-Verifizierung) und automatisieren Sie diese mit Python. Mit wachsendem Vertrauen koennen Sie diese Skripte zu einem umfassenden Automatisierungs-Toolkit zusammenfuegen.

Fuer verwandte Anleitungen zur Absicherung der Server, auf denen diese Skripte ausgefuehrt werden, lesen Sie die Sicherheitscheckliste fuer Linux-Server und SSH-Haertung: 12 Schritte zur Absicherung Ihres Linux-Servers.