228 lines
7.2 KiB
Python
228 lines
7.2 KiB
Python
# -*- coding: UTF-8 -*-
|
|
'''
|
|
@Project :domainScanDemo
|
|
@File :jucha_detector.py
|
|
@IDE :PyCharm
|
|
@Author :梦伴
|
|
@Date :2026/4/9 0:04
|
|
@explain : 聚查检测器
|
|
'''
|
|
|
|
import requests
|
|
from curl_cffi import requests as curl_requests
|
|
import re
|
|
from app.detectors.base import BaseDetector
|
|
|
|
|
|
class JuchaDetector(BaseDetector):
|
|
"""
|
|
聚查检测器
|
|
"""
|
|
|
|
def __init__(self):
|
|
"""
|
|
初始化聚查检测器
|
|
"""
|
|
super().__init__()
|
|
self.url = 'https://www.jucha.com'
|
|
self.whois_url = 'https://www.jucha.com/whois/{domain}'
|
|
self.beian_url = 'https://www.jucha.com/beian/{domain}'
|
|
self.intercept_url = 'https://www.jucha.com/intercept/{domain}'
|
|
|
|
def check_domain(self, domain):
|
|
"""
|
|
检测域名
|
|
|
|
:param domain: 域名
|
|
:return: dict - 检测结果
|
|
"""
|
|
try:
|
|
# 检查WHOIS
|
|
whois_result = self.check_whois(domain)
|
|
|
|
# 检查备案
|
|
beian_result = self.check_beian(domain)
|
|
|
|
# 检查拦截
|
|
intercept_result = self.check_intercept(domain)
|
|
|
|
return {
|
|
'whois': whois_result,
|
|
'beian': beian_result,
|
|
'intercept': intercept_result
|
|
}
|
|
except Exception as e:
|
|
return self._handle_exception(e, domain)
|
|
|
|
def check_whois(self, domain):
|
|
"""
|
|
检查WHOIS
|
|
|
|
:param domain: 域名
|
|
:return: dict - 检测结果
|
|
"""
|
|
try:
|
|
# 构建查询URL
|
|
query_url = self.whois_url.format(domain=domain)
|
|
|
|
headers = {
|
|
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/146.0.0.0 Safari/537.36'
|
|
}
|
|
|
|
# 使用curl_cffi模拟浏览器
|
|
response = curl_requests.get(query_url, headers=headers, impersonate='chrome', timeout=10)
|
|
|
|
if response.status_code == 200:
|
|
content = response.text
|
|
|
|
# 提取WHOIS信息
|
|
whois_info = self._extract_whois_info(content)
|
|
|
|
return whois_info
|
|
else:
|
|
self._log_warning(f"聚查WHOIS查询失败: {response.status_code}")
|
|
return {'status': ''}
|
|
except Exception as e:
|
|
self._handle_exception(e, domain)
|
|
return {'status': ''}
|
|
|
|
def check_beian(self, domain):
|
|
"""
|
|
检查备案
|
|
|
|
:param domain: 域名
|
|
:return: dict - 检测结果
|
|
"""
|
|
try:
|
|
# 构建查询URL
|
|
query_url = self.beian_url.format(domain=domain)
|
|
|
|
headers = {
|
|
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/146.0.0.0 Safari/537.36'
|
|
}
|
|
|
|
# 使用curl_cffi模拟浏览器
|
|
response = curl_requests.get(query_url, headers=headers, impersonate='chrome', timeout=10)
|
|
|
|
if response.status_code == 200:
|
|
content = response.text
|
|
|
|
# 提取备案信息
|
|
beian_info = self._extract_beian_info(content)
|
|
|
|
return beian_info
|
|
else:
|
|
self._log_warning(f"聚查备案查询失败: {response.status_code}")
|
|
return {'has_beian': False, 'beian_year': '', 'is_enterprise': False, 'beian_match': False}
|
|
except Exception as e:
|
|
self._handle_exception(e, domain)
|
|
return {'has_beian': False, 'beian_year': '', 'is_enterprise': False, 'beian_match': False}
|
|
|
|
def check_intercept(self, domain):
|
|
"""
|
|
检查拦截
|
|
|
|
:param domain: 域名
|
|
:return: dict - 检测结果
|
|
"""
|
|
try:
|
|
# 构建查询URL
|
|
query_url = self.intercept_url.format(domain=domain)
|
|
|
|
headers = {
|
|
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/146.0.0.0 Safari/537.36'
|
|
}
|
|
|
|
# 使用curl_cffi模拟浏览器
|
|
response = curl_requests.get(query_url, headers=headers, impersonate='chrome', timeout=10)
|
|
|
|
if response.status_code == 200:
|
|
content = response.text
|
|
|
|
# 检查是否被拦截
|
|
is_normal = self._check_intercept_status(content)
|
|
|
|
return {
|
|
'normal': is_normal
|
|
}
|
|
else:
|
|
self._log_warning(f"聚查拦截查询失败: {response.status_code}")
|
|
return {'normal': False}
|
|
except Exception as e:
|
|
self._handle_exception(e, domain)
|
|
return {'normal': False}
|
|
|
|
def _extract_whois_info(self, content):
|
|
"""
|
|
提取WHOIS信息
|
|
|
|
:param content: 页面内容
|
|
:return: dict - WHOIS信息
|
|
"""
|
|
try:
|
|
# 提取状态信息
|
|
pattern = r'域名状态：<span[^>]+>(.*?)</span>'
|
|
match = re.search(pattern, content)
|
|
status = match.group(1).strip() if match else ''
|
|
|
|
return {
|
|
'status': status
|
|
}
|
|
except Exception as e:
|
|
self._handle_exception(e, 'extract_whois_info')
|
|
return {'status': ''}
|
|
|
|
def _extract_beian_info(self, content):
|
|
"""
|
|
提取备案信息
|
|
|
|
:param content: 页面内容
|
|
:return: dict - 备案信息
|
|
"""
|
|
try:
|
|
# 检查是否有备案
|
|
has_beian = '备案信息' in content
|
|
|
|
# 提取备案年份
|
|
beian_year = ''
|
|
pattern = r'审核时间：(\d{4})-\d{2}-\d{2}'
|
|
match = re.search(pattern, content)
|
|
if match:
|
|
beian_year = match.group(1)
|
|
|
|
# 提取单位性质
|
|
is_enterprise = '企业' in content
|
|
|
|
# 检查首网址和备案网址是否一致
|
|
beian_match = '网站首页网址' in content
|
|
|
|
return {
|
|
'has_beian': has_beian,
|
|
'beian_year': beian_year,
|
|
'is_enterprise': is_enterprise,
|
|
'beian_match': beian_match
|
|
}
|
|
except Exception as e:
|
|
self._handle_exception(e, 'extract_beian_info')
|
|
return {'has_beian': False, 'beian_year': '', 'is_enterprise': False, 'beian_match': False}
|
|
|
|
def _check_intercept_status(self, content):
|
|
"""
|
|
检查拦截状态
|
|
|
|
:param content: 页面内容
|
|
:return: bool - 是否正常
|
|
"""
|
|
try:
|
|
# 检查是否包含正常标识
|
|
if '正常' in content:
|
|
return True
|
|
|
|
# 检查是否包含拦截标识
|
|
if '拦截' in content:
|
|
return False
|
|
|
|
return False
|
|
except Exception as e:
|
|
self._handle_exception(e, 'check_intercept_status')
|
|
return False |