Vladislav Kopnin
34a46b6832
Интерфейс: - после входа на главной появляются «Админка» и «Выйти» — в шапке, перед переключателем темы и после него соответственно (порядок: Админка, тема, Выйти); - «Выйти» теперь красная и заметно реагирует на наведение (темнеет, приподнимается, тень) — и на сайте, и в админке (одна кнопка в шапке, из карточки убрана); - в админке убраны «осталось попыток», путь к файлу настроек и карточка «Данные расписания» (дублировала шапку); блокировка показывается отдельной плашкой только когда действительно активна. Безопасность пути: - путь к панели не отдаётся ни в HTML, ни в JSON (из /api/admin/state убран), ни в JS: на главной для вошедшего только флаг, а кнопка «Админка» ведёт на служебный /api/admin/goto, который отвечает 302 на секретный путь; - без сессии /api/admin/goto и /api/admin/state отвечают 401, неизвестные пути дают 404 — перебором путь не находится; - страница панели: noindex/nofollow/noarchive, Referrer-Policy: no-referrer, X-Frame-Options: DENY, Cache-Control: no-store; - robots.txt без пути (только Disallow: /api/), внешние ссылки в панели — rel=noopener noreferrer. Тесты: 161 (состояние без пути, /goto: 401 без входа и 302 с ним).
296 lines
12 KiB
Python
296 lines
12 KiB
Python
"""Админка расписания.
|
||
|
||
Отдельного входа на сайте нет: панель открывается по секретному пути из
|
||
ADMIN_PATH, логин и пароль задаются в ADMIN_LOGIN/ADMIN_PASSWORD. Если они
|
||
пусты, админка выключена и маршруты /api/admin/* не регистрируются.
|
||
|
||
После нескольких неудачных попыток вход блокируется на время из
|
||
ADMIN_LOCKOUT_MINUTES (по умолчанию: 3 попытки и 15 минут). Каждая попытка
|
||
запоминается (откуда, когда, успешна ли) — это видно в панели и пишется в лог.
|
||
|
||
Что умеет администратор:
|
||
GET /api/admin/state — состояние кэша, ссылка, интервал, история входов;
|
||
POST /api/admin/link — проверить и поставить новую ссылку на таблицу;
|
||
POST /api/admin/interval — интервал обновления в минутах;
|
||
POST /api/admin/refresh — принудительно перечитать таблицу сейчас.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import base64
|
||
import hashlib
|
||
import hmac
|
||
import json
|
||
import logging
|
||
import threading
|
||
import time
|
||
from dataclasses import dataclass
|
||
from datetime import datetime
|
||
|
||
from fastapi import APIRouter, Depends, HTTPException, Request, Response
|
||
from fastapi.responses import RedirectResponse
|
||
from pydantic import BaseModel
|
||
|
||
from schedule import ScheduleCache, ScheduleError
|
||
from schedule.settings import SettingsStore
|
||
|
||
log = logging.getLogger(__name__)
|
||
|
||
COOKIE_NAME = "schedule_admin"
|
||
SESSION_TTL_SECONDS = 12 * 3600
|
||
|
||
DEFAULT_MAX_ATTEMPTS = 3
|
||
DEFAULT_LOCKOUT_MINUTES = 15
|
||
|
||
|
||
@dataclass
|
||
class AdminConfig:
|
||
"""Настройки админки: секретный путь, учётные данные и защита от перебора."""
|
||
|
||
path: str = ""
|
||
login: str = ""
|
||
password: str = ""
|
||
max_attempts: int = DEFAULT_MAX_ATTEMPTS
|
||
lockout_minutes: int = DEFAULT_LOCKOUT_MINUTES
|
||
|
||
@property
|
||
def enabled(self) -> bool:
|
||
return bool(self.login and self.password)
|
||
|
||
|
||
def _b64e(raw: bytes) -> str:
|
||
return base64.urlsafe_b64encode(raw).decode().rstrip("=")
|
||
|
||
|
||
def _b64d(text: str) -> bytes:
|
||
return base64.urlsafe_b64decode(text + "=" * (-len(text) % 4))
|
||
|
||
|
||
def _same(left: str, right: str) -> bool:
|
||
"""Сравнение строк за постоянное время (compare_digest не умеет не-ASCII)."""
|
||
return hmac.compare_digest(left.encode("utf-8"), right.encode("utf-8"))
|
||
|
||
|
||
def make_token(login: str, password: str, ttl: int = SESSION_TTL_SECONDS) -> str:
|
||
payload = json.dumps({"sub": login, "exp": int(time.time()) + ttl}, separators=(",", ":")).encode()
|
||
sig = hmac.new(password.encode(), payload, hashlib.sha256).digest()
|
||
return f"{_b64e(payload)}.{_b64e(sig)}"
|
||
|
||
|
||
def verify_token(token: str, password: str) -> bool:
|
||
try:
|
||
payload_b64, sig_b64 = token.split(".", 1)
|
||
payload = _b64d(payload_b64)
|
||
expected = hmac.new(password.encode(), payload, hashlib.sha256).digest()
|
||
if not hmac.compare_digest(expected, _b64d(sig_b64)):
|
||
return False
|
||
return int(json.loads(payload).get("exp", 0)) > time.time()
|
||
except (ValueError, TypeError):
|
||
return False
|
||
|
||
|
||
class LoginThrottle:
|
||
"""Счётчик неудачных попыток: N промахов — пауза на lockout_minutes."""
|
||
|
||
def __init__(self, max_attempts: int = DEFAULT_MAX_ATTEMPTS, lockout_minutes: int = DEFAULT_LOCKOUT_MINUTES) -> None:
|
||
self.max_attempts = max(1, int(max_attempts))
|
||
self.lockout_seconds = max(1, int(lockout_minutes)) * 60
|
||
self._lock = threading.Lock()
|
||
self._failures: dict[str, int] = {}
|
||
self._blocked_until: dict[str, float] = {}
|
||
|
||
def locked_until(self, key: str) -> float | None:
|
||
with self._lock:
|
||
until = self._blocked_until.get(key)
|
||
if until is None:
|
||
return None
|
||
if until <= time.time():
|
||
self._blocked_until.pop(key, None)
|
||
self._failures.pop(key, None)
|
||
return None
|
||
return until
|
||
|
||
def attempts_left(self, key: str) -> int:
|
||
with self._lock:
|
||
return max(0, self.max_attempts - self._failures.get(key, 0))
|
||
|
||
def register_failure(self, key: str) -> None:
|
||
with self._lock:
|
||
failures = self._failures.get(key, 0) + 1
|
||
if failures >= self.max_attempts:
|
||
self._blocked_until[key] = time.time() + self.lockout_seconds
|
||
self._failures[key] = 0
|
||
else:
|
||
self._failures[key] = failures
|
||
|
||
def reset(self, key: str) -> None:
|
||
with self._lock:
|
||
self._failures.pop(key, None)
|
||
self._blocked_until.pop(key, None)
|
||
|
||
|
||
class LoginBody(BaseModel):
|
||
login: str
|
||
password: str
|
||
|
||
|
||
class LinkBody(BaseModel):
|
||
link: str
|
||
|
||
|
||
class IntervalBody(BaseModel):
|
||
interval_minutes: int
|
||
|
||
|
||
def _client_ip(request: Request) -> str:
|
||
"""Реальный адрес клиента: за nginx он приходит в X-Forwarded-For."""
|
||
forwarded = request.headers.get("x-forwarded-for", "")
|
||
if forwarded:
|
||
return forwarded.split(",")[0].strip() or "неизвестно"
|
||
return request.client.host if request.client else "неизвестно"
|
||
|
||
|
||
def _human_dt(iso: str | None) -> str:
|
||
if not iso:
|
||
return "—"
|
||
try:
|
||
return datetime.fromisoformat(iso).strftime("%d.%m.%Y %H:%M:%S")
|
||
except ValueError:
|
||
return iso
|
||
|
||
|
||
def create_admin_router(
|
||
cache: ScheduleCache,
|
||
settings: SettingsStore | None,
|
||
config: AdminConfig,
|
||
) -> APIRouter:
|
||
router = APIRouter(prefix="/api/admin", tags=["admin"])
|
||
throttle = LoginThrottle(config.max_attempts, config.lockout_minutes)
|
||
|
||
def require_auth(request: Request) -> None:
|
||
token = request.cookies.get(COOKIE_NAME, "")
|
||
if not token or not verify_token(token, config.password):
|
||
raise HTTPException(status_code=401, detail="Требуется вход в админку")
|
||
|
||
def _secure(request: Request) -> bool:
|
||
proto = request.headers.get("x-forwarded-proto", "") or request.url.scheme
|
||
return proto.split(",")[0].strip() == "https"
|
||
|
||
def _remember(ip: str, ok: bool) -> None:
|
||
if settings is not None:
|
||
settings.record_login(ip, ok)
|
||
|
||
@router.post("/login")
|
||
def login(body: LoginBody, request: Request, response: Response) -> dict:
|
||
ip = _client_ip(request)
|
||
until = throttle.locked_until(ip)
|
||
if until is not None:
|
||
minutes = max(1, round((until - time.time()) / 60))
|
||
log.warning("Вход в админку заблокирован (%s), осталось ~%d мин", ip, minutes)
|
||
raise HTTPException(
|
||
status_code=429,
|
||
detail=f"Слишком много неудачных попыток. Вход заблокирован ещё на ~{minutes} мин.",
|
||
)
|
||
ok_login = _same(body.login.strip(), config.login)
|
||
ok_password = _same(body.password, config.password)
|
||
if not (ok_login and ok_password):
|
||
throttle.register_failure(ip)
|
||
_remember(ip, False)
|
||
left = throttle.attempts_left(ip)
|
||
if throttle.locked_until(ip) is not None:
|
||
log.warning(
|
||
"Неудачная попытка входа (%s): исчерпаны попытки, пауза %d мин",
|
||
ip,
|
||
config.lockout_minutes,
|
||
)
|
||
raise HTTPException(
|
||
status_code=429,
|
||
detail=f"Неверный логин или пароль. Попытки исчерпаны — вход заблокирован на {config.lockout_minutes} мин.",
|
||
)
|
||
log.warning("Неудачная попытка входа (%s), осталось попыток: %d", ip, left)
|
||
raise HTTPException(
|
||
status_code=401,
|
||
detail=f"Неверный логин или пароль. Осталось попыток: {left}.",
|
||
)
|
||
throttle.reset(ip)
|
||
previous = settings.last_attempt if settings is not None else None
|
||
_remember(ip, True)
|
||
if previous:
|
||
log.info(
|
||
"Успешный вход в админку (%s). Предыдущая попытка: %s, %s, %s",
|
||
ip,
|
||
_human_dt(previous.get("at")),
|
||
previous.get("ip"),
|
||
"успешная" if previous.get("ok") else "неудачная",
|
||
)
|
||
else:
|
||
log.info("Успешный вход в админку (%s)", ip)
|
||
response.set_cookie(
|
||
COOKIE_NAME,
|
||
make_token(config.login, config.password),
|
||
max_age=SESSION_TTL_SECONDS,
|
||
httponly=True,
|
||
samesite="lax",
|
||
secure=_secure(request),
|
||
path="/",
|
||
)
|
||
return {"ok": True}
|
||
|
||
@router.post("/logout")
|
||
def logout(response: Response) -> dict:
|
||
response.delete_cookie(COOKIE_NAME, path="/")
|
||
return {"ok": True}
|
||
|
||
@router.get("/state", dependencies=[Depends(require_auth)])
|
||
def state(request: Request) -> dict:
|
||
ip = _client_ip(request)
|
||
until = throttle.locked_until(ip)
|
||
payload = cache.state()
|
||
# путь к панели здесь НЕ отдаём: он не должен попадать ни в JSON, ни в
|
||
# HTML — переход в панель идёт через /api/admin/goto (редирект)
|
||
payload["throttle"] = {
|
||
"max_attempts": config.max_attempts,
|
||
"lockout_minutes": config.lockout_minutes,
|
||
"attempts_left": throttle.attempts_left(ip),
|
||
"locked_until": datetime.fromtimestamp(until).astimezone().isoformat() if until else None,
|
||
}
|
||
return payload
|
||
|
||
@router.get("/goto", dependencies=[Depends(require_auth)])
|
||
def goto() -> Response:
|
||
"""Переход в панель.
|
||
|
||
Так секретный путь не хранится в разметке главной страницы: кнопка
|
||
«Админка» дёргает этот адрес, и браузер уже получает редирект.
|
||
Без валидной сессии ответ — 401, то есть ботам тут ничего не видно.
|
||
"""
|
||
if not config.path:
|
||
raise HTTPException(status_code=500, detail="ADMIN_PATH не задан")
|
||
return RedirectResponse(
|
||
url=config.path, status_code=302, headers={"Cache-Control": "no-store"}
|
||
)
|
||
|
||
@router.post("/refresh", dependencies=[Depends(require_auth)])
|
||
def refresh() -> dict:
|
||
try:
|
||
return cache.force_reload()
|
||
except ScheduleError as exc:
|
||
raise HTTPException(status_code=502, detail=str(exc))
|
||
|
||
@router.post("/link", dependencies=[Depends(require_auth)])
|
||
def set_link(body: LinkBody) -> dict:
|
||
try:
|
||
return cache.apply_link(body.link)
|
||
except ScheduleError as exc:
|
||
raise HTTPException(status_code=400, detail=str(exc))
|
||
|
||
@router.post("/interval", dependencies=[Depends(require_auth)])
|
||
def set_interval(body: IntervalBody) -> dict:
|
||
try:
|
||
minutes = cache.set_interval_minutes(body.interval_minutes)
|
||
except ScheduleError as exc:
|
||
raise HTTPException(status_code=400, detail=str(exc))
|
||
return {"interval_minutes": minutes}
|
||
|
||
return router
|