git-annex-remote-gvfs

#!/usr/bin/env python3
# git-annex-remote-gvfs - a Gvfs/Gio backend for git-annex
#
# (c) 2016-2019 Mantas Mikulėnas <grawity@gmail.com>
# Released under the MIT License (dist/LICENSE.mit)
#
# Sadly Gvfs only has a GTK password prompt, not a textual one, and therefore
# requires X11 or Wayland. It *does not* require GNOME, though, and will work
# with any desktop of your choice. (If you use a GUI file manager to make the
# initial connection and ensure the password is saved in keyring, it *might*
# even work without GUI afterwards.)
#
# Parameters to initremote:
#
#   type=external
#   externaltype=gvfs
#   path=smb://<host>/<share>/<dir>
#   path=sftp://<host>/<dir>
#
# Only tested with smb:// and sftp://, but in theory any gvfs-provided
# filesystem should work.
#
# Chunking is not supported.

import os
import sys
import subprocess

use_gio_cmd = None

def _log(*args):
    if os.environ.get("DEBUG"):
        print("\033[1;35m[gvfs]\033[;35m", *args, end="\033[m\n",
              file=sys.stderr, flush=True)

def _say(msg, *args):
    with open("/dev/tty", "w") as fh:
        print("[gvfs: %s]" % msg, *args, file=fh, end=" ", flush=True)

def find_path(cmd):
    path = os.environ.get("PATH", "/usr/bin").split(":")
    for p in path:
        file = os.path.join(p or ".", cmd)
        if os.path.isfile(file) and os.access(file, os.X_OK):
            return file

def GVFS_CMD(cmd, *args):
    cmds = {
        "ls": "list",
        "rm": "remove",
    }
    if use_gio_cmd:
        argv = ["gio", cmds.get(cmd, cmd), *args]
    else:
        argv = ["gvfs-%s" % cmd, *args]
    #_log("command: %r" % argv)
    return argv

class AnnexStdio(object):
    def send(self, *args):
        _log("stdio: --> %r" % (args,))
        print(*args, flush=True)

    def recv(self, nparams=0):
        line = sys.stdin.readline().rstrip("\r\n")
        _log("stdio: <-- %r" % line)
        if nparams:
            return line.split(" ", nparams)
        else:
            return line

    def loop(self):
        for line in sys.stdin:
            line = line.rstrip("\r\n")
            _log("stdio: <-- %r" % line)
            yield line.split(" ")

    def transact(self, *args):
        self.send(*args)
        cmd, rest = self.recv(nparams=1)
        if cmd == "VALUE":
            return rest
        else:
            raise IOError("transact: expected VALUE, got %r" % cmd)

    def set_config(self, var, value):
        return self.send("SETCONFIG", var, value)

    def get_config(self, var):
        return self.transact("GETCONFIG", var)

    def dir_hash(self, key, layout="mixed"):
        if layout == "mixed":
            return self.transact("DIRHASH", key)
        elif layout == "lower":
            return self.transact("DIRHASH-LOWER", key)
        else:
            raise ValueError("unknown dirhash layout %r" % layout)

