Workflow: Automation Scripts¶
Python und Bash Scripts für den MSP/MSSP-Alltag entwickeln.
Übersicht¶
| Aspekt | Details |
|---|---|
| Ziel | Funktionierende Scripts für Automatisierung |
| Features | Iteratives Development, Background Tasks, Testing |
| Sprachen | Python, Bash, PowerShell |
| Output | Production-ready Scripts mit Error Handling |
Feature-Kombination¶
┌─────────────────────────────────────────────────────────┐
│ ANFORDERUNG │
│ "Ich brauche ein Script das X macht" │
└───────────────────────────┬─────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────┐
│ ITERATIVE ENTWICKLUNG │
│ Grundstruktur → Logik → Error Handling → Polish │
└───────────────────────────┬─────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────┐
│ BACKGROUND TASKS │
│ Tests laufen im Hintergrund während du weiterarbeitest │
└─────────────────────────────────────────────────────────┘
Workflow: Iteratives Development¶
Statt "mach alles auf einmal" - schrittweise aufbauen.
Schritt 1: Grundstruktur¶
Erstelle ein Python-Script das:
- FortiGate Configs via API pulled
- In einem Ordner mit Timestamp speichert
Erst nur die Grundstruktur, noch keine Logik.
Output:
#!/usr/bin/env python3
"""FortiGate Config Backup Script"""
import argparse
import logging
from datetime import datetime
from pathlib import Path
def setup_logging():
pass
def parse_args():
pass
def backup_config(host: str, token: str) -> str:
pass
def save_config(config: str, output_dir: Path, hostname: str):
pass
def main():
pass
if __name__ == "__main__":
main()
Schritt 2: Logik füllen¶
Schritt 3: Error Handling¶
Füge Error Handling hinzu:
- Connection Errors (Retry mit Backoff)
- Auth Errors (klare Fehlermeldung)
- Disk Errors (Platz prüfen)
Schritt 4: Polish¶
Finale Verbesserungen:
- Docstrings für alle Funktionen
- Type Hints
- --dry-run Flag
- Config-File Support (.env oder YAML)
Workflow: Mit Background Tasks¶
Für Scripts die Tests brauchen.
Schritt 1: Script erstellen¶
Erstelle ein Bash-Script das:
- Alle Docker Container Health-Checks
- Unhealthy Container neu startet
- Slack-Notification bei Restart
Schritt 2: Tests im Background¶
Teste das Script im Background mit verschiedenen Szenarien.
Simuliere: healthy, unhealthy, nicht erreichbar.
Drücke Ctrl+B wenn der Test läuft - Terminal bleibt frei.
Schritt 3: Ergebnisse holen¶
Beispiel-Scripts aus dem MSP-Alltag¶
FortiGate Backup Script¶
#!/usr/bin/env python3
"""
FortiGate Config Backup via REST API
Speichert Configs mit Timestamp in strukturiertem Ordner.
"""
import argparse
import logging
import sys
from datetime import datetime
from pathlib import Path
from typing import Optional
import requests
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
def get_config(host: str, token: str, vdom: str = "root") -> Optional[str]:
"""Holt Full-Config von FortiGate via API."""
url = f"https://{host}/api/v2/monitor/system/config/backup"
params = {"scope": "global", "vdom": vdom}
headers = {"Authorization": f"Bearer {token}"}
try:
response = requests.get(
url,
headers=headers,
params=params,
verify=False,
timeout=30
)
response.raise_for_status()
return response.text
except requests.exceptions.RequestException as e:
logger.error(f"Fehler bei {host}: {e}")
return None
def save_config(config: str, output_dir: Path, hostname: str) -> Path:
"""Speichert Config mit Timestamp."""
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
filename = f"{hostname}_{timestamp}.conf"
filepath = output_dir / hostname / filename
filepath.parent.mkdir(parents=True, exist_ok=True)
filepath.write_text(config)
logger.info(f"Gespeichert: {filepath}")
return filepath
def main():
parser = argparse.ArgumentParser(description="FortiGate Backup")
parser.add_argument("--host", required=True, help="FortiGate IP/Hostname")
parser.add_argument("--token", required=True, help="API Token")
parser.add_argument("--output", default="./backups", help="Output Directory")
parser.add_argument("--vdom", default="root", help="VDOM Name")
args = parser.parse_args()
output_dir = Path(args.output)
config = get_config(args.host, args.token, args.vdom)
if config:
save_config(config, output_dir, args.host)
sys.exit(0)
else:
sys.exit(1)
if __name__ == "__main__":
main()
Docker Health Monitor (Bash)¶
#!/bin/bash
#
# Docker Container Health Monitor
# Prüft Container, restartet Unhealthy, sendet Slack Alert
#
set -euo pipefail
SLACK_WEBHOOK="${SLACK_WEBHOOK:-}"
LOG_FILE="/var/log/docker-health.log"
log() {
echo "[$(date '+%Y-%m-%d %H:%M:%S')] $1" | tee -a "$LOG_FILE"
}
send_slack() {
local message="$1"
if [[ -n "$SLACK_WEBHOOK" ]]; then
curl -s -X POST -H 'Content-type: application/json' \
--data "{\"text\":\"$message\"}" \
"$SLACK_WEBHOOK" > /dev/null
fi
}
check_containers() {
local unhealthy
unhealthy=$(docker ps --filter "health=unhealthy" --format "{{.Names}}" 2>/dev/null || true)
if [[ -z "$unhealthy" ]]; then
log "INFO: Alle Container healthy"
return 0
fi
for container in $unhealthy; do
log "WARN: Container unhealthy: $container"
log "INFO: Restarting $container..."
if docker restart "$container" > /dev/null 2>&1; then
log "INFO: Restart erfolgreich: $container"
send_slack "🔄 Container restarted: $container"
else
log "ERROR: Restart fehlgeschlagen: $container"
send_slack "🚨 Container restart FAILED: $container"
fi
done
}
main() {
log "INFO: Health Check gestartet"
check_containers
log "INFO: Health Check abgeschlossen"
}
main "$@"
Bulk DNS Update (Python)¶
#!/usr/bin/env python3
"""
Bulk DNS Record Update für Technitium DNS
Liest Records aus CSV, updated via API.
"""
import csv
import sys
from dataclasses import dataclass
from pathlib import Path
import requests
@dataclass
class DNSRecord:
name: str
type: str
value: str
ttl: int = 3600
def load_records(csv_path: Path) -> list[DNSRecord]:
"""Lädt Records aus CSV."""
records = []
with open(csv_path) as f:
reader = csv.DictReader(f)
for row in reader:
records.append(DNSRecord(
name=row['name'],
type=row['type'],
value=row['value'],
ttl=int(row.get('ttl', 3600))
))
return records
def update_record(api_url: str, token: str, record: DNSRecord) -> bool:
"""Updated einen DNS Record via API."""
endpoint = f"{api_url}/api/zones/records/add"
params = {
"token": token,
"domain": record.name,
"type": record.type,
"value": record.value,
"ttl": record.ttl,
"overwrite": "true"
}
try:
response = requests.get(endpoint, params=params, timeout=10)
return response.json().get("status") == "ok"
except Exception as e:
print(f"Error updating {record.name}: {e}")
return False
def main():
if len(sys.argv) < 4:
print("Usage: dns_update.py <api_url> <token> <csv_file>")
sys.exit(1)
api_url, token, csv_file = sys.argv[1:4]
records = load_records(Path(csv_file))
success, failed = 0, 0
for record in records:
if update_record(api_url, token, record):
print(f"✓ {record.name}")
success += 1
else:
print(f"✗ {record.name}")
failed += 1
print(f"\nErgebnis: {success} OK, {failed} Failed")
sys.exit(0 if failed == 0 else 1)
if __name__ == "__main__":
main()
Tipps für gute Ergebnisse¶
DO ✅¶
Erstelle ein Python-Script das:
- Input: CSV mit Hostname, IP, VLAN
- Output: Cisco Switch Config für jeden Host
- Error Handling für malformed CSV
- Logging nach stdout und File
- Dry-run Modus
Bibliotheken: csv, logging, argparse, jinja2
Klare Anforderungen, erwartete Libraries, Modi.
DON'T ❌¶
Zu vage - welcher Vendor, welches Format, welche Features?
TIPP: Bestehenden Code erweitern 💡¶
Hier ist mein bestehendes Script:
@scripts/backup.py
Erweitere es um:
- Parallel-Execution für mehrere Hosts
- Progress Bar (tqdm)
- JSON-Report am Ende
TIPP: Test-Driven 💡¶
Erstelle erst die Tests für diese Funktion:
- Input: Liste von IPs
- Output: Dict mit Hostname-Auflösung
- Edge Cases: Timeout, Invalid IP, DNS Failure
Dann implementiere die Funktion.
Kosten-Optimierung¶
| Szenario | Empfehlung |
|---|---|
| Einfaches Script (<100 Zeilen) | Sonnet direkt |
| Komplexes Script (iterativ) | Sonnet, schrittweise |
| Mehrere Scripts gleichzeitig | Sub-Agents mit Haiku |
| Code Review | Fork Context, Sonnet |
Zeit vs Kosten¶
Selbst wenn du 3 Iterationen brauchst, bist du bei <$2.
Troubleshooting¶
Script läuft nicht¶
Das Script wirft diesen Error:
[Error-Message]
Kontext:
- Python 3.11
- Ubuntu 24.04
- Virtuelle Umgebung
Finde und fixe das Problem.
Performance-Problem¶
Das Script ist zu langsam (5 min für 100 Hosts).
Optimiere für Parallelisierung.
Behalte Error Handling bei.
Dependency-Konflikt¶
Best Practices¶
Error Handling Checkliste¶
Gute Scripts behandeln diese Fehlerklassen:
# 1. Netzwerk-Fehler mit Retry
import time
from functools import wraps
def retry(max_attempts=3, delay=5):
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
for attempt in range(max_attempts):
try:
return func(*args, **kwargs)
except (ConnectionError, TimeoutError) as e:
if attempt == max_attempts - 1:
raise
time.sleep(delay * (attempt + 1))
return wrapper
return decorator
# 2. Auth-Fehler klar melden
except requests.HTTPError as e:
if e.response.status_code == 401:
logger.error("Auth fehlgeschlagen - API Token ungültig?")
sys.exit(2) # Exit-Code 2 für Auth-Fehler
# 3. Partial-Failure tracken (nicht beim ersten Fehler abbrechen)
results = {"success": [], "failed": []}
for host in hosts:
try:
results["success"].append(process(host))
except Exception as e:
results["failed"].append({"host": host, "error": str(e)})
logger.warning(f"Fehler bei {host}: {e}, weiter...")
Claude-Prompt für robuste Scripts:
Erstelle das Script mit:
- Retry-Logik (3 Versuche, exponential backoff) für API-Calls
- Partial-Failure: Weiter bei Einzel-Fehler, Fehler am Ende zusammenfassen
- Exit-Codes: 0=OK, 1=Fehler, 2=Auth-Problem, 3=Config-Fehler
- Dry-run Flag (--dry-run) der keine Änderungen macht
Logging Standards¶
# Standard-Setup das Claude generieren soll:
import logging
import sys
def setup_logging(verbose: bool = False) -> logging.Logger:
level = logging.DEBUG if verbose else logging.INFO
logging.basicConfig(
level=level,
format='%(asctime)s %(levelname)-8s %(message)s',
datefmt='%Y-%m-%d %H:%M:%S',
handlers=[
logging.StreamHandler(sys.stdout),
logging.FileHandler(f'/var/log/{script_name}.log')
]
)
return logging.getLogger(__name__)
Credentials nie als Argumente¶
# SCHLECHT: Token in Command-Line History sichtbar
parser.add_argument("--token", required=True)
# GUT: Aus Environment oder File
import os
token = os.environ.get("FG_API_TOKEN") or \
read_from_file(os.environ.get("FG_TOKEN_FILE"))
if not token:
logger.error("FG_API_TOKEN nicht gesetzt")
sys.exit(3)
Bash-Scripts mit set -euo pipefail¶
Jedes Bash-Script sollte damit anfangen:
#!/usr/bin/env bash
set -euo pipefail
# -e: Bei Fehler sofort abbrechen
# -u: Ungesetzte Variablen als Fehler
# -o pipefail: Pipe-Fehler propagieren
# Dann: Cleanup-Funktion
cleanup() {
rm -f "$TMPFILE"
}
trap cleanup EXIT INT TERM
Claude-Prompt: