Coverage for /pythoncovmergedfiles/medio/medio/src/idna/tests/fuzz_idna_codec.py: 61%

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/env python3 

13"""OSS-Fuzz target for the ``idna2008`` codec. 

14 

15Exercises the one-shot codec and, more importantly, the incremental 

16encoder/decoder: the input is fed in fuzzer-chosen chunk sizes and the 

17concatenated result must equal the one-shot result (or both must raise 

18:class:`idna.IDNAError`). The buffering logic in 

19:mod:`idna.codec` is stateful across calls, which is exactly the kind of 

20code a fuzzer is good at breaking. 

21 

22OSS-Fuzz builds every ``fuzz_*.py`` it finds in the checkout, so this file 

23needs no registration there. To run it locally:: 

24 

25 pip install atheris . 

26 python tests/fuzz_idna_codec.py -max_total_time=60 

27 

28Any libFuzzer flag is accepted; a crash writes a ``crash-*`` file which can be 

29passed back as an argument to reproduce. ``tests/test_idna_fuzz_targets.py`` 

30smoke-tests this harness in the ordinary test suite without atheris. 

31""" 

32 

33import codecs 

34import sys 

35 

36import atheris # ty: ignore[unresolved-import] 

37 

38with atheris.instrument_imports(): 

39 import idna 

40 import idna.codec # registers the "idna2008" codec 

41 

42MAX_INPUT = 1100 # just past idna's 1024-character input cap 

43MAX_CHUNKS = 8 

44 

45 

46def _outcome(fn, *args): 

47 try: 

48 return fn(*args), None 

49 except idna.IDNAError as err: 

50 return None, err 

51 

52 

53def _chunks(fdp, data): 

54 cuts = sorted(fdp.ConsumeIntInRange(0, len(data)) for _ in range(fdp.ConsumeIntInRange(0, MAX_CHUNKS))) 

55 points = [0, *cuts, len(data)] 

56 return [data[i:j] for i, j in zip(points, points[1:])] 

57 

58 

59def fuzz_encoder(fdp): 

60 s = fdp.ConsumeUnicode(MAX_INPUT) 

61 if not s: 

62 return # the codec maps "" to b"" by design; core raises "Empty domain" 

63 one_shot = _outcome(idna.encode, s) 

64 assert _outcome(s.encode, "idna2008") == one_shot or one_shot[1] is not None 

65 

66 chunks = _chunks(fdp, s) 

67 encoder = codecs.getincrementalencoder("idna2008")() 

68 

69 def incremental(): 

70 out = b"".join(encoder.encode(chunk) for chunk in chunks) 

71 return out + encoder.encode("", final=True) 

72 

73 result = _outcome(incremental) 

74 assert (result[1] is None) == (one_shot[1] is None), (s, chunks, one_shot, result) 

75 assert result[0] == one_shot[0], (s, chunks, one_shot, result) 

76 

77 

78def fuzz_decoder(fdp): 

79 b = fdp.ConsumeBytes(MAX_INPUT) 

80 if not b: 

81 return 

82 one_shot = _outcome(idna.decode, b) 

83 assert _outcome(b.decode, "idna2008") == one_shot or one_shot[1] is not None 

84 

85 chunks = _chunks(fdp, b) 

86 decoder = codecs.getincrementaldecoder("idna2008")() 

87 

88 def incremental(): 

89 out = "".join(decoder.decode(chunk) for chunk in chunks) 

90 return out + decoder.decode(b"", final=True) 

91 

92 result = _outcome(incremental) 

93 assert (result[1] is None) == (one_shot[1] is None), (b, chunks, one_shot, result) 

94 assert result[0] == one_shot[0], (b, chunks, one_shot, result) 

95 

96 

97def fuzz_stream(fdp): 

98 # StreamWriter/StreamReader wrap the one-shot codec; make sure they only 

99 # ever raise IDNAError. 

100 import io 

101 

102 if fdp.ConsumeBool(): 

103 writer = codecs.getwriter("idna2008")(io.BytesIO()) 

104 _outcome(writer.write, fdp.ConsumeUnicode(MAX_INPUT)) 

105 else: 

106 reader = codecs.getreader("idna2008")(io.BytesIO(fdp.ConsumeBytes(MAX_INPUT))) 

107 _outcome(reader.read) 

108 

109 

110OPERATIONS = (fuzz_encoder, fuzz_decoder, fuzz_stream) 

111 

112 

113def TestOneInput(data): 

114 fdp = atheris.FuzzedDataProvider(data) 

115 OPERATIONS[fdp.ConsumeIntInRange(0, len(OPERATIONS) - 1)](fdp) 

116 

117 

118def main(): 

119 atheris.Setup(sys.argv, TestOneInput, enable_python_coverage=True) 

120 atheris.Fuzz() 

121 

122 

123if __name__ == "__main__": 

124 main()