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

chore(shebang): python3; style(newline): crlf -> lf

Browse Source
This commit is contained in:
lit
2025-06-25 09:39:12 +08:00
parent 55fc6b90bd
commit dc6e3232be
1 changed files with 212 additions and 212 deletions
Download Patch File Download Diff File Expand all files Collapse all files
xha_auto_login.py
+212 -212
View File
@@ -1,212 +1,212 @@
#!/usr/bin/env python3
from sys import maxsize from sys import maxsize
from enum import IntEnum, auto from enum import IntEnum, auto
from urllib.error import URLError from urllib.error import URLError
from urllib import request from urllib import request
from errno import ENETUNREACH from errno import ENETUNREACH
import json import json
import random import random
import subprocess import subprocess
def request_get_text(url, headers={}): def request_get_text(url, headers={}):
req = request.Request(url, headers=headers) req = request.Request(url, headers=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")
class LoginStatus(IntEnum): class LoginStatus(IntEnum):
unknown = -maxsize-1 unknown = -maxsize-1
no_wifi = -3 no_wifi = -3
not_unlimit = -2 not_unlimit = -2
bad_pwd = -1 bad_pwd = -1
succ = auto() succ = auto()
used_online = auto() used_online = auto()
class Loginer: class Loginer:
def __init__(self, def __init__(self,
interface = None, interface = None,
online_device_limit = 2, online_device_limit = 2,
log = print, log = print,
): ):
self.interface = interface self.interface = interface
self.online_device_limit = online_device_limit self.online_device_limit = online_device_limit
self.log = log self.log = log
self.interface_def = interface is None self.interface_def = interface is None
def check_connectivity(self): def check_connectivity(self):
interface = self.interface interface = self.interface
if not self.interface_def: if not self.interface_def:
command = ["ping", "-I", interface] command = ["ping", "-I", interface]
host = "223.5.5.5" host = "223.5.5.5"
command += ["-c", "1", host] # -n 1 on windows, -c 1 on linux command += ["-c", "1", host] # -n 1 on windows, -c 1 on linux
try: try:
result = subprocess.check_call(command, timeout=2) result = subprocess.check_call(command, timeout=2)
return result == 0 return result == 0
except subprocess.TimeoutExpired: except subprocess.TimeoutExpired:
self.log(f"接口 {interface}：Ping to {host} timed out.") self.log(f"接口 {interface}：Ping to {host} timed out.")
return False return False
except subprocess.CalledProcessError: except subprocess.CalledProcessError:
return False return False
except Exception as e: except Exception as e:
self.log(f"接口 {interface}：Error during ping: {e}") self.log(f"接口 {interface}：Error during ping: {e}")
return False return False
else: else:
try: try:
request_get_text("http://connect.rom.miui.com/generate_204") request_get_text("http://connect.rom.miui.com/generate_204")
return True return True
except: except:
return False return False
referrer = "http://192.168.101.201" referrer = "http://192.168.101.201"
header = { header = {
"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": referrer "Referer": referrer
} }
@staticmethod @staticmethod
def url(last, dr, v): def url(last, dr, v):
return "/eportal/portal/" + last + "?callback=dr" + dr + "&v=" + v + \ return "/eportal/portal/" + last + "?callback=dr" + dr + "&v=" + v + \
"&lang=zh-CN&wlan_user_mac=000000000000&jsVersion=4.1" "&lang=zh-CN&wlan_user_mac=000000000000&jsVersion=4.1"
ip0 = "&wlan_user_ip=0.0.0.0" ip0 = "&wlan_user_ip=0.0.0.0"
index_qs = "&program_index=ctshNw1713845951&page_index=V5fmKw1713845966" index_qs = "&program_index=ctshNw1713845951&page_index=V5fmKw1713845966"
urlloadUserInfo = referrer + ":801" + url("page/loadUserInfo", "1004", "599") + ip0 + index_qs urlloadUserInfo = referrer + ":801" + url("page/loadUserInfo", "1004", "599") + ip0 + index_qs
time_qs = "&start_time=2010-01-01&end_time=2100-01-01&start_rn=1&end_rn=5" time_qs = "&start_time=2010-01-01&end_time=2100-01-01&start_rn=1&end_rn=5"
urlloadOnlineRecord = referrer + ":801" + url("page/loadOnlineRecord", "1006", "2399") + time_qs + index_qs + "&wlan_user_ip=10.169.0.241" urlloadOnlineRecord = referrer + ":801" + url("page/loadOnlineRecord", "1006", "2399") + time_qs + index_qs + "&wlan_user_ip=10.169.0.241"
urllogin = "https://xha.ouc.edu.cn:802" + url("login", "1003", "2425") + ip0 + "&login_method=1&wlan_user_ipv6=&wlan_ac_ip=&wlan_ac_name=&terminal_type=1" urllogin = "https://xha.ouc.edu.cn:802" + url("login", "1003", "2425") + ip0 + "&login_method=1&wlan_user_ipv6=&wlan_ac_ip=&wlan_ac_name=&terminal_type=1"
@staticmethod @staticmethod
def json_from_drnnnn(t): def json_from_drnnnn(t):
"drNNNN(`result`); <- e.g. t.replace('dr1006(', '').replace(')', '').replace(';', '')" "drNNNN(`result`); <- e.g. t.replace('dr1006(', '').replace(')', '').replace(';', '')"
return json.loads(t[7:-2]) return json.loads(t[7:-2])
def login(self, user, pwd): def login(self, user, pwd):
interface = self.interface interface = self.interface
interface_def = self.interface_def interface_def = self.interface_def
u = "&user_account=" + user u = "&user_account=" + user
url = self.urlloadUserInfo + u url = self.urlloadUserInfo + u
try: try:
t = request_get_text(url, headers=self.header) t = request_get_text(url, headers=self.header)
except URLError as e: except URLError as e:
# e.reason may be Error or str # e.reason may be Error or str
if getattr(e.reason, "errno", None) == ENETUNREACH: if getattr(e.reason, "errno", None) == ENETUNREACH:
return LoginStatus.no_wifi return LoginStatus.no_wifi
raise raise
j = self.json_from_drnnnn(t) j = self.json_from_drnnnn(t)
# 检查是否付费 # 检查是否付费
if j["user_info"]["user_state"] == "正常" and j["user_info"]["available_flow"] in ("0MB", "无限制"): if j["user_info"]["user_state"] == "正常" and j["user_info"]["available_flow"] in ("0MB", "无限制"):
url = self.urlloadOnlineRecord + u url = self.urlloadOnlineRecord + u
# 获取在线设备 # 获取在线设备
t = request_get_text(url, headers=self.header) t = request_get_text(url, headers=self.header)
j1 = self.json_from_drnnnn(t) j1 = self.json_from_drnnnn(t)
result = LoginStatus.unknown result = LoginStatus.unknown
if j1['count'] <= self.online_device_limit: if j1['count'] <= self.online_device_limit:
# 判断密码是否正确 # 判断密码是否正确
url = self.urllogin + u + "&user_password=" + pwd url = self.urllogin + u + "&user_password=" + pwd
if interface_def: if interface_def:
res = request_get_text(url) res = request_get_text(url)
else: else:
command = ["curl", url, "--interface", interface] command = ["curl", url, "--interface", interface]
res = subprocess.check_output(command, text=True) res = subprocess.check_output(command, text=True)
j_res = self.json_from_drnnnn(res) j_res = self.json_from_drnnnn(res)
""" """
dr1003({"result":0,"msg":"账号不存在","ret_code":1}); dr1003({"result":0,"msg":"账号不存在","ret_code":1});
dr1003({"result":0,"msg":"密码错误","ret_code":1}); dr1003({"result":0,"msg":"密码错误","ret_code":1});
dr1003({"result":1,"msg":"Portal协议认证成功！"}); dr1003({"result":1,"msg":"Portal协议认证成功！"});
dr1003({"result":0,"msg":"IP: 10.142.5.160 已经在线！","ret_code":2}); dr1003({"result":0,"msg":"IP: 10.142.5.160 已经在线！","ret_code":2});
""" """
msg = "" msg = ""
if not interface_def: if not interface_def:
self.log(f"使用接口 {interface} 进行请求") self.log(f"使用接口 {interface} 进行请求")
msg += f"接口 {interface}： " msg += f"接口 {interface}： "
j_msg = j_res["msg"] j_msg = j_res["msg"]
if "密码错误" in j_msg: if "密码错误" in j_msg:
msg += f"{user} {pwd} 密码错误" msg += f"{user} {pwd} 密码错误"
result = LoginStatus.bad_pwd result = LoginStatus.bad_pwd
elif "已经在线" in j_msg: elif "已经在线" in j_msg:
msg += "正常在线！" msg += "正常在线！"
result = LoginStatus.used_online result = LoginStatus.used_online
elif "认证成功" in j_msg: elif "认证成功" in j_msg:
msg += f"使用账号{user}登录成功!" msg += f"使用账号{user}登录成功!"
result = LoginStatus.succ result = LoginStatus.succ
if msg != "": if msg != "":
self.log(msg) self.log(msg)
return result return result
else: else:
return LoginStatus.not_unlimit return LoginStatus.not_unlimit
def login_till_succ(self, user_pwd_gen, bad_user_callback=lambda u: None): 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.
""" """
not_succ = True not_succ = True
while not_succ: while not_succ:
(user, pwd) = user_pwd_gen() (user, pwd) = user_pwd_gen()
ret = self.login(user, pwd) ret = self.login(user, pwd)
if ret == LoginStatus.no_wifi: if ret == LoginStatus.no_wifi:
raise RuntimeError("not connect to wifi yet") raise RuntimeError("not connect to wifi yet")
not_succ = ret < 0 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(self, file, warn, sep=' '): def main(self, file, warn, sep=' '):
interface_def = self.interface_def interface_def = self.interface_def
if interface_def: if interface_def:
warn("no interface given by -i or --interface, use default route") warn("no interface given by -i or --interface, use default route")
# 检测网络联通性 # 检测网络联通性
if self.check_connectivity(): if self.check_connectivity():
return return
with open(file, 'r') as f: with open(file, 'r') as f:
user_pwd = [i.rstrip("\r\n").split(sep) for i in f.readlines()] user_pwd = [i.rstrip("\r\n").split(sep) for i in f.readlines()]
# login # login
user_pwd_error_or_not_unlimit_combo_idx = set() user_pwd_error_or_not_unlimit_combo_idx = set()
cur_lku_user_idx = 0 cur_lku_user_idx = 0
def user_pwd_getter(): def user_pwd_getter():
# 随机选择，防止前面有密码错误的用户卡死 # 随机选择，防止前面有密码错误的用户卡死
user_index = random.randrange(len(user_pwd)) user_index = random.randrange(len(user_pwd))
if user_index in user_pwd_error_or_not_unlimit_combo_idx: if user_index in user_pwd_error_or_not_unlimit_combo_idx:
return user_pwd_getter() return user_pwd_getter()
(user, pwd) = user_pwd[user_index] (user, pwd) = user_pwd[user_index]
return (user, pwd) return (user, pwd)
def bad_user_callback(_): def bad_user_callback(_):
user_pwd_error_or_not_unlimit_combo_idx.add(cur_lku_user_idx) user_pwd_error_or_not_unlimit_combo_idx.add(cur_lku_user_idx)
self.login_till_succ(user_pwd_getter, bad_user_callback) self.login_till_succ(user_pwd_getter, bad_user_callback)
# 删除密码错误或不满足要求的用户 # 删除密码错误或不满足要求的用户
with open(file, 'w') as f: with open(file, 'w') as f:
for i in range(len(user_pwd)): for i in range(len(user_pwd)):
if i in user_pwd_error_or_not_unlimit_combo_idx: if i in user_pwd_error_or_not_unlimit_combo_idx:
user_pwd_error_or_not_unlimit_combo_idx.remove(i) user_pwd_error_or_not_unlimit_combo_idx.remove(i)
else: else:
t = user_pwd[i] t = user_pwd[i]
f.write(t[0]) f.write(t[0])
f.write(' ') f.write(' ')
f.write(t[1]) f.write(t[1])
f.write('\n') f.write('\n')
if __name__ == "__main__": if __name__ == "__main__":
import argparse import argparse
import warnings import warnings
import syslog import syslog
parser = argparse.ArgumentParser(formatter_class=argparse.ArgumentDefaultsHelpFormatter) parser = argparse.ArgumentParser(formatter_class=argparse.ArgumentDefaultsHelpFormatter)
parser.add_argument("-i", "--interface", help="网卡接口名称. If not given, use default route", default=None) parser.add_argument("-i", "--interface", help="网卡接口名称. If not given, use default route", default=None)
parser.add_argument("-f", "--file", help="用户密码文件路径, one record each line", default="user_pwd.txt") parser.add_argument("-f", "--file", help="用户密码文件路径, one record each line", default="user_pwd.txt")
parser.add_argument("-S", "--no-syslog", action="store_true", help="use print over syslog for log") parser.add_argument("-S", "--no-syslog", action="store_true", help="use print over syslog for log")
parser.add_argument("-s", "--sep", help="separator used in file between user and password", default=' ') parser.add_argument("-s", "--sep", help="separator used in file between user and password", default=' ')
parser.add_argument("-n", "--online-device-limit", help="only connect to it if its online device is not more than ONLINE_DEVICE_LIMIT", type=int, default=2) parser.add_argument("-n", "--online-device-limit", help="only connect to it if its online device is not more than ONLINE_DEVICE_LIMIT", type=int, default=2)
# TODO: support auto means 3 if only one account available but 0 otherwise # TODO: support auto means 3 if only one account available but 0 otherwise
# Or connect the account with least online device number # Or connect the account with least online device number
args = parser.parse_args() args = parser.parse_args()
log = lambda *a: print(*a) log = lambda *a: print(*a)
if args.no_syslog: if args.no_syslog:
log = lambda *a: syslog.syslog(' '.join(map(str, a))) log = lambda *a: syslog.syslog(' '.join(map(str, a)))
n = args.online_device_limit n = args.online_device_limit
Loginer(args.interface, log=log, online_device_limit=n).main(args.file, warnings.warn, sep=args.sep) Loginer(args.interface, log=log, online_device_limit=n).main(args.file, warnings.warn, sep=args.sep)