from __future__ import annotations
__lazy_modules__ = {
"contextlib",
"plumbum.lib",
"plumbum.path",
"plumbum.path.local",
"re",
"tempfile",
}
import contextlib
import re
import typing
from tempfile import NamedTemporaryFile
from typing import TYPE_CHECKING
from plumbum.commands import CommandNotFound, ConcreteCommand, shquote
from plumbum.lib import ProcInfo
from plumbum.machines.base import BaseMachine, PopenWithAddons
from plumbum.machines.env import BaseEnv
from plumbum.path.local import LocalPath
from plumbum.path.remote import RemotePath, RemoteStatRes, RemoteWorkdir
if TYPE_CHECKING:
from collections.abc import Generator, Sequence
from plumbum._compat.typing import Self
from plumbum.commands.async_ import AsyncRemoteCommand
from plumbum.machines.session import ShellSession
[docs]
class RemoteEnv(BaseEnv[RemotePath]):
"""The remote machine's environment; exposes a dict-like interface"""
__slots__ = ["_orig", "remote"]
[docs]
def __init__(self, remote: BaseRemoteMachine) -> None:
session = remote._session
# GNU env has a -0 argument; use it if present. Otherwise,
# fall back to calling printenv on each (possible) variable
# from plain env.
env0 = session.run("env -0; echo")
if env0[0] == 0 and not env0[2].rstrip():
_curr = dict(
line.split("=", 1) for line in env0[1].split("\x00") if "=" in line
)
else:
lines = session.run("env; echo")[1].splitlines()
split = (line.split("=", 1) for line in lines)
keys = (line[0] for line in split if len(line) > 1)
runs = ((key, session.run(f'printenv "{key}"; echo')) for key in keys)
_curr = {
key: run[1].rstrip("\n")
for (key, run) in runs
if run[0] == 0 and run[1].rstrip("\n") and not run[2]
}
super().__init__(remote.path, ":", _curr=_curr)
self.remote = remote
self._orig = dict(self._curr)
[docs]
def __delitem__(self, name: str) -> None:
BaseEnv.__delitem__(self, name)
self.remote._session.run(f"unset {name}")
[docs]
def __setitem__(self, name: str, value: str) -> None:
BaseEnv.__setitem__(self, name, value)
self.remote._session.run(f"export {name}={shquote(value)}")
[docs]
def pop(self, name: str, *default: str) -> str | None:
value = BaseEnv.pop(self, name, *default)
self.remote._session.run(f"unset {name}")
return value
[docs]
def update(self, *args: typing.Any, **kwargs: typing.Any) -> None:
BaseEnv.update(self, *args, **kwargs)
self.remote._session.run(
"export " + " ".join(f"{k}={shquote(v)}" for k, v in self.getdict().items())
)
[docs]
def expand(self, expr: str) -> str:
"""Expands any environment variables and home shortcuts found in ``expr``
(like ``os.path.expanduser`` combined with ``os.path.expandvars``)
:param expr: An expression containing environment variables (as ``$FOO``) or
home shortcuts (as ``~/.bashrc``)
:returns: The expanded string"""
return self.remote.expand(expr)
[docs]
def expanduser(self, expr: str) -> str:
"""Expand home shortcuts (e.g., ``~/foo/bar`` or ``~john/foo/bar``)
:param expr: An expression containing home shortcuts
:returns: The expanded string"""
return self.remote.expanduser(expr)
# def clear(self):
# BaseEnv.clear(self, *args, **kwargs)
# self.remote._session.run("export %s" % " ".join("%s=%s" % (k, v) for k, v in self.getdict()))
[docs]
def getdelta(self) -> dict[str, str]:
"""Returns the difference between the this environment and the original environment of
the remote machine"""
self._curr["PATH"] = self.path.join()
delta = {}
for k, v in self._curr.items():
if k not in self._orig:
delta[k] = str(v)
for k, v in self._orig.items():
if k not in self._curr:
delta[k] = ""
elif v != self._curr[k]:
delta[k] = self._curr[k]
return delta
[docs]
class RemoteCommand(ConcreteCommand):
__slots__ = ("remote",)
QUOTE_LEVEL = 1
[docs]
def __init__(
self, remote: BaseRemoteMachine, executable: RemotePath, encoding: str = "auto"
) -> None:
self.remote = remote
ConcreteCommand.__init__(
self, executable, remote.custom_encoding if encoding == "auto" else encoding
)
@property
def machine(self) -> BaseRemoteMachine:
return self.remote
[docs]
def __repr__(self) -> str:
return f"RemoteCommand({self.remote!r}, {self.executable!r})"
[docs]
def popen(
self, args: Sequence[typing.Any] | str = (), **kwargs: typing.Any
) -> PopenWithAddons[str]:
return self.remote.popen(self[args], **kwargs) # type: ignore[arg-type]
[docs]
def nohup(
self,
cwd: str = ".",
stdout: str = "nohup.out",
stderr: str | None = None,
append: bool = True,
) -> PopenWithAddons[str]:
"""Runs a command detached."""
return self.machine.daemonic_popen(self, cwd, stdout, stderr, append)
[docs]
class ClosedRemoteMachine(Exception):
pass
class ClosedRemote:
__slots__ = ["__weakref__", "_obj"]
def __init__(self, obj: object) -> None:
self._obj = obj
def close(self) -> None:
pass
def __getattr__(
self, name: str
) -> typing.NoReturn: # pragma: no cover - always raises
raise ClosedRemoteMachine(f"{self._obj!r} has been closed")
def _is_recursive_glob(pattern: str) -> bool:
"""Whether ``pattern`` uses ``**`` as a recursive wildcard.
As in :mod:`glob`/:mod:`pathlib`, ``**`` is only special when it is a whole
path segment; ``a**b`` is just two ``*`` wildcards within one segment.
"""
return "**" in pattern.split("/")
def _segment_to_regex(segment: str) -> str:
"""Translate a single (slash-free) glob segment into a regex fragment.
Supports ``*``, ``?`` and ``[...]`` character classes, all confined to a
single path component (they never cross ``/``). A leading wildcard does not
match a leading dot, mirroring :func:`glob.glob` (which skips dotfiles
unless the pattern segment starts with a literal ``.``).
"""
out = []
# A wildcard at the start of a segment must not match a leading dot.
if segment[:1] in ("*", "?", "["):
out.append(r"(?!\.)")
i, n = 0, len(segment)
while i < n:
c = segment[i]
if c == "*":
out.append("[^/]*")
i += 1
elif c == "?":
out.append("[^/]")
i += 1
elif c == "[":
j = i + 1
if j < n and segment[j] == "!":
j += 1
if j < n and segment[j] == "]":
j += 1
while j < n and segment[j] != "]":
j += 1
if j >= n: # no closing bracket -- treat "[" as a literal
out.append(re.escape("["))
i += 1
else:
stuff = segment[i + 1 : j].replace("\\", r"\\")
if stuff[0] == "!":
stuff = "^" + stuff[1:]
elif stuff[0] in ("^", "["):
stuff = "\\" + stuff
out.append(f"[{stuff}]")
i = j + 1
else:
out.append(re.escape(c))
i += 1
return "".join(out)
def _glob_to_regex(pattern: str) -> str:
"""Translate a glob pattern into an anchored regex with pathlib-like ``**``.
``**`` as a whole path segment matches any number of (non-hidden)
directories; ``*``, ``?`` and ``[...]`` match within a single path segment.
Dotfiles are not matched unless the relevant pattern segment starts with a
literal ``.`` -- matching :func:`glob.glob`, which is the local backend.
Used to match recursive globs in Python instead of relying on
shell-specific recursion support.
"""
segments = pattern.split("/")
out = []
need_sep = False # whether a "/" must precede the next fragment
for i, segment in enumerate(segments):
if segment == "**":
if need_sep:
out.append("/")
if i == len(segments) - 1:
# trailing ``**``: one or more non-hidden path components
out.append(r"(?!\.)[^/]+(?:/(?!\.)[^/]+)*")
need_sep = False
else:
# ``**/``: zero or more non-hidden directories (slash included)
out.append(r"(?:(?!\.)[^/]+/)*")
need_sep = False
continue
if need_sep:
out.append("/")
out.append(_segment_to_regex(segment))
need_sep = True
return "(?s:" + "".join(out) + r")\Z"
[docs]
class BaseRemoteMachine(BaseMachine):
"""Represents a *remote machine*; serves as an entry point to everything related to that
remote machine, such as working directory and environment manipulation, command creation,
etc.
Attributes:
* ``cwd`` - the remote working directory
* ``env`` - the remote environment
* ``custom_encoding`` - the remote machine's default encoding (assumed to be UTF8)
* ``connect_timeout`` - the connection timeout
There also is a _cwd attribute that exists if the cwd is not current (del if cwd is changed).
"""
__slots__ = (
"_cwd",
"_program_cache",
"_python",
"_session",
"connect_timeout",
"env",
"uname",
)
# allow inheritors to override the RemoteCommand class
RemoteCommand = RemoteCommand
@property
def cwd(self) -> RemoteWorkdir:
if not hasattr(self, "_cwd"):
self._cwd = RemoteWorkdir(self)
return self._cwd
[docs]
def __init__(
self,
encoding: str = "utf8",
connect_timeout: float | None = 10,
new_session: bool = False,
) -> None:
self.custom_encoding = encoding
self.connect_timeout = connect_timeout
self._session: ShellSession | ClosedRemote = self.session(
new_session=new_session
)
self.uname = self._get_uname()
self.env = RemoteEnv(self)
self._python: ConcreteCommand | None = None
self._program_cache: dict[tuple[str, str], RemotePath] = {}
[docs]
def clear_program_cache(self) -> None:
self._program_cache.clear()
def _get_uname(self) -> str:
rc, out, _ = self._session.run("uname", retcode=None)
if rc == 0:
return out.strip()
rc, out, _ = self._session.run(
"python3 -c 'import platform;print(platform.uname()[0])'", retcode=None
)
if rc == 0:
return out.strip()
# all POSIX systems should have uname. make an educated guess it's Windows
return "Windows"
[docs]
def __repr__(self) -> str:
return f"<{self.__class__.__name__} {self}>"
def __enter__(self) -> Self:
return self
def __exit__(self, t: object, v: object, tb: object) -> None:
self.close()
def __del__(self) -> None:
with contextlib.suppress(Exception):
self.close()
[docs]
def close(self) -> None:
"""closes the connection to the remote machine; all paths and programs will
become defunct"""
self._session.close()
self._session = ClosedRemote(self)
[docs]
def path(self, *parts: str | RemotePath | LocalPath) -> RemotePath:
"""A factory for :class:`RemotePaths <plumbum.path.remote.RemotePath>`.
Usage: ``p = rem.path("/usr", "lib", "python2.7")``
"""
parts2 = [str(self.cwd)]
for p in parts:
if isinstance(p, LocalPath):
raise TypeError(f"Cannot construct RemotePath from {p!r}")
parts2.append(self.expanduser(str(p)))
return RemotePath(self, *parts2)
[docs]
def which(self, progname: str) -> RemotePath:
"""Looks up a program in the ``PATH``. If the program is not found, raises
:class:`CommandNotFound <plumbum.commands.processes.CommandNotFound>`
:param progname: The program's name. Note that if underscores (``_``) are present
in the name, and the exact name is not found, they will be replaced
in turn by hyphens (``-``) then periods (``.``), and the name will
be looked up again for each alternative
:returns: A :class:`RemotePath <plumbum.path.remote.RemotePath>`
"""
key = (progname, self.env.get("PATH", ""))
with contextlib.suppress(KeyError):
return self._program_cache[key]
alternatives = [progname]
if "_" in progname:
alternatives += [progname.replace("_", "-"), progname.replace("_", ".")]
for name in alternatives:
for p in self.env.path:
fn = p / name
if fn.access("x") and not fn.is_dir():
self._program_cache[key] = fn
return fn
raise CommandNotFound(progname, self.env.path)
[docs]
def __getitem__(self, cmd: str | RemotePath | LocalPath) -> ConcreteCommand:
"""Returns a `Command` object representing the given program. ``cmd`` can be a string or
a :class:`RemotePath <plumbum.path.remote.RemotePath>`; if it is a path, a command
representing this path will be returned; otherwise, the program name will be looked up in
the system's ``PATH`` (using ``which``). Usage::
r_ls = rem["ls"]
"""
if isinstance(cmd, RemotePath):
if cmd.remote is self:
return self.RemoteCommand(self, cmd)
raise TypeError(
f"Given path does not belong to this remote machine: {cmd!r}"
)
if not isinstance(cmd, LocalPath):
return self.RemoteCommand(
self, self.path(cmd) if "/" in cmd or "\\" in cmd else self.which(cmd)
)
raise TypeError(f"cmd must not be a LocalPath: {cmd!r}")
@property
def python(self) -> ConcreteCommand:
"""A command that represents the default remote python interpreter"""
if not self._python:
self._python = self["python3"]
return self._python
[docs]
def session(
self, isatty: bool = False, *, new_session: bool = False
) -> ShellSession:
"""Creates a new :class:`ShellSession <plumbum.machines.session.ShellSession>` object; this invokes the user's
shell on the remote machine and executes commands on it over stdin/stdout/stderr
"""
raise NotImplementedError()
[docs]
def download(self, src: str | RemotePath, dst: str | LocalPath) -> None:
"""Downloads a remote file/directory (``src``) to a local destination (``dst``).
``src`` must be a string or a :class:`RemotePath <plumbum.path.remote.RemotePath>`
pointing to this remote machine, and ``dst`` must be a string or a
:class:`LocalPath <plumbum.path.local.LocalPath>`"""
raise NotImplementedError()
[docs]
def upload(self, src: str | LocalPath, dst: str | RemotePath) -> None:
"""Uploads a local file/directory (``src``) to a remote destination (``dst``).
``src`` must be a string or a :class:`LocalPath <plumbum.path.local.LocalPath>`,
and ``dst`` must be a string or a :class:`RemotePath <plumbum.path.remote.RemotePath>`
pointing to this remote machine"""
raise NotImplementedError()
[docs]
def popen(
self, args: Sequence[typing.Any] | str, **kwargs: typing.Any
) -> PopenWithAddons[str]:
"""Spawns the given command on the remote machine, returning a ``Popen``-like object;
do not use this method directly, unless you need "low-level" control on the remote
process"""
raise NotImplementedError()
[docs]
def list_processes(self) -> Generator[ProcInfo, None, None]:
"""
Returns information about all running processes (on POSIX systems: using ``ps``)
.. versionadded:: 1.3
"""
ps = self["ps"]
lines = ps("-e", "-o", "pid,uid,stat,args").splitlines()
lines.pop(0) # header
for line in lines:
parts = line.strip().split()
yield ProcInfo(int(parts[0]), int(parts[1]), parts[2], " ".join(parts[3:]))
[docs]
def pgrep(self, pattern: str) -> Generator[ProcInfo, None, None]:
"""
Process grep: return information about all processes whose command-line args match the given regex pattern
"""
pat = re.compile(pattern)
for procinfo in self.list_processes():
if pat.search(procinfo.args):
yield procinfo
[docs]
@contextlib.contextmanager
def tempdir(self) -> Generator[RemotePath, None, None]:
"""A context manager that creates a remote temporary directory, which is removed when
the context exits"""
_, out, _ = self._session.run(
"mktemp -d 2>/dev/null || mktemp -d tmp.XXXXXXXXXX"
)
local_dir = self.path(out.strip())
try:
yield local_dir
finally:
local_dir.delete()
#
# Path implementation
#
def _path_listdir(self, fn: str) -> list[str]:
files = self._session.run(f"ls -a {shquote(fn)}")[1].splitlines()
files.remove(".")
files.remove("..")
return files
def _path_glob(self, fn: str, pattern: str) -> list[str]:
if _is_recursive_glob(pattern):
# Recursive glob (``**``). The shell loop below cannot do this
# portably: ``/bin/sh`` is often dash, and ``**`` only recurses in
# bash with ``globstar`` enabled. Instead, enumerate the tree with
# POSIX ``find`` and match in Python, so the result is identical
# regardless of the remote shell -- and free of the shell-glob
# quirks that bite paths containing glob metacharacters.
regex = re.compile(_glob_to_regex(pattern))
# ``find`` exits non-zero on partial errors (e.g. an unreadable
# subdirectory) while still printing the matches it did find, so
# match whatever was printed rather than discarding it -- this
# mirrors ``glob.glob``, which does not error on unreadable subdirs.
_, out, _ = self._session.run(f"find {shquote(fn)}", retcode=None)
prefix = fn.rstrip("/") + "/"
return sorted(
line
for line in out.splitlines()
if line.startswith(prefix) and regex.match(line[len(prefix) :])
)
# shquote does not work here due to the way bash loops use space as a separator
pattern = pattern.replace(" ", r"\ ")
fn = fn.replace(" ", r"\ ")
matches = self._session.run(rf"for fn in {fn}/{pattern}; do echo $fn; done")[
1
].splitlines()
if len(matches) == 1 and not self._path_stat(matches[0]):
return [] # pattern expansion failed
return matches
def _path_getuid(self, fn: str) -> list[str]:
stat_cmd = (
"stat -c '%u,%U' "
if self.uname not in ("Darwin", "FreeBSD")
else "stat -f '%u,%Su' "
)
return self._session.run(stat_cmd + shquote(fn))[1].strip().split(",")
def _path_getgid(self, fn: str) -> list[str]:
stat_cmd = (
"stat -c '%g,%G' "
if self.uname not in ("Darwin", "FreeBSD")
else "stat -f '%g,%Sg' "
)
return self._session.run(stat_cmd + shquote(fn))[1].strip().split(",")
def _path_stat(self, fn: str) -> RemoteStatRes | None:
if self.uname not in ("Darwin", "FreeBSD"):
stat_cmd = "stat -c '%F,%f,%i,%d,%h,%u,%g,%s,%X,%Y,%Z' "
else:
stat_cmd = "stat -f '%HT,%Xp,%i,%d,%l,%u,%g,%z,%a,%m,%c' "
rc, out, _ = self._session.run(stat_cmd + shquote(fn), retcode=None)
if rc != 0:
return None
statres = out.strip().split(",")
text_mode = statres.pop(0).lower()
res = RemoteStatRes(
(int(statres[0], 16), *tuple(int(sr) for sr in statres[1:])) # type: ignore[arg-type]
)
res.text_mode = text_mode
return res
def _path_delete(self, fn: str) -> None:
self._session.run(f"rm -rf {shquote(fn)}")
def _path_move(self, src: str, dst: str) -> RemotePath:
self._session.run(f"mv {shquote(src)} {shquote(dst)}")
return RemotePath(self, dst)
def _path_copy(self, src: str, dst: str) -> RemotePath:
self._session.run(f"cp -r {shquote(src)} {shquote(dst)}")
return RemotePath(self, dst)
def _path_mkdir(
self,
fn: str,
mode: int | None = None, # noqa: ARG002
minus_p: bool = True,
) -> None:
p_str = "-p " if minus_p else ""
cmd = f"mkdir {p_str}{shquote(fn)}"
self._session.run(cmd)
def _path_chmod(self, mode: int, fn: str) -> None:
self._session.run(f"chmod {mode:o} {shquote(fn)}")
def _path_touch(self, path: str) -> None:
self._session.run(f"touch {shquote(path)}")
def _path_chown(
self,
fn: str,
owner: int | str | None,
group: int | str | None,
recursive: bool,
) -> None:
args = ["chown"]
if recursive:
args.append("-R")
if owner is not None and group is not None:
args.append(f"{owner}:{group}")
elif owner is not None:
args.append(str(owner))
elif group is not None:
args.append(f":{group}")
args.append(shquote(fn))
self._session.run(" ".join(args))
def _path_read(self, fn: str) -> bytes:
data = self["cat"](fn)
if self.custom_encoding and isinstance(data, str):
return data.encode(self.custom_encoding)
return typing.cast("bytes", data)
def _path_write(self, fn: str, data: bytes | str) -> None:
if self.custom_encoding and isinstance(data, str):
data = data.encode(self.custom_encoding)
assert isinstance(data, (bytes, bytearray))
with NamedTemporaryFile() as f:
f.write(data)
f.flush()
f.seek(0)
self.upload(f.name, fn)
def _path_link(self, src: str, dst: str, symlink: bool) -> None:
symlink_str = "-s " if symlink else ""
self._session.run(f"ln {symlink_str}{shquote(src)} {shquote(dst)}")
def expand(self, expr: str) -> str:
return self._session.run(f"echo {expr}")[1].strip()
def expanduser(self, expr: str) -> str:
if not any(part.startswith("~") for part in expr.split("/")):
return expr
# we escape all $ signs to avoid expanding env-vars
expr_repl = expr.replace("$", "\\$")
return self._session.run(f"echo {expr_repl}")[1].strip()
[docs]
class AsyncRemoteMachine:
"""Async version of BaseRemoteMachine.
This class provides async access to remote commands via SSH.
It wraps a sync RemoteMachine and provides async execution methods.
Example::
from plumbum.machines.ssh_machine import AsyncSshMachine
async with AsyncSshMachine("host") as rem:
ls = rem["ls"]
result = await ls("-la")
.. versionadded:: 2.0
"""
__slots__ = ("_sync_machine",)
[docs]
def __init__(self, sync_machine: BaseRemoteMachine):
"""Initialize with a sync remote machine to wrap.
Args:
sync_machine: The sync remote machine to wrap
"""
self._sync_machine = sync_machine
[docs]
def __getitem__(self, cmd: str | RemotePath | LocalPath) -> AsyncRemoteCommand:
"""Get an async remote command by name or path.
This delegates to the sync machine for command lookup, then wraps it.
Args:
cmd: Command name (will be looked up in PATH) or RemotePath
Returns:
AsyncRemoteCommand instance
Raises:
CommandNotFound: If command is not found in PATH
"""
from plumbum.commands.async_ import AsyncRemoteCommand
sync_cmd = self._sync_machine[cmd]
return AsyncRemoteCommand(sync_cmd)
[docs]
def __contains__(self, cmd: str) -> bool:
"""Check if a command exists in remote PATH."""
return cmd in self._sync_machine
@property
def cwd(self) -> RemoteWorkdir:
"""Current working directory on remote machine."""
return self._sync_machine.cwd
@property
def env(self) -> RemoteEnv:
"""Environment variables on remote machine."""
return self._sync_machine.env
[docs]
def path(self, *parts: str | RemotePath | LocalPath) -> RemotePath:
"""Create a RemotePath from parts."""
return self._sync_machine.path(*parts)
[docs]
def close(self) -> None:
"""Close the connection to the remote machine."""
self._sync_machine.close()
[docs]
async def __aenter__(self) -> Self:
"""Async context manager entry."""
return self
[docs]
async def __aexit__(self, t: object, v: object, tb: object) -> None:
"""Async context manager exit."""
self.close()
__all__ = [
"AsyncRemoteMachine",
"BaseRemoteMachine",
"ClosedRemote",
"ClosedRemoteMachine",
"RemoteCommand",
"RemoteEnv",
]
def __dir__() -> list[str]:
return list(__all__)