class GvfsClient(object):
    def __init__(self):
        self._null = open("/dev/null", "wb")
        self._stderr = open("/dev/tty", "wb")

    def is_mounted(self, uri):
        _log("GvfsClient.is_mounted(%r)" % uri)
        r = subprocess.call(GVFS_CMD("info", "-nfa", "gvfs::backend", uri),
                            stdout=self._null,
                            stderr=self._null)
        return (r == 0)

    def mount(self, uri):
        _log("GvfsClient.mount(%r)" % uri)
        if self.is_mounted(uri):
            return True
        else:
            r = subprocess.call(GVFS_CMD("mount", uri),
                                stdout=self._stderr)
            return (r == 0)

    def has_file(self, uri):
        _log("GvfsClient.has_file(%r)" % uri)
        r = subprocess.call(GVFS_CMD("info", "-a", "access::can-read", uri),
                            stdout=self._null,
                            stderr=self._null)
        return (r == 0)

    def create_dir_p(self, uri):
        _log("GvfsClient.create_dir_p(%r)" % uri)
        if self.has_file(uri):
            return True
        else:
            parent = os.path.dirname(uri)
            if not self.create_dir_p(parent):
                return False
            r = subprocess.call(GVFS_CMD("mkdir", "-p", uri),
                                stdout=self._stderr,
                                stderr=self._stderr)
            return (r == 0)

    def delete_file(self, uri):
        _log("GvfsClient.delete_file(%r)" % uri)
        r = subprocess.call(GVFS_CMD("rm", "-f", uri),
                            stdout=self._stderr,
                            stderr=self._stderr)
        return (r == 0)

    def copy_file(self, src, dst):
        _log("GvfsClient.copy_file(%r -> %r)" % (src, dst))
        r = subprocess.call(GVFS_CMD("copy", "-p", src, dst),
                            stdout=self._stderr,
                            stderr=self._stderr)
        return (r == 0)

    def rename_file(self, src, dst):
        _log("GvfsClient.rename_file(%r -> %r)" % (src, dst))
        r = subprocess.call(GVFS_CMD("move", "-C", src, dst),
                            stdout=self._stderr,
                            stderr=self._stderr)
        return (r == 0)

    def move_file(self, src, dst):
        _log("GvfsClient.move_file(%r -> %r)" % (src, dst))
        r = subprocess.call(GVFS_CMD("move", "-p", src, dst),
                            stdout=self._stderr,
                            stderr=self._stderr)
        return (r == 0)

class AnnexBackend(object):
    def __init__(self, annex):
        self._annex = annex
        self._layout = self._annex.get_config("layout") or "lower"
        self._migrate_layout = False

        if self._layout == "mixed/lower":
            self._layout = "lower"
            self._migrate_layout = True

        if self._layout not in {"mixed", "lower"}:
            raise ValueError("unknown layout %r" % self._layout)

    def key_to_path(self, key, layout=None):
        dir = self._annex.dir_hash(key, layout or self._layout)
        return os.path.join(dir, key, key)

    def has_key(self, key):
        _log("AnnexBackend.has_key(%r)" % key)
        if self.maybe_migrate_key(key):
            return True
        path = self.key_to_path(key)
        return self._has_file(path)

    def store_key(self, key, src):
        _log("AnnexBackend.store_key(%r -> %r)" % (src, key))
        if self.maybe_migrate_key(key):
            return True
        path = self.key_to_path(key)
        return self._import_file(src, path)

    def retrieve_key(self, key, dst):
        _log("AnnexBackend.retrieve_key(%r -> %r)" % (key, dst))
        path = self.key_to_path(key)
        return self._export_file(path, dst)

    def remove_key(self, key):
        _log("AnnexBackend.remove_key(%r)" % key)
        path = self.key_to_path(key)
        if self._has_file(path):
            return self._delete_file(path)
        return True

    def maybe_migrate_key(self, key):
        # return True if key exists and migration has been done
        if self._migrate_layout:
            if self.migrate_layout_key(key):
                _log("maybe_migrate_key: migrated key %r from mixed to lower" % key)
                return True
        return False

    def migrate_layout_key(self, key):
        _log("AnnexBackend.migrate_layout_key(%r)" % key)
        if self._layout == "lower":
            old_layout = "mixed"
        else:
            raise ValueError("only migration from mixed to lower is supported")
        old_path = self.key_to_path(key, layout=old_layout)
        new_path = self.key_to_path(key)
        return self._migrate_key(key, old_path, new_path)

    def _migrate_key(self, key, old_path, new_path):
        if old_path == new_path:
            return False
        if self._has_file(new_path):
            if is_parent_of(old_path, new_path):
                _log("_migrate_key: key present in new, and old is prefix of new; doing nothing")
                return False
            elif self._has_file(old_path):
                _log("_migrate_key: key present in both old & new; deleting old")
                return self._delete_file(old_path)
        elif self._has_file(old_path):
            _log("_migrate_key: renaming key from old to new")
            _say("migrating to new layout")
            return self._rename_file(old_path, new_path)
        else:
            return False

