auth: вход по логину/паролю от веб-админки роутера + сессионные токены

- auth.py: NDM-аутентификация (md5+sha256 challenge) к http://ROUTER_HOST/auth
- Сессионные токены (secrets, TTL 8ч, скользящее окно), заголовок X-Session-Token
- WEB_PASSWORD остаётся как запасной мастер-пароль
- /api/login и /api/logout, форма входа в UI с полем «Логин»
- Шапка автора в UI: GitHub, Telegram, Boosty, Поддержать (Ozon СБП)
- README: NDM-флоу, переменные ROUTER_HOST/ROUTER_LOGIN/ROUTER_TIMEOUT
- install.sh копирует auth.py
This commit is contained in:
Андрей Бобырев
2026-04-26 00:19:09 +03:00
parent 7c7a58b3a9
commit 75cb432e13
6 changed files with 324 additions and 90 deletions

116
auth.py Normal file
View File

@@ -0,0 +1,116 @@
"""Аутентификация против HTTP-админки Keenetic (NDM) + сессионные токены.
Используется протокол NDM:
1) GET /auth → 401 + заголовки X-NDM-Realm и X-NDM-Challenge.
2) hash = sha256(challenge + md5(login:realm:password)).
3) POST /auth с JSON {"login": ..., "password": hash} → 200 OK при успехе.
Не требует внешних библиотек (urllib стандартной библиотеки).
"""
from __future__ import annotations
import hashlib
import http.cookiejar
import json
import logging
import secrets
import threading
import time
import urllib.error
import urllib.request
log = logging.getLogger("kssh.auth")
def _normalize_host(host: str) -> str:
h = (host or "").strip().rstrip("/")
if not h:
return ""
if not h.startswith(("http://", "https://")):
h = "http://" + h
return h
def keenetic_validate(host: str, login: str, password: str, timeout: float = 5.0) -> bool:
"""Проверяет связку login/password против админки Keenetic по NDM-протоколу."""
base = _normalize_host(host)
if not base or not login or password is None:
return False
cj = http.cookiejar.CookieJar()
opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(cj))
realm = ""
challenge = ""
try:
req = urllib.request.Request(base + "/auth", method="GET")
with opener.open(req, timeout=timeout) as r:
if r.status == 200:
return True
realm = r.headers.get("X-NDM-Realm", "") or ""
challenge = r.headers.get("X-NDM-Challenge", "") or ""
except urllib.error.HTTPError as e:
if e.code != 401:
return False
realm = e.headers.get("X-NDM-Realm", "") or ""
challenge = e.headers.get("X-NDM-Challenge", "") or ""
except (urllib.error.URLError, OSError) as e:
log.warning("auth: router unreachable %s: %s", base, e)
return False
if not realm or not challenge:
return False
md5 = hashlib.md5(f"{login}:{realm}:{password}".encode("utf-8")).hexdigest()
sha = hashlib.sha256((challenge + md5).encode("utf-8")).hexdigest()
body = json.dumps({"login": login, "password": sha}).encode("utf-8")
try:
req2 = urllib.request.Request(
base + "/auth",
data=body,
headers={"Content-Type": "application/json"},
method="POST",
)
with opener.open(req2, timeout=timeout) as r2:
return 200 <= r2.status < 300
except urllib.error.HTTPError:
return False
except (urllib.error.URLError, OSError) as e:
log.warning("auth: post failed %s: %s", base, e)
return False
_lock = threading.Lock()
_sessions: dict[str, float] = {}
_TTL = 8 * 3600
def _gc_locked(now: float) -> None:
expired = [t for t, exp in _sessions.items() if exp < now]
for t in expired:
_sessions.pop(t, None)
def issue_token(ttl: int = _TTL) -> str:
now = time.time()
token = secrets.token_urlsafe(32)
with _lock:
_gc_locked(now)
_sessions[token] = now + ttl
return token
def is_token_valid(token: str, ttl: int = _TTL) -> bool:
if not token:
return False
now = time.time()
with _lock:
_gc_locked(now)
exp = _sessions.get(token)
if exp and exp >= now:
_sessions[token] = now + ttl
return True
return False
def revoke_token(token: str) -> None:
if not token:
return
with _lock:
_sessions.pop(token, None)