上传文件至「Points_Based」

2026-08-01 03:29:31 +08:00
parent be602d0929
commit e7a04ae104
1 changed files with 290 additions and 0 deletions
@@ -0,0 +1,290 @@
+# 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()