""" Mozisarok Video Uploader - Biztonsági & Hitelesítési Modul (Admin Auth) Kiemelt biztonság: - PBKDF2-HMAC-SHA256 jelszóhashelés 600.000 iterációval (OWASP ajánlás) - Kriptográfiai 32 bájtos só (salt) generálás - Időzítés-alapú támadások elleni védelem (hmac.compare_digest) - Brute-force védelem (5 sikertelen kísérlet után 15 perc IP-cím alapú kizárás) - Biztonságos 256 bites munkamenet (Session) token kezelés """ import hashlib import hmac import os import secrets import time from datetime import datetime from typing import Any, Dict, Optional, Tuple from database import get_admin_user, has_admin_user, set_admin_user from logger_config import get_logger logger = get_logger("AdminAuth") # Konfiguráció HASH_ALGORITHM = "sha256" HASH_ITERATIONS = 600_000 SALT_BYTES = 32 SESSION_TOKEN_BYTES = 32 SESSION_TIMEOUT_SECONDS = 7200 # 2 óra inaktivitás után lejár MAX_LOGIN_ATTEMPTS = 5 LOCKOUT_DURATION_SECONDS = 900 # 15 perc zárolás 5 sikertelen próbálkozás után # Memória-tár az aktív munkamenetekhez és sikertelen próbálkozásokhoz _ACTIVE_SESSIONS: Dict[str, Dict[str, Any]] = {} _FAILED_ATTEMPTS: Dict[str, Dict[str, Any]] = {} def hash_password(password: str, salt: Optional[str] = None) -> Tuple[str, str]: """ Hasheli a jelszót PBKDF2-HMAC-SHA256 algoritmussal és egyedi sóval. Visszatér: (password_hash_hex, salt_hex) """ if salt is None: salt_bytes = secrets.token_bytes(SALT_BYTES) salt_hex = salt_bytes.hex() else: salt_bytes = bytes.fromhex(salt) salt_hex = salt key = hashlib.pbkdf2_hmac( HASH_ALGORITHM, password.encode("utf-8"), salt_bytes, HASH_ITERATIONS, ) return key.hex(), salt_hex def verify_password(password: str, password_hash: str, salt: str) -> bool: """ Konstans idejű összehasonlítással ellenőrzi a megadott jelszót (timing-attack védelem). """ try: calculated_hash, _ = hash_password(password, salt=salt) return hmac.compare_digest(calculated_hash, password_hash) except Exception as e: logger.error(f"Hiba a jelszó ellenőrzésekor: {e}") return False def is_ip_locked(ip: str) -> Tuple[bool, int]: """ Ellenőrzi, hogy a megadott IP cím zárolva van-e túl sok sikertelen próbálkozás miatt. Visszatér: (is_locked, remaining_seconds) """ now = time.time() record = _FAILED_ATTEMPTS.get(ip) if not record: return False, 0 attempts = record.get("count", 0) last_attempt = record.get("last_attempt", 0) # Ha a zárolási idő letelt, visszaállítjuk if now - last_attempt > LOCKOUT_DURATION_SECONDS: _FAILED_ATTEMPTS.pop(ip, None) return False, 0 if attempts >= MAX_LOGIN_ATTEMPTS: remaining = int(LOCKOUT_DURATION_SECONDS - (now - last_attempt)) return True, max(1, remaining) return False, 0 def record_login_failure(ip: str): """Rögzít egy sikertelen bejelentkezési kísérletet az adott IP címről.""" now = time.time() record = _FAILED_ATTEMPTS.get(ip, {"count": 0, "last_attempt": now}) # Ha régebbi az utolsó kísérlet mint a zárolási ablak, újraszámlálunk if now - record["last_attempt"] > LOCKOUT_DURATION_SECONDS: record["count"] = 1 else: record["count"] += 1 record["last_attempt"] = now _FAILED_ATTEMPTS[ip] = record logger.warning(f"Sikertelen admin bejelentkezés [IP: {ip}] - kísérlet: {record['count']}/{MAX_LOGIN_ATTEMPTS}") def reset_login_failures(ip: str): """Sikeres bejelentkezés után törli a számlálót az IP címről.""" _FAILED_ATTEMPTS.pop(ip, None) def create_session(username: str, ip: str) -> str: """ Létrehoz egy új kriptográfiailag biztonságos munkamenetet. """ token = secrets.token_urlsafe(SESSION_TOKEN_BYTES) now = time.time() _ACTIVE_SESSIONS[token] = { "username": username, "ip": ip, "created_at": now, "last_active": now, } logger.info(f"Admin munkamenet létrehozva [{username}] [IP: {ip}]") return token def validate_session(token: Optional[str]) -> Optional[Dict[str, Any]]: """ Ellenőrzi és frissíti a munkamenetet. Ha érvénytelen vagy lejárt, None-t ad vissza. """ if not token or token not in _ACTIVE_SESSIONS: return None session = _ACTIVE_SESSIONS[token] now = time.time() # Lejárat ellenőrzése if now - session["last_active"] > SESSION_TIMEOUT_SECONDS: _ACTIVE_SESSIONS.pop(token, None) logger.info(f"Admin munkamenet lejárt [{session.get('username')}]") return None # Munkamenet frissítése session["last_active"] = now return session def revoke_session(token: str): """Érvényteleníti a megadott munkamenetet.""" if token in _ACTIVE_SESSIONS: _ACTIVE_SESSIONS.pop(token, None) def revoke_all_sessions(): """Összes munkamenet kiléptetése (pl. jelszóváltoztatáskor).""" _ACTIVE_SESSIONS.clear() def is_admin_initialized() -> bool: """Ellenőrzi, hogy létezik-e már adminisztrátori fiók az adatbázisban.""" try: return has_admin_user() except Exception as e: logger.error(f"Nem sikerült ellenőrizni az admin fiókot: {e}") return False def setup_initial_admin(username: str, password: str) -> bool: """Kezdeti admin fiók létrehozása (csak akkor futhat, ha még nincs admin).""" if is_admin_initialized(): return False if not username.strip() or len(password) < 6: raise ValueError("A felhasználónév nem lehet üres, a jelszónak pedig legalább 6 karakternek kell lennie.") p_hash, salt = hash_password(password) set_admin_user(username.strip(), p_hash, salt) logger.info(f"Kezdeti adminisztrátori fiók létrehozva: {username.strip()}") return True def ensure_default_admin(default_user: str = "admin", default_pass: str = "admin123") -> Tuple[str, str]: """ Ha még nem létezik admin fiók, automatikusan létrehozza az alapértelmezett fiókot (admin / admin123). """ if not is_admin_initialized(): p_hash, salt = hash_password(default_pass) set_admin_user(default_user, p_hash, salt) logger.info(f"Alapértelmezett adminisztrátori fiók inicializálva: {default_user}") return default_user, default_pass def change_admin_credentials(current_password: str, new_username: str, new_password: Optional[str] = None) -> bool: """ Megváltoztatja az adminisztrátor felhasználónevét és/vagy jelszavát. A jelenlegi jelszó kötelezően megadandó a visszaélések elkerülésére. """ admin = get_admin_user() if not admin: return False if not verify_password(current_password, admin["password_hash"], admin["salt"]): return False final_username = new_username.strip() if new_username and new_username.strip() else admin["username"] if new_password and new_password.strip(): if len(new_password.strip()) < 6: raise ValueError("Az új jelszónak legalább 6 karakter hosszúnak kell lennie.") p_hash, salt = hash_password(new_password.strip()) else: p_hash, salt = admin["password_hash"], admin["salt"] set_admin_user(final_username, p_hash, salt) # Biztonsági okokból minden egyéb munkamenetet kiléptetünk revoke_all_sessions() logger.info(f"Adminisztrátori adatok sikeresen megváltoztatva: {final_username}") return True def authenticate_admin(username: str, password: str, ip: str) -> Tuple[bool, Optional[str], Optional[str]]: """ Hitelesíti a bejelentkezni kívánó admint. Visszatér: (sikeres: bool, session_token: Optional[str], hibaüzenet: Optional[str]) """ locked, remaining = is_ip_locked(ip) if locked: mins = max(1, remaining // 60) return False, None, f"Túl sok sikertelen próbálkozás. Az IP cím zárolva van még {mins} percig." admin = get_admin_user() if not admin: return False, None, "Nincs konfigurálva admin fiók. Hozd létre a kezdő fiókot!" if admin["username"] != username or not verify_password(password, admin["password_hash"], admin["salt"]): record_login_failure(ip) return False, None, "Hibás adminisztrátori felhasználónév vagy jelszó." reset_login_failures(ip) token = create_session(username, ip) return True, token, None