Fix printing by regressing to using pwn directly

internalblue/adbcore.py

@@ -1,6 +1,12 @@
 #!/usr/bin/env python2
+import struct
+from time import sleep
 
 from future import standard_library
+
+from pwnlib import adb
+from pwnlib.exception import PwnlibException
+
 standard_library.install_aliases()
 from builtins import str
 import datetime
@@ -10,8 +16,8 @@ import random
 from internalblue import hci
 from internalblue.utils import bytes_to_hex
 
-from pwn import *
 
+from internalblue.utils.pwnlib_wrapper import log, context, u32
 from .core import InternalBlue
 
 

internalblue/cli.py

@@ -29,8 +29,11 @@
 
 
 from __future__ import print_function
+
+import socket
+import sys
 from builtins import str
-from pwn import *
+import internalblue.utils.pwnlib_wrapper as pwnlib
 import os
 import traceback
 import argparse
@@ -62,7 +65,7 @@ def print_banner():
 
 type <help> for usage information!\n\n"""
     for line in banner:
-        term.output(text.blue(line))
+        pwnlib.term.output(pwnlib.text.blue(line))
 
 def commandLoop(internalblue, init_commands=None):
     cmdstack = init_commands.split(';')[::-1] if init_commands else None
@@ -72,37 +75,37 @@ def commandLoop(internalblue, init_commands=None):
             if cmdstack:
                 cmdline = cmdstack.pop().strip()
             else:
-                cmdline = term.readline.readline(prompt='> ').strip().decode('utf-8')
+                cmdline = pwnlib.term.readline.readline(prompt='> ').strip().decode('utf-8')
             cmdword = cmdline.split(' ')[0].split('=')[0]
             if(cmdword == ''):
                 continue
-            log.debug("Command Line: [[" + cmdword + "]] " + cmdline)
+            pwnlib.log.debug("Command Line: [[" + cmdword + "]] " + cmdline)
             matching_cmd = cmds.findCmd(cmdword)
             if matching_cmd == None:
-                log.warn("Command unknown: " + cmdline)
+                pwnlib.log.warn("Command unknown: " + cmdline)
                 continue
             cmd_instance = matching_cmd(cmdline, internalblue)
             if(not cmd_instance.work()):
-                log.warn("Command failed: " + str(cmd_instance))
+                pwnlib.log.warn("Command failed: " + str(cmd_instance))
         except ValueError as e:
-            log.warn("commandLoop: ValueError: " + str(e))
+            pwnlib.log.warn("commandLoop: ValueError: " + str(e))
             raise
         except KeyboardInterrupt:
             if(cmd_instance != None):
                 cmd_instance.abort_cmd()
             else:
-                log.info("Got Ctrl-C; exiting...")
+                pwnlib.log.info("Got Ctrl-C; exiting...")
                 internalblue.exit_requested = True
                 break
         except AssertionError as e:
             raise
         except socket.error as e:
             if e.args == (1, "Operation not permitted"):
-                log.critical("Received an 'Operation not permitted' socket.error, you might need root for the command '{}'".format(cmdline))
-                log.critical(traceback.format_exc())
+                pwnlib.log.critical("Received an 'Operation not permitted' socket.error, you might need root for the command '{}'".format(cmdline))
+                pwnlib.log.critical(traceback.format_exc())
         except Exception as e:
             internalblue.exit_requested = True      # Make sure all threads terminate
-            log.critical("Uncaught exception (%s). Abort." % str(e))
+            pwnlib.log.critical("Uncaught exception (%s). Abort." % str(e))
             print(traceback.format_exc())
             raise
         cmd_instance = None
@@ -146,8 +149,8 @@ def internalblue_cli(argv, args=None):
     for cmd in cmds.getCmdList():
         for keyword in cmd.keywords:
             cmd_keywords.append(keyword)
-    readline_completer = term.completer.LongestPrefixCompleter(words=cmd_keywords)
-    term.readline.set_completer(readline_completer)
+    readline_completer = pwnlib.term.completer.LongestPrefixCompleter(words=cmd_keywords)
+    pwnlib.term.readline.set_completer(readline_completer)
 
 
 
@@ -218,18 +221,18 @@ def internalblue_cli(argv, args=None):
         elif args.device:
             matching_devices = [ dev for dev in devices if dev[1] == args.device]
             if len(matching_devices) > 1:
-                log.critical("Found multiple matching devices")
+                pwnlib.log.critical("Found multiple matching devices")
                 exit(-1)
             elif len(matching_devices) == 1:
-                log.info("Found device is: {}".format(matching_devices[0]))
+                pwnlib.log.info("Found device is: {}".format(matching_devices[0]))
                 device = matching_devices[0]
             else:
-                log.critical("No matching devices found")
+                pwnlib.log.critical("No matching devices found")
                 exit(-1)
         elif len(devices) == 1:
             device = devices[0]
         else:
-            i = options('Please specify device:',  [d[2] for d in devices], 0)
+            i = pwnlib.options('Please specify device:',  [d[2] for d in devices], 0)
             device = devices[i]
 
         # Setup device
@@ -238,16 +241,16 @@ def internalblue_cli(argv, args=None):
 
         # Restore readline history:
         if os.path.exists(reference.data_directory + "/" + HISTFILE):
-            readline_history = read(reference.data_directory + "/" + HISTFILE)
-            term.readline.history = readline_history.split(b'\n')
+            readline_history = pwnlib.read(reference.data_directory + "/" + HISTFILE)
+            pwnlib.term.readline.history = readline_history.split(b'\n')
 
         # Connect to device
         if not reference.connect():
-            log.critical("No connection to target device.")
+            pwnlib.log.critical("No connection to target device.")
             exit(-1)
 
         # Enter command loop (runs until user quits)
-        log.info("Starting commandLoop for reference {}".format(reference))
+        pwnlib.log.info("Starting commandLoop for reference {}".format(reference))
         commandLoop(reference, init_commands=args.commands)
 
         # shutdown connection
@@ -260,7 +263,7 @@ def internalblue_cli(argv, args=None):
         # f.close()
 
     # Cleanup
-    log.info("Goodbye")
+    pwnlib.log.info("Goodbye")
 
 
 if __name__ == "__main__":

internalblue/cmds.py

@@ -24,11 +24,12 @@
 #   Software.
 
 from __future__ import print_function
+
+import re
 from builtins import str
 from builtins import hex
 from builtins import range
 from builtins import object
-from pwn import *
 import os
 import sys
 import inspect
@@ -40,12 +41,16 @@ import struct
 import time
 import select
 import json
+
+from pwnlib.context import context
+from pwnlib.asm import disasm, asm
+from pwnlib.exception import PwnlibException
+from pwnlib.ui import yesno
+from pwnlib.util.fiddling import isprint
+
+from internalblue.utils.pwnlib_wrapper import log, flat, read, p8, p32, u32, p16
 from internalblue.utils import bytes_to_hex
 
-
-
-
-
 try:
     from typing import List, Optional, Any, TYPE_CHECKING, Tuple, Type
 
@@ -191,7 +196,7 @@ class Cmd(object):
                 dumped_sections[section.start_addr] = self.readMem(section.start_addr, section.size(), self.progress_log, bytes_done, bytes_total)
                 bytes_done += section.size()
             self.progress_log.success("Received Data: complete")
-            Cmd.memory_image = fit(dumped_sections, filler='\x00')
+            Cmd.memory_image = flat(dumped_sections, filler='\x00')
             f = open(self.memory_image_template_filename, 'wb')
             f.write(Cmd.memory_image)
             f.close()

internalblue/core.py

@@ -26,7 +26,18 @@
 #   Software.
 
 from __future__ import division
+
+import socket
+import struct
+
 from future import standard_library
+
+import pwnlib
+from pwnlib.asm import asm
+from pwnlib.exception import PwnlibException
+from pwnlib.util.fiddling import bits, unbits
+from .utils.pwnlib_wrapper import p16, p8, u32, u16, p32, u8
+
 standard_library.install_aliases()
 from builtins import hex
 from builtins import str
@@ -35,7 +46,6 @@ from builtins import object
 from past.utils import old_div
 from abc import ABCMeta, abstractmethod
 
-from pwn import *
 from .fw.fw import Firmware
 import datetime
 import time
@@ -46,6 +56,8 @@ from .objects.connection_information import ConnectionInformation
 from future.utils import with_metaclass
 from internalblue.utils import bytes_to_hex
 
+from internalblue.utils.pwnlib_wrapper import log, context, flat
+
 try:
     from typing import List, Optional, Any, TYPE_CHECKING, Tuple, Union, NewType, Callable
     from internalblue import Address, Record, Task, HCI_CMD, FilterFunction, ConnectionNumber, ConnectionDict, \
@@ -371,7 +383,7 @@ class InternalBlue(with_metaclass(ABCMeta, object)):
 
             # Check if this was the last packet
             if len(self.tracepoint_memdump_parts) == self.fw.TRACEPOINT_RAM_DUMP_PKT_COUNT:
-                dump = fit(self.tracepoint_memdump_parts)
+                dump = flat(self.tracepoint_memdump_parts)
                 #TODO: use this to start qemu
                 filename = self.data_directory + "/" + "internalblue_tracepoint_0x%x_%s.bin" % (self.tracepoint_memdump_address, datetime.datetime.now().strftime("%Y-%m-%d_%H-%M-%S"))
                 log.info("Captured Ram Dump for Tracepoint 0x%x to %s" % (self.tracepoint_memdump_address, filename))
@@ -747,7 +759,7 @@ class InternalBlue(with_metaclass(ABCMeta, object)):
             if hasattr(self, 'test_failed'):
                 raise self.test_failed
             return None
-        except queue.Full:
+        except queue2k.Full:
             log.warn("sendHciCommand: send queue is full!")
             return None
 
@@ -764,7 +776,7 @@ class InternalBlue(with_metaclass(ABCMeta, object)):
         try:
             self.sendQueue.put((h4type, data, None, None), timeout=timeout)
             return True
-        except queue.Full:
+        except queue2k.Full:
             log.warn("sendH4: send queue is full!")
             return False
 
@@ -864,7 +876,7 @@ class InternalBlue(with_metaclass(ABCMeta, object)):
                 response_check = self.sendHciCommand(0xfc4d, p32(read_addr) + p8(blocksize))
                 if response != response_check:
                     log.debug("readMem: double checking response failed at 0x%x! retry..." % read_addr)
-                    sleep(0.3)
+                    time.sleep(0.3)
                     retry = retry - 1
                     continue
 
@@ -1045,7 +1057,7 @@ class InternalBlue(with_metaclass(ABCMeta, object)):
         return True
 
     def getPatchramState(self):
-        # type: () -> Tuple[List[Optional[int]], List[Any], List[Any]]
+        # type: () -> Union[bool, Tuple[List[Optional[int]], List[Union[Union[int, bytes, None], Any]], list]]
         """
         Retrieves the current state of the patchram unit. The return value
         is a tuple containing 3 lists which are indexed by the slot number:

