434 lines
13 KiB
Python
434 lines
13 KiB
Python
"""Consultas seguras al ASUS RT-AX88U PRO desde la Raspberry."""
|
|
|
|
import html
|
|
import re
|
|
import shlex
|
|
import subprocess
|
|
import sys
|
|
|
|
|
|
ROUTER = "jjminguez@192.168.50.1"
|
|
KEY = "/home/pi/.ssh/router_control"
|
|
SSH_BASE = [
|
|
"ssh",
|
|
"-i", KEY,
|
|
"-o", "BatchMode=yes",
|
|
"-o", "ConnectTimeout=8",
|
|
ROUTER,
|
|
]
|
|
|
|
# Reservas que ya existian antes de reconstruir la lista desde los leases.
|
|
LEGACY_RESERVATIONS = {
|
|
"82:CC:3C:7D:0F:41": "192.168.50.194",
|
|
"72:2A:0C:FE:5F:04": "192.168.50.158",
|
|
"E0:1F:FC:A9:DC:73": "192.168.50.107",
|
|
}
|
|
|
|
|
|
def _run(command):
|
|
try:
|
|
result = subprocess.run(
|
|
SSH_BASE + [command],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=15,
|
|
)
|
|
except Exception as exc:
|
|
return f"Error conectando al router: {exc}"
|
|
if result.returncode != 0:
|
|
return f"Error del router: {result.stderr.strip()[-300:]}"
|
|
return result.stdout.strip()
|
|
|
|
|
|
def estado():
|
|
output = _run(
|
|
"printf '%s\\n' "
|
|
"'Modelo='$(nvram get productid) "
|
|
"'Firmware='$(nvram get firmver).$(nvram get buildno) "
|
|
"'LAN='$(nvram get lan_ipaddr) "
|
|
"'WAN='$(nvram get wan0_ipaddr) "
|
|
"'WiFi24='$(nvram get wl0_radio) "
|
|
"'WiFi5='$(nvram get wl1_radio); uptime"
|
|
)
|
|
return f"<pre>{html.escape(output)}</pre>"
|
|
|
|
|
|
def wifi():
|
|
output = _run(
|
|
"printf '%s\\n' "
|
|
"'2.4GHz='$(nvram get wl0_radio) "
|
|
"'5GHz='$(nvram get wl1_radio) "
|
|
"'6GHz='$(nvram get wl2_radio)"
|
|
)
|
|
return f"<pre>{html.escape(output)}</pre>"
|
|
|
|
|
|
def clientes():
|
|
output = _run("cat /proc/net/arp")
|
|
return f"<pre>{html.escape(output)}</pre>"
|
|
|
|
|
|
def _reservas_actuales(text):
|
|
return re.findall(r"<([^>]+)>([^>]+)>>", text or "")
|
|
|
|
|
|
def _inventario():
|
|
leases = _run("cat /var/lib/misc/dnsmasq.leases")
|
|
reservas = _reservas_actuales(_run("nvram get dhcp_staticlist"))
|
|
by_mac = {mac.upper(): ip for mac, ip in reservas}
|
|
for mac, ip in LEGACY_RESERVATIONS.items():
|
|
by_mac.setdefault(mac, ip)
|
|
devices = []
|
|
for line in leases.splitlines():
|
|
fields = line.split()
|
|
if len(fields) < 4 or not fields[2].startswith("192.168.50."):
|
|
continue
|
|
mac, ip, name = fields[1].upper(), fields[2], fields[3]
|
|
by_mac.setdefault(mac, ip)
|
|
devices.append((ip, mac, name if name != "*" else "pendiente-identificar"))
|
|
devices.sort(key=lambda item: int(item[0].rsplit(".", 1)[1]))
|
|
return devices, by_mac
|
|
|
|
|
|
def inventario():
|
|
devices, _by_mac = _inventario()
|
|
lines = ["IP MAC NOMBRE"]
|
|
lines.extend(f"{ip:<18} {mac:<22} {name}" for ip, mac, name in devices)
|
|
return f"<pre>{html.escape(chr(10).join(lines))}</pre>"
|
|
|
|
|
|
def reservar_dhcp():
|
|
devices, by_mac = _inventario()
|
|
payload = "".join(f"<{mac}>{ip}>>" for mac, ip in by_mac.items())
|
|
command = (
|
|
f"nvram set dhcp_staticlist={shlex.quote(payload)}; "
|
|
"nvram commit; service restart_dnsmasq"
|
|
)
|
|
result = _run(command)
|
|
if result.startswith("Error "):
|
|
return result
|
|
return f"Reservas DHCP aplicadas: {len(by_mac)} dispositivos. Leases inventariados: {len(devices)}."
|
|
|
|
|
|
def reservas():
|
|
output = _run("nvram get dhcp_staticlist")
|
|
return f"<pre>{html.escape(output)}</pre>"
|
|
|
|
|
|
def control_parental():
|
|
try:
|
|
from asus_api_control import parental_status
|
|
return f"<pre>{html.escape(parental_status())}</pre>"
|
|
except RuntimeError as exc:
|
|
return f"<pre>{html.escape(str(exc))}</pre>"
|
|
except Exception:
|
|
pass
|
|
keys = (
|
|
"yadns_enable_x",
|
|
"yadns_mode",
|
|
"yadns_clientlist",
|
|
"bwdpi_enable",
|
|
"bwdpi_rulelist",
|
|
"keyword_rulelist",
|
|
"url_rulelist",
|
|
"time_zone",
|
|
)
|
|
lines = []
|
|
for key in keys:
|
|
lines.append(f"{key}={_run(f'nvram get {key}')}")
|
|
return f"<pre>{html.escape(chr(10).join(lines))}</pre>"
|
|
|
|
|
|
def reiniciar():
|
|
return _run("reboot")
|
|
|
|
|
|
QOS_TIPOS = {
|
|
"0": "Traditional (por prioridades)",
|
|
"1": "Adaptive (auto gaming/streaming)",
|
|
"2": "Bandwidth limiter",
|
|
"3": "GeForce Now",
|
|
}
|
|
|
|
|
|
def _set_nvram(entradas, servicio=None):
|
|
partes = [
|
|
f"nvram set {clave}={shlex.quote(str(valor))}"
|
|
for clave, valor in entradas
|
|
]
|
|
partes.append("nvram commit")
|
|
if servicio:
|
|
partes.append(f"service {servicio}")
|
|
return _run("; ".join(partes))
|
|
|
|
|
|
def _nombre_por_mac(mac):
|
|
mac = mac.upper()
|
|
try:
|
|
from asus_api_control import DEVICE_ALIASES
|
|
for _alias, info in DEVICE_ALIASES.items():
|
|
if info["mac"].upper() == mac:
|
|
return info["name"]
|
|
except Exception:
|
|
pass
|
|
_devices, _by_mac = _inventario()
|
|
for ip, m, nombre in _devices:
|
|
if m == mac and nombre != "pendiente-identificar":
|
|
return nombre
|
|
return "pendiente-identificar"
|
|
|
|
|
|
def _resolver_mac(dispositivo):
|
|
valor = str(dispositivo or "").strip()
|
|
if valor.lower().count(":") == 5:
|
|
return valor.upper()
|
|
try:
|
|
from asus_api_control import normalizar_nombre, resolver_dispositivo
|
|
except Exception:
|
|
raise ValueError(f"Dispositivo no reconocido: {dispositivo}")
|
|
info = resolver_dispositivo(valor)
|
|
if info:
|
|
return info["mac"].upper()
|
|
_devices, _by_mac = _inventario()
|
|
clave = normalizar_nombre(valor)
|
|
for ip, m, nombre in _devices:
|
|
n = normalizar_nombre(nombre)
|
|
if clave and (n == clave or clave in n or n in clave):
|
|
return m
|
|
raise ValueError(
|
|
f"Dispositivo no reconocido: {dispositivo}. Usa MAC o un alias."
|
|
)
|
|
|
|
|
|
def qos_estado():
|
|
enable = (_run("nvram get qos_enable") or "").strip()
|
|
tipo = (_run("nvram get qos_type") or "").strip()
|
|
activado = "sí" if enable == "1" else "no"
|
|
modo = QOS_TIPOS.get(tipo, f"desconocido ({tipo})")
|
|
texto = f"Activado: {activado}\nModo: {modo}"
|
|
return f"<pre>{html.escape(texto)}</pre>"
|
|
|
|
|
|
def qos_set(modo):
|
|
conf = {
|
|
"adaptive": ("1", "1"),
|
|
"tradicional": ("0", "0"),
|
|
}
|
|
if modo not in conf:
|
|
raise ValueError("Modo no válido: usa adaptive o tradicional")
|
|
qos_enable, qos_type = conf[modo]
|
|
resultado = _set_nvram(
|
|
[("qos_enable", qos_enable), ("qos_type", qos_type)], "restart_qos"
|
|
)
|
|
if resultado.startswith("Error "):
|
|
return resultado
|
|
nombre = QOS_TIPOS[qos_type]
|
|
return (
|
|
f"QoS activado en modo <b>{nombre}</b>. "
|
|
"Ten en cuenta que QoS desactiva la aceleración NAT."
|
|
)
|
|
|
|
|
|
def qos_apagar():
|
|
resultado = _set_nvram([("qos_enable", "0")], "restart_qos")
|
|
if resultado.startswith("Error "):
|
|
return resultado
|
|
return "QoS desactivado. Aceleración NAT restaurada."
|
|
|
|
|
|
def _limite_parse():
|
|
raw = _run("nvram get qos_bw_rulelist") or ""
|
|
reglas = []
|
|
for parte in str(raw).split("<"):
|
|
parte = parte.strip()
|
|
if not parte:
|
|
continue
|
|
campos = parte.rstrip(">").split(">")
|
|
if len(campos) < 4:
|
|
continue
|
|
en, mac, dl, ul = campos[:4]
|
|
if str(mac).count(":") != 5:
|
|
continue
|
|
reglas.append({
|
|
"enable": en,
|
|
"mac": mac.upper(),
|
|
"dl": dl,
|
|
"ul": ul,
|
|
})
|
|
return raw, reglas
|
|
|
|
|
|
def _limite_payload(reglas):
|
|
partes = [
|
|
f"{r['enable']}>{r['mac']}>{r['dl']}>{r['ul']}>{idx}"
|
|
for idx, r in enumerate(reglas)
|
|
]
|
|
return "<".join(partes)
|
|
|
|
|
|
def _limite_guardar(reglas):
|
|
payload = _limite_payload(reglas)
|
|
if reglas:
|
|
entradas = [
|
|
("qos_enable", "1"),
|
|
("qos_type", "2"),
|
|
("qos_bw_rulelist", payload),
|
|
]
|
|
else:
|
|
entradas = [
|
|
("qos_enable", "0"),
|
|
("qos_bw_rulelist", payload),
|
|
]
|
|
return _set_nvram(entradas, "restart_qos")
|
|
|
|
|
|
def limite_listar():
|
|
_raw, reglas = _limite_parse()
|
|
if not reglas:
|
|
return "<pre>No hay reglas de límite activas.</pre>"
|
|
lineas = ["ACTIVADO DISPOSITIVO/MAC BAJADA SUBIDA"]
|
|
for r in reglas:
|
|
nombre = _nombre_por_mac(r["mac"])
|
|
dl = f"{int(r['dl']) / 1024:.0f}" if r["dl"].isdigit() else r["dl"]
|
|
ul = f"{int(r['ul']) / 1024:.0f}" if r["ul"].isdigit() else r["ul"]
|
|
lineas.append(
|
|
f"{'sí' if r['enable'] == '1' else 'no':<9} "
|
|
f"{nombre:<20} {dl:>7} {ul:>7} Mbps"
|
|
)
|
|
return f"<pre>{html.escape(chr(10).join(lineas))}</pre>"
|
|
|
|
|
|
def limite_poner(dispositivo, dl_mbps, ul_mbps):
|
|
mac = _resolver_mac(dispositivo)
|
|
if dl_mbps <= 0 or ul_mbps <= 0:
|
|
raise ValueError("Las velocidades deben ser mayores que 0")
|
|
_raw, reglas = _limite_parse()
|
|
reglas = [r for r in reglas if r["mac"] != mac]
|
|
reglas.append({
|
|
"enable": "1",
|
|
"mac": mac,
|
|
"dl": str(int(dl_mbps * 1024)),
|
|
"ul": str(int(ul_mbps * 1024)),
|
|
})
|
|
resultado = _limite_guardar(reglas)
|
|
if resultado.startswith("Error "):
|
|
return resultado
|
|
return (
|
|
f"Límite aplicado a <b>{_nombre_por_mac(mac)}</b>: "
|
|
f"{dl_mbps:.0f} Mbps bajada / {ul_mbps:.0f} Mbps subida. "
|
|
"El bandwidth limiter desactiva la aceleración NAT."
|
|
)
|
|
|
|
|
|
def limite_quitar(dispositivo):
|
|
mac = _resolver_mac(dispositivo)
|
|
_raw, reglas = _limite_parse()
|
|
if not any(r["mac"] == mac for r in reglas):
|
|
return f"No hay límite activo para {dispositivo}."
|
|
reglas = [r for r in reglas if r["mac"] != mac]
|
|
resultado = _limite_guardar(reglas)
|
|
if resultado.startswith("Error "):
|
|
return resultado
|
|
if not reglas:
|
|
return f"Límite retirado de <b>{_nombre_por_mac(mac)}</b>. "
|
|
return f"Límite retirado de <b>{_nombre_por_mac(mac)}</b>."
|
|
|
|
|
|
def seguridad_estado():
|
|
import json as _json
|
|
|
|
out = _run(
|
|
"printf '%s\\n' "
|
|
"'dnsfilter='$(nvram get dnsfilter_enable_x) "
|
|
"'keywords='$(nvram get keyword_rulelist)"
|
|
)
|
|
lineas = []
|
|
parental = None
|
|
try:
|
|
from asus_api_control import parental_status
|
|
parental = _json.loads(parental_status())
|
|
except Exception:
|
|
parental = None
|
|
if parental is not None:
|
|
estado_p = "activo" if parental.get("state") else "inactivo"
|
|
nreglas = len(parental.get("rules") or {})
|
|
lineas.append(
|
|
f"Control parental: {estado_p} · "
|
|
f"{nreglas} dispositivos con regla"
|
|
)
|
|
if parental.get("block_all"):
|
|
lineas.append("Bloqueo total de Internet: ACTIVADO")
|
|
else:
|
|
lineas.append("Control parental: no consultable")
|
|
for campo in out.splitlines():
|
|
clave, _, valor = campo.partition("=")
|
|
if clave == "dnsfilter":
|
|
if valor == "1":
|
|
lineas.append("Filtro DNS: activado")
|
|
elif valor == "0":
|
|
lineas.append("Filtro DNS: desactivado")
|
|
else:
|
|
lineas.append(f"Filtro DNS: estado {valor!r}")
|
|
elif clave == "keywords":
|
|
palabras = [p for p in str(valor).split("<") if p.strip()]
|
|
lineas.append(
|
|
"Palabras filtradas: "
|
|
+ (", ".join(palabras) if palabras else "(ninguna)")
|
|
)
|
|
lineas.append(
|
|
"AiProtection Trend Micro: no activado "
|
|
"(requiere activacion desde la web del router)"
|
|
)
|
|
return f"<pre>{html.escape(chr(10).join(lineas))}</pre>"
|
|
|
|
|
|
def _keywords_actuales():
|
|
raw = _run("nvram get keyword_rulelist") or ""
|
|
return [p.strip().lower() for p in str(raw).split("<") if p.strip()]
|
|
|
|
|
|
def seguridad_keyword(palabra, activar):
|
|
palabra = palabra.strip().lower()
|
|
palabras = _keywords_actuales()
|
|
if activar:
|
|
if palabra in palabras:
|
|
return f"La palabra <b>{palabra}</b> ya estaba en el filtro."
|
|
palabras.append(palabra)
|
|
else:
|
|
if palabra not in palabras:
|
|
return f"La palabra <b>{palabra}</b> no estaba en el filtro."
|
|
palabras = [p for p in palabras if p != palabra]
|
|
payload = "".join(f"<{p}>" for p in palabras)
|
|
entradas = [("keyword_rulelist", payload)]
|
|
if palabras:
|
|
entradas.insert(0, ("bwdpi_enable", "1"))
|
|
resultado = _set_nvram(entradas, "restart_bwdpi")
|
|
if resultado.startswith("Error "):
|
|
return resultado
|
|
accion = "bloqueada" if activar else "desbloqueada"
|
|
if palabras:
|
|
resto = f" Filtro activo: {', '.join(palabras)}."
|
|
else:
|
|
resto = " Filtro de contenidos vacío."
|
|
return f"Palabra <b>{palabra}</b> {accion}.{resto}"
|
|
|
|
|
|
if __name__ == "__main__":
|
|
action = sys.argv[1] if len(sys.argv) > 1 else "estado"
|
|
actions = {
|
|
"estado": estado,
|
|
"clientes": clientes,
|
|
"wifi": wifi,
|
|
"inventario": inventario,
|
|
"reservar": reservar_dhcp,
|
|
"reservas": reservas,
|
|
"parental": control_parental,
|
|
"qos_estado": qos_estado,
|
|
"qos_adaptive": lambda: qos_set("adaptive"),
|
|
"qos_tradicional": lambda: qos_set("tradicional"),
|
|
"qos_apagar": qos_apagar,
|
|
"limite_listar": limite_listar,
|
|
"seguridad_estado": seguridad_estado,
|
|
}
|
|
print(actions.get(action, estado)())
|