8912ca5a8c3a9725c3ba6d30561607150a6faebe-PR-2205

pending completion

Build # 8912ca5a8c3a9725c3ba6d30561607150a6faebe-PR-2205

Build Type

Pull #2205

github-actions

Committed by

web-flow

Commit Message

Merge 81f463e2c into 8b4cacf8b

Pull Request Pull Request #2205: Fix stable Python 2 installation from a built wheel

Run Details

3878 of 6371 branches covered (60.87%)

12199 of 16604 relevant lines covered (73.47%)

0.73 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

76.63

/pwnlib/filesystem/ssh.py

# -*- coding: utf-8 -*-
"""
Handles file abstraction for remote SSH files

Emulates pathlib as much as possible, but does so through duck typing.
"""
import os
import six
import sys
import tempfile
import time

from pwnlib.context import context
from pwnlib.util.misc import read, write
from pwnlib.util.packing import _encode, _decode

if six.PY3:
    from pathlib import *
else:
    from pathlib2 import *

class SSHPath(PosixPath):
    r"""Represents a file that exists on a remote filesystem.

    See :class:`.ssh` for more information on how to set up an SSH connection.
    See :py:class:`pathlib.Path` for documentation on what members and
    properties this object has.

    Arguments:
        name(str): Name of the file
        ssh(ssh): :class:`.ssh` object for manipulating remote files

    Note:

        You can avoid having to supply ``ssh=`` on every ``SSHPath`` by setting
        :data:`.context.ssh_session`.  
        In these examples we provide ``ssh=`` for clarity.

    Examples:

        First, create an SSH connection to the server.

        >>> ssh_conn = ssh('travis', 'example.pwnme')

        Let's use a temporary directory for our tests
    
        >>> _ = ssh_conn.set_working_directory()

        Next, you can create SSHPath objects to represent the paths to files
        on the remote system.

        >>> f = SSHPath('filename', ssh=ssh_conn)
        >>> f.touch()
        >>> f.exists()
        True
        >>> f.resolve().path # doctests: +ELLIPSIS
        '/tmp/.../filename'
        >>> f.write_text('asdf ❤️')
        >>> f.read_bytes()
        b'asdf \xe2\x9d\xa4\xef\xb8\x8f'

        ``context.ssh_session`` must be set to use the :meth:`.SSHPath.mktemp`
        or :meth:`.SSHPath.mkdtemp` methods.

        >>> context.ssh_session = ssh_conn
        >>> SSHPath.mktemp() # doctest: +ELLIPSIS
        SSHPath('...', ssh=ssh(user='travis', host='127.0.0.1'))
    """

    sep = '/'

    def __init__(self, path, ssh=None):
        self.path = self._s(path)
        self.ssh = ssh or context.ssh_session

        if self.ssh is None:
            raise ValueError('SSHPath requires an ssh session.  Provide onee or set context.ssh_session.')

    def _s(self, other):
        # We want strings
        if isinstance(other, str):
            return other

        # We don't want unicode
        if isinstance(other, six.text_type):
            return str(other)

        # We also don't want binary
        if isinstance(other, six.binary_type):
            return str(_decode(other))

    def _new(self, path, *a, **kw):
        kw['ssh'] = self.ssh
        path = self._s(path)
        return SSHPath(path, *a, **kw)

    def _run(self, *a, **kw):
        with context.silent:
            return self.ssh.run(*a, **kw)

#---------------------------------- PUREPATH ----------------------------------
    def __str__(self):
        return self.path

    def __fspath__(self):
        return str(self)

    def as_posix(self):
        return self.path

    def __bytes__(self):
        return os.fsencode(self)

    def __repr__(self):
        return "{}({!r}, ssh={!r})".format(self.__class__.__name__, self.as_posix(), self.ssh)

    def as_uri(self):
        raise NotImplementedError()

    def __eq__(self, other):
        if not isinstance(other, SSHPath):
            return str(self) == str(other)

        if self.ssh.host != other.ssh.host:
            return False

        if self.path != other.path:
            return False

        return True

    def __hash__(*a, **kw): ""; raise NotImplementedError
    def __lt__(*a, **kw): ""; raise NotImplementedError
    def __le__(*a, **kw): ""; raise NotImplementedError
    def __gt__(*a, **kw): ""; raise NotImplementedError
    def __ge__(*a, **kw): ""; raise NotImplementedError

    @property
    def anchor(self):
        raise NotImplementedError()

    @property
    def name(self):
        """Returns the name of the file.

        >>> f = SSHPath('hello', ssh=ssh_conn)
        >>> f.name
        'hello'
        """
        return os.path.basename(self.path)

    @property
    def suffix(self):
        """Returns the suffix of the file.

        >>> f = SSHPath('hello.tar.gz', ssh=ssh_conn)
        >>> f.suffix
        '.gz'
        """
        if '.' not in self.name:
            return ''
        return self.name[self.name.rindex('.'):]

    @property
    def suffixes(self):
        """Returns the suffixes of a file

        >>> f = SSHPath('hello.tar.gz', ssh=ssh_conn)
        >>> f.suffixes
        '.tar.gz'
        """

        basename = self.name
        if '.' not in basename:
            return ''
        return '.' + self.name.split('.', 1)[1]

    @property
    def stem(self):
        """Returns the stem of a file without any extension
        
        >>> f = SSHPath('hello.tar.gz', ssh=ssh_conn)
        >>> f.stem
        'hello'
        """
        if '.' not in self.name:
            return self.name
        return self.name[:self.name.index('.')]

    def with_name(self, name):
        """Return a new path with the file name changed

        >>> f = SSHPath('hello/world', ssh=ssh_conn)
        >>> f.path
        'hello/world'
        >>> f.with_name('asdf').path
        'hello/asdf'
        """
        if '/' not in self.path:
            return name

        path, _ = self.path.split(self.sep, 1)
        path = self._new(path)
        return path.joinpath(name)

    def with_stem(self, name):
        """Return a new path with the stem changed.

        >>> f = SSHPath('hello/world.tar.gz', ssh=ssh_conn)
        >>> f.with_stem('asdf').path
        'hello/asdf.tar.gz'
        """
        return self.with_name(name + self.suffixes)

    def with_suffix(self, suffix):
        """Return a new path with the file suffix changed

        >>> f = SSHPath('hello/world.tar.gz', ssh=ssh_conn)
        >>> f.with_suffix('.tgz').path
        'hello/world.tgz'
        """
        return self.with_name(self.stem + suffix)

    def relative_to(self, *other):
        raise NotImplementedError()

    def is_relative_to(self, *other):
        raise NotImplementedError()       

    @property
    def parts(self):
        """Return the individual parts of the path
    
        >>> f = SSHPath('hello/world.tar.gz', ssh=ssh_conn)
        >>> f.parts
        ['hello', 'world.tar.gz']
        """
        return self.path.split(self.sep)

    def joinpath(self, *args):
        """Combine this path with one or several arguments.

        >>> f = SSHPath('hello', ssh=ssh_conn)
        >>> f.joinpath('world').path
        'hello/world'
        """
        newpath = os.path.join(self.path, *args)
        return SSHPath(newpath, ssh=self.ssh)
    
    # __truediv__
    # __rtruediv__

    @property
    def parent(self):
        """Return the parent of this path

        >>> f = SSHPath('hello/world/file.txt', ssh=ssh_conn)
        >>> f.parent.path
        'hello/world'
        """
        a, b = self.path.rsplit(self.sep, 1)
        if a:
            return self._new(a)
        return self

    @property
    def parents(self):
        """Return the parents of this path, as individual parts

        >>> f = SSHPath('hello/world/file.txt', ssh=ssh_conn)
        >>> list(p.path for p in f.parents)
        ['hello', 'world']
        """
        if '/' not in self.path:
            return self._new('.')

        return [self._new(p) for p in self.parent.path.split(self.sep)]

    def is_absolute(self):
        """Returns whether a path is absolute or not.

        >>> f = SSHPath('hello/world/file.txt', ssh=ssh_conn)
        >>> f.is_absolute()
        False

        >>> f = SSHPath('/hello/world/file.txt', ssh=ssh_conn)
        >>> f.is_absolute()
        True
        """
        return self.path.startswith(self.sep)

    def is_reserved(self):
        return False

    def match(self, path_pattern):
        raise NotImplementedError()

