tlsfuzzer/extract.py: Add support for {r,s} tuples in sigs file

tests/measurements_test_files/data_r_and_s.bin

@@ -0,0 +1,4 @@
+6-0ßÒ
+ ²Àô"6ßPå·$ÅËÈY`Éu׌¸à0IÉÛI”ß—ÊŽ>î=[EþNØFÈ
+:¯Iš­ßœ
>ðkâ͆§?E‰-»G,Ç[„åÞ|ÝþѸÝBK¬Ò¡?u	Éð˽2`—²§9üIŒe1·3ò‹rÉ{”Òo뿦Q´;K«k™v[Kq_îÙz$¶b©|;$Ip%ÚG#q#èÊSq€¤ÊÐä'‰Ù{K§g%,ñ“Vü{wpõAÌQ¢ÞJb8†p¢.B_ôé•Ï¡Xï®±Mûo8‡
æ*t4¡O.<ÁùÁ¼Ö¡¤òä#;ñ2Ü#lã +ôÒ¹a	 ê58´Ÿ¤•g‰³òxĆò+
+-:œg5—â™1âOV+S•ÊezÁšÍ»æó˜bi

tests/measurements_test_files/priv_key_r_and_s.pem

@@ -0,0 +1,5 @@
+-----BEGIN PRIVATE KEY-----
+MIGHAgEBMBMGByqGSM49AgEGCCqGSM49AwEHBG0wawIBAQQgmBZ3w97A7vU3uUW3R05Xrrtuz25s
+QRMXx05VOo410f+hRANCAAQwiv85bw4LWmyCH3+CXxbnIx5VkhWOeN07XXG0gSiIaFLe36c5aDnN
+vndtWj9hvWxgXK4UIZ3G0MNrtv9pDNW2
+-----END PRIVATE KEY-----

tests/test_tlsfuzzer_extract.py

@@ -16,6 +16,7 @@
 from os.path import join, dirname, abspath
 import hashlib
 from random import choice
+import ecdsa
 
 from tlsfuzzer.utils.log import Log
 
@@ -266,7 +267,8 @@ def test_command_line(self, mock_parse, mock_write, mock_write_pkt,
                     no_quickack=False, delay=None, carriage_return=None,
                     data=None, data_size=None, sigs=None, priv_key=None,
                     key_type=None, frequency=None, hash_func=hashlib.sha256,
-                    workers=None, verbose=False, rsa_keys=None)
+                    workers=None, verbose=False, rsa_keys=None,
+                    sig_format="DER")
                 mock_measurements.assert_not_called()
 
     @mock.patch(
@@ -303,7 +305,8 @@ def test_delay_and_CR(self, mock_parse, mock_write, mock_write_pkt,
                     no_quickack=False, delay=3.5, carriage_return='\n',
                     data=None, data_size=None, sigs=None, priv_key=None,
                     key_type=None, frequency=None, hash_func=hashlib.sha256,
-                    workers=None, verbose=False, rsa_keys=None)
+                    workers=None, verbose=False, rsa_keys=None,
+                    sig_format="DER")
                 mock_measurements.assert_not_called()
 
     @mock.patch(
@@ -339,7 +342,8 @@ def test_no_quickack(self, mock_parse, mock_write, mock_write_pkt,
                     no_quickack=True, delay=None, carriage_return=None,
                     data=None, data_size=None, sigs=None, priv_key=None,
                     key_type=None, frequency=None, hash_func=hashlib.sha256,
-                    workers=None, verbose=False, rsa_keys=None)
+                    workers=None, verbose=False, rsa_keys=None,
+                    sig_format="DER")
                 mock_measurements.assert_not_called()
 
     @mock.patch(
@@ -372,7 +376,8 @@ def test_raw_times(self, mock_parse, mock_write, mock_write_pkt, mock_log,
                     no_quickack=False, delay=None, carriage_return=None,
                     data=None, data_size=None, sigs=None, priv_key=None,
                     key_type=None, frequency=None, hash_func=hashlib.sha256,
-                    workers=None, verbose=False, rsa_keys=None)
+                    workers=None, verbose=False, rsa_keys=None,
+                    sig_format="DER")
                 mock_measurements.assert_not_called()
 
     @mock.patch(
@@ -406,7 +411,8 @@ def test_raw_binary_times(self, mock_parse, mock_write, mock_write_pkt,
                     no_quickack=False, delay=None, carriage_return=None,
                     data=None, data_size=None, sigs=None, priv_key=None,
                     key_type=None, frequency=None, hash_func=hashlib.sha256,
-                    workers=None, verbose=False, rsa_keys=None)
+                    workers=None, verbose=False, rsa_keys=None,
+                    sig_format="DER")
                 mock_measurements.assert_not_called()
 
     @mock.patch('tlsfuzzer.extract.Log')
@@ -510,7 +516,8 @@ def test_raw_times_with_column_name(self, mock_parse, mock_write,
                     no_quickack=False, delay=None, carriage_return=None,
                     data=None, data_size=None, sigs=None, priv_key=None,
                     key_type=None, frequency=None, hash_func=hashlib.sha256,
-                    workers=None, verbose=False, rsa_keys=None)
+                    workers=None, verbose=False, rsa_keys=None,
+                    sig_format="DER")
                 mock_measurements.assert_not_called()
 
     @mock.patch('__main__.__builtins__.print')
@@ -582,7 +589,7 @@ def test_rsa_keys_options(self, mock_parse, mock_write, mock_process,
                     data=None, data_size=None, sigs=None,
                     priv_key=None, key_type=None, frequency=None,
                     hash_func=hashlib.sha256, workers=None, verbose=False,
-                    rsa_keys=priv_key)
+                    rsa_keys=priv_key, sig_format="DER")
                 mock_write.assert_not_called()
                 mock_write_pkt.assert_not_called()
                 mock_log.assert_not_called()
@@ -623,7 +630,7 @@ def test_ecdsa_signs_options(self, mock_parse, mock_process, mock_write,
                     data=raw_data, data_size=data_size, sigs=raw_sigs,
                     priv_key=priv_key, key_type="ecdsa", frequency=None,
                     hash_func=hashlib.sha256, workers=None, verbose=False,
-                    rsa_keys=None)
+                    rsa_keys=None, sig_format="DER")
                 mock_write.assert_not_called()
                 mock_write_pkt.assert_not_called()
                 mock_log.assert_not_called()
@@ -664,7 +671,7 @@ def test_verbose_option(self, mock_parse, mock_process, mock_write,
                     data=raw_data, data_size=data_size, sigs=raw_sigs,
                     priv_key=priv_key, key_type="ecdsa", frequency=None,
                     hash_func=hashlib.sha256, workers=None, verbose=True,
-                    rsa_keys=None)
+                    rsa_keys=None, sig_format="DER")
                 mock_write.assert_not_called()
                 mock_write_pkt.assert_not_called()
                 mock_log.assert_not_called()
@@ -706,7 +713,8 @@ def test_frequency_option(self, mock_parse, mock_process, mock_write,
                     data=raw_data, data_size=data_size, sigs=raw_sigs,
                     priv_key=priv_key, key_type="ecdsa",
                     frequency=frequency * 1e6, hash_func=hashlib.sha256,
-                    workers=None, verbose=False, rsa_keys=None)
+                    workers=None, verbose=False, rsa_keys=None,
+                    sig_format="DER")
                 mock_write.assert_not_called()
                 mock_write_pkt.assert_not_called()
                 mock_log.assert_not_called()
@@ -748,7 +756,8 @@ def test_hash_func_option(self, mock_parse, mock_process, mock_write,
                     data=raw_data, data_size=data_size, sigs=raw_sigs,
                     priv_key=priv_key, key_type="ecdsa",
                     frequency=None, hash_func=hashlib.sha384,
-                    workers=None, verbose=False, rsa_keys=None)
+                    workers=None, verbose=False, rsa_keys=None,
+                    sig_format="DER")
                 mock_write.assert_not_called()
                 mock_write_pkt.assert_not_called()
                 mock_log.assert_not_called()
@@ -789,7 +798,8 @@ def test_prehashed_option(self, mock_parse, mock_process, mock_write,
                     data=raw_data, data_size=data_size, sigs=raw_sigs,
                     priv_key=priv_key, key_type="ecdsa",
                     frequency=None, hash_func=None,
-                    workers=None, verbose=False, rsa_keys=None)
+                    workers=None, verbose=False, rsa_keys=None,
+                    sig_format="DER")
                 mock_write.assert_not_called()
                 mock_write_pkt.assert_not_called()
                 mock_log.assert_not_called()
@@ -831,7 +841,8 @@ def test_workers_option(self, mock_parse, mock_process, mock_write,
                     data=raw_data, data_size=data_size, sigs=raw_sigs,
                     priv_key=priv_key, key_type="ecdsa",
                     frequency=None, hash_func=hashlib.sha256,
-                    workers=workers, verbose=False, rsa_keys=None)
+                    workers=workers, verbose=False, rsa_keys=None,
+                    sig_format="DER")
                 mock_write.assert_not_called()
                 mock_write_pkt.assert_not_called()
                 mock_log.assert_not_called()
@@ -872,7 +883,8 @@ def test_skip_invert_option(self, mock_parse, mock_process, mock_write,
                     data=raw_data, data_size=data_size, sigs=raw_sigs,
                     priv_key=priv_key, key_type="ecdsa",
                     frequency=None, hash_func=hashlib.sha256,
-                    workers=None, verbose=False, rsa_keys=None)
+                    workers=None, verbose=False, rsa_keys=None,
+                    sig_format="DER")
                 mock_write.assert_not_called()
                 mock_write_pkt.assert_not_called()
                 mock_log.assert_not_called()
@@ -882,6 +894,48 @@ def test_skip_invert_option(self, mock_parse, mock_process, mock_write,
                 for mode in files_passes_in_process.values():
                     self.assertNotIn("invert", mode)
 
+    @mock.patch('tlsfuzzer.extract.Log')
+    @mock.patch('tlsfuzzer.extract.Extract._write_pkts')
+    @mock.patch('tlsfuzzer.extract.Extract._write_csv')
+    @mock.patch(
+        'tlsfuzzer.extract.Extract.process_and_create_multiple_csv_files'
+    )
+    @mock.patch('tlsfuzzer.extract.Extract.parse')
+    def test_raw_sig_format_option(self, mock_parse, mock_process, mock_write,
+                              mock_write_pkt, mock_log):
+        output = "/tmp"
+        raw_data = "/tmp/data"
+        data_size = 32
+        raw_sigs = "/tmp/sigs"
+        raw_times = "/tmp/times"
+        priv_key = "/tmp/key"
+        args = ["extract.py",
+                "-o", output,
+                "--raw-data", raw_data,
+                "--data-size", data_size,
+                "--raw-sigs", raw_sigs,
+                "--raw-times", raw_times,
+                "--priv-key-ecdsa", priv_key,
+                "--sig-format", "RAW"]
+        mock_init = mock.Mock()
+        mock_init.return_value = None
+        with mock.patch('tlsfuzzer.extract.Extract.__init__', mock_init):
+            with mock.patch("sys.argv", args):
+                main()
+                mock_init.assert_called_once_with(
+                    mock.ANY, None, output, None, None,
+                    raw_times, None, binary=None, endian="little",
+                    no_quickack=False, delay=None, carriage_return=None,
+                    data=raw_data, data_size=data_size, sigs=raw_sigs,
+                    priv_key=priv_key, key_type="ecdsa",
+                    frequency=None, hash_func=hashlib.sha256,
+                    workers=None, verbose=False, rsa_keys=None,
+                    sig_format="RAW")
+                mock_write.assert_not_called()
+                mock_write_pkt.assert_not_called()
+                mock_log.assert_not_called()
+                mock_process.assert_called_once()
+
     def test_specify_to_private_keys(self):
         args = [
             "extract.py", "-o", "/tmp", "--raw-data", "/tmp/data",
@@ -1383,15 +1437,16 @@ def setUp(self):
         self.times_used_write_on_hamming = 0
         self.k_time_map = []
 
-        out_dir = join(dirname(abspath(__file__)), "measurements_test_files")
+        common_dir = "measurements_test_files"
+        out_dir = join(dirname(abspath(__file__)), common_dir)
         raw_times = join(dirname(abspath(__file__)),
-                         "measurements_test_files", "times.bin")
+                         common_dir, "times.bin")
         raw_sigs = join(dirname(abspath(__file__)),
-                         "measurements_test_files", "sigs.bin")
+                         common_dir, "sigs.bin")
         raw_data = join(dirname(abspath(__file__)),
-                         "measurements_test_files", "data.bin")
+                         common_dir, "data.bin")
         priv_key = join(dirname(abspath(__file__)),
-                         "measurements_test_files", "priv_key.pem")
+                         common_dir, "priv_key.pem")
 
         self.extract = Extract(
             output=out_dir, raw_times=raw_times, binary=8,
@@ -1455,6 +1510,7 @@ def test_measurement_creation_with_verbose_and_frequency(
         ):
         self.extract.frequency = 1
         self.extract.verbose = True
+        self.k_time_map = []
 
         mock_file.side_effect = self.file_emulator
         self.times_used_write = 0
@@ -1471,6 +1527,43 @@ def test_measurement_creation_with_verbose_and_frequency(
         self.times_used_write = 0
         self.extract.frequency = None
         self.extract.verbose = False
+        self.k_time_map = []
+
+    @mock.patch('builtins.print')
+    def test_measurement_creation_raw_sigs(self, mock_print):
+        self.k_time_map = []
+        common_dir = "measurements_test_files"
+        out_dir = join(dirname(abspath(__file__)), common_dir)
+        raw_times = join(dirname(abspath(__file__)),
+                         common_dir, "times_r_and_s.bin")
+        raw_sigs = join(dirname(abspath(__file__)),
+                         common_dir, "sigs_r_and_s.bin")
+        raw_data = join(dirname(abspath(__file__)),
+                         common_dir, "data_r_and_s.bin")
+        priv_key = join(dirname(abspath(__file__)),
+                         common_dir, "priv_key_r_and_s.pem")
+
+        extract = Extract(
+            output=out_dir, raw_times=raw_times, binary=8,
+            sigs=raw_sigs, data=raw_data, data_size=32, priv_key=priv_key,
+            key_type="ecdsa", hash_func=None, sig_format="RAW"
+        )
+
+        self.times_used_write = 0
+
+        with mock.patch('__main__.__builtins__.open') as mock_file:
+            mock_file.side_effect = self.file_emulator
+            extract.process_measurements_and_create_csv_file(
+                extract.ecdsa_iter(), extract.ecdsa_max_value()
+            )
+
+        self.assertGreater(
+            self.times_used_write, 0,
+            "At least one measurement should have been written."
+        )
+
+        self.times_used_write = 0
+        self.k_time_map = []
 
     @mock.patch('__main__.__builtins__.open')
     def test_measurement_creation_with_k_size_invert(
@@ -1641,6 +1734,93 @@ def test_measurement_creation_with_incomplete_times(
 
         self.extract.output = original_output
 
+    @mock.patch('builtins.print')
+    @mock.patch('__main__.__builtins__.open')
+    def test_measurement_creation_with_misformated_sigs(
+            self, mock_file, mock_print
+        ):
+        def custom_file_emulator_creator(misformated_sig):
+            def custom_file_emulator(*args, **kwargs):
+                name = args[0]
+                try:
+                    mode = args[1]
+                except IndexError:
+                    mode = 'r'
+
+                if type(name) == int:
+                    return self.builtin_open(*args, **kwargs)
+
+                if "w" in mode:
+                    r = mock.mock_open()(name, mode)
+                    r.write.side_effect = None
+                    return r
+
+                r = mock.mock_open(
+                    read_data=misformated_sig
+                )(name, mode)
+                # r.write.side_effect = lambda s: (
+                #     self.k_time_map.append(s[:-1])
+                # )
+                return r
+
+            return custom_file_emulator
+
+        # Test 1: No sequence in the beginning
+        mock_file.side_effect = custom_file_emulator_creator(b"\x20")
+
+        with self.assertRaises(ValueError) as e:
+            self.extract.process_measurements_and_create_csv_file(
+                self.extract.ecdsa_iter(), self.extract.ecdsa_max_value()
+            )
+
+        self.assertIn("There was an error in parsing signatures",
+                      str(e.exception))
+
+        # Test 2: No length after sequence
+        mock_file.side_effect = custom_file_emulator_creator(b"\x30")
+
+        with self.assertRaises(ValueError) as e:
+            self.extract.process_measurements_and_create_csv_file(
+                self.extract.ecdsa_iter(), self.extract.ecdsa_max_value()
+            )
+
+        self.assertIn("Couldn't read size of a signature.", str(e.exception))
+
+        # Test 3: Only sequence and length
+        mock_file.side_effect = custom_file_emulator_creator(b"\x30\x23")
+
+        with self.assertRaises(ValueError) as e:
+            self.extract.process_measurements_and_create_csv_file(
+                self.extract.ecdsa_iter(), self.extract.ecdsa_max_value()
+            )
+
+        self.assertIn("Signature file ended unexpectedly.", str(e.exception))
+
+        # Test 4: Sequence and length but not enough data afterwards
+        mock_file.side_effect = custom_file_emulator_creator(
+            b"\x30\x23" + (b"\x10" * 5))
+
+        with self.assertRaises(ValueError) as e:
+            self.extract.process_measurements_and_create_csv_file(
+                self.extract.ecdsa_iter(), self.extract.ecdsa_max_value()
+            )
+
+        self.assertIn("Signature file ended unexpectedly.", str(e.exception))
+
+        # Test 5: Raw signature not enough bytes
+        self.extract.r_or_s_size = 32
+        mock_file.side_effect = custom_file_emulator_creator(
+            b"\x30\x23\x20" * 10)
+
+        with self.assertRaises(ValueError) as e:
+            self.extract.process_measurements_and_create_csv_file(
+                self.extract.ecdsa_iter(), self.extract.ecdsa_max_value()
+            )
+
+        self.assertIn("Incomplete r or s values in binary file.",
+                      str(e.exception))
+        self.extract.r_or_s_size = None
+
     @mock.patch('__main__.__builtins__.print')
     @mock.patch('__main__.__builtins__.open')
     def test_multiple_measurement_creation(