5573614195

pending completion

Build # 5573614195

Build Type

push

github-actions

Committed by

peace-maker

Commit Message

Fix non-pie tests

Run Details

3936 of 6602 branches covered (59.62%)

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

12076 of 16875 relevant lines covered (71.56%)

0.72 hits per line

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

58.7

/pwnlib/util/misc.py

from __future__ import division

import base64
import errno
import os
import re
import signal
import six
import socket
import stat
import string
import subprocess
import sys
import tempfile

from pwnlib import atexit
from pwnlib.context import context
from pwnlib.log import getLogger
from pwnlib.util import fiddling
from pwnlib.util import lists
from pwnlib.util import packing

log = getLogger(__name__)

def align(alignment, x):
    """align(alignment, x) -> int

    Rounds `x` up to nearest multiple of the `alignment`.

    Example:
      >>> [align(5, n) for n in range(15)]
      [0, 5, 5, 5, 5, 5, 10, 10, 10, 10, 10, 15, 15, 15, 15]
    """
    return x + -x % alignment


def align_down(alignment, x):
    """align_down(alignment, x) -> int

    Rounds `x` down to nearest multiple of the `alignment`.

    Example:
        >>> [align_down(5, n) for n in range(15)]
        [0, 0, 0, 0, 0, 5, 5, 5, 5, 5, 10, 10, 10, 10, 10]
    """
    return x - x % alignment


def binary_ip(host):
    """binary_ip(host) -> str

    Resolve host and return IP as four byte string.

    Example:
        >>> binary_ip("127.0.0.1")
        b'\\x7f\\x00\\x00\\x01'
    """
    return socket.inet_aton(socket.gethostbyname(host))


def size(n, abbrev = 'B', si = False):
    """size(n, abbrev = 'B', si = False) -> str

    Convert the length of a bytestream to human readable form.

    Arguments:
      n(int,iterable): The length to convert to human readable form,
        or an object which can have ``len()`` called on it.
      abbrev(str): String appended to the size, defaults to ``'B'``.

    Example:
        >>> size(451)
        '451B'
        >>> size(1000)
        '1000B'
        >>> size(1024)
        '1.00KB'
        >>> size(1024, ' bytes')
        '1.00K bytes'
        >>> size(1024, si = True)
        '1.02KB'
        >>> [size(1024 ** n) for n in range(7)]
        ['1B', '1.00KB', '1.00MB', '1.00GB', '1.00TB', '1.00PB', '1024.00PB']
        >>> size([])
        '0B'
        >>> size([1,2,3])
        '3B'
    """
    if hasattr(n, '__len__'):
        n = len(n)

    base = 1000.0 if si else 1024.0
    if n < base:
        return '%d%s' % (n, abbrev)

    for suffix in ['K', 'M', 'G', 'T']:
        n /= base
        if n < base:
            return '%.02f%s%s' % (n, suffix, abbrev)

    return '%.02fP%s' % (n / base, abbrev)

KB = 1000
MB = 1000 * KB
GB = 1000 * MB

KiB = 1024
MiB = 1024 * KiB
GiB = 1024 * MiB

def read(path, count=-1, skip=0):
    r"""read(path, count=-1, skip=0) -> str

    Open file, return content.

    Examples:
        >>> read('/proc/self/exe')[:4]
        b'\x7fELF'
    """
    path = os.path.expanduser(os.path.expandvars(path))
    with open(path, 'rb') as fd:
        if skip:
            fd.seek(skip)
        return fd.read(count)


def write(path, data = b'', create_dir = False, mode = 'w'):
    """Create new file or truncate existing to zero length and write data."""
    path = os.path.expanduser(os.path.expandvars(path))
    if create_dir:
        path = os.path.realpath(path)
        mkdir_p(os.path.dirname(path))
    if mode == 'w' and isinstance(data, bytes): mode += 'b'
    with open(path, mode) as f:
        f.write(data)

def which(name, all = False, path=None):
    """which(name, flags = os.X_OK, all = False) -> str or str set

    Works as the system command ``which``; searches $PATH for ``name`` and
    returns a full path if found.

    If `all` is :const:`True` the set of all found locations is returned, else
    the first occurrence or :const:`None` is returned.

    Arguments:
      `name` (str): The file to search for.
      `all` (bool):  Whether to return all locations where `name` was found.

    Returns:
      If `all` is :const:`True` the set of all locations where `name` was found,
      else the first location or :const:`None` if not found.

    Example:

        >>> which('sh') # doctest: +ELLIPSIS
        '.../bin/sh'
    """
    # If name is a path, do not attempt to resolve it.
    if os.path.sep in name:
        return name

    isroot = os.getuid() == 0
    out = set()
    try:
        path = path or os.environ['PATH']
    except KeyError:
        log.exception('Environment variable $PATH is not set')
    for p in path.split(os.pathsep):
        p = os.path.join(p, name)
        if os.access(p, os.X_OK):
            st = os.stat(p)
            if not stat.S_ISREG(st.st_mode):
                continue
            # work around this issue: https://bugs.python.org/issue9311
            if isroot and not \
              st.st_mode & (stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH):
                continue
            if all:
                out.add(p)
            else:
                return p
    if all:
        return out
    else:
        return None


