1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
|
"""Integration tests for the Ghidra JSON-lines control bridge."""
from __future__ import annotations
import json
import os
import shutil
import socket
import unittest
import greeneye as ge
from greeneye.gdbridge import GhidraBridge
from greeneye.ghidra import render_ghidra_command
HERE = os.path.dirname(os.path.abspath(__file__))
DEMO_ELF = os.path.join(HERE, "..", "firmware", "uproc_demo", "build", "uproc_demo.elf")
class GhidraCommandTests(unittest.TestCase):
def test_render_command_names_project_program_and_hits(self):
cmd = render_ghidra_command("/proj", "P", "gsp.bin", "/tmp/hits.tsv")
self.assertIn("analyzeHeadless /proj P", cmd)
self.assertIn("-process gsp.bin", cmd)
self.assertIn("MarkGreenEyeHits.java /tmp/hits.tsv", cmd)
def qemu_available() -> bool:
return bool(os.environ.get("GREENEYE_QEMU") or shutil.which("qemu-system-riscv64"))
@unittest.skipUnless(qemu_available(), "qemu-system-riscv64 not installed")
@unittest.skipUnless(os.path.exists(DEMO_ELF), "demo firmware not built")
class BridgeTests(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.session = ge.launch(kernel=DEMO_ELF, record=True, name="unittest-bridge")
cls.bridge = GhidraBridge(cls.session, port=0)
cls.port = cls.bridge.start()
@classmethod
def tearDownClass(cls):
cls.bridge.stop()
cls.session.close()
def call(self, cmd, args=None):
with socket.create_connection(("127.0.0.1", self.port), timeout=10) as sock:
stream = sock.makefile("rwb")
stream.write((json.dumps({"id": 1, "cmd": cmd, "args": args or {}}) + "\n").encode())
stream.flush()
response = json.loads(stream.readline())
self.assertTrue(response["ok"], response)
return response["result"]
def test_info_and_resolve(self):
info = self.call("info")
self.assertEqual(info["name"], "unittest-bridge")
addr = self.call("resolve", {"name": "rpc_sum"})
self.assertEqual(addr, self.session.symbols.lookup("rpc_sum"))
def test_run_until_and_hits(self):
result = self.call("run_until", {"target": "rpc_sum", "timeout": 5.0})
self.assertEqual(result["addr"], self.session.symbols.lookup("rpc_sum"))
hits = self.call("hits")
self.assertTrue(any(entry["addr"] == result["addr"] for entry in hits), hits)
def test_memory_and_registers(self):
ctrl = self.session.symbols.lookup("gsp_ctrl")
hexdata = self.call("read_mem", {"addr": ctrl, "size": 8})
self.assertEqual(len(bytes.fromhex(hexdata)), 8)
self.call("write_reg", {"name": "t3", "value": 0x1234})
self.assertEqual(self.call("read_reg", {"name": "t3"}), 0x1234)
def test_breakpoints_and_call(self):
addr = self.call("bp_set", {"target": "rpc_fill"})
self.assertEqual(addr, self.session.symbols.lookup("rpc_fill"))
self.call("bp_clear", {"addr": addr})
result = self.call("call", {"target": "rpc_ping", "args": [0x1111]})
self.assertEqual(result, 0x1111 ^ 0xBEEF)
def test_unknown_command(self):
with socket.create_connection(("127.0.0.1", self.port), timeout=10) as sock:
stream = sock.makefile("rwb")
stream.write((json.dumps({"id": 2, "cmd": "nope", "args": {}}) + "\n").encode())
stream.flush()
response = json.loads(stream.readline())
self.assertFalse(response["ok"])
self.assertIn("unknown bridge command", response["error"])
if __name__ == "__main__":
unittest.main()
|
