Coverage for /pythoncovmergedfiles/medio/medio/src/fuzz_http.py: 69%

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 

27import os 

28import sys 

29import atheris 

30 

31import six 

32from six.moves import http_client 

33import mock 

34import httplib2 

35import google_auth_httplib2 

36 

37 

38# Mocks for fuzzing. Inspired by testing infra. Main difference is we have 

39# added logic to handle the event where the MockHTTP has run out of 

40# responses an appropriate exception is thrown that we catch in the fuzzer. 

41class FuzzMockHttp(object): 

42 def __init__(self, responses, headers=None): 

43 self.responses = responses 

44 self.requests = [] 

45 self.headers = headers or {} 

46 self.add_certificate = mock.Mock(return_value=None) 

47 

48 def request( 

49 self, 

50 url, 

51 method="GET", 

52 body=None, 

53 headers=None, 

54 redirections=httplib2.DEFAULT_MAX_REDIRECTS, 

55 connection_type=None, 

56 ): 

57 self.requests.append( 

58 (method, url, body, headers, redirections, connection_type) 

59 ) 

60 if len(self.responses) == 0: 

61 raise Exception("FUZZ: No more responses") 

62 return self.responses.pop(0) 

63 

64 

65class FuzzMockResponse(object): 

66 def __init__(self, status=http_client.OK, data=b""): 

67 self.status = status 

68 self.data = data 

69 

70 def __iter__(self): 

71 yield self 

72 yield self.data 

73 

74 

75class FuzzMockCredentials(object): 

76 def __init__(self, token="token"): 

77 self.token = token 

78 

79 def apply(self, headers): 

80 headers["authorization"] = self.token 

81 

82 def before_request(self, request, method, url, headers): 

83 self.apply(headers) 

84 

85 def refresh(self, request): 

86 self.token += "1" 

87 

88 

89def TestOneInput(data): 

90 fdp = atheris.FuzzedDataProvider(data) 

91 

92 mock_credentials = mock.Mock(wraps=FuzzMockCredentials(fdp.ConsumeUnicodeNoSurrogates(10))) 

93 if fdp.ConsumeBool() == True: 

94 mock_credentials.apply({'authorization': ""}) 

95 

96 # Create responses. We need at least two. 

97 http_codes = [400, 401, 403, 404, 405, 413, 415, 429, 500, 503, 504] 

98 mock_responses = [] 

99 for i in range(fdp.ConsumeIntInRange(1, 10)): 

100 mock_responses.append( 

101 FuzzMockResponse( 

102 status = http_codes[fdp.ConsumeIntInRange(0, len(http_codes)-1)], 

103 data = fdp.ConsumeBytes(100) 

104 ) 

105 ) 

106 mock_http = FuzzMockHttp(mock_responses) 

107 

108 authed_http = google_auth_httplib2.AuthorizedHttp( 

109 mock_credentials, http=mock_http 

110 ) 

111 try: 

112 r, d = authed_http.request( 

113 "http://localhost:8001", 

114 fdp.ConsumeUnicodeNoSurrogates(10), 

115 None, 

116 None, 

117 httplib2.DEFAULT_MAX_REDIRECTS, 

118 None 

119 ) 

120 except Exception as e: 

121 if "FUZZ" in str(e): 

122 pass 

123 else: 

124 raise e 

125 

126 

127def main(): 

128 atheris.instrument_all() 

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

130 atheris.Fuzz() 

131 

132 

133if __name__ == "__main__": 

134 main()