Coverage for /pythoncovmergedfiles/medio/medio/src/unparse_parse_fuzzer.py: 58%

Shortcuts on this page

r m x   toggle line displays

j k   next/prev highlighted chunk

0   (zero) top of page

1   (one) first highlighted chunk

76 statements  

1###### Coverage stub 

2import atexit 

3import coverage 

4cov = coverage.coverage(data_file='.coverage', cover_pylib=True) 

5cov.start() 

6# Register an exist handler that will print coverage 

7def exit_handler(): 

8 cov.stop() 

9 cov.save() 

10atexit.register(exit_handler) 

11####### End of coverage stub 

12#!/usr/bin/python3 

13# Copyright 2022 Google LLC 

14# 

15# Licensed under the Apache License, Version 2.0 (the "License"); 

16# you may not use this file except in compliance with the License. 

17# You may obtain a copy of the License at 

18# 

19# http://www.apache.org/licenses/LICENSE-2.0 

20# 

21# Unless required by applicable law or agreed to in writing, software 

22# distributed under the License is distributed on an "AS IS" BASIS, 

23# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 

24# See the License for the specific language governing permissions and 

25# limitations under the License. 

26"""Test dict to xml and back with fuzzing..""" 

27from typing import Dict, Any, Text, List, Callable 

28 

29import atheris 

30import logging 

31import sys 

32 

33from collections import OrderedDict 

34from xmltodict import parse, unparse 

35from xml.parsers.expat import ExpatError 

36 

37_MAX_LENGTH = 1000 

38_MAX_DEPTH = 500 

39 

40 

41def _gen_dict(fdp: atheris.FuzzedDataProvider, depth: int): 

42 """Returns a random dict for fuzzing.""" 

43 length = fdp.ConsumeIntInRange(0, _MAX_LENGTH) 

44 d = OrderedDict() 

45 for _ in range(length): 

46 key_length = fdp.ConsumeIntInRange(0, _MAX_LENGTH) 

47 key = fdp.ConsumeString(key_length) 

48 d[key] = _gen_value(fdp, depth + 1) 

49 return d 

50 

51 

52def _gen_string(fdp: atheris.FuzzedDataProvider): 

53 """Returns a random string for fuzzing.""" 

54 length = fdp.ConsumeIntInRange(0, _MAX_LENGTH) 

55 return fdp.ConsumeString(length) 

56 

57 

58def _gen_list(fdp: atheris.FuzzedDataProvider, depth: int): 

59 """Returns a random list for fuzzing.""" 

60 length = fdp.ConsumeIntInRange(0, _MAX_LENGTH) 

61 return [_gen_value(fdp, depth + 1) for _ in range(length)] 

62 

63 

64def _gen_value(fdp: atheris.FuzzedDataProvider, depth: int) -> Any: 

65 """Returns a random value for fuzzing.""" 

66 consume_next = [ 

67 fdp.ConsumeBool, 

68 fdp.ConsumeFloat, 

69 lambda: fdp.ConsumeInt(4), 

70 lambda: _gen_string(fdp), 

71 lambda: None, 

72 ] 

73 # XML documents can have exactly 1 root so don't add lists when 

74 # depth is exactly 0. 

75 if 0 < depth < _MAX_DEPTH: 

76 consume_next.append(lambda: _gen_list(fdp, depth)) 

77 

78 if depth < _MAX_DEPTH: 

79 consume_next.append(lambda: _gen_dict(fdp, depth)) 

80 return fdp.PickValueInList(consume_next)() 

81 

82 

83@atheris.instrument_func 

84def test_one_input(data: bytes): 

85 fdp = atheris.FuzzedDataProvider(data) 

86 original = OrderedDict() 

87 try: 

88 original[_gen_string(fdp)] = _gen_value(fdp, depth=0) 

89 except RecursionError: 

90 # Not interesting 

91 return 

92 

93 try: 

94 # Not all fuzz-generated data is valid XML. 

95 xml = unparse(original) 

96 except (ExpatError, UnicodeEncodeError): 

97 return 

98 

99 try: 

100 # FIXME: Not all unparsed XML is parsable. 

101 # FIXME: Why is there an _encode_ error in parse? 

102 final = parse(xml) # type: OrderedDict[Text, Any] 

103 except (ExpatError, UnicodeEncodeError): 

104 return 

105 

106 assert len(original) == len(final) 

107 for (k1,v1), (k2, v2) in zip(original.items(), final.items()): 

108 assert k1.strip() == k2, (k1, k2) 

109 

110 if isinstance(v1, str): 

111 # Strings are stripped and '' becomes None. 

112 v1 = v1.strip() or None 

113 if any(isinstance(v1, t) for t in (bool, int, float)): 

114 # Bools and Numbers become strings. 

115 v1 = str(v1) 

116 # Capitalization of booleans is inconsistent. 

117 assert v1.lower() == v2.lower(), (v1, v2) 

118 return 

119 if v1 == OrderedDict(): 

120 # Empty dict => None 

121 assert v2 == None 

122 return 

123 

124 assert v1 == v2, (v1, v2) 

125 

126def main(): 

127 atheris.instrument_all() 

128 atheris.Setup(sys.argv, test_one_input) 

129 atheris.Fuzz() 

130 return 0 

131 

132if __name__ == "__main__": 

133 sys.exit(main())