Python para sysadmins: automatiza tareas de red y sistema

Python es el lenguaje de scripting más usado en administración de sistemas. A diferencia de Bash, ofrece manejo de errores robusto, bibliotecas para prácticamente todo (SNMP, SSH, REST APIs, parseo de archivos) y código más legible. En este artículo vemos los módulos más útiles con ejemplos reales.

Módulos estándar útiles para sysadmins

import os          # operaciones del sistema operativo
import subprocess  # ejecutar comandos del sistema
import shutil      # operaciones de archivos de alto nivel
import pathlib     # rutas de archivos modernas
import socket      # operaciones de red
import json        # parsear/generar JSON
import csv         # leer/escribir CSV
import logging     # sistema de logs
import datetime    # fechas y horas
import re          # expresiones regulares

Ejecutar comandos del sistema

import subprocess
from pathlib import Path

resultado = subprocess.run(
    ["df", "-h", "/"],
    capture_output=True,
    text=True,
    check=True  # lanza excepción si el comando falla
)
print(resultado.stdout)


resultado = subprocess.run(
    "ps aux | grep nginx | grep -v grep",
    shell=True,
    capture_output=True,
    text=True
)
print(resultado.stdout)


def proceso_activo(nombre):
    try:
        resultado = subprocess.run(
            ["pgrep", "-x", nombre],
            capture_output=True, check=True
        )
        return True
    except subprocess.CalledProcessError:
        return False

if not proceso_activo("nginx"):
    subprocess.run(["systemctl", "start", "nginx"])
    print("nginx iniciado")

Operaciones de archivos y directorios

import os
import shutil
from pathlib import Path
from datetime import datetime

def limpiar_logs_antiguos(directorio, dias=30):
    """Elimina archivos .log con más de N días de antigüedad."""
    directorio = Path(directorio)
    ahora = datetime.now().timestamp()
    limite = dias * 86400

    eliminados = 0
    for archivo in directorio.rglob("*.log"):
        if ahora - archivo.stat().st_mtime > limite:
            print(f"Eliminando: {archivo}")
            archivo.unlink()
            eliminados += 1

    print(f"Total eliminados: {eliminados}")

limpiar_logs_antiguos("/var/log/myapp", dias=30)


def crear_backup(origen, destino_base):
    fecha = datetime.now().strftime("%Y%m%d_%H%M%S")
    destino = Path(destino_base) / f"backup_{fecha}"
    shutil.copytree(origen, destino)
    print(f"Backup creado en: {destino}")
    return destino

Operaciones de red

import socket
import subprocess

def verificar_puertos(host, puertos):
    """Verifica si una lista de puertos están abiertos."""
    resultados = {}
    for puerto in puertos:
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.settimeout(2)
        resultado = sock.connect_ex((host, puerto))
        resultados[puerto] = resultado == 0
        sock.close()
    return resultados


host = "192.168.1.10"
servicios = {22: "SSH", 80: "HTTP", 443: "HTTPS", 5432: "PostgreSQL"}
estados = verificar_puertos(host, list(servicios.keys()))

for puerto, nombre in servicios.items():
    estado = "✓ ABIERTO" if estados[puerto] else "✗ CERRADO"
    print(f"{nombre:12} (:{puerto}): {estado}")

Parsear archivos de configuración y logs

import re
from collections import Counter

def analizar_access_log(ruta_log):
    """Extrae estadísticas de un access.log de nginx."""
    patron = r'(\d+\.\d+\.\d+\.\d+).?"[A-Z]+ ([^ ]+) HTTP.?" (\d+)'

    ips = []
    codigos = []
    rutas = []

    with open(ruta_log) as f:
        for linea in f:
            match = re.search(patron, linea)
            if match:
                ip, ruta, codigo = match.groups()
                ips.append(ip)
                codigos.append(codigo)
                rutas.append(ruta)

    print("Top 10 IPs:")
    for ip, count in Counter(ips).most_common(10):
        print(f"  {ip}: {count} peticiones")

    print("\nCódigos de respuesta:")
    for codigo, count in sorted(Counter(codigos).items()):
        print(f"  {codigo}: {count}")

analizar_access_log("/var/log/nginx/access.log")

Inventario de servidores con Paramiko (SSH)

import paramiko
import csv

def obtener_info_servidor(host, usuario, clave_privada):
    """Obtiene información de un servidor por SSH."""
    cliente = paramiko.SSHClient()
    cliente.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    cliente.connect(host, username=usuario, key_filename=clave_privada)

    def ejecutar(cmd):
        _, stdout, _ = cliente.exec_command(cmd)
        return stdout.read().decode().strip()

    info = {
        "host": host,
        "hostname": ejecutar("hostname"),
        "os": ejecutar("lsb_release -ds 2>/dev/null || cat /etc/os-release | grep PRETTY_NAME | cut -d= -f2"),
        "kernel": ejecutar("uname -r"),
        "cpu_cores": ejecutar("nproc"),
        "ram_gb": ejecutar("free -g | awk 'NR==2{print $2}'"),
        "disco_libre": ejecutar("df -h / | awk 'NR==2{print $4}'"),
    }
    cliente.close()
    return info

servidores = ["10.0.0.1", "10.0.0.2", "10.0.0.3"]
inventario = [obtener_info_servidor(s, "admin", "~/.ssh/id_ed25519") for s in servidores]

with open("inventario.csv", "w", newline="") as f:
    writer = csv.DictWriter(f, fieldnames=inventario[0].keys())
    writer.writeheader()
    writer.writerows(inventario)

print("Inventario guardado en inventario.csv")

Conclusión

Python transforma tareas manuales repetitivas en scripts reproducibles y mantenibles. La combinación de módulos estándar, Paramiko para SSH y requests para APIs cubre el 95% de las necesidades de automatización de un sysadmin. Empieza con scripts simples y ve añadiendo complejidad según los necesites.

Entradas