performing-brand-monitoring-for-impersonation监控域名、社交媒体、移动应用和暗网渠道中的品牌仿冒攻击，检测针对组织的网络钓鱼活动、虚假站点和未授权品牌使用行为。
Independent GitHub snapshot · Automated classification · Original documentation governs
品牌仿冒攻击通过仿冒域名、虚假社交媒体账号、假冒移动应用和模仿合法品牌的网络钓鱼站点来利用消费者信任。2025 年，品牌仿冒仍是代价最高的网络威胁之一，AI 生成的网络钓鱼邮件点击率高达 54%。本技能涵盖构建综合品牌监控计划，使用自动化扫描和告警检测域名抢注、社交媒体仿冒、虚假移动应用、未授权 Logo 使用和暗网品牌提及。
dnstwist、requests、beautifulsoup4、Levenshtein、tweepy 库品牌仿冒跨越多个渠道：域名抢注（错字抢注、同形字、TLD 变体）、网络钓鱼站点（克隆使用被盗品牌的网站）、社交媒体（仿冒高管或公司的虚假账号）、移动应用（应用商店中的假冒应用）、电子邮件欺骗（显示名和域名仿冒）以及暗网（论坛和市场中的品牌提及）。
有效的品牌监控结合了：主动扫描（使用 dnstwist 的域名置换、CT 日志监控）、Web 爬取（截图对比、Logo 检测）、社交媒体监控（账号名匹配、帖子内容分析）、应用商店监控（名称和图标相似度检测）以及暗网监控（论坛爬取、市场跟踪）。
并非所有仿冒都是恶意的。风险因素包括：活跃的 Web 内容（尤其是登录页面）、存在 SSL 证书、已配置 MX 记录（具备接收邮件能力）、与合法站点视觉高度相似、近期注册日期，以及托管在与网络犯罪相关的地区。
import subprocess
import requests
import json
from datetime import datetime
from urllib.parse import urlparse
import Levenshtein
class BrandMonitor:
def __init__(self, brand_config):
self.brand_name = brand_config["name"]
self.domains = brand_config["domains"]
self.keywords = brand_config["keywords"]
self.executive_names = brand_config.get("executives", [])
self.logo_hash = brand_config.get("logo_hash", "")
self.findings = []
def scan_domain_squatting(self):
"""检测错字抢注和仿冒域名。"""
all_results = []
for domain in self.domains:
cmd = ["dnstwist", "--registered", "--format", "json",
"--nameservers", "8.8.8.8", "--threads", "30", domain]
try:
result = subprocess.run(cmd, capture_output=True, text=True, timeout=300)
if result.returncode == 0:
domains = json.loads(result.stdout)
registered = [d for d in domains if d.get("dns_a") or d.get("dns_aaaa")]
all_results.extend(registered)
print(f"[+] {domain} 域名抢注扫描：发现 {len(registered)} 个已注册仿冒域名")
except (subprocess.TimeoutExpired, Exception) as e:
print(f"[-] 扫描 {domain} 时出错：{e}")
for entry in all_results:
self.findings.append({
"type": "domain_squatting",
"indicator": entry.get("domain", ""),
"fuzzer": entry.get("fuzzer", ""),
"dns_a": entry.get("dns_a", []),
"ssdeep_score": entry.get("ssdeep_score", 0),
"detected_at": datetime.now().isoformat(),
})
return all_results
def check_google_safe_browsing(self, urls, api_key):
"""通过 Google Safe Browsing API 检查 URL。"""
url = f"https://safebrowsing.googleapis.com/v4/threatMatches:find?key={api_key}"
body = {
"client": {"clientId": "brand-monitor", "clientVersion": "1.0"},
"threatInfo": {
"threatTypes": ["MALWARE", "SOCIAL_ENGINEERING", "UNWANTED_SOFTWARE"],
"platformTypes": ["ANY_PLATFORM"],
"threatEntryTypes": ["URL"],
"threatEntries": [{"url": u} for u in urls],
},
}
resp = requests.post(url, json=body, timeout=15)
if resp.status_code == 200:
matches = resp.json().get("matches", [])
print(f"[+] Google Safe Browsing：发现 {len(matches)} 个威胁")
return matches
return []
def monitor_social_media_impersonation(self, platform="twitter"):
"""检测仿冒品牌或高管的社交媒体账号。"""
suspicious_profiles = []
# 搜索具有相似名称的账号
for name in self.executive_names + [self.brand_name]:
# 通用搜索方法
search_url = f"https://api.twitter.com/2/users/by/username/{name.replace(' ', '')}"
# 注意：生产环境中需使用已认证的 Twitter API
suspicious_profiles.append({
"search_term": name,
"platform": platform,
"note": "完整搜索需要已认证的 API 访问",
})
return suspicious_profiles
def monitor_app_stores(self):
"""检查应用商店中仿冒品牌的虚假移动应用。"""
fake_apps = []
for keyword in self.keywords:
# Google Play Store 搜索（非官方）
url = f"https://play.google.com/store/search?q={keyword}&c=apps"
try:
resp = requests.get(url, timeout=15, headers={
"User-Agent": "Mozilla/5.0"
})
if resp.status_code == 200:
# 解析结果，匹配品牌名称
from bs4 import BeautifulSoup
soup = BeautifulSoup(resp.text, "html.parser")
app_links = soup.find_all("a", href=lambda h: h and "/store/apps/details" in h)
for link in app_links:
app_name = link.get_text(strip=True)
if any(k.lower() in app_name.lower() for k in self.keywords):
fake_apps.append({
"name": app_name,
"url": f"https://play.google.com{link['href']}",
"platform": "google_play",
"keyword": keyword,
})
except Exception as e:
print(f"[-] 应用商店搜索出错：{e}")
return fake_apps
def generate_monitoring_report(self):
report = {
"brand": self.brand_name,
"generated": datetime.now().isoformat(),
"total_findings": len(self.findings),
"findings_by_type": {},
"high_priority": [],
}
for finding in self.findings:
ftype = finding["type"]
if ftype not in report["findings_by_type"]:
report["findings_by_type"][ftype] = 0
report["findings_by_type"][ftype] += 1
# 高优先级：具有较高 Web 相似度或 MX 记录
if finding.get("ssdeep_score", 0) > 50:
report["high_priority"].append(finding)
with open(f"brand_monitoring_{self.brand_name.lower()}.json", "w") as f:
json.dump(report, f, indent=2)
print(f"[+] 品牌监控报告：{len(self.findings)} 个发现")
return report
monitor = BrandMonitor({
"name": "MyCompany",
"domains": ["mycompany.com", "mycompany.org"],
"keywords": ["mycompany", "mybrand", "myproduct"],
"executives": ["CEO Name", "CTO Name"],
})
monitor.scan_domain_squatting()
report = monitor.generate_monitoring_report()
def generate_takedown_request(finding, brand_info):
"""生成针对域名/站点的滥用举报以申请下架。"""
request = f"""主题：滥用举报 - 品牌仿冒 / 网络钓鱼
尊敬的滥用处理团队，
我们正在举报一个仿冒 {brand_info['name']} 的域名，
疑似用于网络钓鱼/欺诈目的。
侵权域名：{finding.get('indicator', '')}
IP 地址：{', '.join(finding.get('dns_a', ['未知']))}
检测方法：{finding.get('fuzzer', '域名相似性分析')}
Web 相似度评分：{finding.get('ssdeep_score', 'N/A')}%
检测日期：{finding.get('detected_at', '')}
我方合法域名：{', '.join(brand_info['domains'])}
该域名通过 {finding.get('fuzzer', '错字抢注')} 方式仿冒我方品牌。
请立即暂停该域名。
侵权证据可应要求提供。
此致
{brand_info['name']} 安全团队
"""
return request