class AnnexGvfsBackend(AnnexBackend):
    """
    AnnexBackend which uses GvfsClient.
    """
    def __init__(self, *args):
        super().__init__(*args)

        # It turns out that current `gio mount` does in fact accept complete URIs,
        # not just the "mountable" prefixes. This allows a single parameter to be
        # used for everything, like in a normal GNOME app.

        self._base = self._annex.get_config("uri")
        if not self._base:
            self._base = self._annex.get_config("path")

        self._volume = self._annex.get_config("mount")
        if not self._volume:
            self._volume = self._base
        elif self._volume == "none":
            self._volume = None

        self.error = None
        self.gvfs = GvfsClient()
        self._mounted = False

    def path_to_uri(self, path):
        assert (path[0] != "/")
        return os.path.join(self._base, path)

    def prepare(self):
        """Remote is being activated for use; don't mount the volume yet because we might
           not need it (e.g. for WHEREIS requests)"""
        if not self._base:
            self.error = "path= must be specified"
            return False
        if self._volume:
            if "://" not in self._volume:
                self.error = "mount= can only be an URL"
                return False
            if self._base.startswith("/"):
                self._base = os.path.join(self._volume, self._base[1:])
                # do the prefix check anyway, for sanity
            if not is_parent_of(self._volume, self._base):
                self.error = "mount= must be a prefix of path="
                return False
        return True

    def mount(self):
        """Delayed on-demand mount"""
        if not self._mounted:
            if not self.prepare():
                return False
            if not self.gvfs.mount(self._volume):
                self.error = "could not mount %r" % self._volume
                return False
            self._mounted = True
        return True

    def initialize(self):
        """Remote is being added; check settings and create base directory if needed."""
        if not self.mount():
            return False
        if not self.gvfs.create_dir_p(self._base):
            self._error = "could not mkdir %r" % self._base
            return False
        return True

    def has_repo(self):
        if not self.mount():
            return False
        return self.gvfs.has_file(self._base)

    def _has_file(self, path):
        if not self.mount():
            return False
        uri = self.path_to_uri(path)
        return self.gvfs.has_file(uri)

    def _delete_file(self, path):
        """
        Delete file from backend.
        """
        if not self.mount():
            return False
        uri = self.path_to_uri(path)
        return self.gvfs.delete_file(uri)

    def _import_file(self, ext_src, dst_path):
        """
        Import & upload a file from local path into the backend.
        """
        if not self.mount():
            return False
        _log("AnnexGvfsBackend._import_file(%r -> %r)" % (ext_src, dst_path))
        tmp_path = "transfer/%s.part" % os.path.basename(dst_path)
        tmp_dir_uri = self.path_to_uri(os.path.dirname(tmp_path))
        dst_dir_uri = self.path_to_uri(os.path.dirname(dst_path))
        tmp_uri = self.path_to_uri(tmp_path)
        dst_uri = self.path_to_uri(dst_path)
        try:
            if not self.gvfs.create_dir_p(tmp_dir_uri):
                raise IOError()
            if not self.gvfs.create_dir_p(dst_dir_uri):
                raise IOError()
            if not self.gvfs.copy_file(ext_src, tmp_uri):
                raise IOError()
            if not self.gvfs.rename_file(tmp_uri, dst_uri):
                raise IOError()
        except IOError:
            self.gvfs.delete_file(tmp_uri)
            return False
        else:
            return True

    def _export_file(self, src_path, ext_dst):
        """
        Retrieve & export a file from the backend to a local path.
        """
        if not self.mount():
            return False
        _log("AnnexGvfsBackend._export_file(%r -> %r)" % (src_path, ext_dst))
        src_uri = self.path_to_uri(src_path)
        ext_tmp = ext_dst + ".part"
        try:
            if not self.gvfs.copy_file(src_uri, ext_tmp):
                raise IOError()
            if not self.gvfs.rename_file(ext_tmp, ext_dst):
                raise IOError()
        except IOError:
            self.gvfs.delete_file(ext_tmp)
            return False
        else:
            return True

    def _rename_file(self, old_path, new_path):
        """
        Rename a file within the backend (when migrating layouts).
        """
        if not self.mount():
            return False
        _log("AnnexGvfsBackend._rename_file(%r -> %r)" % (old_path, new_path))
        old_dir_uri = self.path_to_uri(os.path.dirname(old_path))
        new_dir_uri = self.path_to_uri(os.path.dirname(new_path))
        old_uri = self.path_to_uri(old_path)
        new_uri = self.path_to_uri(new_path)
        try:
            if not self.gvfs.create_dir_p(new_dir_uri):
                raise IOError()
            if not self.gvfs.rename_file(old_uri, new_uri):
                raise IOError()
        except IOError:
            return False
        else:
            return True

