# Copyright (c) 2012 The Native Client Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

import operator
import os
import re
import struct
import subprocess
import sys
import unittest
import xml.etree.ElementTree

import gdb_rsp


if sys.platform == 'win32':
  RETURNCODE_KILL = -9
else:
  RETURNCODE_KILL = -9 & 0xff


NACL_SIGILL = 4
NACL_SIGTRAP = 5
NACL_SIGSEGV = 11

# GDB specific errno constants.
GDB_EBADF = 9


# These are set up by Main().
ARCH = None
NM_TOOL = None
SEL_LDR_COMMAND = None


def AssertEquals(x, y):
  if x != y:
    raise AssertionError('%r != %r' % (x, y))


def DecodeHex(data):
  assert len(data) % 2 == 0, data
  return ''.join([chr(int(data[index * 2 : (index + 1) * 2], 16))
                  for index in xrange(len(data) / 2)])


def EncodeHex(data):
  return ''.join('%02x' % ord(byte) for byte in data)


def DecodeEscaping(data):
  ret = ''
  last = None
  repeat = False
  escape = False
  for byte in data:
    if escape:
      last = chr(ord(byte) ^ 0x20)
      ret += last
      escape = False
    elif repeat:
      count = ord(byte) - 29
      assert count >= 3 and count <= 97
      assert byte != '$' and byte != '#'
      ret += last * count
      repeat = False
    elif byte == '}':
      escape = True
    elif byte == '*':
      assert last is not None
      repeat = True
    else:
      ret += byte
      last = byte
  return ret


X86_32_REG_DEFS = [
    ('eax', 'I'),
    ('ecx', 'I'),
    ('edx', 'I'),
    ('ebx', 'I'),
    ('esp', 'I'),
    ('ebp', 'I'),
    ('esi', 'I'),
    ('edi', 'I'),
    ('eip', 'I'),
    ('eflags', 'I'),
    ('cs', 'I'),
    ('ss', 'I'),
    ('ds', 'I'),
    ('es', 'I'),
    ('fs', 'I'),
    ('gs', 'I'),
]


X86_64_REG_DEFS = [
    ('rax', 'Q'),
    ('rbx', 'Q'),
    ('rcx', 'Q'),
    ('rdx', 'Q'),
    ('rsi', 'Q'),
    ('rdi', 'Q'),
    ('rbp', 'Q'),
    ('rsp', 'Q'),
    ('r8', 'Q'),
    ('r9', 'Q'),
    ('r10', 'Q'),
    ('r11', 'Q'),
    ('r12', 'Q'),
    ('r13', 'Q'),
    ('r14', 'Q'),
    ('r15', 'Q'),
    ('rip', 'Q'),
    ('eflags', 'I'),
    ('cs', 'I'),
    ('ss', 'I'),
    ('ds', 'I'),
    ('es', 'I'),
    ('fs', 'I'),
    ('gs', 'I'),
]


ARM_REG_DEFS = ([('r%d' % regno, 'I') for regno in xrange(16)]
                + [('cpsr', 'I')])


MIPS_REG_DEFS = [
     ('zero', 'I'),
     ('at', 'I'),
     ('v0', 'I'),
     ('v1', 'I'),
     ('a0', 'I'),
     ('a1', 'I'),
     ('a2', 'I'),
     ('a3', 'I'),
     ('t0', 'I'),
     ('t1', 'I'),
     ('t2', 'I'),
     ('t3', 'I'),
     ('t4', 'I'),
     ('t5', 'I'),
     ('t6', 'I'),
     ('t7', 'I'),
     ('s0', 'I'),
     ('s1', 'I'),
     ('s2', 'I'),
     ('s3', 'I'),
     ('s4', 'I'),
     ('s5', 'I'),
     ('s6', 'I'),
     ('s7', 'I'),
     ('t8', 'I'),
     ('t9', 'I'),
     ('k0', 'I'),
     ('k1', 'I'),
     ('global_ptr', 'I'),
     ('stack_ptr', 'I'),
     ('frame_ptr', 'I'),
     ('return_addr', 'I'),
     ('prog_ctr', 'I'),
]


REG_DEFS = {
    'x86-32': X86_32_REG_DEFS,
    'x86-64': X86_64_REG_DEFS,
    'arm': ARM_REG_DEFS,
    'mips32': MIPS_REG_DEFS,
    }


SP_REG = {
    'x86-32': 'esp',
    'x86-64': 'rsp',
    'arm': 'r13',
    'mips32': 'stack_ptr',
    }


IP_REG = {
    'x86-32': 'eip',
    'x86-64': 'rip',
    'arm': 'r15',
    'mips32': 'prog_ctr',
    }


X86_TRAP_FLAG = 1 << 8

# RESET_X86_FLAGS_VALUE is what ASM_WITH_REGS() resets the x86 flags
# to.  Copied from tests/common/register_set.h.
RESET_X86_FLAGS_VALUE = (1 << 2) | (1 << 6)
KNOWN_X86_FLAGS_MASK = (1<<0) | (1<<2) | (1<<6) | (1<<7) | (1<<11) | (1<<8)

# These are the only ARM CPSR bits that user code and untrusted code
# can read and modify, excluding the IT bits which are for Thumb-2
# (for If-Then-Else instructions).  Copied from
# tests/common/register_set.h.
ARM_USER_CPSR_FLAGS_MASK = (
    (1<<31) | # N
    (1<<30) | # Z
    (1<<29) | # C
    (1<<28) | # V
    (1<<27) | # Q
    (1<<19) | (1<<18) | (1<<17) | (1<<16)) # GE bits


def UsingQemu():
  return os.path.basename(SEL_LDR_COMMAND[0]).startswith('run_under_qemu_')


def MainNexe():
  # Assume that the last parameter ending in .nexe is the main nexe.
  for arg in reversed(SEL_LDR_COMMAND):
    if arg.endswith('.nexe'):
      return arg
  assert False


def IrtNexe():
  # Assume that the irt starts with 'irt' and ends with '.nexe'.
  for arg in reversed(SEL_LDR_COMMAND):
    basename = os.path.basename(arg)
    if basename.startswith('irt_') and basename.endswith('.nexe'):
      return arg
  return None


def DecodeRegs(reply):
  defs = REG_DEFS[ARCH]
  names = [reg_name for reg_name, reg_fmt in defs]
  fmt = ''.join([reg_fmt for reg_name, reg_fmt in defs])

  values = struct.unpack_from(fmt, DecodeHex(reply))
  return dict(zip(names, values))


def EncodeRegs(regs):
  defs = REG_DEFS[ARCH]
  names = [reg_name for reg_name, reg_fmt in defs]
  fmt = ''.join([reg_fmt for reg_name, reg_fmt in defs])

  values = [regs[r] for r in names]
  return EncodeHex(struct.pack(fmt, *values))


def PopenDebugStub(test):
  gdb_rsp.EnsurePortIsAvailable()
  return subprocess.Popen(SEL_LDR_COMMAND + ['-g', test])


def KillProcess(process):
  if process.returncode is not None:
    # kill() won't work if we've already wait()'ed on the process.
    return
  try:
    process.kill()
  except OSError:
    if sys.platform == 'win32':
      # If process is already terminated, kill() throws
      # "WindowsError: [Error 5] Access is denied" on Windows.
      pass
    else:
      raise
  process.wait()


def ChangeReg(conn, name, func):
  regs = DecodeRegs(conn.RspRequest('g'))
  regs[name] = func(regs[name])
  return conn.RspRequest('G' + EncodeRegs(regs))


def SkipBreakpoint(connection, stop_reply):
  # Skip past the faulting instruction in debugger_test.c's
  # breakpoint() function.
  regs = DecodeRegs(connection.RspRequest('g'))
  if ARCH in ('x86-32', 'x86-64'):
    AssertReplySignal(stop_reply, NACL_SIGSEGV)
    # Skip past the single-byte HLT instruction.
    regs[IP_REG[ARCH]] += 1
  elif ARCH == 'arm':
    AssertReplySignal(stop_reply, NACL_SIGILL)
    bundle_size = 16
    assert regs['r15'] % bundle_size == 0, regs['r15']
    regs['r15'] += bundle_size
  elif ARCH == 'mips32':
    AssertReplySignal(stop_reply, NACL_SIGTRAP)
    bundle_size = 16
    assert regs['prog_ctr'] % bundle_size == 0, regs['prog_ctr']
    regs['prog_ctr'] += bundle_size
  else:
    raise AssertionError('Unknown architecture')
  AssertEquals(connection.RspRequest('G' + EncodeRegs(regs)), 'OK')


class LaunchDebugStub(object):

  def __init__(self, test):
    self._proc = PopenDebugStub(test)

  def __enter__(self):
    try:
      return gdb_rsp.GdbRspConnection()
    except:
      KillProcess(self._proc)
      raise

  def __exit__(self, exc_type, exc_value, traceback):
    KillProcess(self._proc)


def GetSymbols():
  assert '-f' in SEL_LDR_COMMAND
  nexe_filename = SEL_LDR_COMMAND[SEL_LDR_COMMAND.index('-f') + 1]
  symbols = {}
  proc = subprocess.Popen([NM_TOOL, '--format=posix', nexe_filename],
                          stdout=subprocess.PIPE)
  for line in proc.stdout:
    match = re.match('(\S+) [TtWwBbDd] ([0-9a-fA-F]+)', line)
    if match is not None:
      name = match.group(1)
      addr = int(match.group(2), 16)
      symbols[name] = addr
  result = proc.wait()
  assert result == 0, result
  return symbols


def ParseThreadStopReply(reply):
  match = re.match('T([0-9a-f]{2})thread:([0-9a-f]+);$', reply)
  if match is None:
    raise AssertionError('Bad thread stop reply: %r' % reply)
  return {'signal': int(match.group(1), 16),
          'thread_id': int(match.group(2), 16)}


def AssertReplySignal(reply, signal):
  AssertEquals(ParseThreadStopReply(reply)['signal'], signal)


def ReadMemory(connection, address, size):
  reply = connection.RspRequest('m%x,%x' % (address, size))
  assert not reply.startswith('E'), reply
  return DecodeHex(reply)


def ReadUint32(connection, address):
  return struct.unpack('I', ReadMemory(connection, address, 4))[0]


def SingleSteppingWorks():
  # Single-stepping is not yet supported on ARM and MIPS.
  # TODO(eaeltsin):
  #   http://code.google.com/p/nativeclient/issues/detail?id=2911
  return ARCH in ('x86-32', 'x86-64')


class DebugStubTest(unittest.TestCase):

  def test_initial_breakpoint(self):
    # Any arguments to the nexe would work here because we are only
    # testing that we get a breakpoint at the _start entry point.
    with LaunchDebugStub('test_getting_registers') as connection:
      reply = connection.RspRequest('?')
      AssertReplySignal(reply, NACL_SIGTRAP)

  def CheckTargetXml(self, connection):
    reply = connection.RspRequest('qXfer:features:read:target.xml:0,fff')
    self.assertEquals(reply[0], 'l')
    # Just check that we are given parsable XML.
    xml.etree.ElementTree.fromstring(reply[1:])

  # Test that we can fetch register values.
  # This check corresponds to the last instruction of debugger_test.c
  def CheckReadRegisters(self, connection):
    registers = DecodeRegs(connection.RspRequest('g'))
    if ARCH == 'x86-32':
      self.assertEquals(registers['eax'], 0x11000022)
      self.assertEquals(registers['ebx'], 0x22000033)
      self.assertEquals(registers['ecx'], 0x33000044)
      self.assertEquals(registers['edx'], 0x44000055)
      self.assertEquals(registers['esi'], 0x55000066)
      self.assertEquals(registers['edi'], 0x66000077)
      self.assertEquals(registers['ebp'], 0x77000088)
      self.assertEquals(registers['esp'], 0x88000099)
      self.assertEquals(registers['eflags'] & KNOWN_X86_FLAGS_MASK,
                        RESET_X86_FLAGS_VALUE)
      self.assertEquals(registers['cs'], 0x00000000)
      self.assertEquals(registers['ds'], 0x00000000)
      self.assertEquals(registers['es'], 0x00000000)
      self.assertEquals(registers['fs'], 0x00000000)
      self.assertEquals(registers['gs'], 0x00000000)
      self.assertEquals(registers['ss'], 0x00000000)
    elif ARCH == 'x86-64':
      self.assertEquals(registers['rax'], 0x1100000000000022)
      self.assertEquals(registers['rbx'], 0x2200000000000033)
      self.assertEquals(registers['rcx'], 0x3300000000000044)
      self.assertEquals(registers['rdx'], 0x4400000000000055)
      self.assertEquals(registers['rsi'], 0x5500000000000066)
      self.assertEquals(registers['rdi'], 0x6600000000000077)
      self.assertEquals(registers['r8'],  0x7700000000000088)
      self.assertEquals(registers['r9'],  0x8800000000000099)
      self.assertEquals(registers['r10'], 0x99000000000000aa)
      self.assertEquals(registers['r11'], 0xaa000000000000bb)
      self.assertEquals(registers['r12'], 0xbb000000000000cc)
      self.assertEquals(registers['r13'], 0xcc000000000000dd)
      self.assertEquals(registers['r14'], 0xdd000000000000ee)
      self.assertEquals(registers['rsp'], registers['r15'] + 0x12300321)
      self.assertEquals(registers['rbp'], registers['r15'] + 0x23400432)
      self.assertEquals(registers['eflags'] & KNOWN_X86_FLAGS_MASK,
                        RESET_X86_FLAGS_VALUE)
      self.assertEquals(registers['cs'], 0x00000000)
      self.assertEquals(registers['ds'], 0x00000000)
      self.assertEquals(registers['es'], 0x00000000)
      self.assertEquals(registers['fs'], 0x00000000)
      self.assertEquals(registers['gs'], 0x00000000)
      self.assertEquals(registers['ss'], 0x00000000)
    elif ARCH == 'arm':
      self.assertEquals(registers['r0'], 0x00000001)
      self.assertEquals(registers['r1'], 0x10000002)
      self.assertEquals(registers['r2'], 0x20000003)
      self.assertEquals(registers['r3'], 0x30000004)
      self.assertEquals(registers['r4'], 0x40000005)
      self.assertEquals(registers['r5'], 0x50000006)
      self.assertEquals(registers['r6'], 0x60000007)
      self.assertEquals(registers['r7'], 0x70000008)
      self.assertEquals(registers['r8'], 0x80000009)
      self.assertEquals(registers['r9'], 0x00000000)
      self.assertEquals(registers['r10'], 0xa000000b)
      self.assertEquals(registers['r11'], 0xb000000c)
      self.assertEquals(registers['r12'], 0xc000000d)
      # Ensure sp is masked.
      self.assertEquals(registers['r13'], 0x12345678)
      self.assertEquals(registers['r14'], 0xe000000f)
      # Only the upper 4 bits of cspr are visible.
      self.assertEquals(registers['cpsr'] & ARM_USER_CPSR_FLAGS_MASK,
                        (1 << 29))
    elif ARCH == 'mips32':
      # We skip zero register because it cannot be set.
      self.assertEquals(registers['at'], 0x11000220)
      self.assertEquals(registers['v0'], 0x22000330)
      self.assertEquals(registers['v1'], 0x33000440)
      self.assertEquals(registers['a0'], 0x44000550)
      self.assertEquals(registers['a1'], 0x55000660)
      self.assertEquals(registers['a2'], 0x66000770)
      self.assertEquals(registers['a3'], 0x77000880)
      self.assertEquals(registers['t0'], 0x88000990)
      self.assertEquals(registers['t1'], 0x99000aa0)
      self.assertEquals(registers['t2'], 0xaa000bb0)
      self.assertEquals(registers['t3'], 0xbb000cc0)
      self.assertEquals(registers['t4'], 0xcc000dd0)
      self.assertEquals(registers['t5'], 0xdd000ee0)
      self.assertEquals(registers['t6'], 0x0ffffff0)
      self.assertEquals(registers['t7'], 0x3fffffff)
      # Skip t8 because it cannot be set by untrusted code.
      self.assertEquals(registers['s0'], 0x11100222)
      self.assertEquals(registers['s1'], 0x22200333)
      self.assertEquals(registers['s2'], 0x33300444)
      self.assertEquals(registers['s3'], 0x44400555)
      self.assertEquals(registers['s4'], 0x55500666)
      self.assertEquals(registers['s5'], 0x66600777)
      self.assertEquals(registers['s6'], 0x77700888)
      self.assertEquals(registers['s7'], 0x88800999)
      self.assertEquals(registers['t9'], 0xaaa00bbb)
      # Skip k0 and k1 registers, since they can be changed by kernel.
      self.assertEquals(registers['global_ptr'],  0xddd00eee)
      self.assertEquals(registers['stack_ptr'],   0x2ee00fff)
      self.assertEquals(registers['frame_ptr'],   0xfff00000)
      self.assertEquals(registers['return_addr'], 0x0a0a0a0a)
    else:
      raise AssertionError('Unknown architecture')

    expected_fault_addr = GetSymbols()['fault_addr']
    if ARCH == 'x86-64':
      expected_fault_addr += registers['r15']
    self.assertEquals(registers[IP_REG[ARCH]], expected_fault_addr)

  # Test that we can write registers.
  def CheckWriteRegisters(self, connection):
    if ARCH == 'x86-32':
      reg_name = 'edx'
    elif ARCH == 'x86-64':
      reg_name = 'rdx'
    elif ARCH == 'arm':
      reg_name = 'r0'
    elif ARCH == 'mips32':
      reg_name = 'a0'
    else:
      raise AssertionError('Unknown architecture')

    # Read registers.
    regs = DecodeRegs(connection.RspRequest('g'))

    # Change a register.
    regs[reg_name] += 1
    new_value = regs[reg_name]

    # Write registers.
    self.assertEquals(connection.RspRequest('G' + EncodeRegs(regs)), 'OK')

    # Read registers. Check for a new value.
    regs = DecodeRegs(connection.RspRequest('g'))
    self.assertEquals(regs[reg_name], new_value)

    # TODO: Resume execution and check that changing the registers really
    # influenced the program's execution. This would require changing
    # debugger_test.c.

  def CheckReadOnlyRegisters(self, connection):
    if ARCH == 'x86-32':
      read_only_regs = []
      read_only_zero_regs = ['cs', 'ds', 'es', 'fs', 'gs', 'ss']
    elif ARCH == 'x86-64':
      read_only_regs = ['r15']
      read_only_zero_regs = ['cs', 'ds', 'es', 'fs', 'gs', 'ss']
    elif ARCH == 'arm':
      read_only_regs = []
      read_only_zero_regs = ['r9']
    elif ARCH == 'mips32':
      read_only_regs = ['zero', 'k0', 'k1']
      read_only_zero_regs = []
    else:
      raise AssertionError('Unknown architecture')

    for reg_name in read_only_regs + read_only_zero_regs:
      # Read registers.
      regs = DecodeRegs(connection.RspRequest('g'))

      # Change a register.
      old_value = regs[reg_name]
      if reg_name in read_only_zero_regs:
        self.assertEquals(old_value, 0)
      regs[reg_name] += 1

      # Check cannot write registers.
      self.assertEquals(connection.RspRequest('G' + EncodeRegs(regs)), 'E03')

      # Read registers. Check for an old value.
      regs = DecodeRegs(connection.RspRequest('g'))
      self.assertEquals(regs[reg_name], old_value)

  # Test that reading from an unreadable address gives a sensible error.
  def CheckReadMemoryAtInvalidAddr(self, connection):
    mem_addr = 0
    result = connection.RspRequest('m%x,%x' % (mem_addr, 8))
    self.assertEquals(result, 'E03')

    # Check non-zero address in the first page.
    mem_addr = 0xffff
    resut = connection.RspRequest('m%x,%x' % (mem_addr, 1))
    self.assertEquals(result, 'E03')

  # Run tests on debugger_test.c binary.
  def test_debugger_test(self):
    with LaunchDebugStub('test_getting_registers') as connection:
      # Tell the process to continue, because it starts at the
      # breakpoint set at its start address.
      reply = connection.RspRequest('c')
      if ARCH == 'arm':
        AssertReplySignal(reply, NACL_SIGILL)
      elif ARCH == 'mips32':
        # The process should have stopped on a BKPT instruction.
        AssertReplySignal(reply, NACL_SIGTRAP)
      else:
        # The process should have stopped on a HLT instruction.
        AssertReplySignal(reply, NACL_SIGSEGV)

      self.CheckTargetXml(connection)
      self.CheckReadRegisters(connection)
      self.CheckWriteRegisters(connection)
      self.CheckReadOnlyRegisters(connection)

  def test_jump_to_address_zero(self):
    if UsingQemu():
      # This test hangs under qemu-arm or qemu-mips.
      return
    with LaunchDebugStub('test_jump_to_address_zero') as connection:
      # Continue from initial breakpoint.
      reply = connection.RspRequest('c')
      AssertReplySignal(reply, NACL_SIGSEGV)
      registers = DecodeRegs(connection.RspRequest('g'))
      if ARCH == 'x86-64':
        self.assertEquals(registers[IP_REG[ARCH]], registers['r15'])
      else:
        self.assertEquals(registers[IP_REG[ARCH]], 0)

  def test_reading_and_writing_memory(self):
    # Any arguments to the nexe would work here because we do not run
    # the executable beyond the initial breakpoint.
    with LaunchDebugStub('test_getting_registers') as connection:
      mem_addr = GetSymbols()['g_example_var']
      # Check reading memory.
      expected_data = 'some_debug_stub_test_data\0'
      reply = connection.RspRequest('m%x,%x' % (mem_addr, len(expected_data)))
      self.assertEquals(DecodeHex(reply), expected_data)

      # On x86-64, for reading/writing memory, the debug stub accepts
      # untrusted addresses with or without the %r15 sandbox base
      # address added, because GDB uses both.
      # TODO(eaeltsin): Fix GDB to not use addresses with %r15 added,
      # and change the expected result in the check below.
      if ARCH == 'x86-64':
        registers = DecodeRegs(connection.RspRequest('g'))
        sandbox_base_addr = registers['r15']
        reply = connection.RspRequest('m%x,%x' % (sandbox_base_addr + mem_addr,
                                                  len(expected_data)))
        self.assertEquals(DecodeHex(reply), expected_data)

      # Check writing memory.
      new_data = 'replacement_data\0'
      assert len(new_data) < len(expected_data)
      reply = connection.RspRequest('M%x,%x:%s' % (mem_addr, len(new_data),
                                                   EncodeHex(new_data)))
      self.assertEquals(reply, 'OK')
      # Check that we can read back what we wrote.
      reply = connection.RspRequest('m%x,%x' % (mem_addr, len(new_data)))
      self.assertEquals(DecodeHex(reply), new_data)

      self.CheckReadMemoryAtInvalidAddr(connection)

  def test_exit_code(self):
    with LaunchDebugStub('test_exit_code') as connection:
      reply = connection.RspRequest('c')
      self.assertEquals(reply, 'W02')

  # Single-step and check IP corresponds to debugger_test:test_single_step
  def CheckSingleStep(self, connection, step_command, thread_id):
    if ARCH == 'x86-32':
      instruction_sizes = [1, 2, 3, 6]
    elif ARCH == 'x86-64':
      instruction_sizes = [1, 3, 4, 6]
    else:
      raise AssertionError('Unknown architecture')

    ip = DecodeRegs(connection.RspRequest('g'))[IP_REG[ARCH]]

    for size in instruction_sizes:
      reply = connection.RspRequest(step_command)
      AssertReplySignal(reply, NACL_SIGTRAP)
      self.assertEquals(ParseThreadStopReply(reply)['thread_id'], thread_id)
      ip += size
      regs = DecodeRegs(connection.RspRequest('g'))
      self.assertEqual(regs[IP_REG[ARCH]], ip)
      # The trap flag should be reported as unset.
      self.assertEqual(regs['eflags'] & X86_TRAP_FLAG, 0)

  def test_single_step(self):
    if not SingleSteppingWorks():
      return
    with LaunchDebugStub('test_single_step') as connection:
      # We expect test_single_step() to stop at a HLT instruction.
      reply = connection.RspRequest('c')
      AssertReplySignal(reply, NACL_SIGSEGV)
      tid = ParseThreadStopReply(reply)['thread_id']
      # Skip past the single-byte HLT instruction.
      regs = DecodeRegs(connection.RspRequest('g'))
      regs[IP_REG[ARCH]] += 1
      AssertEquals(connection.RspRequest('G' + EncodeRegs(regs)), 'OK')

      self.CheckSingleStep(connection, 's', tid)
      # Check that we can continue after single-stepping.
      reply = connection.RspRequest('c')
      self.assertEquals(reply, 'W00')

  def test_vCont(self):
    # Basically repeat test_single_step, but using vCont commands.
    if not SingleSteppingWorks():
      return
    with LaunchDebugStub('test_single_step') as connection:
      # Test if vCont is supported.
      reply = connection.RspRequest('vCont?')
      self.assertEqual(reply, 'vCont;s;S;c;C')

      # Continue using vCont.
      # We expect test_single_step() to stop at a HLT instruction.
      reply = connection.RspRequest('vCont;c')
      AssertReplySignal(reply, NACL_SIGSEGV)
      # Get signalled thread id.
      tid = ParseThreadStopReply(reply)['thread_id']
      # Skip past the single-byte HLT instruction.
      regs = DecodeRegs(connection.RspRequest('g'))
      regs[IP_REG[ARCH]] += 1
      AssertEquals(connection.RspRequest('G' + EncodeRegs(regs)), 'OK')

      self.CheckSingleStep(connection, 'vCont;s:%x' % tid, tid)

      # Single step one thread and continue all others.
      reply = connection.RspRequest('vCont;s:%x;c' % tid)
      # WARNING! This check is valid in single-threaded case only!
      # In multi-threaded case another thread might stop first.
      self.assertEqual(reply, 'T05thread:%x;' % tid)

      # Try to continue the thread and to single-step all others.
      reply = connection.RspRequest('vCont;c:%x;s' % tid)
      self.assertTrue(reply.startswith('E'))

      # Try to single-step wrong thread.
      reply = connection.RspRequest('vCont;s:%x' % (tid + 2))
      self.assertTrue(reply.startswith('E'))

      # Try to single-step all threads.
      reply = connection.RspRequest('vCont;s')
      self.assertTrue(reply.startswith('E'))

  def test_interrupt(self):
    if not SingleSteppingWorks():
      return
    func_addr = GetSymbols()['test_interrupt']
    with LaunchDebugStub('test_interrupt') as connection:
      # Single stepping inside syscalls doesn't work. So we need to reach
      # a point where interrupt will not catch the program inside syscall.
      reply = connection.RspRequest('Z0,%x,0' % func_addr)
      self.assertEquals(reply, 'OK')
      reply = connection.RspRequest('c')
      AssertReplySignal(reply, NACL_SIGTRAP)
      reply = connection.RspRequest('z0,%x,0' % func_addr)
      self.assertEquals(reply, 'OK')

      # Continue (program will spin forever), then interrupt.
      connection.RspSendOnly('c')
      reply = connection.RspInterrupt()
      self.assertEqual(reply, 'T00')

      # Single-step.
      reply = connection.RspRequest('s')
      AssertReplySignal(reply, NACL_SIGTRAP)

  def test_modifying_code_is_disallowed(self):
    with LaunchDebugStub('test_setting_breakpoint') as connection:
      # Pick an arbitrary address in the code segment.
      func_addr = GetSymbols()['breakpoint_target_func']
      # Writing to the code area should be disallowed.
      data = '\x00'
      write_command = 'M%x,%x:%s' % (func_addr, len(data), EncodeHex(data))
      reply = connection.RspRequest(write_command)
      self.assertEquals(reply, 'E03')

  def test_kill(self):
    sel_ldr = PopenDebugStub('test_exit_code')
    try:
      connection = gdb_rsp.GdbRspConnection()
      # Request killing the target.
      reply = connection.RspRequest('k')
      self.assertEquals(reply, 'OK')
      self.assertEquals(sel_ldr.wait(), RETURNCODE_KILL)
    finally:
      KillProcess(sel_ldr)

  def test_detach(self):
    sel_ldr = PopenDebugStub('test_exit_code')
    try:
      connection = gdb_rsp.GdbRspConnection()
      # Request detaching from the target.
      # This resumes execution, so we get the nexe's normal exit() status.
      reply = connection.RspRequest('D')
      self.assertEquals(reply, 'OK')
      self.assertEquals(sel_ldr.wait(), 2)
    finally:
      KillProcess(sel_ldr)

  def test_disconnect(self):
    sel_ldr = PopenDebugStub('test_exit_code')
    try:
      # Connect and record the instruction pointer.
      connection = gdb_rsp.GdbRspConnection()
      # Check something basic responds with sane results.
      reply = connection.RspRequest('vCont?')
      self.assertEqual(reply, 'vCont;s;S;c;C')
      # Store the instruction pointer.
      registers = DecodeRegs(connection.RspRequest('g'))
      initial_ip = registers[IP_REG[ARCH]]
      connection.Close()
      # Reconnect 5 times.
      for _ in range(5):
        connection = gdb_rsp.GdbRspConnection()
        # Confirm the instruction pointer stays where it was, indicating that
        # the thread stayed suspended.
        registers = DecodeRegs(connection.RspRequest('g'))
        self.assertEquals(registers[IP_REG[ARCH]], initial_ip)
        connection.Close()
    finally:
      KillProcess(sel_ldr)

  def RemoteGet(self, connection, download_filename, expected_filename):
    """Use the vFile interface to remote get a file, checking the result.

    Args:
      connection: Rsp connection to debug stub to use.
      download_filename: Remote filename to download.
      expected_filename: Local filename to check downloaded result against.
    """
    # Open the nexe.
    reply = connection.RspRequest(
        'vFile:open:%s,0,0' % EncodeHex(download_filename))
    self.assertEqual(reply[0], 'F')
    fd = int(reply[1:], 16)
    self.assertGreaterEqual(fd, 0)
    # Read in the full contents of the file.
    data = ''
    offset = 0
    while True:
      # Read up to 4096 bytes at a time.
      reply = connection.RspRequest(
        'vFile:pread:%x,%x,%x' % (fd, 4096, offset))
      self.assertEqual(reply[0], 'F')
      retcode, chunk = reply[1:].split(';', 1)
      retcode = int(retcode, 16)
      self.assertGreaterEqual(retcode, 0)
      chunk = DecodeEscaping(chunk)
      self.assertEqual(len(chunk), retcode)
      if retcode == 0:
        break
      data += chunk
      offset += retcode
    expected_data = open(expected_filename, 'rb').read()
    # Check that the length matches first, so that large data mismatch spew
    # is only emitted in the case of more subtle mismatches.
    self.assertEqual(len(data), len(expected_data))
    self.assertEqual(data, expected_data)
    # Close the file handle.
    reply = connection.RspRequest('vFile:close:%x' % fd)
    self.assertEqual(reply, 'F0')

  def test_remote_get_main_nexe(self):
    with LaunchDebugStub('test_interrupt') as connection:
      self.RemoteGet(connection, 'nexe', MainNexe())

  def test_remote_get_irt(self):
    if IrtNexe() is None:
      self.skipTest('Does not work in non-irt mode.')
    with LaunchDebugStub('test_interrupt') as connection:
      self.RemoteGet(connection, 'irt', IrtNexe())

  def test_remote_get_bad_fd(self):
    with LaunchDebugStub('test_interrupt') as connection:
      # Test closing a not-yet-opened nexe.
      reply = connection.RspRequest('vFile:close:%x' % 13)
      self.assertEqual(reply, 'F-1,%x' % GDB_EBADF)

  def test_register_constraints(self):
    if ARCH == 'mips32':
      # This has not been implemented for mips.
      return
    with LaunchDebugStub('test_super_instruction') as connection:
      reply = connection.RspRequest('c')
      SkipBreakpoint(connection, reply)

      regs = DecodeRegs(connection.RspRequest('g'))

      if ARCH == 'x86-32':
        valid_regs = ['eax', 'ebx', 'ecx', 'edx', 'esp',
                      'ebp', 'esi', 'edi', 'eflags']
        read_only_regs = ['cs', 'ss', 'ds', 'es', 'fs', 'gs']
        super_inst_len = 5
      elif ARCH == 'x86-64':
        valid_regs = ['rax', 'rbx', 'rcx', 'rdx', 'rsi', 'rdi', 'eflags',
                      'r8', 'r9', 'r10', 'r11', 'r12', 'r13', 'r14']
        read_only_regs = ['cs', 'ss', 'ds', 'es', 'fs', 'gs', 'r15']
        super_inst_len = 8
      elif ARCH == 'arm':
        valid_regs = ['r0', 'r1', 'r2', 'r3', 'r4', 'r5', 'r6',
                      'r7', 'r8', 'r10', 'r11', 'r12', 'r14']
        read_only_regs = ['r9']
        super_inst_len = 8
      else:
        raise AssertionError('Unknown architecture')

      # Allowed to change all valid registers.
      for reg in valid_regs:
        self.assertEquals(ChangeReg(connection, reg,
                                    lambda x: (x + 1) % 0xffffffff), 'OK')

      # Cannot change read only registers.
      for reg in read_only_regs:
        self.assertEquals(ChangeReg(connection, reg,
                                    lambda x: (x + 1) % 0xffffffff), 'E03')

      if ARCH == 'x86-64':
        # Cannot change the upper 32 bits of x86-64 restricted registers
        # to non-zero value.
        for reg in ['rsp', 'rbp']:
          self.assertEquals(ChangeReg(connection, reg, lambda x: 0), 'OK')
          self.assertEquals(ChangeReg(connection, reg, lambda x: x), 'OK')
          self.assertEquals(ChangeReg(connection, reg,
            lambda x: x + 0xf00000000), 'E03')
      elif ARCH == 'arm':
        # Upper 2 bits of r13 (sp) must be 0.
        self.assertEquals(ChangeReg(connection, 'r13', lambda x: 0), 'OK')
        self.assertEquals(ChangeReg(connection, 'r13',
                                    lambda x: 0x80000000), 'E03')
        # Only upper 4 bits (NZCV) of cpsr can be set.
        self.assertEquals(ChangeReg(connection, 'cpsr',
                                    lambda x: 0xF0000000), 'OK')
        self.assertEquals(ChangeReg(connection, 'cpsr', lambda x: 1), 'E03')
        # PC must fall in sandboxed range
        self.assertEquals(ChangeReg(connection, 'r15',
                                    lambda x: 0xF0000000), 'E03')
        # PC must be on inst boundary (insts are 4 bytes)
        self.assertEquals(ChangeReg(connection, 'r15', lambda x: x + 3), 'E03')

      # Next instruction is a super instruction.
      # Therefore cannot jump anywhere in the middle.
      for i in xrange(1, super_inst_len):
        self.assertEquals(ChangeReg(connection, IP_REG[ARCH],
          lambda x: x + i), 'E03')

      # Allowed to jump over the entire super instruction.
      self.assertEquals(ChangeReg(connection, IP_REG[ARCH],
        lambda x: x + super_inst_len), 'OK')

  def test_step_inside_super_instruction(self):
    if not SingleSteppingWorks():
      return
    with LaunchDebugStub('test_super_instruction') as connection:
      if ARCH == 'x86-32':
        reg = 'eax'
      elif ARCH == 'x86-64':
        reg = 'rax'
      else:
        raise AssertionError('Unknown architecture')

      reply = connection.RspRequest('c')
      SkipBreakpoint(connection, reply)

      reply = connection.RspRequest('vCont;s:1')
      self.assertEquals(reply, 'T05thread:1;')

      regs = DecodeRegs(connection.RspRequest('g'))

      # Cannot change registers within super-instruction.
      self.assertEquals(ChangeReg(connection, reg, lambda x: x + 1), 'E03')

class DebugStubBreakpointTest(unittest.TestCase):

  def CheckInstructionPtr(self, connection, expected_ip):
    ip_value = DecodeRegs(connection.RspRequest('g'))[IP_REG[ARCH]]
    if ARCH == 'x86-64':
      # TODO(mseaborn): The debug stub should probably omit the top
      # bits of %rip automatically.
      ip_value &= 0xffffffff
    self.assertEquals(ip_value, expected_ip)

  def test_setting_and_removing_breakpoint(self):
    func_addr = GetSymbols()['breakpoint_target_func']
    with LaunchDebugStub('test_setting_breakpoint') as connection:
      # Set a breakpoint.
      reply = connection.RspRequest('Z0,%x,0' % func_addr)
      self.assertEquals(reply, 'OK')
      # Requesting a breakpoint on an address that already has a
      # breakpoint should return an error.
      reply = connection.RspRequest('Z0,%x,0' % func_addr)
      self.assertEquals(reply, 'E03')

      # When we run the program, we should hit the breakpoint.  When
      # we continue, we should hit the breakpoint again because it has
      # not been removed: the debug stub does not step through
      # breakpoints automatically.
      for i in xrange(2):
        reply = connection.RspRequest('c')
        AssertReplySignal(reply, NACL_SIGTRAP)
        self.CheckInstructionPtr(connection, func_addr)

      # If we continue a single thread, the fault the thread receives
      # should still be recognized as a breakpoint.
      tid = ParseThreadStopReply(reply)['thread_id']
      reply = connection.RspRequest('vCont;c:%x' % tid)
      AssertReplySignal(reply, NACL_SIGTRAP)
      self.CheckInstructionPtr(connection, func_addr)

      # Check that we can remove the breakpoint.
      reply = connection.RspRequest('z0,%x,0' % func_addr)
      self.assertEquals(reply, 'OK')
      # Requesting removing a breakpoint on an address that does not
      # have one should return an error.
      reply = connection.RspRequest('z0,%x,0' % func_addr)
      self.assertEquals(reply, 'E03')
      # After continuing, we should not hit the breakpoint again, and
      # the program should run to completion.
      reply = connection.RspRequest('c')
      self.assertEquals(reply, 'W00')

  def test_setting_breakpoint_on_invalid_address(self):
    with LaunchDebugStub('test_exit_code') as connection:
      # Requesting a breakpoint on an invalid address should give an error.
      reply = connection.RspRequest('Z0,%x,1' % (1 << 32))
      self.assertEquals(reply, 'E03')

  def test_setting_breakpoint_on_data_address(self):
    with LaunchDebugStub('test_exit_code') as connection:
      # Pick an arbitrary address in the data segment.
      data_addr = GetSymbols()['g_main_thread_var']
      # Requesting a breakpoint on a non-code address should give an error.
      reply = connection.RspRequest('Z0,%x,1' % data_addr)
      self.assertEquals(reply, 'E03')

  def test_breakpoint_memory_changes_are_hidden(self):
    func_addr = GetSymbols()['breakpoint_target_func']
    with LaunchDebugStub('test_setting_breakpoint') as connection:
      chunk_size = 32
      old_memory = ReadMemory(connection, func_addr, chunk_size)
      reply = connection.RspRequest('Z0,%x,0' % func_addr)
      self.assertEquals(reply, 'OK')

      # The debug stub should hide the memory modification.
      new_memory = ReadMemory(connection, func_addr, chunk_size)
      self.assertEquals(new_memory, old_memory)
      # Check reading a subset of the range.  (This will only be a
      # proper subset on architectures where the breakpoint size is
      # >1, such as ARM not but x86.)
      new_memory = ReadMemory(connection, func_addr, 1)
      self.assertEquals(new_memory, old_memory[:1])

  def test_setting_breakpoint_in_super_instruction(self):
    if ARCH != 'x86-32' and ARCH != 'x86-64':
      # ARM breakpoints are handled separately, and MIPS has not been
      # implemented.
      return
    with LaunchDebugStub('test_super_instruction') as connection:
      reply = connection.RspRequest('c')
      SkipBreakpoint(connection, reply)

      regs = DecodeRegs(connection.RspRequest('g'))

      if ARCH == 'arm':
        offset = 4
      elif ARCH == 'x86-32' or ARCH == 'x86-64':
        offset = 3
      else:
        raise AssertionError('Unknown architecture')

      # Next instruction is inside super instruction.
      invalid_addr = regs[IP_REG[ARCH]] + offset

      # Setting breakpoint here should not be allowed.
      reply = connection.RspRequest('Z0,%x,0' % invalid_addr)
      self.assertEquals(reply, 'E03')

  def test_setting_breakpoint_in_constant_pools(self):
    if ARCH != 'arm':
      return
    with LaunchDebugStub('test_arm_breakpoint') as connection:
      reply = connection.RspRequest('c')
      SkipBreakpoint(connection, reply)

      regs = DecodeRegs(connection.RspRequest('g'))

      # First constant pool is next bundle.
      pc = regs[IP_REG[ARCH]] + 16

      # Check we cannot set breakpoint anywhere inside both bundles.
      for addr in range(pc, pc + 32):
        reply = connection.RspRequest('Z0,%x,0' % addr)
        self.assertEquals(reply, 'E03')


