Coverage for graphqler / fuzzer / engine / detectors / query_deny_bypass / query_deny_bypass_detector.py: 83%

1from typing import Type, override 

2 

3import requests 

4 

5from graphqler.fuzzer.engine.types import Result, ResultEnum 

6from graphqler.utils.api import API 

7from graphqler.fuzzer.engine.materializers.getter import Getter 

8from graphqler.fuzzer.engine.detectors.detector import Detector 

9from graphqler.fuzzer.engine.materializers.utils.materialization_utils import prettify_graphql_payload 

10from graphqler.fuzzer.engine.materializers.regular_payload_materializer import RegularPayloadMaterializer 

11from graphqler.utils.objects_bucket import ObjectsBucket 

12from graphqler.utils.stats import Stats 

13from graphqler.utils import plugins_handler 

14 

15 

16class QueryDenyBypassMaterializer(RegularPayloadMaterializer): 

17 def __init__(self, api: API, fail_on_hard_dependency_not_met: bool = False, max_depth: int = 20): 

18 super().__init__(api, fail_on_hard_dependency_not_met) 

19 self.api = api 

20 self.fail_on_hard_dependency_not_met = fail_on_hard_dependency_not_met 

21 self.getter = Getter() 

22 

23 def get_non_aliased_payload(self, name: str, objects_bucket: ObjectsBucket, graphql_type: str) -> tuple[str, dict]: 

24 if graphql_type == "Query": 

25 return self._get_query_payload(name, objects_bucket, max_input_depth=3, max_output_depth=3, minimal_materialization=True) 

26 if graphql_type == "Mutation": 

27 return self._get_mutation_payload(name, objects_bucket, max_input_depth=3, max_output_depth=3, minimal_materialization=True) 

28 else: 

29 return "", {} 

30 

31 def get_aliased_payload(self, name: str, objects_bucket: ObjectsBucket, graphql_type: str) -> tuple[str, dict]: 

32 if graphql_type not in ["Query", "Mutation"]: 

33 return "", {} 

34 info = self.api.queries[name] if graphql_type == "Query" else self.api.mutations[name] 

35 inputs = self.materialize_inputs(info, info["inputs"], objects_bucket, max_depth=3) 

36 output = self.materialize_output(info, info["output"], objects_bucket, max_depth=3) 

37 

38 if graphql_type == "Query": 

39 if inputs != "": 

40 inputs = f"({inputs})" 

41 

42 payload = f""" 

43 query {{ 

44 s: {name} {inputs} 

45 {output} 

46 }} 

47 """ 

48 else: 

49 if inputs.strip() == "": 

50 payload = f""" 

51 mutation {{ 

52 {name} 

53 {output} 

54 }} 

55 """ 

56 else: 

57 payload = f""" 

58 mutation {{ 

59 {name} ( 

60 {inputs} 

61 ) 

62 {output} 

63 }} 

64 """ 

65 pretty_payload = prettify_graphql_payload(payload) 

66 return pretty_payload, {} 

67 

68 

69class QueryDenyBypassDetector(Detector): 

70 @property 

71 def DETECTION_NAME(self) -> str: 

72 return "Query deny bypass" 

73 

74 @property 

75 def detect_only_once_for_api(self) -> bool: 

76 return False 

77 

78 @property 

79 def detect_only_once_for_node(self) -> bool: 

80 return True 

81 

82 @property 

83 def materializer(self) -> Type[QueryDenyBypassMaterializer]: 

84 return QueryDenyBypassMaterializer 

85 

86 @override 

87 def detect(self) -> tuple[bool, bool]: 

88 """Detects deny bypass vulnerability 

89 Does this in two phases - first, send the first request, if it's blocked, sends the second request with alias 

90 A deny bypass vulnerability is detected if the second request is successful 

91 

92 Returns: 

93 tuple[bool, bool]: is_vulnerable, is_potentially_vulnerable 

94 """ 

95 # If it's not a query, just skip 

