Coverage for graphqler / compiler / compiler.py: 71%

1"""Compiler class - responsible for: 

2- Getting the introspection query results into various files we can use later on 

3- Resolving dependencies among objects 

4- Tieing queries / mutations to objects 

5- Generating dependency chains for the fuzzer 

6""" 

7 

8from pathlib import Path 

9from graphqler.utils import plugins_handler 

10from graphqler.utils.file_utils import write_dict_to_yaml, write_json_to_file, initialize_file 

11from graphqler.utils.logging_utils import Logger 

12from .introspection_query import introspection_query 

13from .parsers import QueryListParser, ObjectListParser, MutationListParser, InputObjectListParser, EnumListParser, UnionListParser, InterfaceListParser, Parser 

14from .resolvers import ObjectDependencyResolver, ObjectMethodResolver, MutationObjectResolver, QueryObjectResolver, LLMMutationObjectResolver, LLMQueryObjectResolver, ResolverComparison 

15from graphqler.chains import ChainGenerator 

16from graphqler.graph import GraphGenerator 

17from graphqler import config 

18from clairvoyance.cli import blind_introspection 

19 

20import asyncio 

21import json 

22 

23 

24class Compiler: 

25 def __init__(self, save_path: str, url: str): 

26 """Initializes the compiler, 

27 creates all necessary file paths to save the outputs for run if doesn't already exist 

28 

29 Args: 

30 save_path (str): Save directory path 

31 url (str): URL for graphql introspection query to hit 

32 """ 

33 self.save_path = save_path 

34 self.introspection_result_save_path = Path(save_path) / Path(config.INTROSPECTION_RESULT_FILE_NAME) 

35 self.object_list_save_path = Path(save_path) / config.OBJECT_LIST_FILE_NAME 

36 self.input_object_list_save_path = Path(save_path) / config.INPUT_OBJECT_LIST_FILE_NAME 

37 self.mutation_parameter_save_path = Path(save_path) / config.MUTATION_PARAMETER_FILE_NAME 

38 self.query_parameter_save_path = Path(save_path) / config.QUERY_PARAMETER_FILE_NAME 

39 self.enum_list_save_path = Path(save_path) / config.ENUM_LIST_FILE_NAME 

40 self.union_list_save_path = Path(save_path) / config.UNION_LIST_FILE_NAME 

41 self.interface_list_save_path = Path(save_path) / config.INTERFACE_LIST_FILE_NAME 

42 

43 self.compiled_objects_save_path = Path(save_path) / config.COMPILED_OBJECTS_FILE_NAME 

44 self.compiled_mutations_save_path = Path(save_path) / config.COMPILED_MUTATIONS_FILE_NAME 

45 self.compiled_queries_save_path = Path(save_path) / config.COMPILED_QUERIES_FILE_NAME 

46 self.url = url 

47 

48 # Initialize the parsers we will use 

49 self.object_list_parser = ObjectListParser() 

50 self.query_list_parser = QueryListParser() 

51 self.mutation_list_parser = MutationListParser() 

52 self.input_object_list_parser = InputObjectListParser() 

53 self.enum_list_parser = EnumListParser() 

54 self.union_list_parser = UnionListParser() 

55 self.interface_list_parser = InterfaceListParser() 

56 

57 # Initialize the logger 

58 self.logger = Logger().get_compiler_logger() 

59 

60 # Initialize the plugins handler to get request utils 

61 self.request_utils = plugins_handler.get_request_utils() 

62 

63 # ChainGenerator — populated after run() completes 

64 self.chain_generator: ChainGenerator = ChainGenerator() 

65 

66 # Create empty files for these files 

67 Path(self.save_path).mkdir(parents=True, exist_ok=True) 

68 initialize_file(self.introspection_result_save_path) 

69 initialize_file(self.object_list_save_path) 

70 initialize_file(self.input_object_list_save_path) 

