Browse Source

update: detailed exception thrown of ProxyChecker

master
Dnomd343 3 years ago
parent
commit
254d9a545e
  1. 63
      ProxyChecker/Http.py
  2. 128
      check.py
  3. 81
      demo.py

63
ProxyChecker/Http.py

@ -5,28 +5,71 @@ import time
import requests import requests
def httpPing(port, url = 'http://gstatic.com/generate_204', timeout = 30): def httpPing(port, url = 'http://gstatic.com/generate_204', timeout = 30):
'''
HTTP请求延迟测试
检测异常: return None, {reason}
服务错误: return False, {reason}
连接正常: return True, httpDelay
'''
try: try:
startTime = time.time_ns() startTime = time.time_ns() # 请求开始时间
socks5 = 'socks5://127.0.0.1:' + str(port) socks5 = 'socks5://127.0.0.1:' + str(port)
httpRequest = requests.get(url, proxies = { httpRequest = requests.get(url, proxies = { # http请求
'http': socks5, 'http': socks5,
'https': socks5, 'https': socks5,
}, timeout = timeout) }, timeout = timeout)
if httpRequest.status_code == 204: except NameError: # 模块无效
return None, 'Missing modules'
except requests.exceptions.InvalidSchema: # 缺失pysocks包
return None, 'Missing dependencies for SOCKS support'
except requests.exceptions.ConnectionError: # socks端口出错
return None, 'Unable to connect socks5 proxy'
except requests.exceptions.ProxyError: # socks代理出错
return None, 'Unable to connect socks5 proxy'
except requests.exceptions.ConnectTimeout: # 请求超时
return False, 'Request timeout'
except requests.exceptions.ReadTimeout: # 请求超时
return False, 'Request timeout'
except requests.exceptions.Timeout: # 请求超时
return False, 'Request timeout'
except: # 未知错误
return None, 'Unknown error'
try:
statusCode = httpRequest.status_code # 获取http状态码
except:
return None, 'Http request error'
if 'statusCode' in vars() and statusCode == 204: # http测试成功
delay = (time.time_ns() - startTime) / (10 ** 6) delay = (time.time_ns() - startTime) / (10 ** 6)
return round(delay, 2) # 保留小数点后两位 return True, round(delay, 2) # 保留小数点后两位
except: pass else:
return -1 return False, 'Http status code not 204'
def httpCheck(port, url = 'http://gstatic.com/generate_204', timeout = 30): def httpCheck(port, url = 'http://gstatic.com/generate_204', timeout = 30):
'''
HTTP请求测试
测试异常: return None, {reason}
测试完成: return health, delay
'''
result = [] result = []
result.append(httpPing(port, url, timeout / 4))
result.append(httpPing(port, url, timeout / 2))
result.append(httpPing(port, url, timeout / 1))
failNum = 0 failNum = 0
for ret in result: for i in [4, 2, 1]: # 三次测试
status, delay = httpPing(port, url, timeout / i)
if status == None: # 测试异常
return None, delay
elif status == False: # 连接失败
result.append(-1)
else: # 连接成功
result.append(delay)
for ret in result: # 计算失败次数
if ret < 0: if ret < 0:
failNum += 1 failNum += 1
if failNum == 3: # 全部失败 if failNum == 3: # 全部失败
return False, -1 return False, -1
elif failNum == 2: # 仅成功一次 elif failNum == 2: # 仅成功一次

128
check.py

