Module pandare.utils
Misc helper functions
Expand source code
# Helper utilities functions and classes for use in pypanda.
'''
Misc helper functions
'''
from colorama import Fore, Style
from functools import wraps
from os import devnull
from subprocess import check_call, STDOUT
from sys import platform, stdout
from threading import current_thread, main_thread
#for _find_build_path
from os import dup, getenv, environ, path
from os.path import realpath, isfile, dirname, join as pjoin
# for rr2 format
import tarfile
# Set to enable pypanda debugging
debug = False
def progress(msg):
"""
Print a message with a green "[PYPANDA]" prefix if in a tty
otherwise just print the message
"""
if stdout.isatty():
print(Fore.GREEN + '[PYPANDA] ' + Fore.RESET + Style.BRIGHT + msg +Style.RESET_ALL)
else:
print(f"[PYPANDA] {msg}")
def warn(msg):
"""
Print a message with a red "[PYPANDA]" prefix if in a tty
otherwise just print the message
"""
if stdout.isatty():
print(Fore.RED + '[PYPANDA] ' + Fore.RESET + Style.BRIGHT + msg +Style.RESET_ALL)
else:
print(f"[PYPANDA] {msg}")
def make_iso(directory, iso_path):
'''
Generate an iso from a directory
'''
with open(devnull, "w") as DEVNULL:
if platform.startswith('linux'):
check_call([
'genisoimage', '-RJ', '-max-iso9660-filenames', '-o', iso_path, directory
], stderr=STDOUT if debug else DEVNULL)
elif platform == 'darwin':
check_call([
'hdiutil', 'makehybrid', '-hfs', '-joliet', '-iso', '-o', iso_path, directory
], stderr=STDOUT if debug else DEVNULL)
else:
raise NotImplementedError("Unsupported operating system!")
def telescope(panda, cpu, val):
'''
Given a value, check if it's a pointer by seeing if we can map it to physical memory.
If so, recursively print where it points
to until
1) It points to a string (then print the string)
2) It's code (then disassembly the insn)
3) It's an invalid pointer
4) It's the 5th time we've done this, break
TODO Should use memory protections to determine string/code/data
'''
for _ in range(5): # Max chain of 5
print("-> 0x{:0>8x}".format(val), end="\t")
if val == 0:
print()
return
# Consider that val points to a string. Test and print
try:
str_data = panda.virtual_memory_read(cpu, val, 16)
except ValueError:
print()
return
str_val = ""
for d in str_data:
if d >= 0x20 and d < 0x7F:
str_val += chr(d)
else:
break
if len(str_val) > 2:
print("== \"{}\"".format(str_val))
return
data = str_data[:4] # Truncate to 4 bytes
val = int.from_bytes(data, byteorder='little')
print("-> ...")
def blocking(func):
"""
Decorator to ensure a function isn't run in the main thread
"""
@wraps(func)
def wrapper(*args, **kwargs):
assert (current_thread() is not main_thread()), "Blocking function run in main thread"
return func(*args, **kwargs)
wrapper.__blocking__ = True
wrapper.__name__ = func.__name__ + " (with async thread)"
return wrapper
def rr2_name(name):
return name if name.endswith(".rr2") else name + ".rr2"
def rr2_recording(name):
def is_gzip(name):
rr = open(name, "rb")
return rr.read(2) == b'\x1f\x8b'
rr2_filename = rr2_name(name)
if isfile(rr2_filename) and is_gzip(rr2_filename):
return True
return False
def rr2_contains_member(name, member):
rr2_filename = rr2_name(name)
contains_member = False
if rr2_recording(rr2_filename):
try:
tar = tarfile.open(rr2_filename)
tar.getmember(member)
contains_member = True
except (KeyError, IsADirectoryError, FileNotFoundError, tarfile.ReadError):
pass
return contains_member
class GArrayIterator():
'''
Iterator which will run a function on each iteration incrementing
the second argument. Useful for GArrays with an accessor function
that takes arguments of the GArray and list index. e.g., osi's
get_one_module.
'''
def __init__(self, func, garray, garray_len, cleanup_fn):
self.garray = garray
self.garray_len = garray_len
self.current_idx = 0
self.func = func
self.cleanup_func = cleanup_fn
def __iter__(self):
self.current_idx = 0
return self
def __next__(self):
if self.current_idx >= self.garray_len:
raise StopIteration
# Would need to make this configurable before using MappingIter with other types
ret = self.func(self.garray, self.current_idx)
self.current_idx += 1
return ret
def __del__(self):
self.cleanup_func(self.garray)
class plugin_list(dict):
'''
Wrapper class around list of active C plugins
'''
def __init__(self,panda):
self._panda = panda
super().__init__()
def __getitem__(self,plugin_name):
if plugin_name not in self:
self._panda.load_plugin(plugin_name)
return super().__getitem__(plugin_name)
def _find_build_dir(arch_name, find_executable=False):
'''
Internal function to return the build directory for the specified architecture
'''
system_build = "/usr/local/bin/"
python_package = pjoin(*[dirname(__file__), "data"])
local_build = realpath(pjoin(dirname(__file__), "../../../../build"))
arch_dir = f"{arch_name}-softmmu"
file_name = f"panda-system-{arch_name}" if find_executable else \
f"libpanda-{arch_name}.so"
# system path could have panda-system-X or libpanda-X.so. Others would have an arch_name - softmmu directory
pot_paths = [system_build,
pjoin(python_package, arch_dir),
pjoin(local_build, arch_dir)]
if find_executable and 'PATH' in environ:
# If we're looking for the panda executable, also search the user's path
pot_paths.extend(environ.get('PATH').split(":"))
for potential_path in pot_paths:
if isfile(pjoin(potential_path, file_name)):
if not find_executable:
# potential_path may contain [arch]-softmmu/ which
# we shouldn't return unless we're looking for an executable's
# build dir
potential_path = potential_path.replace(arch_dir, "")
return potential_path
searched_paths = "\n".join(["\t"+p for p in pot_paths])
raise RuntimeError((f"Couldn't find {file_name}\n"
f"Did you built PANDA for this architecture?\n"
f"Searched for {arch_dir}/{file_name} in:\n{searched_paths}"))
def find_build_dir(arch_name=None, find_executable=False):
'''
Find directory containing the binaries we care about (i.e., ~git/panda/build). If
find_executable is False, we're looking for [arch]-softmmu/libpanda-[arch].so. If
find_executable is True, we're looking for [arch]-softmmu/panda-system-[arch] and we'll return
the parent dir of the executable (i.e., ~/git/panda/build/x86_64-softmmu/)
We do this by searching paths in the following order:
1) Check relative to file (in the case of installed packages)
2) Check in../ ../../../build/
2) Search path if user is looking for an executable instead of a library
3) Raise RuntimeError if we find nothing
If arch_name is none, we'll search for any supported architecture and return the first
one we find.
'''
arches = ['i386', 'x86_64', 'arm', 'aarch64', 'ppc', 'mips', 'mipsel', 'mips64']
if arch_name is None:
e = None
for arch in arches:
try:
return _find_build_dir(arch, find_executable)
except RuntimeError as _e:
e = _e
if e:
raise e
elif arch_name not in arches:
raise ValueError(f"Unsupported architecture name: {arch_name}, allowed values are: {arches}")
return _find_build_dir(arch_name, find_executable)Functions
def blocking(func)-
Decorator to ensure a function isn't run in the main thread
Expand source code
def blocking(func): """ Decorator to ensure a function isn't run in the main thread """ @wraps(func) def wrapper(*args, **kwargs): assert (current_thread() is not main_thread()), "Blocking function run in main thread" return func(*args, **kwargs) wrapper.__blocking__ = True wrapper.__name__ = func.__name__ + " (with async thread)" return wrapper def find_build_dir(arch_name=None, find_executable=False)-
Find directory containing the binaries we care about (i.e., ~git/panda/build). If find_executable is False, we're looking for [arch]-softmmu/libpanda-[arch].so. If find_executable is True, we're looking for [arch]-softmmu/panda-system-[arch] and we'll return the parent dir of the executable (i.e., ~/git/panda/build/x86_64-softmmu/)
We do this by searching paths in the following order: 1) Check relative to file (in the case of installed packages) 2) Check in../ ../../../build/ 2) Search path if user is looking for an executable instead of a library 3) Raise RuntimeError if we find nothing
If arch_name is none, we'll search for any supported architecture and return the first one we find.
Expand source code
def find_build_dir(arch_name=None, find_executable=False): ''' Find directory containing the binaries we care about (i.e., ~git/panda/build). If find_executable is False, we're looking for [arch]-softmmu/libpanda-[arch].so. If find_executable is True, we're looking for [arch]-softmmu/panda-system-[arch] and we'll return the parent dir of the executable (i.e., ~/git/panda/build/x86_64-softmmu/) We do this by searching paths in the following order: 1) Check relative to file (in the case of installed packages) 2) Check in../ ../../../build/ 2) Search path if user is looking for an executable instead of a library 3) Raise RuntimeError if we find nothing If arch_name is none, we'll search for any supported architecture and return the first one we find. ''' arches = ['i386', 'x86_64', 'arm', 'aarch64', 'ppc', 'mips', 'mipsel', 'mips64'] if arch_name is None: e = None for arch in arches: try: return _find_build_dir(arch, find_executable) except RuntimeError as _e: e = _e if e: raise e elif arch_name not in arches: raise ValueError(f"Unsupported architecture name: {arch_name}, allowed values are: {arches}") return _find_build_dir(arch_name, find_executable) def make_iso(directory, iso_path)-
Generate an iso from a directory
Expand source code
def make_iso(directory, iso_path): ''' Generate an iso from a directory ''' with open(devnull, "w") as DEVNULL: if platform.startswith('linux'): check_call([ 'genisoimage', '-RJ', '-max-iso9660-filenames', '-o', iso_path, directory ], stderr=STDOUT if debug else DEVNULL) elif platform == 'darwin': check_call([ 'hdiutil', 'makehybrid', '-hfs', '-joliet', '-iso', '-o', iso_path, directory ], stderr=STDOUT if debug else DEVNULL) else: raise NotImplementedError("Unsupported operating system!") def progress(msg)-
Print a message with a green "[PYPANDA]" prefix if in a tty otherwise just print the message
Expand source code
def progress(msg): """ Print a message with a green "[PYPANDA]" prefix if in a tty otherwise just print the message """ if stdout.isatty(): print(Fore.GREEN + '[PYPANDA] ' + Fore.RESET + Style.BRIGHT + msg +Style.RESET_ALL) else: print(f"[PYPANDA] {msg}") def rr2_contains_member(name, member)-
Expand source code
def rr2_contains_member(name, member): rr2_filename = rr2_name(name) contains_member = False if rr2_recording(rr2_filename): try: tar = tarfile.open(rr2_filename) tar.getmember(member) contains_member = True except (KeyError, IsADirectoryError, FileNotFoundError, tarfile.ReadError): pass return contains_member def rr2_name(name)-
Expand source code
def rr2_name(name): return name if name.endswith(".rr2") else name + ".rr2" def rr2_recording(name)-
Expand source code
def rr2_recording(name): def is_gzip(name): rr = open(name, "rb") return rr.read(2) == b'\x1f\x8b' rr2_filename = rr2_name(name) if isfile(rr2_filename) and is_gzip(rr2_filename): return True return False def telescope(panda, cpu, val)-
Given a value, check if it's a pointer by seeing if we can map it to physical memory. If so, recursively print where it points to until 1) It points to a string (then print the string) 2) It's code (then disassembly the insn) 3) It's an invalid pointer 4) It's the 5th time we've done this, break
TODO Should use memory protections to determine string/code/data
Expand source code
def telescope(panda, cpu, val): ''' Given a value, check if it's a pointer by seeing if we can map it to physical memory. If so, recursively print where it points to until 1) It points to a string (then print the string) 2) It's code (then disassembly the insn) 3) It's an invalid pointer 4) It's the 5th time we've done this, break TODO Should use memory protections to determine string/code/data ''' for _ in range(5): # Max chain of 5 print("-> 0x{:0>8x}".format(val), end="\t") if val == 0: print() return # Consider that val points to a string. Test and print try: str_data = panda.virtual_memory_read(cpu, val, 16) except ValueError: print() return str_val = "" for d in str_data: if d >= 0x20 and d < 0x7F: str_val += chr(d) else: break if len(str_val) > 2: print("== \"{}\"".format(str_val)) return data = str_data[:4] # Truncate to 4 bytes val = int.from_bytes(data, byteorder='little') print("-> ...") def warn(msg)-
Print a message with a red "[PYPANDA]" prefix if in a tty otherwise just print the message
Expand source code
def warn(msg): """ Print a message with a red "[PYPANDA]" prefix if in a tty otherwise just print the message """ if stdout.isatty(): print(Fore.RED + '[PYPANDA] ' + Fore.RESET + Style.BRIGHT + msg +Style.RESET_ALL) else: print(f"[PYPANDA] {msg}")
Classes
class GArrayIterator (func, garray, garray_len, cleanup_fn)-
Iterator which will run a function on each iteration incrementing the second argument. Useful for GArrays with an accessor function that takes arguments of the GArray and list index. e.g., osi's get_one_module.
Expand source code
class GArrayIterator(): ''' Iterator which will run a function on each iteration incrementing the second argument. Useful for GArrays with an accessor function that takes arguments of the GArray and list index. e.g., osi's get_one_module. ''' def __init__(self, func, garray, garray_len, cleanup_fn): self.garray = garray self.garray_len = garray_len self.current_idx = 0 self.func = func self.cleanup_func = cleanup_fn def __iter__(self): self.current_idx = 0 return self def __next__(self): if self.current_idx >= self.garray_len: raise StopIteration # Would need to make this configurable before using MappingIter with other types ret = self.func(self.garray, self.current_idx) self.current_idx += 1 return ret def __del__(self): self.cleanup_func(self.garray) class plugin_list (panda)-
Wrapper class around list of active C plugins
Expand source code
class plugin_list(dict): ''' Wrapper class around list of active C plugins ''' def __init__(self,panda): self._panda = panda super().__init__() def __getitem__(self,plugin_name): if plugin_name not in self: self._panda.load_plugin(plugin_name) return super().__getitem__(plugin_name)Ancestors
- builtins.dict