--- /dev/null Thu Jan 01 00:00:00 1970 +0000
+++ b/borgend/config.py Sun Jan 28 11:54:46 2018 +0000
@@ -0,0 +1,174 @@
+#!/usr/local/bin/python3
+#
+# Borgend configuration loader
+#
+
+import yaml
+import io
+import os
+import string
+import logging
+import platform
+from functools import reduce
+
+from . import loggers
+from . import branding
+from . import locations
+
+logger=loggers.get(__name__)
+
+#
+# Defaults
+#
+
+defaults={
+ # borg
+ # Default: backup every 6 hours (21600 seconds)
+ 'backup_interval': 21600,
+ # Default: retry every 15 minutes if unable to connect / unfinished backup
+ 'retry_interval': 900,
+ # Extract passphrases at startup or on demand?
+ 'extract_passphrases_at_startup': True,
+ # Do not insert a quit menu entry (useful for installing on computers of
+ # inexperienced users)
+ 'no_quit_menu_entry': False,
+ # Borg settings
+ 'borg': {
+ 'executable': 'borg',
+ 'common_parameters': [],
+ 'create_parameters': [],
+ 'prune_parameters': [],
+ }
+}
+
+#
+# Type checking etc.
+#
+
+def error(x):
+ raise AssertionError(x)
+
+def check_field(cfg, field, descr, loc, default, check):
+ if field in cfg:
+ tmp=cfg[field]
+ if not check(tmp):
+ error("%s is of invalid type for %s" % (field, loc))
+ return tmp
+ else:
+ if default is not None:
+ return default
+ else:
+ error("%s is not configured for %s" % (field, loc))
+
+def check_bool(cfg, field, descr, loc, default=None):
+ return check_field(cfg, field, descr, loc, default,
+ lambda x: isinstance(x, bool))
+
+def check_string(cfg, field, descr, loc, default=None):
+ return check_field(cfg, field, descr, loc, default,
+ lambda x: isinstance(x, str))
+
+def check_dict(cfg, field, descr, loc, default=None):
+ return check_field(cfg, field, descr, loc, default,
+ lambda x: isinstance(x, dict))
+
+def check_list(cfg, field, descr, loc, default=None):
+ return check_field(cfg, field, descr, loc, default,
+ lambda x: isinstance(x, list))
+
+def is_list_of(x, chk):
+ if x is None:
+ return True
+ elif isinstance(x, list):
+ return reduce(lambda y, z: y and chk(z), x, True)
+ else:
+ return False
+
+def check_list_of_dicts(cfg, field, descr, loc, default=None):
+ return check_field(cfg, field, descr, loc, default,
+ lambda x: is_list_of(x, lambda z: isinstance(z, dict)))
+
+def check_list_of_strings(cfg, field, descr, loc, default=None):
+ return check_field(cfg, field, descr, loc, default,
+ lambda x: is_list_of(x, lambda z: isinstance(z, str)))
+
+def check_nonempty_list_of_strings(cfg, field, descr, loc):
+ return check_list_of_strings(cfg, field, descr, loc) and cfg[field]
+
+
+def check_nonneg_int(cfg, field, descr, loc, default=None):
+ return check_field(cfg, field, descr, loc, default,
+ lambda x: isinstance(x, int) and x>=0)
+
+
+#
+# Borg command line parameter configuration helper routines and classes
+#
+
+class BorgParameters:
+ def __init__(self, common, create, prune):
+ self.common=common or []
+ self.create=create or []
+ self.prune=prune or []
+
+ def from_config(cfg, loc):
+ common=check_list_of_dicts(cfg, 'common_parameters',
+ 'Borg parameters', loc, default=[])
+
+ create=check_list_of_dicts(cfg, 'create_parameters',
+ 'Create parameters', loc, default=[])
+
+ prune=check_list_of_dicts(cfg, 'prune_parameters',
+ 'Prune parameters', loc, default=[])
+
+ return BorgParameters(common, create, prune)
+
+ def __add__(self, other):
+ common=self.common+other.common
+ create=self.create+other.create
+ prune=self.prune+other.prune
+ return BorgParameters(common, create, prune)
+
+#
+# Load config on module load
+#
+
+def expand_env(cfg, env):
+ if isinstance(cfg, dict):
+ out={key: expand_env(val, env) for key, val in cfg.items()}
+ elif isinstance(cfg, list):
+ out=[expand_env(val, env) for val in cfg]
+ elif isinstance(cfg, str):
+ out=string.Template(cfg).substitute(os.environ)
+ else:
+ out=cfg
+
+ return out
+
+if not (os.path.exists(locations.cfgfile) and os.path.isfile(locations.cfgfile)):
+ raise SystemExit("Configuration file required: %s" % locations.cfgfile)
+
+logger.info("Reading configuration %s missing" % locations.cfgfile)
+
+with io.open(locations.cfgfile, 'r') as file:
+ settings=expand_env(yaml.load(file), os.environ);
+
+#
+# Verify basic settings
+#
+
+def check_and_set(cfg, field, loc, defa, fn):
+ cfg[field]=fn(cfg, field, field, loc, defa[field])
+
+check_and_set(settings, 'backup_interval', 'top-level', defaults, check_nonneg_int)
+check_and_set(settings, 'retry_interval', 'top-level', defaults, check_nonneg_int)
+check_and_set(settings, 'extract_passphrases_at_startup', 'top-level', defaults, check_nonneg_int)
+check_and_set(settings, 'no_quit_menu_entry', 'top-level', defaults, check_bool)
+check_and_set(settings, 'borg', 'top-level', defaults, check_dict)
+# Check parameters within 'borg'
+if True:
+ check_and_set(settings['borg'], 'executable', 'borg',
+ defaults['borg'], check_string)
+
+ borg_parameters=BorgParameters.from_config(settings['borg'], "top-level")
+