7250412654

Committed 18 Dec 2023 03:44PM UTC coverage: 74.547% (+0.1%) from 74.452%

Build # 7250412654

Build Type

push

github

Committed by

web-flow

Commit Message

Merge branch 'dev' into retguard

Run Details

4565 of 7244 branches covered (0.0%)

350 of 507 new or added lines in 17 files covered. (69.03%)

13 existing lines in 5 files now uncovered.

12843 of 17228 relevant lines covered (74.55%)

0.75 hits per line

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

61.98

/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)
            # Check if = is in the key, Required check since we sometimes call ctypes.execve directly
            # https://github.com/python/cpython/blob/025995feadaeebeef5d808f2564f0fd65b704ea5/Modules/posixmodule.c#L6476
            if b'=' in packing._encode(k):
                log.error('Environment keys may not contain "=": %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='/bin/env ' * (' ' in sys.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()
        try:
            pid = int(out)
        except ValueError:
            pid = None
        if pid is None:
            log.error("Could not parse PID from tmux output (%r). Start tmux first.", 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	# Check if = is in the key, Required check since we sometimes call ctypes.execve directly
233	# https://github.com/python/cpython/blob/025995feadaeebeef5d808f2564f0fd65b704ea5/Modules/posixmodule.c#L6476
234	if b'=' in packing._encode(k):	1!
235	log.error('Environment keys may not contain "=": %r' % (k))	×
236	if not isinstance(v, (bytes, six.text_type)):	1!
237	log.error('Environment values must be strings: %r=%r' % (k,v))	×
238	k = packing._need_bytes(k, level, 0x80) # ASCII text is okay	1✔
239	v = packing._need_bytes(v, level, 0x80) # ASCII text is okay	1✔
240	if b'\x00' in k[:-1]:	1!
241	log.error('Inappropriate nulls in env key: %r' % (k))	×
242	if b'\x00' in v[:-1]:	1!
243	log.error('Inappropriate nulls in env value: %r=%r' % (k, v))	×
244	env2.append((bytearray(k.rstrip(b'\x00')), bytearray(v.rstrip(b'\x00'))))	1✔
245
246	return argv, env2 or env	1✔
247
248
249	def run_in_new_terminal(command, terminal=None, args=None, kill_at_exit=True, preexec_fn=None):	1✔
250	"""run_in_new_terminal(command, terminal=None, args=None, kill_at_exit=True, preexec_fn=None) -> int
251
252	Run a command in a new terminal.
253
254	When ``terminal`` is not set:
255	- If ``context.terminal`` is set it will be used.
256	If it is an iterable then ``context.terminal[1:]`` are default arguments.
257	- If a ``pwntools-terminal`` command exists in ``$PATH``, it is used
258	- If tmux is detected (by the presence of the ``$TMUX`` environment
259	variable), a new pane will be opened.
260	- If GNU Screen is detected (by the presence of the ``$STY`` environment
261	variable), a new screen will be opened.
262	- If ``$TERM_PROGRAM`` is set, that is used.
263	- If X11 is detected (by the presence of the ``$DISPLAY`` environment
264	variable), ``x-terminal-emulator`` is used.
265	- If KDE Konsole is detected (by the presence of the ``$KONSOLE_VERSION``
266	environment variable), a terminal will be split.
267	- If WSL (Windows Subsystem for Linux) is detected (by the presence of
268	a ``wsl.exe`` binary in the ``$PATH`` and ``/proc/sys/kernel/osrelease``
269	containing ``Microsoft``), a new ``cmd.exe`` window will be opened.
270
271	If `kill_at_exit` is :const:`True`, try to close the command/terminal when the
272	current process exits. This may not work for all terminal types.
273
274	Arguments:
275	command (str): The command to run.
276	terminal (str): Which terminal to use.
277	args (list): Arguments to pass to the terminal
278	kill_at_exit (bool): Whether to close the command/terminal on process exit.
279	preexec_fn (callable): Callable to invoke before exec().
280
281	Note:
282	The command is opened with ``/dev/null`` for stdin, stdout, stderr.
283
284	Returns:
285	PID of the new terminal process
286	"""
287	if not terminal:	1!
288	if context.terminal:	1!
289	terminal = context.terminal[0]	1✔
290	args = context.terminal[1:]	1✔
291	elif which('pwntools-terminal'):	×
292	terminal = 'pwntools-terminal'	×
293	args = []	×
294	elif 'TMUX' in os.environ and which('tmux'):	×
295	terminal = 'tmux'	×
296	args = ['splitw']	×
297	elif 'STY' in os.environ and which('screen'):	×
298	terminal = 'screen'	×
299	args = ['-t','pwntools-gdb','bash','-c']	×
300	elif 'TERM_PROGRAM' in os.environ:	×
301	terminal = os.environ['TERM_PROGRAM']	×
302	args = []	×
303	elif 'DISPLAY' in os.environ and which('x-terminal-emulator'):	×
304	terminal = 'x-terminal-emulator'	×
305	args = ['-e']	×
306	elif 'KONSOLE_VERSION' in os.environ and which('qdbus'):	×
307	qdbus = which('qdbus')	×
308	window_id = os.environ['WINDOWID']	×
309	konsole_dbus_service = os.environ['KONSOLE_DBUS_SERVICE']	×
310
311	with subprocess.Popen((qdbus, konsole_dbus_service), stdout=subprocess.PIPE) as proc:	×
312	lines = proc.communicate()[0].decode().split('\n')	×
313
314	# Iterate over all MainWindows
315	for line in lines:	×
316	parts = line.split('/')	×
317	if len(parts) == 3 and parts[2].startswith('MainWindow_'):	×
318	name = parts[2]	×
319	with subprocess.Popen((qdbus, konsole_dbus_service, '/konsole/' + name,	×
320	'org.kde.KMainWindow.winId'), stdout=subprocess.PIPE) as proc:
321	target_window_id = proc.communicate()[0].decode().strip()	×
322	if target_window_id == window_id:	×
323	break	×
324	else:
325	log.error('MainWindow not found')	×
326
327	# Split
328	subprocess.run((qdbus, konsole_dbus_service, '/konsole/' + name,	×
329	'org.kde.KMainWindow.activateAction', 'split-view-left-right'), stdout=subprocess.DEVNULL)
330
331	# Find new session
332	with subprocess.Popen((qdbus, konsole_dbus_service, os.environ['KONSOLE_DBUS_WINDOW'],	×
333	'org.kde.konsole.Window.sessionList'), stdout=subprocess.PIPE) as proc:
334	session_list = map(int, proc.communicate()[0].decode().split())	×
335	last_konsole_session = max(session_list)	×
336
337	terminal = 'qdbus'	×
338	args = [konsole_dbus_service, '/Sessions/{}'.format(last_konsole_session),	×
339	'org.kde.konsole.Session.runCommand']
340
341	else:
342	is_wsl = False	×
343	if os.path.exists('/proc/sys/kernel/osrelease'):	×
344	with open('/proc/sys/kernel/osrelease', 'rb') as f:	×
345	is_wsl = b'icrosoft' in f.read()	×
346	if is_wsl and which('cmd.exe') and which('wsl.exe') and which('bash.exe'):	×
347	terminal = 'cmd.exe'	×
348	args = ['/c', 'start']	×
349	distro_name = os.getenv('WSL_DISTRO_NAME')	×
350
351	# Split pane in Windows Terminal
352	if 'WT_SESSION' in os.environ and which('wt.exe'):	×
353	args.extend(['wt.exe', '-w', '0', 'split-pane', '-d', '.'])	×
354
355	if distro_name:	×
356	args.extend(['wsl.exe', '-d', distro_name, 'bash', '-c'])	×
357	else:
358	args.extend(['bash.exe', '-c'])	×
359
360
361	if not terminal:	1!
362	log.error('Could not find a terminal binary to use. Set context.terminal to your terminal.')	×
363	elif not which(terminal):	1!
364	log.error('Could not find terminal binary %r. Set context.terminal to your terminal.' % terminal)	×
365
366	if isinstance(args, tuple):	1!
367	args = list(args)	×
368
369	# When not specifying context.terminal explicitly, we used to set these flags above.
370	# However, if specifying terminal=['tmux', 'splitw', '-h'], we would be lacking these flags.
371	# Instead, set them here and hope for the best.
372	if terminal == 'tmux':	1!
373	args += ['-F' '#{pane_pid}', '-P']	×
374
375	argv = [which(terminal)] + args	1✔
376
377	if isinstance(command, six.string_types):	1!
378	if ';' in command:	×
379	log.error("Cannot use commands with semicolon. Create a script and invoke that directly.")	×
380	argv += [command]	×
381	elif isinstance(command, (list, tuple)):	1!
382	# Dump the full command line to a temporary file so we can be sure that
383	# it is parsed correctly, and we do not need to account for shell expansion
384	script = '''	1✔
385	#!{executable!s}
386	import os
387	os.execve({argv0!r}, {argv!r}, os.environ)
388	'''
389	script = script.format(executable='/bin/env ' * (' ' in sys.executable) + sys.executable,	1✔
390	argv=command,
391	argv0=which(command[0]))
392	script = script.lstrip()	1✔
393
394	log.debug("Created script for new terminal:\n%s" % script)	1✔
395
396	with tempfile.NamedTemporaryFile(delete=False, mode='wt+') as tmp:	1✔
397	tmp.write(script)	1✔
398	tmp.flush()	1✔
399	os.chmod(tmp.name, 0o700)	1✔
400	argv += [tmp.name]	1✔
401
402
403	log.debug("Launching a new terminal: %r" % argv)	1✔
404
405	stdin = stdout = stderr = open(os.devnull, 'r+b')	1✔
406	if terminal == 'tmux':	1!
407	stdout = subprocess.PIPE	×
408
409	p = subprocess.Popen(argv, stdin=stdin, stdout=stdout, stderr=stderr, preexec_fn=preexec_fn)	1✔
410
411	if terminal == 'tmux':	1!
412	out, _ = p.communicate()	×
NEW 413	try:	×
NEW 414	pid = int(out)	×
NEW 415	except ValueError:	×
NEW 416	pid = None	×
NEW 417	if pid is None:	×
NEW 418	log.error("Could not parse PID from tmux output (%r). Start tmux first.", out)	×
419	elif terminal == 'qdbus':	1!
420	with subprocess.Popen((qdbus, konsole_dbus_service, '/Sessions/{}'.format(last_konsole_session),	×
421	'org.kde.konsole.Session.processId'), stdout=subprocess.PIPE) as proc:
422	pid = int(proc.communicate()[0].decode())	×
423	else:
424	pid = p.pid	1✔
425
426	if kill_at_exit:	1!
427	def kill():	1✔
428	try:	×
429	if terminal == 'qdbus':	×
430	os.kill(pid, signal.SIGHUP)	×
431	else:
432	os.kill(pid, signal.SIGTERM)	×
433	except OSError:	×
434	pass	×
435
436	atexit.register(kill)	1✔
437
438	return pid	1✔
439
440	def parse_ldd_output(output):	1✔
441	"""Parses the output from a run of 'ldd' on a binary.
442	Returns a dictionary of {path: address} for
443	each library required by the specified binary.
444
445	Arguments:
446	output(str): The output to parse
447
448	Example:
449	>>> sorted(parse_ldd_output('''
450	... linux-vdso.so.1 => (0x00007fffbf5fe000)
451	... libtinfo.so.5 => /lib/x86_64-linux-gnu/libtinfo.so.5 (0x00007fe28117f000)
452	... libdl.so.2 => /lib/x86_64-linux-gnu/libdl.so.2 (0x00007fe280f7b000)
453	... libc.so.6 => /lib/x86_64-linux-gnu/libc.so.6 (0x00007fe280bb4000)
454	... /lib64/ld-linux-x86-64.so.2 (0x00007fe2813dd000)
455	... ''').keys())
456	['/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']
457	"""
458	expr_linux = re.compile(r'\s(?P<lib>\S?/\S+)\s+\((?P<addr>0x.+)\)')	1✔
459	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✔
460	libs = {}	1✔
461
462	for s in output.split('\n'):	1✔
463	match = expr_linux.search(s) or expr_openbsd.search(s)	1✔
464	if not match:	1✔
465	continue	1✔
466	lib, addr = match.group('lib'), match.group('addr')	1✔
467	libs[lib] = int(addr, 16)	1✔
468
469	return libs	1✔
470
471	def mkdir_p(path):	1✔
472	"""Emulates the behavior of ``mkdir -p``."""
473
474	try:	1✔
475	os.makedirs(path)	1✔
476	except OSError as exc:	1✔
477	if exc.errno == errno.EEXIST and os.path.isdir(path):	1!
478	pass	1✔
479	else:
480	raise	×
481
482	def dealarm_shell(tube):	1✔
483	"""Given a tube which is a shell, dealarm it.
484	"""
485	tube.clean()	×
486
487	tube.sendline('which python \|\| echo')	×
488	if tube.recvline().startswith('/'):	×
489	tube.sendline('''exec python -c "import signal, os; signal.alarm(0); os.execl('$SHELL','')"''')	×
490	return tube	×
491
492	tube.sendline('which perl \|\| echo')	×
493	if tube.recvline().startswith('/'):	×
494	tube.sendline('''exec perl -e "alarm 0; exec '${SHELL:-/bin/sh}'"''')	×
495	return tube	×
496
497	return None	×
498
499	def register_sizes(regs, in_sizes):	1✔
500	"""Create dictionaries over register sizes and relations
501
502	Given a list of lists of overlapping register names (e.g. ['eax','ax','al','ah']) and a list of input sizes,
503	it returns the following:
504
505	* all_regs : list of all valid registers
506	* sizes[reg] : the size of reg in bits
507	* bigger[reg] : list of overlapping registers bigger than reg
508	* smaller[reg]: list of overlapping registers smaller than reg
509
510	Used in i386/AMD64 shellcode, e.g. the mov-shellcode.
511
512	Example:
513	>>> regs = [['eax', 'ax', 'al', 'ah'],['ebx', 'bx', 'bl', 'bh'],
514	... ['ecx', 'cx', 'cl', 'ch'],
515	... ['edx', 'dx', 'dl', 'dh'],
516	... ['edi', 'di'],
517	... ['esi', 'si'],
518	... ['ebp', 'bp'],
519	... ['esp', 'sp'],
520	... ]
521	>>> all_regs, sizes, bigger, smaller = register_sizes(regs, [32, 16, 8, 8])
522	>>> all_regs
523	['eax', 'ax', 'al', 'ah', 'ebx', 'bx', 'bl', 'bh', 'ecx', 'cx', 'cl', 'ch', 'edx', 'dx', 'dl', 'dh', 'edi', 'di', 'esi', 'si', 'ebp', 'bp', 'esp', 'sp']
524	>>> pprint(sizes)
525	{'ah': 8,
526	'al': 8,
527	'ax': 16,
528	'bh': 8,
529	'bl': 8,
530	'bp': 16,
531	'bx': 16,
532	'ch': 8,
533	'cl': 8,
534	'cx': 16,
535	'dh': 8,
536	'di': 16,
537	'dl': 8,
538	'dx': 16,
539	'eax': 32,
540	'ebp': 32,
541	'ebx': 32,
542	'ecx': 32,
543	'edi': 32,
544	'edx': 32,
545	'esi': 32,
546	'esp': 32,
547	'si': 16,
548	'sp': 16}
549	>>> pprint(bigger)
550	{'ah': ['eax', 'ax', 'ah'],
551	'al': ['eax', 'ax', 'al'],
552	'ax': ['eax', 'ax'],
553	'bh': ['ebx', 'bx', 'bh'],
554	'bl': ['ebx', 'bx', 'bl'],
555	'bp': ['ebp', 'bp'],
556	'bx': ['ebx', 'bx'],
557	'ch': ['ecx', 'cx', 'ch'],
558	'cl': ['ecx', 'cx', 'cl'],
559	'cx': ['ecx', 'cx'],
560	'dh': ['edx', 'dx', 'dh'],
561	'di': ['edi', 'di'],
562	'dl': ['edx', 'dx', 'dl'],
563	'dx': ['edx', 'dx'],
564	'eax': ['eax'],
565	'ebp': ['ebp'],
566	'ebx': ['ebx'],
567	'ecx': ['ecx'],
568	'edi': ['edi'],
569	'edx': ['edx'],
570	'esi': ['esi'],
571	'esp': ['esp'],
572	'si': ['esi', 'si'],
573	'sp': ['esp', 'sp']}
574	>>> pprint(smaller)
575	{'ah': [],
576	'al': [],
577	'ax': ['al', 'ah'],
578	'bh': [],
579	'bl': [],
580	'bp': [],
581	'bx': ['bl', 'bh'],
582	'ch': [],
583	'cl': [],
584	'cx': ['cl', 'ch'],
585	'dh': [],
586	'di': [],
587	'dl': [],
588	'dx': ['dl', 'dh'],
589	'eax': ['ax', 'al', 'ah'],
590	'ebp': ['bp'],
591	'ebx': ['bx', 'bl', 'bh'],
592	'ecx': ['cx', 'cl', 'ch'],
593	'edi': ['di'],
594	'edx': ['dx', 'dl', 'dh'],
595	'esi': ['si'],
596	'esp': ['sp'],
597	'si': [],
598	'sp': []}
599	"""
600	sizes = {}	1✔
601	bigger = {}	1✔
602	smaller = {}	1✔
603
604	for l in regs:	1✔
605	for r, s in zip(l, in_sizes):	1✔
606	sizes[r] = s	1✔
607
608	for r in l:	1✔
609	bigger[r] = [r_ for r_ in l if sizes[r_] > sizes[r] or r == r_]	1✔
610	smaller[r] = [r_ for r_ in l if sizes[r_] < sizes[r]]	1✔
611
612	return lists.concat(regs), sizes, bigger, smaller	1✔
613
614
615	def python_2_bytes_compatible(klass):	1✔
616	"""
617	A class decorator that defines __str__ methods under Python 2.
618	Under Python 3 it does nothing.
619	"""
620	if six.PY2:	1✔
621	if '__str__' not in klass.__dict__:	1!
622	klass.__str__ = klass.__bytes__	1✔
623	return klass	1✔

Gallopsled / pwntools / 7250412654

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