WIP: Add remote debug server zdbserver

Author: Kevin Z. Snow

conftest.py

@@ -1,3 +1,9 @@
+import contextlib
+import os
+
+import filelock
+import pytest
+
 from hypothesis import HealthCheck, settings
 
 
@@ -7,3 +13,18 @@ def pytest_configure(config):
         "patience", settings(suppress_health_check=[HealthCheck.too_slow])
     )
     settings.load_profile("patience")
+
+
+@pytest.fixture(scope="session")
+def lock(tmp_path_factory):
+    base_temp = tmp_path_factory.getbasetemp()
+    lock_file = base_temp.parent / "serial.lock"
+    yield filelock.FileLock(lock_file=str(lock_file))
+    with contextlib.suppress(OSError):
+        os.remove(path=lock_file)
+
+
+@pytest.fixture()
+def serial(lock):
+    with lock.acquire(poll_intervall=0.1):
+        yield

pyproject.toml

@@ -18,7 +18,7 @@ not_skip="__init__.py"
 use_parentheses=true
 
 known_first_party="zelos"
-known_third_party=["capstone", "colorama", "configargparse", "dnslib", "hypothesis", "lief", "pypacker", "recommonmark", "setuptools", "sortedcontainers", "sphinx_rtd_theme", "termcolor", "unicorn", "verboselogs", "zelos"]
+known_third_party=["capstone", "colorama", "configargparse", "dnslib", "filelock", "hypothesis", "lief", "pypacker", "pytest", "recommonmark", "setuptools", "sortedcontainers", "sphinx_rtd_theme", "termcolor", "unicorn", "verboselogs", "zelos"]
 
 [tool.pytest]
 junit_family = "xunit2"

src/zelos/__init__.py

@@ -33,7 +33,7 @@
 
 import colorama
 