internalblue/hci.py

@@ -29,7 +29,10 @@ from __future__ import absolute_import
 from builtins import hex
 from builtins import range
 from builtins import object
-from pwn import *
+
+from internalblue.utils.pwnlib_wrapper import p8, u16, p16, unbits, bits_str, u8, bits, p32, u32
+from internalblue.utils.pwnlib_wrapper import log
+from pwnlib.util.packing import flat
 
 HCI_UART_TYPE_CLASS = {}
 
@@ -965,7 +968,7 @@ class StackDumpReceiver(object):
         log.debug("Stack dump handling addr %08x", addr-self.memdump_addr)
 
     def finishStackDump(self):
-        dump = fit(self.memdumps)
+        dump = flat(self.memdumps)
         log.warn("Stack dump @0x%08x written to %s!" % (self.memdump_addr, self.stack_dump_filename))
         f = open(self.stack_dump_filename, "wb")
         f.write(dump)

internalblue/hcicore.py

@@ -1,6 +1,10 @@
 #!/usr/bin/env python2
 
 from __future__ import absolute_import
+
+import socket
+import struct
+
 from future import standard_library
 standard_library.install_aliases()
 from builtins import str
@@ -8,7 +12,7 @@ from builtins import zip
 from builtins import range
 import subprocess
 import datetime
