63c3cd747d701bb7386a944161c14a247b498068
[samba.git] / lib / fuzzing / decode_ndr_X_crash
1 #!/usr/bin/env python3
2 #
3 # Interpret a file that crashes an fuzz_ndr_X binary.
4 #
5 # Copyright (C) Catalyst IT Ltd. 2019
6
7
8 import sys
9 import os
10 from base64 import b64encode
11 import struct
12 import argparse
13 import re
14
15 TYPE_MASK = 3
16 TYPES = ['struct', 'in', 'out']
17
18 FLAGS = [
19     (4, 'ndr64', '--ndr64'),
20 ]
21
22
23 def print_if_verbose(*args, **kwargs):
24     if verbose:
25         print(*args, **kwargs)
26
27
28 def process_one_file(f):
29     print_if_verbose(f.name)
30     print_if_verbose('-' * len(f.name))
31
32     b = f.read()
33     flags, function = struct.unpack('<HH', b[:4])
34     if opnum is not None and opnum != function:
35         return
36
37     t = TYPES[flags & TYPE_MASK]
38     if ndr_type and ndr_type != t:
39         return
40
41     payload = b[4:]
42     data64 = b64encode(payload).decode('utf-8')
43
44     cmd = ['bin/ndrdump',
45            pipe,
46            str(function),
47            t,
48            '--base64-input',
49            '--input', data64,
50     ]
51
52     for flag, name, option in FLAGS:
53         if flags & flag:
54             print_if_verbose("flag: %s" % name)
55             cmd.append(option)
56
57     print_if_verbose("length: %d\n" % len(payload))
58     print(' '.join(cmd))
59     print_if_verbose()
60
61
62 def main():
63     parser = argparse.ArgumentParser()
64     parser.add_argument('-p', '--pipe', default='$PIPE',
65                         help='pipe name (for output command line)')
66     parser.add_argument('-t', '--type', default=None, choices=TYPES,
67                         help='restrict to this type')
68     parser.add_argument('-o', '--opnum', default=None, type=int,
69                         help='restrict to this function/struct number')
70     parser.add_argument('FILES', nargs='*', default=(),
71                         help="read from these files")
72     parser.add_argument('-k', '--ignore-errors', action='store_true',
73                         help='do not stop on errors')
74     parser.add_argument('-v', '--verbose', action='store_true',
75                         help='say more')
76     parser.add_argument('-H', '--honggfuzz-file',
77                         help="extract crashes from this honggfuzz report")
78     parser.add_argument('-f', '--crash-filter',
79                         help="only print crashes matching this rexexp")
80
81     args = parser.parse_args()
82
83     global pipe, opnum, ndr_type, verbose
84     pipe = args.pipe
85     opnum = args.opnum
86     ndr_type = args.type
87     verbose = args.verbose
88
89     if not args.FILES and not args.honggfuzz_file:
90         parser.print_usage()
91         sys.exit(1)
92
93     for fn in args.FILES:
94         if args.crash_filter is not None:
95             if not re.search(args.crash_filter, fn):
96                 print_if_verbose(f"skipping {fn}")
97                 continue
98         try:
99             if fn == '-':
100                 process_one_file(sys.stdin)
101             else:
102                 with open(fn, 'rb') as f:
103                     process_one_file(f)
104         except Exception:
105             print_if_verbose("Error processing %s\n" % fn)
106             if args.ignore_errors:
107                 continue
108             raise
109
110     if args.honggfuzz_file:
111         print_if_verbose(f"looking at {args.honggfuzz_file}")
112         with open(args.honggfuzz_file) as f:
113             pipe = None
114             crash = None
115             for line in f:
116                 m = re.match(r'^\s*fuzzTarget\s*:\s*bin/fuzz_ndr_(\w+)\s*$', line)
117                 if m:
118                     pipe = m.group(1).split('_TYPE_', 1)[0]
119                     print_if_verbose(f"found pipe {pipe}")
120                 m = re.match(r'^FUZZ_FNAME: (\S+)$', line)
121                 if m:
122                     crash = m.group(1)
123                     if args.crash_filter is not None:
124                         if not re.search(args.crash_filter, crash):
125                             print_if_verbose(f"skipping {crash}")
126                             pipe = None
127                             crash = None
128                             continue
129                     print_if_verbose(f"found crash {crash}")
130                 if pipe is not None and crash is not None:
131                     with open(crash, 'rb') as f:
132                         process_one_file(f)
133                     pipe = None
134                     crash = None
135
136
137 main()