#!/usr/bin/python3# -*- coding: utf-8 -*-# Author: Erdog, Loveforkeeps, Allukaimport base64import jsonimport loggingimport requestsimport osimport sysfrom retry.api import retry_callif __name__ == '__main__': # 当作为脚本直接执行时的处理方式 from exception import FofaError from helper import get_language, encode_queryelse: # 当作为包/模块导入时的处理方式 from .exception import FofaError from .helper import get_language, encode_queryif __name__ == "__main__": client = Client() logging.basicConfig(level=logging.DEBUG) print(client.can_use_next()) print(json.dumps(client.get_userinfo(), ensure_ascii=False)) print(json.dumps(client.search('app="网宿科技-公司产品"', page=1), ensure_ascii=False)) print(json.dumps(client.search_host('78.48.50.249', detail=True), ensure_ascii=False)) print(json.dumps(client.search_stats('domain="baidu.com"', fields='title'), ensure_ascii=False))[docs]class Client: """ A class representing the FOFA client. :param key: The Fofa api key. If not specified, it will be read from the FOFA_KEY environment variable. :type key: str :param base_url: The base URL of the FOFA API. Defaults to 'https://fofa.info'. :type base_url: str :param proxies: A proxies array for the requests library, e.g. {'https': 'your proxy'} :type proxies: dict """ def __init__(self, key='', base_url='', proxies=None): """ Initialize the FOFA client. """ key = os.environ.get('FOFA_KEY', '') if base_url == '': base_url = os.environ.get('FOFA_BASE_URL', 'https://fofa.info') self.key = key self.base_url = base_url.rstrip('/') self.lang = 'en' sys_lang = get_language() if sys_lang != None and sys_lang.startswith('zh'): self.lang = 'zh-CN' self._session = requests.Session() if proxies: self._session.proxies.update(proxies) self._session.trust_env = False # retry config, in seconds self.tries = 5 # Number of retry attempts self.delay = 1 # Initial delay between retries self.max_delay = 60 # Maximum delay between retries self.backoff = 2 # Backoff factor for exponential backoff
[docs] def get_userinfo(self): """ Get user info for current user. :return: User information in JSON format. :rtype: dict :raises FofaException: If an error occurs during the API request. :Example: The returned JSON result will be in the following format: .. code-block:: json { "username": "sample", "fofacli_ver": "4.0.3", "fcoin": 0, "error": false, "fofa_server": true, "avatar": "https://nosec.org/missing.jpg", "vip_level": 0, "is_verified": false, "message": "", "isvip": false, "email": "username@sample.net" } """ return self.__do_req( "/api/v1/info/my")
[docs] def search(self, query_str, page=1, size=100, fields="", opts={}): """ Search data in FOFA. :param query_str: The search query string. Example 1: 'ip=127.0.0.1' Example 2: 'header="thinkphp" || header="think_template"' :type query_str: str :param page: Page number. Default is 1. :type page: int :param size: Number of results to be returned in one page. Default is 100. :type size: int :param fields: Comma-separated list of fields to be included in the query result. Example: 'ip,city' :type fields: str :param opts: Additional options for the query. This should be a dictionary of key-value pairs. :type opts: dict :return: Query result in JSON format. :rtype: dict .. code-block:: json { "results": [ [ "111.**.241.**:8111", "111.**.241.**", "8111" ], [ "210.**.181.**", "210.**.181.**", "80" ] ], "mode": "extended", "error": false, "query": "app=\\"网宿科技-公司产品\\"", "page": 1, "size": 2 } """ param = opts param['qbase64'] = encode_query(query_str) param['page'] = page param['fields'] = fields param['size'] = size logging.debug("search '%s' page:%d size:%d", query_str, page, size) return self.__do_req('/api/v1/search/all', param)
[docs] def can_use_next(self): """ Check if the "search_next" API can be used. :return: True if the "search_next" API can be used, False otherwise. :rtype: bool """ try: self.search_next('bad=query', size=1) except FofaError as e: if e.code == 820000: return True return False
[docs] def search_next(self, query_str, fields='', size=100, next='', full=False, opts={}): """ Query the next page of search results. :param query_str: The search query string. Example 1: 'ip=127.0.0.1' Example 2: 'header="thinkphp" || header="think_template"' :param fields: The fields to be included in the response. Default: 'host,ip,port' :type fields: str :param size: The number of results to be returned per page. Default: 100 Maximum: 10,000 :type size: int :param next: The ID for pagination. The next value is returned in the response of previous search query. If not provided, the first page of results will be returned. :type next: str :param full: Specify if all data should be searched. Default: False (search within the past year) Set to True to search all data. :type full: bool :param opts: Additional options for the search. :type opts: dict :return: The query result in JSON format. :rtype: dict """ param = opts param['qbase64'] = encode_query(query_str) param['fields'] = fields param['size'] = size param['full'] = full if next and next != '': param['next'] = next logging.debug("search next for '%s' size:%d, next:%s", query_str, size, next) return self.__do_req('/api/v1/search/next', param)
[docs] def search_stats(self, query_str, size=5, fields='', opts={}): """ Query the statistics of the search results. :param query_str: The search query string. Example 1: 'ip=127.0.0.1' Example 2: 'header="thinkphp" || header="think_template"' :type query_str: str :param size: The number of results to be aggregated for each item. Default: 5 :type size: int :param fields: The fields to be included in the aggregation. Example: 'ip,city' :type fields: str :param opts: Additional options for the search. :type opts: dict :return: query result in json format .. code-block:: json { "distinct": { "ip": 1717, "title": 411 }, "lastupdatetime": "2022-06-17 13:00:00", "aggs": { "title": [ { "count": 35, "name": "百度一下，你就知道" }, { "count": 25, "name": "百度网盘-免费云盘丨文件共享软件丨超大容量丨存储安全" }, { "count": 16, "name": "百度智能云-登录" }, { "count": 2, "name": "百度翻译开放平台" } ], "countries": [] }, "error": false } """ param = opts param['qbase64'] = encode_query(query_str) param['fields'] = fields param['size'] = size return self.__do_req('/api/v1/search/stats', param)
[docs] def search_host(self, host, detail=False, opts={}): """ Search for host information based on the specified IP address or domain. :param host: The IP address or domain of the host to search for. :type host: str :param detail: Optional. Specifies whether to show detailed information. Default is False. :type detail: bool :param opts: Optional. Additional options for the search. Default is an empty dictionary. :type opts: dict :return: The query result in JSON format. :rtype: dict .. code-block:: json { "error": false, "host": "78.48.50.249", "ip": "78.48.50.249", "asn": 6805, "org": "Telefonica Germany", "country_name": "Germany", "country_code": "DE", "protocol": [ "http", "https" ], "port": [ 80, 443 ], "category": [ "CMS" ], "product": [ "Synology-WebStation" ], "update_time": "2022-06-11 08:00:00" } """ param = opts param['detail'] = detail u = '/api/v1/host/%s' % host return self.__do_req(u, param)
def __do_req(self, path, params=None, method='get'): u = self.base_url + path data = None req_param = {} if not self.key or self.key == '': raise FofaError("Empty fofa api key") if params == None: req_param = { "key": self.key, "lang": self.lang, } else: req_param = params req_param['key'] = self.key req_param['lang'] = self.lang if method == 'post': data = params params = None def make_request(): headers = {"Accept-Encoding": "gzip"} response = self._session.request(url=u, method=method, data=data, params=req_param, headers=headers) if response.status_code != 200: raise Exception("Request failed with status code: {}".format(response.status_code)) return response res = retry_call(make_request, tries = self.tries, delay = self.delay, max_delay = self.max_delay, backoff=self.backoff) data = res.json() if 'error' in data and data['error']: raise FofaError(data['errmsg']) return data
