Support remote UNIX sockets for GDBProtocol

avatar2/protocols/gdb.py

@@ -397,6 +397,15 @@ class GDBProtocol(object):
 
         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'):
         """

tests/test_gdbprotocol.py

@@ -5,6 +5,7 @@ import subprocess
 import os
 import time
 
+from test_utils import unix2tcp
 from nose.tools import *
 
 SLEEP_TIME = 1
@@ -38,6 +39,28 @@ def setup_helloworld():
     # Base addresses can change across kernel versions due to PIE binaries
     MEM_BASE = g.get_symbol("main")[1] & ~0xfff
 
+def setup_helloworld_unix():
+    global p, g, port, MEM_BASE
+
+    socket_path ='/tmp/test_socket'
+    unix2tcp(socket_path, "127.0.0.1", port)
+
+    binary = '%s/tests/binaries/hello_world' % os.getcwd()
+    p = subprocess.Popen(['gdbserver', '--once', '127.0.0.1:%d' % port, binary],
+                        stderr=subprocess.PIPE)
+
+    out = str(p.stderr.readline())
+    assert_equal(binary in out, True)
+    out = str(p.stderr.readline())
+    assert_equal(str(port) in out, True)
+
+    g = GDBProtocol(arch=avatar2.archs.X86_64)
+    g.remote_connect_unix(socket_path)
+
+    # Base addresses can change across kernel versions due to PIE binaries
+    MEM_BASE = g.get_symbol("main")[1] & ~0xfff
+
+
 def setup_inf_loop():
     global p, g, port
 
@@ -73,6 +96,12 @@ def test_register_read_and_write():
     ret = g.read_register('rax')
     assert_equal(ret, 1678)
 
+@with_setup(setup_helloworld_unix, teardown_func)
+def test_register_read_and_write_unix():
+    ret = g.write_register('rax', 1678)
+    assert_equal(ret, True)
+    ret = g.read_register('rax')
+    assert_equal(ret, 1678)
 
 @with_setup(setup_helloworld, teardown_func)
 def test_break_run_and_read_write_mem():

tests/test_utils.py

@@ -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()
+