/
HookDev-Arch
/
ServerMonitor
Обзор
Документация
Войти
/
HookDev-Arch
/
ServerMonitor
Код
Запросы
0
Задачи
Вики
Пакеты
0
Релизы
2
CI/CD
Аналитика
Безопасность
master
core/auth_manager.py
214 строк
8 KB
HookDev-Arch
upload files
09 окт 2025, 02:25
09 окт 2025, 02:25
8bd6873
Код
Авторство
О чём код?
"""Credential storage and session management helpers for the dashboard.""" from __future__ import annotations import hashlib import json import secrets import time from pathlib import Path from typing import Any, Dict, Optional from core.config import settings from core.logger import get_logger class AuthManager: """Handle user credentials backed by a JSON file plus in-memory sessions.""" def __init__(self) -> None: self.auth_file = Path(settings.auth_file) self.logger = get_logger("auth") self.sessions: Dict[str, Dict[str, Any]] = {} self._ensure_auth_file() # --------------------------------------------------------------------- # Persistence helpers # --------------------------------------------------------------------- def _ensure_auth_file(self) -> None: """Create an empty auth store when none exists yet.""" if self.auth_file.exists(): return self.auth_file.parent.mkdir(parents=True, exist_ok=True) self._save_auth_data({"users": []}) self.logger.info( "Initialized empty auth store. Use add_user.py to create the first account." ) def _hash_password(self, password: str) -> str: """Hash a password with a random salt using SHA256.""" salt = secrets.token_hex(16) digest = hashlib.sha256((password + salt).encode()).hexdigest() return f"{salt}:{digest}" def _verify_password(self, password: str, hashed: str) -> bool: """Validate a password against a salt+hash representation.""" try: salt, digest = hashed.split(":", maxsplit=1) except ValueError: return False candidate = hashlib.sha256((password + salt).encode()).hexdigest() return candidate == digest def _load_auth_data(self) -> Dict[str, Any]: """Return the auth JSON payload, or an empty structure on failure.""" try: with self.auth_file.open("r", encoding="utf-8") as handle: return json.load(handle) except Exception as exc: # pragma: no cover - defensive log path self.logger.error("Error loading auth data: %s", exc) return {"users": []} def _save_auth_data(self, data: Dict[str, Any]) -> None: """Persist the auth JSON payload to disk.""" try: with self.auth_file.open("w", encoding="utf-8") as handle: json.dump(data, handle, ensure_ascii=False, indent=2) except Exception as exc: # pragma: no cover - defensive log path self.logger.error("Error saving auth data: %s", exc) # --------------------------------------------------------------------- # User management # --------------------------------------------------------------------- def authenticate(self, username: str, password: str) -> Optional[Dict[str, Any]]: """Return user metadata when credentials are valid.""" auth_data = self._load_auth_data() for user in auth_data.get("users", []): if user.get("username") != username: continue if not user.get("is_active", True): continue if not self._verify_password(password, user.get("password_hash", "")): continue user["last_login"] = time.time() self._save_auth_data(auth_data) self.logger.info("User %s authenticated successfully", username) return { "user_id": user.get("username"), "username": user.get("username"), "created_at": user.get("created_at"), "last_login": user.get("last_login"), } self.logger.warning("Failed authentication attempt for user: %s", username) return None def change_password(self, username: str, old_password: str, new_password: str) -> bool: """Change a user's password when the old password matches.""" auth_data = self._load_auth_data() for user in auth_data.get("users", []): if user.get("username") != username: continue if not self._verify_password(old_password, user.get("password_hash", "")): break user["password_hash"] = self._hash_password(new_password) self._save_auth_data(auth_data) self.logger.info("Password changed for user: %s", username) return True self.logger.warning("Failed password change attempt for user: %s", username) return False def create_user(self, username: str, password: str) -> bool: """Create a new user if the username is not already taken.""" auth_data = self._load_auth_data() for user in auth_data.get("users", []): if user.get("username") == username: self.logger.warning("User %s already exists", username) return False new_user = { "username": username, "password_hash": self._hash_password(password), "created_at": time.time(), "last_login": None, "is_active": True, } auth_data.setdefault("users", []).append(new_user) self._save_auth_data(auth_data) self.logger.info("User %s created successfully", username) return True def delete_user(self, username: str) -> bool: """Delete a user from the auth store.""" auth_data = self._load_auth_data() users = auth_data.get("users", []) remaining = [user for user in users if user.get("username") != username] if len(remaining) == len(users): self.logger.warning("User %s does not exist", username) return False auth_data["users"] = remaining self._save_auth_data(auth_data) self.logger.info("User %s removed", username) return True def list_users(self) -> list[str]: """Return a list of usernames currently stored.""" return [user.get("username", "") for user in self._load_auth_data().get("users", [])] # --------------------------------------------------------------------- # Session management # --------------------------------------------------------------------- def create_session(self, user_data: Dict[str, Any]) -> str: """Create an in-memory session and return its identifier.""" session_id = secrets.token_urlsafe(32) self.sessions[session_id] = { "user": user_data, "created_at": time.time(), "last_activity": time.time(), } self.logger.info("Session created for user: %s", user_data["username"]) return session_id def get_session(self, session_id: str) -> Optional[Dict[str, Any]]: """Return session data when the id is known, refreshing last activity.""" session = self.sessions.get(session_id) if not session: return None session["last_activity"] = time.time() return session def destroy_session(self, session_id: str) -> None: """Remove an in-memory session if it exists.""" session = self.sessions.pop(session_id, None) if session: self.logger.info("Session destroyed for user: %s", session["user"]["username"]) def cleanup_expired_sessions(self, max_age: int = 86400) -> None: """Clean up sessions that have been idle longer than ``max_age`` seconds.""" cutoff = time.time() - max_age expired = [sid for sid, data in self.sessions.items() if data["last_activity"] < cutoff] for session_id in expired: self.destroy_session(session_id) if expired: self.logger.info("Cleaned up %s expired sessions", len(expired)) # Convenience singleton used across the application auth_mgr = AuthManager()