290 lines
9.6 KiB
Python
290 lines
9.6 KiB
Python
# cron: 16 6 * * *
|
||
# new Env("ipzan代理白名单管理")
|
||
import requests
|
||
import os
|
||
import sys
|
||
import time
|
||
from datetime import datetime
|
||
import warnings
|
||
from urllib3.exceptions import InsecureRequestWarning
|
||
from Crypto.Cipher import AES
|
||
from Crypto.Util.Padding import pad
|
||
import json
|
||
|
||
# 禁用SSL警告
|
||
warnings.filterwarnings("ignore", category=InsecureRequestWarning)
|
||
|
||
# ================= 配置区 =================
|
||
# 从单个环境变量中读取所有配置
|
||
CONFIG_STR = os.getenv('API_KEY_PZ')
|
||
if not CONFIG_STR:
|
||
print("错误：未找到API_KEY_PZ环境变量")
|
||
sys.exit(1)
|
||
|
||
# 解析环境变量：USER_ID#LOGIN_PASSWORD#KEY#ORDER_NO#EXTRACT_KEY
|
||
try:
|
||
parts = CONFIG_STR.split('#')
|
||
if len(parts) < 5:
|
||
raise ValueError("环境变量格式不正确")
|
||
|
||
USER_ID = parts[0]
|
||
LOGIN_PASSWORD = parts[1]
|
||
API_KEY = parts[2]
|
||
ORDER_NO = parts[3]
|
||
EXTRACT_KEY = parts[4]
|
||
|
||
# 打印配置信息（调试用）
|
||
print(f"USER_ID: {USER_ID}")
|
||
print(f"LOGIN_PASSWORD: {LOGIN_PASSWORD[:2]}****{LOGIN_PASSWORD[-2:]}")
|
||
print(f"API_KEY: {API_KEY[:2]}****{API_KEY[-2:]}")
|
||
print(f"ORDER_NO: {ORDER_NO}")
|
||
print(f"EXTRACT_KEY: {EXTRACT_KEY[:2]}****{EXTRACT_KEY[-2:]}")
|
||
|
||
except Exception as e:
|
||
print(f"解析环境变量失败：{str(e)}")
|
||
print("请确保环境变量格式为：USER_ID#LOGIN_PASSWORD#KEY#ORDER_NO#EXTRACT_KEY")
|
||
sys.exit(1)
|
||
|
||
BASE_URL = "https://service.ipzan.com"
|
||
IP_SERVICES = [
|
||
{'url': 'http://ifconfig.me/ip', 'verify': False},
|
||
{'url': 'http://ip.3322.net', 'verify': False},
|
||
{'url': 'http://myip.ipip.net', 'verify': False}
|
||
]
|
||
MAX_RETRIES = 3
|
||
REQUEST_DELAY = 2
|
||
# ==========================================
|
||
|
||
def debug_log(msg, level="INFO"):
|
||
"""分级日志系统"""
|
||
timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
|
||
color_code = {
|
||
"DEBUG": "\033[94m", # 蓝色
|
||
"INFO": "\033[92m", # 绿色
|
||
"WARNING": "\033[93m", # 黄色
|
||
"ERROR": "\033[91m", # 红色
|
||
"SUCCESS": "\033[92m", # 绿色（成功）
|
||
"SYSTEM": "\033[95m" # 紫色
|
||
}.get(level, "\033[0m")
|
||
print(f"{color_code}## [{timestamp}] [{level}] {msg}\033[0m")
|
||
|
||
def get_public_ip():
|
||
"""获取公网IP"""
|
||
debug_log("正在获取公网IP...", "DEBUG")
|
||
for service in IP_SERVICES:
|
||
try:
|
||
response = requests.get(service['url'], timeout=5, verify=service['verify'])
|
||
if response.status_code == 200:
|
||
ip = response.text.strip()
|
||
if validate_ip(ip):
|
||
debug_log(f"验证通过的公网IP：{ip}", "SUCCESS")
|
||
return ip
|
||
except Exception as e:
|
||
debug_log(f"{service['url']}查询失败：{str(e)}", "DEBUG")
|
||
debug_log("所有IP服务不可用", "ERROR")
|
||
return None
|
||
|
||
def validate_ip(ip):
|
||
"""IP地址格式验证"""
|
||
parts = ip.split('.')
|
||
return len(parts) == 4 and all(part.isdigit() and 0 <= int(part) < 256 for part in parts)
|
||
|
||
def generate_aes_signature():
|
||
"""生成AES加密签名"""
|
||
try:
|
||
# 获取当前时间戳（秒）
|
||
timestamp = str(int(time.time()))
|
||
# 构造签名内容：登录密码:套餐提取密匙:时间戳
|
||
data = f"{LOGIN_PASSWORD}:{EXTRACT_KEY}:{timestamp}"
|
||
|
||
# 将API_KEY转换为16字节密钥
|
||
key = API_KEY.encode('utf-8')
|
||
# 确保密钥长度为16字节（AES-128）
|
||
key = key.ljust(16, b'\0')[:16]
|
||
|
||
# 创建AES加密器（ECB模式）
|
||
cipher = AES.new(key, AES.MODE_ECB)
|
||
|
||
# 对数据进行PKCS7填充
|
||
padded_data = pad(data.encode('utf-8'), AES.block_size)
|
||
|
||
# 加密数据
|
||
encrypted_data = cipher.encrypt(padded_data)
|
||
|
||
# 返回十六进制字符串
|
||
return encrypted_data.hex()
|
||
except Exception as e:
|
||
debug_log(f"AES签名生成失败：{str(e)}", "ERROR")
|
||
return None
|
||
|
||
def add_ip(ip):
|
||
"""添加IP到白名单（使用AES签名）"""
|
||
url = f"{BASE_URL}/whiteList-add"
|
||
|
||
# 生成AES签名
|
||
sign = generate_aes_signature()
|
||
if not sign:
|
||
debug_log("无法生成AES签名", "ERROR")
|
||
return False
|
||
|
||
params = {
|
||
'no': ORDER_NO,
|
||
'sign': sign,
|
||
'ip': ip,
|
||
'replace': 1 # 启用自动替换功能
|
||
}
|
||
|
||
debug_log(f"正在添加IP：{ip}", "INFO")
|
||
try:
|
||
response = requests.get(url, params=params, timeout=15)
|
||
debug_log(f"添加接口原始响应：{response.text}", "DEBUG")
|
||
|
||
# 尝试解析JSON响应
|
||
try:
|
||
data = response.json()
|
||
except ValueError:
|
||
debug_log("响应不是有效的JSON格式", "ERROR")
|
||
return False
|
||
|
||
if data.get('code') == 0:
|
||
debug_log(f"IP {ip} 添加成功", "SUCCESS")
|
||
return True
|
||
|
||
error_msg = data.get('message') or '未知错误'
|
||
debug_log(f"添加失败：{error_msg}（状态码：{data.get('code')}）", "ERROR")
|
||
except Exception as e:
|
||
debug_log(f"请求添加接口失败：{str(e)}", "ERROR")
|
||
return False
|
||
|
||
def verify_ip(ip):
|
||
"""校验IP是否在白名单中"""
|
||
url = f"{BASE_URL}/whiteList-verify"
|
||
params = {
|
||
'no': ORDER_NO,
|
||
'ip': ip
|
||
}
|
||
|
||
try:
|
||
response = requests.get(url, params=params, timeout=10)
|
||
debug_log(f"校验接口原始响应：{response.text}", "DEBUG")
|
||
|
||
# 尝试解析JSON响应
|
||
try:
|
||
data = response.json()
|
||
except ValueError:
|
||
debug_log("响应不是有效的JSON格式", "ERROR")
|
||
return None
|
||
|
||
if data.get('code') == 0:
|
||
# 根据实际接口返回结构调整
|
||
if data.get('data') is True:
|
||
debug_log(f"IP {ip} 已在白名单中", "SUCCESS")
|
||
return True
|
||
else:
|
||
debug_log(f"IP {ip} 不在白名单中", "INFO")
|
||
return False
|
||
|
||
error_msg = data.get('message') or '未知错误'
|
||
debug_log(f"校验失败：{error_msg}（状态码：{data.get('code')}）", "ERROR")
|
||
except Exception as e:
|
||
debug_log(f"请求校验接口失败：{str(e)}", "ERROR")
|
||
return None
|
||
|
||
def get_balance():
|
||
"""查询套餐余额和余量"""
|
||
url = f"{BASE_URL}/userProduct-get"
|
||
params = {
|
||
'no': ORDER_NO,
|
||
'userId': USER_ID # 保持参数名不变
|
||
}
|
||
|
||
try:
|
||
# 打印完整的请求URL以便调试
|
||
full_url = f"{url}?no={ORDER_NO}&userId={USER_ID}"
|
||
debug_log(f"余额查询请求URL: {full_url}", "DEBUG")
|
||
|
||
response = requests.get(url, params=params, timeout=10)
|
||
debug_log(f"余额查询接口原始响应：{response.text}", "DEBUG")
|
||
|
||
# 尝试解析JSON响应
|
||
try:
|
||
data = response.json()
|
||
except ValueError:
|
||
debug_log("响应不是有效的JSON格式", "ERROR")
|
||
return None
|
||
|
||
if data.get('code') == 0:
|
||
balance_info = data.get('data', {})
|
||
debug_log("余额查询成功", "SUCCESS")
|
||
return balance_info
|
||
|
||
error_msg = data.get('message') or '未知错误'
|
||
debug_log(f"余额查询失败：{error_msg}（状态码：{data.get('code')}）", "ERROR")
|
||
except Exception as e:
|
||
debug_log(f"请求余额查询接口失败：{str(e)}", "ERROR")
|
||
return None
|
||
|
||
def main():
|
||
debug_log("===== ipzan代理白名单管理 =====", "SYSTEM")
|
||
|
||
# 第一步：查询套餐余额（可选步骤）
|
||
debug_log("查询套餐余额...", "INFO")
|
||
try:
|
||
balance_info = get_balance()
|
||
if balance_info:
|
||
# 格式化显示余额信息
|
||
balance_str = []
|
||
if 'balance' in balance_info:
|
||
balance_str.append(f"钱包余额: {balance_info['balance']}")
|
||
if 'limitDaySurplus' in balance_info:
|
||
balance_str.append(f"今日余量: {balance_info['limitDaySurplus']}")
|
||
if 'totalSurplus' in balance_info:
|
||
balance_str.append(f"套餐总量: {balance_info['totalSurplus']}")
|
||
|
||
if balance_str:
|
||
debug_log(" | ".join(balance_str), "INFO")
|
||
else:
|
||
debug_log("余额查询失败，跳过余额显示", "WARNING")
|
||
except Exception as e:
|
||
debug_log(f"余额查询过程中出错：{str(e)}", "ERROR")
|
||
|
||
# 第二步：获取当前公网IP
|
||
public_ip = get_public_ip()
|
||
if not public_ip:
|
||
debug_log("无法获取有效公网IP", "FATAL")
|
||
sys.exit(1)
|
||
|
||
# 第三步：检查当前IP是否已在白名单
|
||
debug_log("检查当前IP状态...", "INFO")
|
||
ip_status = verify_ip(public_ip)
|
||
|
||
if ip_status is None:
|
||
debug_log("IP校验失败，请检查网络或服务状态", "FATAL")
|
||
sys.exit(1)
|
||
|
||
if ip_status:
|
||
debug_log("当前IP已在白名单中，无需操作", "SUCCESS")
|
||
sys.exit(0)
|
||
|
||
debug_log("当前IP不在白名单中，开始添加流程...", "WARNING")
|
||
|
||
# 第四步：添加当前公网IP（启用自动替换功能）
|
||
if not add_ip(public_ip):
|
||
debug_log("新IP添加失败", "FATAL")
|
||
sys.exit(1)
|
||
|
||
# 第五步：最终验证
|
||
debug_log("等待接口数据刷新...", "SYSTEM")
|
||
time.sleep(5)
|
||
|
||
final_status = verify_ip(public_ip)
|
||
if final_status is True:
|
||
debug_log(f"最终验证通过：{public_ip} 已生效", "SUCCESS")
|
||
else:
|
||
debug_log(f"最终验证失败：请手动检查白名单状态", "FATAL")
|
||
sys.exit(1)
|
||
|
||
debug_log("===== 操作成功完成 =====", "SYSTEM")
|
||
|
||
if __name__ == '__main__':
|
||
main() |