Refactor: Encrypted backups to use direct LUKS
Replaced the LVM-on-a-file implementation with a more robust, industry-standard LUKS-on-a-file approach. This change was motivated by persistent and hard-to-debug errors related to LVM state management and duplicate loop device detection during repeated mount/unmount cycles. The new implementation provides several key benefits: - **Robustness:** Eliminates the entire LVM layer, which was the root cause of the mount/unmount failures. - **Improved UX:** Drastically reduces the number of password prompts for encrypted user backups. By changing ownership of the mountpoint, rsync can run with user privileges. - **Enhanced Security:** The file transfer process (rsync) for user backups no longer runs with root privileges. - **Better Usability:** Encrypted containers are now left mounted during the application's lifecycle and are only unmounted on exit, improving workflow for consecutive operations.
This commit is contained in:
File diff suppressed because it is too large
Load Diff
@@ -7,6 +7,7 @@ import subprocess
|
||||
import tempfile
|
||||
import stat
|
||||
import re
|
||||
import math
|
||||
from typing import Optional, List
|
||||
|
||||
from core.pbp_app_config import AppConfig
|
||||
@@ -22,32 +23,17 @@ class EncryptionManager:
|
||||
self.logger = logger
|
||||
self.app = app
|
||||
self.service_id = "py-backup-encryption"
|
||||
self.session_password = None
|
||||
self.mounted_destinations = set()
|
||||
self.auth_method = None
|
||||
self.is_mounting = False
|
||||
|
||||
def get_key_file_path(self, base_dest_path: str) -> str:
|
||||
"""Generates the standard path for the key file for a given destination."""
|
||||
key_filename = f"keyfile_{os.path.basename(base_dest_path.rstrip('/'))}.key"
|
||||
return os.path.join(AppConfig.CONFIG_DIR, key_filename)
|
||||
|
||||
def get_password_from_keyring(self, username: str) -> Optional[str]:
|
||||
try:
|
||||
return keyring.get_password(self.service_id, username)
|
||||
except keyring.errors.InitError as e:
|
||||
self.logger.log(f"Could not initialize keyring: {e}")
|
||||
return None
|
||||
except Exception as e:
|
||||
self.logger.log(f"Could not get password from keyring: {e}")
|
||||
return None
|
||||
|
||||
def is_key_in_keyring(self, username: str) -> bool:
|
||||
try:
|
||||
return self.get_password_from_keyring(username) is not None
|
||||
except Exception as e:
|
||||
self.logger.log(f"Could not check password in keyring: {e}")
|
||||
return False
|
||||
return self.get_password_from_keyring(username) is not None
|
||||
|
||||
def set_password_in_keyring(self, username: str, password: str) -> bool:
|
||||
try:
|
||||
@@ -59,288 +45,195 @@ class EncryptionManager:
|
||||
return False
|
||||
|
||||
def get_password(self, username: str, confirm: bool = True) -> Optional[str]:
|
||||
if self.session_password:
|
||||
return self.session_password
|
||||
|
||||
password = self.get_password_from_keyring(username)
|
||||
if password:
|
||||
self.session_password = password
|
||||
return password
|
||||
|
||||
dialog = PasswordDialog(
|
||||
self.app, title=f"Enter password for {username}", confirm=confirm)
|
||||
dialog = PasswordDialog(self.app, title=f"Enter password for {username}", confirm=confirm)
|
||||
password, save_to_keyring = dialog.get_password()
|
||||
if password and save_to_keyring:
|
||||
self.set_password_in_keyring(username, password)
|
||||
|
||||
if password:
|
||||
self.session_password = password
|
||||
|
||||
return password
|
||||
|
||||
def is_encrypted(self, base_dest_path: str) -> bool:
|
||||
if os.path.basename(base_dest_path) == "pybackup":
|
||||
pybackup_dir = base_dest_path
|
||||
else:
|
||||
pybackup_dir = os.path.join(base_dest_path, "pybackup")
|
||||
def get_container_path(self, base_dest_path: str) -> str:
|
||||
pybackup_dir = os.path.join(base_dest_path, "pybackup")
|
||||
user_encrypt_dir = os.path.join(pybackup_dir, "user_encrypt")
|
||||
container_path = os.path.join(user_encrypt_dir, "pybackup_lvm.img")
|
||||
return os.path.exists(container_path)
|
||||
return os.path.join(user_encrypt_dir, "pybackup_luks.img")
|
||||
|
||||
def is_encrypted(self, base_dest_path: str) -> bool:
|
||||
return os.path.exists(self.get_container_path(base_dest_path))
|
||||
|
||||
def is_mounted(self, base_dest_path: str) -> bool:
|
||||
if os.path.basename(base_dest_path) == "pybackup":
|
||||
pybackup_dir = base_dest_path
|
||||
else:
|
||||
pybackup_dir = os.path.join(base_dest_path, "pybackup")
|
||||
pybackup_dir = os.path.join(base_dest_path, "pybackup")
|
||||
mount_point = os.path.join(pybackup_dir, "encrypted")
|
||||
return os.path.ismount(mount_point)
|
||||
|
||||
def mount(self, base_dest_path: str, queue=None, size_gb: int = 0) -> Optional[str]:
|
||||
if self.is_mounting:
|
||||
self.logger.log("Mount process already in progress. Aborting new request.")
|
||||
return None
|
||||
|
||||
self.is_mounting = True
|
||||
try:
|
||||
if self.is_mounted(base_dest_path):
|
||||
self.mounted_destinations.add(base_dest_path)
|
||||
pybackup_dir = os.path.join(base_dest_path, "pybackup")
|
||||
return os.path.join(pybackup_dir, "encrypted")
|
||||
|
||||
username = os.path.basename(base_dest_path.rstrip('/'))
|
||||
|
||||
# Use a dummy queue if none is provided
|
||||
if queue is None:
|
||||
from queue import Queue
|
||||
queue = Queue()
|
||||
|
||||
# 1. Try keyring
|
||||
password = self.get_password_from_keyring(username)
|
||||
if password:
|
||||
self.logger.log("Found password in keyring. Attempting to mount.")
|
||||
mount_point = self._setup_encrypted_backup(queue, base_dest_path, size_gb, password=password)
|
||||
if mount_point:
|
||||
self.auth_method = "keyring"
|
||||
self.mounted_destinations.add(base_dest_path)
|
||||
return mount_point
|
||||
else:
|
||||
# If mounting with keyring key fails, stop here and report error.
|
||||
self.logger.log("Mounting with keyring password failed. Aborting mount attempt.")
|
||||
return None
|
||||
|
||||
# 2. Try key file
|
||||
key_file_path = self.get_key_file_path(base_dest_path)
|
||||
if os.path.exists(key_file_path):
|
||||
self.logger.log(f"Found key file at {key_file_path}. Attempting to mount.")
|
||||
mount_point = self._setup_encrypted_backup(queue, base_dest_path, size_gb, key_file=key_file_path)
|
||||
if mount_point:
|
||||
self.auth_method = "keyfile"
|
||||
self.mounted_destinations.add(base_dest_path)
|
||||
return mount_point
|
||||
|
||||
# 3. Prompt for password
|
||||
self.logger.log("No password in keyring or key file found. Prompting user.")
|
||||
password = self.get_password(username, confirm=False)
|
||||
if not password:
|
||||
self.logger.log("No password provided, cannot mount container.")
|
||||
self.auth_method = None
|
||||
return None
|
||||
|
||||
mount_point = self._setup_encrypted_backup(queue, base_dest_path, size_gb, password=password)
|
||||
if mount_point:
|
||||
self.auth_method = "password"
|
||||
self.mounted_destinations.add(base_dest_path)
|
||||
return mount_point
|
||||
finally:
|
||||
self.is_mounting = False
|
||||
|
||||
def unmount(self, base_dest_path: str):
|
||||
if base_dest_path in self.mounted_destinations:
|
||||
self._unmount_encrypted_backup(base_dest_path)
|
||||
self.mounted_destinations.remove(base_dest_path)
|
||||
|
||||
def unmount_all(self):
|
||||
self.logger.log(f"Unmounting all mounted backup destinations: {self.mounted_destinations}")
|
||||
# Create a copy for safe iteration
|
||||
for path in list(self.mounted_destinations):
|
||||
self.unmount(path)
|
||||
|
||||
def _unmount_encrypted_backup(self, base_dest_path: str):
|
||||
""" Gently unmounts the container without destroying LVM structures. """
|
||||
pybackup_dir = os.path.join(base_dest_path, "pybackup")
|
||||
user_encrypt_dir = os.path.join(pybackup_dir, "user_encrypt")
|
||||
mount_point = os.path.join(pybackup_dir, "encrypted")
|
||||
container_path = os.path.join(user_encrypt_dir, "pybackup_lvm.img")
|
||||
base_name = os.path.basename(base_dest_path.rstrip('/'))
|
||||
vg_name = f"pybackup_vg_{base_name}"
|
||||
mapper_name = f"pybackup_luks_{base_name}"
|
||||
self.logger.log(f"Unmounting encrypted LVM backup for {base_dest_path}")
|
||||
find_loop_device_cmd = f"losetup -j {container_path} | cut -d: -f1"
|
||||
script = f"""
|
||||
LOOP_DEVICE=$({find_loop_device_cmd} | head -n 1)
|
||||
if mountpoint -q {mount_point}; then
|
||||
umount {mount_point} || echo "Umount failed, continuing..."
|
||||
fi
|
||||
if [ -e /dev/mapper/{mapper_name} ]; then
|
||||
cryptsetup luksClose {mapper_name} || echo "luksClose failed, continuing..."
|
||||
fi
|
||||
if vgdisplay {vg_name} >/dev/null 2>&1; then
|
||||
vgchange -an {vg_name} || echo "Could not deactivate volume group {vg_name}."
|
||||
fi
|
||||
if [ -n "$LOOP_DEVICE" ] && losetup $LOOP_DEVICE >/dev/null 2>&1; then
|
||||
losetup -d $LOOP_DEVICE || echo "Could not detach loop device $LOOP_DEVICE."
|
||||
fi
|
||||
"""
|
||||
if not self._execute_as_root(script):
|
||||
self.logger.log("Encrypted LVM backup unmount script failed.")
|
||||
|
||||
def _setup_encrypted_backup(self, queue, base_dest_path: str, size_gb: int, password: Optional[str] = None, key_file: Optional[str] = None) -> Optional[str]:
|
||||
self.logger.log(f"Setting up LVM-based encrypted container at {base_dest_path}")
|
||||
|
||||
for tool in ["cryptsetup", "losetup", "pvcreate", "vgcreate", "lvcreate", "lvextend", "resize2fs"]:
|
||||
if not shutil.which(tool):
|
||||
self.logger.log(f"Error: Required tool '{tool}' is not installed.")
|
||||
queue.put(('error', f"Required tool '{tool}' is not installed."))
|
||||
return None
|
||||
|
||||
pybackup_dir = os.path.join(base_dest_path, "pybackup")
|
||||
user_encrypt_dir = os.path.join(pybackup_dir, "user_encrypt")
|
||||
encrypted_dir = os.path.join(pybackup_dir, "encrypted")
|
||||
container_path = os.path.join(user_encrypt_dir, "pybackup_lvm.img")
|
||||
mount_point = encrypted_dir
|
||||
|
||||
base_name = os.path.basename(base_dest_path.rstrip('/'))
|
||||
vg_name = f"pybackup_vg_{base_name}"
|
||||
lv_name = "backup_lv"
|
||||
mapper_name = f"pybackup_luks_{base_name}"
|
||||
lv_path = f"/dev/{vg_name}/{lv_name}"
|
||||
|
||||
if not password and not key_file:
|
||||
self.logger.log("No password or key file provided for encryption.")
|
||||
queue.put(('error', "No password or key file provided for encryption."))
|
||||
return None
|
||||
|
||||
if os.path.ismount(mount_point):
|
||||
self.logger.log(f"Mount point {mount_point} already in use. Assuming it's correctly mounted.")
|
||||
return mount_point
|
||||
|
||||
def prepare_encrypted_destination(self, base_dest_path: str, is_system: bool, source_size: int, queue) -> Optional[str]:
|
||||
container_path = self.get_container_path(base_dest_path)
|
||||
if os.path.exists(container_path):
|
||||
auth_part = f"echo -n '{password}' | cryptsetup luksOpen {lv_path} {mapper_name} -" if password else f"cryptsetup luksOpen {lv_path} {mapper_name} --key-file {key_file}"
|
||||
script = f"""
|
||||
mkdir -p {user_encrypt_dir}
|
||||
LOOP_DEVICE=$(losetup -f --show {container_path})
|
||||
pvscan --cache
|
||||
vgchange -ay {vg_name}
|
||||
{auth_part}
|
||||
mount /dev/mapper/{mapper_name} {mount_point}
|
||||
echo "LOOP_DEVICE_PATH=$LOOP_DEVICE"
|
||||
"""
|
||||
if not self._execute_as_root(script):
|
||||
self.logger.log("Failed to unlock existing LVM container.")
|
||||
self._destroy_encrypted_structures(base_dest_path)
|
||||
return None
|
||||
self.logger.log(f"Encrypted LVM container unlocked and mounted at {mount_point}")
|
||||
return mount_point
|
||||
return self._handle_existing_container(base_dest_path, is_system, source_size, queue)
|
||||
else:
|
||||
format_auth_part = f"echo -n '{password}' | cryptsetup luksFormat {lv_path} -" if password else f"cryptsetup luksFormat {lv_path} --key-file {key_file}"
|
||||
open_auth_part = f"echo -n '{password}' | cryptsetup luksOpen {lv_path} {mapper_name} -" if password else f"cryptsetup luksOpen {lv_path} {mapper_name} --key-file {key_file}"
|
||||
script = f"""
|
||||
mkdir -p {encrypted_dir}
|
||||
mkdir -p {user_encrypt_dir}
|
||||
fallocate -l {size_gb}G {container_path}
|
||||
LOOP_DEVICE=$(losetup -f --show {container_path})
|
||||
pvcreate $LOOP_DEVICE
|
||||
vgcreate {vg_name} $LOOP_DEVICE
|
||||
lvcreate -n {lv_name} -l 100%FREE {vg_name}
|
||||
{format_auth_part}
|
||||
{open_auth_part}
|
||||
mkfs.ext4 /dev/mapper/{mapper_name}
|
||||
mount /dev/mapper/{mapper_name} {mount_point}
|
||||
echo "LOOP_DEVICE_PATH=$LOOP_DEVICE"
|
||||
"""
|
||||
if not self._execute_as_root(script):
|
||||
self.logger.log("Failed to create and setup LVM-based encrypted container.")
|
||||
self._destroy_encrypted_structures(base_dest_path)
|
||||
if os.path.exists(container_path):
|
||||
self._execute_as_root(f"rm -f {container_path}")
|
||||
queue.put(('error', "Failed to setup LVM-based encrypted container."))
|
||||
return None
|
||||
self.logger.log(f"Encrypted LVM container is ready and mounted at {mount_point}")
|
||||
return mount_point
|
||||
return self._handle_new_container(base_dest_path, is_system, source_size, queue)
|
||||
|
||||
def _destroy_encrypted_structures(self, base_dest_path: str):
|
||||
pybackup_dir = os.path.join(base_dest_path, "pybackup")
|
||||
user_encrypt_dir = os.path.join(pybackup_dir, "user_encrypt")
|
||||
mount_point = os.path.join(pybackup_dir, "encrypted")
|
||||
container_path = os.path.join(user_encrypt_dir, "pybackup_lvm.img")
|
||||
base_name = os.path.basename(base_dest_path.rstrip('/'))
|
||||
vg_name = f"pybackup_vg_{base_name}"
|
||||
mapper_name = f"pybackup_luks_{base_name}"
|
||||
self.logger.log(f"Cleaning up encrypted LVM backup for {base_dest_path}")
|
||||
find_loop_device_cmd = f"losetup -j {container_path} | cut -d: -f1"
|
||||
lv_name = "backup_lv"
|
||||
lv_path = f"/dev/{vg_name}/{lv_name}"
|
||||
def _handle_existing_container(self, base_dest_path: str, is_system: bool, source_size: int, queue) -> Optional[str]:
|
||||
self.logger.log("Handling existing container.")
|
||||
mount_point = os.path.join(os.path.dirname(self.get_container_path(base_dest_path)), "..", "encrypted")
|
||||
|
||||
if not self._open_and_mount(base_dest_path, is_system):
|
||||
self.logger.log("Failed to mount container for size check.")
|
||||
return None
|
||||
|
||||
free_space = shutil.disk_usage(mount_point).free
|
||||
required_space = int(source_size * 1.15)
|
||||
|
||||
if required_space > free_space:
|
||||
self.logger.log(f"Resize needed. Free: {free_space}, Required: {required_space}")
|
||||
queue.put(('status_update', "Container zu klein. Vergrößere..."))
|
||||
|
||||
current_total = shutil.disk_usage(mount_point).total
|
||||
needed_additional = required_space - free_space
|
||||
new_total_size = current_total + needed_additional
|
||||
|
||||
self.unmount_and_reset_owner(base_dest_path)
|
||||
|
||||
if not self._resize_container(base_dest_path, int(new_total_size)):
|
||||
self.logger.log("Failed to resize container.")
|
||||
return None
|
||||
|
||||
if not self._open_and_mount(base_dest_path, is_system):
|
||||
self.logger.log("Failed to remount after resize.")
|
||||
return None
|
||||
|
||||
self.mounted_destinations.add(base_dest_path)
|
||||
return mount_point
|
||||
|
||||
def _handle_new_container(self, base_dest_path: str, is_system: bool, source_size: int, queue) -> Optional[str]:
|
||||
self.logger.log("Handling new container creation.")
|
||||
size_gb = math.ceil((source_size * 1.2) / (1024**3)) + 5
|
||||
|
||||
username = os.path.basename(base_dest_path.rstrip('/'))
|
||||
password = self.get_password(username, confirm=True)
|
||||
if not password: return None
|
||||
|
||||
container_path = self.get_container_path(base_dest_path)
|
||||
mount_point = os.path.join(os.path.dirname(container_path), "..", "encrypted")
|
||||
mapper_name = f"pybackup_luks_{username}"
|
||||
chown_cmd = self._get_chown_command(mount_point, is_system)
|
||||
|
||||
script = f"""
|
||||
set -x # Log executed commands
|
||||
LOOP_DEVICE=$({find_loop_device_cmd} | head -n 1)
|
||||
if mountpoint -q {mount_point}; then
|
||||
umount {mount_point} || echo "Umount failed, continuing cleanup..."
|
||||
fi
|
||||
if [ -e /dev/mapper/{mapper_name} ]; then
|
||||
cryptsetup luksClose {mapper_name} || echo "luksClose failed, continuing cleanup..."
|
||||
fi
|
||||
# Deactivate and remove all LVM structures associated with the VG
|
||||
if vgdisplay {vg_name} >/dev/null 2>&1; then
|
||||
lvchange -an {lv_path} >/dev/null 2>&1 || echo "lvchange failed, continuing..."
|
||||
vgchange -an {vg_name} || echo "vgchange -an failed, continuing cleanup..."
|
||||
lvremove -f {vg_name} || echo "lvremove failed, continuing cleanup..."
|
||||
vgremove -f {vg_name} || echo "vgremove failed, continuing cleanup..."
|
||||
fi
|
||||
if [ -n "$LOOP_DEVICE" ] && losetup $LOOP_DEVICE >/dev/null 2>&1; then
|
||||
pvremove -f $LOOP_DEVICE || echo "pvremove failed, continuing cleanup..."
|
||||
losetup -d $LOOP_DEVICE || echo "losetup -d failed, continuing cleanup..."
|
||||
fi
|
||||
"""
|
||||
if not self._execute_as_root(script):
|
||||
self.logger.log("Encrypted LVM backup cleanup script failed.")
|
||||
mkdir -p \"{os.path.dirname(container_path)}\"\n mkdir -p \"{mount_point}\"\n truncate -s {int(size_gb)}G \"{container_path}\"\n echo -n \"$LUKSPASS\" | cryptsetup luksFormat \"{container_path}\" -
|
||||
echo -n \"$LUKSPASS\" | cryptsetup luksOpen \"{container_path}\" {mapper_name} -
|
||||
mkfs.ext4 \"/dev/mapper/{mapper_name}\"
|
||||
mount \"/dev/mapper/{mapper_name}\" \"{mount_point}\"\n {chown_cmd}\n """
|
||||
if not self._execute_as_root(script, password):
|
||||
return None
|
||||
|
||||
self.mounted_destinations.add(base_dest_path)
|
||||
return mount_point
|
||||
|
||||
def _execute_as_root(self, script_content: str) -> bool:
|
||||
def _open_and_mount(self, base_dest_path: str, is_system: bool) -> bool:
|
||||
username = os.path.basename(base_dest_path.rstrip('/'))
|
||||
password = self.get_password(username, confirm=False)
|
||||
if not password: return False
|
||||
|
||||
container_path = self.get_container_path(base_dest_path)
|
||||
mount_point = os.path.join(os.path.dirname(container_path), "..", "encrypted")
|
||||
mapper_name = f"pybackup_luks_{username}"
|
||||
chown_cmd = self._get_chown_command(mount_point, is_system)
|
||||
|
||||
script = f"""
|
||||
umount -l \"{mount_point}\" || true
|
||||
cryptsetup luksClose {mapper_name} || true
|
||||
mkdir -p \"{mount_point}\"\n echo -n \"$LUKSPASS\" | cryptsetup luksOpen \"{container_path}\" {mapper_name} -
|
||||
mount \"/dev/mapper/{mapper_name}\" \"{mount_point}\"\n {chown_cmd}\n """
|
||||
return self._execute_as_root(script, password)
|
||||
|
||||
def _resize_container(self, base_dest_path: str, new_size_bytes: int) -> bool:
|
||||
username = os.path.basename(base_dest_path.rstrip('/'))
|
||||
password = self.get_password(username, confirm=False)
|
||||
if not password: return False
|
||||
|
||||
container_path = self.get_container_path(base_dest_path)
|
||||
mapper_name = f"pybackup_luks_{username}"
|
||||
|
||||
script = f"""
|
||||
truncate -s {new_size_bytes} \"{container_path}\"\n echo -n \"$LUKSPASS\" | cryptsetup luksOpen \"{container_path}\" {mapper_name} -
|
||||
e2fsck -fy \"/dev/mapper/{mapper_name}\"\n resize2fs \"/dev/mapper/{mapper_name}\"\n cryptsetup luksClose {mapper_name}
|
||||
"""
|
||||
return self._execute_as_root(script, password)
|
||||
|
||||
def unmount_and_reset_owner(self, base_dest_path: str, force_unmap=False):
|
||||
mapper_name = f"pybackup_luks_{os.path.basename(base_dest_path.rstrip('/'))}"
|
||||
if not os.path.exists(f"/dev/mapper/{mapper_name}") and not self.is_mounted(base_dest_path):
|
||||
if not force_unmap: return
|
||||
|
||||
self.logger.log(f"Unmounting and resetting owner for {base_dest_path}")
|
||||
container_path = self.get_container_path(base_dest_path)
|
||||
mount_point = os.path.join(os.path.dirname(container_path), "..", "encrypted")
|
||||
|
||||
script = f"""
|
||||
chown root:root \"{mount_point}\" || true
|
||||
umount -l \"{mount_point}\" || true
|
||||
cryptsetup luksClose {mapper_name} || true
|
||||
"""
|
||||
self._execute_as_root(script)
|
||||
|
||||
if base_dest_path in self.mounted_destinations:
|
||||
self.mounted_destinations.remove(base_dest_path)
|
||||
|
||||
def unmount_all(self):
|
||||
self.logger.log(f"Unmounting all: {self.mounted_destinations}")
|
||||
for path in list(self.mounted_destinations):
|
||||
self.unmount_and_reset_owner(path, force_unmap=True)
|
||||
|
||||
def _get_chown_command(self, mount_point: str, is_system: bool) -> str:
|
||||
if not is_system:
|
||||
try:
|
||||
uid = os.getuid()
|
||||
gid = os.getgid()
|
||||
return f"chown {uid}:{gid} \"{mount_point}\""
|
||||
except Exception as e:
|
||||
self.logger.log(f"Could not get current user UID/GID for chown: {e}")
|
||||
return ""
|
||||
|
||||
def _execute_as_root(self, script_content: str, password_for_stdin: Optional[str] = None) -> bool:
|
||||
script_path = ''
|
||||
try:
|
||||
if password_for_stdin:
|
||||
escaped_password = password_for_stdin.replace("'", "'\\''")
|
||||
script_content = f"LUKSPASS='{escaped_password}'\n{script_content}"
|
||||
|
||||
with tempfile.NamedTemporaryFile(mode='w', delete=False, prefix="pybackup_script_", suffix=".sh", dir="/tmp") as tmp_script:
|
||||
tmp_script.write("#!/bin/bash\n\n")
|
||||
tmp_script.write("set -e\n\n")
|
||||
tmp_script.write(script_content)
|
||||
tmp_script.write("#!/bin/bash\n" + script_content)
|
||||
script_path = tmp_script.name
|
||||
|
||||
os.chmod(script_path, stat.S_IRWXU | stat.S_IRGRP |
|
||||
stat.S_IXGRP | stat.S_IROTH | stat.S_IXOTH)
|
||||
|
||||
os.chmod(script_path, stat.S_IRWXU)
|
||||
command = ['pkexec', script_path]
|
||||
|
||||
sanitized_script_content = re.sub(
|
||||
r"echo -n \'.*?\'", "echo -n \'[REDACTED]\'", script_content)
|
||||
self.logger.log(
|
||||
f"Executing privileged command via script: {script_path}")
|
||||
self.logger.log(
|
||||
f"Script content:\n---\n{sanitized_script_content}\n---")
|
||||
log_lines = []
|
||||
for line in script_content.split('\n'):
|
||||
if "LUKSPASS=" in line:
|
||||
log_lines.append("LUKSPASS='[REDACTED]'")
|
||||
else:
|
||||
log_lines.append(line)
|
||||
sanitized_script_content = "\n".join(log_lines)
|
||||
|
||||
result = subprocess.run(
|
||||
command, capture_output=True, text=True, check=False)
|
||||
self.logger.log(f"Executing privileged command via script: {script_path}")
|
||||
self.logger.log(f"Script content:\n---\n{sanitized_script_content}\n---")
|
||||
|
||||
result = subprocess.run(command, capture_output=True, text=True, check=False)
|
||||
|
||||
if result.returncode == 0:
|
||||
self.logger.log(
|
||||
f"Privileged script executed successfully. Output:\n{result.stdout}")
|
||||
self.logger.log(f"Privileged script executed successfully. Output:\n{result.stdout}")
|
||||
return True
|
||||
else:
|
||||
self.logger.log(
|
||||
f"Privileged script failed. Return code: {result.returncode}\nStderr: {result.stderr}\nStdout: {result.stdout}")
|
||||
self.logger.log(f"Privileged script failed. Return code: {result.returncode}\nStderr: {result.stderr}\nStdout: {result.stdout}")
|
||||
return False
|
||||
except Exception as e:
|
||||
self.logger.log(
|
||||
f"Failed to set up or execute privileged command: {e}")
|
||||
self.logger.log(f"Failed to set up or execute privileged command: {e}")
|
||||
return False
|
||||
finally:
|
||||
if script_path and os.path.exists(script_path):
|
||||
os.remove(script_path)
|
||||
try:
|
||||
os.remove(script_path)
|
||||
except OSError as e:
|
||||
self.logger.log(f"Error removing temp script {script_path}: {e}")
|
||||
@@ -141,7 +141,7 @@ class BackupContentFrame(ttk.Frame):
|
||||
button.configure(style="Gray.Toolbutton")
|
||||
self.nav_progress_bars[i].pack_forget()
|
||||
|
||||
def show(self, backup_path):
|
||||
def show(self, backup_path, initial_tab_index=0):
|
||||
app_logger.log(
|
||||
f"BackupContentFrame: show called with path {backup_path}")
|
||||
self.grid(row=2, column=0, sticky="nsew")
|
||||
@@ -151,21 +151,7 @@ class BackupContentFrame(ttk.Frame):
|
||||
# Check if the destination is encrypted and trigger mount if necessary
|
||||
is_encrypted = self.backup_manager.encryption_manager.is_encrypted(
|
||||
backup_path)
|
||||
if is_encrypted and not self.backup_manager.encryption_manager.is_mounted(backup_path):
|
||||
app_logger.log(
|
||||
"Encrypted destination is not mounted. Attempting to mount.")
|
||||
mount_point = self.backup_manager.encryption_manager.mount(
|
||||
backup_path)
|
||||
if not mount_point:
|
||||
app_logger.log("Mount failed. Cannot display backup content.")
|
||||
MessageDialog(
|
||||
message_type="error", title=Msg.STR["error"], text=Msg.STR["err_unlock_failed"])
|
||||
# Clear views and return if mount fails
|
||||
self.system_backups_frame.show(backup_path, [])
|
||||
self.user_backups_frame.show(backup_path, [])
|
||||
return
|
||||
# Refresh header status after successful mount
|
||||
self.app.header_frame.refresh_status()
|
||||
self.viewing_encrypted = is_encrypted # Set this flag for remembering the view
|
||||
|
||||
pybackup_dir = os.path.join(backup_path, "pybackup")
|
||||
|
||||
@@ -187,9 +173,8 @@ class BackupContentFrame(ttk.Frame):
|
||||
self.system_backups_frame.show(backup_path, [])
|
||||
self.user_backups_frame.show(backup_path, [])
|
||||
|
||||
config_key = "last_encrypted_backup_content_view" if self.viewing_encrypted else "last_backup_content_view"
|
||||
last_view = self.app.config_manager.get_setting(config_key, 0)
|
||||
self._switch_view(last_view)
|
||||
# Use the passed index to switch to the correct view
|
||||
self._switch_view(initial_tab_index)
|
||||
|
||||
def hide(self):
|
||||
self.grid_remove()
|
||||
|
||||
@@ -269,10 +269,9 @@ class Navigation:
|
||||
self.app.top_bar.grid()
|
||||
self._update_task_bar_visibility("settings")
|
||||
|
||||
def toggle_backup_content_frame(self, active_index=None):
|
||||
def toggle_backup_content_frame(self, initial_tab_index=0):
|
||||
self._cancel_calculation()
|
||||
if active_index is not None:
|
||||
self.app.drawing.update_nav_buttons(active_index)
|
||||
self.app.drawing.update_nav_buttons(2) # Index 2 for Backup Content
|
||||
|
||||
if not self.app.destination_path:
|
||||
MessageDialog(master=self.app, message_type="info",
|
||||
@@ -283,7 +282,9 @@ class Navigation:
|
||||
# Mount the destination if it is encrypted and not already mounted
|
||||
if self.app.backup_manager.encryption_manager.is_encrypted(self.app.destination_path):
|
||||
if not self.app.backup_manager.encryption_manager.is_mounted(self.app.destination_path):
|
||||
mount_point = self.app.backup_manager.encryption_manager.mount(self.app.destination_path)
|
||||
is_system = (initial_tab_index == 0)
|
||||
mount_point = self.app.backup_manager.encryption_manager.prepare_encrypted_destination(
|
||||
self.app.destination_path, is_system=is_system, source_size=0, queue=self.app.queue)
|
||||
if not mount_point:
|
||||
MessageDialog(master=self.app, message_type="error",
|
||||
title=Msg.STR["error"], text=Msg.STR.get("err_mount_failed", "Mounting failed."))
|
||||
@@ -301,8 +302,6 @@ class Navigation:
|
||||
self.app.target_size_frame.grid_remove()
|
||||
self.app.restore_size_frame_before.grid_remove()
|
||||
self.app.restore_size_frame_after.grid_remove()
|
||||
self.app.backup_content_frame.show(self.app.destination_path)
|
||||
self.app.backup_content_frame.show(self.app.destination_path, initial_tab_index)
|
||||
self.app.top_bar.grid()
|
||||
self._update_task_bar_visibility("scheduler")
|
||||
|
||||
|
||||
self._update_task_bar_visibility("scheduler")
|
||||
Reference in New Issue
Block a user