ccdef065df
Bash scripts were untestable and duplicated device/LUKS/mount logic; the lim/ package centralizes it behind one subprocess wrapper and a YAML image catalog (single point of truth). BREAKING CHANGE: scripts/*.sh removed. Use `lim --type <cmd>`; new types mount/umount/single-boot/raid1-boot/lock/unlock/import/export replace direct script calls. --extra is deprecated and ignored. - distributions.yml + lim/catalog.py hold the image catalog (PyYAML) - pytest suite: 102 tests with mocked subprocess (tests/unit) and a 250-line max file-length guard (tests/lint) - ruff strict (select ALL), GitHub Actions CI, Dependabot; Travis gone - Makefile: install (symlink ~/.local/bin/lim) and test targets - fixes over bash: SUDO_USER-aware chown, mmcblk/nvme partition paths, sha512 checksum support, whole-pipeline failure detection, blkid UUID fallback for pre-mounted images, conditional fstab seeding for PARTUUID/LABEL images, clean errors for missing binaries
136 lines
4.5 KiB
Python
136 lines
4.5 KiB
Python
"""Import personal data from the system into the encrypted store, or export it back."""
|
|
|
|
import time
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
|
|
from lim import config, runner, system, ui
|
|
from lim.data import crypt
|
|
from lim.errors import LimError
|
|
|
|
# Paths relative to $HOME; trailing "/" marks a directory tree.
|
|
BACKUP_ITEMS = [
|
|
".ssh/",
|
|
".gitconfig",
|
|
".atom/config.cson",
|
|
".projectlibre/projectlibre.conf",
|
|
".local/share/rhythmbox/rhythmdb.xml",
|
|
".config/keepassxc/keepassxc.ini",
|
|
"Documents/certificates/",
|
|
"Documents/security/",
|
|
"Documents/identity/",
|
|
"Documents/health/",
|
|
"Documents/licenses/",
|
|
]
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class SyncOperation:
|
|
source: str
|
|
destination: str
|
|
backup_dir: str
|
|
is_directory: bool
|
|
|
|
|
|
def build_sync_plan(
|
|
mode: str,
|
|
home: Path,
|
|
data_path: Path,
|
|
backup_folder: Path,
|
|
) -> list[SyncOperation]:
|
|
"""One rsync operation per existing backup item; missing sources are skipped."""
|
|
operations = []
|
|
for item in BACKUP_ITEMS:
|
|
is_directory = item.endswith("/")
|
|
system_path = home / item
|
|
# The store mirrors the absolute system path below DATA_PATH.
|
|
data_path_item = Path(f"{data_path}{system_path}")
|
|
if mode == "export":
|
|
source, destination = data_path_item, system_path
|
|
elif mode == "import":
|
|
source, destination = system_path, data_path_item
|
|
else:
|
|
raise LimError(f"Unknown sync mode: {mode}")
|
|
ui.info(f"{mode.capitalize()} data from {source} to {destination}...")
|
|
if not (source.is_file() or source.is_dir()):
|
|
ui.warning(f"{source} doesn't exist. Copying data is not possible.")
|
|
continue
|
|
if is_directory:
|
|
backup_dir = Path(f"{backup_folder}{system_path}")
|
|
else:
|
|
backup_dir = Path(f"{backup_folder}{system_path}").parent
|
|
operations.append(
|
|
SyncOperation(
|
|
source=f"{source}/" if is_directory else str(source),
|
|
destination=f"{destination}/" if is_directory else str(destination),
|
|
backup_dir=str(backup_dir),
|
|
is_directory=is_directory,
|
|
)
|
|
)
|
|
return operations
|
|
|
|
|
|
def rsync_command(operation: SyncOperation) -> list[str]:
|
|
command = ["rsync", "-abcEPuvW"]
|
|
if operation.is_directory:
|
|
command.append("--delete")
|
|
command += [f"--backup-dir={operation.backup_dir}", operation.source, operation.destination]
|
|
return command
|
|
|
|
|
|
def execute_sync_plan(operations: list[SyncOperation]) -> None:
|
|
for operation in operations:
|
|
Path(operation.backup_dir).mkdir(parents=True, exist_ok=True)
|
|
destination_dir = (
|
|
Path(operation.destination)
|
|
if operation.is_directory
|
|
else Path(operation.destination).parent
|
|
)
|
|
destination_dir.mkdir(parents=True, exist_ok=True)
|
|
if not operation.is_directory and Path(operation.destination).is_file():
|
|
ui.info("The destination file already exists!")
|
|
ui.info("Difference:")
|
|
runner.run(
|
|
["diff", operation.destination, operation.source], check=False
|
|
)
|
|
runner.run(rsync_command(operation))
|
|
|
|
|
|
def ensure_unlocked() -> None:
|
|
mounts = runner.output(["mount"], check=False)
|
|
if str(config.DECRYPTED_PATH) not in mounts:
|
|
ui.info(
|
|
f"The decrypted folder {config.DECRYPTED_PATH} is locked. "
|
|
"You need to unlock it!"
|
|
)
|
|
crypt.unlock()
|
|
|
|
|
|
def import_from_system(mode: str = "import") -> None:
|
|
ensure_unlocked()
|
|
backup_folder = (
|
|
config.BACKUP_PATH / mode / time.strftime("%Y%m%d%H%M%S")
|
|
)
|
|
backup_folder.mkdir(parents=True, exist_ok=True)
|
|
operations = build_sync_plan(
|
|
mode, system.real_home(), config.DATA_PATH, backup_folder
|
|
)
|
|
execute_sync_plan(operations)
|
|
|
|
|
|
def export_to_system() -> None:
|
|
import_from_system(mode="export")
|
|
user = system.real_user()
|
|
home = system.real_home()
|
|
ui.info("Setting right permissions for imported files...")
|
|
try:
|
|
(home / ".ssh").chmod(0o700)
|
|
for child in (home / ".ssh").rglob("*"):
|
|
child.chmod(0o700 if child.is_dir() else 0o600)
|
|
except OSError as exc:
|
|
raise LimError(f"Failed to set correct ssh permissions: {exc}") from exc
|
|
# Best effort without privilege escalation, like the original script.
|
|
chowned = runner.run(["chown", "-R", f"{user}:{user}", str(home)], check=False)
|
|
if chowned.returncode != 0:
|
|
ui.warning(f'Not all files could be owned by user "{user}"...')
|