#------------------------------------ PATH ------------------------------------

    @property
    def cwd(self):
        return self._new(self.ssh.cwd)

    @property
    def home(self):
        """Returns the home directory for the SSH connection

        >>> f = SSHPath('...', ssh=ssh_conn)
        >>> f.home # doctest: +ELLIPSIS
        SSHPath('/home/...', ssh=ssh(user='...', host='127.0.0.1'))
        """
        path = self._run('echo ~').recvall().rstrip()
        return self._new(path)

    def samefile(self, other_path):
        """Returns whether two files are the same

        >>> a = SSHPath('a', ssh=ssh_conn)
        >>> A = SSHPath('a', ssh=ssh_conn)
        >>> x = SSHPath('x', ssh=ssh_conn)

        >>> a.samefile(A)
        True
        >>> a.samefile(x)
        False
        """
        if not isinstance(other_path, SSHPath):
            return False

        return self.absolute() == other_path.absolute()

    def iterdir(self):
        """Iterates over the contents of the directory

        >>> directory = SSHPath('iterdir', ssh=ssh_conn)
        >>> directory.mkdir()
        >>> fileA = directory.joinpath('fileA')
        >>> fileA.touch()
        >>> fileB = directory.joinpath('fileB')
        >>> fileB.touch()
        >>> dirC = directory.joinpath('dirC')
        >>> dirC.mkdir()
        >>> [p.name for p in directory.iterdir()]
        ['dirC', 'fileA', 'fileB']
        """
        for directory in sorted(self.ssh.sftp.listdir(self.path)):
            yield self._new(directory)

    def glob(self, pattern):
        raise NotImplementedError()

    def rglob(self, pattern):
        raise NotImplementedError()

    def absolute(self):
        """Return the absolute path to a file, preserving e.g. "../".
        The current working directory is determined via the :class:`.ssh`
        member :attr:`.ssh.cwd`.

        Example:
            
            >>> f = SSHPath('absA/../absB/file', ssh=ssh_conn)
            >>> f.absolute().path # doctest: +ELLIPSIS
            '/.../absB/file'
        """
        path = os.path.normpath(self.path)

        if self.is_absolute():
            return self._new(path)

        return self._new(os.path.join(self.ssh.cwd, path))

    def resolve(self, strict=False):
        """Return the absolute path to a file, resolving any '..' or symlinks.
        The current working directory is determined via the :class:`.ssh`
        member :attr:`.ssh.cwd`.

        Note:

            The file must exist to call resolve().

        Examples:

            >>> f = SSHPath('resA/resB/../resB/file', ssh=ssh_conn)

            >>> f.resolve().path # doctest: +ELLIPSIS
            Traceback (most recent call last):
            ...
            ValueError: Could not normalize path: '/.../resA/resB/file'

            >>> f.parent.absolute().mkdir(parents=True)
            >>> list(f.parent.iterdir())
            []

            >>> f.touch()
            >>> f.resolve() # doctest: +ELLIPSIS
            SSHPath('/.../resA/resB/file', ssh=ssh(user='...', host='127.0.0.1'))
        """
        path = self.absolute().path
        path = os.path.normpath(path)

        if six.PY2:
            error_type = IOError
        else:
            error_type = FileNotFoundError

        try:
            return self._new(self.ssh.sftp.normalize(path))
        except error_type as e:
            raise ValueError("Could not normalize path: %r" % path)

    def stat(self):
        """Returns the permissions and other information about the file

        >>> f = SSHPath('filename', ssh=ssh_conn)
        >>> f.touch()
        >>> stat = f.stat()
        >>> stat.st_size
        0
        >>> '%o' % stat.st_mode # doctest: +ELLIPSIS
        '...664'
        """
        return self.ssh.sftp.stat(self.path)

    def owner(self):
        raise NotImplementedError()

    def group(self):
        raise NotImplementedError()

    def open(self, *a, **kw):
        """Return a file-like object for this path.

        This currently seems to be broken in Paramiko.

        >>> f = SSHPath('filename', ssh=ssh_conn)
        >>> f.write_text('Hello')
        >>> fo = f.open(mode='r+')
        >>> fo                      # doctest: +ELLIPSIS
        <paramiko.sftp_file.SFTPFile object at ...>
        >>> fo.read('asdfasdf')     # doctest: +SKIP
        b'Hello'
        """
        return self.ssh.sftp.open(self.path, *a, **kw)

    def read_bytes(self):
        """Read bytes from the file at this path

        >>> f = SSHPath('/etc/passwd', ssh=ssh_conn)
        >>> f.read_bytes()[:10]
        b'root:x:0:0'
        """
        return self.ssh.read(str(self.absolute()))

    def read_text(self):
        """Read text from the file at this path

        >>> f = SSHPath('/etc/passwd', ssh=ssh_conn)
        >>> f.read_text()[:10]
        'root:x:0:0'
        """
        return self._s(self.read_bytes())

    def write_bytes(self, data):
        r"""Write bytes to the file at this path

        >>> f = SSHPath('somefile', ssh=ssh_conn)
        >>> f.write_bytes(b'\x00HELLO\x00')
        >>> f.read_bytes()
        b'\x00HELLO\x00'
        """
        self.ssh.write(str(self.absolute()), data)

    def write_text(self, data):
        r"""Write text to the file at this path

        >>> f = SSHPath('somefile', ssh=ssh_conn)
        >>> f.write_text("HELLO 😭")
        >>> f.read_bytes()
        b'HELLO \xf0\x9f\x98\xad'
        >>> f.read_text()
        'HELLO 😭'
        """
        data = _encode(data)
        self.write_bytes(data)

    def readlink(self):
        data = self.ssh.readlink(self.path)
        if data == b'':
            data = self.path
        return self._new(data)

    def touch(self):
        """Touch a file (i.e. make it exist)

        >>> f = SSHPath('touchme', ssh=ssh_conn)
        >>> f.exists()
        False
        >>> f.touch()
        >>> f.exists()
        True
        """
        self.ssh.write(self.path, b'')
        # self.ssh.sftp.truncate(self.path, 0)

    def mkdir(self, mode=0o777, parents=False, exist_ok=True):
        r"""Make a directory at the specified path

        >>> f = SSHPath('dirname', ssh=ssh_conn)
        >>> f.mkdir()
        >>> f.exists()
        True

        >>> f = SSHPath('dirA/dirB/dirC', ssh=ssh_conn)
        >>> f.mkdir(parents=True)
        >>> ssh_conn.run(['ls', '-la', f.absolute().path], env={'LC_ALL': 'C.UTF-8'}).recvline()
        b'total 8\n'
        """
        if exist_ok and self.is_dir():
            return

        if not parents:
            self.ssh.sftp.mkdir(self.path, mode=mode)
            return

        if not self.is_absolute():
            path = self._new(self.ssh.cwd)
        else:
            path = self._new('/')

        parts = self.path.split(self.sep)

        for part in parts:
            # Catch against common case, need to normalize path
            if part == '..':
                raise ValueError("Cannot create directory '..'")

            path = path.joinpath(part)

            # Don't create directories that already exist            
            try:
                path.mkdir(mode=mode)
            except OSError:
                raise OSError("Could not create directory %r" % path)

    def chmod(self, mode):
        """Change the permissions of a file

        >>> f = SSHPath('chmod_me', ssh=ssh_conn)
        >>> f.touch() # E
        >>> '0o%o' % f.stat().st_mode
        '0o100664'
        >>> f.chmod(0o777)
        >>> '0o%o' % f.stat().st_mode
        '0o100777'
        """
        self.ssh.sftp.chmod(self.path, mode)

    def lchmod(*a, **kw):
        raise NotImplementedError()

    def unlink(self, missing_ok=False):
        """Remove an existing file.

        TODO:

            This test fails catastrophically if the file name is unlink_me
            (note the underscore)

        Example:

            >>> f = SSHPath('unlink_me', ssh=ssh_conn)
            >>> f.exists()
            False
            >>> f.touch()
            >>> f.exists()
            True
            >>> f.unlink()
            >>> f.exists()
            False

            Note that unlink only works on files.

            >>> f.mkdir()
            >>> f.unlink()
            Traceback (most recent call last):
            ...
            ValueError: Cannot unlink SSHPath(...)): is not a file
        """
        try:
            self.ssh.sftp.remove(str(self))
        except (IOError, OSError) as e:
            if self.exists() and not self.is_file():
                raise ValueError("Cannot unlink %r: is not a file" % self)
            if not missing_ok:
                raise e

    def rmdir(self):
        """Remove an existing directory.

        Example:

            >>> f = SSHPath('rmdir_me', ssh=ssh_conn)
            >>> f.mkdir()
            >>> f.is_dir()
            True
            >>> f.rmdir()
            >>> f.exists()
            False
        """
        if not self.exists():
            return

        if not self.is_dir():
            raise ValueError("Cannot rmdir %r: not a directory" % self)

        self.ssh.sftp.rmdir(self.path)

    def link_to(self, target):
        raise NotImplementedError()

    def symlink_to(self, target):
        r"""Create a symlink at this path to the provided target

        Todo:

            Paramiko's documentation is wrong and inverted.
            https://github.com/paramiko/paramiko/issues/1821

        Example:

            >>> a = SSHPath('link_name', ssh=ssh_conn)
            >>> b = SSHPath('link_target', ssh=ssh_conn)
            >>> a.symlink_to(b)
            >>> a.write_text("Hello")
            >>> b.read_text()
            'Hello'
        """
        if isinstance(target, SSHPath):
            target = target.path

        self.ssh.sftp.symlink(target, self.path)

    def rename(self, target):
        """Rename a file to the target path

        Example:

            >>> a = SSHPath('rename_from', ssh=ssh_conn)
            >>> b = SSHPath('rename_to', ssh=ssh_conn)
            >>> a.touch()
            >>> b.exists()
            False
            >>> a.rename(b)
            >>> b.exists()
            True
        """
        if isinstance(target, SSHPath):
            target = target.path

        self.ssh.sftp.rename(self.path, target)

    def replace(self, target):
        """Replace target file with file at this path

        Example:

            >>> a = SSHPath('rename_from', ssh=ssh_conn)
            >>> a.write_text('A')
            >>> b = SSHPath('rename_to', ssh=ssh_conn)
            >>> b.write_text('B')
            >>> a.replace(b)
            >>> b.read_text()
            'A'
        """
        if isinstance(target, SSHPath):
            target = target.path

        self._new(target).unlink(missing_ok=True)
        self.rename(target)

    def exists(self):
        """Returns True if the path exists

        Example:

            >>> a = SSHPath('exists', ssh=ssh_conn)
            >>> a.exists()
            False
            >>> a.touch()
            >>> a.exists()
            True
            >>> a.unlink()
            >>> a.exists()
            False
        """
        try:
            self.stat()
            return True
        except IOError:
            return False

    def is_dir(self):
        """Returns True if the path exists and is a directory
        
        Example:

            >>> f = SSHPath('is_dir', ssh=ssh_conn)
            >>> f.is_dir()
            False
            >>> f.touch()
            >>> f.is_dir()
            False
            >>> f.unlink()
            >>> f.mkdir()
            >>> f.is_dir()
            True
        """
        if not self.exists():
            return False

        if self.stat().st_mode & 0o040000:
            return True

        return False

    def is_file(self):
        """Returns True if the path exists and is a file
        
        Example:

            >>> f = SSHPath('is_file', ssh=ssh_conn)
            >>> f.is_file()
            False
            >>> f.touch()
            >>> f.is_file()
            True
            >>> f.unlink()
            >>> f.mkdir()
            >>> f.is_file()
            False
        """
        if not self.exists():
            return False

        if self.stat().st_mode & 0o040000:
            return False

        return True

    def is_symlink(self):
        raise NotImplementedError()

    def is_block_device(self):
        raise NotImplementedError()

    def is_char_device(self):
        raise NotImplementedError()

    def is_fifo(self):
        raise NotImplementedError()

    def is_socket(self):
        raise NotImplementedError()

    def expanduser(self):
        """Expands a path that starts with a tilde

        Example:

            >>> f = SSHPath('~/my-file', ssh=ssh_conn)
            >>> f.path
            '~/my-file'
            >>> f.expanduser().path # doctest: +ELLIPSIS
            '/home/.../my-file'
        """
        if not self.path.startswith('~/'):
            return self
        
        home = self.home
        subpath = self.path.replace('~/', '')
        return home.joinpath(subpath)

