13600950642

Committed 01 Mar 2025 04:10AM UTC coverage: 74.211% (+3.2%) from 71.055%

Build # 13600950642

Build Type

Pull #2546

github

Committed by

web-flow

Commit Message

Merge 77df40314 into 60cff2437

Pull Request Pull Request #2546: ssh: Allow passing `disabled_algorithms` keyword argument from `ssh` to paramiko

Run Details

3812 of 6380 branches covered (59.75%)

0 of 1 new or added line in 1 file covered. (0.0%)

1243 existing lines in 37 files now uncovered.

13352 of 17992 relevant lines covered (74.21%)

0.74 hits per line

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

82.61

/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 sys
import tempfile
import time

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

from pathlib 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 binary
        return _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)

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

Gallopsled / pwntools / 13600950642

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