import json import logging import secrets from datetime import datetime, timedelta from pathlib import Path from typing import Dict, Optional from config.settings import settings logger = logging.getLogger(__name__) ADMIN_PASSWORD = settings.ADMIN_PASSWORD _SESSIONS_FILE = Path("data/sessions.json") class AuthManager: """Управляет сессиями и ролями. Личность пользователя подтверждает Telegram, поэтому пароля для входа нет. ADMIN_PASSWORD остаётся единственным секретом — им повышают роль до admin. """ def __init__(self, session_timeout_minutes: int = 120): self._sessions: Dict[int, datetime] = {} self._roles: Dict[int, str] = {} # user_id -> "admin" or "user" self._failed_attempts: Dict[int, list] = {} self._max_attempts = 5 self._lockout_duration = 300 self._session_timeout = timedelta(minutes=session_timeout_minutes) self._restore_sessions() @staticmethod def verify_admin_password(password: str) -> bool: """Проверяет пароль администратора. Пустой ADMIN_PASSWORD означает, что админ-режим отключён — иначе ненастроенный бот пускал бы в админку по пустой строке. """ if not ADMIN_PASSWORD: logger.warning("ADMIN_PASSWORD not set — admin mode is disabled") return False return secrets.compare_digest(password, ADMIN_PASSWORD) def is_authenticated(self, user_id: int) -> bool: if user_id not in self._sessions: return False # Обычные пользователи — сессия бесконечная if self._roles.get(user_id) != "admin": return True # Админы — таймаут применяется if datetime.now() - self._sessions[user_id] > self._session_timeout: self.logout(user_id) return False return True def is_admin(self, user_id: int) -> bool: return self._roles.get(user_id) == "admin" def login(self, user_id: int, is_admin: bool = False): self._sessions[user_id] = datetime.now() self._roles[user_id] = "admin" if is_admin else "user" self._failed_attempts.pop(user_id, None) self._save_sessions() logger.info(f"User {user_id} logged in as {self._roles[user_id]}") def logout(self, user_id: int): self._sessions.pop(user_id, None) self._roles.pop(user_id, None) self._failed_attempts.pop(user_id, None) self._save_sessions() logger.info(f"User {user_id} logged out") def _save_sessions(self) -> None: try: _SESSIONS_FILE.parent.mkdir(parents=True, exist_ok=True) data = {str(uid): role for uid, role in self._roles.items()} _SESSIONS_FILE.write_text(json.dumps(data), encoding="utf-8") except Exception as e: logger.warning(f"Failed to save sessions: {e}") def _restore_sessions(self) -> None: try: if not _SESSIONS_FILE.exists(): return data = json.loads(_SESSIONS_FILE.read_text(encoding="utf-8")) for uid_str, role in data.items(): uid = int(uid_str) self._sessions[uid] = datetime.now() self._roles[uid] = role logger.info(f"Restored {len(data)} sessions") except Exception as e: logger.warning(f"Failed to restore sessions: {e}") def is_locked_out(self, user_id: int) -> bool: if user_id not in self._failed_attempts: return False attempts = self._failed_attempts[user_id] recent = [t for t in attempts if datetime.now() - t < timedelta(minutes=5)] self._failed_attempts[user_id] = recent if len(recent) >= self._max_attempts: return True return False def record_failed_attempt(self, user_id: int): if user_id not in self._failed_attempts: self._failed_attempts[user_id] = [] self._failed_attempts[user_id].append(datetime.now()) def get_session_info(self, user_id: int) -> Optional[dict]: if user_id not in self._sessions: return None login_time = self._sessions[user_id] elapsed = datetime.now() - login_time return { "user_id": user_id, "login_time": login_time.isoformat(), "elapsed_minutes": int(elapsed.total_seconds() // 60), }