class DebugStubThreadSuspensionTest(unittest.TestCase):

  def WaitForTestThreadsToStart(self, connection, symbols):
    # Wait until:
    #  * The main thread starts to modify g_main_thread_var.
    #  * The child thread executes a breakpoint.
    old_value = ReadUint32(connection, symbols['g_main_thread_var'])
    while True:
      reply = connection.RspRequest('c')
      SkipBreakpoint(connection, reply)
      child_thread_id = ParseThreadStopReply(reply)['thread_id']
      if ReadUint32(connection, symbols['g_main_thread_var']) != old_value:
        break
    return child_thread_id

  def test_continuing_thread_with_others_suspended(self):
    if UsingQemu():
      # Suspending a running thread doesn't work under qemu-arm or qemu-mips,
      # so disable this test there.
      return
    with LaunchDebugStub('test_suspending_threads') as connection:
      symbols = GetSymbols()
      child_thread_id = self.WaitForTestThreadsToStart(connection, symbols)

      # Test continuing a single thread while other threads remain
      # suspended.
      for _ in range(3):
        main_thread_val = ReadUint32(connection, symbols['g_main_thread_var'])
        child_thread_val = ReadUint32(connection, symbols['g_child_thread_var'])
        reply = connection.RspRequest('vCont;c:%x' % child_thread_id)
        SkipBreakpoint(connection, reply)
        self.assertEquals(ParseThreadStopReply(reply)['thread_id'],
                          child_thread_id)
        # The main thread should not be allowed to run, so should not
        # modify g_main_thread_var.
        self.assertEquals(
            ReadUint32(connection, symbols['g_main_thread_var']),
            main_thread_val)
        # The child thread should always modify g_child_thread_var
        # between each breakpoint.
        self.assertNotEquals(
            ReadUint32(connection, symbols['g_child_thread_var']),
            child_thread_val)

  def test_single_stepping_thread_with_others_suspended(self):
    if UsingQemu():
      # Suspending a running thread doesn't work under qemu-arm or qemu-mips,
      # so disable this test there.
      return
    with LaunchDebugStub('test_suspending_threads') as connection:
      symbols = GetSymbols()
      child_thread_id = self.WaitForTestThreadsToStart(connection, symbols)

      # Test single-stepping a single thread while other threads
      # remain suspended.
      for _ in range(3):
        main_thread_val = ReadUint32(connection, symbols['g_main_thread_var'])
        child_thread_val = ReadUint32(connection, symbols['g_child_thread_var'])
        while True:
          reply = connection.RspRequest('vCont;s:%x' % child_thread_id)
          if (ARCH in ('x86-32', 'x86-64') and
              ParseThreadStopReply(reply)['signal'] == NACL_SIGTRAP):
            # We single-stepped through an instruction without
            # otherwise faulting.  We did not hit the breakpoint, so
            # there is nothing to do.
            pass
          else:
            SkipBreakpoint(connection, reply)
          self.assertEquals(ParseThreadStopReply(reply)['thread_id'],
                            child_thread_id)
          # The main thread should not be allowed to run, so should not
          # modify g_main_thread_var.
          self.assertEquals(
              ReadUint32(connection, symbols['g_main_thread_var']),
              main_thread_val)
          # Eventually, the child thread should modify g_child_thread_var.
          if (ReadUint32(connection, symbols['g_child_thread_var'])
              != child_thread_val):
            break


def Main():
  # TODO(mseaborn): Clean up to remove the global variables.  They are
  # currently here because unittest does not help with making
  # parameterised tests.
  index = sys.argv.index('--')
  args = sys.argv[index + 1:]
  # The remaining arguments go to unittest.main().
  sys.argv = sys.argv[:index]
  global ARCH
  global NM_TOOL
  global SEL_LDR_COMMAND
  ARCH = args.pop(0)
  NM_TOOL = args.pop(0)
  SEL_LDR_COMMAND = args
  unittest.main()


if __name__ == '__main__':
  Main()