""" Mozisarok Video Uploader - Frissítés- és Fájlszinkronizáló Modul Támogatja a privát PHP 8.2 frissítési szerver (test.nyamnet.hu) integrációját: - Fájl alapú differenciális szinkronizáció (csak a módosult/hiányzó fájlok letöltése) - Automatikus SHA-256 integritás ellenőrzés fájlonként - Automatikus biztonsági mentés (backup) az aktuális fájlokról frissítés előtt - A .env fájl és a helyi felhasználói adatok szigorú védelme (sosem íródnak felül) """ import hashlib import os import re import shutil import time import zipfile from pathlib import Path from typing import Any, Dict, List, Optional import requests from config import BASE_DIR from database import get_setting, set_setting from logger_config import get_logger from version import __version__ logger = get_logger("UpdateManager") DEFAULT_UPDATE_SERVER_URL = "https://test.nyamnet.hu" BACKUP_DIR = BASE_DIR / "backups" # Védett fájlok és mappák listája, amelyeket a szinkronizáció SOHA nem írhat felül PROTECTED_PATHS = { ".env", "logs", "backups", ".git", "config.zip", "update_server", "scratch", ".system_generated", "sync_push.py", "sync_push.bat", } def get_update_server_url() -> str: """Visszaadja a beállított frissítési szerver alapértelmezett URL-jét.""" url = get_setting("update_server_url", DEFAULT_UPDATE_SERVER_URL) return str(url).strip().rstrip("/") if url else DEFAULT_UPDATE_SERVER_URL def set_update_server_url(url: str) -> str: """Menti a frissítési szerver URL-jét a központi beállításokba.""" clean_url = str(url).strip().rstrip("/") if not clean_url: clean_url = DEFAULT_UPDATE_SERVER_URL set_setting("update_server_url", clean_url) logger.info(f"Frissítési szerver URL beállítva: {clean_url}") return clean_url def parse_version_tuple(ver_str: str) -> tuple: """Számpárokká alakítja a verziószámot az összehasonlításhoz (pl. '1.2.0' -> (1, 2, 0)).""" clean = re.sub(r"[^\d.]", "", str(ver_str).strip()) parts = [] for p in clean.split("."): if p.isdigit(): parts.append(int(p)) return tuple(parts) or (0,) def compare_versions(v1: str, v2: str) -> int: """ Összehasonlít két verziószámot. Visszatérési érték: -1 : ha v1 < v2 (v2 újabb) 0 : ha v1 == v2 (azonos verzió) 1 : ha v1 > v2 (v1 újabb) """ t1 = parse_version_tuple(v1) t2 = parse_version_tuple(v2) max_len = max(len(t1), len(t2)) t1 += (0,) * (max_len - len(t1)) t2 += (0,) * (max_len - len(t2)) if t1 < t2: return -1 elif t1 > t2: return 1 return 0 def calculate_local_sha256(path: Path) -> Optional[str]: """Kiszámítja egy helyi fájl SHA-256 lenyomatát.""" try: if not path.is_file(): return None return hashlib.sha256(path.read_bytes()).hexdigest().lower() except Exception: return None def is_ssl_error(exc: Exception) -> bool: """Megállapítja, hogy a hiba SSL/TLS tanúsítvány vagy kézfogási problémából ered-e.""" if isinstance(exc, requests.exceptions.SSLError): return True err_text = str(exc).lower() ssl_keywords = ("ssl", "certificate", "cert", "handshake", "tls", "verify failed") if any(k in err_text for k in ssl_keywords): return True cur = exc while cur.__cause__ or cur.__context__: cur = cur.__cause__ or cur.__context__ if isinstance(cur, requests.exceptions.SSLError) or any(k in str(cur).lower() for k in ssl_keywords): return True return False def check_for_updates(server_url: Optional[str] = None) -> Dict[str, Any]: """ Lekérdezi a megadott vagy beállított frissítési szervertől a fájljegyzéket (manifest). Összehasonlítja a helyi fájlokkal, és megállapítja, mely fájlok módosultak vagy hiányoznak. Ha a HTTPS lekérdezés SSL hiba miatt hiúsul meg, automatikusan visszavált HTTP-re. """ target_url = (server_url or get_update_server_url()).strip().rstrip("/") urls_to_try = [ f"{target_url}/api/manifest", f"{target_url}/index.php?api=manifest", f"{target_url}/api/version", f"{target_url}/index.php?api=version", ] last_error = None response_data = None for url in urls_to_try: try: logger.info(f"Fájljegyzék lekérdezése: {url}...") resp = requests.get( url, headers={"Accept": "application/json", "User-Agent": f"MozisarokUploader/{__version__}"}, timeout=8, ) if resp.status_code == 200: data = resp.json() if isinstance(data, dict) and data.get("status") in ("success", "empty"): response_data = data break except Exception as e: if url.startswith("https://") and is_ssl_error(e): http_url = "http://" + url[8:] logger.warning(f"HTTPS SSL hiba a(z) {url} elérésekor ({e}). Váltás HTTP-re: {http_url}...") try: resp = requests.get( http_url, headers={"Accept": "application/json", "User-Agent": f"MozisarokUploader/{__version__}"}, timeout=8, ) if resp.status_code == 200: data = resp.json() if isinstance(data, dict) and data.get("status") in ("success", "empty"): response_data = data if target_url.startswith("https://"): target_url = "http://" + target_url[8:] logger.info(f"Sikeres HTTP kapcsolat a frissítési szerverrel: {target_url}") break except Exception as http_err: last_error = str(http_err) logger.debug(f"HTTP hiba is felmerült ({http_url}): {http_err}") else: last_error = str(e) logger.debug(f"Hiba a(z) {url} lekérdezésekor: {e}") if not response_data: err_msg = f"Nem sikerült kapcsolódni a frissítési szerverhez ({target_url}). Hiba: {last_error or 'Ismeretlen hiba'}" logger.warning(err_msg) return { "success": False, "update_available": False, "current_version": __version__, "latest_version": None, "server_url": target_url, "error": err_msg, } latest_ver = response_data.get("latest_version") or response_data.get("version") if not latest_ver: return { "success": True, "update_available": False, "current_version": __version__, "latest_version": __version__, "server_url": target_url, "message": "Jelenleg nincs közzétett frissítés a szerveren.", "changelog": [], "files_to_update": [], } remote_files = response_data.get("files", {}) files_to_update: List[Dict[str, Any]] = [] up_to_date_files: List[str] = [] total_download_size = 0 # Fájlonkénti differenciális vizsgálat if isinstance(remote_files, dict) and remote_files: for rel_path, meta in remote_files.items(): # Védett fájlokat sosem vizsgálunk/írunk felül parts = Path(rel_path).parts if not parts: continue if parts[0] in PROTECTED_PATHS or rel_path in PROTECTED_PATHS: continue local_file = BASE_DIR / rel_path expected_sha = (meta.get("sha256") or "").lower() file_size = int(meta.get("size") or 0) if not local_file.exists(): files_to_update.append({ "path": rel_path, "status": "missing", "size": file_size, "sha256": expected_sha, }) total_download_size += file_size else: local_sha = calculate_local_sha256(local_file) if local_sha != expected_sha: files_to_update.append({ "path": rel_path, "status": "modified", "size": file_size, "sha256": expected_sha, }) total_download_size += file_size else: up_to_date_files.append(rel_path) # Verziók összehasonlítása ver_cmp = compare_versions(__version__, latest_ver) update_available = (len(files_to_update) > 0) or (ver_cmp < 0) download_base = response_data.get("download_base_url") or f"{target_url}/file.php?path=" return { "success": True, "update_available": update_available, "sync_type": "differential", "current_version": __version__, "latest_version": latest_ver, "release_date": response_data.get("release_date") or response_data.get("updated_at", "")[:10], "min_python_version": response_data.get("min_python_version", "3.10"), "download_url": response_data.get("download_url"), "download_base_url": download_base, "total_remote_files": len(remote_files), "files_to_update_count": len(files_to_update), "files_to_update": files_to_update, "total_download_size": total_download_size, "up_to_date_count": len(up_to_date_files), "changelog": response_data.get("changelog", []), "mandatory": bool(response_data.get("mandatory", False)), "server_url": target_url, } def create_safety_backup() -> Path: """ Automatikus biztonsági mentést készít az aktuális kódokról frissítés előtt. A mentés a backups/ mappába kerül zip formátumban. """ BACKUP_DIR.mkdir(parents=True, exist_ok=True) timestamp = time.strftime("%Y%m%d_%H%M%S") backup_file = BACKUP_DIR / f"backup_v{__version__}_{timestamp}.zip" logger.info(f"Biztonsági mentés készítése: {backup_file.name}...") with zipfile.ZipFile(backup_file, "w", compression=zipfile.ZIP_DEFLATED) as z: for p in BASE_DIR.rglob("*"): if not p.is_file(): continue rel = p.relative_to(BASE_DIR) first_part = rel.parts[0] if rel.parts else "" if first_part in ("logs", "backups", ".git", "__pycache__", "update_server", "scratch", ".system_generated"): continue if p.name == ".env": continue try: z.write(p, arcname=str(rel.as_posix())) except Exception as e: logger.warning(f"Nem sikerült a mentésbe tenni a fájlt ({rel}): {e}") logger.info(f"Biztonsági mentés sikeresen elkészült: {backup_file}") return backup_file def apply_update(server_url: Optional[str] = None, force: bool = False) -> Dict[str, Any]: """ Differenciális szinkronizáció: csak a ténylegesen megváltozott vagy hiányzó fájlokat tölti le a PHP szerverről, fájlonkénti SHA-256 ellenőrzéssel. A .env és más privát konfigurációk szigorúan védettek és NEM íródnak felül! """ # 1. Frissítések vizsgálata check_info = check_for_updates(server_url) if not check_info.get("success"): raise RuntimeError(check_info.get("error", "Nem sikerült ellenőrizni a frissítéseket.")) latest_ver = check_info.get("latest_version") or __version__ files_to_update = check_info.get("files_to_update", []) if not check_info.get("update_available") and not force and not files_to_update: return { "success": True, "updated": False, "message": f"A rendszer minden fájlja naprakész (v{__version__}).", "current_version": __version__, } # 2. Biztonsági mentés készítése frissítés előtt backup_path = create_safety_backup() # 3. Differenciális letöltés fájlonként target_url = check_info.get("server_url", DEFAULT_UPDATE_SERVER_URL).rstrip("/") download_base = check_info.get("download_base_url") or f"{target_url}/file.php?path=" if target_url.startswith("http://") and download_base.startswith("https://"): download_base = "http://" + download_base[8:] updated_files: List[str] = [] downloaded_bytes = 0 logger.info(f"Differenciális fájlszinkronizáció megkezdése ({len(files_to_update)} db fájl)...") for item in files_to_update: rel_path = item["path"] expected_sha = (item.get("sha256") or "").lower() # Biztonsági ellenőrzés parts = Path(rel_path).parts if not parts or parts[0] in PROTECTED_PATHS or rel_path in PROTECTED_PATHS: logger.info(f"Védett fájl kihagyva: {rel_path}") continue file_url = f"{download_base}{rel_path}" logger.info(f"Letöltés: {rel_path} ({file_url})...") try: resp = requests.get( file_url, headers={"User-Agent": f"MozisarokUploaderSync/{__version__}"}, timeout=30, ) except Exception as e: if file_url.startswith("https://") and is_ssl_error(e): http_file_url = "http://" + file_url[8:] logger.warning(f"Letöltési SSL hiba ({file_url}), újrapróbálás HTTP-vel ({http_file_url})...") resp = requests.get( http_file_url, headers={"User-Agent": f"MozisarokUploaderSync/{__version__}"}, timeout=30, ) else: raise resp.raise_for_status() content = resp.content # SHA-256 ellenőrzés if expected_sha: actual_sha = hashlib.sha256(content).hexdigest().lower() if actual_sha != expected_sha: raise ValueError( f"Integritás hiba a(z) '{rel_path}' fájlnál! " f"Letöltött SHA: {actual_sha}, elvárt: {expected_sha}." ) target_file = BASE_DIR / rel_path target_file.parent.mkdir(parents=True, exist_ok=True) target_file.write_bytes(content) updated_files.append(rel_path) downloaded_bytes += len(content) logger.info(f"Differenciális szinkronizáció befejezve! Frissített fájlok: {len(updated_files)} db.") return { "success": True, "updated": True, "sync_type": "differential", "old_version": __version__, "new_version": latest_ver, "backup_created": str(backup_path.name), "updated_files_count": len(updated_files), "updated_files": updated_files[:20], "downloaded_bytes": downloaded_bytes, "message": f"Sikeres szinkronizáció! {len(updated_files)} db fájl frissítve a(z) v{latest_ver} állapotra.", }