Coverage for graphqler / fuzzer / engine / dengine.py: 73%

1import graphqler.config as config 

2from graphqler.graph.node import Node 

3from graphqler.utils.api import API 

4from graphqler.utils.logging_utils import Logger 

5from graphqler.utils.objects_bucket import ObjectsBucket 

6 

7from .detectors import api_detectors, injection_detectors, misc_detectors 

8from .detectors.detector import Detector 

9 

10 

11class DEngine: 

12 """Detector engine module 

13 -- used to detect vulnerabilities in the API 

14 """ 

15 

16 def __init__(self, api: API): 

17 """The intiialization of the DEngine 

18 

19 Args: 

20 api (API): The API object 

21 """ 

22 self.api = api 

23 self.logger = Logger().get_detector_logger() 

24 self.nodes_ran: dict[str, dict[str, bool]] = {} # {node_name: {detection_name: True/False}} 

25 

26 def run_detections_on_api(self): 

27 """Run detections on the API 

28 - Uses API as the key for nodes_ran marking 

29 """ 

30 for api_detector in api_detectors: 

31 node = Node(graphql_type='misc', name=self.api.url, body={}) # Create a dummy node object for the API 

32 detector = api_detector(api=self.api, node=node, objects_bucket=ObjectsBucket(self.api), graphql_type="") 

33 if not self.__should_run_detection(detector, self.api.url): 

34 continue 

35 try: 

36 is_vulnerable, potentially_vulnerable = detector.detect() 

37 self.logger.info(f"Detector {detector.DETECTION_NAME} finished detecting - is_vulnerable: {is_vulnerable} - potentially_vulnerable: {potentially_vulnerable}") 

38 self.__add_ran_node(self.api.url, detector.DETECTION_NAME) 

39 except Exception as e: 

40 self.logger.error(f"Detector {detector.DETECTION_NAME} failed with error: {e}") 

41 

42 def run_detections_on_graphql_object(self, node: Node, objects_bucket: ObjectsBucket, graphql_type: str): 

43 """Runs all detectors on a specific GraphQL object (either QUERY or MUTATION) 

44 

45 Args: 

46 node (Node): The node object 

47 objects_bucket (ObjectsBucket): The objects bucket 

48 graphql_type (str): The GraphQL type 

49 """ 

50 if not config.SKIP_INJECTION_ATTACKS: 

51 self.__run_injection_detections(node, objects_bucket, graphql_type) 

52 

53 if not config.SKIP_MISC_ATTACKS: 

54 self.__run_misc_detections(node, objects_bucket, graphql_type) 

55 

56 def __run_misc_detections(self, node: Node, objects_bucket: ObjectsBucket, graphql_type: str): 

57 """Runs miscellaneous detections 

58 

59 Args: 

60 node (Node): The node object 

61 objects_bucket (ObjectsBucket): The objects bucket 

62 graphql_type (str): The type of the GraphQL operation 

63 """ 

64 for misc_detector in misc_detectors: 

65 detector = misc_detector(api=self.api, node=node, objects_bucket=objects_bucket, graphql_type=graphql_type) 

66 if not self.__should_run_detection(detector, node.name): 

67 continue 

68 try: 

69 is_vulnerable, potentially_vulnerable = detector.detect() 

70 self.logger.info(f"Detector {detector.DETECTION_NAME} finished detecting - is_vulnerable: {is_vulnerable} - potentially_vulnerable: {potentially_vulnerable}") 

71 self.__add_ran_node(node.name, detector.DETECTION_NAME) 

72 except Exception as e: 

73 self.logger.error(f"Detector {detector.DETECTION_NAME} failed with error: {e}") 

74 

75 def __run_injection_detections(self, node: Node, objects_bucket: ObjectsBucket, graphql_type: str): 

76 """Runs injection detections 

77 

78 Args: 

79 node (Node): The node object 

80 objects_bucket (ObjectsBucket): The objects bucket 

81 graphql_type (str): The type of the GraphQL operation 

82 """ 

83 for injection_detector in injection_detectors: 

84 detector = injection_detector(api=self.api, node=node, objects_bucket=objects_bucket, graphql_type=graphql_type) 

85 if not self.__should_run_detection(detector, node.name): 

86 continue 

87 try: 

88 is_vulnerable, potentially_vulnerable = detector.detect() 

89 self.logger.info(f"Detector {detector.DETECTION_NAME} finished detecting - is_vulnerable: {is_vulnerable} - potentially_vulnerable: {potentially_vulnerable}") 

90 self.__add_ran_node(node.name, detector.DETECTION_NAME) 

91 except Exception as e: 

92 self.logger.error(f"Detector {detector.DETECTION_NAME} failed with error: {e}") 

93 

94 def __add_ran_node(self, name: str, detection_name: str): 

95 """Adds the node to the ran nodes list 

96 

97 Args: 

98 name (str): The name of the node 

99 detection_name (str): The name of the detection 

100 """ 

101 if name not in self.nodes_ran: 

102 self.nodes_ran[name] = {} 

103 self.nodes_ran[name][detection_name] = True 

104 

105 def __should_run_detection(self, detector: Detector, name: str) -> bool: 

106 """Whether the detection should be ran 

107 

108 Args: 

109 detector (Detector): The detector object 

110 name (str): The name of the node 

111 

112 Returns: 

113 bool: Whether the detection should be ran 

114 """ 

115 # First, check if the detector should be ran only once on the node 

116 if detector.detect_only_once_for_node: 

117 if name in self.nodes_ran and detector.DETECTION_NAME in self.nodes_ran[name]: 

118 return False 

119 

120 # Next, check if the detector should be run only once on the API 

121 if detector.detect_only_once_for_api: 

122 for node in self.nodes_ran: 

123 if detector.DETECTION_NAME in self.nodes_ran[node]: 

124 return False 

125 return True