Coverage for graphqler / fuzzer / fuzzer.py: 78%
166 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-03-20 10:09 -0400
1"""Class for fuzzer
31. Loads pre-generated chains from the compilation step
42. Pass 1: Run chains that contain only CREATE/QUERY nodes
53. Pass 2: Run chains that also allow UPDATE nodes
64. Pass 3: Run all chains (including DELETE/UNKNOWN)
75. Clean up
8"""
10import multiprocessing
11import random
12import threading
13import time
15import typing
17from graphqler import config
18from graphqler.chains import Chain, ChainGenerator
19from graphqler.graph import GraphGenerator, Node
20from graphqler.utils.api import API
21from graphqler.utils.logging_utils import Logger
22from graphqler.utils.objects_bucket import ObjectsBucket
23from graphqler.utils.stats import Stats
25from .engine.fengine import FEngine
26from .engine.dengine import DEngine
27from .engine.types import Result, ResultEnum
30class Fuzzer(object):
31 def __init__(self, save_path: str, url: str, objects_bucket: typing.Optional[ObjectsBucket] = None):
32 """Initializes the fuzzer, reading information from the compiled files
34 Args:
35 save_path (str): Save directory path
36 url (str): URL for graphql introspection query to hit
37 """
38 self.save_path = save_path
39 self.url = url
40 self.logger = Logger().get_fuzzer_logger()
41 self.stats = Stats()
42 self.api = API(url, save_path)
44 self.dependency_graph = GraphGenerator(save_path).get_dependency_graph()
45 self.fengine = FEngine(self.api)
46 self.dengine = DEngine(self.api)
48 if objects_bucket:
49 self.objects_bucket = objects_bucket
50 else:
51 self.objects_bucket = ObjectsBucket(self.api)
53 # Load pre-generated chains produced during compilation
54 self.chains: list[Chain] = ChainGenerator().load_from_yaml(save_path, self.dependency_graph)
56 # Stats about the run
57 self.stats.number_of_queries = self.api.get_num_queries()
58 self.stats.number_of_mutations = self.api.get_num_mutations()
59 self.stats.number_of_objects = self.api.get_num_objects()
61 def run(self):
62 """Main function to run the fuzzer"""
63 queue = multiprocessing.Queue()
64 if config.DEBUG:
65 p = threading.Thread(target=self.__run_steps, args=(queue,))
66 p.daemon = True
67 else:
68 p = multiprocessing.Process(target=self.__run_steps, args=(queue,))
69 p.start()
70 p.join(config.MAX_TIME)
72 if p.is_alive() and isinstance(p, multiprocessing.Process):
73 print(f"(+) Terminating the fuzzer process - reached max time {config.MAX_TIME}s")
74 p.terminate()
76 if not queue.empty():
77 _ = queue.get()
79 def run_single(self, node_name: str):
80 """Runs a single node
82 Args:
83 node_name (str): The name of the node
84 """
85 node = [n for n in self.dependency_graph.nodes if n.name == node_name]
86 if len(node) == 0:
87 print(f"(F) Node `{node_name}` not found")
88 self.logger.error(f"Node `{node_name}` not found")
89 return
91 self.stats.start_time = time.time()
92 self.__run_nodes(node)
93 self.logger.info("Completed fuzzing")
94 self.stats.print_results()
95 self.stats.save()
96 self.objects_bucket.save()
98 def run_no_dfs(self):
99 """Runs the fuzzer without using the dependency graph. Just uses each node and tests against the server
101 Returns:
102 dict: The objects bucket
103 """
104 nodes_to_run = list(self.dependency_graph.nodes)
105 self.__run_nodes(nodes_to_run)
106 self.logger.info("Completed fuzzing")
107 self.stats.print_results()
108 self.stats.save()
109 self.objects_bucket.save()
111 def __run_steps(self, queue: multiprocessing.Queue):
112 """Runs the fuzzer using pre-generated chains. Steps:
113 1. Execute each chain in order (pass/filter ordering handled by the compiler)
114 2. Run any nodes not covered by the chains (island nodes)
115 3. Run detections on the overall API
116 4. Finish
118 Args:
119 queue (multiprocessing.Queue): Queue for communicating back to the parent process
120 """
121 self.stats.start_time = time.time()
123 if self.chains:
124 self.logger.info(f"Running {len(self.chains)} pre-generated chains")
125 for chain in self.chains:
126 self.__run_chain(chain)
127 self.logger.info("Completed all chains")
129 # Run any nodes not covered by any chain (e.g. isolated nodes)
130 chained_nodes: set[Node] = {node for chain in self.chains for node in chain.nodes}
131 uncovered_nodes = [node for node in self.dependency_graph.nodes if node not in chained_nodes]
132 else:
133 # Fallback: no chains available (compiler not run or old compilation), execute all nodes
134 self.logger.warning("No chains found — falling back to running all nodes directly")
135 uncovered_nodes = list(self.dependency_graph.nodes)
137 if uncovered_nodes:
138 self.logger.info(f"Running {len(uncovered_nodes)} uncovered node(s)")
139 self.__run_nodes(uncovered_nodes)
141 # Detections
142 self.dengine.run_detections_on_api()
143 self.logger.info("Completed running detections on the overall API")
145 # Finish
146 self.logger.info("Completed fuzzing")
147 self.logger.info(f"Objects bucket: {self.objects_bucket}")
148 self.stats.print_results()
149 self.stats.save()
150 self.objects_bucket.save()
152 def __run_chain(self, chain: Chain):
153 """Executes every node in the chain sequentially using a fresh, isolated ObjectsBucket.
155 Each chain is fully self-sufficient (generated by :class:`AllDependenciesChainStrategy`),
156 so its bucket starts completely empty — no objects from previous chains are copied in.
157 This guarantees true isolation between chains.
159 Args:
160 chain (Chain): The chain to execute.
161 """
162 fresh_bucket: ObjectsBucket = ObjectsBucket.__wrapped__(self.api)
164 self.logger.info(f"Running chain: {chain}")
165 for node in chain.nodes:
166 if node.name in config.SKIP_NODES:
167 continue
168 self.stats.print_running_stats()
169 self.logger.info(f"[chain] Running node: {node}")
170 node_start = time.time()
171 _next_paths, result = self.__evaluate(node, list(chain.nodes[:chain.nodes.index(node) + 1]),
172 objects_bucket=fresh_bucket)
173 self.stats.record_node_timing(node, time.time() - node_start)
174 self.stats.update_stats_from_result(node, result)
176 self.__fuzz(node, list(chain.nodes[:chain.nodes.index(node) + 1]), objects_bucket=fresh_bucket)
177 self.__detect_vulnerabilities_on_node(node, fresh_bucket)
179 if not result.success:
180 # If a prerequisite node fails, the rest of the chain cannot proceed
181 self.logger.info(f"[chain] Node {node} failed — stopping chain execution early")
182 break
184 def __run_nodes(self, nodes: list[Node]):
185 """Runs the nodes given in the list
187 Args:
188 nodes (list[Node]): List of nodes to run
190 Raises:
191 Exception: If the GraphQL type of the node is unknown
192 """
193 for current_node in nodes:
194 self.stats.print_running_stats()
195 self.logger.info(f"Running node: {current_node}")
196 node_start = time.time()
197 _next_visit_path, result = self.__run_node(current_node, [current_node], check_hard_depends_on=False)
198 self.stats.record_node_timing(current_node, time.time() - node_start)
199 self.stats.update_stats_from_result(current_node, result)
201 if result.success:
202 self.logger.info(f"Node was successful: {current_node}")
204 def __run_node(self, node: Node, visit_path: list[Node], check_hard_depends_on: bool = True) -> tuple[list[list[Node]], Result]:
205 """Runs the node, evaluating it and return the next visit paths.
206 - The return will be based on the positive testing of the node
207 - The side effects will be the fuzzed node and the detection of any vulnerabilities on the node
209 Args:
210 node (Node): The node
211 visit_path (list[Node]): The visit path
212 check_hard_depends_on (bool, optional): Whether to check the dependencies. Defaults to True.
214 Returns:
215 tuple[list[list[Node]], Result]: The results of the positive node evaluation
216 """
217 if node.name in config.SKIP_NODES:
218 return ([], Result(ResultEnum.GENERAL_SUCCESS))
219 new_paths_to_evaluate, res = self.__evaluate(node, visit_path, check_hard_depends_on=check_hard_depends_on)
220 self.__fuzz(node, visit_path)
221 self.__detect_vulnerabilities_on_node(node)
222 return (new_paths_to_evaluate, res)
224 def __evaluate(self, node: Node, visit_path: list[Node], check_hard_depends_on: bool = True,
225 objects_bucket: typing.Optional[ObjectsBucket] = None) -> tuple[list[list[Node]], Result]:
226 """Evaluates the path, performing the following based on the type of node:
227 Case 1: If it's an object node, then we should check if the object is in our bucket. If not, fail, if it is,
228 then queue up the next neighboring nodes to visit
229 Case 2: If it's an query node or mutation node, run the payload with the required objects, then store the results in the object bucket
231 Args:
232 node (Node): Node to be evaluated
233 visit_path (list[Node]): The list of visited paths to arrive at the node
234 check_hard_depends_on (bool): The check hard depends on flag for materializing the object
235 objects_bucket (ObjectsBucket | None): Bucket to use; defaults to self.objects_bucket
237 Returns:
238 tuple[list[list[Node]], Result]: A list of the next to_visit paths, and the result of the node evaluation
239 """
240 bucket = objects_bucket if objects_bucket is not None else self.objects_bucket
241 neighboring_nodes = self._get_neighboring_nodes(node)
242 new_visit_paths = self._get_new_visit_path_with_neighbors(neighboring_nodes, visit_path)
244 if node.graphql_type == "Object" and check_hard_depends_on:
245 if bucket.is_object_in_bucket(node.name):
246 return (new_visit_paths, Result(ResultEnum.GENERAL_SUCCESS))
247 else:
248 return ([], Result(ResultEnum.INTERNAL_FAILURE))
249 else:
250 _graphql_response, res = self.fengine.run_minimal_payload(node.name, bucket, node.graphql_type, check_hard_depends_on=check_hard_depends_on)
251 if res.success:
252 return (new_visit_paths, res)
253 else:
254 return ([], res)
256 def __fuzz(self, node: Node, visit_path: list[Node], objects_bucket: typing.Optional[ObjectsBucket] = None):
257 """Fuzzes a node by running the node and storing the results. Currently runs:
258 - DOS Query / Mutation (from size 0 to MAX_INPUT_DEPTH or HARD_CUTOFF_DEPTH, whichever is smaller)
260 Args:
261 node (Node): The node to fuzz
262 visit_path (list[Node]): The list of visited paths to arrive at the node
263 objects_bucket (ObjectsBucket | None): Bucket to use; defaults to self.objects_bucket
264 """
265 bucket = objects_bucket if objects_bucket is not None else self.objects_bucket
266 if node.graphql_type not in ["Query", "Mutation"]:
267 return
268 if not config.SKIP_DOS_ATTACKS and config.MAX_FUZZING_ITERATIONS != 0:
269 random_numbers = [random.randint(1, min(config.HARD_CUTOFF_DEPTH, config.MAX_INPUT_DEPTH)) for _ in range(0, config.MAX_FUZZING_ITERATIONS)]
270 random_number = random.choice(random_numbers)
271 self.logger.info(f"Running DOS {node.graphql_type}: {node.name} with depth: {random_number}")
272 results = self.fengine.run_dos_payloads(node.name, bucket, node.graphql_type, random_number)
273 for _graphql_response, res in results:
274 self.stats.update_stats_from_result(node, res)
276 if not config.SKIP_MAXIMAL_PAYLOADS:
277 self.fengine.run_maximal_payload(node.name, bucket, node.graphql_type, check_hard_depends_on=False)
279 def __detect_vulnerabilities_on_node(self, node: Node, objects_bucket: typing.Optional[ObjectsBucket] = None):
280 bucket = objects_bucket if objects_bucket is not None else self.objects_bucket
281 if node.graphql_type in ["Query", "Mutation"]:
282 self.dengine.run_detections_on_graphql_object(node, bucket, node.graphql_type)
284 # ------------------- Helpers -------------------
286 def _get_new_visit_path_with_neighbors(self, neighboring_nodes: list[Node], visit_path: list[Node]) -> list[list[Node]]:
287 """Gets the new visit path with the neighbors by creating a new path for each neighboring node
289 Args:
290 neighboring_nodes (list[Node]): The list of neighboring nodes
291 visit_path (list[Node]): The visit path that the current iteration is on
293 Returns:
294 list[list[Node]]: A list of visit_paths where each visit_path is just the visit_path + neighboring_node
295 """
296 new_visit_paths = []
297 for node in neighboring_nodes:
298 new_visit_paths.append(visit_path + [node])
299 return new_visit_paths
301 def _get_neighboring_nodes(self, node: Node) -> list[Node]:
302 """Get nodes that this node goes out of
304 Args:
305 node (Node): The node we want to find that is pointing to this node
307 Returns:
308 list[Node]: List of nodes that are dependent on the input node
309 """
310 return [n for n in self.dependency_graph.successors(node)]
312 def _get_starter_nodes(self) -> list[Node]:
313 """Gets a list of starter nodes to start the fuzzing with.
314 First, looks for nodes with no incoming edges (in-degree == 0).
315 If none exist, returns nodes with the minimum in-degree.
317 Returns:
318 list[Node]: A list of starter nodes
319 """
320 in_degrees = dict(self.dependency_graph.in_degree())
321 if not in_degrees:
322 self.logger.error("No nodes in dependency graph")
323 return []
325 min_degree = min(in_degrees.values())
326 nodes = [node for node, degree in in_degrees.items() if degree == min_degree]
327 if nodes:
328 return nodes
330 self.logger.error("No starter nodes found, choosing a random node")
331 return [random.choice(list(self.dependency_graph.nodes))]