connect to the API and pull all the IP addresses that are associated with the given query
(self)
| 55 | return token |
| 56 | |
| 57 | def search(self): |
| 58 | """ |
| 59 | connect to the API and pull all the IP addresses that are associated with the |
| 60 | given query |
| 61 | """ |
| 62 | start_animation("searching ZoomEye with given query '{}'".format(self.query)) |
| 63 | discovered_zoomeye_hosts = set() |
| 64 | try: |
| 65 | token = self.__get_auth() |
| 66 | if self.user_agent is None: |
| 67 | headers = {"Authorization": "JWT {}".format(str(token["access_token"]))} |
| 68 | else: |
| 69 | headers = { |
| 70 | "Authorization": "JWT {}".format(str(token["access_token"])), |
| 71 | "User-Agent": self.user_agent["User-Agent"] # oops |
| 72 | } |
| 73 | params = {"query": self.query, "page": "1", "facet": "ipv4"} |
| 74 | req = requests.get( |
| 75 | API_URLS["zoomeye"][1].format(query=self.query), |
| 76 | params=params, headers=headers, proxies=self.proxy |
| 77 | ) |
| 78 | _json_data = req.json() |
| 79 | for item in _json_data["matches"]: |
| 80 | if len(item["ip"]) > 1: |
| 81 | for ip in item["ip"]: |
| 82 | discovered_zoomeye_hosts.add(ip) |
| 83 | else: |
| 84 | discovered_zoomeye_hosts.add(str(item["ip"][0])) |
| 85 | write_to_file(discovered_zoomeye_hosts, self.host_file, mode=self.save_mode) |
| 86 | return True |
| 87 | except Exception as e: |
| 88 | raise AutoSploitAPIConnectionError(str(e)) |
| 89 |
nothing calls this directly
no test coverage detected