-
Notifications
You must be signed in to change notification settings - Fork 2
/
Copy pathmain.py
324 lines (299 loc) · 12.1 KB
/
main.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
import urllib3
import requests
import download
import asyncio
from launch import *
# 定义主类
class Main:
url = "https://www.locyanfrp.cn/api/"
proxy_temp = "" # 用户所持有隧道ID暂存
proxy_list_id_type = {}
r = None
def __init__(self):
# 关闭keep-alive
self.r = requests.session()
self.r.keep_alive = False
# 设置重连次数为5
self.r.DEFAULT_RETRIES = 5
# 关闭urllib3不验证SSL产生的报错
urllib3.disable_warnings()
async def async_init(self):
# 检查更新
version = "version_1.0.1"
check_update_url = (
f"https://api.ymbit.cn/public/check_update/{version}.html"
)
check_update = requests.get(check_update_url, verify=False)
if check_update.status_code == 200:
await download.check_frp_exists()
elif check_update.status_code == 404:
print("目前 LoCyanPyLauncher 非最新版本，请前往 GitHub 更新。")
else:
print("检查更新失败，请检查网络连接。")
sys.exit(1)
print("欢迎使用LoCyan Frp Application")
result = self.r.get("https://api.locyanfrp.cn/App", verify=False)
result = result.json()
contents = result["contents"]
print("------------------------------------------------")
print("公告：")
print(contents)
print("------------------------------------------------")
def get_user_info(
self, node: str, code: str
): # 使用FrpPanel自带的API进行鉴权，判断用户是否有权限使用
result = self.r.get(
self.url
+ "?apitoken=LoCyanToken|"
+ node
+ "&user="
+ code
+ "&action=checktoken",
verify=False,
)
return result.text
def get_server_list(self): # 获取所有服务器列表
url = "https://api.locyanfrp.cn/api/v2/proxies/getlist"
rs = self.r.get(url, verify=False).json()
for i in rs:
server_id = i["id"]
server_name = i["name"]
server_ip = i["hostname"]
status = i["status"]
if status == "200":
print("ID:", server_id, "|", server_name, "[", server_ip, "]")
def get_name_by_code(self, code): # 通过token获取用户名
url = "https://api.locyanfrp.cn/Account/GetUserNameByFrpToken?token=" + code
rs = self.r.get(url, verify=False).json()
username = rs["username"]
return username
def getproxy(self, node, user): # 通过用户名获取用户隧道
url = (
"https://api.locyanfrp.cn/Proxies/GetProxiesListByNode?node="
+ node
+ "&username="
+ user
)
rs = self.r.get(url, verify=False).json()
# 检查是否有隧道存在
if rs["count"] == 0:
return None
else:
# rs["proxies"] 是个列表
for i in rs["proxies"]:
# 鬼知道为什么本地端口是整数，然后远程是字符串
print(
"ID:",
i["id"],
"|",
i["proxy_name"],
"[",
i["proxy_type"],
"]",
str(i["local_port"]),
"->",
i["remote_port"],
)
if len(self.proxy_temp) == 0:
self.proxy_list_id_type[str(i["id"])] = i["proxy_type"]
self.proxy_temp = str(i["id"])
else:
self.proxy_list_id_type[str(i["id"])] = i["proxy_type"]
self.proxy_temp = self.proxy_temp + " | " + str(i["id"])
def is_proxy_in_user(self): # 检查该隧道是否为该用户所有
if self.proxy_temp.find(proxyid) >= 0:
return 0
else:
return -1
def get_server_info(self, id):
url = "https://api.locyanfrp.cn/Proxies/GetServerInfoByNode?node=" + id
rs = self.r.get(url, verify=False).json()
server_id = rs["id"]
server_name = rs["name"]
server_ip = rs["hostname"]
server_port = rs["port"]
status = rs["status"]
return str(server_ip) + "|" + str(server_port)
def get_proxy_info(self, user, id1):
url = (
"https://api.locyanfrp.cn/Proxies/GetProxiesListByID?username="
+ user
+ "&id="
+ id1
)
rs = self.r.get(url, verify=False).json()
# type1, lip, lp, rp, proxyname, cd
# rs["proxies"]是列表
for type in rs["proxies"]:
# type[3]:type[4]:lip type[5]:lp type[11] rp type[8]:自定义域名(cd)
if (
self.proxy_list_id_type[id1] == "http"
or self.proxy_list_id_type[id1] == "https"
):
return (
str(type["proxy_type"])
+ "|"
+ str(type["local_ip"])
+ "|"
+ str(type["local_port"])
+ "|"
+ str(type["remote_port"])
+ "|"
+ str(type["proxy_name"])
+ "|"
+ type["domain"]
)
else:
return (
str(type["proxy_type"])
+ "|"
+ str(type["local_ip"])
+ "|"
+ str(type["local_port"])
+ "|"
+ str(type["remote_port"])
+ "|"
+ str(type["proxy_name"])
)
main = Main()
main.get_server_list()
token_user = ""
if os.path.exists("cfg.ini"):
f = open("cfg.ini", "r", encoding="utf-8")
token_user = f.read()
print(
"已读取到您上次使用的token: ",
token_user,
"如果想要更换的话请在选择服务器的时候输入ct",
)
f.close()
while True:
choose = input("请选择需要映射的服务器：")
if choose == "ct":
ct = input("请输入您要更换的token: ")
f = open("cfg.ini", "w", encoding="utf-8")
f.write(ct)
f.close()
print("更换成功！")
choose = input("请选择需要映射的服务器：")
token_user = ct
if token_user == "":
token_user = input("请输入您的连接Token: ")
f = open("cfg.ini", "w", encoding="utf-8")
f.write(token_user)
f.close()
print("开始进行鉴权，请稍等...")
a = main.get_user_info(choose, token_user)
if "success" in a:
token_user_name = main.get_name_by_code(token_user)
print("鉴权成功，欢迎您，指挥官：" + token_user_name)
print("支持多开, 请用空格分隔每一个隧道ID, 目前仅支持同一节点的多开")
while True: # 反复弹出，直到输入正确
main.getproxy(node=choose, user=token_user_name)
proxyid = input(">> ")
print("选择ID: " + proxyid)
print("准备启动...")
ip, port = main.get_server_info(choose).split("|")
if " " in proxyid:
print("=========================================")
print("你可以用这个地址连接你的隧道！")
print("检测到多个隧道多开，以下为链接方式列表：")
proxyid = proxyid.split(" ")
# 会清除frpc.ini内的全部内容，需要第一个执行
add_proxy_header(ip=ip, port=port, user=token_user)
for i in proxyid:
# 判断是否是HTTPS/HTTP映射，以便判断是否需要获取domain域名
try:
if (
main.proxy_list_id_type[i] == "http"
or main.proxy_list_id_type[i] == "https"
):
type1, lip, lp, rp, proxyname, cd = main.get_proxy_info(
token_user_name, i
).split("|")
else:
type1, lip, lp, rp, proxyname = main.get_proxy_info(
token_user_name, i
).split("|")
except KeyError as e:
print("隧道不存在!")
continue
if type1 == "http" or type1 == "https":
print("ID: " + i + "连接地址：" + cd)
more_proxy_http(
ip=ip,
port=port,
type1=type1,
lip=lip,
lp=lp,
rp=rp,
proxyname=proxyname,
cd=cd,
)
else:
print("ID: " + i + "连接地址：" + ip + ":" + rp)
more_proxy_nohttp(
ip=ip,
port=port,
type1=type1,
lip=lip,
lp=lp,
rp=rp,
proxyname=proxyname,
)
print("=========================================")
launch_frp()
else:
print("=========================================")
print("你可以用这个地址连接你的隧道！")
# 判断是否是HTTPS/HTTP映射，以便判断是否需要获取domain域名
try:
if (
main.proxy_list_id_type[proxyid] == "http"
or main.proxy_list_id_type[proxyid] == "https"
):
type1, lip, lp, rp, proxyname, cd = main.get_proxy_info(
token_user_name, proxyid
).split("|")
else:
type1, lip, lp, rp, proxyname = main.get_proxy_info(
token_user_name, proxyid
).split("|")
except KeyError as e:
print("隧道不存在!")
continue
if type1 == "http" or type1 == "https":
print("ID: " + proxyid + "连接地址：" + cd)
print("=========================================")
get_launch_conf_http(
ip=ip,
port=port,
user=token_user,
type1=type1,
lip=lip,
lp=lp,
rp=rp,
proxyname=proxyname,
cd=cd,
)
else:
print("ID: " + proxyid + "连接地址：" + ip + ":" + rp)
print("=========================================")
get_launch_conf_no_http(
ip=ip,
port=port,
user=token_user,
type1=type1,
lip=lip,
lp=lp,
rp=rp,
proxyname=proxyname,
)
launch_frp()
break
break
else:
print(
"鉴权失败, 请检查您的信息以及你是否有权限使用该服务器, 您可以尝试输入ct更换token秘钥"
)
continue