Find the cURL command that contains the dynamic parts
(self, state: AgentState)
| 290 | return simplest_curl |
| 291 | |
| 292 | def find_curl_from_content(self, state: AgentState) -> AgentState: |
| 293 | """ |
| 294 | Find the cURL command that contains the dynamic parts |
| 295 | """ |
| 296 | search_string_list = state[self.IN_PROCESS_NODE_DYNAMIC_PARTS_KEY] |
| 297 | search_string_list_leftovers = search_string_list.copy() |
| 298 | |
| 299 | in_process_node_id = state[self.IN_PROCESS_NODE_KEY] |
| 300 | new_to_be_processed_nodes = [] |
| 301 | |
| 302 | # Handle cookies |
| 303 | for search_string in search_string_list_leftovers[:]: |
| 304 | cookie_key = self.find_key_by_string_in_value( |
| 305 | self.cookie_dict, search_string |
| 306 | ) |
| 307 | if cookie_key: |
| 308 | search_string_list_leftovers.remove(search_string) |
| 309 | if cookie_key in self.cookie_to_id_dict: |
| 310 | cookie_node_id = self.cookie_to_id_dict[cookie_key] |
| 311 | else: |
| 312 | cookie_node_id = self.dag_manager.add_node( |
| 313 | node_type="cookie", # Specify node type |
| 314 | content={ |
| 315 | "key": cookie_key, |
| 316 | "value": search_string |
| 317 | }, |
| 318 | extracted_parts=[search_string] |
| 319 | ) |
| 320 | self.cookie_to_id_dict[cookie_key] = cookie_node_id |
| 321 | #dont need to add node to to_be_processed_nodes because cookies dont need further processing |
| 322 | self.dag_manager.add_edge(in_process_node_id, cookie_node_id) |
| 323 | |
| 324 | # Handle curls |
| 325 | if search_string_list_leftovers: |
| 326 | for search_string in search_string_list_leftovers[:]: |
| 327 | requests_with_search_string = [] |
| 328 | |
| 329 | for request, response in self.req_to_res_map.items(): |
| 330 | curl = str(request) |
| 331 | if ( |
| 332 | ( |
| 333 | isinstance(curl, str) |
| 334 | and search_string.lower() in response["text"].lower() |
| 335 | ) |
| 336 | and (search_string.lower() not in curl.lower()) |
| 337 | ) or ( |
| 338 | urllib.parse.unquote(search_string) in curl |
| 339 | and (urllib.parse.unquote(search_string) not in curl) |
| 340 | ): |
| 341 | requests_with_search_string.append(request) |
| 342 | simplest_request = "" |
| 343 | |
| 344 | # Get simplest curl to reduce number of dependencies |
| 345 | if len(requests_with_search_string) > 1: |
| 346 | simplest_request = self.get_simplest_request(requests_with_search_string) |
| 347 | elif len(requests_with_search_string) == 1: |
| 348 | simplest_request = requests_with_search_string[0] |
| 349 | else: |
nothing calls this directly
no test coverage detected