pwnlib.tubes.process
— Processes
- class pwnlib.tubes.process.process(argv=None, shell=False, executable=None, cwd=None, env=None, ignore_environ=None, stdin=-1, stdout=<pwnlib.tubes.process.PTY object>, stderr=-2, close_fds=True, preexec_fn=<function process.<lambda>>, raw=True, aslr=None, setuid=None, where='local', display=None, alarm=None, creationflags=0, *args, **kwargs)[source]
Bases:
tube
Spawns a new process, and wraps it with a tube for communication.
- Parameters
argv (list) – List of arguments to pass to the spawned process.
shell (bool) – Set to True to interpret argv as a string to pass to the shell for interpretation instead of as argv.
executable (str) – Path to the binary to execute. If
None
, usesargv[0]
. Cannot be used withshell
.cwd (str) – Working directory. Uses the current working directory by default.
env (dict) – Environment variables to add to the environment.
ignore_environ (bool) – Ignore Python’s environment. By default use Python’s environment iff env not specified.
stdin (int) – File object or file descriptor number to use for
stdin
. By default, a pipe is used. A pty can be used instead by setting this toPTY
. This will cause programs to behave in an interactive manner (e.g..,python
will show a>>>
prompt). If the application reads from/dev/tty
directly, use a pty.stdout (int) – File object or file descriptor number to use for
stdout
. By default, a pty is used so that any stdout buffering by libc routines is disabled. May also bePIPE
to use a normal pipe.stderr (int) – File object or file descriptor number to use for
stderr
. By default,STDOUT
is used. May also bePIPE
to use a separate pipe, although thepwnlib.tubes.tube.tube
wrapper will not be able to read this data.close_fds (bool) – Close all open file descriptors except stdin, stdout, stderr. By default,
True
is used.preexec_fn (callable) – Callable to invoke immediately before calling
execve
.raw (bool) – Set the created pty to raw mode (i.e. disable echo and control characters).
True
by default. If no pty is created, this has no effect.aslr (bool) –
If set to
False
, disable ASLR viapersonality
(setarch -R
) andsetrlimit
(ulimit -s unlimited
).This disables ASLR for the target process. However, the
setarch
changes are lost if asetuid
binary is executed.The default value is inherited from
context.aslr
. Seesetuid
below for additional options and information.setuid (bool) –
Used to control setuid status of the target binary, and the corresponding actions taken.
By default, this value is
None
, so no assumptions are made.If
True
, treat the target binary assetuid
. This modifies the mechanisms used to disable ASLR on the process ifaslr=False
. This is useful for debugging locally, when the exploit is asetuid
binary.If
False
, preventsetuid
bits from taking effect on the target binary. This is only supported on Linux, with kernels v3.5 or greater.where (str) – Where the process is running, used for logging purposes.
display (list) – List of arguments to display, instead of the main executable name.
alarm (int) – Set a SIGALRM alarm timeout on the process.
creationflags (int) – Windows only. Flags to pass to
CreateProcess
.
Examples
>>> p = process('python') >>> p.sendline(b"print('Hello world')") >>> p.sendline(b"print('Wow, such data')") >>> b'' == p.recv(timeout=0.01) True >>> p.shutdown('send') >>> p.proc.stdin.closed True >>> p.connected('send') False >>> p.recvline() b'Hello world\n' >>> p.recvuntil(b',') b'Wow,' >>> p.recvregex(b'.*data') b' such data' >>> p.recv() b'\n' >>> p.recv() Traceback (most recent call last): ... EOFError
>>> p = process('cat') >>> d = open('/dev/urandom', 'rb').read(4096) >>> p.recv(timeout=0.1) b'' >>> p.write(d) >>> p.recvrepeat(0.1) == d True >>> p.recv(timeout=0.1) b'' >>> p.shutdown('send') >>> p.wait_for_close() >>> p.poll() 0
>>> p = process('cat /dev/zero | head -c8', shell=True, stderr=open('/dev/null', 'w+b')) >>> p.recv() b'\x00\x00\x00\x00\x00\x00\x00\x00'
>>> p = process(['python','-c','import os; print(os.read(2,1024).decode())'], ... preexec_fn = lambda: os.dup2(0,2)) >>> p.sendline(b'hello') >>> p.recvline() b'hello\n'
>>> stack_smashing = ['python','-c','open("/dev/tty","wb").write(b"stack smashing detected")'] >>> process(stack_smashing).recvall() b'stack smashing detected'
>>> process(stack_smashing, stdout=PIPE).recvall() b''
>>> getpass = ['python','-c','import getpass; print(getpass.getpass("XXX"))'] >>> p = process(getpass, stdin=PTY) >>> p.recv() b'XXX' >>> p.sendline(b'hunter2') >>> p.recvall() b'\nhunter2\n'
>>> process('echo hello 1>&2', shell=True).recvall() b'hello\n'
>>> process('echo hello 1>&2', shell=True, stderr=PIPE).recvall() b''
>>> a = process(['cat', '/proc/self/maps']).recvall() >>> b = process(['cat', '/proc/self/maps'], aslr=False).recvall() >>> with context.local(aslr=False): ... c = process(['cat', '/proc/self/maps']).recvall() >>> a == b False >>> b == c True
>>> process(['sh','-c','ulimit -s'], aslr=0).recvline() b'unlimited\n'
>>> io = process(['sh','-c','sleep 10; exit 7'], alarm=2) >>> io.poll(block=True) == -signal.SIGALRM True
>>> binary = ELF.from_assembly('nop', arch='mips') >>> p = process(binary.path) >>> binary_dir, binary_name = os.path.split(binary.path) >>> p = process('./{}'.format(binary_name), cwd=binary_dir) >>> p = process(binary.path, cwd=binary_dir) >>> p = process('./{}'.format(binary_name), cwd=os.path.relpath(binary_dir)) >>> p = process(binary.path, cwd=os.path.relpath(binary_dir))
- __getattr__(attr)[source]
Permit pass-through access to the underlying process object for fields like
pid
andstdin
.
- __init__(argv=None, shell=False, executable=None, cwd=None, env=None, ignore_environ=None, stdin=-1, stdout=<pwnlib.tubes.process.PTY object>, stderr=-2, close_fds=True, preexec_fn=<function process.<lambda>>, raw=True, aslr=None, setuid=None, where='local', display=None, alarm=None, creationflags=0, *args, **kwargs)[source]
- __on_enoexec(exception)[source]
We received an ‘exec format’ error (ENOEXEC)
This implies that the user tried to execute e.g. an ARM binary on a non-ARM system, and does not have binfmt helpers installed for QEMU.
- __preexec_fn()[source]
Routine executed in the child process before invoking execve().
Handles setting the controlling TTY as well as invoking the user- supplied preexec_fn.
- __pty_make_controlling_tty(tty_fd)[source]
This makes the pseudo-terminal the controlling tty. This should be more portable than the pty.fork() function. Specifically, this should work on Solaris.
- _validate(cwd, executable, argv, env)[source]
Perform extended validation on the executable path, argv, and envp.
Mostly to make Python happy, but also to prevent common pitfalls.
- address_mapping(address) mapping [source]
Returns the mapping at the specified address.
Example
>>> p = process(['cat']) >>> p.sendline(b'meow') >>> p.recvline() b'meow\n' >>> libc = p.libc_mapping().address >>> heap = p.heap_mapping().address >>> elf = p.elf_mapping().address >>> p.address_mapping(libc).path '.../libc...' >>> p.address_mapping(heap + 0x123).path '[heap]' >>> p.address_mapping(elf + 0x1234).path '.../cat' >>> p.address_mapping(elf - 0x1234) == None True
- can_recv_raw(timeout) bool [source]
Should not be called directly. Returns True, if there is data available within the timeout, but ignores the buffer on the object.
- communicate(stdin=None) str [source]
Calls
subprocess.Popen.communicate()
method on the process.
- connected_raw(direction)[source]
connected(direction = ‘any’) -> bool
Should not be called directly. Returns True iff the tube is connected in the given direction.
- elf_mapping(single=True) mapping [source]
- elf_mapping(False) [mapping]
- Parameters
single (bool=True) – Whether to only return the first mapping matched, or all of them.
Returns
process.get_mapping()
with theprocess.elf()
path and single as arguments.Example
>>> p = process(['cat']) >>> p.sendline(b'meow') >>> p.recvline() b'meow\n' >>> mapping = p.elf_mapping() >>> mapping.path '...cat...' >>> mapping.perms.execute False >>> mapping.perms.write False >>> hex(mapping.address) '0x55a2abba0000' >>> mappings = p.elf_mapping(single=False) >>> len(mappings) > 1 True >>> hex(mappings[1].address) '0x55a2abba2000' >>> mappings[0].end == mappings[1].start True >>> mappings[1].perms.execute True
- get_mapping(path_value, single=True) mapping [source]
- get_mapping(path_value, False) [mapping]
- Parameters
path_value (str) – The exact path of the requested mapping, valid values are also [stack], [heap], etc..
single (bool=True) – Whether to only return the first mapping matched, or all of them.
Returns found mapping(s) in process memory according to path_value.
Example
>>> p = process(['cat']) >>> mapping = p.get_mapping('[stack]') >>> mapping.path == '[stack]' True >>> mapping.perms.execute False >>> >>> mapping = p.get_mapping('does not exist') >>> print(mapping) None >>> >>> mappings = p.get_mapping(which('cat'), single=False) >>> len(mappings) > 1 True
- heap_mapping(single=True) mapping [source]
- heap_mapping(False) [mapping]
- Parameters
single (bool=True) – Whether to only return the first mapping matched, or all of them.
Returns
process.get_mapping()
with ‘[heap]’ and single as arguments.Example
>>> p = process(['cat']) >>> p.sendline(b'meow') >>> p.recvline() b'meow\n' >>> mapping = p.heap_mapping() >>> mapping.path '[heap]' >>> mapping.perms.execute False >>> mapping.perms.write True >>> hex(mapping.address) '0x557650fae000' >>> mappings = p.heap_mapping(single=False) >>> len(mappings) 1
- leak(address, count=1)[source]
Leaks memory within the process at the specified address.
- Parameters
Example
>>> e = ELF(which('bash-static')) >>> p = process(e.path)
In order to make sure there’s not a race condition against the process getting set up…
>>> p.sendline(b'echo hello') >>> p.recvuntil(b'hello') b'hello'
Now we can leak some data!
>>> p.leak(e.address, 4) b'\x7fELF'
- lib_size(path_value) int [source]
- Parameters
path_value (str) – The exact path of the shared library
process (loaded by the) –
Returns the size of the shared library in process memory. If the library is not found, zero is returned.
Example
>>> from pwn import * >>> p = process(['cat']) >>> libc_size = p.lib_size(p.libc.path) >>> hex(libc_size) '0x1d5000' >>> libc_mappings = p.libc_mapping(single=False) >>> libc_size == (libc_mappings[-1].end - libc_mappings[0].start) True
- libc_mapping(single=True) mapping [source]
- libc_mapping(False) [mapping]
- Parameters
single (bool=True) – Whether to only return the first mapping matched, or all of them.
Returns either the first libc mapping found in process memory, or all libc mappings, depending on “single”.
Example
>>> p = process(['cat']) >>> p.sendline(b'meow') >>> p.recvline() b'meow\n' >>> mapping = p.libc_mapping() >>> mapping.path '...libc...' >>> mapping.perms.execute False >>> mapping.perms.write False >>> hex(mapping.address) '0x7fbde7fd7000' >>> >>> mappings = p.libc_mapping(single=False) >>> len(mappings) > 1 True >>> hex(mappings[1].address) '0x7fbde7ffd000' >>> mappings[0].end == mappings[1].start True >>> mappings[1].perms.execute True
- libs() dict [source]
Return a dictionary mapping the path of each shared library loaded by the process to the address it is loaded at in the process’ address space.
- maps() [mapping] [source]
Returns a list of process mappings. A mapping object has the following fields:
addr, address (addr alias), start (addr alias), end, size, perms, path, rss, pss, shared_clean, shared_dirty, private_clean, private_dirty, referenced, anonymous, swap
- perms is a permissions object, with the following fields:
read, write, execute, private, shared, string
Example
>>> p = process(['cat']) >>> p.sendline(b"meow") >>> p.recvline() b'meow\n' >>> proc_maps = open("/proc/" + str(p.pid) + "/maps", "r").readlines() >>> pwn_maps = p.maps() >>> len(proc_maps) == len(pwn_maps) True >>> checker_arr = [] >>> for proc, pwn in zip(proc_maps, pwn_maps): ... proc = proc.split(' ') ... p_addrs = proc[0].split('-') ... checker_arr.append(int(p_addrs[0], 16) == pwn.addr == pwn.address == pwn.start) ... checker_arr.append(int(p_addrs[1], 16) == pwn.end) ... checker_arr.append(pwn.size == pwn.end - pwn.start) ... checker_arr.append(pwn.perms.string == proc[1]) ... proc_path = proc[-1].strip() ... checker_arr.append(pwn.path == proc_path or (pwn.path == '[anon]' and proc_path == '')) ... >>> checker_arr == [True] * len(proc_maps) * 5 True
- musl_mapping(single=True) mapping [source]
- musl_mapping(False) [mapping]
- Parameters
single (bool=True) – Whether to only return the first mapping matched, or all of them.
Returns either the first musl mapping found in process memory, or all musl mappings, depending on “single”.
- poll(block=False) int [source]
- Parameters
block (bool) – Wait for the process to exit
Poll the exit code of the process. Will return None, if the process has not yet finished and the exit code otherwise.
- readmem(address, count=1)[source]
Leaks memory within the process at the specified address.
- Parameters
Example
>>> e = ELF(which('bash-static')) >>> p = process(e.path)
In order to make sure there’s not a race condition against the process getting set up…
>>> p.sendline(b'echo hello') >>> p.recvuntil(b'hello') b'hello'
Now we can leak some data!
>>> p.leak(e.address, 4) b'\x7fELF'
- recv_raw(numb) str [source]
Should not be called directly. Receives data without using the buffer on the object.
Unless there is a timeout or closed connection, this should always return data. In case of a timeout, it should return None, in case of a closed connection it should raise an
exceptions.EOFError
.
- send_raw(data)[source]
Should not be called directly. Sends data to the tube.
Should return
exceptions.EOFError
, if it is unable to send any more, because of a closed tube.
- shutdown_raw(direction)[source]
Should not be called directly. Closes the tube for further reading or writing.
- stack_mapping(single=True) mapping [source]
- stack_mapping(False) [mapping]
- Parameters
single (bool=True) – Whether to only return the first mapping matched, or all of them.
Returns
process.get_mapping()
with ‘[stack]’ and single as arguments.Example
>>> p = process(['cat']) >>> mapping = p.stack_mapping() >>> mapping.path '[stack]' >>> mapping.perms.execute False >>> mapping.perms.write True >>> hex(mapping.address) '0x7fffd99fe000' >>> mappings = p.stack_mapping(single=False) >>> len(mappings) 1
- vdso_mapping(single=True) mapping [source]
- vdso_mapping(False) [mapping]
- Parameters
single (bool=True) – Whether to only return the first mapping matched, or all of them.
Returns
process.get_mapping()
with ‘[vdso]’ and single as arguments.Example
>>> p = process(['cat']) >>> mapping = p.vdso_mapping() >>> mapping.path '[vdso]' >>> mapping.perms.execute True >>> mapping.perms.write False >>> hex(mapping.address) '0x7ffcf13af000' >>> mappings = p.vdso_mapping(single=False) >>> len(mappings) 1
- vvar_mapping(single=True) mapping [source]
- vvar_mapping(False) [mapping]
- Parameters
single (bool=True) – Whether to only return the first mapping matched, or all of them.
Returns
process.get_mapping()
with ‘[vvar]’ and single as arguments.Example
>>> p = process(['cat']) >>> mapping = p.vvar_mapping() >>> mapping.path '[vvar]' >>> mapping.perms.execute False >>> mapping.perms.write False >>> hex(mapping.address) '0x7ffee5f60000' >>> mappings = p.vvar_mapping(single=False) >>> len(mappings) 1
- writemem(address, data)[source]
Writes memory within the process at the specified address.
Example
Let’s write data to the beginning of the mapped memory of the ELF.
>>> context.clear(arch='i386') >>> address = 0x100000 >>> data = cyclic(32) >>> assembly = shellcraft.nop() * len(data)
Wait for one byte of input, then write the data to stdout
>>> assembly += shellcraft.write(1, address, 1) >>> assembly += shellcraft.read(0, 'esp', 1) >>> assembly += shellcraft.write(1, address, 32) >>> assembly += shellcraft.exit() >>> asm(assembly)[32:] b'j\x01[\xb9\xff\xff\xef\xff\xf7\xd1\x89\xdaj\x04X\xcd\x801\xdb\x89\xe1j\x01Zj\x03X\xcd\x80j\x01[\xb9\xff\xff\xef\xff\xf7\xd1j Zj\x04X\xcd\x801\xdbj\x01X\xcd\x80'
Assemble the binary and test it
>>> elf = ELF.from_assembly(assembly, vma=address) >>> io = elf.process() >>> _ = io.recvuntil(b'\x90') >>> _ = io.writemem(address, data) >>> io.send(b'X') >>> io.recvall() b'aaaabaaacaaadaaaeaaafaaagaaahaaa'
- property corefile[source]
Returns a corefile for the process.
If the process is alive, attempts to create a coredump with GDB.
If the process is dead, attempts to locate the coredump created by the kernel.
- property cwd[source]
Directory that the process is working in.
Example
>>> p = process('sh') >>> p.sendline(b'cd /tmp; echo AAA') >>> _ = p.recvuntil(b'AAA') >>> p.cwd == '/tmp' True >>> p.sendline(b'cd /proc; echo BBB;') >>> _ = p.recvuntil(b'BBB') >>> p.cwd '/proc'
- property libc[source]
Returns an ELF for the libc for the current process. If possible, it is adjusted to the correct address automatically.
Example:
>>> p = process("/bin/cat") >>> libc = p.libc >>> libc ELF('/lib64/libc-...so') >>> p.close()
- proc = None[source]
subprocess.Popen
object that backs this process
- property program[source]
Alias for
executable
, for backward compatibility.Example
>>> p = process('/bin/true') >>> p.executable == '/bin/true' True >>> p.executable == p.program True
- property stderr[source]
Shorthand for
self.proc.stderr
See:
process.proc
- property stdin[source]
Shorthand for
self.proc.stdin
See:
process.proc
- property stdout[source]
Shorthand for
self.proc.stdout
See:
process.proc