+'Shared logic of orgbackup, orgrestore: ~/org in borg repos across servers.'
+from getpass import getpass
+from json import dumps as json_dumps, loads as json_loads
+from os import environ
+from pathlib import Path
+from stat import S_ISDIR, S_ISLNK
+from subprocess import CompletedProcess, run as subprocess_run
+from sys import exit as sys_exit, stderr
+from typing import Callable, Optional
+
+DIRNAME_ORG = 'org'
+PATH_HOME = Path.home()
+PATH_ORG = PATH_HOME / DIRNAME_ORG
+PATH_CONF = PATH_HOME / '.config' / 'orgsync'
+PATH_SERVERS = PATH_CONF / 'servers'
+PATH_KEYS = PATH_CONF / 'keys'
+PATH_STATE = PATH_HOME / '.local' / 'state' / 'orgsync' / 'sync.json'
+
+# keep in sync with scripts/setup_borg_server.sh
+BORG_ACCOUNT = 'borg'
+PATH_REMOTE_REPO = '/srv/borg/org'
+
+BORG_ENCRYPTION = 'keyfile'
+BORG_RCS_OK = (0, 1) # 1: warning, e.g. file changed while being read
+BORG_RSH = 'ssh -o StrictHostKeyChecking=accept-new'
+# where borg itself keeps local per-repo data (its defaults), to narrate that
+PATH_BORG_CACHE = PATH_HOME / '.cache' / 'borg'
+PATH_BORG_SECURITY = PATH_HOME / '.config' / 'borg' / 'security'
+
+# snapshot of PATH_ORG's tree, cheap to compute (no reading of file contents),
+# stored on each sync, to detect local changes since by comparing against a
+# fresh one; maps each path below PATH_ORG (relative to it) to:
+# - directories: [st_mode]
+# - symlinks: [st_mode, link target]
+# - anything else: [st_mode, st_size, st_mtime_ns]
+_Manifest = dict[str, list[int | str]]
+
+
+class SyncError(Exception):
+ 'Condition to abort on with a plain message rather than a traceback.'
+
+
+def run_against_syncerror(
+ f: Callable[[], None]
+ ) -> None:
+ 'Run f, turning a SyncError into a plain message and sys_exit.'
+ try:
+ f()
+ except SyncError as e:
+ sys_exit(f'error: {e}')
+
+
+def warn(
+ text: str
+ ) -> None:
+ 'Print text to stderr.'
+ print(text, file=stderr)
+
+
+class Server:
+ 'Server holding a borg repo of PATH_ORG.'
+
+ def __init__(
+ self,
+ name: str
+ ) -> None:
+ self.name = name
+ self.path_key = PATH_KEYS / name
+ self.archives: list[str] = []
+ self.reachable = False
+ self.repo_id = ''
+
+ def location(
+ self,
+ archive_name: Optional[str] = None
+ ) -> str:
+ 'Repo URL, with "::archive_name" appended if given.'
+ url = f'ssh://{BORG_ACCOUNT}@{self.name}{PATH_REMOTE_REPO}'
+ return url if archive_name is None else f'{url}::{archive_name}'
+
+ def borg(
+ self,
+ *args,
+ check: bool = True,
+ **kwargs
+ ) -> CompletedProcess:
+ 'Run borg with args against own keyfile, fail on rc not BORG_RCS_OK.'
+ result = subprocess_run(
+ ('borg', *args),
+ env=environ | {'BORG_KEY_FILE': str(self.path_key),
+ 'BORG_RSH': BORG_RSH},
+ check=False,
+ **kwargs)
+ if check and result.returncode not in BORG_RCS_OK:
+ raise SyncError(f'{self.name}: borg {args[0]} failed')
+ return result
+
+ def probe(
+ self
+ ) -> None:
+ 'Fill .archives, set .reachable – or warn why the server is skipped.'
+ if not self.path_key.exists():
+ warn(f'{self.name}: skipped, no keyfile at {self.path_key}')
+ return
+ result = self.borg('list', '--json', self.location(),
+ check=False, capture_output=True, text=True)
+ if result.returncode not in BORG_RCS_OK:
+ warn(f'{self.name}: skipped, borg list failed:\n'
+ + result.stderr.strip())
+ return
+ listing = json_loads(result.stdout)
+ self.archives = sorted(archive['name']
+ for archive in listing['archives'])
+ self.repo_id = listing['repository']['id']
+ self.reachable = True
+ print(f'{self.name}: borg keeps its records of this repo at'
+ f' {self.path_borg_security}')
+
+ @property
+ def path_borg_cache(
+ self
+ ) -> Path:
+ "Borg's local cache for repo, as written by borg create."
+ return PATH_BORG_CACHE / self.repo_id
+
+ @property
+ def path_borg_security(
+ self
+ ) -> Path:
+ "Borg's local records of repo, as updated on any access."
+ return PATH_BORG_SECURITY / self.repo_id
+
+ @property
+ def newest(
+ self
+ ) -> Optional[str]:
+ 'Name of newest archive, if any.'
+ return self.archives[-1] if self.archives else None
+
+ def init(
+ self
+ ) -> None:
+ 'Create repo on server and its keyfile locally, refuse overwriting.'
+ if self.path_key.exists():
+ raise SyncError(f'{self.path_key} exists, refusing to overwrite')
+ PATH_KEYS.mkdir(mode=0o700, parents=True, exist_ok=True)
+ self.borg('init', '--encryption', BORG_ENCRYPTION, self.location())
+
+
+def ensure_passphrase(
+ confirm: bool = False
+ ) -> None:
+ 'Unless borg has a passphrase source already, ask for one for all repos.'
+ if 'BORG_PASSPHRASE' in environ or 'BORG_PASSCOMMAND' in environ:
+ return
+ passphrase = getpass('borg passphrase: ')
+ if confirm and getpass('borg passphrase, again: ') != passphrase:
+ raise SyncError('passphrases differ')
+ environ['BORG_PASSPHRASE'] = passphrase
+
+
+def load_servers(
+ ) -> list[Server]:
+ 'Read PATH_SERVERS: one server per line; blank, "#" lines skipped.'
+ if not PATH_SERVERS.exists():
+ raise SyncError(f'no server list at {PATH_SERVERS}')
+ lines = [line.strip() for line
+ in PATH_SERVERS.read_text(encoding='utf8').splitlines()]
+ return [Server(line) for line in lines
+ if line and not line.startswith('#')]
+
+
+def probe_servers(
+ ) -> list[Server]:
+ 'Load and probe servers, return reachable ones, fail if none.'
+ servers = load_servers()
+ for server in servers:
+ server.probe()
+ reachable = [server for server in servers if server.reachable]
+ if not reachable:
+ raise SyncError('no server reachable')
+ return reachable
+
+
+def newest_archive(
+ servers: list[Server]
+ ) -> Optional[str]:
+ 'Name of newest archive across servers, if any.'
+ return max((server.newest for server in servers if server.newest),
+ default=None)
+
+
+def current_manifest() -> _Manifest:
+ 'Snapshot PATH_ORG entries: mode, plus size and mtime or link target.'
+ manifest: _Manifest = {}
+ if not PATH_ORG.exists():
+ return manifest
+ for dirpath, dirnames, filenames in PATH_ORG.walk():
+ for path in (dirpath / name for name in dirnames + filenames):
+ stat = path.lstat()
+ key = str(path.relative_to(PATH_ORG))
+ if S_ISLNK(stat.st_mode):
+ manifest[key] = [stat.st_mode, str(path.readlink())]
+ elif S_ISDIR(stat.st_mode):
+ manifest[key] = [stat.st_mode]
+ else:
+ manifest[key] = [stat.st_mode, stat.st_size, stat.st_mtime_ns]
+ return manifest
+
+
+def load_state() -> tuple[Optional[str], Optional[_Manifest]]:
+ 'Last synced archive name and PATH_ORG manifest, if ever synced.'
+ if not PATH_STATE.exists():
+ return None, None
+ state = json_loads(PATH_STATE.read_text(encoding='utf8'))
+ return state['archive_name'], state['manifest']
+
+
+def save_state(
+ archive_name: str,
+ manifest: _Manifest
+ ) -> None:
+ 'Store archive name and PATH_ORG manifest as last synced state.'
+ print(f'saving sync state to {PATH_STATE} …')
+ PATH_STATE.parent.mkdir(parents=True, exist_ok=True)
+ PATH_STATE.write_text(
+ json_dumps({'archive_name': archive_name, 'manifest': manifest}),
+ encoding='utf8')
projects / confplom / commitdiff