#!/usr/bin/env python3 """ Program: Rod's Incremental Backup System (ribs.py) Author: Rod Wright Date: 03/10/2025 """ import subprocess import sys import os import shutil import glob from datetime import datetime, timedelta import textwrap import argparse import configparser from pathlib import Path # App version APP_VERSION="1.3.2" # File paths SYS_EN_CONF_DIR="/etc/ribs/conf-enabled/" SYS_AV_CONF_DIR="/etc/ribs/conf-available/" USER_EN_CONF_DIR=str(Path.home())+"/.ribs/conf-enabled/" USER_AV_CONF_DIR=str(Path.home())+"/.ribs/conf-available/" # Determine if I have superuser privileges if os.geteuid()==0: superuser=True else: superuser=False # ****** BEGIN function definitions ****** def main(): """ main function """ print("") # Parse arguments and take required actions parser=argparse.ArgumentParser(description='Rod\'s Incremental Backup System, version '+APP_VERSION) group=parser.add_mutually_exclusive_group() parser.add_argument( '-s', '--simulate', help='simulate operations to be performed', action="store_true" ) group.add_argument( '-l', '--list-configs', help='list available and enabled config files', action="store_true" ) group.add_argument( '-e', '--enable', metavar='CONF_FILE', help='enable a config file' ) group.add_argument('-d', '--disable', metavar='CONF_FILE', help='disable a config file' ) group.add_argument( "conf_file", help='Run RIBS using the specified config file. \ If none specified, run all enabled config files, looking in \'~/.ribs/conf-enabled\' \ for standard users, or in \'/etc/ribs/conf-enabled\' for superusers.', nargs='?' ) args=parser.parse_args() if args.simulate: simulate=True else: simulate=False if args.enable: if args.enable!=None: enable_config(args.enable,simulate) elif args.disable: if args.disable!=None: disable_config(args.disable,simulate) elif args.list_configs: if args.list_configs!=None: list_configs() else: run_backup(args.conf_file,simulate) def run_rsync(source, destination, options=None, excludes=None): """ Runs rsync with the given source, destination, options, and excludes. Args: source (str): The source path. destination (str): The destination path. options (list, optional): A list of rsync options. Defaults to None. excludes (list, optional): A list of patterns to exclude. Defaults to None. Returns: int: The return code of the rsync command. """ command = ["rsync"] if options: command.extend(options) if excludes: for pattern in excludes: command.extend(['--exclude='+pattern]) command.extend([source, destination]) #print("complete rsync command is:",command) process = subprocess.run(command, capture_output=True, text=True) if '-v' in options or '--verbose' in options: print(process.stdout) if process.returncode != 0: print(f"Error running rsync: {process.stderr}") return process.returncode def mount_offline(mountpoint,mountparams,mode): """ Handle mounting and unmounting of offline storage. mountpoint is backup_root. mountparams are the mount parameters from config file. mode is ro, rw, or u. """ pass def list_configs(): """ Show a list of config files. """ if superuser: ena_conf_dir=SYS_EN_CONF_DIR avail_conf_dir=SYS_AV_CONF_DIR enabled_configs=find_configs('system','enabled') available_configs=find_configs('system','available') else: ena_conf_dir=USER_EN_CONF_DIR avail_conf_dir=USER_AV_CONF_DIR enabled_configs=find_configs('user','enabled') available_configs=find_configs('user','available') print("Available config files:\n") if len(available_configs)==0: infostring="No available config files found. Refer to the sample config file "+SYS_AV_CONF_DIR+"ribs.conf.sample to create one. Note that config files must have the .conf extension to be recognized." infostringlist=textwrap.wrap(infostring,width=80,break_on_hyphens=False) for line in infostringlist: print(line) else: for config in available_configs: print(Path(config).name) print("\n") print("Enabled config files:\n") if len(enabled_configs)==0: infostring="No enabled config files found. Refer to the list of available config files above and enable one using ribs -e. Type ribs -h for help." infostringlist=textwrap.wrap(infostring,width=80,break_on_hyphens=False) for line in infostringlist: print(line) else: for config in enabled_configs: print(Path(config).name) def enable_config(configfile,simulate=True): """ Enable the specified config file """ if simulate: print("All actions will be simulated") if superuser: conftype='system' availconfdir=SYS_AV_CONF_DIR enaconfdir=SYS_EN_CONF_DIR else: conftype='user' availconfdir=USER_AV_CONF_DIR enaconfdir=USER_EN_CONF_DIR # Look for configs to enable configs=find_configs(conftype,'available') if len(configs)>0: for config in configs: if Path(config).name == configfile: link=Path(enaconfdir+configfile) target=Path(availconfdir+configfile) if simulate: print("Simulating enabling",Path(configfile).name) else: print("Enabling the user config:",configfile) link.symlink_to(target) def disable_config(configfile,simulate=True): """ Disable the specified config file """ if simulate: print("All actions will be simulated") if superuser: conftype='system' enaconfdir=SYS_EN_CONF_DIR else: conftype='user' enaconfdir=USER_EN_CONF_DIR # Look for configs to disable configs=find_configs(conftype,'enabled') if len(configs)>0: for config in configs: if Path(config).name == configfile: link=Path(enaconfdir+configfile) if simulate: print("Simulating disabling",Path(configfile).name) else: print("Disabling the user config:",configfile) link.unlink(link) def find_configs(conftype,confstate): """ Find and produce a list of all enabled config files """ if confstate=="enabled": if conftype=="user": dirpath=USER_EN_CONF_DIR elif conftype=="system": dirpath=SYS_EN_CONF_DIR elif confstate=="available": if conftype=="user": dirpath=USER_AV_CONF_DIR elif conftype=="system": dirpath=SYS_AV_CONF_DIR pattern="*.conf" configs = [] matching_files=glob.glob(os.path.join(dirpath,pattern)) if matching_files: for filepath in matching_files: configs.append(filepath) return(configs) def parse_config_file(conf_file): """ Read the specified config file and return a dictionary of the parameters """ config_errors=[] config=configparser.ConfigParser(allow_no_value=True) config.read(conf_file) if config.has_section('General'): # Pull the general options if config.has_option('General','source_dir'): source_dir=config['General']['source_dir'] else: config_errors.append('source_dir option missing from config file') if config.has_option('General','backup_root'): backup_root=config['General']['backup_root'] else: config_errors.append('backup_root option missing from config file') if config.has_option('General','backup_subdir'): backup_subdir=config['General']['backup_subdir'] else: backup_subdir=None if config.has_option('General','exclude_list'): exclude_list=config['General']['exclude_list'].split(',') exclude_list=[pattern.strip() for pattern in exclude_list] else: exclude_list=None if config.has_option('General','day_limit'): day_limit=config['General']['day_limit'] else: config_errors.append('day_limit option missing from config file') if config.has_option('General','storage_type'): storage_type=config['General']['storage_type'] else: config_errors.append('storage_type option missing from config file') else: config_errors.append('General section missing from config file') if len(config_errors)>0: print("The following configuration file errors were found") for err_msg in config_errors: print(err_msg) print("These are fatal errors and ribs cannot continue.") print("Please check the config file:",conf_file) sys.exit(1) config_errors=[] if storage_type=="local": if config.has_section('local'): # Reserved for future use pass else: pass elif storage_type=="offline": if config.has_section('offline'): if config.has_option('offline','mount_parameters'): mount_parameters=config['offline']['mount_parameters'] else: config_errors.append('mount_parameters option missing from config file') if config.has_option('offline','remount_ro'): remount_ro=config['offline']['remount_ro'] else: remount_ro=False else: config_errors.append('offline section missing from config file') elif storage_type=="remote": if config.has_section('remote'): if config.has_option('remote','remote_user_host'): remote_user_host=config['remote']['remote_user_host'] else: config_errors.append('remote_user_host option missing from config file') else: config_errors.append('remote section missing from config file') else: config_errors.append('invalid storage_type. Must be local, offline, or remote.') if len(config_errors)>0: print("The following configuration file errors were found") for err_msg in config_errors: print(err_msg) print("These are fatal errors and ribs cannot continue.") print("Please check the config file:",conf_file) sys.exit(1) # QC option formatting config_errors=[] if not source_dir.startswith('/'): config_errors.append('source_dir must be an absolute path, but doesn\'t start with a \'/\'') if not backup_root.startswith('/'): config_errors.append('backup_root must be an absolute path, but doesn\'t start with a \'/\'') if day_limit=='0': config_errors.append('day_limit cannot be 0') if storage_type=='remote' and not "@" in remote_user_host: config_errors.append('remote_user_host must be in the format\'user@host\'') if len(config_errors)>0: print("The following configuration file errors were found") for err_msg in config_errors: print(err_msg) print("These are fatal errors and ribs cannot continue.") print("Please check the config file:",conf_file) sys.exit(1) # Clean up variables if backup_root.endswith('/'): backup_root=backup_root.rstrip('/') if backup_subdir: if backup_subdir.startswith('/'): backup_subdir=backup_subdir.lstrip('/') if backup_subdir.endswith('/'): backup_subdir=backup_subdir.rstrip('/') # Return a parameter:option dictionary if storage_type=="local": return { 'storage_type':storage_type, 'source_dir':source_dir, 'backup_root':backup_root, 'backup_subdir':backup_subdir, 'exclude_list':exclude_list, 'day_limit':day_limit } elif storage_type=="offline": return { 'storage_type':storage_type, 'source_dir':source_dir, 'backup_root':backup_root, 'backup_subdir':backup_subdir, 'exclude_list':exclude_list, 'day_limit':day_limit, 'mount_parameters':mount_parameters, 'remount_ro':remount_ro } elif storage_type=="remote": return { 'storage_type':storage_type, 'source_dir':source_dir, 'backup_root':backup_root, 'backup_subdir':backup_subdir, 'exclude_list':exclude_list, 'day_limit':day_limit, 'remote_user_host':remote_user_host } else: # Not sure how we got here, but if we did, something's wrong return False def run_backup(configfile=None,simulate=True): """ Run backup using the specified configfile, or all enabled config files if none specified. """ if simulate: print("All actions will be simulated") if configfile==None: # Find and run all enabled configs print("Finding enabled config files...") userconfigs=find_configs("user","enabled") systemconfigs=find_configs("system","enabled") if len(userconfigs)>0: configs=userconfigs print("Found user config files:") for config in configs: print(config) elif len(systemconfigs)>0: print("No user config files found.") configs=systemconfigs print("Found system config files:") for config in configs: print(config) else: print("No enabled config files found") else: # Run only the specified config configs=[configfile] for config in configs: print("Running backup from the config file:",config) # Parse config file and create variables conf_parameters=parse_config_file(config) source_dir=Path(conf_parameters.get('source_dir')) backup_root=Path(conf_parameters.get('backup_root')) backup_subdir=conf_parameters.get('backup_subdir') if backup_subdir: backup_subdir=Path(conf_parameters.get('backup_subdir')) exclude_list=conf_parameters.get('exclude_list') day_limit=conf_parameters.get('day_limit') storage_type=conf_parameters.get('storage_type') mount_parameters=conf_parameters.get('mount_parameters') remount_ro=conf_parameters.get('remount_ro') remote_user_host=conf_parameters.get('remote_user_host') # Generate exclude list if exclude_list != None: for exclude_pattern in exclude_list: print("excluding",exclude_pattern) # If storage_type is offline, mount offline storage read-write if storage_type=='offline': try: if not backup_root.is_dir(): raise Exception("Error: The mount point "+backup_root+" does not exist.") except Exception as errmsg: print("") print(errmsg) print("This is a fatal error and ribs cannot continue.") print("Please check the config file:",config,", or create the mountpoint.") sys.exit(1) else: mount_offline(backup_root,mount_parameters,'rw') # Make sure backup_root exists, is writable, and supports hard links if storage_type=='local' or storage_type=='offline': testdir=backup_root.joinpath(Path('testdir')) testfile=testdir.joinpath(Path('testfile')) testlink=testdir.joinpath(Path('testlink')) # Test for existance if storage_type=='local': print("Testing for existance of",backup_root,"... ",end="") try: if not backup_root.is_dir(): raise Exception(backup_root+" is not a directory.") except Exception as errmsg: print("FAILED") print("") print(errmsg) print("") print("This is a fatal error and ribs cannot continue.") print("Please check the config file:",config) sys.exit(1) else: print("OK") # Test for writability if simulate: print("Skipping all write testing since we're only simulating.") else: print("Testing for writability of",backup_root,"... ",end="") try: testdir.mkdir(exist_ok=True) testfile.touch(exist_ok=True) testfile.unlink(missing_ok=True) shutil.rmtree(testdir) except Exception as errmsg: print("FAILED") print("") print(errmsg) print("") print("This is a fatal error and ribs cannot continue.") print("Please check the config file:",config) if storage_type=='offline': mount_offline(backup_root,mountparams,'u') sys.exit(1) else: print("OK") # Test for hard link support print("Testing for hard link support on",backup_root,"... ",end="") try: testdir.mkdir(exist_ok=True) testfile.touch(exist_ok=True) testlink.hardlink_to(testfile) testlink.unlink(missing_ok=True) testfile.unlink(missing_ok=True) shutil.rmtree(testdir) except Exception as errmsg: print("FAILED") print("") print(errmsg) print("") print("This is a fatal error and ribs cannot continue.") print("Please check the config file:",config) if storage_type=='offline': mount_offline(backup_root,mountparams,'u') sys.exit(1) else: print("OK") # Determine today's date and generate current datetime directory name datetimedir_format='%Y-%m-%d_%H:%M:%S' now=datetime.now() current_dirname=now.strftime(datetimedir_format) this_dirname=Path(current_dirname) # Find previous backup's directory, if present, to use as link destination dirname_pattern='????-??-??*' prev_backups = [] if storage_type=='remote': # get remote listing pass else: # get local or offline listing if backup_subdir: print("The backup subdirectory is",backup_subdir) backups_location=backup_root.joinpath(backup_subdir) else: backups_location=backup_root matches=backups_location.glob(dirname_pattern) #matching_dirs=glob.glob(os.path.join(backup_root+backup_subdir+"/",dirname_pattern)) matching_dirs=[match for match in matches if match.is_dir()] if matching_dirs: for dirpath in matching_dirs: prev_backups.append(dirpath) if len(prev_backups)==0: print("No previous backups were found") has_previous=False else: has_previous=True prev_backups.sort() last_dirname=prev_backups[-1] last_backup=Path(last_dirname) print("The last backup was in:",last_backup) this_backup=backups_location.joinpath(this_dirname) print("This backup will be in:",this_backup) # Set rsync target and link destination directories if storage_type=='remote': target_dir=remote_user_host+":"+this_backup link_dir=last_backup+'/' else: target_dir=this_backup if has_previous: link_dir=last_backup else: link_dir=None # Generate rsync options rsync_options=[] #rsync_options.append('--verbose') rsync_options.append('--secluded-args') rsync_options.append('--archive') rsync_options.append('--human-readable') rsync_options.append('--delete') if has_previous: rsync_options.append('--link-dest='+str(link_dir)) if simulate: rsync_options.append('-n') # Create backup_subdir if necessary if simulate: print("Unable to create backup subdirectory since we're just simulating.") else: if backup_subdir: backups_location.mkdir(exist_ok=True) # Run the rsync command try: print("Running rsync... ",end="") run_rsync(str(source_dir), str(this_backup)+'/', options=rsync_options, excludes=exclude_list) except Exception as errmsg: print("FAILED") print("") print(errmsg) else: print("OK") # Delete backup directories older than day_limit try: if simulate: print("Simulating ",end="") print("Deleting backups that are more than",day_limit,"days old.") if has_previous: would_delete=False for old_backup in prev_backups: backup_date=datetime.strptime(os.path.basename(old_backup),datetimedir_format) if now-backup_date>timedelta(days=int(day_limit)): would_delete=True if simulate: print("Simulating deleting",os.path.basename(old_backup)) else: print("Deleting",os.path.basename(old_backup)) shutil.rmtree(old_backup) if not would_delete: print("No backups are more than",day_limit,"days old, so not deleting.") else: print("There are no previous backups to delete") except Exception as errmsg: print("") print(errmsg) # Synchronize disks # If STORAGE_TYPE is OFFLINE, update the capacity file # If STORAGE_TYPE is OFFLINE, unmount, and remount read-only if required # ****** END function definitions ****** # Call main function main()