# 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()