From b52f164004a11aea0d8be67012c896cade391557 Mon Sep 17 00:00:00 2001 From: xwings Date: Sun, 16 Oct 2022 10:48:44 +0800 Subject: [PATCH] more detailed tenda CI test and cleanup elf multithrad http test --- tests/test_elf_multithread.py | 23 +++++++++++++---------- tests/test_tendaac15_httpd.py | 35 ++++++++++++++++++++++++----------- 2 files changed, 37 insertions(+), 21 deletions(-) diff --git a/tests/test_elf_multithread.py b/tests/test_elf_multithread.py index fc7a62f3a..6188b8111 100644 --- a/tests/test_elf_multithread.py +++ b/tests/test_elf_multithread.py @@ -3,7 +3,7 @@ # Cross Platform and Multi Architecture Advanced Binary Emulation Framework # -import platform, sys, unittest, os, threading, time +import http.client, platform, sys, os, threading, time, unittest sys.path.append("..") from qiling import Qiling @@ -365,42 +365,45 @@ def picohttpd(): ql = Qiling(["../examples/rootfs/x8664_linux/bin/picohttpd","12911"], "../examples/rootfs/x8664_linux", multithread=True, verbose=QL_VERBOSE.DEBUG) ql.run() - picohttpd_therad = threading.Thread(target=picohttpd, daemon=True) picohttpd_therad.start() time.sleep(1) - f = os.popen("curl http://127.0.0.1:12911") - self.assertEqual("httpd_test_successful", f.read()) + f = http.client.HTTPConnection('localhost', 12911, timeout=10) + f.request("GET", "/") + response = f.getresponse() + self.assertEqual("httpd_test_successful", response.read().decode()) def test_http_elf_linux_arm(self): def picohttpd(): ql = Qiling(["../examples/rootfs/arm_linux/bin/picohttpd","12912"], "../examples/rootfs/arm_linux", multithread=True, verbose=QL_VERBOSE.DEBUG) ql.run() - picohttpd_therad = threading.Thread(target=picohttpd, daemon=True) picohttpd_therad.start() time.sleep(1) - f = os.popen("curl http://127.0.0.1:12912") - self.assertEqual("httpd_test_successful", f.read()) + f = http.client.HTTPConnection('localhost', 12912, timeout=10) + f.request("GET", "/") + response = f.getresponse() + self.assertEqual("httpd_test_successful", response.read().decode()) def test_http_elf_linux_armeb(self): def picohttpd(): ql = Qiling(["../examples/rootfs/armeb_linux/bin/picohttpd"], "../examples/rootfs/armeb_linux", multithread=True, verbose=QL_VERBOSE.DEBUG) ql.run() - picohttpd_thread = threading.Thread(target=picohttpd, daemon=True) picohttpd_thread.start() time.sleep(1) - f = os.popen("curl http://127.0.0.1:12913") - self.assertEqual("httpd_test_successful", f.read()) + f = http.client.HTTPConnection('localhost', 12913, timeout=10) + f.request("GET", "/") + response = f.getresponse() + self.assertEqual("httpd_test_successful", response.read().decode()) if __name__ == "__main__": diff --git a/tests/test_tendaac15_httpd.py b/tests/test_tendaac15_httpd.py index 05f73c965..fd8632b98 100644 --- a/tests/test_tendaac15_httpd.py +++ b/tests/test_tendaac15_httpd.py @@ -13,7 +13,7 @@ # notes: we are using rootfs in this example, so rootfs = squashfs-root # -import os, socket, sys, threading, unittest +import http.client, json, os, socket, sys, time, threading, unittest sys.path.append("..") from qiling import Qiling @@ -57,24 +57,37 @@ def patcher(ql): for addr in br0_addr: ql.mem.write(addr, b'lo\x00') - - def check_pc(ql): - ql.emu_stop() - - - def my_sandbox(path, rootfs): - ql = Qiling(path, rootfs, verbose=QL_VERBOSE.DEBUG) + def my_tenda(): + ql = Qiling(["../examples/rootfs/arm_tendaac15/bin/httpd"], "../examples/rootfs/arm_tendaac15", verbose=QL_VERBOSE.DEBUG) ql.add_fs_mapper("/dev/urandom","/dev/urandom") ql.hook_address(patcher, ql.loader.elf_entry) - ql.hook_address(check_pc, 0x901e24cc) ql.run() del ql if __name__ == "__main__": + + threadLock = threading.Lock() + threads = [] + nvram_listener_therad = threading.Thread(target=nvram_listener, daemon=True) - nvram_listener_therad.start() - my_sandbox(["../examples/rootfs/arm_tendaac15/bin/httpd"], "../examples/rootfs/arm_tendaac15") + mytenda_therad = threading.Thread(target=my_tenda, daemon=True) + nvram_listener_therad.start() + mytenda_therad.start() + + threads.append(nvram_listener_therad) + threads.append(mytenda_therad) + + time.sleep(5) + + conn = http.client.HTTPConnection('localhost', 8080, timeout=10) + headers = {'X-Requested-With': 'XMLHttpRequest', 'Content-Type': 'application/x-www-form-urlencoded'} + web_data = {'page': 'CCCCAAAA', 'entrys':'sync'} + json_data = json.dumps(web_data) + conn.request('POST', '/goform/addressNat', json_data, headers) + response = conn.getresponse() + self.assertIn("Please update your documents to reflect the new location.", response.read().decode()) + if __name__ == "__main__": unittest.main() \ No newline at end of file