-from .api.zelos_api import Zelos
+from .api.zelos_api import Zelos, ZelosCmdline
 from .engine import Engine
 from .exceptions import (
     InvalidHookTypeException,
@@ -51,6 +51,7 @@
 
 __all__ = [
     "Zelos",
+    "ZelosCmdline",
     "Engine",
     "ZelosException",
     "ZelosLoadException",

src/zelos/api/regs_api.py

@@ -54,6 +54,12 @@ def __getattr__(self, attr):
     def __setattr__(self, attr, value):
         self._current_thread.set_reg(attr, value)
 
+    def __getitem__(self, key):
+        return self._current_thread.get_reg(key)
+
+    def __setitem__(self, key, value):
+        self._current_thread.set_reg(key, value)
+
     def getIP(self) -> int:
         """
         Returns the platform-agnostic instruction pointer. On x86, this

src/zelos/tools/zdbserver/README.md

@@ -0,0 +1,19 @@
+# Zelos Remote Debug Server (zdb)
+
+The `zdbserver` enables remote debugging with `zelos` over an HTTP/XML-based RPC protocol, i.e. the python `xmlrpc` protocol.
+
+## Basic Usage
+
+To remotely debug a binary with default options:
+
+```console
+$ python -m zelos.tools.zdbserver my_binary
+```
+
+All the standard `zelos` flags can be used here as well. By default, the debug server is hosted on http://localhost:62433. The port can be changed:
+
+```console
+$ python -m zelos.tools.zdbserver --debug_port 555 my_binary
+```
+
+Currently, the only `zdb` client is an `angr` [tool](https://github.com/zeropointdynamics/angr-zelos-target) that integrates symbolic execution with `zelos`.

src/zelos/tools/zdbserver/__init__.py

@@ -0,0 +1,26 @@
+# Copyright (C) 2020 Zeropoint Dynamics
+
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as
+# published by the Free Software Foundation, either version 3 of the
+# License, or (at your option) any later version.
+
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+
+# You should have received a copy of the GNU Affero General Public
+# License along with this program.  If not, see
+# <http://www.gnu.org/licenses/>.
+# ======================================================================
+
+from .zdbserver import (
+    DEFAULT_INTERFACE,
+    DEFAULT_PORT,
+    ZdbServer,
+    create_server,
+)
+
+
+__all__ = ["ZdbServer", "create_server", "DEFAULT_INTERFACE", "DEFAULT_PORT"]

src/zelos/tools/zdbserver/__main__.py

@@ -0,0 +1,24 @@
+# Copyright (C) 2020 Zeropoint Dynamics
+
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as
+# published by the Free Software Foundation, either version 3 of the
+# License, or (at your option) any later version.
+
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+
+# You should have received a copy of the GNU Affero General Public
+# License along with this program.  If not, see
+# <http://www.gnu.org/licenses/>.
+# ======================================================================
+
+if __name__ == "__main__":
+    import sys
+
+    from .zdbserver import create_server
+
+    server = create_server(sys.argv[1:])
+    server.serve_forever()

src/zelos/tools/zdbserver/test_zdbserver.py

@@ -0,0 +1,206 @@
+# Copyright (C) 2020 Zeropoint Dynamics
+
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as
+# published by the Free Software Foundation, either version 3 of the
+# License, or (at your option) any later version.
+
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+
+
+# You should have received a copy of the GNU Affero General Public
+# License along with this program.  If not, see
+# <http://www.gnu.org/licenses/>.
+# ======================================================================
+
+import multiprocessing
+import unittest
+import xmlrpc
+
+from os import path
+
+import pytest
+
+from zelos.tools.zdbserver import DEFAULT_PORT, create_server
+
+
+DATA_DIR = path.join(
+    path.dirname(path.abspath(__file__)), "../../../../tests/data"
+)
+
+
+class Transport(xmlrpc.client.Transport):
+    def __init__(self, use_datetime=False, use_builtin_types=False, timeout=1):
+        super().__init__(
+            use_datetime=use_datetime, use_builtin_types=use_builtin_types
+        )
+        self.timeout = timeout
+
+    def make_connection(self, host):
+        conn = super(Transport, self).make_connection(host)
+        conn.timeout = self.timeout
+        return conn
+
+
+class TestZdbServer(unittest.TestCase):
+    def start_server(
+        self,
+        server_url=f"http://localhost:{DEFAULT_PORT}",
+        target=path.join(DATA_DIR, "static_elf_helloworld"),
+    ) -> (multiprocessing.Process, xmlrpc.client.ServerProxy):
+
+        srv_ready = multiprocessing.Event()
+        srv_error = multiprocessing.Event()
+
+        def server_process():
+            try:
+                rpc_server = create_server(target)
+                srv_ready.set()
+                rpc_server.serve_forever()
+            except Exception as e:
+                print("start_server Exception:", e)
+                srv_error.set()
+                srv_ready.set()
+
+        srv = multiprocessing.Process(target=server_process)
+        srv.start()
+        srv_ready.wait(10)
+        self.assertTrue(srv_ready.is_set())
+        self.assertFalse(srv_error.is_set())
+
+        rpc = xmlrpc.client.ServerProxy(
+            server_url, transport=Transport(timeout=10)
+        )
+
+        return srv, rpc
+
+    def stop_server(
+        self, srv: multiprocessing.Process, rpc: xmlrpc.client.ServerProxy
+    ) -> None:
+        rpc.server_shutdown()
+        srv.join(5)
+        self.assertFalse(srv.is_alive())
+        if srv.is_alive():
+            srv.terminate()
+
+    # @pytest.mark.usefixtures("serial")
+    # def test_server_connect(self):
+    #     proc, rpc = self.start_server()
+    #     self.stop_server(proc, rpc)
+
+    @pytest.mark.usefixtures("serial")
+    def test_server_api(self):
+        # Do this all in one test so we're not creating many
+        # sub-processes.
+        proc, zdb = self.start_server()
+
+        zdb_exception = None
+
+        try:
+            self.assertEqual(
+                zdb.get_filepath(),
+                path.join(DATA_DIR, "static_elf_helloworld"),
+            )
+
+            # Test syscall breaks
+            zdb.set_syscall_breakpoint("brk")
+            break_state = zdb.run()
+            del break_state["syscall"]["retval"]  # address may change
+            self.assertEqual(
+                str(break_state),
+                "{'pc': '0x815b577', 'syscall': {'name': 'brk', 'args': "
+                "[{'type': 'void*', 'name': 'addr', 'value': '0x0'}], "
+                "'retval_register': 'eax'}, 'bits': 32}",
+            )
+
+            # Test syscall break argument values
+            break_state = zdb.run()
+            brk_target_addr = int(
+                break_state["syscall"]["args"][0]["value"], 16
+            )
+            # Address may change, so just check it is non-zero
+            self.assertNotEqual(0, brk_target_addr)
+
+            zdb.remove_syscall_breakpoint("brk")
+
+            # Test address breaks
+            zdb.set_breakpoint("0x080eccf7", False)
+            break_state = zdb.run()
+            self.assertEqual(
+                str(break_state),
+                "{'pc': '0x80eccf7', 'syscall': {}, 'bits': 32}",
+            )
+            zdb.remove_breakpoint("0x080eccf7")
+
+            # Test read/write register
+            pc = int(zdb.read_register("pc"), 16)
+            self.assertEqual(pc, 0x80ECCF7)
+            zdb.write_register("pc", "0xdeadbeef")
+            pc = int(zdb.read_register("eip"), 16)
+            self.assertEqual(pc, 0xDEADBEEF)
+            zdb.write_register("eip", "0x80eccf7")
+            pc = int(zdb.read_register("pc"), 16)
+            self.assertEqual(pc, 0x80ECCF7)
+
+            ecx = int(zdb.read_register("ecx"), 16)
+            self.assertEqual(ecx, 0x81E9CA0)
+
+            # Test `zdbserver` exceptions
+            # ...
+
+            # Test memory read after break
+            stack_var = int(zdb.read_register("esp"), 16) + 0x1C
+            data = zdb.read_memory(f"0x{stack_var:x}", 4)
+            print(data.data)
+            # self.assertEqual(data.data, b"\xc0\x00\x00\x90")  # after inst
+            self.assertEqual(data.data, b"\x80\x00\x00\x90")  # before inst
+
+            # Test memory map list
+            mappings = zdb.get_mappings()
+            self.assertEqual(len(mappings), 21)
+            mr = mappings[0]
+            self.assertTrue("start_address" in mr)
+            self.assertTrue("end_address" in mr)
+
+            # Test memory write by modifying a `write` buffer just
+            # before the syscall.
+            zdb.set_breakpoint("0x08106b04", True)
+            break_state = zdb.run()
+            buf = zdb.read_register("edi")
+            zdb.write_memory(buf, b"Hello World! I'm a Zelos Test!")
+
+            # Test watchpoint
+            zdb.set_watchpoint(buf, True, True, False)
+            # break_state = zdb.run()
+            # self.assertEqual(break_state, "asdf")
+            zdb.remove_watchpoint(buf)
+
+            # Test memory write, continued
+            zdb.set_syscall_breakpoint("write")
+            break_state = zdb.run()
+            buf = break_state["syscall"]["args"][1]["value"]
+            count = int(break_state["syscall"]["args"][2]["value"], 16)
+            stdout = zdb.read_memory(buf, count)
+            self.assertEqual(stdout, b"Hello World! I'm a Zelos Test!")
+
+            # Run until program end
+            break_state = zdb.run()
+            self.assertEqual(break_state, {})
+
+            zdb.stop()
+
+        except Exception as e:
+            zdb_exception = e
+
+        finally:
+            self.stop_server(proc, zdb)
+
+        if zdb_exception is not None:
+            self.fail("test_server_api error: " + str(zdb_exception))
+
+
+if __name__ == "__main__":
+    unittest.main()