Coverage for graphqler / fuzzer / engine / materializers / dos / dos_alias_materializer.py: 31%

1"""DOSAliasMaterializer: Generates a payload that uses field aliases to invoke the same resolver many times 

2in a single GraphQL request. This tests whether the server protects against alias-based amplification attacks. 

3""" 

4 

5from ..materializer import Materializer 

6from ..utils.materialization_utils import prettify_graphql_payload 

7from graphqler.utils.api import API 

8from graphqler.utils.objects_bucket import ObjectsBucket 

9 

10ALIAS_REPEAT_COUNT = 10 

11 

12 

13class DOSAliasMaterializer(Materializer): 

14 def __init__(self, api: API, fail_on_hard_dependency_not_met: bool = False, max_depth: int = 5): 

15 super().__init__(api, fail_on_hard_dependency_not_met) 

16 self.fail_on_hard_dependency_not_met = fail_on_hard_dependency_not_met 

17 self.max_depth = max_depth 

18 

19 def get_payload(self, name: str, objects_bucket: ObjectsBucket, graphql_type: str = "") -> tuple[str, dict]: 

20 """Builds a single query/mutation containing ALIAS_REPEAT_COUNT aliased copies of the operation. 

21 

22 Args: 

23 name (str): The query or mutation name 

24 objects_bucket (ObjectsBucket): The objects bucket 

25 graphql_type (str): "Query" or "Mutation" 

26 

27 Returns: 

28 tuple[str, dict]: The payload string and used objects map 

29 """ 

30 if graphql_type == "Query": 

31 return self._get_query_payload(name, objects_bucket) 

32 elif graphql_type == "Mutation": 

33 return self._get_mutation_payload(name, objects_bucket) 

34 else: 

35 raise ValueError("Invalid graphql_type provided") 

36 

37 def _get_query_payload(self, query_name: str, objects_bucket: ObjectsBucket) -> tuple[str, dict]: 

38 query_info = self.api.queries[query_name] 

39 query_inputs = self.materialize_inputs(query_info, query_info["inputs"], objects_bucket, max_depth=self.max_depth) 

40 query_output = self.materialize_output(query_info, query_info["output"], objects_bucket, max_depth=self.max_depth) 

41 

42 inputs_str = f"({query_inputs})" if query_inputs.strip() else "" 

43 

44 aliases = "\n".join( 

45 f"alias{i}: {query_name} {inputs_str} {query_output}" 

46 for i in range(ALIAS_REPEAT_COUNT) 

47 ) 

48 payload = f"query {{\n{aliases}\n}}" 

49 return prettify_graphql_payload(payload), self.used_objects 

50 

51 def _get_mutation_payload(self, mutation_name: str, objects_bucket: ObjectsBucket) -> tuple[str, dict]: 

52 mutation_info = self.api.mutations[mutation_name] 

53 mutation_inputs = self.materialize_inputs(mutation_info, mutation_info["inputs"], objects_bucket, max_depth=self.max_depth) 

54 mutation_output = self.materialize_output(mutation_info, mutation_info["output"], objects_bucket, max_depth=self.max_depth) 

55 

56 inputs_str = f"({mutation_inputs})" if mutation_inputs.strip() else "" 

57 

58 aliases = "\n".join( 

59 f"alias{i}: {mutation_name} {inputs_str} {mutation_output}" 

60 for i in range(ALIAS_REPEAT_COUNT) 

61 ) 

62 payload = f"mutation {{\n{aliases}\n}}" 

63 return prettify_graphql_payload(payload), self.used_objects