| import datetime |
| import logging |
| import os |
| import types |
| import uuid |
| from stat import S_ISDIR, S_ISLNK |
|
|
| import paramiko |
|
|
| from .. import AbstractFileSystem |
| from ..utils import infer_storage_options |
|
|
| logger = logging.getLogger("fsspec.sftp") |
|
|
| |
| |
| HOST_KEY_POLICIES = { |
| "auto_add": paramiko.AutoAddPolicy, |
| "reject": paramiko.RejectPolicy, |
| "warning": paramiko.WarningPolicy, |
| } |
|
|
|
|
| def _resolve_host_key_policy(policy): |
| """Turn a ``host_key_policy`` argument into a paramiko policy instance.""" |
| if policy is None: |
| return paramiko.AutoAddPolicy() |
| if isinstance(policy, str): |
| try: |
| return HOST_KEY_POLICIES[policy]() |
| except KeyError: |
| raise ValueError( |
| f"Unknown host_key_policy {policy!r}; expected one of " |
| f"{sorted(HOST_KEY_POLICIES)} or a paramiko.MissingHostKeyPolicy" |
| ) from None |
| if isinstance(policy, type) and issubclass(policy, paramiko.MissingHostKeyPolicy): |
| return policy() |
| if isinstance(policy, paramiko.MissingHostKeyPolicy): |
| return policy |
| raise TypeError( |
| "host_key_policy must be one of " |
| f"{sorted(HOST_KEY_POLICIES)}, a paramiko.MissingHostKeyPolicy subclass " |
| f"or instance, not {type(policy).__name__}" |
| ) |
|
|
|
|
| class SFTPFileSystem(AbstractFileSystem): |
| """Files over SFTP/SSH |
| |
| Peer-to-peer filesystem over SSH using paramiko. |
| |
| Note: if using this with the ``open`` or ``open_files``, with full URLs, |
| there is no way to tell if a path is relative, so all paths are assumed |
| to be absolute. |
| """ |
|
|
| protocol = "sftp", "ssh" |
|
|
| def __init__(self, host, **ssh_kwargs): |
| """ |
| |
| Parameters |
| ---------- |
| host: str |
| Hostname or IP as a string |
| temppath: str |
| Location on the server to put files, when within a transaction |
| host_key_policy: str or paramiko.MissingHostKeyPolicy, optional |
| Policy applied when the server presents a host key that is not |
| known yet, passed to |
| ``paramiko.SSHClient.set_missing_host_key_policy``. Either one of |
| the strings ``"auto_add"`` (default, ``paramiko.AutoAddPolicy``: |
| accept and remember any unknown key), ``"warning"`` |
| (``paramiko.WarningPolicy``: log a warning and accept) or |
| ``"reject"`` (``paramiko.RejectPolicy``: refuse to connect), or a |
| ``paramiko.MissingHostKeyPolicy`` subclass or instance. |
| |
| The default, ``"auto_add"``, trusts any host key presented on the |
| first connection to a host, so a man-in-the-middle attacker can |
| impersonate the server. Pass ``"reject"`` to only connect to hosts |
| whose keys are already in the known hosts of the process. |
| ssh_kwargs: dict |
| Parameters passed on to connection. See details in |
| https://docs.paramiko.org/en/3.3/api/client.html#paramiko.client.SSHClient.connect |
| May include port, username, password... |
| """ |
| if self._cached: |
| return |
| super().__init__(**ssh_kwargs) |
| self.temppath = ssh_kwargs.pop("temppath", "/tmp") |
| self.host_key_policy = _resolve_host_key_policy( |
| ssh_kwargs.pop("host_key_policy", None) |
| ) |
| self.host = host |
| self.ssh_kwargs = ssh_kwargs |
| self._connect() |
|
|
| def _connect(self): |
| logger.debug("Connecting to SFTP server %s", self.host) |
| self.client = paramiko.SSHClient() |
| self.client.set_missing_host_key_policy(self.host_key_policy) |
| self.client.connect(self.host, **self.ssh_kwargs) |
| self.ftp = self.client.open_sftp() |
|
|
| @classmethod |
| def _strip_protocol(cls, path): |
| return infer_storage_options(path)["path"] |
|
|
| @staticmethod |
| def _get_kwargs_from_urls(urlpath): |
| out = infer_storage_options(urlpath) |
| out.pop("path", None) |
| out.pop("protocol", None) |
| return out |
|
|
| def mkdir(self, path, create_parents=True, mode=511): |
| path = self._strip_protocol(path) |
| logger.debug("Creating folder %s", path) |
| if self.exists(path): |
| raise FileExistsError(f"File exists: {path}") |
|
|
| if create_parents: |
| self.makedirs(path) |
| else: |
| self.ftp.mkdir(path, mode) |
|
|
| def makedirs(self, path, exist_ok=False, mode=511): |
| if self.exists(path) and not exist_ok: |
| raise FileExistsError(f"File exists: {path}") |
|
|
| parts = path.split("/") |
| new_path = "/" if path[:1] == "/" else "" |
|
|
| for part in parts: |
| if part: |
| new_path = f"{new_path}/{part}" if new_path else part |
| if not self.exists(new_path): |
| self.ftp.mkdir(new_path, mode) |
|
|
| def rmdir(self, path): |
| path = self._strip_protocol(path) |
| logger.debug("Removing folder %s", path) |
| self.ftp.rmdir(path) |
|
|
| def info(self, path): |
| path = self._strip_protocol(path) |
| stat = self._decode_stat(self.ftp.stat(path)) |
| stat["name"] = path |
| return stat |
|
|
| def created(self, path): |
| """Return the created timestamp of a file as a datetime.datetime""" |
| |
| info = self.info(path) |
| mtime = info.get("mtime", None) |
| if mtime is not None: |
| return mtime |
| raise RuntimeError("Could not retrieve creation time (modification time).") |
|
|
| def modified(self, path): |
| """Return the modified timestamp of a file as a datetime.datetime""" |
| info = self.info(path) |
| mtime = info.get("mtime", None) |
| if mtime is not None: |
| return mtime |
| raise RuntimeError("Could not retrieve modification time.") |
|
|
| @staticmethod |
| def _decode_stat(stat, parent_path=None): |
| if S_ISDIR(stat.st_mode): |
| t = "directory" |
| elif S_ISLNK(stat.st_mode): |
| t = "link" |
| else: |
| t = "file" |
| out = { |
| "name": "", |
| "size": stat.st_size, |
| "type": t, |
| "uid": stat.st_uid, |
| "gid": stat.st_gid, |
| "time": datetime.datetime.fromtimestamp( |
| stat.st_atime, tz=datetime.timezone.utc |
| ), |
| "mtime": datetime.datetime.fromtimestamp( |
| stat.st_mtime, tz=datetime.timezone.utc |
| ), |
| } |
| if parent_path: |
| out["name"] = "/".join([parent_path.rstrip("/"), stat.filename]) |
| return out |
|
|
| def ls(self, path, detail=False): |
| path = self._strip_protocol(path) |
| logger.debug("Listing folder %s", path) |
| stats = [self._decode_stat(stat, path) for stat in self.ftp.listdir_iter(path)] |
| if detail: |
| return stats |
| else: |
| paths = [stat["name"] for stat in stats] |
| return sorted(paths) |
|
|
| def put_file(self, lpath, rpath, callback=None, **kwargs): |
| self.mkdirs(self._parent(os.fspath(rpath)), exist_ok=True) |
| logger.debug("Put file %s into %s", lpath, rpath) |
| self.ftp.put(lpath, rpath) |
|
|
| def get_file(self, rpath, lpath, **kwargs): |
| if self.isdir(rpath): |
| os.makedirs(lpath, exist_ok=True) |
| else: |
| self.ftp.get(self._strip_protocol(rpath), lpath) |
|
|
| def _open(self, path, mode="rb", block_size=None, **kwargs): |
| """ |
| block_size: int or None |
| If 0, no buffering, if 1, line buffering, if >1, buffer that many |
| bytes, if None use default from paramiko. |
| """ |
| logger.debug("Opening file %s", path) |
| if kwargs.get("autocommit", True) is False: |
| |
| path2 = "/".join([self.temppath, str(uuid.uuid4())]) |
| f = self.ftp.open(path2, mode, bufsize=block_size if block_size else -1) |
| f.temppath = path2 |
| f.targetpath = path |
| f.fs = self |
| f.commit = types.MethodType(commit_a_file, f) |
| f.discard = types.MethodType(discard_a_file, f) |
| else: |
| f = self.ftp.open(path, mode, bufsize=block_size if block_size else -1) |
| return f |
|
|
| def _rm(self, path): |
| if self.isdir(path): |
| self.ftp.rmdir(path) |
| else: |
| self.ftp.remove(path) |
|
|
| def mv(self, old, new): |
| new = self._strip_protocol(new) |
| old = self._strip_protocol(old) |
| logger.debug("Renaming %s into %s", old, new) |
| self.ftp.posix_rename(old, new) |
|
|
|
|
| def commit_a_file(self): |
| self.fs.mv(self.temppath, self.targetpath) |
|
|
|
|
| def discard_a_file(self): |
| self.fs._rm(self.temppath) |
|
|
