Coverage for /pythoncovmergedfiles/medio/medio/src/fuzz_actions_parser.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/python3 

13# Copyright 2023 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 

30from botocore.exceptions import ProfileNotFound 

31 

32with atheris.instrument_imports(): 

33 from c7n import policy as c7n_policy 

34 from c7n import exceptions, manager 

35 

36 from c7n.filters import FilterRegistry 

37 from c7n.actions import ActionRegistry 

38 

39 from c7n.resources import aws, rdsparamgroup, elasticache, ec2 

40 from c7n.resources import emr, account, apigw, elb, s3, glue 

41 from c7n.resources import appelb 

42 

43def TestOneInput(data): 

44 """Fuzz encode and decode""" 

45 registry_type = [ 

46 'c7n.data', 'rds-param-group', 'elasticache', 'ec2', 'emr', 

47 'aws.account', 'rest-account', 'elb', 's3', 'iac', 'rds', 

48 'glue-catalog', 'app-elb-target-group' 

49 ] 

50 provider = 'aws' 

51 

52 fdp = atheris.FuzzedDataProvider(data) 

53 

54 option = FuzzOption(fdp) 

55 data = _generate_random_dict(fdp) 

56 manager_data = _generate_random_dict(fdp) 

57 type = fdp.PickValueInList(registry_type) 

58 action_registry = ActionRegistry("%s.actions" % type) 

59 filter_registry = FilterRegistry("%s.filters" % type) 

60 resource_manager = manager.ResourceManager(FuzzContext(provider, option), manager_data) 

61 resource_manager.action_registry = action_registry 

62 resource_manager.filter_registry = filter_registry 

63 resource_manager.type = type 

64 resources = manager.resources 

65 

66 try: 

67 action_registry.parse(data, resource_manager) 

68 except (exceptions.PolicyValidationError, ProfileNotFound): 

69 pass 

70 except (KeyError, TypeError): 

71 pass 

72 

73 

74def _generate_random_dict(fdp): 

75 map = dict() 

76 

77 for count in range(fdp.ConsumeIntInRange(1, 100)): 

78 map[fdp.ConsumeUnicodeNoSurrogates(1024)] = fdp.ConsumeUnicodeNoSurrogates(1024) 

79 

80 map["name"] = fdp.ConsumeUnicodeNoSurrogates(1024) 

81 

82 return map 

83 

84 

85def initializeProviders(): 

86 aws.AWS() 

87 

88 

89def main(): 

90 initializeProviders() 

91 

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

93 atheris.Fuzz() 

94 

95 

96class FuzzContext: 

97 def __init__(self, name, option): 

98 self.options = None 

99 self.session_factory = c7n_policy.get_session_factory(name, option) 

100 self.policy = FuzzPolicy(name) 

101 

102 

103class FuzzPolicy: 

104 def __init__(self, provider_name): 

105 self.provider_name = provider_name 

106 self.name = "FuzzName" 

107 

108 

109class FuzzOption: 

110 def __init__(self, fdp): 

111 self.region = fdp.ConsumeUnicodeNoSurrogates(1024) 

112 self.profile = fdp.ConsumeUnicodeNoSurrogates(1024) 

113 self.assume_role = fdp.ConsumeUnicodeNoSurrogates(1024) 

114 self.external_id = fdp.ConsumeUnicodeNoSurrogates(1024) 

115 

116 

117if __name__ == "__main__": 

118 main()