• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1#!/usr/bin/env python3
2#
3#  Copyright The Mbed TLS Contributors
4#  SPDX-License-Identifier: Apache-2.0
5#
6#  Licensed under the Apache License, Version 2.0 (the "License"); you may
7#  not use this file except in compliance with the License.
8#  You may obtain a copy of the License at
9#
10#  http://www.apache.org/licenses/LICENSE-2.0
11#
12#  Unless required by applicable law or agreed to in writing, software
13#  distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
14#  WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15#  See the License for the specific language governing permissions and
16#  limitations under the License.
17#
18
19"""
20Make fuzz like testing for pkcs7 tests
21Given a valid DER pkcs7 file add tests to the test_suite_pkcs7.data file
22 - It is expected that the pkcs7_asn1_fail( data_t *pkcs7_buf )
23    function is defined in test_suite_pkcs7.function
24 - This is not meant to be portable code, if anything it is meant to serve as
25   documentation for showing how those ugly tests in test_suite_pkcs7.data were created
26"""
27
28
29import sys
30from os.path import exists
31
32PKCS7_TEST_FILE = "../suites/test_suite_pkcs7.data"
33
34class Test: # pylint: disable=too-few-public-methods
35    """
36    A instance of a test in test_suite_pkcs7.data
37    """
38    def __init__(self, name, depends, func_call):
39        self.name = name
40        self.depends = depends
41        self.func_call = func_call
42
43    # pylint: disable=no-self-use
44    def to_string(self):
45        return "\n" + self.name + "\n" + self.depends + "\n" + self.func_call + "\n"
46
47class TestData:
48    """
49    Take in test_suite_pkcs7.data file.
50    Allow for new tests to be added.
51    """
52    mandatory_dep = "MBEDTLS_SHA256_C"
53    test_name = "PKCS7 Parse Failure Invalid ASN1"
54    test_function = "pkcs7_asn1_fail:"
55    def __init__(self, file_name):
56        self.file_name = file_name
57        self.last_test_num, self.old_tests = self.read_test_file(file_name)
58        self.new_tests = []
59
60    # pylint: disable=no-self-use
61    def read_test_file(self, file):
62        """
63        Parse the test_suite_pkcs7.data file.
64        """
65        tests = []
66        if not exists(file):
67            print(file + " Does not exist")
68            sys.exit()
69        with open(file, "r", encoding='UTF-8') as fp:
70            data = fp.read()
71        lines = [line.strip() for line in data.split('\n') if len(line.strip()) > 1]
72        i = 0
73        while i < len(lines):
74            if "depends" in lines[i+1]:
75                tests.append(Test(lines[i], lines[i+1], lines[i+2]))
76                i += 3
77            else:
78                tests.append(Test(lines[i], None, lines[i+1]))
79                i += 2
80        latest_test_num = float(tests[-1].name.split('#')[1])
81        return latest_test_num, tests
82
83    def add(self, name, func_call):
84        self.last_test_num += 1
85        self.new_tests.append(Test(self.test_name + ": " + name +  " #" + \
86                str(self.last_test_num), "depends_on:" + self.mandatory_dep, \
87                self.test_function + '"' + func_call + '"'))
88
89    def write_changes(self):
90        with open(self.file_name, 'a', encoding='UTF-8') as fw:
91            fw.write("\n")
92            for t in self.new_tests:
93                fw.write(t.to_string())
94
95
96def asn1_mutate(data):
97    """
98    We have been given an asn1 structure representing a pkcs7.
99    We want to return an array of slightly modified versions of this data
100    they should be modified in a way which makes the structure invalid
101
102    We know that asn1 structures are:
103    |---1 byte showing data type---|----byte(s) for length of data---|---data content--|
104    We know that some data types can contain other data types.
105    Return a dictionary of reasons and mutated data types.
106    """
107
108    # off the bat just add bytes to start and end of the buffer
109    mutations = []
110    reasons = []
111    mutations.append(["00"] + data)
112    reasons.append("Add null byte to start")
113    mutations.append(data + ["00"])
114    reasons.append("Add null byte to end")
115    # for every asn1 entry we should attempt to:
116    #    - change the data type tag
117    #    - make the length longer than actual
118    #    - make the length shorter than actual
119    i = 0
120    while i < len(data):
121        tag_i = i
122        leng_i = tag_i + 1
123        data_i = leng_i + 1 + (int(data[leng_i][1], 16) if data[leng_i][0] == '8' else 0)
124        if data[leng_i][0] == '8':
125            length = int(''.join(data[leng_i + 1: data_i]), 16)
126        else:
127            length = int(data[leng_i], 16)
128
129        tag = data[tag_i]
130        print("Looking at ans1: offset " + str(i) + " tag = " + tag + \
131                ", length = " + str(length)+ ":")
132        print(''.join(data[data_i:data_i+length]))
133        # change tag to something else
134        if tag == "02":
135            # turn integers into octet strings
136            new_tag = "04"
137        else:
138            # turn everything else into an integer
139            new_tag = "02"
140        mutations.append(data[:tag_i] + [new_tag] + data[leng_i:])
141        reasons.append("Change tag " + tag + " to " + new_tag)
142
143        # change lengths to too big
144        # skip any edge cases which would cause carry over
145        if int(data[data_i - 1], 16) < 255:
146            new_length = str(hex(int(data[data_i - 1], 16) + 1))[2:]
147            if len(new_length) == 1:
148                new_length = "0"+new_length
149            mutations.append(data[:data_i -1] + [new_length] + data[data_i:])
150            reasons.append("Change length from " + str(length) + " to " \
151                    + str(length + 1))
152            # we can add another test here for tags that contain other tags \
153            # where they have more data than there containing tags account for
154            if tag in ["30", "a0", "31"]:
155                mutations.append(data[:data_i -1] + [new_length] + \
156                        data[data_i:data_i + length] + ["00"] + \
157                        data[data_i + length:])
158                reasons.append("Change contents of tag " + tag + " to contain \
159                        one unaccounted extra byte")
160        # change lengths to too small
161        if int(data[data_i - 1], 16) > 0:
162            new_length = str(hex(int(data[data_i - 1], 16) - 1))[2:]
163            if len(new_length) == 1:
164                new_length = "0"+new_length
165            mutations.append(data[:data_i -1] + [new_length] + data[data_i:])
166            reasons.append("Change length from " + str(length) + " to " + str(length - 1))
167
168        # some tag types contain other tag types so we should iterate into the data
169        if tag in ["30", "a0", "31"]:
170            i = data_i
171        else:
172            i = data_i + length
173
174    return list(zip(reasons, mutations))
175
176if __name__ == "__main__":
177    if len(sys.argv) < 2:
178        print("USAGE: " + sys.argv[0] + " <pkcs7_der_file>")
179        sys.exit()
180
181    DATA_FILE = sys.argv[1]
182    TEST_DATA = TestData(PKCS7_TEST_FILE)
183    with open(DATA_FILE, 'rb') as f:
184        DATA_STR = f.read().hex()
185    # make data an array of byte strings eg ['de','ad','be','ef']
186    HEX_DATA = list(map(''.join, [[DATA_STR[i], DATA_STR[i+1]] for i in range(0, len(DATA_STR), \
187            2)]))
188    # returns tuples of test_names and modified data buffers
189    MUT_ARR = asn1_mutate(HEX_DATA)
190
191    print("made " + str(len(MUT_ARR)) + " new tests")
192    for new_test in MUT_ARR:
193        TEST_DATA.add(new_test[0], ''.join(new_test[1]))
194
195    TEST_DATA.write_changes()
196