def normalize_argv_env(argv, env, log, level=2):
    #
    # Validate argv
    #
    # - Must be a list/tuple of strings
    # - Each string must not contain '\x00'
    #
    argv = argv or []
    if isinstance(argv, (six.text_type, six.binary_type)):
        argv = [argv]

    if not isinstance(argv, (list, tuple)):
        log.error('argv must be a list or tuple: %r' % argv)

    if not all(isinstance(arg, (six.text_type, bytes, bytearray)) for arg in argv):
        log.error("argv must be strings or bytes: %r" % argv)

    # Create a duplicate so we can modify it
    argv = list(argv)

    for i, oarg in enumerate(argv):
        arg = packing._need_bytes(oarg, level, 0x80)  # ASCII text is okay
        if b'\x00' in arg[:-1]:
            log.error('Inappropriate nulls in argv[%i]: %r' % (i, oarg))
        argv[i] = bytearray(arg.rstrip(b'\x00'))

    #
    # Validate environment
    #
    # - Must be a dictionary of {string:string}
    # - No strings may contain '\x00'
    #

    # Create a duplicate so we can modify it safely
    env2 = []
    if hasattr(env, 'items'):
        env_items = env.items()
    else:
        env_items = env
    if env:
        for k,v in env_items:
            if not isinstance(k, (bytes, six.text_type)):
                log.error('Environment keys must be strings: %r' % k)
            if not isinstance(v, (bytes, six.text_type)):
                log.error('Environment values must be strings: %r=%r' % (k,v))
            k = packing._need_bytes(k, level, 0x80)  # ASCII text is okay
            v = packing._need_bytes(v, level, 0x80)  # ASCII text is okay
            if b'\x00' in k[:-1]:
                log.error('Inappropriate nulls in env key: %r' % (k))
            if b'\x00' in v[:-1]:
                log.error('Inappropriate nulls in env value: %r=%r' % (k, v))
            env2.append((bytearray(k.rstrip(b'\x00')), bytearray(v.rstrip(b'\x00'))))

    return argv, env2 or env


def run_in_new_terminal(command, terminal=None, args=None, kill_at_exit=True, preexec_fn=None):
    """run_in_new_terminal(command, terminal=None, args=None, kill_at_exit=True, preexec_fn=None) -> int

    Run a command in a new terminal.

    When ``terminal`` is not set:
        - If ``context.terminal`` is set it will be used.
          If it is an iterable then ``context.terminal[1:]`` are default arguments.
        - If a ``pwntools-terminal`` command exists in ``$PATH``, it is used
        - If tmux is detected (by the presence of the ``$TMUX`` environment
          variable), a new pane will be opened.
        - If GNU Screen is detected (by the presence of the ``$STY`` environment
          variable), a new screen will be opened.
        - If ``$TERM_PROGRAM`` is set, that is used.
        - If X11 is detected (by the presence of the ``$DISPLAY`` environment
          variable), ``x-terminal-emulator`` is used.
        - If KDE Konsole is detected (by the presence of the ``$KONSOLE_VERSION``
          environment variable), a terminal will be split.
        - If WSL (Windows Subsystem for Linux) is detected (by the presence of
          a ``wsl.exe`` binary in the ``$PATH`` and ``/proc/sys/kernel/osrelease``
          containing ``Microsoft``), a new ``cmd.exe`` window will be opened.

    If `kill_at_exit` is :const:`True`, try to close the command/terminal when the
    current process exits. This may not work for all terminal types.

    Arguments:
        command (str): The command to run.
        terminal (str): Which terminal to use.
        args (list): Arguments to pass to the terminal
        kill_at_exit (bool): Whether to close the command/terminal on process exit.
        preexec_fn (callable): Callable to invoke before exec().

    Note:
        The command is opened with ``/dev/null`` for stdin, stdout, stderr.

    Returns:
      PID of the new terminal process
    """
    if not terminal:
        if context.terminal:
            terminal = context.terminal[0]
            args     = context.terminal[1:]
        elif which('pwntools-terminal'):
            terminal = 'pwntools-terminal'
            args     = []
        elif 'TMUX' in os.environ and which('tmux'):
            terminal = 'tmux'
            args     = ['splitw']
        elif 'STY' in os.environ and which('screen'):
            terminal = 'screen'
            args     = ['-t','pwntools-gdb','bash','-c']
        elif 'TERM_PROGRAM' in os.environ:
            terminal = os.environ['TERM_PROGRAM']
            args     = []
        elif 'DISPLAY' in os.environ and which('x-terminal-emulator'):
            terminal = 'x-terminal-emulator'
            args     = ['-e']
        elif 'KONSOLE_VERSION' in os.environ and which('qdbus'):
            qdbus = which('qdbus')
            window_id = os.environ['WINDOWID']
            konsole_dbus_service = os.environ['KONSOLE_DBUS_SERVICE']

            with subprocess.Popen((qdbus, konsole_dbus_service), stdout=subprocess.PIPE) as proc:
                lines = proc.communicate()[0].decode().split('\n')

            # Iterate over all MainWindows
            for line in lines:
                parts = line.split('/')
                if len(parts) == 3 and parts[2].startswith('MainWindow_'):
                    name = parts[2]
                    with subprocess.Popen((qdbus, konsole_dbus_service, '/konsole/' + name,
                                           'org.kde.KMainWindow.winId'), stdout=subprocess.PIPE) as proc:
                        target_window_id = proc.communicate()[0].decode().strip()
                        if target_window_id == window_id:
                            break
            else:
                log.error('MainWindow not found')

            # Split
            subprocess.run((qdbus, konsole_dbus_service, '/konsole/' + name,
                            'org.kde.KMainWindow.activateAction', 'split-view-left-right'), stdout=subprocess.DEVNULL)

            # Find new session
            with subprocess.Popen((qdbus, konsole_dbus_service, os.environ['KONSOLE_DBUS_WINDOW'],
                                   'org.kde.konsole.Window.sessionList'), stdout=subprocess.PIPE) as proc:
                session_list = map(int, proc.communicate()[0].decode().split())
            last_konsole_session = max(session_list)

            terminal = 'qdbus'
            args = [konsole_dbus_service, '/Sessions/{}'.format(last_konsole_session),
                    'org.kde.konsole.Session.runCommand']

        else:
            is_wsl = False
            if os.path.exists('/proc/sys/kernel/osrelease'):
                with open('/proc/sys/kernel/osrelease', 'rb') as f:
                    is_wsl = b'icrosoft' in f.read()
            if is_wsl and which('cmd.exe') and which('wsl.exe') and which('bash.exe'):
                terminal    = 'cmd.exe'
                args        = ['/c', 'start']
                distro_name = os.getenv('WSL_DISTRO_NAME')

                # Split pane in Windows Terminal
                if 'WT_SESSION' in os.environ and which('wt.exe'):
                    args.extend(['wt.exe', '-w', '0', 'split-pane', '-d', '.'])

                if distro_name:
                    args.extend(['wsl.exe', '-d', distro_name, 'bash', '-c'])
                else:
                    args.extend(['bash.exe', '-c'])
                

    if not terminal:
        log.error('Could not find a terminal binary to use. Set context.terminal to your terminal.')
    elif not which(terminal):
        log.error('Could not find terminal binary %r. Set context.terminal to your terminal.' % terminal)

    if isinstance(args, tuple):
        args = list(args)

    # When not specifying context.terminal explicitly, we used to set these flags above.
    # However, if specifying terminal=['tmux', 'splitw', '-h'], we would be lacking these flags.
    # Instead, set them here and hope for the best.
    if terminal == 'tmux':
        args += ['-F' '#{pane_pid}', '-P']

    argv = [which(terminal)] + args

    if isinstance(command, six.string_types):
        if ';' in command:
            log.error("Cannot use commands with semicolon.  Create a script and invoke that directly.")
        argv += [command]
    elif isinstance(command, (list, tuple)):
        # Dump the full command line to a temporary file so we can be sure that
        # it is parsed correctly, and we do not need to account for shell expansion
        script = '''
#!{executable!s}
import os
os.execve({argv0!r}, {argv!r}, os.environ)
'''
        script = script.format(executable=sys.executable,
                               argv=command,
                               argv0=which(command[0]))
        script = script.lstrip()

        log.debug("Created script for new terminal:\n%s" % script)

        with tempfile.NamedTemporaryFile(delete=False, mode='wt+') as tmp:
          tmp.write(script)
          tmp.flush()
          os.chmod(tmp.name, 0o700)
          argv += [tmp.name]


    log.debug("Launching a new terminal: %r" % argv)

    stdin = stdout = stderr = open(os.devnull, 'r+b')
    if terminal == 'tmux':
        stdout = subprocess.PIPE

    p = subprocess.Popen(argv, stdin=stdin, stdout=stdout, stderr=stderr, preexec_fn=preexec_fn)

    if terminal == 'tmux':
        out, _ = p.communicate()
        pid = int(out)
    elif terminal == 'qdbus':
        with subprocess.Popen((qdbus, konsole_dbus_service, '/Sessions/{}'.format(last_konsole_session),
                               'org.kde.konsole.Session.processId'), stdout=subprocess.PIPE) as proc:
            pid = int(proc.communicate()[0].decode())
    else:
        pid = p.pid

    if kill_at_exit:
        def kill():
            try:
                if terminal == 'qdbus':
                    os.kill(pid, signal.SIGHUP)
                else:
                    os.kill(pid, signal.SIGTERM)
            except OSError:
                pass

        atexit.register(kill)

    return pid

