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

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 

35from itertools import pairwise 

36 

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

38 

39with atheris.instrument_imports(): 

40 import idna 

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

42 

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

44MAX_CHUNKS = 8 

45 

46 

47def _outcome(fn, *args): 

48 try: 

49 return fn(*args), None 

50 except idna.IDNAError as err: 

51 return None, err 

52 

53 

54def _chunks(fdp, data): 

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

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

57 return [data[i:j] for i, j in pairwise(points)] 

58 

59 

60def fuzz_encoder(fdp): 

61 s = fdp.ConsumeUnicode(MAX_INPUT) 

62 if not s: 

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

64 one_shot = _outcome(idna.encode, s) 

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

66 

67 chunks = _chunks(fdp, s) 

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

69 

70 def incremental(): 

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

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

73 

74 result = _outcome(incremental) 

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

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

77 

78 

79def fuzz_decoder(fdp): 

80 b = fdp.ConsumeBytes(MAX_INPUT) 

81 if not b: 

82 return 

83 one_shot = _outcome(idna.decode, b) 

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

85 

86 chunks = _chunks(fdp, b) 

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

88 

89 def incremental(): 

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

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

92 

93 result = _outcome(incremental) 

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

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

96 

97 

98def fuzz_stream(fdp): 

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

100 # ever raise IDNAError. 

101 import io 

102 

103 if fdp.ConsumeBool(): 

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

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

106 else: 

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

108 _outcome(reader.read) 

109 

110 

111OPERATIONS = (fuzz_encoder, fuzz_decoder, fuzz_stream) 

112 

113 

114def TestOneInput(data): 

115 fdp = atheris.FuzzedDataProvider(data) 

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

117 

118 

119def main(): 

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

121 atheris.Fuzz() 

122 

123 

124if __name__ == "__main__": 

125 main()