Coverage for graphqler / fuzzer / engine / detectors / detector.py: 91%

1from abc import ABC, abstractmethod 

2from typing import Type 

3 

4import requests 

5 

6from graphqler.graph.node import Node 

7from graphqler.utils.api import API 

8from graphqler.utils.logging_utils import Logger 

9from graphqler.utils.objects_bucket import ObjectsBucket 

10from graphqler.utils.stats import Stats 

11from graphqler.utils import plugins_handler 

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

13 

14from ..materializers.materializer import Materializer 

15 

16 

17class Detector(ABC): 

18 """Base Detector class that implements common functionality for all detectors. 

19 

20 Subclasses must implement: 

21 - DETECTION_NAME (class attribute) 

22 - materializer (property) 

23 - _is_vulnerable (method) 

24 """ 

25 

26 @property 

27 @abstractmethod 

28 def DETECTION_NAME(self) -> str: 

29 """Name of the detection type""" 

30 pass 

31 

32 @property 

33 def detect_only_once_for_api(self) -> bool: 

34 """Whether the detector should be run only once on the API""" 

35 return False 

36 

37 @property 

38 def detect_only_once_for_node(self) -> bool: 

39 """Whether the detector should be run only once on the node""" 

40 return True 

41 

42 @property 

43 @abstractmethod 

44 def materializer(self) -> Type[Materializer]: 

45 """Materializer class to be used for payload generation""" 

46 pass 

47 

48 def __init__(self, api: API, node: Node, objects_bucket: ObjectsBucket, graphql_type: str): 

49 self.api = api 

50 self.node = node 

51 self.name = node.name 

52 self.objects_bucket = objects_bucket 

53 self.graphql_type = graphql_type 

54 self.detector_logger = Logger().get_detector_logger() 

55 self.fuzzer_logger = Logger().get_fuzzer_logger() 

56 self.payload = "" 

57 self.confirmed_vulnerable = False 

58 self.potentially_vulnerable = False 

59 

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

61 """Main function to run to detect the vulnerability. 

62 

63 Returns: 

64 tuple[bool, bool]: (confirmed_vulnerable, potentially_vulnerable) 

65 """ 

66 self.payload = self.get_payload() 

67 self.fuzzer_logger.debug(f"[Fuzzer] Payload:\n{self.payload}") 

68 self.detector_logger.info(f"[Detector] Payload:\n{self.payload}") 

69 

70 # Send the GraphQL request 

71 graphql_response, request_response = plugins_handler.get_request_utils().send_graphql_request(self.api.url, self.payload) 

72 result = Result( 

73 result_enum=ResultEnum.GENERAL_SUCCESS, 

74 payload=self.payload, 

75 status_code=request_response.status_code, 

76 graphql_response=graphql_response, 

77 raw_response_text=request_response.text 

78 ) 

79 Stats().add_http_status_code(self.name, request_response.status_code) 

80 Stats().update_stats_from_result(self.node, result) 

81 

82 self.detector_logger.info(f"[{request_response.status_code}]Response: {request_response.text}") 

83 self.fuzzer_logger.info(f"[{request_response.status_code}]Response: {graphql_response}") 

84 

85 self._parse_response(graphql_response, request_response) 

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

87 return (self.confirmed_vulnerable, self.potentially_vulnerable) 

88 

89 def get_payload(self) -> str: 

90 """Gets the materialized payload to be sent to the API""" 

91 materializer_instance = self.materializer( 

92 api=self.api, 

93 fail_on_hard_dependency_not_met=False, 

94 max_depth=3 

95 ) 

96 payload, used_objects = materializer_instance.get_payload(self.name, self.objects_bucket, self.graphql_type) 

97 return payload 

98 

99 def _parse_response(self, graphql_response: dict, request_response: requests.Response): 

100 """Parses the response and checks for vulnerability""" 

101 if "errors" in graphql_response: 

102 self.detector_logger.info(f"Got errors: {graphql_response['errors']}") 

103 # self.potentially_vulnerable = True 

104 if self._is_vulnerable(graphql_response, request_response): 

105 self.detector_logger.info(f"Vulnerable to {self.DETECTION_NAME}") 

106 self.confirmed_vulnerable = True 

107 if self._is_potentially_vulnerable(graphql_response, request_response): 

108 self.detector_logger.info(f"Potentially vulnerable to {self.DETECTION_NAME}") 

109 self.potentially_vulnerable = True 

110 

111 @abstractmethod 

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

113 """Checks if the response indicates a vulnerability. 

114 

115 Must be implemented by subclasses. 

116 """ 

117 pass 

118 

119 @abstractmethod 

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

121 """Checks if the response indicates a potential vulnerability. 

122 

123 Must be implemented by subclasses. 

124 """ 

125 pass