Yuriy Yuriev
115 lines
4.3 KiB
Python
115 lines
4.3 KiB
Python
import asyncio
|
|
import hashlib
|
|
import json
|
|
import secrets
|
|
import logging
|
|
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:
|
|
"""Управляет авторизацией: хеширование паролей, сессии, верификация, роли."""
|
|
|
|
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 _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)
|
|
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),
|
|
} |