-from pwn import *
+from internalblue.utils.pwnlib_wrapper import log, context, p32, u16, p16, u32
 import fcntl
 from .core import InternalBlue
 from . import hci

internalblue/ioscore.py

@@ -1,6 +1,9 @@
 #!/usr/bin/env python2
 
 from __future__ import absolute_import
+
+import struct
+
 from future import standard_library
 standard_library.install_aliases()
 from builtins import str
@@ -8,7 +11,7 @@ import socket
 import queue as queue2k
 from . import hci
 
-from pwn import *
+from internalblue.utils.pwnlib_wrapper import log, context
 
 from .core import InternalBlue
 

internalblue/macoscore.py

@@ -1,6 +1,10 @@
 #!/usr/bin/env python2
 
 from __future__ import absolute_import
+
+import random
+import time
+
 from future import standard_library
 standard_library.install_aliases()
 from builtins import str
@@ -8,14 +12,16 @@ import socket
 import queue as queue2k
 from . import hci
 
-from pwn import *
 
+from internalblue.utils.pwnlib_wrapper import log, context, p8
 from .core import InternalBlue
 
 import binascii
 import os
 filepath = os.path.dirname(os.path.abspath(__file__))
 
+IOBE = None
+
 class macOSCore(InternalBlue):
 
     def __init__(self, queue_size=1000, btsnooplog_filename='btsnoop.log', log_level='info', fix_binutils='True', data_directory=".", replay=False):

internalblue/objects/connection_information.py

