149 lines
5.0 KiB
Python
149 lines
5.0 KiB
Python
# -*- coding: utf-8 -*-
|
||
import json
|
||
import hmac
|
||
import time
|
||
import random
|
||
import hashlib
|
||
import requests
|
||
from urllib.parse import urlparse
|
||
|
||
# 示例参数
|
||
secret_key = "3yc8c8bg1dym0zaiwjh867al"
|
||
end_point = "https://openapi-hk.armcloud.net"
|
||
access_key_id = "gz8f1u0t63byzdu6ozbx8r5qs3e5lipt"
|
||
default_bypass_domain = ['rr1---sn-i3b7knsl.gvt1.com']
|
||
|
||
|
||
def calculate_signature(timestamp, path, body):
|
||
string_to_sign = timestamp + path + (body if body else "")
|
||
signature = hmac.new(
|
||
secret_key.encode('utf-8'),
|
||
string_to_sign.encode('utf-8'),
|
||
hashlib.sha256
|
||
).hexdigest()
|
||
return signature
|
||
|
||
|
||
def sig_request(req: requests.Request):
|
||
re_qu = urlparse(req.url)
|
||
query_string = ""
|
||
if req.method == "GET":
|
||
query_string = re_qu.query
|
||
elif req.method == "POST":
|
||
query_string = req.data
|
||
# 生成时间戳
|
||
timestamp = str(int(time.time() * 1000))
|
||
signature = calculate_signature(timestamp=timestamp, path=re_qu.path, body=query_string)
|
||
req.headers["authver"] = "2.0"
|
||
req.headers["x-ak"] = access_key_id
|
||
req.headers["x-timestamp"] = timestamp
|
||
req.headers["x-sign"] = signature
|
||
|
||
|
||
def api(api_endpoint, payload):
|
||
try:
|
||
headers = {"Content-Type": "application/json"}
|
||
payloads = json.dumps(payload)
|
||
|
||
# 3. 发送请求
|
||
r = requests.Request('POST', api_endpoint,
|
||
headers=headers, data=payloads)
|
||
sig_request(r)
|
||
session = requests.Session()
|
||
response = session.send(r.prepare())
|
||
response.raise_for_status()
|
||
# 4. 解析响应
|
||
result = response.json()
|
||
# print("API %s %s %s", api_endpoint, payloads, result)
|
||
if result.get("code") == 200:
|
||
if isinstance(result['data'], dict):
|
||
return [result['data']]
|
||
elif isinstance(result['data'], list):
|
||
return result['data']
|
||
# 5. 业务状态码检查
|
||
error_map = {
|
||
110003: "ADB命令执行失败，请联系管理员",
|
||
110012: "命令执行超时，请稍后重试"
|
||
}
|
||
error_code = result.get("code")
|
||
err_msg = " %s %s %s" % (error_map.get(
|
||
error_code, f"未知错误: {result.get('msg')}"), error_code, result.get("ts"))
|
||
except requests.exceptions.RequestException as e:
|
||
err_msg = f"网络请求异常: {str(e)} 500"
|
||
except json.JSONDecodeError:
|
||
err_msg = "响应解析失败 502"
|
||
except KeyError as e:
|
||
err_msg = f"响应字段缺失: {str(e)} 503"
|
||
print(err_msg)
|
||
return []
|
||
|
||
|
||
def set_network_proxy(pad_codes, account, password, ip, port, proxy_type, proxy_name,
|
||
bypass_package_list=None,
|
||
bypass_ip_list=None,
|
||
bypass_domain_list=None,
|
||
enable=True,
|
||
s_uot=False):
|
||
"""
|
||
设置网络代理配置
|
||
|
||
参数说明：
|
||
pad_codes (list): 实例列表 (必填)
|
||
enable (bool): 启用代理 (必填)
|
||
account (str): 账号
|
||
password (str): 密码
|
||
ip (str): 代理IP
|
||
port (int): 代理端口
|
||
proxy_type (str): 代理类型 (proxy/vpn)
|
||
proxy_name (str): 代理协议 (socks5/http-relay)
|
||
bypass_package_list (list): 不走代理的包名列表
|
||
bypass_ip_list (list): 不走代理的IP列表
|
||
bypass_domain_list (list): 不走代理的域名列表
|
||
s_uot (bool): 是否开启UDP连接
|
||
|
||
返回: API响应数据或错误信息
|
||
"""
|
||
# 构造请求URL (替换为实际域名)
|
||
url = end_point + "/openapi/open/network/proxy/set"
|
||
# 构造请求体
|
||
if isinstance(pad_codes, str):
|
||
pad_codes = [pad_codes]
|
||
payload = {
|
||
"padCodes": pad_codes,
|
||
"enable": enable,
|
||
"sUoT": s_uot
|
||
}
|
||
# 添加可选参数
|
||
if account:
|
||
payload["account"] = account
|
||
if password:
|
||
payload["password"] = password
|
||
if ip:
|
||
payload["ip"] = ip
|
||
if port:
|
||
payload["port"] = port
|
||
if proxy_type:
|
||
payload["proxyType"] = proxy_type
|
||
if proxy_name:
|
||
payload["proxyName"] = proxy_name
|
||
if bypass_package_list:
|
||
payload["bypassPackageList"] = bypass_package_list
|
||
if bypass_ip_list:
|
||
payload["bypassIpList"] = bypass_ip_list
|
||
if bypass_domain_list:
|
||
bypass_domain_list.extend(default_bypass_domain)
|
||
payload["bypassDomainList"] = bypass_domain_list
|
||
return api(url, payload)
|
||
|
||
|
||
# tasks = set_network_proxy(pad_codes=["ACP250915XAGEH03"],
|
||
# account=f'uniproxy-zone-custom-region-US-session-{random.randint(1000000, 1000000000)}-sessTime-20',
|
||
# password='uniproxy',
|
||
# ip='adv1.uipoxy.com', port=3000,
|
||
# proxy_name='socks5',
|
||
# proxy_type='proxy')
|
||
# print(tasks)
|
||
# set_cmd = f"/data/wyproxy/run.sh -proxy socks5://6255452-2ff229fb:uniproxy@adv1.uipoxy.com:3000 -dns 8.8.8.8 -noudp"
|
||
# # curl -x socks5://uniproxy-zone-custom:uniproxy@adv1.ipmoyu.com:3000 ipinfo.io
|
||
# # curl -x socks5://uniproxy-zone-custom-region:uniproxy@adv1.ipmoyu.com:3000 ipinfo.io
|