724 lines
33 KiB
Python
724 lines
33 KiB
Python
# backup_unifi.py
|
|
#
|
|
# Respaldo automático de UniFi OS Server 5.1.21 (LXC en Proxmox)
|
|
# con UniFi Network Application 10.5.67
|
|
#
|
|
# Dos estrategias en cascada — sin depender de la nube de UniFi:
|
|
#
|
|
# Estrategia 1 — SSH/SFTP (Principal):
|
|
# Conecta por SSH al LXC y copia el archivo .unf más reciente desde
|
|
# /var/lib/unifi/backup/autobackup/ directamente. No usa ninguna API.
|
|
# Requiere: SSH habilitado en el LXC y pip install paramiko
|
|
#
|
|
# Estrategia 2 — API HTTP (Fallback):
|
|
# Autenticación por sesión + CSRF token y descarga del último backup
|
|
# vía /api/backup/download. Si no existe, reintenta con /cmd/backup.
|
|
# No depende de la nube: todo es contra la IP local del LXC.
|
|
#
|
|
import os
|
|
import re
|
|
import sys
|
|
import time
|
|
import subprocess
|
|
from datetime import datetime
|
|
import requests
|
|
import urllib3
|
|
|
|
# ── Importar paramiko (solo necesario para Estrategia 1 — SSH) ──────────────
|
|
try:
|
|
import paramiko
|
|
PARAMIKO_AVAILABLE = True
|
|
except ImportError:
|
|
PARAMIKO_AVAILABLE = False
|
|
|
|
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
# CONFIGURACIÓN
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
|
|
# ── UniFi OS Server (LXC en Proxmox) ────────────────────────────────────────
|
|
UNIFI_HOST = "192.168.1.10"
|
|
UNIFI_PORT = "11443"
|
|
UNIFI_USER = "admin"
|
|
UNIFI_PASS = "@Lasalle2599*"
|
|
UNIFI_SITE = "default"
|
|
UNIFI_BASE_URL = f"https://{UNIFI_HOST}:{UNIFI_PORT}"
|
|
|
|
# ── SSH — Estrategia 1 ───────────────────────────────────────────────────────
|
|
# UniFi OS Server en LXC: el usuario SSH es normalmente "root"
|
|
SSH_USER = "root"
|
|
SSH_PASS = "@Lasalle2599*" # Contraseña root del LXC (ajustar si difiere)
|
|
SSH_PORT = 22
|
|
SSH_KEY_PATH = "" # Ruta a clave privada (.pem / id_rsa). Dejar vacío para usar contraseña.
|
|
|
|
# Rutas de backup del OS Server (.unifi) — análisis del instalador:
|
|
#
|
|
# server.conf: /var/lib/uosserver/server.conf (línea 5534 del .sh)
|
|
# WEB_PORT leido de: grep '^WEB_PORT=' /var/lib/uosserver/server.conf → default 11443
|
|
# API del OS Server: https://HOST:11443/api/backup (POST = trigger)
|
|
# https://HOST:11443/api/backup/download (GET = descarga .unifi)
|
|
# API de sistema: https://HOST:11443/api/system (GET = health check)
|
|
#
|
|
# Rutas .unifi en el filesystem del LXC (buscadas por SSH):
|
|
# /var/lib/uosserver/data/backups/ ← OS Server backups (.unifi)
|
|
# /home/uosserver/.local/share/uosserver/backups/
|
|
# /data/unifi-os/backups/
|
|
#
|
|
# Rutas .unf (Network App backups, fallback):
|
|
# /var/lib/unifi/backup/autobackup/ ← CONFIRMADO: symlink real
|
|
# /usr/lib/unifi/data/backup/autobackup/ ← CONFIRMADO: default instalador
|
|
|
|
# Rutas SSH para OS Server backups (.unifi) — se prueban primero
|
|
SSH_OS_SERVER_PATHS = [
|
|
"/var/lib/uosserver/data/backups", # ← OS Server (ruta principal)
|
|
"/home/uosserver/.local/share/uosserver/backups", # OS Server (home alternativo)
|
|
"/data/unifi-os/backups", # OS Server (variante)
|
|
"/var/lib/uosserver/backups", # OS Server (variante plana)
|
|
]
|
|
|
|
# Rutas SSH para Network App backups (.unf) — fallback
|
|
SSH_NETWORK_PATHS = [
|
|
"/var/lib/unifi/backup/autobackup", # ← CONFIRMADO: resolución real del symlink
|
|
"/usr/lib/unifi/data/backup/autobackup", # ← CONFIRMADO: default del instalador
|
|
"/var/lib/unifi/backup", # Directorio padre alternativo
|
|
]
|
|
|
|
# Si el backup más reciente es más viejo que esto (horas), se considera stale
|
|
SSH_MAX_BACKUP_AGE_HOURS = 72
|
|
|
|
# Puerto del OS Server (confirmado: WEB_PORT en /var/lib/uosserver/server.conf, default 11443)
|
|
UNIFI_NETWORK_PORT = "8443" # Puerto directo Network App (legacy fallback)
|
|
UNIFI_NETWORK_URL = f"https://{UNIFI_HOST}:{UNIFI_NETWORK_PORT}"
|
|
|
|
# ── NAS / Destino ────────────────────────────────────────────────────────────
|
|
if os.name == 'nt':
|
|
DEFAULT_NAS_PATH = r"\\10.0.0.6\bak-unifi"
|
|
else:
|
|
DEFAULT_NAS_PATH = "/mnt/bak-unifi"
|
|
|
|
NAS_PATH = os.getenv("NAS_PATH", DEFAULT_NAS_PATH)
|
|
NAS_USER = "jenkins"
|
|
NAS_PASS = "LSJenkins2026*"
|
|
RETENTION_DAYS = 7
|
|
|
|
# ── Timeouts API ─────────────────────────────────────────────────────────────
|
|
SYSINFO_TIMEOUT = (10, 15)
|
|
BACKUP_CREATE_TIMEOUT = (15, 180) # Crear backup OS Server puede tardar ~2 min
|
|
BACKUP_CMD_TIMEOUT = (15, 120) # /cmd/backup Network App
|
|
DOWNLOAD_TIMEOUT = (15, 120)
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
# UTILIDADES
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
|
|
def sanitize_filename(text: str) -> str:
|
|
"""Elimina caracteres inválidos para nombres de archivos."""
|
|
return re.sub(r'[\\/*?:"<>| ]', '_', text)
|
|
|
|
|
|
def _sep(title: str = ""):
|
|
"""Separador visual de sección."""
|
|
if title:
|
|
print(f"\n{'─' * 4} {title} {'─' * (50 - len(title))}")
|
|
else:
|
|
print("─" * 60)
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
# NAS
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
|
|
def authenticate_nas_share(path: str, username: str, password: str) -> bool:
|
|
"""Asegura la disponibilidad del recurso NAS en Windows o Linux."""
|
|
print(f"[*] Verificando acceso al recurso NAS: {path}")
|
|
|
|
if os.path.exists(path):
|
|
print("[+] Conexión al recurso NAS activa y accesible.")
|
|
return True
|
|
|
|
if os.name == 'nt':
|
|
cmd = f'net use "{path}" "{password}" /user:"{username}"'
|
|
try:
|
|
res = subprocess.run(cmd, shell=True, capture_output=True, text=True)
|
|
if res.returncode == 0 or os.path.exists(path):
|
|
print("[+] Conexión SMB establecida con éxito en Windows.")
|
|
return True
|
|
else:
|
|
print(f"[!] Advertencia 'net use': {res.stderr.strip()}")
|
|
except Exception as e:
|
|
print(f"[!] Error al ejecutar 'net use': {e}")
|
|
else:
|
|
print(f"[!] La ruta '{path}' no existe o no está montada.")
|
|
try:
|
|
os.makedirs(path, exist_ok=True)
|
|
if os.path.exists(path):
|
|
print("[+] Directorio creado/verificado exitosamente.")
|
|
return True
|
|
except Exception as e:
|
|
print(f"[!] No se pudo crear el directorio {path}: {e}")
|
|
|
|
return os.path.exists(path)
|
|
|
|
|
|
def cleanup_old_backups(directory_path: str, days_to_keep: int = 7):
|
|
"""Elimina archivos de backup (.unf, .unifi) que superen los días de retención."""
|
|
print(f"\n[*] Ejecutando limpieza de archivos antiguos (Retención: {days_to_keep} días)...")
|
|
if not os.path.exists(directory_path):
|
|
print(f"[!] La ruta {directory_path} no está disponible para limpieza.")
|
|
return
|
|
|
|
cutoff_time = datetime.now().timestamp() - (days_to_keep * 86400)
|
|
deleted_count = 0
|
|
kept_count = 0
|
|
|
|
try:
|
|
files = [
|
|
f for f in os.listdir(directory_path)
|
|
if f.endswith(".unf") or f.endswith(".unifi")
|
|
]
|
|
for file_name in files:
|
|
file_path = os.path.join(directory_path, file_name)
|
|
if not os.path.isfile(file_path):
|
|
continue
|
|
if os.path.getmtime(file_path) < cutoff_time:
|
|
try:
|
|
os.remove(file_path)
|
|
print(f" [-] Eliminado por antigüedad (>{days_to_keep}d): {file_name}")
|
|
deleted_count += 1
|
|
except Exception as err:
|
|
print(f" [!] Error al eliminar {file_name}: {err}")
|
|
else:
|
|
kept_count += 1
|
|
print(f"[+] Limpieza finalizada: {deleted_count} eliminado(s), {kept_count} conservado(s).")
|
|
except Exception as e:
|
|
print(f"[!] Error al escanear directorio de backups: {e}")
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
# ESTRATEGIA 1 — SSH / SFTP
|
|
# Accede directamente al filesystem del LXC. No depende de ninguna API.
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
|
|
def _sftp_find_files(sftp, paths: list[str], extensions: tuple[str, ...]) -> tuple[str, list] | None:
|
|
"""
|
|
Busca en las rutas dadas el primer directorio que contenga archivos
|
|
con alguna de las extensiones indicadas. Retorna (ruta, lista_de_entries) o None.
|
|
"""
|
|
for remote_path in paths:
|
|
try:
|
|
entries = sftp.listdir_attr(remote_path)
|
|
found = [e for e in entries if any(e.filename.endswith(ext) for ext in extensions)]
|
|
if found:
|
|
exts_found = set(os.path.splitext(e.filename)[1] for e in found)
|
|
print(f"[+] Directorio de backup encontrado: {remote_path} "
|
|
f"({len(found)} archivo(s): {', '.join(sorted(exts_found))})")
|
|
return remote_path, found
|
|
else:
|
|
print(f"[i] {remote_path} existe pero no contiene {extensions}.")
|
|
except IOError:
|
|
print(f"[i] {remote_path} no encontrado en el LXC.")
|
|
return None
|
|
|
|
|
|
def backup_via_ssh() -> tuple[bytes, str] | None:
|
|
"""
|
|
Estrategia 1: SSH → SFTP al LXC de Proxmox.
|
|
|
|
Busca en este orden:
|
|
1. Backup OS Server (.unifi) en SSH_OS_SERVER_PATHS ← PRIORITARIO
|
|
2. Backup Network App (.unf) en SSH_NETWORK_PATHS ← Fallback
|
|
|
|
Retorna (contenido_bytes, extensión) o None si falló.
|
|
"""
|
|
if not PARAMIKO_AVAILABLE:
|
|
print("[!] Librería 'paramiko' no instalada. Estrategia SSH omitida.")
|
|
print(" → Instalar con: pip install paramiko")
|
|
return None
|
|
|
|
print(f"[*] Conectando por SSH a {UNIFI_HOST}:{SSH_PORT} (usuario: {SSH_USER})...")
|
|
|
|
ssh = paramiko.SSHClient()
|
|
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
|
|
|
|
try:
|
|
connect_kwargs: dict = {
|
|
"hostname": UNIFI_HOST,
|
|
"port": SSH_PORT,
|
|
"username": SSH_USER,
|
|
"timeout": 15,
|
|
"allow_agent": False,
|
|
"look_for_keys": False,
|
|
}
|
|
if SSH_KEY_PATH and os.path.exists(SSH_KEY_PATH):
|
|
connect_kwargs["key_filename"] = SSH_KEY_PATH
|
|
print(f"[i] Usando clave privada: {SSH_KEY_PATH}")
|
|
else:
|
|
connect_kwargs["password"] = SSH_PASS
|
|
|
|
ssh.connect(**connect_kwargs)
|
|
print("[+] Conexión SSH establecida correctamente.")
|
|
sftp = ssh.open_sftp()
|
|
|
|
# ── Paso 1: buscar backups del OS Server (.unifi) ───────────────────────────
|
|
print("[*] Buscando backups del OS Server (.unifi)...")
|
|
result = _sftp_find_files(sftp, SSH_OS_SERVER_PATHS, (".unifi",))
|
|
|
|
if not result:
|
|
# ── Paso 2 (fallback): buscar backups de la Network App (.unf) ──────────
|
|
print("[!] No se encontraron backups .unifi del OS Server.")
|
|
print("[*] Buscando backups de la Network App (.unf) como alternativa...")
|
|
result = _sftp_find_files(sftp, SSH_NETWORK_PATHS, (".unf",))
|
|
|
|
if not result:
|
|
print("[!] No se encontró ninguna ruta de backups en el LXC.")
|
|
print(" Para OS Server backups (.unifi): habilitar en OS Server UI → System → Backups")
|
|
print(" Para Network App backups (.unf): Settings → System → Backups → Auto Backup → ON")
|
|
sftp.close()
|
|
ssh.close()
|
|
return None
|
|
|
|
remote_path, found_entries = result
|
|
|
|
# Determinar extensión del tipo encontrado
|
|
file_ext = ".unifi" if any(e.filename.endswith(".unifi") for e in found_entries) else ".unf"
|
|
backup_type = "OS Server" if file_ext == ".unifi" else "Network App"
|
|
|
|
# Seleccionar el archivo más reciente de ese tipo
|
|
typed_entries = sorted(
|
|
[e for e in found_entries if e.filename.endswith(file_ext)],
|
|
key=lambda e: e.st_mtime or 0,
|
|
reverse=True,
|
|
)
|
|
newest = typed_entries[0]
|
|
age_hours = (time.time() - (newest.st_mtime or 0)) / 3600
|
|
|
|
print(f"[i] Backup {backup_type} más reciente: {newest.filename} (hace {age_hours:.1f}h)")
|
|
|
|
if age_hours > SSH_MAX_BACKUP_AGE_HOURS:
|
|
print(f"[!] El backup tiene {age_hours:.1f}h (límite: {SSH_MAX_BACKUP_AGE_HOURS}h). "
|
|
f"Puede estar desactualizado. Descargando de todas formas...")
|
|
|
|
# Descargar vía SFTP
|
|
remote_file_path = f"{remote_path}/{newest.filename}"
|
|
print(f"[*] Descargando por SFTP: {remote_file_path}")
|
|
t0 = time.time()
|
|
with sftp.open(remote_file_path, "rb") as rf:
|
|
content = rf.read()
|
|
elapsed = time.time() - t0
|
|
|
|
sftp.close()
|
|
ssh.close()
|
|
|
|
print(f"[+] Descarga SSH completada en {elapsed:.1f}s — {len(content) / 1024:.1f} KB ({backup_type})")
|
|
return content, file_ext
|
|
|
|
except paramiko.AuthenticationException:
|
|
print("[!] Fallo de autenticación SSH.")
|
|
print(" Verificar SSH_USER y SSH_PASS en la configuración del script.")
|
|
except paramiko.SSHException as e:
|
|
print(f"[!] Error de protocolo SSH: {e}")
|
|
except (TimeoutError, OSError) as e:
|
|
print(f"[!] No se pudo conectar a {UNIFI_HOST}:{SSH_PORT} — {e}")
|
|
print(" Verificar que SSH esté habilitado en el LXC de Proxmox.")
|
|
except Exception as e:
|
|
print(f"[!] Error inesperado en Estrategia SSH: {e}")
|
|
finally:
|
|
try:
|
|
ssh.close()
|
|
except Exception:
|
|
pass
|
|
|
|
return None
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
# ESTRATEGIA 2 — API HTTP (Fallback)
|
|
# Autenticación local por sesión — sin nube, sin UI de Ubiquiti.
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
|
|
def _create_authenticated_session(base_url: str, username: str, password: str) -> requests.Session:
|
|
"""
|
|
Autentica en UniFi OS y retorna la sesión con CSRF token listo.
|
|
UniFi OS 3.x/4.x/5.x requiere el CSRF token en todos los POST.
|
|
"""
|
|
session = requests.Session()
|
|
# User-Agent de navegador para evitar rechazos por agente no reconocido
|
|
session.headers.update({
|
|
"User-Agent": (
|
|
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
|
|
"AppleWebKit/537.36 (KHTML, like Gecko) "
|
|
"Chrome/126.0.0.0 Safari/537.36"
|
|
),
|
|
"Accept": "application/json",
|
|
"Content-Type": "application/json",
|
|
})
|
|
|
|
login_url = f"{base_url}/api/auth/login"
|
|
print(f"[*] Autenticando en UniFi OS: POST {login_url}")
|
|
resp = session.post(
|
|
login_url,
|
|
json={"username": username, "password": password},
|
|
verify=False,
|
|
timeout=15,
|
|
)
|
|
if resp.status_code not in (200, 201):
|
|
raise PermissionError(
|
|
f"Fallo de autenticación en UniFi OS (HTTP {resp.status_code}): {resp.text[:300]}"
|
|
)
|
|
|
|
# Extraer CSRF token — necesario para POST en UniFi OS 3.x/4.x/5.x
|
|
csrf_token = (
|
|
resp.headers.get("X-CSRF-Token")
|
|
or resp.headers.get("x-csrf-token")
|
|
or resp.headers.get("X-Csrf-Token")
|
|
)
|
|
if csrf_token:
|
|
session.headers.update({"X-CSRF-Token": csrf_token})
|
|
print(f"[i] CSRF token obtenido: {csrf_token[:20]}...")
|
|
else:
|
|
print("[i] Sin CSRF token en la respuesta (puede no ser requerido en esta versión).")
|
|
|
|
print("[+] Autenticación por sesión exitosa.")
|
|
return session
|
|
|
|
|
|
def _get_system_info(session: requests.Session, base_url: str) -> tuple[str, str]:
|
|
"""Obtiene nombre y versión del sistema para el nombre del archivo. No crítico."""
|
|
model = "UniFi-OS-Server-5.1.21"
|
|
version = "Network-10.5.67"
|
|
try:
|
|
url = f"{base_url}/proxy/network/api/s/{UNIFI_SITE}/stat/sysinfo"
|
|
res = session.get(url, verify=False, timeout=SYSINFO_TIMEOUT)
|
|
if res.status_code == 200:
|
|
data = res.json().get("data", [{}])[0]
|
|
name = data.get("name", "UniFi-OS-Server")
|
|
ver = data.get("version", "10.5.67")
|
|
model = sanitize_filename(f"UniFi_{name}")
|
|
version = sanitize_filename(f"v{ver}")
|
|
print(f"[+] Sistema: {model} — {version}")
|
|
else:
|
|
print(f"[i] sysinfo retornó HTTP {res.status_code}. Usando valores por defecto.")
|
|
except Exception as e:
|
|
print(f"[i] No se pudo obtener sysinfo: {e}. Usando valores por defecto.")
|
|
return model, version
|
|
|
|
|
|
def _try_create_os_server_backup(session: requests.Session, base_url: str) -> bool:
|
|
"""
|
|
Solicita al OS Server que cree un nuevo backup (.unifi).
|
|
POST /api/backup — el OS Server genera el archivo y lo deja disponible
|
|
para descargar con GET /api/backup/download.
|
|
Retorna True si el trigger fue exitoso, False si falló.
|
|
"""
|
|
url = f"{base_url}/api/backup"
|
|
print(f"[*] Solicitando creación de backup OS Server: POST {url}")
|
|
try:
|
|
t0 = time.time()
|
|
resp = session.post(url, json={}, verify=False, timeout=BACKUP_CREATE_TIMEOUT)
|
|
elapsed = time.time() - t0
|
|
print(f"[i] Respuesta en {elapsed:.1f}s — HTTP {resp.status_code}")
|
|
if resp.status_code in (200, 201, 202):
|
|
print(f"[+] Backup OS Server solicitado correctamente.")
|
|
if elapsed < 5:
|
|
# El servidor aceptó rápido: esperar que termine de generarlo
|
|
print("[*] Esperando 10s para que el OS Server genere el archivo...")
|
|
time.sleep(10)
|
|
return True
|
|
elif resp.status_code == 403:
|
|
print("[!] HTTP 403 en POST /api/backup — permisos insuficientes.")
|
|
elif resp.status_code == 404:
|
|
print("[i] POST /api/backup no existe en esta versión. Continuando con descarga directa.")
|
|
else:
|
|
print(f"[!] HTTP {resp.status_code} al crear backup: {resp.text[:200]}")
|
|
except requests.exceptions.Timeout:
|
|
# Timeout puede ser normal si el servidor tardó en generar el backup
|
|
print(f"[!] Timeout esperando respuesta de POST /api/backup. El backup puede haberse generado.")
|
|
return True # Intentar descarga de todas formas
|
|
except Exception as e:
|
|
print(f"[!] Error en POST /api/backup: {e}")
|
|
return False
|
|
|
|
|
|
def _try_direct_download(session: requests.Session, base_url: str) -> bytes | None:
|
|
"""
|
|
Intenta GET /api/backup/download — descarga el último backup sin generar uno nuevo.
|
|
Este endpoint descarga el archivo existente y no sufre el timeout silencioso de /cmd/backup.
|
|
"""
|
|
url = f"{base_url}/api/backup/download"
|
|
print(f"[*] Intentando descarga directa: GET {url}")
|
|
try:
|
|
t0 = time.time()
|
|
resp = session.get(url, verify=False, timeout=DOWNLOAD_TIMEOUT, stream=True)
|
|
elapsed = time.time() - t0
|
|
print(f"[i] Respuesta en {elapsed:.1f}s — HTTP {resp.status_code}")
|
|
|
|
if resp.status_code == 200:
|
|
content = resp.content
|
|
# Verificar que sea binario (.unf), no un JSON de error
|
|
if len(content) > 1024 and not content.lstrip().startswith(b"{"):
|
|
print(f"[+] Descarga directa exitosa — {len(content) / 1024:.1f} KB")
|
|
return content
|
|
else:
|
|
print(f"[i] La respuesta parece JSON/error, no un archivo binario: {content[:150]}")
|
|
elif resp.status_code == 404:
|
|
print("[i] Endpoint /api/backup/download no existe en esta versión de UniFi OS.")
|
|
elif resp.status_code == 403:
|
|
print("[!] HTTP 403 en /api/backup/download — permisos insuficientes.")
|
|
else:
|
|
print(f"[!] HTTP {resp.status_code} en /api/backup/download.")
|
|
except requests.exceptions.Timeout:
|
|
print("[!] Timeout esperando /api/backup/download.")
|
|
except Exception as e:
|
|
print(f"[!] Error en /api/backup/download: {e}")
|
|
return None
|
|
|
|
|
|
def _try_cmd_backup(session: requests.Session, base_url: str) -> tuple[bytes, str] | None:
|
|
"""
|
|
Último recurso: endpoint clásico /cmd/backup.
|
|
En UniFi Network 10.5.x puede funcionar si los permisos son correctos.
|
|
Timeout reducido a BACKUP_CMD_TIMEOUT[1]s — si tarda más, es fallo silencioso.
|
|
"""
|
|
url = f"{base_url}/proxy/network/api/s/{UNIFI_SITE}/cmd/backup"
|
|
print(f"[*] Intentando /cmd/backup (timeout: {BACKUP_CMD_TIMEOUT[1]}s): POST {url}")
|
|
|
|
try:
|
|
t0 = time.time()
|
|
resp = session.post(
|
|
url,
|
|
json={"cmd": "backup", "days": 0},
|
|
verify=False,
|
|
timeout=BACKUP_CMD_TIMEOUT,
|
|
)
|
|
elapsed = time.time() - t0
|
|
print(f"[i] Respuesta en {elapsed:.1f}s — HTTP {resp.status_code}")
|
|
|
|
if resp.status_code == 200:
|
|
try:
|
|
res_json = resp.json()
|
|
data_list = res_json.get("data", [])
|
|
if data_list and "url" in data_list[0]:
|
|
relative_url = data_list[0]["url"]
|
|
download_url = f"{base_url}{relative_url}"
|
|
ext = ".unifi" if relative_url.endswith(".unifi") else ".unf"
|
|
print(f"[+] Backup generado por /cmd/backup: {relative_url}")
|
|
print("[*] Descargando archivo generado...")
|
|
t1 = time.time()
|
|
dl = session.get(download_url, verify=False, timeout=DOWNLOAD_TIMEOUT)
|
|
print(f"[i] Descarga en {time.time() - t1:.1f}s — HTTP {dl.status_code}")
|
|
if dl.status_code == 200:
|
|
print(f"[+] /cmd/backup exitoso — {len(dl.content) / 1024:.1f} KB")
|
|
return dl.content, ext
|
|
else:
|
|
print(f"[!] Respuesta inesperada de /cmd/backup: {res_json}")
|
|
except Exception as e:
|
|
print(f"[!] Error procesando respuesta de /cmd/backup: {e}")
|
|
elif resp.status_code == 403:
|
|
print("[!] HTTP 403 en /cmd/backup — el usuario necesita 'Full Management' en Network.")
|
|
else:
|
|
print(f"[!] HTTP {resp.status_code} en /cmd/backup: {resp.text[:200]}")
|
|
|
|
except requests.exceptions.ReadTimeout:
|
|
print(f"[!] /cmd/backup no respondió en {BACKUP_CMD_TIMEOUT[1]}s (fallo silencioso conocido).")
|
|
print(" → Habilitar SSH en el LXC para que la Estrategia 1 funcione.")
|
|
except requests.exceptions.ConnectionError as e:
|
|
print(f"[!] Error de conexión en /cmd/backup: {e}")
|
|
except Exception as e:
|
|
print(f"[!] Error inesperado en /cmd/backup: {e}")
|
|
|
|
return None
|
|
|
|
|
|
def backup_via_api() -> tuple[bytes, str, str, str] | None:
|
|
"""
|
|
Estrategia 2: backup vía API HTTP local (sin nube).
|
|
|
|
Prueba en este orden:
|
|
[OS] POST /api/backup → trigger creación backup OS Server (.unifi)
|
|
GET /api/backup/download → descarga el .unifi generado
|
|
[A] GET /api/backup/download → descarga el último .unifi disponible (sin trigger)
|
|
[B] POST /proxy/network/.../cmd/backup → backup Network App (.unf) vía proxy
|
|
[C] Puerto 8443 directo → /cmd/backup sin proxy (Network App)
|
|
|
|
Referencia: instalador línea 5534: WEB_PORT en /var/lib/uosserver/server.conf → 11443
|
|
"""
|
|
model, version = "UniFi-OS-Server-5.1.21", "Network-10.5.67"
|
|
|
|
# ── Autenticación única para todos los intentos vía 11443 ───────────────
|
|
print(f"[*] Autenticando en OS Server: {UNIFI_BASE_URL}")
|
|
try:
|
|
session = _create_authenticated_session(UNIFI_BASE_URL, UNIFI_USER, UNIFI_PASS)
|
|
model, version = _get_system_info(session, UNIFI_BASE_URL)
|
|
except PermissionError as e:
|
|
print(f"[!] Autenticación fallida: {e}")
|
|
return None
|
|
except Exception as e:
|
|
print(f"[!] No se pudo autenticar: {e}")
|
|
return None
|
|
|
|
# ── [OS] Intentar crear + descargar backup del OS Server (.unifi) ────────
|
|
print("\n[*] [OS] Intentando backup del OS Server (.unifi)...")
|
|
triggered = _try_create_os_server_backup(session, UNIFI_BASE_URL)
|
|
if triggered:
|
|
content = _try_direct_download(session, UNIFI_BASE_URL)
|
|
if content:
|
|
print("[+] [OS] Backup OS Server (.unifi) obtenido correctamente.")
|
|
return content, ".unifi", model, version
|
|
print("[!] [OS] Trigger aceptado pero descarga falló. Continuando...")
|
|
|
|
# ── [A] Intentar descarga directa del último backup disponible ───────────
|
|
print("\n[*] [A] Descarga directa del último backup disponible...")
|
|
content = _try_direct_download(session, UNIFI_BASE_URL)
|
|
if content:
|
|
# Determinar extensión por el contenido
|
|
ext = ".unifi" if b"unifi_os_backup" in content[:200] else ".unf"
|
|
print(f"[+] [A] Backup descargado directamente ({ext}).")
|
|
return content, ext, model, version
|
|
|
|
# ── [B] Fallback: backup Network App vía proxy (puerto 11443) ───────────
|
|
print("\n[*] [B] Intentando backup Network App vía proxy (puerto 11443)...")
|
|
result = _try_cmd_backup(session, UNIFI_BASE_URL)
|
|
if result:
|
|
content, ext = result
|
|
return content, ext, model, version
|
|
|
|
# ── [C] Fallback: Network App directa (puerto 8443) ─────────────────────
|
|
print(f"\n[*] [C] Intentando Network App directa: {UNIFI_NETWORK_URL}")
|
|
for login_path in ["/api/auth/login", "/api/login"]:
|
|
try:
|
|
session_c = requests.Session()
|
|
session_c.headers.update({
|
|
"User-Agent": (
|
|
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
|
|
"AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36"
|
|
),
|
|
"Accept": "application/json",
|
|
"Content-Type": "application/json",
|
|
})
|
|
resp_login = session_c.post(
|
|
f"{UNIFI_NETWORK_URL}{login_path}",
|
|
json={"username": UNIFI_USER, "password": UNIFI_PASS},
|
|
verify=False, timeout=15,
|
|
)
|
|
if resp_login.status_code not in (200, 201):
|
|
continue
|
|
csrf = resp_login.headers.get("X-CSRF-Token") or resp_login.headers.get("x-csrf-token")
|
|
if csrf:
|
|
session_c.headers.update({"X-CSRF-Token": csrf})
|
|
print(f"[+] [C] Autenticación exitosa en {login_path}")
|
|
|
|
result = _try_cmd_backup(session_c, UNIFI_NETWORK_URL)
|
|
if result:
|
|
content, ext = result
|
|
return content, ext, model, version
|
|
|
|
content = _try_direct_download(session_c, UNIFI_NETWORK_URL)
|
|
if content:
|
|
return content, ".unifi", model, version
|
|
break
|
|
except Exception as e:
|
|
print(f"[i] [C] Error con {login_path}: {e}")
|
|
continue
|
|
|
|
print("[!] [Estrategia 2 — API] Todos los intentos fallaron.")
|
|
return None
|
|
|
|
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
# MAIN
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
|
|
def main():
|
|
print("=" * 60)
|
|
print(" RESPALDO UNIFI OS SERVER 5.1.21 / NETWORK 10.5.67")
|
|
print(f" LXC Proxmox — {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
|
print("=" * 60)
|
|
|
|
# ── 1. Acceso al NAS ────────────────────────────────────────────────────
|
|
if not authenticate_nas_share(NAS_PATH, NAS_USER, NAS_PASS):
|
|
print(f"[ERROR CRÍTICO] No se puede acceder a la ruta destino: {NAS_PATH}")
|
|
sys.exit(1)
|
|
|
|
backup_content: bytes | None = None
|
|
file_ext = ".unf"
|
|
model = "UniFi-OS-Server-5.1.21"
|
|
version = "Network-10.5.67"
|
|
|
|
# ── 2. Estrategia 1: SSH / SFTP ─────────────────────────────────────────
|
|
_sep("Estrategia 1: SSH / SFTP (principal)")
|
|
result_ssh = backup_via_ssh()
|
|
if result_ssh:
|
|
backup_content, file_ext = result_ssh
|
|
print("[+] Backup obtenido por SSH exitosamente.")
|
|
else:
|
|
print("[!] Estrategia 1 (SSH) no disponible o sin autobackups. Continuando...")
|
|
|
|
# ── 3. Estrategia 2: API HTTP ────────────────────────────────────────────
|
|
if backup_content is None:
|
|
_sep("Estrategia 2: API HTTP (fallback)")
|
|
result_api = backup_via_api()
|
|
if result_api:
|
|
backup_content, file_ext, model, version = result_api
|
|
print("[+] Backup obtenido por API exitosamente.")
|
|
else:
|
|
print("[!] Estrategia 2 (API) también falló.")
|
|
|
|
# ── 4. Verificar que tenemos contenido ──────────────────────────────────
|
|
if backup_content is None:
|
|
print()
|
|
print("=" * 60)
|
|
print("[ERROR CRÍTICO] RESPALDO FALLIDO — Ninguna estrategia tuvo éxito.")
|
|
print()
|
|
print(" Pasos para resolver:")
|
|
print()
|
|
print(" [SSH] 1. Habilitar SSH en el LXC de Proxmox (si no está activo)")
|
|
print(" y asegurarse que SSH_PASS en este script sea correcto.")
|
|
print()
|
|
print(" [SSH] 2. Habilitar autobackups en UniFi UI:")
|
|
print(" Settings → System → Backups → Auto Backup → ON")
|
|
print(" Esperar a que genere el primer archivo .unf.")
|
|
print()
|
|
print(" [API] 3. Verificar permisos del usuario admin:")
|
|
print(" Settings → Admins & Users → admin")
|
|
print(" → Network: Full Management (no solo View)")
|
|
print("=" * 60)
|
|
sys.exit(1)
|
|
|
|
# ── 5. Guardar en NAS ───────────────────────────────────────────────────
|
|
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
|
|
# Prefijo según tipo de backup: OS Server (.unifi) o Network App (.unf)
|
|
if file_ext == ".unifi":
|
|
prefix = "UniFi_OS_Server"
|
|
else:
|
|
prefix = "UniFi_Network_App"
|
|
|
|
safe_version = sanitize_filename(version)
|
|
filename = f"{prefix}_{safe_version}_{timestamp}{file_ext}"
|
|
destination_file = os.path.join(NAS_PATH, filename)
|
|
|
|
print(f"\n[*] Guardando en NAS: {filename}")
|
|
try:
|
|
with open(destination_file, "wb") as f:
|
|
f.write(backup_content)
|
|
size_kb = len(backup_content) / 1024
|
|
print()
|
|
print("=" * 60)
|
|
print("[ÉXITO] RESPALDO COMPLETADO")
|
|
print(f" Ruta : {destination_file}")
|
|
print(f" Tamaño: {size_kb:.2f} KB")
|
|
print("=" * 60)
|
|
except Exception as e:
|
|
print(f"[ERROR CRÍTICO] Falló la escritura del archivo en el NAS: {e}")
|
|
sys.exit(1)
|
|
|
|
# ── 6. Limpieza por retención ────────────────────────────────────────────
|
|
cleanup_old_backups(NAS_PATH, RETENTION_DAYS)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main() |