diff --git a/tools/scanmalware_search.yaml b/tools/scanmalware_search.yaml new file mode 100644 index 000000000..90831c15f --- /dev/null +++ b/tools/scanmalware_search.yaml @@ -0,0 +1,514 @@ +name: "scanmalware_search" +command: "python3" +args: + - "-c" + - | + import argparse + import ipaddress + import json + import sys + from urllib.parse import quote, urlsplit, urlunsplit + + try: + import requests + except ImportError: + print(json.dumps({"status": "error", "message": "缺少依赖: requests,请执行 pip install requests"}, ensure_ascii=False, indent=2)) + sys.exit(1) + + # ScanMalware 公开 API:无需 API Key,匿名限速 600 次/分钟 + BASE_URL = "https://scanmalware.com/api/v1" + REPORT_URL = "https://scanmalware.com/result/{}" + USER_AGENT = "scanmalware-cyberstrikeai/1.0 (+https://github.com/AIPentest/CyberStrikeAI)" + TIMEOUT = 30 + # 冷查询可能需要数秒,结果获取次数设上限以控制请求量 + MAX_RESULT_FETCHES = 3 + + # security_verdict.risk_level 的实际取值:low / medium / high / malicious / critical / unknown + # "malicious" 是扫描器给出的最强结论,必须计入恶意;"low" 只表示未发现问题,不代表安全 + MALICIOUS_LEVELS = ("malicious", "critical", "high") + SUSPICIOUS_LEVELS = ("medium",) + + + class ApiError(Exception): + def __init__(self, status_code, detail): + super().__init__(f"HTTP {status_code}: {detail}") + self.status_code = status_code + self.detail = detail + + + def api_get(path, params=None): + response = requests.get( + BASE_URL + path, + params=params, + headers={"User-Agent": USER_AGENT, "Accept": "application/json"}, + timeout=TIMEOUT, + ) + if response.status_code != 200: + detail = response.text[:300] + try: + body = response.json() + if isinstance(body, dict) and body.get("detail") is not None: + detail = body["detail"] + except ValueError: + pass + raise ApiError(response.status_code, detail) + return response.json() + + + def to_int(value, default, low, high): + try: + number = int(float(value)) + except (TypeError, ValueError): + return default + return max(low, min(high, number)) + + + def normalize_host(value): + value = value.strip() + if "://" in value: + value = urlsplit(value).hostname or "" + else: + value = value.split("/")[0].split("?")[0] + if value.count(":") == 1: + value = value.split(":")[0] + return value.strip().rstrip(".").lower() + + + def normalize_url(value): + parts = urlsplit(value.strip()) + scheme = parts.scheme.lower() + host = (parts.hostname or "").rstrip(".").lower() + port = parts.port + if port and not ((scheme == "http" and port == 80) or (scheme == "https" and port == 443)): + host = f"{host}:{port}" + return urlunsplit((scheme, host, parts.path or "/", parts.query, "")) + + + def summarize_verdict(result): + verdict = result.get("security_verdict") or {} + level = str(verdict.get("risk_level") or "unknown").lower() + if level in MALICIOUS_LEVELS: + assessment = "malicious" + elif level in SUSPICIOUS_LEVELS: + assessment = "suspicious" + elif level == "low": + assessment = "no_findings" + elif level == "unknown": + assessment = "not_assessed" + else: + # 未知的新取值原样返回,不跳过也不猜测 + assessment = "unrecognized" + technologies = (result.get("technologies") or {}).get("detected") or [] + return { + "scan_id": result.get("scan_id"), + "scanned_url": result.get("url"), + "final_url": result.get("final_url"), + "submitted_at": result.get("submitted_at"), + "title": result.get("title"), + "assessment": assessment, + "verdict": verdict.get("verdict"), + "risk_level": level, + "confidence": verdict.get("confidence"), + "threat_categories": verdict.get("threat_categories") or [], + "risk_factors": (verdict.get("risk_factors") or [])[:5], + "primary_ip": result.get("primary_ip"), + "primary_asn": result.get("primary_asn"), + "asn_names": result.get("asn_names") or [], + "technologies": [t.get("name") for t in technologies if isinstance(t, dict)][:15], + "report_url": REPORT_URL.format(result.get("scan_id")), + } + + + def latest_assessed_verdict(rows): + """按时间从新到旧获取扫描结果,跳过未完成评估(risk_level=unknown)的扫描。""" + skipped = [] + for row in rows[:MAX_RESULT_FETCHES]: + summary = summarize_verdict(api_get(f"/result/{row['scan_id']}")) + if summary["assessment"] != "not_assessed": + return summary, skipped + skipped.append(row["scan_id"]) + return None, skipped + + + def compact_scan_row(row): + return { + "scan_id": row.get("scan_id"), + "url": row.get("url"), + "final_url": row.get("final_url"), + "matched_on": row.get("matched_on"), + "title": row.get("title"), + "submitted_at": row.get("submitted_at"), + "report_url": REPORT_URL.format(row.get("scan_id")), + } + + + def domain_scans(host, limit): + return api_get(f"/domains/{quote(host)}/scans", {"limit": limit, "status": "completed"}) + + + def lookup_domain(target, limit, quick): + host = normalize_host(target) + rows = domain_scans(host, limit) + # matched_on 包含 "url" 才表示扫描地址本身属于该主机;只有 final_url 的是重定向到该主机的其他页面 + own = [r for r in rows if "url" in (r.get("matched_on") or [])] + redirected = [r for r in rows if "url" not in (r.get("matched_on") or [])] + output = { + "status": "success", + "mode": "domain", + "target": host, + "scans_of_target": len(own), + "scans_redirecting_here": len(redirected), + "recent_scans": [compact_scan_row(r) for r in own], + "redirected_here": [compact_scan_row(r) for r in redirected], + "note": "verdict 只取自扫描地址本身属于该主机的扫描;仅重定向到该主机的扫描不代表该主机,不附带 verdict", + } + if not quick: + if own: + verdict, skipped = latest_assessed_verdict(own) + output["latest_verdict"] = verdict + output["skipped_not_assessed"] = skipped + if verdict is None: + output["verdict_note"] = "最近的扫描均未完成评估(页面未能加载等),没有可用的 verdict" + else: + output["latest_verdict"] = None + output["verdict_note"] = "ScanMalware 从未直接扫描过该主机,没有 verdict" + if output.get("latest_verdict") and output["latest_verdict"]["assessment"] == "no_findings": + output["verdict_note"] = "低风险仅表示扫描未发现问题,不代表目标安全" + output["message"] = f"{host}: {len(own)} 次直接扫描,{len(redirected)} 次重定向到该主机" + return output + + + def lookup_url(target, limit, quick): + if "://" not in target: + raise ValueError("url 模式需要完整 URL,例如 https://example.com/login") + wanted = normalize_url(target) + host = normalize_host(target) + rows = domain_scans(host, 100) + exact = [ + r for r in rows + if "url" in (r.get("matched_on") or []) and r.get("url") and normalize_url(r["url"]) == wanted + ] + output = { + "status": "success", + "mode": "url", + "target": wanted, + "host": host, + "exact_scans": [compact_scan_row(r) for r in exact[:limit]], + "other_scans_on_host": len(rows) - len(exact), + "note": "只匹配扫描地址与目标 URL 完全一致的扫描;同一主机上其他 URL 的结论不会被套用到该 URL", + } + if not exact: + output["latest_verdict"] = None + output["verdict_note"] = "该 URL 未被扫描过(仅检查该主机最近 100 次扫描),不借用同主机其他页面的 verdict" + elif not quick: + verdict, skipped = latest_assessed_verdict(exact) + output["latest_verdict"] = verdict + output["skipped_not_assessed"] = skipped + if verdict is None: + output["verdict_note"] = "最近的扫描均未完成评估,没有可用的 verdict" + elif verdict["assessment"] == "no_findings": + output["verdict_note"] = "低风险仅表示扫描未发现问题,不代表目标安全" + output["message"] = f"{wanted}: {len(exact)} 次完全匹配的扫描" + return output + + + def lookup_ip(target, limit): + address = str(ipaddress.ip_address(target.strip())) + ct = api_get(f"/ct/ip/{quote(address)}") + scans = api_get(f"/search/ip/{quote(address)}", {"limit": min(limit, 100)}) + ct_domains = ct.get("domains") or [] + rows = scans.get("results") or [] + return { + "status": "success", + "mode": "ip", + "target": address, + "ct_domains_total": ct.get("total", len(ct_domains)), + "ct_domains": ct_domains[:limit], + "scans_contacting_ip_total": scans.get("total", len(rows)), + "scans_contacting_ip": [ + { + "scan_id": r.get("scan_id"), + "url": r.get("url"), + "title": r.get("title"), + "submitted_at": r.get("submitted_at"), + "ip_is_primary": r.get("primary_ip") == address, + "report_url": REPORT_URL.format(r.get("scan_id")), + } + for r in rows + ], + "note": "IP 只提供上下文,不给出 verdict:页面访问过或解析到该地址,不代表该地址本身是恶意的", + "message": f"{address}: 证书透明度记录中 {ct.get('total', len(ct_domains))} 个域名,{scans.get('total', len(rows))} 次扫描访问过该地址", + } + + + def in_scope(host, domain): + return host == domain or host.endswith("." + domain) + + + def lookup_subdomains(target, limit): + domain = normalize_host(target) + found = {} + sources = {} + warnings = [] + failures = 0 + + def add(host, source): + host = (host or "").strip().rstrip(".").lower() + if not host or host.startswith("*.") or not in_scope(host, domain): + return + found.setdefault(host, set()).add(source) + + # 1) 证书透明度 DNS 记录 + try: + ct = api_get(f"/ct/dns/{quote(domain)}", {"subdomain_limit": min(limit, 5000)}) + for host in ct.get("subdomains") or []: + add(host, "ct_dns") + sources["ct_dns"] = len(ct.get("subdomains") or []) + if ct.get("subdomains_truncated"): + warnings.append("ct_dns 结果已截断,调大 limit(最大 5000)可获取更多") + if ct.get("degraded"): + warnings.append(f"ct_dns 暂时无法完整回答({ct.get('degraded_reason') or 'unspecified'}),结果不完整,请稍后重试") + except (ApiError, requests.exceptions.RequestException) as e: + failures += 1 + warnings.append(f"ct_dns 查询失败: {e}") + + # 2) 扫描归档中的主机名 + try: + pages = max(1, min(10, (limit + 99) // 100)) + count = 0 + for page in range(1, pages + 1): + data = api_get("/search/smql", {"q": f"domain:*.{domain}", "limit": 100, "page": page}) + rows = data.get("results") or [] + for row in rows: + for key in ("url", "final_url"): + if row.get(key): + add(urlsplit(row[key]).hostname, "scan_archive") + count += len(rows) + if not (data.get("pagination") or {}).get("has_next"): + break + else: + warnings.append(f"scan_archive 在第 {pages} 页停止,可能还有更多") + sources["scan_archive_rows"] = count + except (ApiError, requests.exceptions.RequestException) as e: + failures += 1 + warnings.append(f"scan_archive 查询失败: {e}") + + # 3) 浏览器在扫描中实际观察到的主机 + source_meaning = {} + try: + hosts = api_get(f"/hosts/{quote(domain)}", {"subdomains_only": "true"}) + for name, values in (hosts.get("sources") or {}).items(): + for host in values or []: + add(host, f"hosts:{name}") + sources[f"hosts:{name}"] = len(values or []) + source_meaning = hosts.get("source_meaning") or {} + for name, truncated in (hosts.get("truncated") or {}).items(): + if truncated: + warnings.append(f"hosts:{name} 结果已截断") + except (ApiError, requests.exceptions.RequestException) as e: + failures += 1 + warnings.append(f"hosts 查询失败: {e}") + + names = sorted(found) + return { + "status": "error" if failures == 3 else "success", + "mode": "subdomains", + "target": domain, + "total_found": len(names), + "subdomains": [{"host": h, "sources": sorted(found[h])} for h in names[:limit]], + "per_source": sources, + "source_meaning": source_meaning, + "warnings": warnings, + "note": "来源:证书透明度 DNS 记录、ScanMalware 公开扫描归档、扫描时浏览器实际访问的主机。结果不是完整枚举", + "message": f"{domain}: 发现 {len(names)} 个主机名" + ("(返回前 %d 个)" % limit if len(names) > limit else ""), + } + + + def lookup_smql(query, limit, page): + data = api_get("/search/smql", {"q": query, "limit": min(limit, 100), "page": page}) + rows = data.get("results") or [] + return { + "status": "success", + "mode": "smql", + "query": query, + "pagination": data.get("pagination"), + "results": [ + { + "scan_id": r.get("scan_id"), + "url": r.get("url"), + "final_url": r.get("final_url"), + "title": r.get("title"), + "submitted_at": r.get("submitted_at"), + "primary_asn": r.get("primary_asn"), + "asn_names": r.get("asn_names"), + "countries": r.get("countries"), + "report_url": REPORT_URL.format(r.get("scan_id")), + } + for r in rows + ], + "note": "SMQL 结果行不含 verdict;需要结论时请对具体 URL 或域名使用 url / domain 模式", + "message": f"共 {(data.get('pagination') or {}).get('total_items', len(rows))} 条匹配,本页返回 {len(rows)} 条", + } + + + def main(): + parser = argparse.ArgumentParser(description="ScanMalware public scan archive lookup") + parser.add_argument("--mode", required=True, choices=["domain", "url", "ip", "subdomains", "smql"]) + parser.add_argument("--target", required=True) + parser.add_argument("--limit", default="20") + parser.add_argument("--page", default="1") + parser.add_argument("--quick", action="store_true") + args = parser.parse_args() + + target = args.target.strip() + if not target: + raise ValueError("target 不能为空") + + if args.mode == "subdomains": + limit = to_int(args.limit, 20, 1, 5000) + else: + limit = to_int(args.limit, 20, 1, 100) + + if args.mode == "domain": + return lookup_domain(target, limit, args.quick) + if args.mode == "url": + return lookup_url(target, limit, args.quick) + if args.mode == "ip": + return lookup_ip(target, limit) + if args.mode == "subdomains": + return lookup_subdomains(target, limit) + return lookup_smql(target, limit, to_int(args.page, 1, 1, 1000)) + + + try: + result = main() + print(json.dumps(result, ensure_ascii=False, indent=2)) + if result.get("status") != "success": + sys.exit(1) + except ApiError as e: + print(json.dumps({ + "status": "error", + "message": f"ScanMalware API 返回 {e.status_code}", + "detail": e.detail, + "suggestion": "400/422 表示输入无效(例如 SMQL 语法或取值错误、非法域名或 IP);429 表示触发限速,请稍后重试", + }, ensure_ascii=False, indent=2)) + sys.exit(1) + except requests.exceptions.RequestException as e: + print(json.dumps({ + "status": "error", + "message": f"请求失败: {e}", + "suggestion": "请检查网络连接,或稍后重试(冷查询可能需要数秒)", + }, ensure_ascii=False, indent=2)) + sys.exit(1) + except ValueError as e: + print(json.dumps({"status": "error", "message": str(e)}, ensure_ascii=False, indent=2)) + sys.exit(1) + except Exception as e: + print(json.dumps({"status": "error", "message": f"执行出错: {e}", "type": type(e).__name__}, ensure_ascii=False, indent=2)) + sys.exit(1) +enabled: false +short_description: "ScanMalware 公开扫描归档查询:URL/域名结论、被动子域名、IP 关联与 SMQL 检索,无需 API Key" +description: | + ScanMalware(https://scanmalware.com)公开扫描归档查询工具。ScanMalware 在沙箱浏览器中渲染提交的 URL,并将每次扫描的结果(钓鱼/恶意判定、网络请求、TLS/JARM、favicon 哈希、技术栈、截图等)编入可检索的公开归档。本工具只读取归档,不提交新的扫描。 + + **无需 API Key:** 匿名即可使用,公开 API 限速为 600 次/分钟。 + + **查询模式(mode):** + - `domain`:主机的近期扫描,以及最近一次已完成评估的 verdict + - `url`:扫描地址与目标 URL 完全一致的扫描及其 verdict + - `ip`:证书透明度 DNS 记录中解析到该 IP 的域名,以及访问过该 IP 的扫描(仅上下文,不给出 verdict) + - `subdomains`:被动子域名收集,合并证书透明度 DNS 记录、扫描归档中的主机名、扫描时浏览器实际访问的主机 + - `smql`:用 SMQL 语法检索整个扫描归档,按 JARM、favicon 哈希、ASN、技术栈、OCR 文本等横向关联 + + **SMQL 示例:** + - `verdict:CONFIRMED_SCAM` + - `domain:*.github.io` + - `ip:104.20.23.154` + - `asn:13335` + - `technology:WordPress` + - `title:"Login"` + - `ocr:paypal` + - `favicon_hash:1983356674`(favicon mmh3) + - `jarm:<62 位 JARM 指纹>` + - 组合与排除:`domain:*.github.io -verdict:LOW_RISK` + - verdict 取值:LEGITIMATE、LOW_RISK、MODERATE_RISK、HIGH_RISK、CONFIRMED_SCAM、NOT_ASSESSED + - 完整过滤器列表:https://scanmalware.com/api/v1/search/smql/filters + + **结论归属规则:** + - verdict 只取自扫描地址本身属于目标的扫描;仅重定向到目标的扫描单独列出,不附带 verdict + - `url` 模式不会把同一主机上其他页面的结论套用到目标 URL + - IP 只提供上下文,不给出 verdict + - 风险等级 `low` 只表示扫描未发现问题,不代表目标安全;`malicious`、`high`、`critical` 记为恶意;无法识别的新取值原样返回(`assessment: unrecognized`) + - 最近的扫描如果未完成评估(例如页面未能加载),会跳过并使用更早的已评估扫描,最多检查 3 次 + + **注意事项:** + - **隐私:** 查询的域名、URL、IP 会发送到 scanmalware.com。根据该 API 在 `/hosts` 返回中的说明(`source_meaning.stream`),被查询过且能在公共 DNS 中解析的主机名,可能以仅存在性的 `stream` 来源出现在其结果中。请勿查询不应外泄的内部或未公开目标 + - 默认 `enabled: false`,确认可以向第三方发送查询目标后再开启 + - 结果基于历史扫描,不代表目标当前状态;子域名结果不是完整枚举 + - 冷查询可能需要数秒 + - 请仅对拥有合法授权的目标进行查询 +parameters: + - name: "mode" + type: "string" + description: | + 查询模式(必需)。 + + - `domain`:主机的扫描记录与 verdict,target 为域名或主机名 + - `url`:完全匹配的 URL 扫描与 verdict,target 为完整 URL + - `ip`:IP 关联的域名与扫描,target 为 IPv4 或 IPv6 地址 + - `subdomains`:被动子域名收集,target 为根域名 + - `smql`:SMQL 检索,target 为 SMQL 查询语句 + required: true + flag: "--mode" + format: "flag" + options: + - "domain" + - "url" + - "ip" + - "subdomains" + - "smql" + - name: "target" + type: "string" + description: | + 查询目标(必需),含义取决于 mode。 + + **示例值:** + - domain:`example.com` + - url:`https://example.com/login` + - ip:`104.20.23.154` + - subdomains:`example.com` + - smql:`verdict:CONFIRMED_SCAM`、`domain:*.github.io -verdict:LOW_RISK` + required: true + flag: "--target" + format: "combined" + - name: "limit" + type: "int" + description: | + 每个列表返回的最大条数(可选)。 + + - 默认 20 + - `domain` 模式取该主机最近 limit 条扫描,再区分直接扫描与重定向 + - `subdomains` 模式最大 5000,其余模式最大 100 + required: false + flag: "--limit" + format: "flag" + default: 20 + - name: "page" + type: "int" + description: | + 页码(可选,仅 smql 模式生效),从 1 开始,默认 1。 + required: false + flag: "--page" + format: "flag" + default: 1 + - name: "quick" + type: "bool" + description: | + 快速模式(可选,仅 domain 与 url 模式生效)。 + + - `false`(默认):额外获取扫描结果以返回 verdict + - `true`:只返回扫描列表,不获取 verdict,减少请求数 + required: false + flag: "--quick" + format: "flag" + default: false