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