248 lines
9.8 KiB
Python
248 lines
9.8 KiB
Python
"""Авторизация по логину/паролю + роли (оператор / админ / суперадмин).
|
|
|
|
Пользователи читаются из .env.json (раздел "users"). Если его нет — он
|
|
выводится из старого раздела "auth" (как один суперадмин) или admin/admin по
|
|
умолчанию. Сессии — в памяти (перезапуск сервиса разлогинивает всех). Пароли
|
|
хранятся в открытом виде, как и раньше; для нескольких операторов этого
|
|
достаточно, при необходимости позже заменим на хеши.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import secrets
|
|
|
|
from fastapi import APIRouter, HTTPException, Request, Response
|
|
from fastapi.responses import JSONResponse
|
|
|
|
COOKIE_NAME = "nvr_session"
|
|
|
|
ROLES = ("operator", "admin", "superadmin")
|
|
_RANK = {"operator": 1, "admin": 2, "superadmin": 3}
|
|
|
|
|
|
def role_at_least(role: str | None, minimum: str) -> bool:
|
|
return bool(role) and _RANK.get(role, 0) >= _RANK[minimum]
|
|
|
|
|
|
class AuthManager:
|
|
def __init__(self, settings_path: str | None = None):
|
|
self.path = settings_path or os.environ.get("NVR_SETTINGS", ".env.json")
|
|
self.users: list[dict] = []
|
|
self._sessions: dict[str, str] = {} # token -> username
|
|
self.reload()
|
|
|
|
# ── загрузка/сохранение ─────────────────────────────────────
|
|
def _read(self) -> dict:
|
|
try:
|
|
with open(self.path, "r", encoding="utf-8") as fh:
|
|
return json.load(fh)
|
|
except (OSError, json.JSONDecodeError):
|
|
return {}
|
|
|
|
def reload(self) -> None:
|
|
data = self._read()
|
|
users = data.get("users")
|
|
if not users:
|
|
auth = data.get("auth") or {}
|
|
users = [{
|
|
"username": auth.get("username", "admin"),
|
|
"password": auth.get("password", "admin"),
|
|
"role": "superadmin",
|
|
}]
|
|
norm: list[dict] = []
|
|
for u in users:
|
|
role = u.get("role", "operator")
|
|
if role not in ROLES:
|
|
role = "operator"
|
|
norm.append({
|
|
"username": u.get("username", ""),
|
|
"password": u.get("password", ""),
|
|
"role": role,
|
|
})
|
|
# гарантируем хотя бы одного суперадмина
|
|
if norm and not any(u["role"] == "superadmin" for u in norm):
|
|
norm[0]["role"] = "superadmin"
|
|
self.users = norm
|
|
|
|
def _write_users(self) -> None:
|
|
data = self._read()
|
|
data["users"] = self.users
|
|
with open(self.path, "w", encoding="utf-8") as fh:
|
|
json.dump(data, fh, ensure_ascii=False, indent=2)
|
|
|
|
def _find(self, username: str) -> dict | None:
|
|
return next((u for u in self.users if u["username"] == username), None)
|
|
|
|
def _superadmin_count(self) -> int:
|
|
return sum(1 for u in self.users if u["role"] == "superadmin")
|
|
|
|
# ── сессии ──────────────────────────────────────────────────
|
|
def login(self, username: str, password: str) -> str | None:
|
|
u = self._find(username)
|
|
if u and secrets.compare_digest(password, u["password"]):
|
|
token = secrets.token_urlsafe(32)
|
|
self._sessions[token] = username
|
|
return token
|
|
return None
|
|
|
|
def validate(self, token: str | None) -> bool:
|
|
return bool(token) and token in self._sessions and \
|
|
self._find(self._sessions[token]) is not None
|
|
|
|
def user_for(self, token: str | None) -> dict | None:
|
|
if not token:
|
|
return None
|
|
username = self._sessions.get(token)
|
|
return self._find(username) if username else None
|
|
|
|
def role_for(self, token: str | None) -> str | None:
|
|
u = self.user_for(token)
|
|
return u["role"] if u else None
|
|
|
|
def logout(self, token: str | None) -> None:
|
|
if token:
|
|
self._sessions.pop(token, None)
|
|
|
|
def verify_token_password(self, token: str | None, password: str) -> bool:
|
|
u = self.user_for(token)
|
|
return bool(u) and secrets.compare_digest(password or "", u["password"])
|
|
|
|
# ── управление пользователями ───────────────────────────────
|
|
def list_users(self) -> list[dict]:
|
|
return [{"username": u["username"], "role": u["role"]} for u in self.users]
|
|
|
|
def add_user(self, username: str, password: str, role: str) -> None:
|
|
username = (username or "").strip()
|
|
if not username:
|
|
raise ValueError("пустой логин")
|
|
if role not in ROLES:
|
|
raise ValueError("недопустимая роль")
|
|
if not password:
|
|
raise ValueError("пустой пароль")
|
|
if self._find(username):
|
|
raise ValueError("пользователь с таким логином уже существует")
|
|
self.users.append({"username": username, "password": password, "role": role})
|
|
self._write_users()
|
|
|
|
def update_user(self, username: str, password: str | None = None,
|
|
role: str | None = None) -> None:
|
|
u = self._find(username)
|
|
if not u:
|
|
raise ValueError("пользователь не найден")
|
|
if role is not None:
|
|
if role not in ROLES:
|
|
raise ValueError("недопустимая роль")
|
|
if u["role"] == "superadmin" and role != "superadmin" and \
|
|
self._superadmin_count() == 1:
|
|
raise ValueError("нельзя снять роль с последнего суперадмина")
|
|
u["role"] = role
|
|
if password:
|
|
u["password"] = password
|
|
self._write_users()
|
|
|
|
def delete_user(self, username: str) -> None:
|
|
u = self._find(username)
|
|
if not u:
|
|
raise ValueError("пользователь не найден")
|
|
if u["role"] == "superadmin" and self._superadmin_count() == 1:
|
|
raise ValueError("нельзя удалить последнего суперадмина")
|
|
self.users = [x for x in self.users if x["username"] != username]
|
|
self._sessions = {t: n for t, n in self._sessions.items() if n != username}
|
|
self._write_users()
|
|
|
|
|
|
auth_manager = AuthManager()
|
|
|
|
|
|
def require_role(request: Request, minimum: str) -> str:
|
|
"""Проверяет, что текущая сессия имеет роль не ниже minimum, иначе 403."""
|
|
role = auth_manager.role_for(request.cookies.get(COOKIE_NAME))
|
|
if not role_at_least(role, minimum):
|
|
raise HTTPException(403, "недостаточно прав")
|
|
return role
|
|
|
|
|
|
router = APIRouter(prefix="/api", tags=["auth"])
|
|
|
|
|
|
@router.post("/login")
|
|
async def login(request: Request) -> Response:
|
|
body = await request.json()
|
|
token = auth_manager.login(body.get("username", ""), body.get("password", ""))
|
|
if not token:
|
|
return JSONResponse({"detail": "неверный логин или пароль"}, status_code=401)
|
|
resp = JSONResponse({"ok": True})
|
|
resp.set_cookie(COOKIE_NAME, token, httponly=True, samesite="lax", max_age=7 * 86400)
|
|
return resp
|
|
|
|
|
|
@router.get("/me")
|
|
async def me(request: Request) -> Response:
|
|
u = auth_manager.user_for(request.cookies.get(COOKIE_NAME))
|
|
if not u:
|
|
return JSONResponse({"detail": "unauthorized"}, status_code=401)
|
|
return JSONResponse({"username": u["username"], "role": u["role"]})
|
|
|
|
|
|
@router.post("/verify-password")
|
|
async def verify_password(request: Request) -> Response:
|
|
"""Подтверждение пароля текущего оператора (для защищённых настроек)."""
|
|
body = await request.json()
|
|
token = request.cookies.get(COOKIE_NAME)
|
|
if not auth_manager.verify_token_password(token, body.get("password", "")):
|
|
return JSONResponse({"detail": "неверный пароль"}, status_code=403)
|
|
return JSONResponse({"ok": True})
|
|
|
|
|
|
@router.post("/logout")
|
|
async def logout(request: Request) -> Response:
|
|
auth_manager.logout(request.cookies.get(COOKIE_NAME))
|
|
resp = JSONResponse({"ok": True})
|
|
resp.delete_cookie(COOKIE_NAME)
|
|
return resp
|
|
|
|
|
|
# ── управление пользователями (только суперадмин) ───────────────
|
|
@router.get("/users")
|
|
async def users_list(request: Request) -> dict:
|
|
require_role(request, "superadmin")
|
|
return {"users": auth_manager.list_users()}
|
|
|
|
|
|
@router.post("/users")
|
|
async def users_create(request: Request) -> dict:
|
|
require_role(request, "superadmin")
|
|
body = await request.json()
|
|
try:
|
|
auth_manager.add_user(body.get("username", ""), body.get("password", ""),
|
|
body.get("role", "operator"))
|
|
except ValueError as e:
|
|
raise HTTPException(400, str(e))
|
|
return {"ok": True}
|
|
|
|
|
|
@router.put("/users/{username}")
|
|
async def users_update(username: str, request: Request) -> dict:
|
|
require_role(request, "superadmin")
|
|
body = await request.json()
|
|
try:
|
|
auth_manager.update_user(
|
|
username,
|
|
password=(body.get("password") or None),
|
|
role=(body.get("role") or None),
|
|
)
|
|
except ValueError as e:
|
|
raise HTTPException(400, str(e))
|
|
return {"ok": True}
|
|
|
|
|
|
@router.delete("/users/{username}")
|
|
async def users_delete(username: str, request: Request) -> dict:
|
|
require_role(request, "superadmin")
|
|
try:
|
|
auth_manager.delete_user(username)
|
|
except ValueError as e:
|
|
raise HTTPException(400, str(e))
|
|
return {"ok": True}
|