96 if self.graphql_type != "Query": 

97 return (False, False) 

98 

99 materializer_instance = self.materializer(api=self.api, fail_on_hard_dependency_not_met=False, max_depth=3) 

100 non_aliased_payload, _ = materializer_instance.get_non_aliased_payload(self.name, self.objects_bucket, self.graphql_type) 

101 self.fuzzer_logger.debug(f"Non-aliased Payload:\n{non_aliased_payload}") 

102 self.detector_logger.info(f"Non-aliased Payload:\n{non_aliased_payload}") 

103 non_aliased_graphql_response, non_aliased_request_response = plugins_handler.get_request_utils().send_graphql_request(self.api.url, non_aliased_payload) 

104 self.fuzzer_logger.debug(f"[{non_aliased_request_response.status_code}]Non-aliased Response: {non_aliased_graphql_response}") 

105 self.detector_logger.info(f"[{non_aliased_request_response.status_code}]Non-aliased Response: {non_aliased_request_response.text}") 

106 

107 aliased_payload, _ = materializer_instance.get_aliased_payload(self.name, self.objects_bucket, self.graphql_type) 

108 self.fuzzer_logger.debug(f"Aliased Payload:\n{aliased_payload}") 

109 self.detector_logger.info(f"Aliased Payload:\n{aliased_payload}") 

110 aliased_graphql_response, aliased_request_response = plugins_handler.get_request_utils().send_graphql_request(self.api.url, aliased_payload) 

111 self.fuzzer_logger.debug(f"[{aliased_request_response.status_code}]Aliased Response: {aliased_graphql_response}") 

112 self.detector_logger.info(f"[{aliased_request_response.status_code}]Aliased Response: {aliased_request_response.text}") 

113 

114 if (("400" in non_aliased_request_response.text and 'errors' in non_aliased_graphql_response) 

115 or non_aliased_request_response.status_code == 400): 

116 if (aliased_request_response.status_code == 200 and 'data' in aliased_graphql_response and aliased_graphql_response['data']): 

117 if 'errors' in aliased_graphql_response and aliased_graphql_response['errors'] and len(aliased_graphql_response) != 0: 

118 self.potentially_vulnerable = True 

119 self.confirmed_vulnerable = False 

120 else: 

121 self.confirmed_vulnerable = True 

122 self.potentially_vulnerable = True 

123 else: 

124 self.potentially_vulnerable = False 

125 self.confirmed_vulnerable = False 

126 

127 non_aliased_result = Result(ResultEnum.GENERAL_SUCCESS, 

128 payload=non_aliased_payload, 

129 status_code=non_aliased_request_response.status_code, 

130 graphql_response=non_aliased_graphql_response, 

131 raw_response_text=non_aliased_request_response.text) 

132 aliased_result = Result(ResultEnum.GENERAL_SUCCESS, 

133 payload=aliased_payload, 

134 status_code=aliased_request_response.status_code, 

135 graphql_response=aliased_graphql_response, 

136 raw_response_text=aliased_request_response.text) 

137 

138 Stats().add_http_status_code(self.name, non_aliased_request_response.status_code) 

139 Stats().add_http_status_code(self.name, aliased_request_response.status_code) 

140 Stats().update_stats_from_result(self.node, non_aliased_result) 

141 Stats().update_stats_from_result(self.node, aliased_result) 

142 Stats().add_vulnerability(self.DETECTION_NAME, self.name, self.confirmed_vulnerable, self.potentially_vulnerable) 

143 return (self.confirmed_vulnerable, self.potentially_vulnerable) 

144 

145 def _is_vulnerable(self, graphql_response: dict, request_response: requests.Response) -> bool: 

146 # Intentionally left blank -- as detect is overriden 

147 raise NotImplementedError() 

148 

149 def _is_potentially_vulnerable(self, graphql_response: dict, request_response: requests.Response) -> bool: 

150 # Intentionally left blank -- as detect is overriden 

151 raise NotImplementedError()