@ -0,0 +1,128 @@
#!/usr/bin/python
# -*- coding:utf-8 -*-
import json
import os
import time
import ProxyBuilder as Builder
import ProxyChecker as Checker
workDir = '/tmp/ProxyC'
ssTest = {
'tag': 'f43c9bae21ae8693',
'check': [
'http'
],
'info': {
'type': 'ss',
'server': '127.0.0.1',
'port': 12345,
'password': 'dnomd343',
'method': 'aes-256-ctr',
'plugin': '',
'pluginParam': '',
}
}
ssrTest = {
'tag': 'f43c9bae21ae8693',
'check': [
'http'
],
'info': {
'type': 'ssr',
"server": "127.0.0.1",
"port": 23456,
"password": "dnomd343",
"method": "table",
"protocol": "auth_aes128_md5",
"protocolParam": "",
"obfs": "tls1.2_ticket_auth",
"obfsParam": ""
}
}
def loadDir(folderPath): # 创建文件夹
try:
if os.path.exists(folderPath): # 文件 / 文件夹 存在
if not os.path.isdir(folderPath): # 文件
return False # 无法创建
else: # 不存在
os.makedirs(folderPath) # 递归创建文件夹
return True # 文件夹正常
except:
return False
def proxyHttpCheck(socksPort): # Http检测
try:
health, httpDelay = Checker.httpCheck(socksPort, timeout = 10)
if health == None: # 连接失败
return None
return {
'delay': httpDelay,
'health': health
}
except: # 未知错误
return None
def proxyTest(rawInfo, startDelay = 1):
'''
代理检测入口
异常错误: return None
'''
if loadDir(workDir) == False: # 工作文件夹无效
return None
if not 'info' in rawInfo: # 缺少代理服务器信息
return None
try:
status, client = Builder.build(rawInfo['info'], workDir)
except: # 构建发生未知错误
Builder.destroy(client)
return None
if status == None: # 构建错误
Builder.destroy(client)
return None
elif status == False: # 节点信息有误
return {
'success': False
}
time.sleep(startDelay) # 延迟等待客户端启动
status = Builder.check(client) # 检查客户端状态
if status == None: # 检测失败
Builder.destroy(client)
return None
elif status == False: # 客户端异常退出
Builder.destroy(client)
return {
'success': False
}
if not 'check' in rawInfo: # 缺少检测项目
return None
checkItem = rawInfo['check']
checkResult = {}
for item in checkItem:
if item == 'http': # http检测
result = proxyHttpCheck(client['port'])
else: # 未知检测项目
result = None
if result == None: # 检测出错
Builder.destroy(client)
return None
checkResult[item] = result
Builder.destroy(client) # 销毁客户端
return {
'success': True,
'result': checkResult
}
ret = proxyTest(ssTest)
print(json.dumps(ret))

81
demo.py

@ -1,81 +0,0 @@
#!/usr/bin/python
# -*- coding:utf-8 -*-
import os
import time
import ProxyBuilder as Builder
import ProxyChecker as Checker
workDir = '/tmp/ProxyC'
ssTest = {
'tag': 'f43c9bae21ae8693',
'check': [
'http'
],
'info': {
'type': 'ss',
'server': '127.0.0.1',
'port': 12345,
'password': 'dnomd343',
'method': 'aes-256-ctr',
'plugin': '',
'pluginParam': '',
}
}
ssrTest = {
'tag': 'f43c9bae21ae8693',
'check': [
'http'
],
'info': {
'type': 'ssr',
"server": "127.0.0.1",
"port": 23456,
"password": "dnomd343",
"method": "table",
"protocol": "auth_aes128_md5",
"protocolParam": "",
"obfs": "tls1.2_ticket_auth",
"obfsParam": ""
}
}
def loadDir(folderPath): # 创建文件夹
try:
if os.path.exists(folderPath): # 文件 / 文件夹 存在
if not os.path.isdir(folderPath): # 文件
return False # 无法创建
else: # 不存在
os.makedirs(folderPath) # 递归创建文件夹
return True # 文件夹正常
except:
return False
def proxyTest(rawInfo, startDelay = 1, destroyDelay = 0.5):
if loadDir(workDir) == False: # 工作文件夹无效
return None
if not 'info' in rawInfo:
return None
status, client = Builder.build(rawInfo['info'], workDir)
if status != True:
print(client)
return None
time.sleep(startDelay)
if Builder.check(client) != True:
print("client error")
return None
health, httpDelay = Checker.httpCheck(client['port'])
print("health = " + str(health))
if httpDelay < 0:
print("http error")
else:
print("delay = " + format(httpDelay, '.2f') + 'ms')
if Builder.destroy(client) != True:
print("client destroy error")
time.sleep(destroyDelay)
print("done")
proxyTest(ssrTest)
Loading…
Cancel
Save