avatartwo/avatar2
1
0
Fork 0
mirror of https://github.com/avatartwo/avatar2.git synced 2026-10-11 22:29:59 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merging with rework on unit tests

Browse Source
This commit is contained in:
sampl
2022-01-26 13:45:49 +01:00
parent 37fa74d1ff f4ad6fc6e8
commit 4f98afb4f4
6 changed files with 157 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files
avatar2/protocols/gdb.py
+41 -52
View File
@@ -337,15 +337,7 @@ class GDBProtocol(object):
return ret
def remote_connect(self, ip='127.0.0.1', port=3333):
"""
connect to a remote gdb server
:param ip: ip of the remote gdb-server (default: localhost)
:param port: port of the remote gdb-server (default: port)
:returns: True on successful connection
"""
def _remote_connect_common(self, remote_string, transport_setup=None):
req = ['-gdb-set', 'target-async', 'on']
ret, resp = self._sync_request(req, GDB_PROT_DONE)
if not ret:
@@ -356,8 +348,6 @@ class GDBProtocol(object):
req = ['-gdb-set', 'architecture', self._arch.gdb_name]
ret, resp = self._sync_request(req, GDB_PROT_DONE)
if not ret:
self.log.critical(
"Unable to set architecture, received response: %s" %
@@ -379,7 +369,11 @@ class GDBProtocol(object):
resp)
raise Exception("GDBProtocol was unable to set endianness")
req = ['-target-select', 'remote', '%s:%d' % (ip, int(port))]
# transport unique setup, if applicable
if transport_setup:
transport_setup()
req = ['-target-select', 'remote', remote_string]
ret, resp = self._sync_request(req, GDB_PROT_CONN)
self.log.debug(
@@ -393,6 +387,26 @@ class GDBProtocol(object):
return ret
def remote_connect(self, ip='127.0.0.1', port=3333):
"""
connect to a remote gdb server via TCP
:param ip: ip of the remote gdb-server (default: localhost)
:param port: port of the remote gdb-server (default: port)
:returns: True on successful connection
"""
return self._remote_connect_common('%s:%d' % (ip, int(port)))
def remote_connect_unix(self, unix_path):
"""
connect to a remote gdb server via a UNIX domain socket path
:param unix_path: path of the UNIX domain socket
:returns: True on successful connection
"""
return self._remote_connect_common(unix_path)
def remote_connect_serial(self, device='/dev/ttyACM0', baud_rate=38400,
parity='none'):
"""
@@ -404,49 +418,24 @@ class GDBProtocol(object):
:returns: True on successful connection
"""
req = ['-gdb-set', 'architecture', self._arch.gdb_name]
ret, resp = self._sync_request(req, GDB_PROT_DONE)
if not ret:
self.log.critical(
"Unable to set architecture, received response: %s" %
resp)
raise Exception(("GDBProtocol was unable to set the architecture\n"
"Did you select the right gdb_executable?"))
def serial_setup():
if parity not in ['none', 'even', 'odd']:
self.log.critical("Parity must be none, even or odd")
raise Exception("Cannot set parity to %s" % parity)
if parity not in ['none', 'even', 'odd']:
self.log.critical("Parity must be none, even or odd")
raise Exception("Cannot set parity to %s" % parity)
req = ['-gdb-set', 'serial', 'parity', '%s' % parity]
ret, resp = self._sync_request(req, GDB_PROT_DONE)
if not ret:
self.log.critical("Unable to set parity")
raise Exception("GDBProtocol was unable to set parity")
req = ['-gdb-set', 'mi-async', 'on']
ret, resp = self._sync_request(req, GDB_PROT_DONE)
if not ret:
self.log.critical(
"Unable to set GDB/MI to async, received response: %s" %
resp)
raise Exception("GDBProtocol was unable to connect")
req = ['-gdb-set', 'serial', 'baud', '%i' % baud_rate]
ret, resp = self._sync_request(req, GDB_PROT_DONE)
if not ret:
self.log.critical("Unable to set baud rate")
raise Exception("GDBProtocol was unable to set Baudrate")
req = ['-gdb-set', 'serial', 'parity', '%s' % parity]
ret, resp = self._sync_request(req, GDB_PROT_DONE)
if not ret:
self.log.critical("Unable to set parity")
raise Exception("GDBProtocol was unable to set parity")
req = ['-gdb-set', 'serial', 'baud', '%i' % baud_rate]
ret, resp = self._sync_request(req, GDB_PROT_DONE)
if not ret:
self.log.critical("Unable to set baud rate")
raise Exception("GDBProtocol was unable to set Baudrate")
req = ['-target-select', 'remote', '%s' % device]
ret, resp = self._sync_request(req, GDB_PROT_CONN)
self.log.debug(
"Attempted to connect to target. Received response: %s" %
resp)
self.update_target_regs()
return ret
return self._remote_connect_common(device, transport_setup=serial_setup)
def remote_disconnect(self):
"""
avatar2/targets/gdb_target.py
+8 -1
View File
@@ -8,6 +8,7 @@ class GDBTarget(Target):
def __init__(self, avatar,
gdb_executable=None, gdb_additional_args=None,
gdb_ip='127.0.0.1', gdb_port=3333,
gdb_unix_socket_path=None,
gdb_serial_device='/dev/ttyACM0',
gdb_serial_baud_rate=38400,
gdb_serial_parity='none',
@@ -26,6 +27,7 @@ class GDBTarget(Target):
self.gdb_additional_args = gdb_additional_args if gdb_additional_args else []
self.gdb_ip = gdb_ip
self.gdb_port = gdb_port
self.gdb_unix_socket_path = gdb_unix_socket_path
self.gdb_serial_device = gdb_serial_device
self.gdb_serial_baud_rate = gdb_serial_baud_rate
self.gdb_serial_parity = gdb_serial_parity
@@ -49,7 +51,12 @@ class GDBTarget(Target):
# If we are debugging a program locally,
# we do not need to establish any connections
if not self._local_binary:
if not self._serial:
if self.gdb_unix_socket_path is not None:
if gdb.remote_connect_unix(self.gdb_unix_socket_path):
self.log.info("Connected to Target")
else:
self.log.warning("Connecting failed")
elif not self._serial:
if gdb.remote_connect(ip=self.gdb_ip, port=self.gdb_port):
self.log.info("Connected to Target")
else:
avatar2/targets/qemu_target.py
+17 -2
View File
@@ -25,6 +25,7 @@ class QemuTarget(Target):
firmware=None,
gdb_executable=None,
gdb_port=3333,
gdb_unix_socket_path=None,
additional_args=None,
gdb_additional_args=None,
gdb_verbose=False,
@@ -58,6 +59,7 @@ class QemuTarget(Target):
)
self.gdb_port = gdb_port
self.gdb_unix_socket_path = gdb_unix_socket_path
self.gdb_additional_args = gdb_additional_args if gdb_additional_args else []
self.gdb_verbose = gdb_verbose
@@ -90,7 +92,10 @@ class QemuTarget(Target):
machine = ["-machine", "configurable"]
kernel = ["-kernel", self.qemu_config_file]
gdb_option = ["-gdb", "tcp::" + str(self.gdb_port)]
if self.gdb_unix_socket_path:
gdb_option = ["-gdb", "unix:%s,server,nowait" % str(self.gdb_unix_socket_path)]
else:
gdb_option = ["-gdb", "tcp::" + str(self.gdb_port)]
stop_on_startup = ["-S"]
nographic = ["-nographic"] # , "-monitor", "/dev/null"]
qmp = ["-qmp", "tcp:127.0.0.1:%d,server,nowait" % self.qmp_port]
@@ -301,10 +306,20 @@ class QemuTarget(Target):
self.protocols.monitor = qmp
self.protocols.remote_memory = rmp
if gdb.remote_connect(port=self.gdb_port) and qmp.connect():
connect_success = True
if self.gdb_unix_socket_path:
connect_success = connect_success and gdb.remote_connect_unix(self.gdb_unix_socket_path)
else:
connect_success = connect_success and gdb.remote_connect(port=self.gdb_port)
connect_success = connect_success and qmp.connect()
if connect_success:
self.log.info("Connected to remote target")
else:
self.log.warning("Connection to remote target failed")
if rmp:
rmp.connect()
self.wait()
tests/test_gdbprotocol.py
+26 -12
View File
@@ -1,8 +1,5 @@
import unittest
from avatar2.protocols.gdb import GDBProtocol
import avatar2
import subprocess
import os
import time
@@ -10,11 +7,15 @@ import re
from os.path import dirname, realpath
import avatar2
from avatar2.protocols.gdb import GDBProtocol
from utils import unix2tcp
SLEEP_TIME = .1
PORT = 4444
X86_REGS = [u'rax', u'rbx', u'rcx', u'rdx', u'rsi', u'rdi', u'rbp', u'rsp',
u'r8', u'r9', u'r10', u'r11', u'r12', u'r13', u'r14', u'r15',
u'rip', u'eflags', u'cs', u'ss', u'ds', u'es', u'fs', u'gs']
@@ -26,7 +27,7 @@ class GdbProtocolTestCase(unittest.TestCase):
def setUp(self):
pass
def setup_env(self, binary):
def setup_env(self, binary, unix_socket=False):
self.process = subprocess.Popen(['gdbserver', '--once', '127.0.0.1:%d' % PORT, binary],
stderr=subprocess.PIPE)
@@ -37,12 +38,18 @@ class GdbProtocolTestCase(unittest.TestCase):
self.assertEqual(str(PORT) in out, True, out)
self.gdb = GDBProtocol(arch=avatar2.archs.X86_64)
self.gdb.remote_connect(port=PORT)
# let's resolve the base address of the binary
ret, out = self.gdb.console_command("p &main")
main_addr = int(re.search("0x[0-9a-f]+", out).group(0), 16)
self.base_address = main_addr - main_addr % 0x1000
if unix_socket is True:
socket_path ='/tmp/test_socket'
unix2tcp(socket_path, "127.0.0.1", PORT)
self.gdb.remote_connect_unix(socket_path)
else:
self.gdb.remote_connect(port=PORT)
# Base addresses can change across kernel versions due to PIE binaries
self.base_address = self.gdb.get_symbol("main")[1] & ~0xfff
def wait_stopped(self):
# As we do not have access to avatar synchronizing target states
@@ -113,8 +120,15 @@ class GDBProtocolTestCaseOnHelloWorld(GdbProtocolTestCase):
self.assertEqual(ret, 0x464c457f, ret)
class GDBProtocolTestCaseOnInfiniteLoop(GdbProtocolTestCase):
class GDBProtocolWithUnixSocketTestCase(GDBProtocolTestCaseOnHelloWorld):
def setUp(self):
dir_path = dirname(realpath(__file__))
binary = '%s/binaries/hello_world' % dir_path
self.setup_env(binary, unix_socket=True)
class GDBProtocolTestCaseOnInfiniteLoop(GdbProtocolTestCase):
def setUp(self):
dir_path = dirname(realpath(__file__))
tests/test_qemutarget.py
+15 -1
View File
@@ -75,15 +75,18 @@ class QemuTargetTestCase(unittest.TestCase):
def setUp(self):
self.setup_env()
def setup_env(self, gdb_unix_socket_path=None):
self.rom_addr = None
self.arch = None
self.setup_arch()
self.avatar = Avatar(arch=self.arch, output_directory=TEST_DIR)
self.avatar = Avatar(arch=self.arch, output_directory=TEST_DIR, configure_logging=False)
self.qemu = QemuTarget(self.avatar, name='qemu_test',
#firmware="./tests/binaries/qemu_arm_test",
firmware='%s/firmware' % TEST_DIR,
gdb_unix_socket_path=gdb_unix_socket_path,
)
self.fake_target = FakeTarget()
@@ -182,6 +185,17 @@ class QemuTargetTestCase(unittest.TestCase):
self.assertEqual(remote_memory_read, 0xdeadbeef, remote_memory_read)
class QemuTargetWithUnixSocketTestCase(QemuTargetTestCase):
def setUp(self):
self.setup_env(gdb_unix_socket_path="/tmp/test_sock")
#def test_initialization():
# self.qemu.init()
# self.qemu.wait()
# self.assertEqual(qemu.state, TargetStates.STOPPED, self.qemu.state)
if __name__ == '__main__':
unittest.main()
tests/utils.py
+50
View File
@@ -0,0 +1,50 @@
import socket
import threading
import os
import time
# oneshot UNIX to TCP socket proxy. Stops after first connection
def unix2tcp(unix_socket_path, tcp_host, tcp_port):
try:
os.unlink(unix_socket_path)
except OSError:
pass
usock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
tsock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
usock.bind(unix_socket_path)
usock.listen(1)
def proxy_loop():
uconn, addr = usock.accept()
tsock.connect((tcp_host, tcp_port))
uconn.setblocking(False)
tsock.setblocking(False)
while True:
data = None
try:
data = uconn.recv(1000)
if len(data) == 0:
break
tsock.sendall(data)
except BlockingIOError:
pass
try:
data = tsock.recv(1000)
if len(data) == 0:
break
uconn.sendall(data)
except BlockingIOError:
pass
usock.close()
uconn.close()
tsock.close()
threading.Thread(target=proxy_loop, daemon=True).start()