feat: scan interfaces via RCI; show Keenetic login + password status (no secret in UI)

Made-with: Cursor
This commit is contained in:
Андрей Бобырев
2026-04-25 00:45:55 +03:00
parent e1d6e6bb81
commit ef946e37f2
4 changed files with 149 additions and 5 deletions

View File

@@ -55,6 +55,40 @@ async def get_data(x_admin_password: str = Header("")):
return load_store()
@app.get("/api/keenetic-env")
async def keenetic_env(x_admin_password: str = Header("")):
"""Логин и факт наличия пароля (сам пароль в ответ не кладём — только из .env на сервере)."""
_chk(x_admin_password)
return {
"login": config.KEENETIC_LOGIN,
"password_configured": bool(config.KEENETIC_PASSWORD),
"hint": "Пароль смотри только в server/.env (KEENETIC_PASSWORD); в браузер не передаётся.",
}
@app.get("/api/routers/{rid}/interfaces")
async def router_interfaces(rid: str, x_admin_password: str = Header("")):
_chk(x_admin_password)
if not config.KEENETIC_PASSWORD:
raise HTTPException(400, "Задайте KEENETIC_PASSWORD в .env")
cur = load_store()
r = next((x for x in cur.get("routers") or [] if x.get("id") == rid), None)
if not r:
raise HTTPException(404, "Роутер не найден")
def _run():
k = KeeneticRCI(
r["rci_base_url"], config.KEENETIC_LOGIN, config.KEENETIC_PASSWORD
)
return k.list_interfaces()
try:
items = await asyncio.to_thread(_run)
except KeeneticRCIError as e:
raise HTTPException(502, str(e)) from e
return {"interfaces": items}
class PutDataBody(BaseModel):
groups: dict[str, dict] | None = None
routers: list[dict] | None = None

View File

@@ -90,6 +90,42 @@ class KeeneticRCI:
if r2.status_code not in (200, 201, 202):
raise KeeneticRCIError(f"POST /auth HTTP {r2.status_code}")
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}")
data = r.json()
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():