def parse_ldd_output(output):
    """Parses the output from a run of 'ldd' on a binary.
    Returns a dictionary of {path: address} for
    each library required by the specified binary.

    Arguments:
      output(str): The output to parse

    Example:
        >>> sorted(parse_ldd_output('''
        ...     linux-vdso.so.1 =>  (0x00007fffbf5fe000)
        ...     libtinfo.so.5 => /lib/x86_64-linux-gnu/libtinfo.so.5 (0x00007fe28117f000)
        ...     libdl.so.2 => /lib/x86_64-linux-gnu/libdl.so.2 (0x00007fe280f7b000)
        ...     libc.so.6 => /lib/x86_64-linux-gnu/libc.so.6 (0x00007fe280bb4000)
        ...     /lib64/ld-linux-x86-64.so.2 (0x00007fe2813dd000)
        ... ''').keys())
        ['/lib/x86_64-linux-gnu/libc.so.6', '/lib/x86_64-linux-gnu/libdl.so.2', '/lib/x86_64-linux-gnu/libtinfo.so.5', '/lib64/ld-linux-x86-64.so.2']
    """
    expr_linux   = re.compile(r'\s(?P<lib>\S?/\S+)\s+\((?P<addr>0x.+)\)')
    expr_openbsd = re.compile(r'^\s+(?P<addr>[0-9a-f]+)\s+[0-9a-f]+\s+\S+\s+[01]\s+[0-9]+\s+[0-9]+\s+(?P<lib>\S+)$')
    libs = {}

    for s in output.split('\n'):
        match = expr_linux.search(s) or expr_openbsd.search(s)
        if not match:
            continue
        lib, addr = match.group('lib'), match.group('addr')
        libs[lib] = int(addr, 16)

    return libs

def mkdir_p(path):
    """Emulates the behavior of ``mkdir -p``."""

    try:
        os.makedirs(path)
    except OSError as exc:
        if exc.errno == errno.EEXIST and os.path.isdir(path):
            pass
        else:
            raise

def dealarm_shell(tube):
    """Given a tube which is a shell, dealarm it.
    """
    tube.clean()

    tube.sendline('which python || echo')
    if tube.recvline().startswith('/'):
        tube.sendline('''exec python -c "import signal, os; signal.alarm(0); os.execl('$SHELL','')"''')
        return tube

    tube.sendline('which perl || echo')
    if tube.recvline().startswith('/'):
        tube.sendline('''exec perl -e "alarm 0; exec '${SHELL:-/bin/sh}'"''')
        return tube

    return None

