diff --git a/Points_Based/PZdaili_interface.py b/Points_Based/Proxy_iPzan.py similarity index 97% rename from Points_Based/PZdaili_interface.py rename to Points_Based/Proxy_iPzan.py index 7ccd091..f3c961c 100644 --- a/Points_Based/PZdaili_interface.py +++ b/Points_Based/Proxy_iPzan.py @@ -1,290 +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__': +# 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() \ No newline at end of file