mirror of
https://github.com/andrey271192/keenetic-dns-routes.git
synced 2026-09-20 14:42:01 +00:00
323 lines
14 KiB
Python
323 lines
14 KiB
Python
"""Keenetic NDMS RCI — авторизация как в gokeenapi / Keenetic Unified."""
|
||
from __future__ import annotations
|
||
|
||
import hashlib
|
||
import logging
|
||
import re
|
||
from typing import Any
|
||
from urllib.parse import urlparse
|
||
|
||
import httpx
|
||
|
||
logger = logging.getLogger("kdns.rci")
|
||
|
||
_MAX_PARSE = 90 # ниже лимита gokeenapi (100), с запасом под save
|
||
|
||
|
||
def _norm_lines(lines: list[str]) -> list[str]:
|
||
seen: set[str] = set()
|
||
out: list[str] = []
|
||
for raw in lines:
|
||
s = raw.strip()
|
||
if not s or s.startswith("#"):
|
||
continue
|
||
if s in seen:
|
||
continue
|
||
seen.add(s)
|
||
out.append(s)
|
||
return out
|
||
|
||
|
||
def _is_ipish(s: str) -> bool:
|
||
if re.match(r"^\d{1,3}(\.\d{1,3}){3}(/\d+)?$", s):
|
||
return True
|
||
if "/" in s and re.match(r"^[0-9a-fA-F:.]+/\d+$", s):
|
||
return True
|
||
if re.match(r"^\d{1,3}(\.\d{1,3}){3}-\d{1,3}(\.\d{1,3}){3}$", s):
|
||
return True
|
||
return False
|
||
|
||
|
||
def _valid_entry(s: str) -> bool:
|
||
if _is_ipish(s):
|
||
return True
|
||
if "." in s and not s.startswith(".") and ".." not in s:
|
||
return True
|
||
return False
|
||
|
||
|
||
class KeeneticRCIError(RuntimeError):
|
||
pass
|
||
|
||
|
||
class KeeneticRCI:
|
||
def __init__(self, base_url: str, login: str, password: str):
|
||
bu = base_url.rstrip("/")
|
||
# user:pass@ в base_url ломает httpx (дубли с NDMS-auth) и даёт 500 / странные ответы
|
||
rest = bu.split("://", 1)[-1] if "://" in bu else bu
|
||
if "@" in rest:
|
||
raise KeeneticRCIError(
|
||
"В base URL не должно быть user:pass@ — сохрани роутер ещё раз "
|
||
"(логин/пароль только в полях или перенесутся из URL при сохранении)."
|
||
)
|
||
self.base_url = bu
|
||
self.login = login
|
||
self.password = password
|
||
self._client: httpx.Client | None = None
|
||
|
||
def _client_ctx(self) -> httpx.Client:
|
||
return httpx.Client(
|
||
base_url=self.base_url,
|
||
verify=False,
|
||
timeout=httpx.Timeout(60.0),
|
||
follow_redirects=True,
|
||
)
|
||
|
||
def _auth(self, client: httpx.Client) -> None:
|
||
r = client.get("/auth")
|
||
if r.status_code == 200:
|
||
return
|
||
if r.status_code == 404:
|
||
raise KeeneticRCIError(
|
||
"/auth HTTP 404: на этом адресе нет NDMS /auth — проверь хост и порт прокси "
|
||
"(часто нужен явный порт, например :81 или :443 для https), без лишнего пути в base URL."
|
||
)
|
||
# 401 — стандартный challenge; 403 иногда даёт прокси до входа, но с теми же заголовками
|
||
if r.status_code not in (401, 403):
|
||
raise KeeneticRCIError(f"/auth HTTP {r.status_code}")
|
||
realm = r.headers.get("X-NDM-Realm", "") or r.headers.get("x-ndm-realm", "")
|
||
challenge = r.headers.get("X-NDM-Challenge", "") or r.headers.get("x-ndm-challenge", "")
|
||
set_cookie = r.headers.get("Set-Cookie") or r.headers.get("set-cookie") or ""
|
||
cookie_pair = set_cookie.split(";")[0].strip()
|
||
if not realm or not challenge or not cookie_pair:
|
||
if r.status_code == 403:
|
||
raise KeeneticRCIError(
|
||
"/auth HTTP 403 без NDMS challenge: доступ к HTTP Proxy с IP этого сервера "
|
||
"запрещён в настройках роутера, либо открыт не тот сервис. "
|
||
"В веб-интерфейсе Keenetic: разрешённые адреса для API / прокси — добавь IP VPS."
|
||
)
|
||
raise KeeneticRCIError("Нет заголовков X-NDM-Realm / Challenge или Set-Cookie")
|
||
md5_hex = hashlib.md5(
|
||
f"{self.login}:{realm}:{self.password}".encode()
|
||
).hexdigest()
|
||
sha_hex = hashlib.sha256(f"{challenge}{md5_hex}".encode()).hexdigest()
|
||
# Не писать Cookie в headers вручную — httpx иначе не подмешивает новую сессию из
|
||
# Set-Cookie после успешного POST /auth, и /rci/* отвечает 401.
|
||
client.cookies.update(r.cookies)
|
||
if not client.cookies and "=" in cookie_pair:
|
||
host = urlparse(self.base_url).hostname or ""
|
||
name, _, value = cookie_pair.partition("=")
|
||
client.cookies.set(name.strip(), value.strip(), domain=host)
|
||
r2 = client.post(
|
||
"/auth",
|
||
json={"login": self.login, "password": sha_hex},
|
||
)
|
||
if r2.status_code in (401, 403):
|
||
raise KeeneticRCIError(
|
||
"Неверный логин или пароль Keenetic (POST /auth). "
|
||
"Проверь учётку с доступом к HTTP Proxy / API и base URL (хост и порт как в настройках KeenDNS)."
|
||
)
|
||
if r2.status_code not in (200, 201, 202):
|
||
raise KeeneticRCIError(f"POST /auth HTTP {r2.status_code}")
|
||
client.cookies.update(r2.cookies)
|
||
|
||
def list_interfaces(self) -> list[dict[str, Any]]:
|
||
"""GET /rci/show/interface — id, type, description, state (как gokeenapi)."""
|
||
with self._client_ctx() as client:
|
||
self._auth(client)
|
||
r = client.get("/rci/show/interface")
|
||
if r.status_code != 200:
|
||
raise KeeneticRCIError(f"show/interface HTTP {r.status_code}")
|
||
try:
|
||
data = r.json()
|
||
except ValueError as e:
|
||
raise KeeneticRCIError(
|
||
f"show/interface: ответ не JSON (возможно неверный URL прокси). "
|
||
f"Начало тела: {r.text[:160]!r}"
|
||
) from e
|
||
if not isinstance(data, dict):
|
||
raise KeeneticRCIError("show/interface: ожидался объект JSON")
|
||
rows: list[dict[str, Any]] = []
|
||
for key, body in data.items():
|
||
if not isinstance(body, dict) or str(key).startswith("_"):
|
||
continue
|
||
iid = str(body.get("id") or body.get("Id") or key)
|
||
typ = str(body.get("type") or body.get("Type") or "")
|
||
desc = str(body.get("description") or body.get("Description") or "")
|
||
state = str(body.get("state") or body.get("State") or "")
|
||
link = str(body.get("link") or body.get("Link") or "")
|
||
conn = str(body.get("connected") or body.get("Connected") or "")
|
||
addr = str(body.get("address") or body.get("Address") or "")
|
||
rows.append(
|
||
{
|
||
"id": iid,
|
||
"type": typ,
|
||
"description": desc,
|
||
"state": state,
|
||
"link": link,
|
||
"connected": conn,
|
||
"address": addr,
|
||
"label": f"{iid} — {desc or typ or 'интерфейс'}",
|
||
}
|
||
)
|
||
rows.sort(key=lambda x: x["id"].lower())
|
||
return rows
|
||
|
||
def _parse_fqdn_response(self, data: dict[str, Any]) -> dict[str, list[str]]:
|
||
out: dict[str, list[str]] = {}
|
||
for name, body in data.items():
|
||
if not isinstance(body, dict) or name.startswith("_"):
|
||
continue
|
||
inc = body.get("include") or body.get("Include") or []
|
||
addrs: list[str] = []
|
||
for item in inc:
|
||
if isinstance(item, dict):
|
||
a = item.get("address") or item.get("Address")
|
||
if a:
|
||
addrs.append(str(a))
|
||
elif item:
|
||
addrs.append(str(item))
|
||
out[name] = addrs
|
||
return out
|
||
|
||
def get_fqdn_groups(self, client: httpx.Client) -> dict[str, list[str]]:
|
||
r = client.get("/rci/object-group/fqdn")
|
||
if r.status_code != 200:
|
||
raise KeeneticRCIError(f"object-group/fqdn HTTP {r.status_code}: {r.text[:200]}")
|
||
data = r.json()
|
||
if not isinstance(data, dict):
|
||
raise KeeneticRCIError("object-group/fqdn: не JSON-объект")
|
||
return self._parse_fqdn_response(data)
|
||
|
||
def get_dns_routes(self, client: httpx.Client) -> dict[str, str]:
|
||
r = client.get("/rci/dns-proxy/route")
|
||
if r.status_code != 200:
|
||
raise KeeneticRCIError(f"dns-proxy/route HTTP {r.status_code}")
|
||
data = r.json()
|
||
if not isinstance(data, list):
|
||
raise KeeneticRCIError("dns-proxy/route: ожидался массив")
|
||
out: dict[str, str] = {}
|
||
for row in data:
|
||
if not isinstance(row, dict):
|
||
continue
|
||
g = row.get("group") or row.get("Group")
|
||
iface = row.get("interface") or row.get("Interface")
|
||
if g and iface:
|
||
out[str(g)] = str(iface)
|
||
return out
|
||
|
||
def _post_parse(self, client: httpx.Client, commands: list[str]) -> list[dict[str, Any]]:
|
||
all_resp: list[dict[str, Any]] = []
|
||
for i in range(0, len(commands), _MAX_PARSE):
|
||
chunk = commands[i : i + _MAX_PARSE]
|
||
body = [{"parse": c} for c in chunk]
|
||
r = client.post("/rci/", json=body)
|
||
if r.status_code != 200:
|
||
raise KeeneticRCIError(f"POST /rci/ HTTP {r.status_code}: {r.text[:500]}")
|
||
part = r.json()
|
||
if not isinstance(part, list):
|
||
raise KeeneticRCIError("POST /rci/: ответ не массив")
|
||
all_resp.extend(part)
|
||
for item in part:
|
||
p = item.get("parse") or item.get("Parse") or {}
|
||
for s in p.get("status") or p.get("Status") or []:
|
||
if not isinstance(s, dict):
|
||
continue
|
||
sv = (s.get("status") or s.get("Status") or "").lower()
|
||
if sv == "error":
|
||
raise KeeneticRCIError(
|
||
f"RCI: {s.get('code')} {s.get('ident', '')} — {s.get('message', '')}"
|
||
)
|
||
return all_resp
|
||
|
||
def apply_groups(
|
||
self,
|
||
groups: dict[str, dict[str, Any]],
|
||
*,
|
||
group_names: tuple[str, ...] = ("US", "RU"),
|
||
) -> list[str]:
|
||
"""
|
||
Синхронизирует object-group fqdn + dns-proxy route для указанных групп.
|
||
Логика как в gokeenapi AddDnsRoutingGroups (инкрементально).
|
||
"""
|
||
log: list[str] = []
|
||
with self._client_ctx() as client:
|
||
self._auth(client)
|
||
existing = self.get_fqdn_groups(client)
|
||
routes = self.get_dns_routes(client)
|
||
cmds: list[str] = []
|
||
|
||
for gname in group_names:
|
||
spec = groups.get(gname) or {}
|
||
iface = (spec.get("interface_id") or "").strip()
|
||
raw_lines = spec.get("lines") or []
|
||
if not isinstance(raw_lines, list):
|
||
raw_lines = []
|
||
want = [x for x in _norm_lines([str(x) for x in raw_lines]) if _valid_entry(x)]
|
||
if not iface:
|
||
if want:
|
||
log.append(f"{gname}: пропуск — не задан interface_id")
|
||
continue
|
||
if not want:
|
||
log.append(f"{gname}: пропуск — пустой список строк")
|
||
continue
|
||
|
||
have = set(existing.get(gname, []))
|
||
want_set = set(want)
|
||
|
||
if gname not in existing:
|
||
cmds.append(f"object-group fqdn {gname}")
|
||
|
||
for ex in existing.get(gname, []):
|
||
if ex not in want_set:
|
||
cmds.append(f"no object-group fqdn {gname} include {ex}")
|
||
|
||
for w in want:
|
||
if w not in have:
|
||
cmds.append(f"object-group fqdn {gname} include {w}")
|
||
|
||
cur_if = routes.get(gname)
|
||
if cur_if != iface:
|
||
if cur_if:
|
||
cmds.append(f"no dns-proxy route object-group {gname} {cur_if}")
|
||
cmds.append(f"dns-proxy route object-group {gname} {iface} auto")
|
||
|
||
if not cmds:
|
||
log.append("Изменений нет (уже совпадает с роутером)")
|
||
return log
|
||
|
||
cmds.append("system configuration save")
|
||
logger.info("RCI %s команд на %s", len(cmds), self.base_url)
|
||
self._post_parse(client, cmds)
|
||
log.append(f"Применено команд: {len(cmds)}")
|
||
return log
|
||
|
||
|
||
def test_connection(base_url: str, login: str, password: str) -> tuple[bool, str]:
|
||
try:
|
||
with httpx.Client(
|
||
base_url=base_url.rstrip("/"),
|
||
verify=False,
|
||
timeout=httpx.Timeout(15.0),
|
||
follow_redirects=True,
|
||
) as c:
|
||
k = KeeneticRCI(base_url, login, password)
|
||
k._auth(c)
|
||
r = c.get("/rci/show/version")
|
||
if r.status_code != 200:
|
||
if r.status_code == 401:
|
||
return (
|
||
False,
|
||
"RCI /rci/show/version → 401: сессия не принята "
|
||
"(часто неверный логин/пароль или устаревший клиент; обновите сервис).",
|
||
)
|
||
return False, f"version HTTP {r.status_code}"
|
||
try:
|
||
j = r.json()
|
||
except ValueError:
|
||
return False, f"version: ответ не JSON (проверь URL прокси): {r.text[:120]!r}"
|
||
title = j.get("title") or j.get("Title") or "?"
|
||
return True, str(title)
|
||
except Exception as e:
|
||
return False, str(e)
|