71 initialize_file(self.mutation_parameter_save_path) 

72 initialize_file(self.query_parameter_save_path) 

73 initialize_file(self.enum_list_save_path) 

74 initialize_file(self.union_list_save_path) 

75 initialize_file(self.interface_list_save_path) 

76 initialize_file(self.compiled_objects_save_path) 

77 initialize_file(self.compiled_mutations_save_path) 

78 initialize_file(self.compiled_queries_save_path) 

79 

80 def run(self): 

81 """The only function required to be run from the caller, will perform: 

82 1. Introspection query 

83 2. Trying clairvoyance if introspection query fails 

84 3. Run the parsers, storing files into objects / query / mutations 

85 4. Creating dependencies between objects and attaching methods (query/mutations) to objects 

86 """ 

87 introspection_result = self.get_introspection_query_results() 

88 if introspection_result is None or introspection_result == {}: 

89 print("(C) Introspection query failed, trying clairvoyance") 

90 introspection_result = self.get_clairvoyance_results() 

91 

92 if introspection_result is None or introspection_result == {}: 

93 raise SystemExit("(E) Couldn't get schema of the API. Exiting") 

94 

95 self.run_parsers_and_save(introspection_result) 

96 self.run_resolvers_and_save(introspection_result) 

97 

98 def get_introspection_query_results(self) -> dict: 

99 """Run the introspection query, grab results and output to file. Raises error if introspection query wasn't successful 

100 

101 Returns: 

102 dict: Dictionary of the resulting JSON from the introspection query 

103 """ 

104 result, response = self.request_utils.send_graphql_request(self.url, introspection_query) 

105 if "introspection is not allowed" in response.text.lower(): 

106 self.logger.warning("GraphQL Introspection is not allowed") 

107 return {} 

108 elif "is not allowed" in response.text.lower(): 

109 self.logger.warning("GraphQL Introspection is not allowed") 

110 return {} 

111 elif response.status_code != 200: 

112 error_message = f"Introspection query failed with status code {response.status_code}" 

113 self.logger.error(error_message) 

114 raise SystemExit(error_message) 

115 else: 

116 write_json_to_file(result, self.introspection_result_save_path) 

117 return result 

118 

119 def get_clairvoyance_results(self) -> dict: 

120 """Runs clairvoyance to get an introspection query output 

121 

122 Returns: 

123 dict: The introspection result using clairvoyance 

124 """ 

125 wordlist = [] 

126 if config.WORDLIST_PATH != "": 

127 with open(config.WORDLIST_PATH, "r") as file: 

128 wordlist = file.read().splitlines() 

129 

130 schema_str = asyncio.run( 

131 blind_introspection( 

132 url=self.url, 

133 logger=self.logger, 

134 wordlist=wordlist, 

135 headers=self.request_utils.get_headers(), 

136 input_document=None, 

137 input_schema_path=None, 

138 output_path=str(self.introspection_result_save_path), 

139 ) 

140 ) 

141 schema = json.loads(schema_str) 

142 return schema 

143 

144 def run_parsers_and_save(self, introspection_result: dict): 

145 """Runs all the parsers (parses introspection result sections out) and saves them to a YAML file 

146 

147 Args: 

148 introspection_result (dict): Introspection results as a dict 

149 """ 

150 self.run_parser_and_save_list(self.object_list_parser, self.object_list_save_path, introspection_result) 

151 self.run_parser_and_save_list(self.query_list_parser, self.query_parameter_save_path, introspection_result) 

152 self.run_parser_and_save_list(self.mutation_list_parser, self.mutation_parameter_save_path, introspection_result) 

153 self.run_parser_and_save_list(self.input_object_list_parser, self.input_object_list_save_path, introspection_result) 

154 self.run_parser_and_save_list(self.enum_list_parser, self.enum_list_save_path, introspection_result) 

155 self.run_parser_and_save_list(self.union_list_parser, self.union_list_save_path, introspection_result) 

