ITStudio_OUC/xha_auto_login_wifi
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refact: class Loginer

Browse Source
This commit is contained in:
lit
2025-06-24 18:53:48 +08:00
parent b53576a273
commit 6a82d46a6b
1 changed files with 130 additions and 123 deletions
Download Patch File Download Diff File Expand all files Collapse all files
get_xha_user_pwd.py
+130 -123
View File
@@ -1,4 +1,5 @@
from enum import IntEnum, auto from enum import IntEnum, auto
from urllib import request from urllib import request
import json import json
@@ -10,32 +11,6 @@ def request_get_text(url, headers={}):
with request.urlopen(req) as response: with request.urlopen(req) as response:
return response.read().decode('utf-8') return response.read().decode('utf-8')
def check_connectivity(interface):
interface_def = interface is None
if not interface_def:
command = ['ping']
if not interface_def:
command += ["-I", interface]
host = "223.5.5.5"
command += ['-c', '1', host] # -n 1 on windows, -c 1 on linux
try:
result = subprocess.check_call(command, timeout=2)
return result == 0
except subprocess.TimeoutExpired:
print(f"接口 {interface}：Ping to {host} timed out.")
return False
except subprocess.CalledProcessError:
return False
except Exception as e:
print(f"接口 {interface}：Error during ping: {e}")
return False
else:
try:
request_get_text("http://connect.rom.miui.com/generate_204")
return True
except:
return False
class LoginStatus(IntEnum): class LoginStatus(IntEnum):
unknown = -3 unknown = -3
not_unlimit = -2 not_unlimit = -2
@@ -43,116 +18,148 @@ class LoginStatus(IntEnum):
succ = auto() succ = auto()
used_online = auto() used_online = auto()
def login(user, pwd, interface, log, device): class Loginer:
interface_def = interface is None def __init__(self,
url = f'http://192.168.101.201:801/eportal/portal/page/loadUserInfo?callback=dr1004&lang=zh-CN&program_index=ctshNw1713845951&page_index=V5fmKw1713845966&user_account={user}&wlan_user_ip=0.0.0.0&wlan_user_mac=000000000000&jsVersion=4.1&v=599&lang=zh' interface = None,
t = request_get_text(url, headers={ device = 0,
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/105.0.0.0 Safari/537.36 Edg/105.0.1343.33', log = print,
'Referer': 'http://192.168.101.201/' ):
}) self.interface = interface
text = t.replace('dr1004(', '').replace(')', '').replace(';', '') self.device = device
j = json.loads(text) self.log = log
if j['user_info']['user_state'] == "正常" and j['user_info']['available_flow'] in ("0MB", "无限制"): self.interface_def = interface is None
url = f"http://192.168.101.201:801/eportal/portal/page/loadOnlineRecord?callback=dr1006&lang=zh-CN&program_index=ctshNw1713845951&page_index=V5fmKw1713845966&user_account={user}&wlan_user_ip=10.169.0.241&wlan_user_mac=000000000000&start_time=2010-01-01&end_time=2100-01-01&start_rn=1&end_rn=5&jsVersion=4.1&v=2399&lang=zh" def check_connectivity(self):
# 获取在线设备 interface = self.interface
if not self.interface_def:
command = ['ping']
if not self.interface_def:
command += ["-I", interface]
host = "223.5.5.5"
command += ['-c', '1', host] # -n 1 on windows, -c 1 on linux
try:
result = subprocess.check_call(command, timeout=2)
return result == 0
except subprocess.TimeoutExpired:
self.log(f"接口 {interface}：Ping to {host} timed out.")
return False
except subprocess.CalledProcessError:
return False
except Exception as e:
self.log(f"接口 {interface}：Error during ping: {e}")
return False
else:
try:
request_get_text("http://connect.rom.miui.com/generate_204")
return True
except:
return False
def login(self, user, pwd):
interface = self.interface
interface_def = self.interface_def
url = f'http://192.168.101.201:801/eportal/portal/page/loadUserInfo?callback=dr1004&lang=zh-CN&program_index=ctshNw1713845951&page_index=V5fmKw1713845966&user_account={user}&wlan_user_ip=0.0.0.0&wlan_user_mac=000000000000&jsVersion=4.1&v=599&lang=zh'
t = request_get_text(url, headers={ t = request_get_text(url, headers={
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/105.0.0.0 Safari/537.36 Edg/105.0.1343.33', 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/105.0.0.0 Safari/537.36 Edg/105.0.1343.33',
'Referer': 'http://192.168.101.201/' 'Referer': 'http://192.168.101.201/'
}) })
text = t.replace('dr1006(', '').replace(')', '').replace(';', '') text = t.replace('dr1004(', '').replace(')', '').replace(';', '')
j1 = json.loads(text) j = json.loads(text)
if int(j1['count']) == device: # 检查是否付费
# 判断密码是否正确 if j['user_info']['user_state'] == "正常" and j['user_info']['available_flow'] in ("0MB", "无限制"):
url = f"https://xha.ouc.edu.cn:802/eportal/portal/login?callback=dr1003&login_method=1&user_account={user}&user_password={pwd}&wlan_user_ip=0.0.0.0&wlan_user_ipv6=&wlan_user_mac=000000000000&wlan_ac_ip=&wlan_ac_name=&jsVersion=4.1&terminal_type=1&lang=zh-cn&v=2425&lang=zh" url = f"http://192.168.101.201:801/eportal/portal/page/loadOnlineRecord?callback=dr1006&lang=zh-CN&program_index=ctshNw1713845951&page_index=V5fmKw1713845966&user_account={user}&wlan_user_ip=10.169.0.241&wlan_user_mac=000000000000&start_time=2010-01-01&end_time=2100-01-01&start_rn=1&end_rn=5&jsVersion=4.1&v=2399&lang=zh"
if interface_def: # 获取在线设备
res = request_get_text(url) t = request_get_text(url, headers={
else: 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/105.0.0.0 Safari/537.36 Edg/105.0.1343.33',
command = ["curl", url, "--interface", interface] 'Referer': 'http://192.168.101.201/'
res = subprocess.check_output(command, text=True) })
result = json.loads(res[7:-2]) text = t.replace('dr1006(', '').replace(')', '').replace(';', '')
""" j1 = json.loads(text)
dr1003({"result":0,"msg":"账号不存在","ret_code":1}); if int(j1['count']) == self.device:
dr1003({"result":0,"msg":"密码错误","ret_code":1}); # 判断密码是否正确
dr1003({"result":1,"msg":"Portal协议认证成功！"}); url = f"https://xha.ouc.edu.cn:802/eportal/portal/login?callback=dr1003&login_method=1&user_account={user}&user_password={pwd}&wlan_user_ip=0.0.0.0&wlan_user_ipv6=&wlan_user_mac=000000000000&wlan_ac_ip=&wlan_ac_name=&jsVersion=4.1&terminal_type=1&lang=zh-cn&v=2425&lang=zh"
dr1003({"result":0,"msg":"IP: 10.142.5.160 已经在线！","ret_code":2}); if interface_def:
""" res = request_get_text(url)
msg = "" else:
if not interface_def: command = ["curl", url, "--interface", interface]
log(f"使用接口 {interface} 进行请求") res = subprocess.check_output(command, text=True)
msg += f"接口 {interface}： " result = json.loads(res[7:-2])
if '密码错误' in result['msg']: """
log(f"{user} {pwd} 密码错误") dr1003({"result":0,"msg":"账号不存在","ret_code":1});
return LoginStatus.bad_pwd dr1003({"result":0,"msg":"密码错误","ret_code":1});
elif '已经在线' in result['msg']: dr1003({"result":1,"msg":"Portal协议认证成功！"});
msg += "正常在线！" dr1003({"result":0,"msg":"IP: 10.142.5.160 已经在线！","ret_code":2});
return LoginStatus.used_online """
elif '认证成功' in result['msg']: msg = ""
msg += f"使用账号{user}登录成功!" if not interface_def:
return LoginStatus.succ self.log(f"使用接口 {interface} 进行请求")
log(msg) msg += f"接口 {interface}： "
return LoginStatus.unknown if '密码错误' in result['msg']:
else: self.log(f"{user} {pwd} 密码错误")
return LoginStatus.not_unlimit return LoginStatus.bad_pwd
elif '已经在线' in result['msg']:
msg += "正常在线！"
return LoginStatus.used_online
elif '认证成功' in result['msg']:
msg += f"使用账号{user}登录成功!"
return LoginStatus.succ
self.log(msg)
return LoginStatus.unknown
else:
return LoginStatus.not_unlimit
def login_till_succ(user_pwd_gen, bad_user_callback=lambda u: None, interface=None, log=print): def login_till_succ(self, user_pwd_gen, bad_user_callback=lambda u: None):
''' '''
user_pwd_gen will be called multiply times until login succeeds. user_pwd_gen will be called multiply times until login succeeds.
''' '''
device = 0 # Final[int] not_succ = True
# 检查是否付费 while not_succ:
not_succ = True (user, pwd) = user_pwd_gen()
while not_succ: ret = self.login(user, pwd)
(user, pwd) = user_pwd_gen() not_succ = ret < 0
ret = login(user, pwd, interface, log, device)
not_succ = ret < 0
if ret in {LoginStatus.bad_pwd, LoginStatus.not_unlimit}: if ret in {LoginStatus.bad_pwd, LoginStatus.not_unlimit}:
bad_user_callback(user) bad_user_callback(user)
def main(args, warn, log): def main(self, file, warn):
interface = args.interface interface_def = self.interface_def
file = args.file if interface_def:
warn("no interface given by -i or --interface, use default route")
interface_def = interface is None # 检测网络联通性
if interface_def: if self.check_connectivity():
warn("no interface given by -i or --interface, use default route") exit()
# 检测网络联通性 with open(file, 'r') as f:
if check_connectivity(interface): user_pwd = [i.rstrip('\n\r').split() for i in f.readlines()]
exit()
with open(file, 'r') as f: # login
user_pwd = [i.rstrip('\n\r').split() for i in f.readlines()] user_pwd_error_or_not_unlimit_combo_idx = set()
cur_lku_user_idx = 0
def user_pwd_getter():
# 随机选择，防止前面有密码错误的用户卡死
user_index = random.randrange(len(user_pwd))
if user_index in user_pwd_error_or_not_unlimit_combo_idx:
return user_pwd_getter()
(user, pwd) = user_pwd[user_index]
return (user, pwd)
# login def bad_user_callback(_):
user_pwd_error_or_not_unlimit_combo_idx = set() user_pwd_error_or_not_unlimit_combo_idx.add(cur_lku_user_idx)
cur_lku_user_idx = 0
def user_pwd_getter():
# 随机选择，防止前面有密码错误的用户卡死
user_index = random.randrange(len(user_pwd))
if user_index in user_pwd_error_or_not_unlimit_combo_idx:
return user_pwd_getter()
(user, pwd) = user_pwd[user_index]
return (user, pwd)
def bad_user_callback(_): self.login_till_succ(user_pwd_getter, bad_user_callback)
user_pwd_error_or_not_unlimit_combo_idx.add(cur_lku_user_idx)
login_till_succ(user_pwd_getter, bad_user_callback, interface=interface, log=log) # 删除密码错误或不满足要求的用户
with open(file, 'w') as f:
for i in range(len(user_pwd)):
# 删除密码错误或不满足要求的用户 if i in user_pwd_error_or_not_unlimit_combo_idx:
with open(file, 'w') as f: user_pwd_error_or_not_unlimit_combo_idx.remove(i)
for i in range(len(user_pwd)): else:
if i in user_pwd_error_or_not_unlimit_combo_idx: t = user_pwd[i]
user_pwd_error_or_not_unlimit_combo_idx.remove(i) f.write(t[0])
else: f.write(' ')
t = user_pwd[i] f.write(t[1])
f.write(t[0]) f.write('\n')
f.write(' ')
f.write(t[1])
f.write('\n')
if __name__ == "__main__": if __name__ == "__main__":
import argparse import argparse
@@ -163,4 +170,4 @@ if __name__ == "__main__":
parser.add_argument("-i", "--interface", help="网卡接口名称 interface") parser.add_argument("-i", "--interface", help="网卡接口名称 interface")
parser.add_argument("-f", "--file", help="用户密码文件路径", default="user_pwd.txt") parser.add_argument("-f", "--file", help="用户密码文件路径", default="user_pwd.txt")
args = parser.parse_args() args = parser.parse_args()
main(args, warnings.warn, log) Loginer(args.interface, log=log).main(args.file, warnings.warn)