import asyncio import hashlib import secrets import logging from datetime import datetime, timedelta from typing import Dict, Optional from config.settings import settings logger = logging.getLogger(__name__) ADMIN_PASSWORD = settings.ADMIN_PASSWORD class AuthManager: """Управляет авторизацией: хеширование паролей, сессии, верификация, роли.""" 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) @staticmethod def _hash_password(password: str, salt: str = None) -> tuple: if salt is None: salt = secrets.token_hex(16) result = hashlib.sha256((salt + password).encode()).hexdigest() return result, salt def verify_password(self, password: str, password_hash: str, salt: str) -> bool: computed, _ = self._hash_password(password, salt) return secrets.compare_digest(computed, password_hash) 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) 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) logger.info(f"User {user_id} logged out") 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), }