Yuriy Yuriev
118 lines
4.6 KiB
Python
118 lines
4.6 KiB
Python
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),
|
|
} |