05-sep: [vigilancia] comandos /aiMesh /qos /limite /seguridad para el ASUS desde Telegram
This commit is contained in:
juanjo
@@ -143,6 +143,245 @@ 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):
|
||||
_devices, _by_mac = _inventario()
|
||||
for ip, m, nombre in _devices:
|
||||
if m == mac.upper():
|
||||
return nombre
|
||||
try:
|
||||
from asus_api_control import DEVICE_ALIASES
|
||||
for _alias, info in DEVICE_ALIASES.items():
|
||||
if info["mac"].upper() == mac.upper():
|
||||
return info["name"]
|
||||
except Exception:
|
||||
pass
|
||||
return f"unknown-{mac}"
|
||||
|
||||
|
||||
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():
|
||||
claves = (
|
||||
"bwdpi_enable",
|
||||
"yadns_enable_x",
|
||||
"yadns_mode",
|
||||
)
|
||||
lineas = []
|
||||
for clave in claves:
|
||||
lineas.append(f"{clave}={_run(f'nvram get {clave}')}")
|
||||
keywords = _run("nvram get keyword_rulelist") or ""
|
||||
palabras = [p for p in str(keywords).split("<") if p.strip()]
|
||||
lineas.append(
|
||||
"palabras filtradas: "
|
||||
+ (", ".join(palabras) if palabras else "(ninguna)")
|
||||
)
|
||||
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 = {
|
||||
@@ -153,5 +392,11 @@ if __name__ == "__main__":
|
||||
"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)())
|
||||
|
||||
Reference in New Issue
Block a user