156 self.run_parser_and_save_list(self.interface_list_parser, self.interface_list_save_path, introspection_result) 

157 

158 def run_parser_and_save_list(self, parser_instance: Parser, save_path: str | Path, introspection_result: dict): 

159 """Runs the given parser instance on the introspection result and saves to the save_path 

160 

161 Args: 

162 parser_instance (Parser): Parser instance 

163 save_path (str): Path to save parsed results (in YAML format) 

164 introspection_result (dict): Introspection result as a dict 

165 """ 

166 parsed_result = parser_instance.parse(introspection_result) 

167 write_dict_to_yaml(parsed_result, save_path) 

168 

169 def run_resolvers_and_save(self, introspection_result: dict): 

170 """Resolves objects, mutations and queries together so make it a "compiled" look: 

171 1. Enriches object-object dependency 

172 2. Enriches object-method dependency 

173 3. Enriches mutation-object dependency (classic or LLM-based) 

174 4. Enriches query-object dependency (classic or LLM-based) 

175 5. When USE_LLM=True, saves a side-by-side comparison JSON 

176 6. Write enriched objects to "compiled" directory in a yaml file 

177 

178 Args: 

179 introspection_result (dict): Introspection query result 

180 """ 

181 objects = self.object_list_parser.parse(introspection_result) 

182 queries = self.query_list_parser.parse(introspection_result) 

183 mutations = self.mutation_list_parser.parse(introspection_result) 

184 input_objects = self.input_object_list_parser.parse(introspection_result) 

185 

186 objects = ObjectDependencyResolver().resolve(objects) 

187 objects = ObjectMethodResolver().resolve(objects, queries, mutations) 

188 

189 if config.USE_LLM: 

190 print(f"(C) Using LLM resolver ({config.LLM_MODEL}) for dependency graph inference …") 

191 mut_resolver = LLMMutationObjectResolver() 

192 qry_resolver = LLMQueryObjectResolver() 

193 mutations = mut_resolver.resolve(objects, mutations, input_objects) 

194 queries = qry_resolver.resolve(objects, queries, input_objects) 

195 

196 if config.LLM_RESOLVER_SAVE_COMPARISON: 

197 comparison = ResolverComparison(mut_resolver.comparison, qry_resolver.comparison) 

198 comparison.save(self.save_path) 

199 comparison.print_diff_summary() 

200 else: 

201 mutations = MutationObjectResolver().resolve(objects, mutations, input_objects) 

202 queries = QueryObjectResolver().resolve(objects, queries, input_objects) 

203 

204 write_dict_to_yaml(objects, self.compiled_objects_save_path) 

205 write_dict_to_yaml(mutations, self.compiled_mutations_save_path) 

206 write_dict_to_yaml(queries, self.compiled_queries_save_path) 

207 

208 def run_chain_generation_and_save(self): 

209 """Builds the dependency graph from compiled files, generates chains via the configured 

210 :class:`ChainGenerator`, and persists them as a YAML file for human inspection. 

211 

212 The generated chains are also available via ``self.chain_generator.chains`` for 

213 immediate inspection without reloading from disk. 

214 """ 

215 dependency_graph = GraphGenerator(self.save_path).get_dependency_graph() 

216 in_degrees = dict(dependency_graph.in_degree()) 

217 if not in_degrees: 

218 self.logger.warning("Dependency graph is empty — no chains generated") 

219 return 

220 

221 min_degree = min(in_degrees.values()) 

222 starter_nodes = [node for node, degree in in_degrees.items() if degree == min_degree] 

223 

224 self.chain_generator.generate(dependency_graph, starter_nodes) 

225 self.logger.info(f"Generated {len(self.chain_generator.chains)} chains") 

226 

227 self.chain_generator.save_to_yaml(self.save_path) 

228 self.logger.info(f"Chains saved to {self.save_path}/{config.CHAINS_FILE_NAME}")