Commit de6bac60 authored by Luke Macken's avatar Luke Macken

Remove pyrasite.utils. Most of them were moved into the gui.

parent e9f701ff
...@@ -30,8 +30,7 @@ Authors: ...@@ -30,8 +30,7 @@ Authors:
import os import os
import warnings import warnings
import subprocess
from .utils import run
class CodeInjector(object): class CodeInjector(object):
...@@ -60,5 +59,6 @@ class CodeInjector(object): ...@@ -60,5 +59,6 @@ class CodeInjector(object):
self.filename, self.filename,
'PyGILState_Release($1)', 'PyGILState_Release($1)',
] ]
run('%sgdb -p %d -batch %s' % (self.gdb_prefix, self.pid, subprocess.call('%sgdb -p %d -batch %s' % (self.gdb_prefix, self.pid,
' '.join(["-eval-command='call %s'" % cmd for cmd in gdb_cmds]))) ' '.join(["-eval-command='call %s'" % cmd for cmd in gdb_cmds])),
shell=True)
...@@ -15,7 +15,7 @@ ...@@ -15,7 +15,7 @@
# #
# Copyright (C) 2011, 2012 Red Hat, Inc. # Copyright (C) 2011, 2012 Red Hat, Inc.
from .utils import run import subprocess
class ObjectInspector(object): class ObjectInspector(object):
...@@ -30,6 +30,7 @@ class ObjectInspector(object): ...@@ -30,6 +30,7 @@ class ObjectInspector(object):
'gdb --quiet -p %s -batch' % self.pid, 'gdb --quiet -p %s -batch' % self.pid,
'-eval-command="print (PyObject *)%s"' % address, '-eval-command="print (PyObject *)%s"' % address,
]) ])
for line in run(cmd)[1].split('\n'): p = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE)
for line in p.communicate()[0].split('\n'):
if line.startswith('$1 = '): if line.startswith('$1 = '):
return line[5:] return line[5:]
...@@ -20,7 +20,6 @@ import sys ...@@ -20,7 +20,6 @@ import sys
import argparse import argparse
from inject import CodeInjector from inject import CodeInjector
from utils import setup_logger
def main(): def main():
...@@ -44,26 +43,25 @@ def main(): ...@@ -44,26 +43,25 @@ def main():
args = parser.parse_args() args = parser.parse_args()
log = setup_logger()
try: try:
pid = int(args.pid) pid = int(args.pid)
except ValueError: except ValueError:
log.error("Error: The first argument must be a pid") print("Error: The first argument must be a pid")
sys.exit(2) sys.exit(2)
filename = args.filename filename = args.filename
if filename: if filename:
if not os.path.exists(filename): if not os.path.exists(filename):
log.error("Error: Invalid path or file doesn't exist") print("Error: Invalid path or file doesn't exist")
sys.exit(3) sys.exit(3)
else: else:
log.error("Error: The second argument must be a filename") print("Error: The second argument must be a filename")
sys.exit(4) sys.exit(4)
injector = CodeInjector(pid, verbose=args.verbose, injector = CodeInjector(pid, verbose=args.verbose,
gdb_prefix=args.gdb_prefix) gdb_prefix=args.gdb_prefix)
injector.inject(filename) injector.inject(filename)
if __name__ == '__main__': if __name__ == '__main__':
main() main()
# http://code.activestate.com/recipes/577081-humanized-representation-of-a-number-of-bytes/
from __future__ import division
def humanize_bytes(bytes, precision=1):
"""Return a humanized string representation of a number of bytes."""
abbrevs = (
(1 << 50, 'PB'),
(1 << 40, 'TB'),
(1 << 30, 'GB'),
(1 << 20, 'MB'),
(1 << 10, 'kB'),
(1, 'bytes')
)
if bytes == 1:
return '1 byte'
for factor, suffix in abbrevs:
if bytes >= factor:
break
return '%.*f %s' % (precision, bytes / factor, suffix)
# Some useful functions based on code from Will Maier's 'ideal Python script'
# https://github.com/wcmaier/python-script
#
# Copyright (c) 2010 Will Maier <willmaier@ml1.net>
#
# Permission to use, copy, modify, and distribute this software for any
# purpose with or without fee is hereby granted, provided that the above
# copyright notice and this permission notice appear in all copies.
#
# THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
# WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
# MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
# ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
# WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
# ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
# OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
import logging
import subprocess
def run(*args, **kwargs):
"""Run a subprocess.
Returns a tuple (*process*, *stdout*, *stderr*). If the *communicate*
keyword argument is True, *stdout* and *stderr* will be strings.
Otherwise, they will be None. *process* is a :class:`subprocess.Popen`
instance. By default, the path to the script itself will be used as the
executable and *args* will be passed as arguments to it.
.. note::
The value of *executable* will be prepended to *args*.
:param args: arguments to be passed to :class:`subprocess.Popen`.
:param kwargs: keyword arguments to be passed to :class:`subprocess.Popen`.
:param communicate: if True, call :meth:`subprocess.Popen.communicate`
after creating the subprocess.
:param executable: if present, the path to a program to execute instead of
this script.
"""
_kwargs = {
"stdin": subprocess.PIPE,
"stdout": subprocess.PIPE,
"stderr": subprocess.PIPE,
"shell": True,
}
communicate = kwargs.pop("communicate", True)
_kwargs.update(kwargs)
kwargs = _kwargs
process = subprocess.Popen(args, **kwargs)
if communicate is True:
stdout, stderr = process.communicate()
else:
stdout, stderr = None, None
return process, stdout, stderr
def setup_logger(verbose=False):
# NullHandler was added in Python 3.1.
try:
NullHandler = logging.NullHandler
except AttributeError:
class NullHandler(logging.Handler):
def emit(self, record):
pass
# Add a do-nothing NullHandler to the module logger to prevent "No handlers
# could be found" errors. The calling code can still add other, more useful
# handlers, or otherwise configure logging.
log = logging.getLogger('pyrasite')
log.addHandler(NullHandler())
level = logging.INFO
if verbose:
level = logging.DEBUG
handler = logging.StreamHandler()
handler.setFormatter(logging.Formatter('%(message)s'))
handler.setLevel(level)
log.addHandler(handler)
log.setLevel(level)
return log
...@@ -16,15 +16,16 @@ ...@@ -16,15 +16,16 @@
# Copyright (C) 2011, 2012 Red Hat, Inc. # Copyright (C) 2011, 2012 Red Hat, Inc.
import unittest import unittest
import subprocess
from pyrasite.inject import CodeInjector from pyrasite.inject import CodeInjector
from pyrasite.utils import run
class TestCodeInjection(unittest.TestCase): class TestCodeInjection(unittest.TestCase):
def test_injection(self): def test_injection(self):
cmd = 'python -c "import time; time.sleep(0.5)"' cmd = 'python -c "import time; time.sleep(0.5)"'
p = run(cmd, communicate=False)[0] p = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE,
stderr=subprocess.PIPE)
ci = CodeInjector(p.pid, verbose=True) ci = CodeInjector(p.pid, verbose=True)
ci.inject('pyrasite/payloads/helloworld.py') ci.inject('pyrasite/payloads/helloworld.py')
...@@ -38,7 +39,8 @@ class TestCodeInjection(unittest.TestCase): ...@@ -38,7 +39,8 @@ class TestCodeInjection(unittest.TestCase):
'snooze = lambda: time.sleep(0.5)', 'snooze = lambda: time.sleep(0.5)',
'threading.Thread(target=snooze).start()', 'threading.Thread(target=snooze).start()',
] ]
p = run('python -c "%s"' % ';'.join(cmd), communicate=False)[0] p = subprocess.Popen('python -c "%s"' % ';'.join(cmd), shell=True,
stdout=subprocess.PIPE, stderr=subprocess.PIPE)
ci = CodeInjector(p.pid, verbose=True) ci = CodeInjector(p.pid, verbose=True)
ci.inject('pyrasite/payloads/helloworld.py') ci.inject('pyrasite/payloads/helloworld.py')
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment