diff --git a/tools/quake_search.yaml b/tools/quake_search.yaml index 5aa41145..c52a2070 100644 --- a/tools/quake_search.yaml +++ b/tools/quake_search.yaml @@ -16,7 +16,7 @@ args: # ================================================== # Quake API基础URL - base_url = "https://quake.360.cn/api/v3/search/quake_service" + base_url = "https://quake.360.net/api/v3/search/quake_service" # 解析参数（从JSON字符串或命令行参数） def parse_args(): @@ -178,6 +178,16 @@ args: sys.exit(1) results = result_data.get("data", []) + if isinstance(results, dict): + nested = None + for key in ("data", "items", "matches", "results", "list", "records"): + value = results.get(key) + if isinstance(value, list): + nested = value + break + results = nested if nested is not None else [] + if not isinstance(results, list): + results = [] meta = result_data.get("meta", {}) pagination = meta.get("pagination", {}) if isinstance(meta, dict) else {} diff --git a/tools/rustscan.yaml b/tools/rustscan.yaml index b25eae85..181ffae5 100644 --- a/tools/rustscan.yaml +++ b/tools/rustscan.yaml @@ -25,7 +25,6 @@ parameters: required: true flag: "-a" format: "flag" - # -p, --ports；范围请用 range (-r)，勿把 1-1000 填进 ports - name: "ports" type: "string" @@ -36,7 +35,6 @@ parameters: required: false flag: "-p" format: "flag" - # -r, --range（与 ports 列表二选一或按官方说明组合） - name: "range" type: "string" @@ -47,7 +45,6 @@ parameters: required: false flag: "-r" format: "flag" - # -u, --ulimit - name: "ulimit" type: "int" @@ -56,7 +53,6 @@ parameters: flag: "-u" format: "flag" default: 5000 - # --scripts：none | default | custom（CLI 默认 default） - name: "scripts" type: "string" @@ -73,7 +69,6 @@ parameters: - "none" - "default" - "custom" - # -g, --greppable：仅端口列表，无 Nmap 输出 - name: "greppable" type: "bool" @@ -82,7 +77,6 @@ parameters: flag: "-g" format: "flag" default: false - # -b, --batch-size [default: 4500] - name: "batch_size" type: "int" @@ -90,7 +84,6 @@ parameters: required: false flag: "-b" format: "flag" - # -t, --timeout ms [default: 1500] - name: "timeout_ms" type: "int" @@ -98,7 +91,6 @@ parameters: required: false flag: "-t" format: "flag" - # --scan-order serial | random [default: serial] - name: "scan_order" type: "string" @@ -109,7 +101,6 @@ parameters: options: - "serial" - "random" - # --top：top 1000 端口 - name: "top_ports" type: "bool" @@ -118,7 +109,6 @@ parameters: flag: "--top" format: "flag" default: false - # -e, --exclude-ports - name: "exclude_ports" type: "string" @@ -126,7 +116,6 @@ parameters: required: false flag: "-e" format: "flag" - # -x, --exclude-addresses - name: "exclude_addresses" type: "string" @@ -134,7 +123,6 @@ parameters: required: false flag: "-x" format: "flag" - # --tries [default: 1] - name: "tries" type: "int" @@ -142,7 +130,6 @@ parameters: required: false flag: "--tries" format: "flag" - - name: "additional_args" type: "string" description: | diff --git a/tools/virustotal_search.yaml b/tools/virustotal_search.yaml index fc236e3f..41d4771b 100644 --- a/tools/virustotal_search.yaml +++ b/tools/virustotal_search.yaml @@ -2,296 +2,70 @@ name: "virustotal_search" command: "python3" args: - "-c" - - | - import sys - import json - import requests - import os - import time - - # ==================== VirusTotal 配置 ==================== - # 请在此处配置您的 VirusTotal API 密钥 - # 您也可以在环境变量中设置：VT_API_KEY - # enable 默认为 false，需开启才能调用该MCP - VT_API_KEY = "" # 请填写您的 VirusTotal API 密钥 - # ======================================================= - - # VirusTotal API 基础 URL - BASE_URL = "https://www.virustotal.com/api/v3" - - def parse_args(): - """解析命令行参数""" - # 尝试从第一个参数读取 JSON 配置 - if len(sys.argv) > 1: - try: - arg1 = str(sys.argv[1]) - config = json.loads(arg1) - if isinstance(config, dict): - return config - except (json.JSONDecodeError, TypeError, ValueError): - pass - - # 传统位置参数方式 - config = {} - if len(sys.argv) > 1: - config['domain'] = str(sys.argv[1]) - if len(sys.argv) > 2: - try: - config['limit'] = int(sys.argv[2]) - except (ValueError, TypeError): - pass - if len(sys.argv) > 3: - config['include_ips'] = sys.argv[3].lower() in ('true', '1', 'yes') - return config - - def query_virustotal_subdomains(domain, api_key, limit=100, include_ips=False): - """ - 查询 VirusTotal 的子域名信息 - - Args: - domain: 要查询的域名 - api_key: VirusTotal API 密钥 - limit: 返回结果数量限制 - include_ips: 是否包含 IP 地址信息 - - Returns: - dict: 包含查询结果的字典 - """ - # 构建 API 请求 URL - url = f"{BASE_URL}/domains/{domain}/subdomains" - - headers = { - "x-apikey": api_key, - "accept": "application/json" - } - - params = { - "limit": min(limit, 40) # API 限制最大 40 - } - - all_results = [] - next_url = None - - try: - # 处理分页 - while True: - if next_url: - response = requests.get(next_url, headers=headers, timeout=30) - else: - response = requests.get(url, headers=headers, params=params, timeout=30) - - response.raise_for_status() - data = response.json() - - # 提取子域名数据 - if 'data' in data and data['data']: - for item in data['data']: - if 'id' in item: - subdomain_info = { - 'subdomain': item['id'], - 'type': item.get('type', 'domain'), - } - - # 如果 include_ips 为 True，尝试获取解析 IP - if include_ips and 'attributes' in item: - attributes = item.get('attributes', {}) - # 这里简化处理，实际可能需要额外的 API 调用 - subdomain_info['last_dns_records'] = attributes.get('last_dns_records', []) - - all_results.append(subdomain_info) - - # 检查是否有下一页 - if 'links' in data and 'next' in data['links'] and len(all_results) < limit: - next_url = data['links']['next'] - # 避免请求过快 - time.sleep(0.5) - else: - break - else: - break - - # 如果已达到限制，停止获取 - if len(all_results) >= limit: - break - - # 处理返回结果 - if all_results: - return { - "status": "success", - "domain": domain, - "total_found": len(all_results), - "results": all_results[:limit], - "message": f"成功获取 {len(all_results[:limit])} 个子域名" - } - else: - return { - "status": "success", - "domain": domain, - "total_found": 0, - "results": [], - "message": f"未找到 {domain} 的子域名" - } - - except requests.exceptions.RequestException as e: - error_msg = str(e) - error_result = { - "status": "error", - "message": f"API 请求失败: {error_msg}", - "suggestion": "请检查网络连接、API 密钥是否正确，或 VirusTotal API 服务是否可用" - } - - # 处理特定 HTTP 状态码 - if hasattr(e, 'response') and e.response: - status_code = e.response.status_code - if status_code == 401: - error_result["message"] = "API 密钥无效或未授权" - error_result["suggestion"] = "请检查 VirusTotal API 密钥是否正确，或在 https://www.virustotal.com/ 获取有效密钥" - elif status_code == 429: - error_result["message"] = "API 请求频率超限" - error_result["suggestion"] = "请稍后再试，VirusTotal API 有严格的速率限制（免费版每分钟4次）" - elif status_code == 404: - error_result["message"] = f"域名 '{domain}' 不存在或未找到" - - return error_result - - try: - config = parse_args() - - if not isinstance(config, dict): - error_result = { - "status": "error", - "message": f"参数解析错误: 期望字典类型，但得到 {type(config).__name__}", - "type": "TypeError" - } - print(json.dumps(error_result, ensure_ascii=False, indent=2)) - sys.exit(1) - - # 获取 API 密钥（从配置或环境变量） - api_key = os.getenv('VT_API_KEY', VT_API_KEY).strip() - - if not api_key: - error_result = { - "status": "error", - "message": "缺少 VirusTotal API 密钥", - "required_config": ["VT_API_KEY"], - "note": "请在 YAML 文件的 VT_API_KEY 配置项中填写您的 VirusTotal API 密钥，或在环境变量 VT_API_KEY 中设置。API 密钥可在 https://www.virustotal.com/ 注册获取" - } - print(json.dumps(error_result, ensure_ascii=False, indent=2)) - sys.exit(1) - - # 获取必需参数 - domain = config.get('domain', '').strip() - if not domain: - error_result = { - "status": "error", - "message": "缺少必需参数: domain（要查询的域名）", - "required_params": ["domain"], - "examples": [ - "example.com", - "google.com", - "baidu.com" - ] - } - print(json.dumps(error_result, ensure_ascii=False, indent=2)) - sys.exit(1) - - # 获取可选参数 - limit = config.get('limit', 100) - try: - limit = int(limit) - if limit < 1: - limit = 100 - elif limit > 1000: - limit = 1000 # 限制最大 1000 - except (ValueError, TypeError): - limit = 100 - - include_ips = config.get('include_ips', False) - if isinstance(include_ips, str): - include_ips = include_ips.lower() in ('true', '1', 'yes') - - # 执行查询 - result = query_virustotal_subdomains(domain, api_key, limit, include_ips) - - # 输出结果 - print(json.dumps(result, ensure_ascii=False, indent=2)) - - except Exception as e: - error_result = { - "status": "error", - "message": f"执行出错: {str(e)}", - "type": type(e).__name__ - } - print(json.dumps(error_result, ensure_ascii=False, indent=2)) - sys.exit(1) - + - "import sys\nimport json\nimport requests\nimport os\nimport time\n\n# ==================== VirusTotal 配置 ====================\n# 请在此处配置您的 VirusTotal API 密钥\n# 您也可以在环境变量中设置：VT_API_KEY\n# enable 默认为 false，需开启才能调用该MCP\nVT_API_KEY = \"\" # 请填写您的 VirusTotal API 密钥\n# =======================================================\n\n# VirusTotal API 基础 URL\nBASE_URL = \"https://www.virustotal.com/api/v3\"\n\ndef parse_args():\n \"\"\"解析命令行参数\"\"\"\n # 尝试从第一个参数读取 JSON 配置\n if len(sys.argv) > 1:\n try:\n arg1 = str(sys.argv[1])\n config = json.loads(arg1)\n if isinstance(config, dict):\n return config\n except (json.JSONDecodeError, TypeError, ValueError):\n pass\n\n # 传统位置参数方式\n config = {}\n if len(sys.argv) > 1:\n config['domain'] = str(sys.argv[1])\n if len(sys.argv) > 2:\n try:\n config['limit'] = int(sys.argv[2])\n except (ValueError, TypeError):\n pass\n if len(sys.argv) > 3:\n config['include_ips'] = sys.argv[3].lower() in ('true', '1', 'yes')\n return config\n\ndef query_virustotal_subdomains(domain, api_key, limit=100, include_ips=False):\n \"\"\"\n 查询 VirusTotal 的子域名信息\n \n Args:\n domain: 要查询的域名\n api_key: VirusTotal API 密钥\n limit: 返回结果数量限制\n include_ips: 是否包含 IP 地址信息\n \n Returns:\n dict: 包含查询结果的字典\n \"\"\"\n # 构建 API 请求 URL\n url = f\"{BASE_URL}/domains/{domain}/subdomains\"\n \n headers = {\n \"x-apikey\": api_key,\n \"accept\": \"application/json\"\n }\n \n params = {\n \"limit\": min(limit, 40) # API 限制最大 40\n }\n \n all_results = []\n next_url = None\n \n try:\n # 处理分页\n while True:\n if next_url:\n response = requests.get(next_url, headers=headers, timeout=30)\n else:\n response = requests.get(url, headers=headers, params=params, timeout=30)\n \n response.raise_for_status()\n data = response.json()\n \n # 提取子域名数据\n if 'data' in data and data['data']:\n for item in data['data']:\n if 'id' in item:\n subdomain_info = {\n 'subdomain': item['id'],\n 'type': item.get('type', 'domain'),\n }\n \n # 如果 include_ips 为 True，尝试获取解析 IP\n if include_ips and 'attributes' in item:\n attributes = item.get('attributes', {})\n # 这里简化处理，实际可能需要额外的 API 调用\n subdomain_info['last_dns_records'] = attributes.get('last_dns_records', [])\n \n all_results.append(subdomain_info)\n \n # 检查是否有下一页\n if 'links' in data and 'next' in data['links'] and len(all_results) < limit:\n next_url = data['links']['next']\n # 避免请求过快\n time.sleep(0.5)\n else:\n break\n else:\n break\n \n # 如果已达到限制，停止获取\n if len(all_results) >= limit:\n break\n \n # 处理返回结果\n if all_results:\n return {\n \"status\": \"success\",\n \"domain\": domain,\n \"total_found\": len(all_results),\n \"results\": all_results[:limit],\n \"message\": f\"成功获取 {len(all_results[:limit])} 个子域名\"\n }\n else:\n return {\n \"status\": \"success\",\n \"domain\": domain,\n \"total_found\": 0,\n \"results\": [],\n \"message\": f\"未找到 {domain} 的子域名\"\n }\n \n except requests.exceptions.RequestException as e:\n error_msg = str(e)\n error_result = {\n \"status\": \"error\",\n \"message\": f\"API 请求失败: {error_msg}\",\n \"suggestion\": \"请检查网络连接、API 密钥是否正确，或 VirusTotal API 服务是否可用\"\n }\n \n # 处理特定 HTTP 状态码\n if hasattr(e, 'response') and e.response:\n status_code = e.response.status_code\n if status_code == 401:\n error_result[\"message\"] = \"API 密钥无效或未授权\"\n error_result[\"suggestion\"] = \"请检查 VirusTotal API 密钥是否正确，或在 https://www.virustotal.com/ 获取有效密钥\"\n elif status_code == 429:\n error_result[\"message\"] = \"API 请求频率超限\"\n error_result[\"suggestion\"] = \"请稍后再试，VirusTotal API 有严格的速率限制（免费版每分钟4次）\"\n elif status_code == 404:\n error_result[\"message\"] = f\"域名 '{domain}' 不存在或未找到\"\n \n return error_result\n\ntry:\n config = parse_args()\n \n if not isinstance(config, dict):\n error_result = {\n \"status\": \"error\",\n \"message\": f\"参数解析错误: 期望字典类型，但得到 {type(config).__name__}\",\n \"type\": \"TypeError\"\n }\n print(json.dumps(error_result, ensure_ascii=False, indent=2))\n sys.exit(1)\n \n # 获取 API 密钥（从配置或环境变量）\n api_key = os.getenv('VT_API_KEY', VT_API_KEY).strip()\n \n if not api_key:\n error_result = {\n \"status\": \"error\",\n \"message\": \"缺少 VirusTotal API 密钥\",\n \"required_config\": [\"VT_API_KEY\"],\n \"note\": \"请在 YAML 文件的 VT_API_KEY 配置项中填写您的 VirusTotal API 密钥，或在环境变量 VT_API_KEY 中设置。API 密钥可在 https://www.virustotal.com/ 注册获取\"\n }\n print(json.dumps(error_result, ensure_ascii=False, indent=2))\n sys.exit(1)\n \n # 获取必需参数\n domain = config.get('domain', '').strip()\n if not domain:\n error_result = {\n \"status\": \"error\",\n \"message\": \"缺少必需参数: domain（要查询的域名）\",\n \"required_params\": [\"domain\"],\n \"examples\": [\n \"example.com\",\n \"google.com\",\n \"baidu.com\"\n ]\n }\n print(json.dumps(error_result, ensure_ascii=False, indent=2))\n sys.exit(1)\n \n # 获取可选参数\n limit = config.get('limit', 100)\n try:\n limit = int(limit)\n if limit < 1:\n limit = 100\n elif limit > 1000:\n limit = 1000 # 限制最大 1000\n except (ValueError, TypeError):\n limit = 100\n \n include_ips = config.get('include_ips', False)\n if isinstance(include_ips, str):\n include_ips = include_ips.lower() in ('true', '1', 'yes')\n \n # 执行查询\n result = query_virustotal_subdomains(domain, api_key, limit, include_ips)\n \n # 输出结果\n print(json.dumps(result, ensure_ascii=False, indent=2))\n \nexcept Exception as e:\n error_result = {\n \"status\": \"error\",\n \"message\": f\"执行出错: {str(e)}\",\n \"type\": type(e).__name__\n }\n print(json.dumps(error_result, ensure_ascii=False, indent=2))\n sys.exit(1)\n" enabled: false - short_description: "VirusTotal 子域名查询工具，通过 VirusTotal API 被动收集域名子域名" - description: | VirusTotal 子域名查询工具，利用 VirusTotal 聚合的历史 DNS 数据来发现目标域名的子域名。 - + **主要功能：** - 被动子域名收集：从 VirusTotal 历史 DNS 数据中检索子域名 - 分页查询：支持大量子域名的获取 - IP 关联：可选包含 DNS 解析记录 - 去重处理：自动去重返回结果 - + **使用场景：** - 安全测试前期信息收集 - 企业网络资产发现 - 攻击面分析 - 威胁情报收集 - 渗透测试信息收集 - + **数据来源：** VirusTotal 聚合了来自多个来源的 DNS 数据，包括： - 历史 DNS 解析记录 - 被动 DNS 数据库 - 证书透明度日志 - 安全扫描数据 - + **注意事项：** - **API 密钥必需**：需要在 VirusTotal 注册账号并获取 API 密钥 - **速率限制**：免费版 API 每分钟限制 4 次请求 - **数据时效性**：数据基于历史扫描记录，可能不是实时的 - **使用授权**：仅允许对您拥有合法授权的目标进行查询 - **配额限制**：免费版每月有查询配额限制 - parameters: - name: "domain" type: "string" description: | 要查询的目标域名（必需）。 - + **格式要求：** - 仅输入主域名，不要包含协议头（http://）或路径 - 支持二级域名查询 - + **示例值：** - "example.com" - "google.com" - "baidu.com" - "github.com" - + **注意事项：** - 域名格式必须正确 - 查询结果可能包含跨域子域名 required: true position: 2 format: "positional" - - name: "limit" type: "int" description: | 返回结果数量限制（可选）。 - + **说明：** - 默认值：40 - 最大值：1000（API 限制） - 建议值：100-500 - + **注意事项：** - 设置过大的值可能导致请求超时 - API 单次返回限制为 40 条，超过会自动分页 @@ -299,20 +73,19 @@ parameters: position: 3 format: "positional" default: 40 - - name: "include_ips" type: "bool" description: | 是否包含 IP 地址信息（可选）。 - + **说明：** - true：在结果中包含 DNS 解析记录 - false：仅返回子域名列表 - + **注意事项：** - 包含 IP 信息会增加 API 调用次数 - 可能包含历史解析 IP，不一定准确 required: false position: 4 format: "positional" - default: false \ No newline at end of file + default: false diff --git a/tools/zoomeye_search.yaml b/tools/zoomeye_search.yaml index 45a043b9..544e9fb2 100644 --- a/tools/zoomeye_search.yaml +++ b/tools/zoomeye_search.yaml @@ -17,7 +17,7 @@ args: # ================================================== # ZoomEye API基础URL - base_url = "https://api.zoomeye.org/v2/search" + base_url = "https://api.zoomeye.ai/v2/search" # 解析参数（从JSON字符串或命令行参数） def parse_args(): @@ -78,7 +78,7 @@ args: "status": "error", "message": "缺少ZoomEye配置: api_key（ZoomEye API密钥）", "required_config": ["api_key"], - "note": "请在YAML文件的ZOOMEYE_API_KEY配置项中填写您的API密钥，或在环境变量ZOOMEYE_API_KEY中设置。API密钥可在ZoomEye个人中心获取: https://www.zoomeye.org/user" + "note": "请在YAML文件的ZOOMEYE_API_KEY配置项中填写您的API密钥，或在环境变量ZOOMEYE_API_KEY中设置。API密钥可在ZoomEye个人中心获取: https://www.zoomeye.ai/user" } print(json.dumps(error_result, ensure_ascii=False, indent=2)) sys.exit(1) @@ -174,6 +174,16 @@ args: # 格式化输出结果 # ZoomEye API 返回的数据在 data 字段中 data_list = result_data.get('data', []) + if isinstance(data_list, dict): + nested = None + for key in ('data', 'items', 'matches', 'results', 'list', 'records'): + value = data_list.get(key) + if isinstance(value, list): + nested = value + break + data_list = nested if nested is not None else [] + if not isinstance(data_list, list): + data_list = [] output = { "status": "success", "query": result_data.get('query', query),