299 lines
10 KiB
Python
299 lines
10 KiB
Python
from __future__ import annotations
|
|
|
|
import hmac
|
|
import json
|
|
import os
|
|
import secrets
|
|
import time
|
|
from dataclasses import dataclass
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from threading import RLock
|
|
from typing import Any
|
|
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
|
|
|
|
from fastapi import APIRouter, HTTPException, Request
|
|
from fastapi.responses import JSONResponse
|
|
from pydantic import BaseModel, Field
|
|
|
|
|
|
DEFAULT_SECURITY: dict[str, Any] = {
|
|
# BUILD98: editor access is a fixed local PIN. The older daily PIN and
|
|
# master PIN fields are retained only so existing config files still parse.
|
|
"fixed_pin": "1993",
|
|
"daily_mask": "4827",
|
|
"master_pin": "638194",
|
|
"timezone": "Europe/Moscow",
|
|
"session_minutes": 30,
|
|
"max_attempts": 5,
|
|
"lock_seconds": 30,
|
|
"cookie_secure": False,
|
|
}
|
|
|
|
COOKIE_NAME = "ui_builder_editor_session"
|
|
|
|
|
|
class PinPayload(BaseModel):
|
|
pin: str = Field(min_length=4, max_length=12)
|
|
|
|
|
|
@dataclass
|
|
class FailureState:
|
|
attempts: int = 0
|
|
locked_until: float = 0.0
|
|
|
|
|
|
class EditorAuthManager:
|
|
"""Local PIN guard for the visual editor.
|
|
|
|
BUILD98 uses one fixed PIN (1993 by default). The old daily-PIN helpers are
|
|
kept for backward-compatible diagnostics only and are no longer accepted
|
|
by the login endpoint.
|
|
"""
|
|
|
|
def __init__(self, settings_dir: Path) -> None:
|
|
self.settings_dir = Path(settings_dir)
|
|
self.security_file = self.settings_dir / "editor_security.json"
|
|
self.settings_dir.mkdir(parents=True, exist_ok=True)
|
|
self._lock = RLock()
|
|
self._sessions: dict[str, float] = {}
|
|
self._failures: dict[str, FailureState] = {}
|
|
self.settings = self._load_settings()
|
|
|
|
def _load_settings(self) -> dict[str, Any]:
|
|
if not self.security_file.exists():
|
|
self.security_file.write_text(
|
|
json.dumps(DEFAULT_SECURITY, ensure_ascii=False, indent=2),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
try:
|
|
raw = json.loads(self.security_file.read_text(encoding="utf-8"))
|
|
except (OSError, json.JSONDecodeError, TypeError):
|
|
raw = {}
|
|
|
|
result = dict(DEFAULT_SECURITY)
|
|
if isinstance(raw, dict):
|
|
result.update(raw)
|
|
|
|
overrides = {
|
|
"fixed_pin": os.getenv("EDITOR_FIXED_PIN"),
|
|
"daily_mask": os.getenv("EDITOR_PIN_MASK"),
|
|
"master_pin": os.getenv("EDITOR_MASTER_PIN"),
|
|
"timezone": os.getenv("EDITOR_PIN_TIMEZONE"),
|
|
"session_minutes": os.getenv("EDITOR_SESSION_MINUTES"),
|
|
"max_attempts": os.getenv("EDITOR_MAX_ATTEMPTS"),
|
|
"lock_seconds": os.getenv("EDITOR_LOCK_SECONDS"),
|
|
}
|
|
for key, value in overrides.items():
|
|
if value not in (None, ""):
|
|
result[key] = value
|
|
|
|
cookie_override = os.getenv("EDITOR_COOKIE_SECURE")
|
|
if cookie_override not in (None, ""):
|
|
result["cookie_secure"] = cookie_override.strip().lower() in {
|
|
"1", "true", "yes", "on"
|
|
}
|
|
|
|
fixed = "".join(character for character in str(result.get("fixed_pin", "1993")) if character.isdigit())
|
|
result["fixed_pin"] = fixed if 4 <= len(fixed) <= 12 else "1993"
|
|
|
|
mask = "".join(character for character in str(result["daily_mask"]) if character.isdigit())
|
|
result["daily_mask"] = mask[:4].ljust(4, "0") if mask else "4827"
|
|
|
|
master = "".join(character for character in str(result["master_pin"]) if character.isdigit())
|
|
result["master_pin"] = master if 4 <= len(master) <= 12 else "638194"
|
|
|
|
try:
|
|
ZoneInfo(str(result["timezone"]))
|
|
except ZoneInfoNotFoundError:
|
|
result["timezone"] = "UTC"
|
|
|
|
for key, low, high, fallback in (
|
|
("session_minutes", 1, 1440, 30),
|
|
("max_attempts", 1, 20, 5),
|
|
("lock_seconds", 5, 3600, 30),
|
|
):
|
|
try:
|
|
value = int(result[key])
|
|
except (TypeError, ValueError):
|
|
value = fallback
|
|
result[key] = max(low, min(high, value))
|
|
|
|
return result
|
|
|
|
@property
|
|
def timezone(self) -> ZoneInfo:
|
|
return ZoneInfo(str(self.settings["timezone"]))
|
|
|
|
@property
|
|
def session_seconds(self) -> int:
|
|
return int(self.settings["session_minutes"]) * 60
|
|
|
|
def now(self) -> datetime:
|
|
return datetime.now(self.timezone)
|
|
|
|
def daily_pin(self, moment: datetime | None = None) -> str:
|
|
current = moment.astimezone(self.timezone) if moment else self.now()
|
|
date_digits = current.strftime("%d%m")
|
|
mask = str(self.settings["daily_mask"])
|
|
return "".join(
|
|
str((int(date_digit) + int(mask_digit)) % 10)
|
|
for date_digit, mask_digit in zip(date_digits, mask, strict=True)
|
|
)
|
|
|
|
def explain_daily_pin(self, moment: datetime | None = None) -> dict[str, Any]:
|
|
current = moment.astimezone(self.timezone) if moment else self.now()
|
|
date_digits = current.strftime("%d%m")
|
|
mask = str(self.settings["daily_mask"])
|
|
return {
|
|
"date": current.strftime("%d.%m.%Y"),
|
|
"date_digits": date_digits,
|
|
"mask": mask,
|
|
"pin": self.daily_pin(current),
|
|
"timezone": str(self.settings["timezone"]),
|
|
"steps": [
|
|
f"{date_digit} + {mask_digit} → {(int(date_digit) + int(mask_digit)) % 10}"
|
|
for date_digit, mask_digit in zip(date_digits, mask, strict=True)
|
|
],
|
|
}
|
|
|
|
@staticmethod
|
|
def _client_key(request: Request) -> str:
|
|
return request.client.host if request.client else "local"
|
|
|
|
def _cleanup_sessions(self) -> None:
|
|
now = time.time()
|
|
for token, expires_at in list(self._sessions.items()):
|
|
if expires_at <= now:
|
|
self._sessions.pop(token, None)
|
|
|
|
def is_token_valid(self, token: str | None) -> bool:
|
|
if not token:
|
|
return False
|
|
with self._lock:
|
|
self._cleanup_sessions()
|
|
return self._sessions.get(token, 0) > time.time()
|
|
|
|
def is_request_authenticated(self, request: Request) -> bool:
|
|
return self.is_token_valid(request.cookies.get(COOKIE_NAME))
|
|
|
|
async def require_editor(self, request: Request) -> None:
|
|
if not self.is_request_authenticated(request):
|
|
raise HTTPException(status_code=401, detail="Сессия конструктора не активна")
|
|
|
|
def status(self, request: Request) -> dict[str, Any]:
|
|
client_key = self._client_key(request)
|
|
token = request.cookies.get(COOKIE_NAME)
|
|
now = time.time()
|
|
|
|
with self._lock:
|
|
self._cleanup_sessions()
|
|
failure = self._failures.get(client_key, FailureState())
|
|
expires_at = self._sessions.get(token or "", 0)
|
|
authenticated = expires_at > now
|
|
|
|
return {
|
|
"authenticated": authenticated,
|
|
"expires_in": max(0, int(expires_at - now)) if authenticated else 0,
|
|
"attempts_remaining": max(
|
|
0,
|
|
int(self.settings["max_attempts"]) - failure.attempts,
|
|
),
|
|
"locked_for": max(0, int(round(failure.locked_until - now))),
|
|
"timezone": str(self.settings["timezone"]),
|
|
"date": self.now().strftime("%d.%m.%Y"),
|
|
"session_minutes": int(self.settings["session_minutes"]),
|
|
}
|
|
|
|
def login(self, request: Request, pin: str) -> tuple[str, int]:
|
|
client_key = self._client_key(request)
|
|
now = time.time()
|
|
supplied = str(pin).strip()
|
|
|
|
with self._lock:
|
|
failure = self._failures.setdefault(client_key, FailureState())
|
|
if failure.locked_until > now:
|
|
raise HTTPException(
|
|
status_code=429,
|
|
detail={
|
|
"message": "Вход временно заблокирован",
|
|
"locked_for": max(1, int(round(failure.locked_until - now))),
|
|
},
|
|
)
|
|
|
|
valid = hmac.compare_digest(supplied, str(self.settings["fixed_pin"]))
|
|
if not valid:
|
|
failure.attempts += 1
|
|
remaining = max(
|
|
0,
|
|
int(self.settings["max_attempts"]) - failure.attempts,
|
|
)
|
|
if failure.attempts >= int(self.settings["max_attempts"]):
|
|
failure.attempts = 0
|
|
failure.locked_until = now + int(self.settings["lock_seconds"])
|
|
raise HTTPException(
|
|
status_code=429,
|
|
detail={
|
|
"message": "Вход временно заблокирован",
|
|
"locked_for": int(self.settings["lock_seconds"]),
|
|
},
|
|
)
|
|
raise HTTPException(
|
|
status_code=401,
|
|
detail={
|
|
"message": "Неверный PIN",
|
|
"attempts_remaining": remaining,
|
|
},
|
|
)
|
|
|
|
self._failures.pop(client_key, None)
|
|
token = secrets.token_urlsafe(32)
|
|
self._sessions[token] = now + self.session_seconds
|
|
return token, self.session_seconds
|
|
|
|
def logout(self, request: Request) -> None:
|
|
token = request.cookies.get(COOKIE_NAME)
|
|
if token:
|
|
with self._lock:
|
|
self._sessions.pop(token, None)
|
|
|
|
|
|
def create_editor_auth_router(
|
|
auth: EditorAuthManager,
|
|
*,
|
|
prefix: str = "/api/ui-builder/auth",
|
|
editor_url: str = "/editor",
|
|
) -> APIRouter:
|
|
router = APIRouter(prefix=prefix, tags=["UI Builder Auth"])
|
|
|
|
@router.get("/status")
|
|
async def auth_status(request: Request) -> dict[str, Any]:
|
|
return auth.status(request)
|
|
|
|
@router.post("/login")
|
|
async def auth_login(request: Request, payload: PinPayload) -> JSONResponse:
|
|
token, max_age = auth.login(request, payload.pin)
|
|
response = JSONResponse(
|
|
{"ok": True, "editor_url": editor_url, "expires_in": max_age}
|
|
)
|
|
response.set_cookie(
|
|
COOKIE_NAME,
|
|
token,
|
|
max_age=max_age,
|
|
httponly=True,
|
|
samesite="strict",
|
|
secure=bool(auth.settings["cookie_secure"]),
|
|
path="/",
|
|
)
|
|
return response
|
|
|
|
@router.post("/logout")
|
|
async def auth_logout(request: Request) -> JSONResponse:
|
|
auth.logout(request)
|
|
response = JSONResponse({"ok": True})
|
|
response.delete_cookie(COOKIE_NAME, path="/")
|
|
return response
|
|
|
|
return router
|