Tout administrateur systeme finit par atteindre un point ou les scripts Bash deviennent ingeres. Une alerte d’utilisation disque qui a commence comme un script de cinq lignes s’etend desormais sur trois cents lignes, gere mal les cas limites et echoue silencieusement sur une distribution differente. Python comble le fosse entre les commandes shell rapides et les outils complets de gestion de configuration comme Ansible. Il offre une gestion des erreurs appropriee, des structures de donnees riches, des milliers de bibliotheques et un code lisible des mois apres l’avoir ecrit.
Cet article parcourt les competences Python essentielles dont un administrateur systeme Linux a besoin : operations sur les fichiers, gestion des sous-processus, automatisation SSH, integration d’API REST, analyse de logs, surveillance systeme et notifications. Chaque section comprend du code pret pour la production que vous pouvez adapter a votre environnement.
Prerequis
Avant de commencer, assurez-vous de disposer des elements suivants :
- Python 3.10+ — verifiez avec
python3 --version - pip — le gestionnaire de paquets Python (generalement fourni avec Python)
- venv — le module integre d’environnements virtuels
- Un serveur Linux (Ubuntu 22.04/24.04, Debian 12 ou RHEL 9 recommandes)
- Une connaissance de base de la ligne de commande Linux et de Bash
- Une paire de cles SSH configuree pour l’acces aux serveurs distants (voir Securisation SSH)
# Verifier l'installation de Python
python3 --version
# Python 3.12.3
# Verifier pip
pip3 --version
# pip 24.0 from /usr/lib/python3/dist-packages/pip (python 3.12)
Configurer un Environnement Python
N’installez jamais de paquets Python globalement sur un serveur de production. Utilisez des environnements virtuels pour isoler les dependances de chaque projet.
Creer un environnement virtuel
# Creer un repertoire de projet
mkdir -p ~/sysadmin-scripts
cd ~/sysadmin-scripts
# Creer un environnement virtuel
python3 -m venv venv
# L'activer
source venv/bin/activate
# Votre invite de commande change pour afficher (venv)
(venv) user@server:~/sysadmin-scripts$
Installer les bibliotheques essentielles
# Installer toutes les bibliotheques utilisees dans cet article
pip install paramiko requests psutil python-dotenv jinja2
# Figer les dependances pour la reproductibilite
pip freeze > requirements.txt
Structure du projet
Un projet de scripts d’administration systeme bien organise ressemble a ceci :
sysadmin-scripts/
├── venv/
├── requirements.txt
├── .env # Secrets (cles API, mots de passe)
├── config.yaml # Listes de serveurs, seuils
├── scripts/
│ ├── disk_monitor.py
│ ├── log_parser.py
│ ├── remote_exec.py
│ └── backup_rotation.py
└── logs/
└── automation.log
Operations sur les Fichiers et Repertoires
Le module pathlib de Python (dans la bibliotheque standard depuis Python 3.4) fournit une interface orientee objet pour les chemins du systeme de fichiers. Combine avec shutil, il gere pratiquement toutes les operations sur les fichiers dont un administrateur systeme a besoin.
Travailler avec pathlib
from pathlib import Path
import shutil
from datetime import datetime, timedelta
# Creer des chemins (fonctionne sur tout OS)
log_dir = Path("/var/log/myapp")
backup_dir = Path("/backup/logs")
# Creer des repertoires (parents=True agit comme mkdir -p)
backup_dir.mkdir(parents=True, exist_ok=True)
# Lister tous les fichiers .log
for log_file in log_dir.glob("*.log"):
print(f"{log_file.name} {log_file.stat().st_size / 1024:.1f} KB")
# Rechercher des fichiers recursivement
for conf in Path("/etc").rglob("*.conf"):
print(conf)
Script de rotation automatique des logs
#!/usr/bin/env python3
"""Rotation et compression des fichiers de log plus anciens que N jours."""
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:
# Compresser le fichier
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"Archive : {log_file.name} -> {gz_path.name}")
Script de sauvegarde avec repertoires horodates
#!/usr/bin/env python3
"""Creer des sauvegardes horodatees et supprimer celles de plus de 30 jours."""
import shutil
from pathlib import Path
from datetime import datetime, timedelta
SOURCE = Path("/etc/nginx")
BACKUP_ROOT = Path("/backup/nginx")
RETENTION_DAYS = 30
# Creer une sauvegarde horodatee
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
backup_path = BACKUP_ROOT / timestamp
shutil.copytree(SOURCE, backup_path)
print(f"Sauvegarde creee : {backup_path}")
# Nettoyer les anciennes sauvegardes
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"Ancienne sauvegarde supprimee : {old_backup.name}")
Executer des Commandes Systeme avec subprocess
Le module subprocess permet d’executer des commandes shell depuis Python avec un controle total sur l’entree, la sortie et la gestion des erreurs. Preferez toujours subprocess.run() a l’ancien os.system().
Execution basique de commandes
import subprocess
# Executer une commande et capturer la sortie
result = subprocess.run(
["df", "-h", "/"],
capture_output=True,
text=True,
timeout=30
)
print(result.stdout)
print(f"Code de retour : {result.returncode}")
# Verifier les erreurs
if result.returncode != 0:
print(f"Erreur : {result.stderr}")
Executer plusieurs commandes en toute securite
import subprocess
import sys
def run_cmd(cmd, description=""):
"""Executer une commande et gerer les erreurs."""
try:
result = subprocess.run(
cmd,
capture_output=True,
text=True,
timeout=60,
check=True # Leve CalledProcessError sur un code de sortie non nul
)
print(f"[OK] {description}")
return result.stdout
except subprocess.CalledProcessError as e:
print(f"[ECHEC] {description}: {e.stderr.strip()}")
sys.exit(1)
except subprocess.TimeoutExpired:
print(f"[TIMEOUT] {description}: la commande a depasse 60s")
sys.exit(1)
# Exemple : sequence de mise a jour du systeme
run_cmd(["sudo", "apt", "update"], "Mettre a jour les listes de paquets")
run_cmd(["sudo", "apt", "upgrade", "-y"], "Mettre a jour les paquets")
run_cmd(["sudo", "apt", "autoremove", "-y"], "Supprimer les paquets inutilises")
Analyser la sortie des commandes
import subprocess
def get_listening_ports():
"""Retourner la liste des ports avec des services en ecoute."""
result = subprocess.run(
["ss", "-tlnp"],
capture_output=True,
text=True,
check=True
)
ports = []
for line in result.stdout.strip().split("\n")[1:]: # Ignorer l'en-tete
parts = line.split()
if len(parts) >= 4:
addr = parts[3]
port = addr.rsplit(":", 1)[-1]
ports.append(port)
return sorted(set(ports))
print("Ports en ecoute :", get_listening_ports())
Automatisation SSH avec Paramiko
Paramiko est la bibliotheque Python standard pour les connexions SSH2. Elle permet d’executer des commandes sur des serveurs distants, de transferer des fichiers via SFTP et de gerer les cles SSH — le tout depuis Python.
Installer Paramiko
pip install paramiko
Executer des commandes a distance
import paramiko
def ssh_exec(hostname, username, command, key_path="~/.ssh/id_ed25519"):
"""Executer une commande sur un serveur distant via SSH."""
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()
# Exemple d'utilisation
result = ssh_exec("web01.example.com", "deploy", "uptime")
print(result["stdout"])
Executer des commandes sur plusieurs serveurs
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):
"""Verifier le temps de fonctionnement d'un serveur."""
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"ERREUR : {e}"
finally:
client.close()
# Executer en parallele avec ThreadPoolExecutor
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}")
Transferts de fichiers par SFTP
import paramiko
from pathlib import Path
def upload_file(hostname, username, local_path, remote_path):
"""Envoyer un fichier vers un serveur distant via SFTP."""
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"Envoye {local_path} -> {hostname}:{remote_path}")
finally:
sftp.close()
transport.close()
# Envoyer un fichier de configuration
upload_file(
"web01.example.com",
"deploy",
Path("./configs/nginx.conf"),
"/tmp/nginx.conf"
)
Travailler avec les API REST
La bibliotheque requests simplifie les appels HTTP. Les administrateurs systeme utilisent les API pour interagir avec les plateformes de surveillance, les fournisseurs cloud, les services DNS et les outils internes.
Requetes GET — recuperer des donnees
import requests
# Verifier l'etat du serveur depuis une API de surveillance
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"Erreur API : {response.status_code} {response.text}")
Requetes POST — envoyer des donnees
import requests
import json
# Creer un enregistrement DNS via API
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"Enregistrement DNS cree : {result['result']['id']}")
else:
print(f"Erreur : {result.get('errors')}")
Client API robuste avec reessais
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
def create_api_session(base_url, api_key, retries=3):
"""Creer une session requests avec logique de reessai."""
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
# Utilisation
api = create_api_session("https://api.example.com", "YOUR_KEY")
response = api.get("https://api.example.com/v1/health", timeout=10)
print(response.json())
Analyse et Traitement des Logs
Les administrateurs systeme consacrent une part importante de leur temps a l’analyse des fichiers de logs. Python facilite l’analyse, le filtrage et l’agregation des donnees de logs.
Analyser syslog avec des expressions regulieres
#!/usr/bin/env python3
"""Analyser syslog et signaler les tentatives de connexion SSH echouees."""
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 des IP attaquantes :")
for ip, count in ip_counter.most_common(10):
print(f" {ip:20s} {count:5d} tentatives")
print("\nTop 10 des noms d'utilisateur cibles :")
for user, count in user_counter.most_common(10):
print(f" {user:20s} {count:5d} tentatives")
Analyser des logs structures (JSON)
#!/usr/bin/env python3
"""Analyser des logs applicatifs au format JSON."""
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("Distribution des codes de statut :")
for code, count in sorted(status_codes.items()):
print(f" {code}: {count}")
print(f"\nRequetes lentes (>2s) : {len(slow_requests)}")
for req in slow_requests[:5]:
print(f" {req['method']} {req['path']} - {req['response_time']:.2f}s")
Suivi des logs en temps reel
#!/usr/bin/env python3
"""Suivre un fichier de log et alerter sur les motifs d'erreur."""
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):
"""Produire les nouvelles lignes au fur et a mesure de leur ajout au fichier."""
with open(filepath) as f:
f.seek(0, 2) # Aller a la fin du fichier
while True:
line = f.readline()
if line:
yield line.strip()
else:
time.sleep(interval)
print(f"Suivi de {LOG_FILE}... (Ctrl+C pour arreter)")
for line in tail_file(LOG_FILE):
for pattern in ALERT_PATTERNS:
if pattern in line:
print(f"[ALERTE] {line}")
# Ici vous pourriez envoyer une notification
break
Scripts de Surveillance Systeme
La bibliotheque psutil fournit un acces multiplateforme aux metriques systeme — CPU, memoire, disque, reseau et informations sur les processus. C’est la base pour construire des outils de surveillance personnalises.
Installer psutil
pip install psutil
Verification complete de l’etat du systeme
#!/usr/bin/env python3
"""Script complet de verification de l'etat du systeme."""
import psutil
import socket
from datetime import datetime
def bytes_to_human(n):
"""Convertir des octets en format lisible."""
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():
"""Executer toutes les verifications systeme et retourner les resultats."""
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],
}
# Memoire
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,
}
# Disque
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
# Reseau
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"=== Rapport d'Etat du Systeme : {hostname} ===")
print(f"Horodatage : {datetime.now().isoformat()}\n")
print(f"CPU : {health['cpu']['usage_percent']}% "
f"({health['cpu']['cores']} coeurs, "
f"charge : {health['cpu']['load_avg_1m']:.2f})")
print(f"Memoire : {health['memory']['percent']}% "
f"({health['memory']['used']} / {health['memory']['total']}) "
f"Swap : {health['memory']['swap_percent']}%")
for disk in health["disks"]:
status = "ATTENTION" if disk["used_percent"] > 85 else "OK"
print(f"Disque {disk['mount']}: {disk['used_percent']}% "
f"de {disk['total']} [{status}]")
print(f"Reseau : envoyes={health['network']['bytes_sent']}, "
f"recus={health['network']['bytes_recv']}")
Alertes d’utilisation disque avec seuils
#!/usr/bin/env python3
"""Surveiller l'utilisation disque et alerter lorsque les seuils sont depasses."""
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("ALERTES D'UTILISATION DISQUE :")
for alert in alerts:
print(f" {alert['mount']}: {alert['usage']}% utilise "
f"(seuil : {alert['threshold']}%, "
f"libre : {alert['free_gb']:.1f} Go)")
else:
print("Toutes les partitions sont dans les seuils normaux.")
Surveillance des processus
#!/usr/bin/env python3
"""Surveiller des processus specifiques et alerter s'ils cessent de fonctionner."""
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" [EN COURS] {proc_name}")
else:
print(f" [ARRETE] {proc_name} -- ACTION REQUISE")
Envoi de Notifications
L’automatisation n’est utile que si vous savez quand quelque chose ne va pas. Python peut envoyer des alertes par e-mail, Slack ou tout service base sur des webhooks.
Envoyer des notifications par e-mail
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):
"""Envoyer une notification par e-mail."""
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)
# Exemple d'utilisation
send_email(
subject="[ALERTE] Utilisation disque superieure a 90% sur web01",
body="La partition /var est a 92% d'utilisation. Espace libre : 3.2 Go.",
to_addr="admin@example.com",
from_addr="alertes@example.com",
smtp_server="smtp.example.com",
username="alertes@example.com",
password="mot-de-passe-application"
)
Notifications par webhook Slack
import requests
import json
def send_slack_alert(webhook_url, message, severity="warning"):
"""Envoyer une alerte vers un canal Slack via webhook."""
colors = {
"info": "#36a64f",
"warning": "#ff9900",
"critical": "#ff0000",
}
payload = {
"attachments": [{
"color": colors.get(severity, "#36a64f"),
"title": f"Alerte Serveur ({severity.upper()})",
"text": message,
"footer": "Automatisation SysAdmin",
}]
}
response = requests.post(
webhook_url,
json=payload,
timeout=10
)
return response.status_code == 200
# Exemple d'utilisation
send_slack_alert(
webhook_url="https://hooks.slack.com/services/T00/B00/XXXX",
message="L'utilisation disque de /var a depasse 90% sur web01.example.com",
severity="critical"
)
Combiner surveillance et notifications
#!/usr/bin/env python3
"""Surveiller les ressources systeme et envoyer des alertes en cas de depassement des seuils."""
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"):
"""Envoyer une alerte vers Slack."""
payload = {
"text": f"*[{severity.upper()}]* `{HOSTNAME}` - {message}\n"
f"_Horodatage : {datetime.now().isoformat()}_"
}
requests.post(SLACK_WEBHOOK, json=payload, timeout=10)
# Verifier le CPU
cpu = psutil.cpu_percent(interval=2)
if cpu > CPU_THRESHOLD:
alert(f"Utilisation CPU a {cpu}% (seuil : {CPU_THRESHOLD}%)", "critical")
# Verifier la memoire
mem = psutil.virtual_memory()
if mem.percent > MEMORY_THRESHOLD:
alert(f"Utilisation memoire a {mem.percent}% (seuil : {MEMORY_THRESHOLD}%)", "critical")
# Verifier les disques
for part in psutil.disk_partitions():
try:
usage = psutil.disk_usage(part.mountpoint)
if usage.percent > DISK_THRESHOLD:
alert(f"Disque {part.mountpoint} a {usage.percent}% "
f"(seuil : {DISK_THRESHOLD}%)", "warning")
except PermissionError:
continue
Planifier les Scripts Python
Utiliser cron
La methode la plus simple pour planifier des scripts Python sous Linux est d’utiliser cron.
# Editer votre crontab
crontab -e
# Executer la surveillance disque toutes les 15 minutes
*/15 * * * * /home/deploy/sysadmin-scripts/venv/bin/python /home/deploy/sysadmin-scripts/scripts/disk_monitor.py >> /var/log/disk_monitor.log 2>&1
# Executer la verification complete toutes les heures
0 * * * * /home/deploy/sysadmin-scripts/venv/bin/python /home/deploy/sysadmin-scripts/scripts/health_check.py >> /var/log/health_check.log 2>&1
# Executer l'analyse des logs quotidiennement a 6h
0 6 * * * /home/deploy/sysadmin-scripts/venv/bin/python /home/deploy/sysadmin-scripts/scripts/log_parser.py >> /var/log/log_analysis.log 2>&1
Utilisez toujours le chemin complet vers le binaire Python de votre environnement virtuel. Cela garantit que le script utilise la bonne version de Python et a acces aux paquets installes.
Utiliser les timers systemd
Pour une planification plus avancee avec integration de la journalisation et gestion des dependances, utilisez les timers systemd.
# /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
# Activer et demarrer le timer
sudo systemctl daemon-reload
sudo systemctl enable health-check.timer
sudo systemctl start health-check.timer
# Verifier l'etat du timer
systemctl list-timers --all | grep health
Tableau de Reference des Bibliotheques Utiles
| Bibliotheque | Fonction | Installation | Documentation |
|---|---|---|---|
pathlib | Operations sur les chemins de fichiers/repertoires | Integree | docs.python.org/3/library/pathlib |
subprocess | Executer des commandes systeme | Integree | docs.python.org/3/library/subprocess |
shutil | Operations de haut niveau sur les fichiers (copie, deplacement, archivage) | Integree | docs.python.org/3/library/shutil |
os | Interface bas niveau avec le SE, variables d’environnement | Integree | docs.python.org/3/library/os |
paramiko | Protocole SSH2, execution de commandes a distance, SFTP | pip install paramiko | paramiko.org |
requests | Client HTTP pour les API REST | pip install requests | requests.readthedocs.io |
psutil | Surveillance systeme (CPU, memoire, disque, reseau) | pip install psutil | psutil.readthedocs.io |
python-dotenv | Charger les variables d’environnement depuis des fichiers .env | pip install python-dotenv | pypi.org/project/python-dotenv |
jinja2 | Moteur de templates pour la generation de fichiers de configuration | pip install jinja2 | jinja.palletsprojects.com |
pyyaml | Analyser et ecrire des fichiers de configuration YAML | pip install pyyaml | pyyaml.org |
fabric | Automatisation SSH de haut niveau (construit sur Paramiko) | pip install fabric | fabfile.org |
schedule | Planification de taches conviviale | pip install schedule | schedule.readthedocs.io |
logging | Journalisation structuree des applications | Integree | docs.python.org/3/library/logging |
Depannage
ModuleNotFoundError lors de l’execution depuis cron
Cron utilise un environnement minimal. Si votre script echoue avec ModuleNotFoundError, vous n’utilisez probablement pas le chemin complet vers le binaire Python de l’environnement virtuel.
# Incorrect -- utilise le Python du systeme, qui ne dispose pas de vos paquets
* * * * * python3 /home/deploy/scripts/monitor.py
# Correct -- utilise le Python du venv
* * * * * /home/deploy/sysadmin-scripts/venv/bin/python /home/deploy/scripts/monitor.py
Connexion refusee avec Paramiko
Si Paramiko leve une ConnectionRefusedError, verifiez que SSH est en cours d’execution sur la cible et que le port est correct.
# Specifier un port personnalise si votre serveur utilise un port SSH non standard
client.connect(hostname="web01.example.com", port=2222, username="deploy",
key_filename="/home/deploy/.ssh/id_ed25519")
Permission refusee sur les operations de fichiers
Lorsque les scripts s’executent en tant qu’utilisateur non root, ils peuvent ne pas avoir acces aux fichiers sous /var/log ou /etc. Executez le script avec sudo ou ajustez les permissions des fichiers.
# Accorder l'acces en lecture a auth.log pour l'utilisateur deploy
sudo usermod -aG adm deploy
# Ou utilisez les ACL pour un controle plus precis
sudo setfacl -m u:deploy:r /var/log/auth.log
psutil retourne 0.0% de CPU au premier appel
psutil.cpu_percent() retourne 0.0 au premier appel car il a besoin de deux mesures pour calculer l’utilisation. Passez toujours un intervalle ou appelez-le deux fois.
# Correct : passer un intervalle pour une mesure bloquante
cpu = psutil.cpu_percent(interval=1)
# Ou l'appeler deux fois avec un delai
psutil.cpu_percent()
import time; time.sleep(1)
cpu = psutil.cpu_percent()
Erreurs de timeout ou SSL avec requests
Lors de la connexion a des API derriere des proxys d’entreprise ou avec des certificats auto-signes, vous pouvez rencontrer des erreurs SSL.
# Desactiver la verification SSL (developpement uniquement -- jamais en production)
response = requests.get("https://internal-api.local", verify=False)
# Ou specifier un bundle CA personnalise
response = requests.get("https://api.example.com", verify="/etc/ssl/custom-ca.pem")
Resume
Python est le couteau suisse de l’administration systeme. Dans ce guide, vous avez appris a :
- Configurer des environnements Python isoles avec
venvpour les projets d’administration systeme - Gerer des fichiers et repertoires par programmation avec
pathlibetshutil - Executer et analyser des commandes systeme avec
subprocess - Automatiser les connexions SSH et les transferts de fichiers avec
paramiko - Interagir avec des API REST en utilisant
requestsavec une logique de reessai appropriee - Analyser et traiter des fichiers de logs avec les expressions regulieres et le traitement JSON
- Creer des scripts de surveillance avec
psutilpour les metriques CPU, memoire, disque et reseau - Envoyer des notifications par e-mail et webhooks Slack
- Planifier des scripts de maniere fiable avec cron et les timers systemd
Commencez modestement : choisissez une tache repetitive que vous effectuez chaque semaine (rotation des logs, verifications de sante, verification des sauvegardes) et automatisez-la avec Python. A mesure que votre confiance grandit, enchainez ces scripts pour constituer une boite a outils d’automatisation complete.
Pour des guides complementaires sur la securisation des serveurs ou ces scripts s’executeront, consultez la Liste de Verification de Securite des Serveurs Linux et Securisation SSH : 12 Etapes pour Securiser Votre Serveur Linux.