def register_sizes(regs, in_sizes):
    """Create dictionaries over register sizes and relations

    Given a list of lists of overlapping register names (e.g. ['eax','ax','al','ah']) and a list of input sizes,
    it returns the following:

    * all_regs    : list of all valid registers
    * sizes[reg]  : the size of reg in bits
    * bigger[reg] : list of overlapping registers bigger than reg
    * smaller[reg]: list of overlapping registers smaller than reg

    Used in i386/AMD64 shellcode, e.g. the mov-shellcode.

    Example:
        >>> regs = [['eax', 'ax', 'al', 'ah'],['ebx', 'bx', 'bl', 'bh'],
        ... ['ecx', 'cx', 'cl', 'ch'],
        ... ['edx', 'dx', 'dl', 'dh'],
        ... ['edi', 'di'],
        ... ['esi', 'si'],
        ... ['ebp', 'bp'],
        ... ['esp', 'sp'],
        ... ]
        >>> all_regs, sizes, bigger, smaller = register_sizes(regs, [32, 16, 8, 8])
        >>> all_regs
        ['eax', 'ax', 'al', 'ah', 'ebx', 'bx', 'bl', 'bh', 'ecx', 'cx', 'cl', 'ch', 'edx', 'dx', 'dl', 'dh', 'edi', 'di', 'esi', 'si', 'ebp', 'bp', 'esp', 'sp']
        >>> pprint(sizes)
        {'ah': 8,
         'al': 8,
         'ax': 16,
         'bh': 8,
         'bl': 8,
         'bp': 16,
         'bx': 16,
         'ch': 8,
         'cl': 8,
         'cx': 16,
         'dh': 8,
         'di': 16,
         'dl': 8,
         'dx': 16,
         'eax': 32,
         'ebp': 32,
         'ebx': 32,
         'ecx': 32,
         'edi': 32,
         'edx': 32,
         'esi': 32,
         'esp': 32,
         'si': 16,
         'sp': 16}
        >>> pprint(bigger)
        {'ah': ['eax', 'ax', 'ah'],
         'al': ['eax', 'ax', 'al'],
         'ax': ['eax', 'ax'],
         'bh': ['ebx', 'bx', 'bh'],
         'bl': ['ebx', 'bx', 'bl'],
         'bp': ['ebp', 'bp'],
         'bx': ['ebx', 'bx'],
         'ch': ['ecx', 'cx', 'ch'],
         'cl': ['ecx', 'cx', 'cl'],
         'cx': ['ecx', 'cx'],
         'dh': ['edx', 'dx', 'dh'],
         'di': ['edi', 'di'],
         'dl': ['edx', 'dx', 'dl'],
         'dx': ['edx', 'dx'],
         'eax': ['eax'],
         'ebp': ['ebp'],
         'ebx': ['ebx'],
         'ecx': ['ecx'],
         'edi': ['edi'],
         'edx': ['edx'],
         'esi': ['esi'],
         'esp': ['esp'],
         'si': ['esi', 'si'],
         'sp': ['esp', 'sp']}
        >>> pprint(smaller)
        {'ah': [],
         'al': [],
         'ax': ['al', 'ah'],
         'bh': [],
         'bl': [],
         'bp': [],
         'bx': ['bl', 'bh'],
         'ch': [],
         'cl': [],
         'cx': ['cl', 'ch'],
         'dh': [],
         'di': [],
         'dl': [],
         'dx': ['dl', 'dh'],
         'eax': ['ax', 'al', 'ah'],
         'ebp': ['bp'],
         'ebx': ['bx', 'bl', 'bh'],
         'ecx': ['cx', 'cl', 'ch'],
         'edi': ['di'],
         'edx': ['dx', 'dl', 'dh'],
         'esi': ['si'],
         'esp': ['sp'],
         'si': [],
         'sp': []}
    """
    sizes = {}
    bigger = {}
    smaller = {}

    for l in regs:
        for r, s in zip(l, in_sizes):
            sizes[r] = s

        for r in l:
            bigger[r] = [r_ for r_ in l if sizes[r_] > sizes[r] or r == r_]
            smaller[r] = [r_ for r_ in l if sizes[r_] < sizes[r]]

    return lists.concat(regs), sizes, bigger, smaller


def python_2_bytes_compatible(klass):
    """
    A class decorator that defines __str__ methods under Python 2.
    Under Python 3 it does nothing.
    """
    if six.PY2:
        if '__str__' not in klass.__dict__:
            klass.__str__ = klass.__bytes__
    return klass

