vidhar: borg copy

4 files changed, 347 insertions, 3 deletions

diff --git a/hosts/vidhar/borg/append.borgbase b/hosts/vidhar/borg/append.borgbase
new file mode 100644
index 00000000..d7091871
--- /dev/null
+++ b/hosts/vidhar/borg/append.borgbase
@@ -0,0 +1,26 @@
+{
+        "data": "ENC[AES256_GCM,data:+TKVqaaf8MYqzLgJ4DFYg6CaGmrSkNlGZiK3PNLusVwZhxAk19Xd/ouFEpI4+Gsecah9kyDZgMRM7x4bXz6RhhR9cEyPbm9lIYQz6OGoqW/xMIui1+DwCIfrtByXZ1SQKLz4052+fUwQvGgdG2/i9twF1SPUUA0NxNKyqjVfZpAVia8aJkl7Zu74UzjUPAqaORYAoZcZRVcUDH1cUQpDMzUza3SsDv9/FbnG/ZIify8sKVXaNsNjmuKQ1WEshcSLNcQQzvQIM0jCwpN8KE9C4mj0ANf3JGehoP39ticvzZl2iOCfBwNuL4feWUepBXuRxtn4nbv8OxmQkdJrqR4VwDzgbP3XzpI5zraLIMH093BIQSLbuNXPOHVDh4QxpWrj5+IZYq9ZjUHw3zCg66AGFZRMKPs11N9RVnxp/LpZRMRSYNhzh12ik7YVqyGt32MkDXrSdGFw96ae1PsTSe5V1rY1yevnznwY6OtXj20cneT/CMGFyj9DkcZ6ClT5KDYP1xPwR2D1A56cBDgvPeNPfhPWhRHoYLU90eHkF16N9zLrSBBhKtoNUUbB9K46hiSWaTwdQVWNR2ZLafR3,iv:6eDgcQuPi9Z5tVbr1BEVzCRu5IZC6hYfZtXfuwIKaZQ=,tag:wdYKzdGkaKOv3jHd6TzA9w==,type:str]",
+        "sops": {
+                "kms": null,
+                "gcp_kms": null,
+                "azure_kv": null,
+                "hc_vault": null,
+                "age": null,
+                "lastmodified": "2022-02-17T10:32:27Z",
+                "mac": "ENC[AES256_GCM,data:7i7zPalH3J/VtNGzHrECMHt02WTGOuT5KL+HQT6b5zLCpYyTTBit/HM4xW4hONxnEaEgPpkUr0cTNvsGrrKE0zyj784xLDASvaBcTinkmNvkBEVsB9ACPG3G5YmexvYC9Z3tVFjfLPdQeXgy5nAm6Q8TMANgp+xIQTnc0IGVjS8=,iv:u3UL8XrR6UKEDOWgMa4nzEYl8vmguZmcANmby2kgxWM=,tag:2k7+DKVTtmI7ldk/ktXzbQ==,type:str]",
+                "pgp": [
+                        {
+                                "created_at": "2022-02-17T10:32:26Z",
+                                "enc": "-----BEGIN PGP MESSAGE-----\n\nhF4DbYDvGI0HDr0SAQdArVlDVy8NhAS3QNPoul2iE05B1Su1r8fnmFm4k4+ORX4w\nddhRrT+TAkbPM0Zl1nDyazqJKWlq2DXZ8DZ6qEFAS0bYN0x/QiazIqH4NfWWwa0N\n0l4BunqQtbQNkv0qjqBmkhDnlVNainmEPv4ChXrJc1z6HXYdvv4CkfzwdvEfr5yO\ndpHERmg5O5mX29BnM6LHk6CdFtJS1jN3qZy3oa0KEECoZm8Ak2hlBw/PeEGk3eBM\n=SHc3\n-----END PGP MESSAGE-----\n",
+                                "fp": "A1C7C95E6CAF0A965CB47277BCF50A89C1B1F362"
+                        },
+                        {
+                                "created_at": "2022-02-17T10:32:26Z",
+                                "enc": "-----BEGIN PGP MESSAGE-----\n\nhF4DXxoViZlp6dISAQdApIVC5MIiWio4yRZ0hIVWntk2Xk7LKHPwJP2pw73SgREw\nCJZDCT91QEMPDwobh8eq4R8vook9fUJ0t+M5TUN5IwjSwSdmwiQ+mpArmyStIi4W\n0l4BiEt5Giar+H8V62bBiN38EbqHF4jN7jfjt6QrU4Nr/kP7DokI26TvXPqOtOOh\nTGOhiTvroHu4zRGMZKqq/IDE9FLA1SXRU3rXmHtVjz2U2Zmnj0Hj9iR+bZmy+TuY\n=ugF2\n-----END PGP MESSAGE-----\n",
+                                "fp": "30D3453B8CD02FE2A3E7C78C0FB536FB87AE8F51"
+                        }
+                ],
+                "unencrypted_suffix": "_unencrypted",
+                "version": "3.7.1"
+        }
+}
\ No newline at end of file
diff --git a/hosts/vidhar/borg/copy.py b/hosts/vidhar/borg/copy.py
new file mode 100755
index 00000000..b99e301a
--- /dev/null
+++ b/hosts/vidhar/borg/copy.py
@@ -0,0 +1,216 @@
+#!@python@/bin/python
+
+import json
+import os
+import subprocess
+import re
+import sys
+from sys import stderr
+from humanize import naturalsize
+
+from tempfile import TemporaryDirectory
+
+from datetime import (datetime, timedelta)
+from dateutil.tz import (tzlocal, tzutc)
+import dateutil.parser
+import argparse
+
+from tqdm import tqdm
+
+from xdg import xdg_runtime_dir
+import pathlib
+
+import unshare
+from time import sleep
+
+from halo import Halo
+
+
+parser = argparse.ArgumentParser()
+parser.add_argument('source', metavar='REPO_OR_ARCHIVE')
+parser.add_argument('target', metavar='REPO_OR_ARCHIVE')
+args = parser.parse_args()
+
+def read_repo(path):
+    with Halo(text=f'Listing {path}', spinner='arc') as sp:
+        res = None
+        with subprocess.Popen(['borg', 'list', '--info', '--lock-wait', '120', '--json', path], stdout=subprocess.PIPE) as proc:
+            res = json.load(proc.stdout)['archives']
+        sp.succeed(f'{len(res)} archives in {path}')
+    return res
+
+class ToSync:
+    def __iter__(self):
+        return self
+
+    def __next__(self):
+        while True:
+            try:
+                src = read_repo(args.source)
+                dst = read_repo(args.target)
+                for entry in src:
+                    if entry['name'] not in {dst_entry['name'] for dst_entry in dst} and not entry['name'].endswith('.checkpoint'):
+                        return entry
+                raise StopIteration
+            except (subprocess.CalledProcessError, json.decoder.JSONDecodeError) as err:
+                print(err, file=stderr)
+                continue
+
+def copy_archive(src_repo_path, dst_repo_path, entry):
+    cache_suffix = None
+    with Halo(text=f'Determine archive parameters', spinner='arc') as sp:
+        match = re.compile('^(.*)-[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}(\.(checkpoint|recreate)(\.[0-9]+)?)?').fullmatch(entry['name'])
+        if match:
+            repo_id = None
+            with subprocess.Popen(['borg', 'info', '--info', '--lock-wait', '120', '--json', src_repo_path], stdout=subprocess.PIPE) as proc:
+                repo_id = json.load(proc.stdout)['repository']['id']
+            if repo_id:
+                cache_suffix = f'{repo_id}_{match.group(1)}'
+        sp.succeed(f'Will process {entry["name"]} ({dateutil.parser.isoparse(entry["start"])}, cache_suffix={cache_suffix})')
+    with TemporaryDirectory(prefix=f'borg-mount_{entry["name"]}_') as tmpdir:
+        child = os.fork()
+        if child == 0:
+            # print('unshare/chroot', file=stderr)
+            unshare.unshare(unshare.CLONE_NEWNS)
+            subprocess.run(['mount', '--make-rprivate', '/'], check=True)
+            chroot = pathlib.Path(tmpdir) / 'chroot'
+            upper = pathlib.Path(tmpdir) / 'upper'
+            work = pathlib.Path(tmpdir) / 'work'
+            for path in [chroot,upper,work]:
+                path.mkdir()
+            subprocess.run(['mount', '-t', 'overlay', 'overlay', '-o', f'lowerdir=/,upperdir={upper},workdir={work}', chroot], check=True)
+            bindMounts = ['nix', 'run', 'proc', 'dev', 'sys', pathlib.Path(os.path.expanduser('~')).relative_to('/')]
+            if not ":" in src_repo_path:
+                bindMounts.append(pathlib.Path(src_repo_path).relative_to('/'))
+            if 'SSH_AUTH_SOCK' in os.environ:
+                bindMounts.append(pathlib.Path(os.environ['SSH_AUTH_SOCK']).parent.relative_to('/'))
+            for bindMount in bindMounts:
+                (chroot / bindMount).mkdir(parents=True,exist_ok=True)
+                subprocess.run(['mount', '--bind', pathlib.Path('/') / bindMount, chroot / bindMount], check=True)
+            os.chroot(chroot)
+            os.chdir('/')
+            dir = pathlib.Path('/borg')
+            dir.mkdir(parents=True,exist_ok=True)
+            with Halo(text=f'Determine size', spinner='arc') as sp:
+                total_size = None
+                total_files = None
+                with subprocess.Popen(['borg', 'info', '--info', '--json', '--lock-wait', '120', f'{src_repo_path}::{entry["name"]}'], stdout=subprocess.PIPE, text=True) as proc:
+                    stats = json.load(proc.stdout)['archives'][0]['stats']
+                    total_size = stats['original_size']
+                    total_files = stats['nfiles']
+                sp.succeed(f'{total_files} files, {naturalsize(total_size, binary=True)}')
+            # print(f'Mounting to {dir}', file=stderr)
+            with subprocess.Popen(['borg', 'mount', '--foreground', '--progress', '--lock-wait', '120', f'{src_repo_path}::{entry["name"]}', dir]) as mount_proc:
+                with Halo(text='Waiting for mount', spinner='arc') as sp:
+                    wait_start = datetime.now()
+                    while True:
+                        ret = subprocess.run(['mountpoint', '-q', dir])
+                        if ret.returncode == 0:
+                            break
+                        elif datetime.now() - wait_start > timedelta(minutes=10):
+                            ret.check_returncode()
+                        sleep(0.1)
+                    sp.succeed('Mounted')
+                while True:
+                    try:
+                        with tqdm(total=total_size, unit_scale=True, unit_divisor=1024, unit='B', smoothing=0.01, disable=None, dynamic_ncols=True, maxinterval=0.5, miniters=1) as progress:
+                            seen = 0
+                            env = os.environ.copy()
+                            create_args = ['borg',
+                                           'create',
+                                           '--lock-wait=120',
+                                           '--one-file-system',
+                                           '--compression=auto,zstd,10',
+                                           '--chunker-params=10,23,16,4095',
+                                           '--files-cache=ctime,size',
+                                           '--show-rc',
+                                           # '--remote-ratelimit=20480',
+                                           '--log-json',
+                                           '--progress',
+                                           '--list',
+                                           '--filter=AMEi-x?',
+                                           '--stats'
+                                           ]
+                            archive_time = datetime.strptime(entry["time"], "%Y-%m-%dT%H:%M:%S.%f").replace(tzinfo=tzlocal()).astimezone(tzutc())
+                            create_args += [f'--timestamp={archive_time.strftime("%Y-%m-%dT%H:%M:%S")}']
+                            if cache_suffix:
+                                env['BORG_FILES_CACHE_SUFFIX'] = cache_suffix
+                            else:
+                                create_args += ['--files-cache=disabled']
+                            create_args += [f'{dst_repo_path}::{entry["name"]}', '.']
+                            with subprocess.Popen(create_args, cwd=dir, stdin=subprocess.DEVNULL, stderr=subprocess.PIPE, text=True, env=env) as proc:
+                                last_list = None
+                                last_list_time = None
+                                for line in proc.stderr:
+                                    try:
+                                        json_line = json.loads(line)
+                                    except json.decoder.JSONDecodeError:
+                                        tqdm.write(line)
+                                        continue
+
+                                    t = ''
+                                    if 'time' in json_line:
+                                        ts = datetime.fromtimestamp(json_line['time']).replace(tzinfo=tzlocal())
+                                        t = f'{ts.isoformat(timespec="minutes")} '
+                                    if json_line['type'] == 'archive_progress':
+                                        if last_list_time is None or ((datetime.now() - last_list_time) // timedelta(seconds=3)) % 2 == 1:
+                                            if 'path' in json_line and json_line['path']:
+                                                progress.set_description(f'… {json_line["path"]}', refresh=False)
+                                            else:
+                                                progress.set_description(None, refresh=False)
+                                        elif last_list is not None:
+                                            progress.set_description(last_list, refresh=False)
+                                        progress.set_postfix(compressed=naturalsize(json_line['compressed_size'], binary=True), deduplicated=naturalsize(json_line['deduplicated_size'], binary=True), nfiles=f'{json_line["nfiles"]}/{total_files}', refresh=False)
+                                        progress.update(json_line["original_size"] - seen)
+                                        seen = json_line["original_size"]
+                                    elif json_line['type'] == 'file_status':
+                                        # tqdm.write(t + f'{json_line["status"]} {json_line["path"]}')
+                                        last_list = f'{json_line["status"]} {json_line["path"]}'
+                                        last_list_time = datetime.now()
+                                        progress.set_description(last_list, refresh=False)
+                                    elif (json_line['type'] == 'log_message' or json_line['type'] == 'progress_message' or json_line['type'] == 'progress_percent') and ('message' in json_line or 'msgid' in json_line):
+                                        if 'message' in json_line:
+                                            tqdm.write(t + json_line['message'])
+                                        elif 'msgid' in json_line:
+                                            tqdm.write(t + json_line['msgid'])
+                                    else:
+                                        tqdm.write(t + line)
+                                progress.set_description(None)
+                                if proc.wait() != 0:
+                                    continue
+                    except subprocess.CalledProcessError as err:
+                        print(err, file=stderr)
+                        continue
+                    else:
+                        break
+                mount_proc.terminate()
+            os._exit(0)
+        else:
+            while True:
+                waitpid, waitret = os.wait()
+                if waitret != 0:
+                    sys.exit(waitret)
+                if waitpid == child:
+                    break
+
+def main():
+    if "::" in args.source:
+        (src_repo_path, _, src_archive) = args.source.partition("::")
+        entry = None
+        for candidate_entry in read_repo(src_repo_path):
+            if entry['name'] != src_archive:
+                continue
+            entry = candidate_entry
+            break
+
+        if entry is None:
+            print("Did not find archive", file=stderr)
+            os.exit(1)
+
+        copy_archive(src_repo_path, args.target, entry)
+    else:
+        for entry in ToSync():
+            copy_archive(args.source, args.target, entry)
+
+if __name__ == "__main__":
+    sys.exit(main())
diff --git a/hosts/vidhar/borg/default.nix b/hosts/vidhar/borg/default.nix
index ee5856c9..65c309da 100644
--- a/hosts/vidhar/borg/default.nix
+++ b/hosts/vidhar/borg/default.nix
@@ -1,15 +1,72 @@
-{ pkgs, lib, ... }:
+{ config, pkgs, lib, ... }:
 
 with lib;
 
-{
+let
+  copyService = { repo, repoEscaped }: let
+    serviceName = "copy-borg@${repoEscaped}";
+    sshConfig = pkgs.writeText "config" ''
+      Host yggdrasil.borgbase
+        HostName nx69hpl8.repo.borgbase.com
+        User nx69hpl8
+        IdentityFile /run/credentials/${serviceName}.service/ssh-identity
+        IdentitiesOnly yes
+
+        BatchMode yes
+        ServerAliveInterval 10
+        ServerAliveCountMax 30
+    '';
+  in nameValuePair serviceName {
+    serviceConfig = {
+      Type = "oneshot";
+      ExecStart = "${copyBorg}/bin/copy ${escapeShellArg repo} yggdrasil.borgbase:repo";
+      User = "borg";
+      Group = "borg";
+      StateDirectory = "borg";
+      Environment = [
+        "BORG_RSH=\"${pkgs.openssh}/bin/ssh -F ${sshConfig}\""
+        "BORG_CACHE_DIR=/var/lib/borg/cache"
+        "BORG_SECURITY_DIR=/var/lib/borg/security"
+        "BORG_KEYS_DIR=/var/lib/borg/keys"
+        "BORG_KEY_FILE=/run/credentials/${serviceName}.service/keyfile"
+      ];
+      LoadCredential = [
+        "ssh-identity:${config.sops.secrets."append.borgbase".path}"
+        "keyfile:${config.sops.secrets."yggdrasil.borgkey".path}"
+      ];
+    };
+  };
+
+  copyBorg = pkgs.stdenv.mkDerivation rec {
+    name = "copy";
+    src = ./copy.py;
+
+    phases = ["buildPhase" "checkPhase" "installPhase"];
+
+    python = pkgs.python39.withPackages (ps: with ps; [humanize tqdm dateutil xdg python-unshare halo]);
+
+    buildPhase = ''
+      substituteAll $src copy
+    '';
+
+    doCheck = true;
+    checkPhase = ''
+      ${python}/bin/python -m py_compile copy
+    '';
+
+    installPhase = ''
+      install -m 0755 -D -t $out/bin \
+        copy
+    '';
+  };
+in {
   config = {
     services.borgbackup.repos.jotnar = {
       path = "/srv/backup/borg/jotnar";
       authorizedKeysAppendOnly = let
         dir = ./jotnar;
         toAuthKey = fname: ftype: if ftype != "regular" || !(hasSuffix ".pub" fname) then null else builtins.readFile (dir + "/${fname}");
-      in filter (v: v != null) (lib.mapAttrsToList toAuthKey (builtins.readDir dir));
+      in filter (v: v != null) (mapAttrsToList toAuthKey (builtins.readDir dir));
     };
 
     boot.postBootCommands = mkBefore ''
@@ -25,5 +82,24 @@ with lib;
 
       Match All
     '';
+
+    sops.secrets."append.borgbase" = {
+      format = "binary";
+      sopsFile = ./append.borgbase;
+    };
+    sops.secrets."yggdrasil.borgkey" = {
+      format = "binary";
+      sopsFile = ./yggdrasil.borgkey;
+    };
+
+    systemd.services = listToAttrs (map copyService [{ repo = "/srv/backup/borg/jotnar"; repoEscaped = "srv-backup-borg-jotnar"; }]);
+
+    # systemd.timers."copy-borg@srv-backup-borg-jotnar" = {
+    #   wantedBy = ["multi-user.target"];
+      
+    #   timerConfig = {
+    #     OnCalendar = "*-*-* 00/4:00:00 Europe/Berlin";
+    #   };
+    # };
   };
 }
diff --git a/hosts/vidhar/borg/yggdrasil.borgkey b/hosts/vidhar/borg/yggdrasil.borgkey
new file mode 100644
index 00000000..3540792b
--- /dev/null
+++ b/hosts/vidhar/borg/yggdrasil.borgkey
@@ -0,0 +1,26 @@
+{
+        "data": "ENC[AES256_GCM,data:/iyvIA5kkXfeyOb7tVakXHI0UN36GQCB7iyPmC9TsTecYWWo9FM4H3jYI10s7NV0ljysnV9IlPqjGGRIbZJ5M9uZluXOmir5uxPZXO7mn6ZWTPKux6OWkWzLOzgebXBx/wqt3y7EwvPg4wKlIC4+xN5Qyn1YT0RuviMNSB7yVRFn5J6OsE3fVssGw5KB7cw3169JhNpNn8LR6zR/3kai5A7FeFxgjvehDwi/t/gIMjWMsfMgdimILVLyTmE1Q2CKHYIZH69/V3/lzThoxNz3EGgj+XTs116DTZaE0nRadvbUfT+mh0CMV/1RKhfaRZ5qbLWDJdihxcKse4f/k0264upzcnwnbDVz3CYgzTaby9vcMra1/Heb6s0lEV8dyXQUiaBsO6CYnTsNC0ogmwIqINZfGiJOQLBIk4ja5dj60UcO9uKYXr3ZkGYuaVowpWOmgpAfy3PE2eLIj1ZTKAE2Llw2vQFUG9X4i9YI2EHjXCnars88Uoa16Hb6/zbypMh3IKw/Yor/3CKkTIdtiuGZ10/09qlIHnfVjy1tL9lPS92vvoDPA2o1z0utrd8qEvsvS4JiZvz9x3JWUKZN2/DQedarzg53FiFhuvSc4BRtGD5Dqo/2kHGDzzL1BmBVrBfiYtzQmMaXmyOe4WvnK/umujVP7LwhrQheER0Jsg8atQmiPGp4fUQEpo9TPuHx0DRnpEpfVSVXqROGF6Q3KiV492hUGb0JISgfUFpN2ZhvhQS2lgaBGnwfb1CJGgkSKhMQIrvWWVqvwHYE90vIGNZLnsSKedHXxy+jbJqyU4uCKVaRRdXOBVtcFOeNekQO3Ayl0YQTZqi7U174h/6oS2ECePUsKWyYXGiQoNKjyd+edG//IIIFvBgjqBeMmsdzGlxxJE47gMRefSXwk8qk/QZG320uiJtz+G0jVpUUrGwL4ujPXwi1Nj099Oi0Q3SJJ1qvZy4kekvFaoyRlTmXHziupyWKhgnTPj6DCWqI03ORvB5m4y5eWdg90yuj+FyT3FSuIJrbbk8r8HaiirCXEfDE/oTK2A4OCfxph/ebi+WrDQYEXNSCbAMmVQAhSINs,iv:VPlUsquBSSxWcQFQG9Eb/kN/VCRp1ULs2Mixe/M9dNo=,tag:D8isf1CanHzXSNSJ8EK+5g==,type:str]",
+        "sops": {
+                "kms": null,
+                "gcp_kms": null,
+                "azure_kv": null,
+                "hc_vault": null,
+                "age": null,
+                "lastmodified": "2022-02-17T11:07:24Z",
+                "mac": "ENC[AES256_GCM,data:5dmDZTT0+xwtUMLRHxQ8O8pviyzZOtcZXufdRkpbQrCImhk1B4eSm2gaT8GavJYswu3I/Z7Yt6BNeiKkccf/PXWAFsOn7L6R2B52X5TdgUD49HXiLcu9V5Sy2/YDqlCcC1IpxwylilxypP1ht/M19VdPl/vFClQTwsQcwpBujtE=,iv:u90ozqlzOnvp0ly/x1hZAnR67XPo5pWGSvPSbzI5eA4=,tag:WKc64wNitiU/x0Baugky9w==,type:str]",
+                "pgp": [
+                        {
+                                "created_at": "2022-02-17T11:07:24Z",
+                                "enc": "-----BEGIN PGP MESSAGE-----\n\nhF4DbYDvGI0HDr0SAQdAnvb/5Kls/HsLN0dWxkew1E9ppPKI3IDS8fHUun+emnYw\nNJ4XjE2VbM2sdPaAsdeEtmONof8r8k0EEmvV8YFb2iH4EvuwB/LE3sb5Ldjp2QHm\n0l4BS/e7YzESnua/yHA26caeRaqBBbD8mXpKjTaA40v9mbOkpcQpqqP62WO1ox6J\nXLBuV7O1gGjaoWfN/xjkzB2PVsAs5WeTBelMQc0M0/RmlPgOQmOD19SWQop+4npR\n=qLw1\n-----END PGP MESSAGE-----\n",
+                                "fp": "A1C7C95E6CAF0A965CB47277BCF50A89C1B1F362"
+                        },
+                        {
+                                "created_at": "2022-02-17T11:07:24Z",
+                                "enc": "-----BEGIN PGP MESSAGE-----\n\nhF4DXxoViZlp6dISAQdACICrWK61VYsHz08d5cUN4S0zOsmas6/KMs9Eok2+hyAw\n5H5cWdUMtKXCG70Cws3pP9Xq0fRrAQ4ta+HBd38w+fDhm/y4HQPcdv7T7ekcEMHH\n0l4BDO10UfkHAiVrhp5jbpdolkH/0uOb90tZPvN1RGJkDoyJjqp5XTn13c9kfsFg\ni5txaJPTp7XvIBiLLwhmb2z3a1XCDjd1qS2hiaD9c7+fxcanU5a9QwlT5ANnzm/X\n=/xps\n-----END PGP MESSAGE-----\n",
+                                "fp": "30D3453B8CD02FE2A3E7C78C0FB536FB87AE8F51"
+                        }
+                ],
+                "unencrypted_suffix": "_unencrypted",
+                "version": "3.7.1"
+        }
+}
\ No newline at end of file