315 lines
9.8 KiB
Python
315 lines
9.8 KiB
Python
#!/usr/bin/env python
|
||
# -*- coding: utf-8 -*-
|
||
|
||
"""
|
||
HTTP客户端模块，负责发送HTTP请求
|
||
"""
|
||
|
||
import logging
|
||
import requests
|
||
import random
|
||
import time
|
||
from urllib.parse import urlparse, urljoin
|
||
from requests.exceptions import RequestException, Timeout, ConnectionError
|
||
from requests.packages.urllib3.exceptions import InsecureRequestWarning
|
||
|
||
# 禁用不安全请求的警告
|
||
requests.packages.urllib3.disable_warnings(InsecureRequestWarning)
|
||
|
||
logger = logging.getLogger('xss_scanner')
|
||
|
||
class HttpClient:
|
||
"""HTTP客户端类，负责处理HTTP请求"""
|
||
|
||
def __init__(self, timeout=10, user_agent=None, proxy=None, cookies=None, headers=None, verify_ssl=False):
|
||
"""
|
||
初始化HTTP客户端
|
||
|
||
Args:
|
||
timeout: 请求超时时间
|
||
user_agent: 用户代理
|
||
proxy: 代理
|
||
cookies: Cookies
|
||
headers: 自定义HTTP头
|
||
verify_ssl: 是否验证SSL证书
|
||
"""
|
||
self.timeout = timeout
|
||
self.user_agent = user_agent
|
||
self.proxy = proxy
|
||
self.cookies = cookies or {}
|
||
self.headers = headers or {}
|
||
self.verify_ssl = verify_ssl
|
||
self.session = requests.Session()
|
||
|
||
# 设置默认User-Agent
|
||
if not self.user_agent:
|
||
self.user_agent = (
|
||
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
|
||
"AppleWebKit/537.36 (KHTML, like Gecko) "
|
||
"Chrome/91.0.4472.124 Safari/537.36"
|
||
)
|
||
|
||
# 设置默认请求头
|
||
if 'User-Agent' not in self.headers:
|
||
self.headers['User-Agent'] = self.user_agent
|
||
|
||
def get(self, url, params=None, headers=None, cookies=None, allow_redirects=True, timeout=None):
|
||
"""
|
||
发送GET请求
|
||
|
||
Args:
|
||
url: 请求的URL
|
||
params: 请求参数
|
||
headers: 请求头
|
||
cookies: Cookies
|
||
allow_redirects: 是否允许重定向
|
||
timeout: 超时时间
|
||
|
||
Returns:
|
||
requests.Response: 响应对象
|
||
"""
|
||
merged_headers = self._merge_headers(headers)
|
||
merged_cookies = self._merge_cookies(cookies)
|
||
timeout = timeout or self.timeout
|
||
|
||
try:
|
||
response = self.session.get(
|
||
url=url,
|
||
params=params,
|
||
headers=merged_headers,
|
||
cookies=merged_cookies,
|
||
proxies=self._get_proxies(),
|
||
allow_redirects=allow_redirects,
|
||
timeout=timeout,
|
||
verify=self.verify_ssl
|
||
)
|
||
return response
|
||
except Timeout:
|
||
logger.warning(f"请求超时: {url}")
|
||
except ConnectionError:
|
||
logger.warning(f"连接错误: {url}")
|
||
except RequestException as e:
|
||
logger.warning(f"请求异常: {url} - {str(e)}")
|
||
except Exception as e:
|
||
logger.error(f"发送GET请求时发生错误: {url} - {str(e)}")
|
||
|
||
return None
|
||
|
||
def post(self, url, data=None, json=None, headers=None, cookies=None, allow_redirects=True, timeout=None):
|
||
"""
|
||
发送POST请求
|
||
|
||
Args:
|
||
url: 请求的URL
|
||
data: 表单数据
|
||
json: JSON数据
|
||
headers: 请求头
|
||
cookies: Cookies
|
||
allow_redirects: 是否允许重定向
|
||
timeout: 超时时间
|
||
|
||
Returns:
|
||
requests.Response: 响应对象
|
||
"""
|
||
merged_headers = self._merge_headers(headers)
|
||
merged_cookies = self._merge_cookies(cookies)
|
||
timeout = timeout or self.timeout
|
||
|
||
try:
|
||
response = self.session.post(
|
||
url=url,
|
||
data=data,
|
||
json=json,
|
||
headers=merged_headers,
|
||
cookies=merged_cookies,
|
||
proxies=self._get_proxies(),
|
||
allow_redirects=allow_redirects,
|
||
timeout=timeout,
|
||
verify=self.verify_ssl
|
||
)
|
||
return response
|
||
except Timeout:
|
||
logger.warning(f"请求超时: {url}")
|
||
except ConnectionError:
|
||
logger.warning(f"连接错误: {url}")
|
||
except RequestException as e:
|
||
logger.warning(f"请求异常: {url} - {str(e)}")
|
||
except Exception as e:
|
||
logger.error(f"发送POST请求时发生错误: {url} - {str(e)}")
|
||
|
||
return None
|
||
|
||
def head(self, url, headers=None, cookies=None, allow_redirects=True, timeout=None):
|
||
"""
|
||
发送HEAD请求
|
||
|
||
Args:
|
||
url: 请求的URL
|
||
headers: 请求头
|
||
cookies: Cookies
|
||
allow_redirects: 是否允许重定向
|
||
timeout: 超时时间
|
||
|
||
Returns:
|
||
requests.Response: 响应对象
|
||
"""
|
||
merged_headers = self._merge_headers(headers)
|
||
merged_cookies = self._merge_cookies(cookies)
|
||
timeout = timeout or self.timeout
|
||
|
||
try:
|
||
response = self.session.head(
|
||
url=url,
|
||
headers=merged_headers,
|
||
cookies=merged_cookies,
|
||
proxies=self._get_proxies(),
|
||
allow_redirects=allow_redirects,
|
||
timeout=timeout,
|
||
verify=self.verify_ssl
|
||
)
|
||
return response
|
||
except Exception as e:
|
||
logger.error(f"发送HEAD请求时发生错误: {url} - {str(e)}")
|
||
|
||
return None
|
||
|
||
def request(self, method, url, **kwargs):
|
||
"""
|
||
发送自定义请求
|
||
|
||
Args:
|
||
method: 请求方法
|
||
url: 请求的URL
|
||
**kwargs: 其他参数
|
||
|
||
Returns:
|
||
requests.Response: 响应对象
|
||
"""
|
||
# 合并请求头和Cookies
|
||
if 'headers' in kwargs:
|
||
kwargs['headers'] = self._merge_headers(kwargs['headers'])
|
||
else:
|
||
kwargs['headers'] = self._merge_headers({})
|
||
|
||
if 'cookies' in kwargs:
|
||
kwargs['cookies'] = self._merge_cookies(kwargs['cookies'])
|
||
else:
|
||
kwargs['cookies'] = self._merge_cookies({})
|
||
|
||
# 设置代理
|
||
kwargs['proxies'] = self._get_proxies()
|
||
|
||
# 设置默认参数
|
||
kwargs.setdefault('timeout', self.timeout)
|
||
kwargs.setdefault('verify', self.verify_ssl)
|
||
|
||
try:
|
||
response = self.session.request(method, url, **kwargs)
|
||
return response
|
||
except Exception as e:
|
||
logger.error(f"发送{method}请求时发生错误: {url} - {str(e)}")
|
||
|
||
return None
|
||
|
||
def download_file(self, url, save_path, chunk_size=8192):
|
||
"""
|
||
下载文件
|
||
|
||
Args:
|
||
url: 文件URL
|
||
save_path: 保存路径
|
||
chunk_size: 块大小
|
||
|
||
Returns:
|
||
bool: 下载是否成功
|
||
"""
|
||
try:
|
||
response = self.get(url, stream=True)
|
||
if not response or response.status_code != 200:
|
||
logger.error(f"下载文件失败, 状态码: {response.status_code if response else 'None'}")
|
||
return False
|
||
|
||
with open(save_path, 'wb') as f:
|
||
for chunk in response.iter_content(chunk_size=chunk_size):
|
||
if chunk:
|
||
f.write(chunk)
|
||
|
||
return True
|
||
except Exception as e:
|
||
logger.error(f"下载文件时发生错误: {url} - {str(e)}")
|
||
return False
|
||
|
||
def submit_form(self, url, form_data, method='POST', headers=None, cookies=None):
|
||
"""
|
||
提交表单
|
||
|
||
Args:
|
||
url: 表单提交URL
|
||
form_data: 表单数据
|
||
method: 提交方法
|
||
headers: 请求头
|
||
cookies: Cookies
|
||
|
||
Returns:
|
||
requests.Response: 响应对象
|
||
"""
|
||
if method.upper() == 'POST':
|
||
return self.post(url, data=form_data, headers=headers, cookies=cookies)
|
||
else:
|
||
return self.get(url, params=form_data, headers=headers, cookies=cookies)
|
||
|
||
def _merge_headers(self, headers=None):
|
||
"""
|
||
合并请求头
|
||
|
||
Args:
|
||
headers: 请求头
|
||
|
||
Returns:
|
||
dict: 合并后的请求头
|
||
"""
|
||
merged = self.headers.copy()
|
||
if headers:
|
||
merged.update(headers)
|
||
return merged
|
||
|
||
def _merge_cookies(self, cookies=None):
|
||
"""
|
||
合并Cookies
|
||
|
||
Args:
|
||
cookies: Cookies
|
||
|
||
Returns:
|
||
dict: 合并后的Cookies
|
||
"""
|
||
merged = self.cookies.copy()
|
||
if cookies:
|
||
merged.update(cookies)
|
||
return merged
|
||
|
||
def _get_proxies(self):
|
||
"""
|
||
获取代理配置
|
||
|
||
Returns:
|
||
dict: 代理配置
|
||
"""
|
||
if not self.proxy:
|
||
return {}
|
||
|
||
return {
|
||
'http': self.proxy,
|
||
'https': self.proxy
|
||
}
|
||
|
||
def delay_request(self, min_delay=1, max_delay=3):
|
||
"""
|
||
延迟请求以避免被目标网站检测为爬虫
|
||
|
||
Args:
|
||
min_delay: 最小延迟时间(秒)
|
||
max_delay: 最大延迟时间(秒)
|
||
"""
|
||
delay = random.uniform(min_delay, max_delay)
|
||
time.sleep(delay) |