netinvent/npbackup
1
1
Fork 0
mirror of https://github.com/netinvent/npbackup.git synced 2025-11-10 14:11:23 +08:00
Code Issues Projects Releases Packages Wiki Activity

Merge branch 'main' into rc1-bughunt

Browse source
This commit is contained in:
Orsiris de Jong 2024-06-17 16:32:33 +02:00 • committed by GitHub
commit e0cc6af034
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
4 changed files with 66 additions and 28 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

50
npbackup/configuration.py
View file

@ -129,6 +129,7 @@ ENCRYPTED_OPTIONS = [
# This is what a config file looks like
empty_config_dict = {
"conf_version": MAX_CONF_VERSION,
"audience": None,
"repos": {
"default": {
"repo_uri": None,
@ -233,6 +234,13 @@ empty_config_dict = {
},
}
def convert_to_commented_map(
source_dict,
):
if isinstance(source_dict, dict):
return CommentedMap({k: convert_to_commented_map(v) for k, v in source_dict.items()})
else:
return source_dict
def get_default_config() -> dict:
"""
@ -240,15 +248,23 @@ def get_default_config() -> dict:
"""
full_config = deepcopy(empty_config_dict)
def convert_to(
source_dict,
):
if isinstance(source_dict, dict):
return CommentedMap({k: convert_to(v) for k, v in source_dict.items()})
else:
return source_dict
return convert_to_commented_map(full_config)
return convert_to(full_config)
def get_default_repo_config() -> dict:
"""
Returns a repo config dict as nested CommentedMaps (used by ruamel.yaml to keep comments intact)
"""
repo_config = deepcopy(empty_config_dict["repos"]["default"])
return convert_to_commented_map(repo_config)
def get_default_group_config() -> dict:
"""
Returns a group config dict as nested CommentedMaps (used by ruamel.yaml to keep comments intact)
"""
group_config = deepcopy(empty_config_dict["groups"]["default_group"])
return convert_to_commented_map(group_config)
def key_should_be_encrypted(key: str, encrypted_options: List[str]):
@ -471,6 +487,9 @@ def extract_permissions_from_full_config(full_config: dict) -> dict:
full_config.s(f"{object_type}.{object_name}.manager_password", manager_password)
else:
logger.info(f"No extra information for {object_type} {object_name} found")
# If no permissions are set, we get to use default permissions
full_config.s(f"repos.{repo}.permissions", empty_config_dict["repos"]["default"]["permissions"])
full_config.s(f"repos.{repo}.manager_password", None)
return full_config
@ -595,15 +614,15 @@ def get_repo_config(
_config_inheritance.g(key)[v] = False
else:
# repo_config may or may not already contain data
if _repo_config is None:
if _repo_config is None or _repo_config == "":
_repo_config = CommentedMap()
_config_inheritance = CommentedMap()
if _repo_config.g(key) is None:
if _repo_config.g(key) is None or _repo_config.g(key) == "":
_repo_config.s(key, value)
_config_inheritance.s(key, True)
# Case where repo_config contains list but group info has single str
elif (
isinstance(_repo_config.g(key), list) and value is not None
isinstance(_repo_config.g(key), list) and value is not None and value != ""
):
merged_lists = _repo_config.g(key) + [value]
@ -826,9 +845,9 @@ def load_config(config_file: Path) -> Optional[dict]:
def save_config(config_file: Path, full_config: dict) -> bool:
try:
full_config = inject_permissions_into_full_config(full_config)
full_config.s("audience", "private" if IS_PRIV_BUILD else "public")
with open(config_file, "w", encoding="utf-8") as file_handle:
full_config = inject_permissions_into_full_config(full_config)
if not is_encrypted(full_config):
full_config = crypt_config(
full_config, AES_KEY, ENCRYPTED_OPTIONS, operation="encrypt"
@ -894,8 +913,9 @@ def get_anonymous_repo_config(repo_config: dict, show_encrypted: bool = False) -
value = "__(o_O)__"
return value
# NPF-SEC-00008: Don't show manager password / sensible data with --show-config
repo_config.pop("manager_password", None)
# NPF-SEC-00008: Don't show manager password / sensible data with --show-config unless it's empty
if repo_config.get("manager_password", None):
repo_config["manager_password"] = "__(x_X)__"
repo_config.pop("update_manager_password", None)
if show_encrypted:
return repo_config

5
npbackup/core/runner.py
View file

@ -472,6 +472,7 @@ class NPBackupRunner:
"restore": ["restore", "full"],
"dump": ["restore", "full"],
"check": ["restore", "full"],
"init": ["full"],
"list": ["full"],
"unlock": ["full"],
"repair": ["full"],
@ -759,13 +760,13 @@ class NPBackupRunner:
self.write_logs("Bogus additional parameters given", level="warning")
try:
env_variables = self.repo_config.g("env.variables")
env_variables = self.repo_config.g("env.env_variables")
if not isinstance(env_variables, list):
env_variables = [env_variables]
except KeyError:
env_variables = []
try:
encrypted_env_variables = self.repo_config.g("env.encrypted_variables")
encrypted_env_variables = self.repo_config.g("env.encrypted_env_variables")
if not isinstance(encrypted_env_variables, list):
encrypted_env_variables = [encrypted_env_variables]
except KeyError:

29
npbackup/gui/config.py
View file

@ -7,7 +7,7 @@ __intname__ = "npbackup.gui.config"
__author__ = "Orsiris de Jong"
__copyright__ = "Copyright (C) 2022-2024 NetInvent"
__license__ = "GPL-3.0-only"
__build__ = "2024051001"
__build__ = "2024061601"
from typing import List, Tuple
@ -157,18 +157,22 @@ def config_gui(full_config: dict, config_file: str):
)
continue
full_config.s(f"{object_type}.{object_name}", CommentedMap())
elif object_type == "groups":
if full_config.g(f"{object_type}.{object_name}"):
full_config.s(f"{object_type}.{object_name}", configuration.get_default_repo_config())
elif object_type == "groups":
if full_config.g(f"{object_type}.{object_name}"):
sg.PopupError(
_t("config_gui.group_already_exists"), keep_on_top=True
)
continue
full_config.s(f"groups.{object_name}", CommentedMap())
full_config.s(f"groups.{object_name}", configuration.get_default_group_config())
else:
raise ValueError("Bogus object type given")
break
window.close()
update_object_gui(full_config, None, unencrypted=False)
return full_config
return full_config, object_name, object_type
def delete_object(full_config: dict, object_name: str) -> dict:
object_type, object_name = get_object_from_combo(object_name)
@ -180,10 +184,17 @@ def config_gui(full_config: dict, config_file: str):
update_object_gui(full_config, None, unencrypted=False)
return full_config
def update_object_selector() -> None:
objects = get_objects()
window["-OBJECT-SELECT-"].Update(objects)
window["-OBJECT-SELECT-"].Update(value=objects[0])
def update_object_selector(object_name: str = None, object_type: str = None) -> None:
object_list = get_objects()
if not object_name or not object_type:
object = object_list[0]
else:
object = f"{object_type.capitalize()}: {object_name}"
print(object_list)
print(object)
window["-OBJECT-SELECT-"].Update(values=object_list)
window["-OBJECT-SELECT-"].Update(value=object)
def get_object_from_combo(combo_value: str) -> Tuple[str, str]:
"""
@ -1926,8 +1937,8 @@ Google Cloud storage: GOOGLE_PROJECT_ID GOOGLE_APPLICATION_CREDENTIALS\n\
update_object_selector()
continue
if event == "-OBJECT-CREATE-":
full_config = create_object(full_config)
update_object_selector()
full_config, object_name, object_type = create_object(full_config)
update_object_selector(object_name, object_type)
continue
if event == "--SET-PERMISSIONS--":
manager_password = configuration.get_manager_password(

10
npbackup/restic_wrapper/__init__.py
View file

@ -7,8 +7,8 @@ __intname__ = "npbackup.restic_wrapper"
__author__ = "Orsiris de Jong"
__copyright__ = "Copyright (C) 2022-2024 NetInvent"
__license__ = "GPL-3.0-only"
__build__ = "2024060101"
__version__ = "2.2.0"
__build__ = "2024061601"
__version__ = "2.2.1"
from typing import Tuple, List, Optional, Callable, Union
@ -25,6 +25,7 @@ from command_runner import command_runner
from ofunctions.misc import BytesConverter, fn_name
from npbackup.__debug__ import _DEBUG
from npbackup.__env__ import FAST_COMMANDS_TIMEOUT, CHECK_INTERVAL
from npbackup.path_helper import CURRENT_DIR
logger = getLogger()
@ -823,6 +824,11 @@ class ResticRunner:
if exclude_file:
if os.path.isfile(exclude_file):
cmd += f' --{case_ignore_param}exclude-file "{exclude_file}"'
elif os.path.isfile(os.path.join(CURRENT_DIR, os.path.basename(exclude_file))):
cmd += f' --{case_ignore_param}exclude-file "{os.path.join(CURRENT_DIR, os.path.basename(exclude_file))}"'
self.write_logs(
f"Expanding exclude file path to {CURRENT_DIR}", level="info"
)
else:
self.write_logs(
f"Exclude file '{exclude_file}' not found", level="error"