feat: add borg-wrapper

borg-wrapper/borg-wrapper.py

@@ -0,0 +1,189 @@
+import argparse
+import json
+import os
+import subprocess
+
+"""
+Very primitive and unpolished borg wrapper for certain actions.
+
+It relies on a JSON config from which it reads borg vault locations and passwords.
+"""
+
+
+def get_destination_host(destination_hosts, backup_destination):
+    for entry in destination_hosts:
+        if entry['alias'] == backup_destination:
+            return entry['host']
+    raise Exception(f"Destination host '{backup_destination}' is missing its configuration.")
+
+
+def sync_backup(args):
+    operation = args.operation
+
+    print("Opening password file ...")
+    # Read Borg archive passwords from JSON file using sudo
+    password_file_path = '/root/borg_passwords.json'
+    password_command = f'sudo cat {password_file_path}'
+    password_data = os.popen(password_command).read().strip()
+
+    # Parse JSON data
+    passwords = json.loads(password_data)
+
+    if operation == 'create_all':
+        for entry in passwords['archives']:
+            password = entry['password']
+            destination_host_alias = entry['destination_host_alias']
+            destination_host = get_destination_host(passwords['destination_hosts'], destination_host_alias)
+            local_archive_root_path = entry['local_archive_root_path']
+            remote_archive_path = entry['remote_archive_path']
+            local_backup_source_dir = entry['local_backup_source_dir']
+
+            if args.backup_destination is not None and args.backup_destination != destination_host_alias:
+                print(f'skipping archive {entry['archive_name']} of destination {destination_host_alias}')
+                continue
+            if not args.skip_missing_source and not os.path.exists(local_archive_root_path):
+                raise RuntimeError(f"local archive root path '{local_archive_root_path}' doesn't exist")
+            elif args.skip_missing_source:
+                print(f"local archive root path is missing, skipping")
+                continue
+
+            # Set up Borg command with operation as the first argument
+            borg_env_var = f'export BORG_PASSPHRASE="{password}"'
+            borg_command = f'{borg_env_var}; cd {local_archive_root_path} && borg create --rsh \'ssh -i ~/.ssh/id_ed25519\''
+
+            if args.dry_run is True:
+                print("dry run enabled")
+            dry_run_or_stats = '--dry-run' if args.dry_run is True else '--stats'
+            if args.ignore_inode:
+                borg_command += ' --files-cache ctime,size'  # ignore inode (useful for external disks)
+            borg_command += f' --iec --progress {dry_run_or_stats} -v --compression none --checkpoint-interval=900'
+            borg_command += f' {destination_host}{remote_archive_path}::' + '{utcnow}Z ' + f'{local_backup_source_dir}'
+
+            # Execute Borg command
+            print(f'[CMD] {";".join(borg_command.split(";")[1:])}')
+            retcode = subprocess.call(borg_command, shell=True)
+            if retcode != 0:
+                raise RuntimeError("exited with non-zero return code")
+    elif operation == 'prune_all':
+        for entry in passwords['archives']:
+            password = entry['password']
+            destination_host = get_destination_host(passwords['destination_hosts'], entry['destination_host_alias'])
+            local_archive_root_path = entry['local_archive_root_path']
+            remote_archive_path = entry['remote_archive_path']
+            #local_backup_source_dir = entry['local_backup_source_dir']
+
+            if not args.skip_missing_source and not os.path.exists(local_archive_root_path):
+                raise RuntimeError(f"local archive root path '{local_archive_root_path}' doesn't exist")
+            elif args.skip_missing_source:
+                print(f"local archive root path is missing, skipping")
+                continue
+
+            # Set up Borg command with operation as the first argument
+            borg_env_var = f'export BORG_PASSPHRASE="{password}"'
+            borg_command = f'{borg_env_var}; cd {local_archive_root_path} && borg prune --rsh \'ssh -i ~/.ssh/id_ed25519\''
+
+            if args.dry_run is True:
+                print("dry run enabled")
+            dry_run_or_stats = '--dry-run' if args.dry_run is True else '--stats'
+            borg_command += f' --iec --progress {dry_run_or_stats} -v'
+            borg_command += f' --list --keep-last 1 {destination_host}{remote_archive_path}'
+
+            # Execute Borg command
+            print(f'[CMD] {";".join(borg_command.split(";")[1:])}')
+            retcode = subprocess.call(borg_command, shell=True)
+            if retcode != 0:
+                raise RuntimeError("exited with non-zero return code")
+    elif operation == 'compact_all':
+        for entry in passwords['archives']:
+            password = entry['password']
+            destination_host = get_destination_host(passwords['destination_hosts'], entry['destination_host_alias'])
+            local_archive_root_path = entry['local_archive_root_path']
+            remote_archive_path = entry['remote_archive_path']
+            #local_backup_source_dir = entry['local_backup_source_dir']
+
+            if not args.skip_missing_source and not os.path.exists(local_archive_root_path):
+                raise RuntimeError(f"local archive root path '{local_archive_root_path}' doesn't exist")
+            elif args.skip_missing_source:
+                print(f"local archive root path is missing, skipping")
+                continue
+
+            # Set up Borg command with operation as the first argument
+            borg_env_var = f'export BORG_PASSPHRASE="{password}"'
+            borg_command = f'{borg_env_var}; cd {local_archive_root_path} && borg compact --rsh \'ssh -i ~/.ssh/id_ed25519\''
+
+            borg_command += f' --iec --progress -v {destination_host}{remote_archive_path}'
+
+            # Execute Borg command
+            print(f'[CMD] {";".join(borg_command.split(";")[1:])}')
+            retcode = subprocess.call(borg_command, shell=True)
+            if retcode != 0:
+                raise RuntimeError("exited with non-zero return code")
+    else:
+        found = False
+        for entry in passwords['archives']:
+            if entry['archive_name'] == args.archive_name and entry['destination_host_alias'] == args.backup_destination:
+                found = True
+                password = entry['password']
+                destination_host = get_destination_host(passwords['destination_hosts'], args.backup_destination)
+                local_archive_root_path = entry['local_archive_root_path']
+                remote_archive_path = entry['remote_archive_path']
+                local_backup_source_dir = entry['local_backup_source_dir']
+
+                if not args.skip_missing_source and not os.path.exists(local_archive_root_path):
+                    raise RuntimeError(f"local archive root path '{local_archive_root_path}' doesn't exist")
+                elif args.skip_missing_source:
+                    print(f"local archive root path is missing, skipping")
+                    continue
+
+                # Set up Borg command with operation as the first argument
+                borg_env_var = f'export BORG_PASSPHRASE="{password}"'
+                borg_command = f'{borg_env_var}; cd {local_archive_root_path} && borg {operation} --rsh \'ssh -i ~/.ssh/id_ed25519\''
+
+                if operation == 'create':
+                    dry_run_or_stats = '--dry-run' if args.dry_run is True else '--stats'
+                    if args.ignore_inode:
+                        borg_command += ' --files-cache ctime,size'  # ignore inode (useful for external disks)
+                    if not args.skip_list:
+                        borg_command += ' --list'
+                    borg_command += f' --iec --progress {dry_run_or_stats} -v --compression none --checkpoint-interval=900'
+                    borg_command += f' {destination_host}{remote_archive_path}::' + '{utcnow}Z ' + f'{local_backup_source_dir}'
+                elif operation == 'list':
+                    borg_command += f' {destination_host}{remote_archive_path}'
+                elif operation == 'info':
+                    borg_command += f' --iec {destination_host}{remote_archive_path}'
+                elif operation == 'prune':
+                    borg_command += f' --list --keep-last 1 {destination_host}{remote_archive_path}'
+                elif operation == 'compact':
+                    borg_command += f' --iec --progress {destination_host}{remote_archive_path}'
+                # elif operation == 'delete':
+                else:
+                    print(f"Invalid operation '{operation}'")
+                    return
+
+                # Execute Borg command
+                print(f'[CMD] {";".join(borg_command.split(";")[1:])}')
+                os.system(borg_command)
+                break
+        if not found:
+            print(f"Couldn't find a pair of archive '{args.archive_name}' and destination '{args.backup_destination}'.")
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description="Sync backups using Borg via CLI")
+    parser.add_argument("operation", type=str, help="Borg operation (create, list, prune, info, compact, create_all, prune_all, compact_all)")
+    parser.add_argument("-a", "--archive-name", type=str, help="Name of the Borg archive", required=False)
+    parser.add_argument("-d", "--backup-destination", type=str, help="Destination for the backup", required=False)
+    parser.add_argument("--dry-run", type=bool, help="If specified, shows only what would happen", required=False)
+    parser.add_argument("--skip-missing-source", action=argparse.BooleanOptionalAction, help="Skips syncing a missing archive instead of crashing", required=False)
+    parser.add_argument("--ignore-inode", action=argparse.BooleanOptionalAction, help="Ignores inodes when creating a backup", required=False)
+    parser.add_argument("--skip-list", action=argparse.BooleanOptionalAction, help="Skips printing list of synced files", required=False)
+    arguments = parser.parse_args()
+    print(f'debug: {arguments.operation}')
+    if arguments.operation in ["create_all", "prune_all", "compact_all"]:
+        if arguments.archive_name is not None:
+            raise RuntimeError(f"Argument -a cannot be used with operation '{arguments.operation}'.")
+    else:
+        if arguments.archive_name is None or arguments.backup_destination is None:
+            raise RuntimeError(f"Arguments -a and -d must be used with operation '{arguments.operation}'.")
+    sync_backup(arguments)
+