import json from logging import getLogger from pathlib import Path from typing import Literal from defence360agent.contracts.config import ANTIVIRUS_MODE from defence360agent.contracts.plugins import Scope logger = getLogger(__name__) BASE_DIR = Path("/var/imunify360") PERSISTENT_STATE_DIR = BASE_DIR / ".persistent_state" LOCK_FILES = set() def register_lock_file( lock_file: str, scope: Literal[Scope.AV, Scope.IM360, Scope.AV_IM360] ) -> Path: """Register lock file for further usage.""" _lock_file = PERSISTENT_STATE_DIR / f".{lock_file}.lock" if scope == Scope.AV_IM360: LOCK_FILES.add(_lock_file) elif scope == Scope.AV and ANTIVIRUS_MODE: LOCK_FILES.add(_lock_file) elif scope == Scope.IM360 and not ANTIVIRUS_MODE: LOCK_FILES.add(_lock_file) return _lock_file def save_state(class_name: str, values: dict): """Save state to a file in .persistent_state folder.""" folder_path = PERSISTENT_STATE_DIR try: folder_path.mkdir(parents=True, exist_ok=True) file_path = folder_path / f"{class_name}.state" json.dump(values, file_path.open("w")) except (AttributeError, OSError) as e: logger.error("Failed to save state: %s %s", class_name, e) def load_state(class_name) -> dict: """Load state from a file in .persistent_state folder.""" folder_path = PERSISTENT_STATE_DIR file_path = folder_path / f"{class_name}.state" if file_path.exists(): try: return json.load(file_path.open("r")) except (json.JSONDecodeError, OSError, UnicodeDecodeError) as e: logger.error("Failed to load state: %s %s", class_name, e) return dict() def remove_unused_locks(): """Remove all unused lock files from .persistent_state folder.""" for lock_file in PERSISTENT_STATE_DIR.glob("*.lock"): if lock_file not in LOCK_FILES: lock_file.unlink()