@@ -1,5 +1,5 @@
 from builtins import object
-from pwnlib.util.packing import u32, u16, u8
+from internalblue.utils.pwnlib_wrapper import u32, u16, u8
 
 
 class ConnectionInformation(object):

internalblue/testcore.py

@@ -2,13 +2,21 @@
 
 from __future__ import print_function
 from __future__ import absolute_import
+
+import os
+import random
+import time
+
 from future import standard_library
+
+from pwnlib.context import context
+
 standard_library.install_aliases()
 import socket
 import queue as queue2k
 from . import hci
 
-from pwn import *
+from internalblue.utils.pwnlib_wrapper import log, p8
 
 from .core import InternalBlue
 import binascii

internalblue/utils.py

@@ -1,3 +0,0 @@
-def bytes_to_hex(bytes):
-    # type: (bytearray) -> str
-    return ''.join(format(x, '02x') for x in bytearray(bytes))

internalblue/utils/__init__.py

@@ -0,0 +1,11 @@
+
+
+#from pwnlib.util.packing import *
+
+def bytes_to_hex(bytes):
+    # type: (bytearray) -> str
+    return ''.join(format(x, '02x') for x in bytearray(bytes))
+
+
+
+

internalblue/utils/pwnlib_wrapper.py

@@ -0,0 +1,91 @@
+"""
+The following proxies various utilities from pwnlibs by explicitly importing them
+To replace a "from pwn import *" remove it and let your IDE highlight all missing methods (Hint: F2 in PyCharm goes to next error)
+import the missing (and only the missing!) methods from this module, e.g. with "from internalblue.utils import term, read, log, text, options"
+In some cases like "from pwn import socket" this just imports another module.
+Use an IPython shell to run "from pwn import *" and check where some method/module actually comes from and either import it directly or add it to this module
+"""
+
+
+
+# Imports that used to be imported via 'from pwn import *'
+import pwnlib
+from pwnlib import term
+from pwnlib.util import iters
+from pwnlib.util.misc import read
+from pwnlib.context import context
+
+#TODO: Logging via pwnlib doesn't work yet, so for now it is still used via pwn
+# import pwnlib.log
+# pwnlib.log.install_default_handler()
+# log = pwnlib.log.getLogger('internalbue')
+
+from pwn import log
+
+
+from pwnlib.term import text
+from pwnlib.ui import options, yesno
+from pwnlib.util.packing import flat
+from pwnlib.asm import disasm
+from pwnlib.util.fiddling import isprint, unbits, bits_str, bits
+
+
+
+
+"""
+The packers like u8 are generated in a fairly convoluted way that breaks IDE introspection.
+The following code remedies this by:
+- Explicitly defining a stub function with type annotations
+- Generating all the packers like pwnlibs would
+- Only if if the current module already has the name of the packer as an attribute (i.e. has a stub function defined) it will be replaced with the pwnlibs version
+
+This means:
+- All import issues in the rest of the code are genuine as the imports are only available if an explicit stub function is added
+- The functions can be easily replaced by just implementing them and removing the for loop at the end
+
+"""
+
+# Imports needed for this hack
+from pwnlib.util.packing import ops, sizes, make_multi
+import sys
+try:
+    from typing import Union, Optional, Literal
+    endianess = Union[Literal['big']]
+
+except ImportError:
+    pass
+mod = sys.modules[__name__]
+
+
+_DEFINES = ['u8', 'p8', 'u32', 'u16', 'p32']
+
+
+
+def u8(data, endian = None):
+    # type: (bytes, Optional[endianess]) -> int
+    pass
+
+def p8(number, endian = None):
+    # type: (int, Optional[endianess]) -> bytes
+    pass
+def u16(data, endian = None):
+    # type: (bytes, Optional[endianess]) -> int
+    pass
+
+def p16(number, endian = None):
+    # type: (int, Optional[endianess]) -> bytes
+    pass
+
+def u32(data, endian = None):
+    # type: (bytes, Optional[endianess]) -> int
+    pass
+
+def p32(number, endian = None):
+    # type: (int, Optional[endianess]) -> bytes
+    pass
+
+
+for op, size in iters.product(ops, sizes):
+    name, routine = make_multi(op, size)
+    if hasattr(mod, name):
+        setattr(mod, name, routine)

setup.py

@@ -15,7 +15,8 @@ setup(name='internalblue',
           'pyelftools',
       ],
       extras_require={
-	"macoscore": ["pyobjc"],
+            "macoscore": ["pyobjc"],
+            "ipython": ["IPython"]
 	},
       entry_points = {
         'console_scripts': ['internalblue=internalblue.cli:internalblue_cli']