more detailed tenda CI test and cleanup elf multithrad http test

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__":

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