#----------------------------- PWNTOOLS ADDITIONS -----------------------------
    @classmethod
    def mktemp(cls):
        temp = _decode(context.ssh_session.mktemp())
        return SSHPath(temp, ssh=context.ssh_session)

    @classmethod
    def mkdtemp(self):
        temp = _decode(context.ssh_session.mkdtemp())
        return SSHPath(temp, ssh=context.ssh_session)

__all__ = ['SSHPath']

1	# -- coding: utf-8 --
2	"""	1✔
3	Handles file abstraction for remote SSH files
4
5	Emulates pathlib as much as possible, but does so through duck typing.
6	"""
7	import os	1✔
8	import six	1✔
9	import sys	1✔
10	import tempfile	1✔
11	import time	1✔
12
13	from pwnlib.context import context	1✔
14	from pwnlib.util.misc import read, write	1✔
15	from pwnlib.util.packing import _encode, _decode	1✔
16
17	if six.PY3:	1!
18	from pathlib import *	×
19	else:
20	from pathlib2 import *	1✔
21
22	class SSHPath(PosixPath):	1✔
23	r"""Represents a file that exists on a remote filesystem.
24
25	See :class:`.ssh` for more information on how to set up an SSH connection.
26	See :py:class:`pathlib.Path` for documentation on what members and
27	properties this object has.
28
29	Arguments:
30	name(str): Name of the file
31	ssh(ssh): :class:`.ssh` object for manipulating remote files
32
33	Note:
34
35	You can avoid having to supply ``ssh=`` on every ``SSHPath`` by setting
36	:data:`.context.ssh_session`.
37	In these examples we provide ``ssh=`` for clarity.
38
39	Examples:
40
41	First, create an SSH connection to the server.
42
43	>>> ssh_conn = ssh('travis', 'example.pwnme')
44
45	Let's use a temporary directory for our tests
46
47	>>> _ = ssh_conn.set_working_directory()
48
49	Next, you can create SSHPath objects to represent the paths to files
50	on the remote system.
51
52	>>> f = SSHPath('filename', ssh=ssh_conn)
53	>>> f.touch()
54	>>> f.exists()
55	True
56	>>> f.resolve().path # doctests: +ELLIPSIS
57	'/tmp/.../filename'
58	>>> f.write_text('asdf ❤️')
59	>>> f.read_bytes()
60	b'asdf \xe2\x9d\xa4\xef\xb8\x8f'
61
62	``context.ssh_session`` must be set to use the :meth:`.SSHPath.mktemp`
63	or :meth:`.SSHPath.mkdtemp` methods.
64
65	>>> context.ssh_session = ssh_conn
66	>>> SSHPath.mktemp() # doctest: +ELLIPSIS
67	SSHPath('...', ssh=ssh(user='travis', host='127.0.0.1'))
68	"""
69
70	sep = '/'	1✔
71
72	def __init__(self, path, ssh=None):	1✔
73	self.path = self._s(path)	1✔
74	self.ssh = ssh or context.ssh_session	1✔
75
76	if self.ssh is None:	1!
77	raise ValueError('SSHPath requires an ssh session. Provide onee or set context.ssh_session.')	×
78
79	def _s(self, other):	1✔
80	# We want strings
81	if isinstance(other, str):	1✔
82	return other	1✔
83
84	# We don't want unicode
85	if isinstance(other, six.text_type):	1!
86	return str(other)	1✔
87
88	# We also don't want binary
89	if isinstance(other, six.binary_type):	×
90	return str(_decode(other))	×
91
92	def _new(self, path, a, *kw):	1✔
93	kw['ssh'] = self.ssh	1✔
94	path = self._s(path)	1✔
95	return SSHPath(path, a, *kw)	1✔
96
97	def _run(self, a, *kw):	1✔
98	with context.silent:	1✔
99	return self.ssh.run(a, *kw)	1✔
100
101	#---------------------------------- PUREPATH ----------------------------------
102	def __str__(self):	1✔
103	return self.path	1✔
104
105	def __fspath__(self):	1✔
106	return str(self)	×
107
108	def as_posix(self):	1✔
109	return self.path	1✔
110
111	def __bytes__(self):	1✔
112	return os.fsencode(self)	×
113
114	def __repr__(self):	1✔
115	return "{}({!r}, ssh={!r})".format(self.__class__.__name__, self.as_posix(), self.ssh)	1✔
116
117	def as_uri(self):	1✔
118	raise NotImplementedError()	×
119
120	def __eq__(self, other):	1✔
121	if not isinstance(other, SSHPath):	1!
122	return str(self) == str(other)	×
123
124	if self.ssh.host != other.ssh.host:	1!
125	return False	×
126
127	if self.path != other.path:	1✔
128	return False	1✔
129
130	return True	1✔
131
132	def __hash__(a, *kw): ""; raise NotImplementedError	1!
133	def __lt__(a, *kw): ""; raise NotImplementedError	1!
134	def __le__(a, *kw): ""; raise NotImplementedError	1!
135	def __gt__(a, *kw): ""; raise NotImplementedError	1!
136	def __ge__(a, *kw): ""; raise NotImplementedError	1!
137
138	@property	1✔
139	def anchor(self):
140	raise NotImplementedError()	×
141
142	@property	1✔
143	def name(self):
144	"""Returns the name of the file.
145
146	>>> f = SSHPath('hello', ssh=ssh_conn)
147	>>> f.name
148	'hello'
149	"""
150	return os.path.basename(self.path)	1✔
151
152	@property	1✔
153	def suffix(self):
154	"""Returns the suffix of the file.
155
156	>>> f = SSHPath('hello.tar.gz', ssh=ssh_conn)
157	>>> f.suffix
158	'.gz'
159	"""
160	if '.' not in self.name:	1!
161	return ''	×
162	return self.name[self.name.rindex('.'):]	1✔
163
164	@property	1✔
165	def suffixes(self):
166	"""Returns the suffixes of a file
167
168	>>> f = SSHPath('hello.tar.gz', ssh=ssh_conn)
169	>>> f.suffixes
170	'.tar.gz'
171	"""
172
173	basename = self.name	1✔
174	if '.' not in basename:	1!
175	return ''	×
176	return '.' + self.name.split('.', 1)[1]	1✔
177
178	@property	1✔
179	def stem(self):
180	"""Returns the stem of a file without any extension
181
182	>>> f = SSHPath('hello.tar.gz', ssh=ssh_conn)
183	>>> f.stem
184	'hello'
185	"""
186	if '.' not in self.name:	1!
187	return self.name	×
188	return self.name[:self.name.index('.')]	1✔
189
190	def with_name(self, name):	1✔
191	"""Return a new path with the file name changed
192
193	>>> f = SSHPath('hello/world', ssh=ssh_conn)
194	>>> f.path
195	'hello/world'
196	>>> f.with_name('asdf').path
197	'hello/asdf'
198	"""
199	if '/' not in self.path:	1!
200	return name	×
201
202	path, _ = self.path.split(self.sep, 1)	1✔
203	path = self._new(path)	1✔
204	return path.joinpath(name)	1✔
205
206	def with_stem(self, name):	1✔
207	"""Return a new path with the stem changed.
208
209	>>> f = SSHPath('hello/world.tar.gz', ssh=ssh_conn)
210	>>> f.with_stem('asdf').path
211	'hello/asdf.tar.gz'
212	"""
213	return self.with_name(name + self.suffixes)	1✔
214
215	def with_suffix(self, suffix):	1✔
216	"""Return a new path with the file suffix changed
217
218	>>> f = SSHPath('hello/world.tar.gz', ssh=ssh_conn)
219	>>> f.with_suffix('.tgz').path
220	'hello/world.tgz'
221	"""
222	return self.with_name(self.stem + suffix)	1✔
223
224	def relative_to(self, *other):	1✔
225	raise NotImplementedError()	×
226
227	def is_relative_to(self, *other):	1✔
228	raise NotImplementedError()	×
229
230	@property	1✔
231	def parts(self):
232	"""Return the individual parts of the path
233
234	>>> f = SSHPath('hello/world.tar.gz', ssh=ssh_conn)
235	>>> f.parts
236	['hello', 'world.tar.gz']
237	"""
238	return self.path.split(self.sep)	1✔
239
240	def joinpath(self, *args):	1✔
241	"""Combine this path with one or several arguments.
242
243	>>> f = SSHPath('hello', ssh=ssh_conn)
244	>>> f.joinpath('world').path
245	'hello/world'
246	"""
247	newpath = os.path.join(self.path, *args)	1✔
248	return SSHPath(newpath, ssh=self.ssh)	1✔
249
250	# __truediv__
251	# __rtruediv__
252
253	@property	1✔
254	def parent(self):
255	"""Return the parent of this path
256
257	>>> f = SSHPath('hello/world/file.txt', ssh=ssh_conn)
258	>>> f.parent.path
259	'hello/world'
260	"""
261	a, b = self.path.rsplit(self.sep, 1)	1✔
262	if a:	1!
263	return self._new(a)	1✔
264	return self	×
265
266	@property	1✔
267	def parents(self):
268	"""Return the parents of this path, as individual parts
269
270	>>> f = SSHPath('hello/world/file.txt', ssh=ssh_conn)
271	>>> list(p.path for p in f.parents)
272	['hello', 'world']
273	"""
274	if '/' not in self.path:	1!
275	return self._new('.')	×
276
277	return [self._new(p) for p in self.parent.path.split(self.sep)]	1✔
278
279	def is_absolute(self):	1✔
280	"""Returns whether a path is absolute or not.
281
282	>>> f = SSHPath('hello/world/file.txt', ssh=ssh_conn)
283	>>> f.is_absolute()
284	False
285
286	>>> f = SSHPath('/hello/world/file.txt', ssh=ssh_conn)
287	>>> f.is_absolute()
288	True
289	"""
290	return self.path.startswith(self.sep)	1✔
291
292	def is_reserved(self):	1✔
293	return False	×
294
295	def match(self, path_pattern):	1✔
296	raise NotImplementedError()	×
297
298	#------------------------------------ PATH ------------------------------------
299
300	@property	1✔
301	def cwd(self):
302	return self._new(self.ssh.cwd)	×
303
304	@property	1✔
305	def home(self):
306	"""Returns the home directory for the SSH connection
307
308	>>> f = SSHPath('...', ssh=ssh_conn)
309	>>> f.home # doctest: +ELLIPSIS
310	SSHPath('/home/...', ssh=ssh(user='...', host='127.0.0.1'))
311	"""
312	path = self._run('echo ~').recvall().rstrip()	1✔
313	return self._new(path)	1✔
314
315	def samefile(self, other_path):	1✔
316	"""Returns whether two files are the same
317
318	>>> a = SSHPath('a', ssh=ssh_conn)
319	>>> A = SSHPath('a', ssh=ssh_conn)
320	>>> x = SSHPath('x', ssh=ssh_conn)
321
322	>>> a.samefile(A)
323	True
324	>>> a.samefile(x)
325	False
326	"""
327	if not isinstance(other_path, SSHPath):	1!
328	return False	×
329
330	return self.absolute() == other_path.absolute()	1✔
331
332	def iterdir(self):	1✔
333	"""Iterates over the contents of the directory
334
335	>>> directory = SSHPath('iterdir', ssh=ssh_conn)
336	>>> directory.mkdir()
337	>>> fileA = directory.joinpath('fileA')
338	>>> fileA.touch()
339	>>> fileB = directory.joinpath('fileB')
340	>>> fileB.touch()
341	>>> dirC = directory.joinpath('dirC')
342	>>> dirC.mkdir()
343	>>> [p.name for p in directory.iterdir()]
344	['dirC', 'fileA', 'fileB']
345	"""
346	for directory in sorted(self.ssh.sftp.listdir(self.path)):	1✔
347	yield self._new(directory)	1✔
348
349	def glob(self, pattern):	1✔
350	raise NotImplementedError()	×
351
352	def rglob(self, pattern):	1✔
353	raise NotImplementedError()	×
354
355	def absolute(self):	1✔
356	"""Return the absolute path to a file, preserving e.g. "../".
357	The current working directory is determined via the :class:`.ssh`
358	member :attr:`.ssh.cwd`.
359
360	Example:
361
362	>>> f = SSHPath('absA/../absB/file', ssh=ssh_conn)
363	>>> f.absolute().path # doctest: +ELLIPSIS
364	'/.../absB/file'
365	"""
366	path = os.path.normpath(self.path)	1✔
367
368	if self.is_absolute():	1✔
369	return self._new(path)	1✔
370
371	return self._new(os.path.join(self.ssh.cwd, path))	1✔
372
373	def resolve(self, strict=False):	1✔
374	"""Return the absolute path to a file, resolving any '..' or symlinks.
375	The current working directory is determined via the :class:`.ssh`
376	member :attr:`.ssh.cwd`.
377
378	Note:
379
380	The file must exist to call resolve().
381
382	Examples:
383
384	>>> f = SSHPath('resA/resB/../resB/file', ssh=ssh_conn)
385
386	>>> f.resolve().path # doctest: +ELLIPSIS
387	Traceback (most recent call last):
388	...
389	ValueError: Could not normalize path: '/.../resA/resB/file'
390
391	>>> f.parent.absolute().mkdir(parents=True)
392	>>> list(f.parent.iterdir())
393	[]
394
395	>>> f.touch()
396	>>> f.resolve() # doctest: +ELLIPSIS
397	SSHPath('/.../resA/resB/file', ssh=ssh(user='...', host='127.0.0.1'))
398	"""
399	path = self.absolute().path	1✔
400	path = os.path.normpath(path)	1✔
401
402	if six.PY2:	1!
403	error_type = IOError	1✔
404	else:
405	error_type = FileNotFoundError	×
406
407	try:	1✔
408	return self._new(self.ssh.sftp.normalize(path))	1✔
409	except error_type as e:	1✔
410	raise ValueError("Could not normalize path: %r" % path)	1✔
411
412	def stat(self):	1✔
413	"""Returns the permissions and other information about the file
414
415	>>> f = SSHPath('filename', ssh=ssh_conn)
416	>>> f.touch()
417	>>> stat = f.stat()
418	>>> stat.st_size
419	0
420	>>> '%o' % stat.st_mode # doctest: +ELLIPSIS
421	'...664'
422	"""
423	return self.ssh.sftp.stat(self.path)	1✔
424
425	def owner(self):	1✔
426	raise NotImplementedError()	×
427
428	def group(self):	1✔
429	raise NotImplementedError()	×
430
431	def open(self, a, *kw):	1✔
432	"""Return a file-like object for this path.
433
434	This currently seems to be broken in Paramiko.
435
436	>>> f = SSHPath('filename', ssh=ssh_conn)
437	>>> f.write_text('Hello')
438	>>> fo = f.open(mode='r+')
439	>>> fo # doctest: +ELLIPSIS
440	<paramiko.sftp_file.SFTPFile object at ...>
441	>>> fo.read('asdfasdf') # doctest: +SKIP
442	b'Hello'
443	"""
444	return self.ssh.sftp.open(self.path, a, *kw)	1✔
445
446	def read_bytes(self):	1✔
447	"""Read bytes from the file at this path
448
449	>>> f = SSHPath('/etc/passwd', ssh=ssh_conn)
450	>>> f.read_bytes()[:10]
451	b'root:x:0:0'
452	"""
453	return self.ssh.read(str(self.absolute()))	1✔
454
455	def read_text(self):	1✔
456	"""Read text from the file at this path
457
458	>>> f = SSHPath('/etc/passwd', ssh=ssh_conn)
459	>>> f.read_text()[:10]
460	'root:x:0:0'
461	"""
462	return self._s(self.read_bytes())	1✔
463
464	def write_bytes(self, data):	1✔
465	r"""Write bytes to the file at this path
466
467	>>> f = SSHPath('somefile', ssh=ssh_conn)
468	>>> f.write_bytes(b'\x00HELLO\x00')
469	>>> f.read_bytes()
470	b'\x00HELLO\x00'
471	"""
472	self.ssh.write(str(self.absolute()), data)	1✔
473
474	def write_text(self, data):	1✔
475	r"""Write text to the file at this path
476
477	>>> f = SSHPath('somefile', ssh=ssh_conn)
478	>>> f.write_text("HELLO 😭")
479	>>> f.read_bytes()
480	b'HELLO \xf0\x9f\x98\xad'
481	>>> f.read_text()
482	'HELLO 😭'
483	"""
484	data = _encode(data)	1✔
485	self.write_bytes(data)	1✔
486
487	def readlink(self):	1✔
488	data = self.ssh.readlink(self.path)	×
489	if data == b'':	×
490	data = self.path	×
491	return self._new(data)	×
492
493	def touch(self):	1✔
494	"""Touch a file (i.e. make it exist)
495
496	>>> f = SSHPath('touchme', ssh=ssh_conn)
497	>>> f.exists()
498	False
499	>>> f.touch()
500	>>> f.exists()
501	True
502	"""
503	self.ssh.write(self.path, b'')	1✔
504	# self.ssh.sftp.truncate(self.path, 0)
505
506	def mkdir(self, mode=0o777, parents=False, exist_ok=True):	1✔
507	r"""Make a directory at the specified path
508
509	>>> f = SSHPath('dirname', ssh=ssh_conn)
510	>>> f.mkdir()
511	>>> f.exists()
512	True
513
514	>>> f = SSHPath('dirA/dirB/dirC', ssh=ssh_conn)
515	>>> f.mkdir(parents=True)
516	>>> ssh_conn.run(['ls', '-la', f.absolute().path], env={'LC_ALL': 'C.UTF-8'}).recvline()
517	b'total 8\n'
518	"""
519	if exist_ok and self.is_dir():	1✔
520	return	1✔
521
522	if not parents:	1✔
523	self.ssh.sftp.mkdir(self.path, mode=mode)	1✔
524	return	1✔
525
526	if not self.is_absolute():	1✔
527	path = self._new(self.ssh.cwd)	1✔
528	else:
529	path = self._new('/')	1✔
530
531	parts = self.path.split(self.sep)	1✔
532
533	for part in parts:	1✔
534	# Catch against common case, need to normalize path
535	if part == '..':	1!
536	raise ValueError("Cannot create directory '..'")	×
537
538	path = path.joinpath(part)	1✔
539
540	# Don't create directories that already exist
541	try:	1✔
542	path.mkdir(mode=mode)	1✔
543	except OSError:	×
544	raise OSError("Could not create directory %r" % path)	×
545
546	def chmod(self, mode):	1✔
547	"""Change the permissions of a file
548
549	>>> f = SSHPath('chmod_me', ssh=ssh_conn)
550	>>> f.touch() # E
551	>>> '0o%o' % f.stat().st_mode
552	'0o100664'
553	>>> f.chmod(0o777)
554	>>> '0o%o' % f.stat().st_mode
555	'0o100777'
556	"""
557	self.ssh.sftp.chmod(self.path, mode)	1✔
558
559	def lchmod(a, *kw):	1✔
560	raise NotImplementedError()	×
561
562	def unlink(self, missing_ok=False):	1✔
563	"""Remove an existing file.
564
565	TODO:
566
567	This test fails catastrophically if the file name is unlink_me
568	(note the underscore)
569
570	Example:
571
572	>>> f = SSHPath('unlink_me', ssh=ssh_conn)
573	>>> f.exists()
574	False
575	>>> f.touch()
576	>>> f.exists()
577	True
578	>>> f.unlink()
579	>>> f.exists()
580	False
581
582	Note that unlink only works on files.
583
584	>>> f.mkdir()
585	>>> f.unlink()
586	Traceback (most recent call last):
587	...
588	ValueError: Cannot unlink SSHPath(...)): is not a file
589	"""
590	try:	1✔
591	self.ssh.sftp.remove(str(self))	1✔
592	except (IOError, OSError) as e:	1✔
593	if self.exists() and not self.is_file():	1!
594	raise ValueError("Cannot unlink %r: is not a file" % self)	1✔
595	if not missing_ok:	×
596	raise e	×
597
598	def rmdir(self):	1✔
599	"""Remove an existing directory.
600
601	Example:
602
603	>>> f = SSHPath('rmdir_me', ssh=ssh_conn)
604	>>> f.mkdir()
605	>>> f.is_dir()
606	True
607	>>> f.rmdir()
608	>>> f.exists()
609	False
610	"""
611	if not self.exists():	1!
612	return	×
613
614	if not self.is_dir():	1!
615	raise ValueError("Cannot rmdir %r: not a directory" % self)	×
616
617	self.ssh.sftp.rmdir(self.path)	1✔
618
619	def link_to(self, target):	1✔
620	raise NotImplementedError()	×
621
622	def symlink_to(self, target):	1✔
623	r"""Create a symlink at this path to the provided target
624
625	Todo:
626
627	Paramiko's documentation is wrong and inverted.
628	https://github.com/paramiko/paramiko/issues/1821
629
630	Example:
631
632	>>> a = SSHPath('link_name', ssh=ssh_conn)
633	>>> b = SSHPath('link_target', ssh=ssh_conn)
634	>>> a.symlink_to(b)
635	>>> a.write_text("Hello")
636	>>> b.read_text()
637	'Hello'
638	"""
639	if isinstance(target, SSHPath):	1!
640	target = target.path	1✔
641
642	self.ssh.sftp.symlink(target, self.path)	1✔
643
644	def rename(self, target):	1✔
645	"""Rename a file to the target path
646
647	Example:
648
649	>>> a = SSHPath('rename_from', ssh=ssh_conn)
650	>>> b = SSHPath('rename_to', ssh=ssh_conn)
651	>>> a.touch()
652	>>> b.exists()
653	False
654	>>> a.rename(b)
655	>>> b.exists()
656	True
657	"""
658	if isinstance(target, SSHPath):	1✔
659	target = target.path	1✔
660
661	self.ssh.sftp.rename(self.path, target)	1✔
662
663	def replace(self, target):	1✔
664	"""Replace target file with file at this path
665
666	Example:
667
668	>>> a = SSHPath('rename_from', ssh=ssh_conn)
669	>>> a.write_text('A')
670	>>> b = SSHPath('rename_to', ssh=ssh_conn)
671	>>> b.write_text('B')
672	>>> a.replace(b)
673	>>> b.read_text()
674	'A'
675	"""
676	if isinstance(target, SSHPath):	1!
677	target = target.path	1✔
678
679	self._new(target).unlink(missing_ok=True)	1✔
680	self.rename(target)	1✔
681
682	def exists(self):	1✔
683	"""Returns True if the path exists
684
685	Example:
686
687	>>> a = SSHPath('exists', ssh=ssh_conn)
688	>>> a.exists()
689	False
690	>>> a.touch()
691	>>> a.exists()
692	True
693	>>> a.unlink()
694	>>> a.exists()
695	False
696	"""
697	try:	1✔
698	self.stat()	1✔
699	return True	1✔
700	except IOError:	1✔
701	return False	1✔
702
703	def is_dir(self):	1✔
704	"""Returns True if the path exists and is a directory
705
706	Example:
707
708	>>> f = SSHPath('is_dir', ssh=ssh_conn)
709	>>> f.is_dir()
710	False
711	>>> f.touch()
712	>>> f.is_dir()
713	False
714	>>> f.unlink()
715	>>> f.mkdir()
716	>>> f.is_dir()
717	True
718	"""
719	if not self.exists():	1✔
720	return False	1✔
721
722	if self.stat().st_mode & 0o040000:	1✔
723	return True	1✔
724
725	return False	1✔
726
727	def is_file(self):	1✔
728	"""Returns True if the path exists and is a file
729
730	Example:
731
732	>>> f = SSHPath('is_file', ssh=ssh_conn)
733	>>> f.is_file()
734	False
735	>>> f.touch()
736	>>> f.is_file()
737	True
738	>>> f.unlink()
739	>>> f.mkdir()
740	>>> f.is_file()
741	False
742	"""
743	if not self.exists():	1✔
744	return False	1✔
745
746	if self.stat().st_mode & 0o040000:	1✔
747	return False	1✔
748
749	return True	1✔
750
751	def is_symlink(self):	1✔
752	raise NotImplementedError()	×
753
754	def is_block_device(self):	1✔
755	raise NotImplementedError()	×
756
757	def is_char_device(self):	1✔
758	raise NotImplementedError()	×
759
760	def is_fifo(self):	1✔
761	raise NotImplementedError()	×
762
763	def is_socket(self):	1✔
764	raise NotImplementedError()	×
765
766	def expanduser(self):	1✔
767	"""Expands a path that starts with a tilde
768
769	Example:
770
771	>>> f = SSHPath('~/my-file', ssh=ssh_conn)
772	>>> f.path
773	'~/my-file'
774	>>> f.expanduser().path # doctest: +ELLIPSIS
775	'/home/.../my-file'
776	"""
777	if not self.path.startswith('~/'):	1!
778	return self	×
779
780	home = self.home	1✔
781	subpath = self.path.replace('~/', '')	1✔
782	return home.joinpath(subpath)	1✔
783
784	#----------------------------- PWNTOOLS ADDITIONS -----------------------------
785	@classmethod	1✔
786	def mktemp(cls):
787	temp = _decode(context.ssh_session.mktemp())	1✔
788	return SSHPath(temp, ssh=context.ssh_session)	1✔
789
790	@classmethod	1✔
791	def mkdtemp(self):
792	temp = _decode(context.ssh_session.mkdtemp())	×
793	return SSHPath(temp, ssh=context.ssh_session)	×
794
795	__all__ = ['SSHPath']	1✔

Gallopsled / pwntools / 8912ca5a8c3a9725c3ba6d30561607150a6faebe-PR-2205

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous