diff --git a/npbackup/core/upgrade_runner.py b/npbackup/core/upgrade_runner.py index 0093189..7bea344 100644 --- a/npbackup/core/upgrade_runner.py +++ b/npbackup/core/upgrade_runner.py @@ -7,24 +7,93 @@ __intname__ = "npbackup.gui.core.upgrade_runner" __author__ = "Orsiris de Jong" __copyright__ = "Copyright (C) 2022-2024 NetInvent" __license__ = "GPL-3.0-only" -__build__ = "2023040401" +__build__ = "2024041801" +import os +from typing import Optional +import tempfile from logging import getLogger -from npbackup import configuration from npbackup.upgrade_client.upgrader import auto_upgrader, _check_new_version from npbackup.__version__ import __version__ as npbackup_version +from npbackup.path_helper import CURRENT_DIR logger = getLogger() +def need_upgrade(upgrade_interval: int) -> bool: + """ + Basic counter which allows an upgrade only every X times this is called so failed operations won't end in an endless upgrade loop + + We need to make to select a write counter file that is writable + So we actually test a local file and a temp file (less secure for obvious reasons) + We just have to make sure that once we can write to one file, we stick to it unless proven otherwise + + The for loop logic isn't straight simple, but allows file fallback + """ + # file counter, local, home, or temp if not available + counter_file = "npbackup.autoupgrade.log" + + def _write_count(file: str, count: int) -> bool: + try: + with open(file, "w") as fpw: + fpw.write(str(count)) + return True + except OSError: + # We may not have write privileges, hence we need a backup plan + return False + + def _get_count(file: str) -> Optional[int]: + try: + with open(file, "r") as fpr: + count = int(fpr.read()) + return count + except OSError: + # We may not have read privileges + None + except ValueError: + logger.error("Bogus upgrade counter in %s", file) + return None + + try: + upgrade_interval = int(upgrade_interval) + except ValueError: + logger.error("Bogus upgrade interval given. Will not upgrade") + return False + + for file in [ + os.path.join(CURRENT_DIR, counter_file), + os.path.join(tempfile.gettempdir(), counter_file), + ]: + if not os.path.isfile(file): + if _write_count(file, 1): + logger.debug("Initial upgrade counter written to %s", file) + else: + logger.debug("Cannot write to upgrade counter file %s", file) + continue + count = _get_count(file) + # Make sure we can write to the file before we make any assumptions + result = _write_count(file, count + 1) + if result: + if count >= upgrade_interval: + # Reinitialize upgrade counter before we actually approve upgrades + if _write_count(file, 1): + logger.info("Auto upgrade has decided upgrade check is required") + return True + break + else: + logger.debug("Cannot write upgrade counter to %s", file) + continue + return False + + def check_new_version(full_config: dict) -> bool: upgrade_url = full_config.g("global_options.auto_upgrade_server_url") username = full_config.g("global_options.auto_upgrade_server_username") password = full_config.g("global_options.auto_upgrade_server_password") if not upgrade_url or not username or not password: - logger.error(f"Missing auto upgrade info, cannot launch auto upgrade") + logger.warning(f"Missing auto upgrade info, cannot launch auto upgrade") return None else: return _check_new_version(upgrade_url, username, password) @@ -35,7 +104,7 @@ def run_upgrade(full_config: dict) -> bool: username = full_config.g("global_options.auto_upgrade_server_username") password = full_config.g("global_options.auto_upgrade_server_password") if not upgrade_url or not username or not password: - logger.error(f"Missing auto upgrade info, cannot launch auto upgrade") + logger.warning(f"Missing auto upgrade info, cannot launch auto upgrade") return False auto_upgrade_host_identity = full_config.g( diff --git a/npbackup/upgrade_client/upgrader.py b/npbackup/upgrade_client/upgrader.py index 9e7de79..ab508a4 100644 --- a/npbackup/upgrade_client/upgrader.py +++ b/npbackup/upgrade_client/upgrader.py @@ -7,7 +7,7 @@ __intname__ = "npbackup.upgrade_client.upgrader" __author__ = "Orsiris de Jong" __copyright__ = "Copyright (C) 2023-2024 NetInvent" __license__ = "BSD-3-Clause" -__build__ = "20263040401" +__build__ = "2024041801" from typing import Optional @@ -21,7 +21,7 @@ from ofunctions.platform import get_os, python_arch from ofunctions.process import kill_childs from command_runner import deferred_command from npbackup.upgrade_client.requestor import Requestor -from npbackup.path_helper import CURRENT_DIR, CURRENT_EXECUTABLE +from npbackup.path_helper import CURRENT_EXECUTABLE from npbackup.core.nuitka_helper import IS_COMPILED from npbackup.__version__ import __version__ as npbackup_version @@ -41,72 +41,6 @@ def sha256sum_data(data): return sha256.hexdigest() -def need_upgrade(upgrade_interval: int) -> bool: - """ - Basic counter which allows an upgrade only every X times this is called so failed operations won't end in an endless upgrade loop - - We need to make to select a write counter file that is writable - So we actually test a local file and a temp file (less secure for obvious reasons) - We just have to make sure that once we can write to one file, we stick to it unless proven otherwise - - The for loop logic isn't straight simple, but allows file fallback - """ - # file counter, local, home, or temp if not available - counter_file = "npbackup.autoupgrade.log" - - def _write_count(file: str, count: int) -> bool: - try: - with open(file, "w") as fpw: - fpw.write(str(count)) - return True - except OSError: - # We may not have write privileges, hence we need a backup plan - return False - - def _get_count(file: str) -> Optional[int]: - try: - with open(file, "r") as fpr: - count = int(fpr.read()) - return count - except OSError: - # We may not have read privileges - None - except ValueError: - logger.error("Bogus upgrade counter in %s", file) - return None - - try: - upgrade_interval = int(upgrade_interval) - except ValueError: - logger.error("Bogus upgrade interval given. Will not upgrade") - return False - - for file in [ - os.path.join(CURRENT_DIR, counter_file), - os.path.join(tempfile.gettempdir(), counter_file), - ]: - if not os.path.isfile(file): - if _write_count(file, 1): - logger.debug("Initial upgrade counter written to %s", file) - else: - logger.debug("Cannot write to upgrade counter file %s", file) - continue - count = _get_count(file) - # Make sure we can write to the file before we make any assumptions - result = _write_count(file, count + 1) - if result: - if count >= upgrade_interval: - # Reinitialize upgrade counter before we actually approve upgrades - if _write_count(file, 1): - logger.info("Auto upgrade has decided upgrade check is required") - return True - break - else: - logger.debug("Cannot write upgrade counter to %s", file) - continue - return False - - def _check_new_version(upgrade_url: str, username: str, password: str) -> bool: """ Check if we have a newer version of npbackup