Coverage for /pythoncovmergedfiles/medio/medio/src/fuzz_requests.py: 62%

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# 

14# Copyright 2024 Google LLC 

15# 

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

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

18# You may obtain a copy of the License at 

19# 

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

21# 

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

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

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

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

26# limitations under the License. 

27# 

28################################################################################ 

29 

30import atheris 

31import sys 

32 

33# urllib3 slows down the initial startup and analysis phases of fuzz target runs 

34# because of how it is imported in requests.compat so it is excluded here. 

35with atheris.instrument_imports( 

36 exclude=['urllib3', 'urllib3.util', 'urllib.parse', 'urllib.request']): 

37 import requests_mock 

38 import requests 

39 from requests.auth import HTTPDigestAuth 

40 from requests.cookies import cookiejar_from_dict, CookieConflictError 

41 from requests.exceptions import RequestException 

42 

43 

44def is_expected_error(error_content_list, error_msg): 

45 for error in error_content_list: 

46 if error in error_msg: 

47 return True 

48 return False 

49 

50 

51def TestOneInput(data): 

52 fdp = atheris.FuzzedDataProvider(data) 

53 http_methods = ['GET', 'POST', 'PUT', 'DELETE', 'HEAD', 'OPTIONS', 'PATCH'] 

54 

55 try: 

56 cookie_jar = cookiejar_from_dict({ 

57 fdp.ConsumeString(10): fdp.ConsumeString(20) 

58 for _ in range(fdp.ConsumeIntInRange(1, 3)) 

59 }) 

60 except CookieConflictError: 

61 return -1 

62 

63 try: 

64 with requests_mock.Mocker() as global_mock: 

65 global_mock.request(method=requests_mock.ANY, 

66 url=requests_mock.ANY, 

67 status_code=fdp.ConsumeIntInRange(0, 599), 

68 reason=fdp.ConsumeString(fdp.ConsumeIntInRange( 

69 0, 100)), 

70 text=fdp.ConsumeString(fdp.ConsumeIntInRange(0, 100)), 

71 headers={ 

72 fdp.ConsumeString(10): fdp.ConsumeString(20) 

73 for _ in range(fdp.ConsumeIntInRange(1, 3)) 

74 }, 

75 cookies={ 

76 fdp.ConsumeString(10): fdp.ConsumeString(20) 

77 for _ in range(fdp.ConsumeIntInRange(1, 3)) 

78 }) 

79 

80 r1 = requests.request( 

81 fdp.PickValueInList(http_methods), 

82 url=fdp.ConsumeString(fdp.ConsumeIntInRange(0, 100)), 

83 allow_redirects=fdp.ConsumeBool(), 

84 auth=HTTPDigestAuth(fdp.ConsumeString(fdp.ConsumeIntInRange(0, 100)), 

85 fdp.ConsumeString(fdp.ConsumeIntInRange(0, 100))), 

86 params=fdp.ConsumeBytes(fdp.ConsumeIntInRange(1, 512)), 

87 timeout=fdp.ConsumeFloatInRange(0.1, 5.0), 

88 headers={ 

89 fdp.ConsumeString(10): fdp.ConsumeString(20) 

90 for _ in range(fdp.ConsumeIntInRange(1, 3)) 

91 }, 

92 cookies=cookie_jar) 

93 _ = r1.status_code 

94 _ = r1.reason 

95 _ = r1.headers 

96 _ = r1.cookies 

97 _ = r1.encoding 

98 _ = r1.text 

99 r1.close() 

100 

101 s = requests.Session() 

102 s.auth = (fdp.ConsumeString(fdp.ConsumeIntInRange(0, 100)), 

103 fdp.ConsumeString(fdp.ConsumeIntInRange(0, 100))) 

104 s.headers.update({ 

105 fdp.ConsumeString(10): fdp.ConsumeString(20) 

106 for _ in range(fdp.ConsumeIntInRange(1, 5)) 

107 }) 

108 

109 proxies = { 

110 'http': fdp.ConsumeString(fdp.ConsumeIntInRange(0, 100)), 

111 'https': fdp.ConsumeString(fdp.ConsumeIntInRange(0, 100)), 

112 } 

113 s.proxies.update(proxies) 

114 

115 custom_method = fdp.ConsumeString(fdp.ConsumeIntInRange(0, 20)) 

116 url_with_port = f"'https://'{fdp.ConsumeString(fdp.ConsumeIntInRange(0, 100))}:{fdp.ConsumeIntInRange(0, 10000)}/" 

117 req = requests.Request(custom_method, 

118 url=url_with_port, 

119 data=fdp.ConsumeBytes( 

120 fdp.ConsumeIntInRange(1, 1024))) 

121 prepped_request = req.prepare() 

122 

123 with requests_mock.Mocker(session=s) as session_mock: 

124 session_mock.request(method=requests_mock.ANY, 

125 url=requests_mock.ANY, 

126 status_code=fdp.ConsumeIntInRange(0, 599), 

127 content=fdp.ConsumeBytes( 

128 fdp.ConsumeIntInRange(0, sys.maxsize))) 

129 r2 = s.send(prepped_request) 

130 _ = r2.content 

131 r2.close() 

132 except (RequestException, ValueError) as e: 

133 expected_error_message_content = ["Invalid IPV4 URL", "Invalid IPV6 URL"] 

134 if (isinstance(e, RequestException) or (isinstance(e, ValueError)) and 

135 is_expected_error(expected_error_message_content, str(e))): 

136 return -1 

137 

138 

139def main(): 

140 atheris.Setup(sys.argv, TestOneInput) 

141 atheris.Fuzz() 

142 

143 

144if __name__ == "__main__": 

145 main()