1	from __future__ import division	1✔
2
3	import base64	1✔
4	import errno	1✔
5	import os	1✔
6	import re	1✔
7	import signal	1✔
8	import six	1✔
9	import socket	1✔
10	import stat	1✔
11	import string	1✔
12	import subprocess	1✔
13	import sys	1✔
14	import tempfile	1✔
15
16	from pwnlib import atexit	1✔
17	from pwnlib.context import context	1✔
18	from pwnlib.log import getLogger	1✔
19	from pwnlib.util import fiddling	1✔
20	from pwnlib.util import lists	1✔
21	from pwnlib.util import packing	1✔
22
23	log = getLogger(__name__)	1✔
24
25	def align(alignment, x):	1✔
26	"""align(alignment, x) -> int
27
28	Rounds `x` up to nearest multiple of the `alignment`.
29
30	Example:
31	>>> [align(5, n) for n in range(15)]
32	[0, 5, 5, 5, 5, 5, 10, 10, 10, 10, 10, 15, 15, 15, 15]
33	"""
34	return x + -x % alignment	1✔
35
36
37	def align_down(alignment, x):	1✔
38	"""align_down(alignment, x) -> int
39
40	Rounds `x` down to nearest multiple of the `alignment`.
41
42	Example:
43	>>> [align_down(5, n) for n in range(15)]
44	[0, 0, 0, 0, 0, 5, 5, 5, 5, 5, 10, 10, 10, 10, 10]
45	"""
46	return x - x % alignment	1✔
47
48
49	def binary_ip(host):	1✔
50	"""binary_ip(host) -> str
51
52	Resolve host and return IP as four byte string.
53
54	Example:
55	>>> binary_ip("127.0.0.1")
56	b'\\x7f\\x00\\x00\\x01'
57	"""
58	return socket.inet_aton(socket.gethostbyname(host))	1✔
59
60
61	def size(n, abbrev = 'B', si = False):	1✔
62	"""size(n, abbrev = 'B', si = False) -> str
63
64	Convert the length of a bytestream to human readable form.
65
66	Arguments:
67	n(int,iterable): The length to convert to human readable form,
68	or an object which can have ``len()`` called on it.
69	abbrev(str): String appended to the size, defaults to ``'B'``.
70
71	Example:
72	>>> size(451)
73	'451B'
74	>>> size(1000)
75	'1000B'
76	>>> size(1024)
77	'1.00KB'
78	>>> size(1024, ' bytes')
79	'1.00K bytes'
80	>>> size(1024, si = True)
81	'1.02KB'
82	>>> [size(1024 ** n) for n in range(7)]
83	['1B', '1.00KB', '1.00MB', '1.00GB', '1.00TB', '1.00PB', '1024.00PB']
84	>>> size([])
85	'0B'
86	>>> size([1,2,3])
87	'3B'
88	"""
89	if hasattr(n, '__len__'):	1✔
90	n = len(n)	1✔
91
92	base = 1000.0 if si else 1024.0	1✔
93	if n < base:	1✔
94	return '%d%s' % (n, abbrev)	1✔
95
96	for suffix in ['K', 'M', 'G', 'T']:	1✔
97	n /= base	1✔
98	if n < base:	1✔
99	return '%.02f%s%s' % (n, suffix, abbrev)	1✔
100
101	return '%.02fP%s' % (n / base, abbrev)	1✔
102
103	KB = 1000	1✔
104	MB = 1000 * KB	1✔
105	GB = 1000 * MB	1✔
106
107	KiB = 1024	1✔
108	MiB = 1024 * KiB	1✔
109	GiB = 1024 * MiB	1✔
110
111	def read(path, count=-1, skip=0):	1✔
112	r"""read(path, count=-1, skip=0) -> str
113
114	Open file, return content.
115
116	Examples:
117	>>> read('/proc/self/exe')[:4]
118	b'\x7fELF'
119	"""
120	path = os.path.expanduser(os.path.expandvars(path))	1✔
121	with open(path, 'rb') as fd:	1✔
122	if skip:	1!
123	fd.seek(skip)	×
124	return fd.read(count)	1✔
125
126
127	def write(path, data = b'', create_dir = False, mode = 'w'):	1✔
128	"""Create new file or truncate existing to zero length and write data."""
129	path = os.path.expanduser(os.path.expandvars(path))	1✔
130	if create_dir:	1!
131	path = os.path.realpath(path)	×
132	mkdir_p(os.path.dirname(path))	×
133	if mode == 'w' and isinstance(data, bytes): mode += 'b'	1✔
134	with open(path, mode) as f:	1✔
135	f.write(data)	1✔
136
137	def which(name, all = False, path=None):	1✔
138	"""which(name, flags = os.X_OK, all = False) -> str or str set
139
140	Works as the system command ``which``; searches $PATH for ``name`` and
141	returns a full path if found.
142
143	If `all` is :const:`True` the set of all found locations is returned, else
144	the first occurrence or :const:`None` is returned.
145
146	Arguments:
147	`name` (str): The file to search for.
148	`all` (bool): Whether to return all locations where `name` was found.
149
150	Returns:
151	If `all` is :const:`True` the set of all locations where `name` was found,
152	else the first location or :const:`None` if not found.
153
154	Example:
155
156	>>> which('sh') # doctest: +ELLIPSIS
157	'.../bin/sh'
158	"""
159	# If name is a path, do not attempt to resolve it.
160	if os.path.sep in name:	1✔
161	return name	1✔
162
163	isroot = os.getuid() == 0	1✔
164	out = set()	1✔
165	try:	1✔
166	path = path or os.environ['PATH']	1✔
167	except KeyError:	×
168	log.exception('Environment variable $PATH is not set')	×
169	for p in path.split(os.pathsep):	1✔
170	p = os.path.join(p, name)	1✔
171	if os.access(p, os.X_OK):	1✔
172	st = os.stat(p)	1✔
173	if not stat.S_ISREG(st.st_mode):	1!
174	continue	×
175	# work around this issue: https://bugs.python.org/issue9311
176	if isroot and not \	1!
177	st.st_mode & (stat.S_IXUSR \| stat.S_IXGRP \| stat.S_IXOTH):
178	continue	×
179	if all:	1!
180	out.add(p)	×
181	else:
182	return p	1✔
183	if all:	1!
184	return out	×
185	else:
186	return None	1✔
187
188
189	def normalize_argv_env(argv, env, log, level=2):	1✔
190	#
191	# Validate argv
192	#
193	# - Must be a list/tuple of strings
194	# - Each string must not contain '\x00'
195	#
196	argv = argv or []	1✔
197	if isinstance(argv, (six.text_type, six.binary_type)):	1✔
198	argv = [argv]	1✔
199
200	if not isinstance(argv, (list, tuple)):	1!
201	log.error('argv must be a list or tuple: %r' % argv)	×
202
203	if not all(isinstance(arg, (six.text_type, bytes, bytearray)) for arg in argv):	1!
204	log.error("argv must be strings or bytes: %r" % argv)	×
205
206	# Create a duplicate so we can modify it
207	argv = list(argv)	1✔
208
209	for i, oarg in enumerate(argv):	1✔
210	arg = packing._need_bytes(oarg, level, 0x80) # ASCII text is okay	1✔
211	if b'\x00' in arg[:-1]:	1!
212	log.error('Inappropriate nulls in argv[%i]: %r' % (i, oarg))	×
213	argv[i] = bytearray(arg.rstrip(b'\x00'))	1✔
214
215	#
216	# Validate environment
217	#
218	# - Must be a dictionary of {string:string}
219	# - No strings may contain '\x00'
220	#
221
222	# Create a duplicate so we can modify it safely
223	env2 = []	1✔
224	if hasattr(env, 'items'):	1✔
225	env_items = env.items()	1✔
226	else:
227	env_items = env	1✔
228	if env:	1✔
229	for k,v in env_items:	1✔
230	if not isinstance(k, (bytes, six.text_type)):	1!
231	log.error('Environment keys must be strings: %r' % k)	×
232	if not isinstance(v, (bytes, six.text_type)):	1!
233	log.error('Environment values must be strings: %r=%r' % (k,v))	×
234	k = packing._need_bytes(k, level, 0x80) # ASCII text is okay	1✔
235	v = packing._need_bytes(v, level, 0x80) # ASCII text is okay	1✔
236	if b'\x00' in k[:-1]:	1!
237	log.error('Inappropriate nulls in env key: %r' % (k))	×
238	if b'\x00' in v[:-1]:	1!
239	log.error('Inappropriate nulls in env value: %r=%r' % (k, v))	×
240	env2.append((bytearray(k.rstrip(b'\x00')), bytearray(v.rstrip(b'\x00'))))	1✔
241
242	return argv, env2 or env	1✔
243
244
245	def run_in_new_terminal(command, terminal=None, args=None, kill_at_exit=True, preexec_fn=None):	1✔
246	"""run_in_new_terminal(command, terminal=None, args=None, kill_at_exit=True, preexec_fn=None) -> int
247
248	Run a command in a new terminal.
249
250	When ``terminal`` is not set:
251	- If ``context.terminal`` is set it will be used.
252	If it is an iterable then ``context.terminal[1:]`` are default arguments.
253	- If a ``pwntools-terminal`` command exists in ``$PATH``, it is used
254	- If tmux is detected (by the presence of the ``$TMUX`` environment
255	variable), a new pane will be opened.
256	- If GNU Screen is detected (by the presence of the ``$STY`` environment
257	variable), a new screen will be opened.
258	- If ``$TERM_PROGRAM`` is set, that is used.
259	- If X11 is detected (by the presence of the ``$DISPLAY`` environment
260	variable), ``x-terminal-emulator`` is used.
261	- If KDE Konsole is detected (by the presence of the ``$KONSOLE_VERSION``
262	environment variable), a terminal will be split.
263	- If WSL (Windows Subsystem for Linux) is detected (by the presence of
264	a ``wsl.exe`` binary in the ``$PATH`` and ``/proc/sys/kernel/osrelease``
265	containing ``Microsoft``), a new ``cmd.exe`` window will be opened.
266
267	If `kill_at_exit` is :const:`True`, try to close the command/terminal when the
268	current process exits. This may not work for all terminal types.
269
270	Arguments:
271	command (str): The command to run.
272	terminal (str): Which terminal to use.
273	args (list): Arguments to pass to the terminal
274	kill_at_exit (bool): Whether to close the command/terminal on process exit.
275	preexec_fn (callable): Callable to invoke before exec().
276
277	Note:
278	The command is opened with ``/dev/null`` for stdin, stdout, stderr.
279
280	Returns:
281	PID of the new terminal process
282	"""
283	if not terminal:	1!
284	if context.terminal:	1!
285	terminal = context.terminal[0]	1✔
286	args = context.terminal[1:]	1✔
287	elif which('pwntools-terminal'):	×
288	terminal = 'pwntools-terminal'	×
289	args = []	×
290	elif 'TMUX' in os.environ and which('tmux'):	×
291	terminal = 'tmux'	×
292	args = ['splitw']	×
293	elif 'STY' in os.environ and which('screen'):	×
294	terminal = 'screen'	×
295	args = ['-t','pwntools-gdb','bash','-c']	×
296	elif 'TERM_PROGRAM' in os.environ:	×
297	terminal = os.environ['TERM_PROGRAM']	×
298	args = []	×
299	elif 'DISPLAY' in os.environ and which('x-terminal-emulator'):	×
300	terminal = 'x-terminal-emulator'	×
301	args = ['-e']	×
302	elif 'KONSOLE_VERSION' in os.environ and which('qdbus'):	×
303	qdbus = which('qdbus')	×
304	window_id = os.environ['WINDOWID']	×
305	konsole_dbus_service = os.environ['KONSOLE_DBUS_SERVICE']	×
306
307	with subprocess.Popen((qdbus, konsole_dbus_service), stdout=subprocess.PIPE) as proc:	×
308	lines = proc.communicate()[0].decode().split('\n')	×
309
310	# Iterate over all MainWindows
311	for line in lines:	×
312	parts = line.split('/')	×
313	if len(parts) == 3 and parts[2].startswith('MainWindow_'):	×
314	name = parts[2]	×
315	with subprocess.Popen((qdbus, konsole_dbus_service, '/konsole/' + name,	×
316	'org.kde.KMainWindow.winId'), stdout=subprocess.PIPE) as proc:
317	target_window_id = proc.communicate()[0].decode().strip()	×
318	if target_window_id == window_id:	×
319	break	×
320	else:
321	log.error('MainWindow not found')	×
322
323	# Split
324	subprocess.run((qdbus, konsole_dbus_service, '/konsole/' + name,	×
325	'org.kde.KMainWindow.activateAction', 'split-view-left-right'), stdout=subprocess.DEVNULL)
326
327	# Find new session
328	with subprocess.Popen((qdbus, konsole_dbus_service, os.environ['KONSOLE_DBUS_WINDOW'],	×
329	'org.kde.konsole.Window.sessionList'), stdout=subprocess.PIPE) as proc:
330	session_list = map(int, proc.communicate()[0].decode().split())	×
331	last_konsole_session = max(session_list)	×
332
333	terminal = 'qdbus'	×
334	args = [konsole_dbus_service, '/Sessions/{}'.format(last_konsole_session),	×
335	'org.kde.konsole.Session.runCommand']
336
337	else:
338	is_wsl = False	×
339	if os.path.exists('/proc/sys/kernel/osrelease'):	×
340	with open('/proc/sys/kernel/osrelease', 'rb') as f:	×
341	is_wsl = b'icrosoft' in f.read()	×
342	if is_wsl and which('cmd.exe') and which('wsl.exe') and which('bash.exe'):	×
343	terminal = 'cmd.exe'	×
344	args = ['/c', 'start']	×
345	distro_name = os.getenv('WSL_DISTRO_NAME')	×
346
347	# Split pane in Windows Terminal
348	if 'WT_SESSION' in os.environ and which('wt.exe'):	×
349	args.extend(['wt.exe', '-w', '0', 'split-pane', '-d', '.'])	×
350
351	if distro_name:	×
352	args.extend(['wsl.exe', '-d', distro_name, 'bash', '-c'])	×
353	else:
354	args.extend(['bash.exe', '-c'])	×
355
356
357	if not terminal:	1!
358	log.error('Could not find a terminal binary to use. Set context.terminal to your terminal.')	×
359	elif not which(terminal):	1!
360	log.error('Could not find terminal binary %r. Set context.terminal to your terminal.' % terminal)	×
361
362	if isinstance(args, tuple):	1!
363	args = list(args)	×
364
365	# When not specifying context.terminal explicitly, we used to set these flags above.
366	# However, if specifying terminal=['tmux', 'splitw', '-h'], we would be lacking these flags.
367	# Instead, set them here and hope for the best.
368	if terminal == 'tmux':	1!
369	args += ['-F' '#{pane_pid}', '-P']	×
370
371	argv = [which(terminal)] + args	1✔
372
373	if isinstance(command, six.string_types):	1!
374	if ';' in command:	×
375	log.error("Cannot use commands with semicolon. Create a script and invoke that directly.")	×
376	argv += [command]	×
377	elif isinstance(command, (list, tuple)):	1!
378	# Dump the full command line to a temporary file so we can be sure that
379	# it is parsed correctly, and we do not need to account for shell expansion
380	script = '''	1✔
381	#!{executable!s}
382	import os
383	os.execve({argv0!r}, {argv!r}, os.environ)
384	'''
385	script = script.format(executable=sys.executable,	1✔
386	argv=command,
387	argv0=which(command[0]))
388	script = script.lstrip()	1✔
389
390	log.debug("Created script for new terminal:\n%s" % script)	1✔
391
392	with tempfile.NamedTemporaryFile(delete=False, mode='wt+') as tmp:	1✔
393	tmp.write(script)	1✔
394	tmp.flush()	1✔
395	os.chmod(tmp.name, 0o700)	1✔
396	argv += [tmp.name]	1✔
397
398
399	log.debug("Launching a new terminal: %r" % argv)	1✔
400
401	stdin = stdout = stderr = open(os.devnull, 'r+b')	1✔
402	if terminal == 'tmux':	1!
403	stdout = subprocess.PIPE	×
404
405	p = subprocess.Popen(argv, stdin=stdin, stdout=stdout, stderr=stderr, preexec_fn=preexec_fn)	1✔
406
407	if terminal == 'tmux':	1!
408	out, _ = p.communicate()	×
409	pid = int(out)	×
410	elif terminal == 'qdbus':	1!
411	with subprocess.Popen((qdbus, konsole_dbus_service, '/Sessions/{}'.format(last_konsole_session),	×
412	'org.kde.konsole.Session.processId'), stdout=subprocess.PIPE) as proc:
413	pid = int(proc.communicate()[0].decode())	×
414	else:
415	pid = p.pid	1✔
416
417	if kill_at_exit:	1!
418	def kill():	1✔
419	try:	×
420	if terminal == 'qdbus':	×
421	os.kill(pid, signal.SIGHUP)	×
422	else:
423	os.kill(pid, signal.SIGTERM)	×
424	except OSError:	×
425	pass	×
426
427	atexit.register(kill)	1✔
428
429	return pid	1✔
430
431	def parse_ldd_output(output):	1✔
432	"""Parses the output from a run of 'ldd' on a binary.
433	Returns a dictionary of {path: address} for
434	each library required by the specified binary.
435
436	Arguments:
437	output(str): The output to parse
438
439	Example:
440	>>> sorted(parse_ldd_output('''
441	... linux-vdso.so.1 => (0x00007fffbf5fe000)
442	... libtinfo.so.5 => /lib/x86_64-linux-gnu/libtinfo.so.5 (0x00007fe28117f000)
443	... libdl.so.2 => /lib/x86_64-linux-gnu/libdl.so.2 (0x00007fe280f7b000)
444	... libc.so.6 => /lib/x86_64-linux-gnu/libc.so.6 (0x00007fe280bb4000)
445	... /lib64/ld-linux-x86-64.so.2 (0x00007fe2813dd000)
446	... ''').keys())
447	['/lib/x86_64-linux-gnu/libc.so.6', '/lib/x86_64-linux-gnu/libdl.so.2', '/lib/x86_64-linux-gnu/libtinfo.so.5', '/lib64/ld-linux-x86-64.so.2']
448	"""
449	expr_linux = re.compile(r'\s(?P<lib>\S?/\S+)\s+\((?P<addr>0x.+)\)')	1✔
450	expr_openbsd = re.compile(r'^\s+(?P<addr>[0-9a-f]+)\s+[0-9a-f]+\s+\S+\s+[01]\s+[0-9]+\s+[0-9]+\s+(?P<lib>\S+)$')	1✔
451	libs = {}	1✔
452
453	for s in output.split('\n'):	1✔
454	match = expr_linux.search(s) or expr_openbsd.search(s)	1✔
455	if not match:	1✔
456	continue	1✔
457	lib, addr = match.group('lib'), match.group('addr')	1✔
458	libs[lib] = int(addr, 16)	1✔
459
460	return libs	1✔
461
462	def mkdir_p(path):	1✔
463	"""Emulates the behavior of ``mkdir -p``."""
464
465	try:	1✔
466	os.makedirs(path)	1✔
467	except OSError as exc:	1✔
468	if exc.errno == errno.EEXIST and os.path.isdir(path):	1!
469	pass	1✔
470	else:
471	raise	×
472
473	def dealarm_shell(tube):	1✔
474	"""Given a tube which is a shell, dealarm it.
475	"""
476	tube.clean()	×
477
478	tube.sendline('which python \|\| echo')	×
479	if tube.recvline().startswith('/'):	×
480	tube.sendline('''exec python -c "import signal, os; signal.alarm(0); os.execl('$SHELL','')"''')	×
481	return tube	×
482
483	tube.sendline('which perl \|\| echo')	×
484	if tube.recvline().startswith('/'):	×
485	tube.sendline('''exec perl -e "alarm 0; exec '${SHELL:-/bin/sh}'"''')	×
486	return tube	×
487
488	return None	×
489
490	def register_sizes(regs, in_sizes):	1✔
491	"""Create dictionaries over register sizes and relations
492
493	Given a list of lists of overlapping register names (e.g. ['eax','ax','al','ah']) and a list of input sizes,
494	it returns the following:
495
496	* all_regs : list of all valid registers
497	* sizes[reg] : the size of reg in bits
498	* bigger[reg] : list of overlapping registers bigger than reg
499	* smaller[reg]: list of overlapping registers smaller than reg
500
501	Used in i386/AMD64 shellcode, e.g. the mov-shellcode.
502
503	Example:
504	>>> regs = [['eax', 'ax', 'al', 'ah'],['ebx', 'bx', 'bl', 'bh'],
505	... ['ecx', 'cx', 'cl', 'ch'],
506	... ['edx', 'dx', 'dl', 'dh'],
507	... ['edi', 'di'],
508	... ['esi', 'si'],
509	... ['ebp', 'bp'],
510	... ['esp', 'sp'],
511	... ]
512	>>> all_regs, sizes, bigger, smaller = register_sizes(regs, [32, 16, 8, 8])
513	>>> all_regs
514	['eax', 'ax', 'al', 'ah', 'ebx', 'bx', 'bl', 'bh', 'ecx', 'cx', 'cl', 'ch', 'edx', 'dx', 'dl', 'dh', 'edi', 'di', 'esi', 'si', 'ebp', 'bp', 'esp', 'sp']
515	>>> pprint(sizes)
516	{'ah': 8,
517	'al': 8,
518	'ax': 16,
519	'bh': 8,
520	'bl': 8,
521	'bp': 16,
522	'bx': 16,
523	'ch': 8,
524	'cl': 8,
525	'cx': 16,
526	'dh': 8,
527	'di': 16,
528	'dl': 8,
529	'dx': 16,
530	'eax': 32,
531	'ebp': 32,
532	'ebx': 32,
533	'ecx': 32,
534	'edi': 32,
535	'edx': 32,
536	'esi': 32,
537	'esp': 32,
538	'si': 16,
539	'sp': 16}
540	>>> pprint(bigger)
541	{'ah': ['eax', 'ax', 'ah'],
542	'al': ['eax', 'ax', 'al'],
543	'ax': ['eax', 'ax'],
544	'bh': ['ebx', 'bx', 'bh'],
545	'bl': ['ebx', 'bx', 'bl'],
546	'bp': ['ebp', 'bp'],
547	'bx': ['ebx', 'bx'],
548	'ch': ['ecx', 'cx', 'ch'],
549	'cl': ['ecx', 'cx', 'cl'],
550	'cx': ['ecx', 'cx'],
551	'dh': ['edx', 'dx', 'dh'],
552	'di': ['edi', 'di'],
553	'dl': ['edx', 'dx', 'dl'],
554	'dx': ['edx', 'dx'],
555	'eax': ['eax'],
556	'ebp': ['ebp'],
557	'ebx': ['ebx'],
558	'ecx': ['ecx'],
559	'edi': ['edi'],
560	'edx': ['edx'],
561	'esi': ['esi'],
562	'esp': ['esp'],
563	'si': ['esi', 'si'],
564	'sp': ['esp', 'sp']}
565	>>> pprint(smaller)
566	{'ah': [],
567	'al': [],
568	'ax': ['al', 'ah'],
569	'bh': [],
570	'bl': [],
571	'bp': [],
572	'bx': ['bl', 'bh'],
573	'ch': [],
574	'cl': [],
575	'cx': ['cl', 'ch'],
576	'dh': [],
577	'di': [],
578	'dl': [],
579	'dx': ['dl', 'dh'],
580	'eax': ['ax', 'al', 'ah'],
581	'ebp': ['bp'],
582	'ebx': ['bx', 'bl', 'bh'],
583	'ecx': ['cx', 'cl', 'ch'],
584	'edi': ['di'],
585	'edx': ['dx', 'dl', 'dh'],
586	'esi': ['si'],
587	'esp': ['sp'],
588	'si': [],
589	'sp': []}
590	"""
591	sizes = {}	1✔
592	bigger = {}	1✔
593	smaller = {}	1✔
594
595	for l in regs:	1✔
596	for r, s in zip(l, in_sizes):	1✔
597	sizes[r] = s	1✔
598
599	for r in l:	1✔
600	bigger[r] = [r_ for r_ in l if sizes[r_] > sizes[r] or r == r_]	1✔
601	smaller[r] = [r_ for r_ in l if sizes[r_] < sizes[r]]	1✔
602
603	return lists.concat(regs), sizes, bigger, smaller	1✔
604
605
606	def python_2_bytes_compatible(klass):	1✔
607	"""
608	A class decorator that defines __str__ methods under Python 2.
609	Under Python 3 it does nothing.
610	"""
611	if six.PY2:	1!
612	if '__str__' not in klass.__dict__:	×
613	klass.__str__ = klass.__bytes__	×
614	return klass	1✔

Gallopsled / pwntools / 5573614195

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