def is_parent_of(a, b):
    """
    Check whether 'a' is a parent directory of 'b'.
    """
    a = a.rstrip("/") + "/"
    b = b.rstrip("/") + "/"
    return b.startswith(a)

annex = AnnexStdio()
annex.send("VERSION", 1)

if find_path("gio"):
    use_gio_cmd = True
elif find_path("gvfs-mount"):
    use_gio_cmd = False
else:
    annex.send("ERROR", "neither gio nor gvfs tools found")
    sys.exit(1)

for cmd, *rest in annex.loop():
    if cmd == "INITREMOTE":
        backend = AnnexGvfsBackend(annex)
        if backend.initialize():
            annex.send("INITREMOTE-SUCCESS")
        else:
            annex.send("INITREMOTE-FAILURE", backend.error)
    elif cmd == "PREPARE":
        backend = AnnexGvfsBackend(annex)
        if backend.prepare():
            annex.send("PREPARE-SUCCESS")
        else:
            annex.send("PREPARE-FAILURE", backend.error)
    elif cmd == "TRANSFER":
        action, key, *file = rest
        file = " ".join(file)
        if action == "STORE":
            if backend.store_key(key, file):
                annex.send("TRANSFER-SUCCESS", action, key)
            else:
                annex.send("TRANSFER-FAILURE", action, key)
        elif action == "RETRIEVE":
            if backend.retrieve_key(key, file):
                annex.send("TRANSFER-SUCCESS", action, key)
            else:
                annex.send("TRANSFER-FAILURE", action, key)
        else:
            annex.send("UNSUPPORTED-REQUEST", cmd, action)
    elif cmd == "CHECKPRESENT":
        key, *rest = rest
        if backend.has_repo():
            if backend.has_key(key):
                annex.send("CHECKPRESENT-SUCCESS", key)
            else:
                annex.send("CHECKPRESENT-FAILURE", key)
        else:
            annex.send("CHECKPRESENT-UNKNOWN", key, "remote not available")
    elif cmd == "REMOVE":
        key, *rest = rest
        if backend.has_key(key):
            if backend.remove_key(key):
                annex.send("REMOVE-SUCCESS", key)
            else:
                annex.send("REMOVE-FAILURE", key)
        else:
            annex.send("REMOVE-SUCCESS", key)
    elif cmd == "WHEREIS":
        key, *rest = rest
        backend.maybe_migrate_key(key)
        # http://git-annex.branchable.com/todo/external_remote_querying_transition/
        # ???
        annex.send("WHEREIS-FAILURE")
    else:
        annex.send("UNSUPPORTED-REQUEST")