163 lines
5.1 KiB
Python
163 lines
5.1 KiB
Python
"""星空代理(xkdaili)白名单适配器。
|
|
|
|
API格式：
|
|
基础URL: http://api2.xkdaili.com/tools/XApi.ashx
|
|
白名单认证参数: apikey + sign(白名单专用) + flag
|
|
获取代理认证参数: apikey + sign(提取专用) — 由用户在代理API地址中填写完整URL
|
|
|
|
白名单接口的 sign 和获取代理的 sign 不同！
|
|
|
|
响应格式:
|
|
status=100: 操作成功
|
|
status=508: 白名单已存在（视为成功）
|
|
status=202: 秘钥验证失败
|
|
status=808: 其它错误
|
|
|
|
注意：没有查询白名单列表的接口
|
|
"""
|
|
|
|
import json
|
|
|
|
import requests
|
|
from loguru import logger
|
|
|
|
from .base import BaseWhitelistAdapter
|
|
|
|
|
|
class XkdailiAdapter(BaseWhitelistAdapter):
|
|
"""星空代理(xkdaili)白名单适配器。"""
|
|
|
|
BASE_URL = "http://api2.xkdaili.com/tools/XApi.ashx"
|
|
|
|
def __init__(self, credentials: dict):
|
|
self.apikey = credentials.get("apikey", "")
|
|
self.wl_sign = credentials.get("wl_sign", "")
|
|
self.flag = credentials.get("flag", "")
|
|
|
|
@property
|
|
def platform_name(self) -> str:
|
|
return "xkdaili"
|
|
|
|
@property
|
|
def platform_label(self) -> str:
|
|
return "星空"
|
|
|
|
def _build_url(self, **params) -> str:
|
|
"""构建白名单请求URL"""
|
|
base_params = {
|
|
"apikey": self.apikey,
|
|
"sign": self.wl_sign,
|
|
"flag": self.flag,
|
|
}
|
|
base_params.update(params)
|
|
from urllib.parse import urlencode
|
|
return f"{self.BASE_URL}?{urlencode(base_params)}"
|
|
|
|
def _parse_response(self, text: str) -> tuple[bool, str]:
|
|
"""解析星空API响应。
|
|
|
|
status=100: 成功
|
|
status=508: 白名单已存在（视为成功）
|
|
status=202: 秘钥验证失败
|
|
status=808: 其它错误
|
|
"""
|
|
try:
|
|
data = json.loads(text)
|
|
status = data.get("status", -1)
|
|
info = data.get("info", "")
|
|
|
|
# 100=成功
|
|
if status == 100:
|
|
return True, info
|
|
|
|
# 508=已存在，视为成功
|
|
if status == 508:
|
|
return True, info
|
|
|
|
# 202=秘钥验证失败
|
|
if status == 202:
|
|
return False, f"认证失败: {info}"
|
|
|
|
# 其他错误
|
|
return False, f"失败(status={status}): {info}"
|
|
|
|
except (json.JSONDecodeError, ValueError):
|
|
# 非JSON响应，按文本判断
|
|
if "成功" in text or "ok" in text.lower():
|
|
return True, text
|
|
if "已存在" in text:
|
|
return True, text
|
|
return False, text
|
|
|
|
def get_whitelist(self) -> list[dict]:
|
|
"""
|
|
星空不支持查询白名单列表，返回空列表。
|
|
|
|
sync_ip 会检测到空列表后直接走添加逻辑。
|
|
"""
|
|
return []
|
|
|
|
def add_ip(self, ip: str) -> tuple[bool, str]:
|
|
"""添加IP到白名单。多个IP用英文逗号隔开。"""
|
|
try:
|
|
url = self._build_url(type="addwhiteip", ip=ip)
|
|
response = requests.get(url, timeout=10)
|
|
response.raise_for_status()
|
|
|
|
text = response.text.strip()
|
|
logger.debug(f"星空添加白名单响应: {text}")
|
|
|
|
ok, msg = self._parse_response(text)
|
|
if ok:
|
|
logger.info(f"星空白名单添加成功: {ip} - {msg}")
|
|
else:
|
|
logger.warning(f"星空白名单添加失败: {msg}")
|
|
return ok, msg
|
|
|
|
except Exception as e:
|
|
logger.error(f"星空添加白名单失败: {e}")
|
|
return False, str(e)
|
|
|
|
def delete_ip(self, ip: str) -> tuple[bool, str]:
|
|
"""删除白名单IP。多个IP用英文逗号隔开。"""
|
|
try:
|
|
url = self._build_url(type="delwhiteip", ip=ip)
|
|
response = requests.get(url, timeout=10)
|
|
response.raise_for_status()
|
|
|
|
text = response.text.strip()
|
|
logger.debug(f"星空删除白名单响应: {text}")
|
|
|
|
ok, msg = self._parse_response(text)
|
|
if ok:
|
|
logger.info(f"星空白名单删除成功: {ip} - {msg}")
|
|
else:
|
|
logger.warning(f"星空白名单删除失败: {msg}")
|
|
return ok, msg
|
|
|
|
except Exception as e:
|
|
logger.error(f"星空删除白名单失败: {e}")
|
|
return False, str(e)
|
|
|
|
def test_connection(self) -> tuple[bool, str]:
|
|
"""测试白名单API连接。"""
|
|
try:
|
|
# 尝试添加无效IP来验证认证
|
|
url = self._build_url(type="addwhiteip", ip="0.0.0.0")
|
|
response = requests.get(url, timeout=10)
|
|
response.raise_for_status()
|
|
|
|
text = response.text.strip()
|
|
ok, msg = self._parse_response(text)
|
|
|
|
if ok:
|
|
return True, "连接成功，API认证正常"
|
|
# 认证失败
|
|
if "认证失败" in msg:
|
|
return False, msg
|
|
# 其他错误（如无效IP），说明认证通过了
|
|
return True, f"连接成功，API可访问: {msg}"
|
|
|
|
except Exception as e:
|
|
return False, f"连接失败: {e}"
|