From 6b782817837063b4a8584da462edaeab85e02ead Mon Sep 17 00:00:00 2001 From: RRRosmontis Date: Tue, 18 Aug 2026 07:54:05 +0000 Subject: [PATCH] =?UTF-8?q?=E4=B8=8A=E4=BC=A0=E6=96=87=E4=BB=B6=E8=87=B3?= =?UTF-8?q?=E3=80=8C/=E3=80=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- login.py | 704 +++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 704 insertions(+) create mode 100644 login.py diff --git a/login.py b/login.py new file mode 100644 index 0000000..8ce88db --- /dev/null +++ b/login.py @@ -0,0 +1,704 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- + +import requests +import time +import urllib.parse +import sys +import io +import uuid +import subprocess +import concurrent.futures +import re +import threading +import json +import os +from io import BytesIO + +if sys.platform == 'win32': + try: + sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', line_buffering=True) + except Exception: + pass + +if getattr(sys, 'frozen', False): + BASE_DIR = os.path.dirname(sys.executable) +else: + BASE_DIR = os.path.dirname(os.path.abspath(__file__)) +CONFIG_FILE = os.path.join(BASE_DIR, "config.json") +STATE_FILE = os.path.join(BASE_DIR, ".eportal_state.json") + +# 模块级默认值，main() 会从 config.json 覆盖 +PORTAL_HOST = "http://192.168.200.100" +PORTAL_BASE_URL = PORTAL_HOST + "/eportal" +USERNAME = "" +PASSWORD = "" +MAX_ATTEMPTS = 60 +SLEEP_SECONDS = 3 +PING_HOST = "www.dufe.edu.cn" +PUBLIC_KEY_FALLBACK = "10001&94dd2a8675fb779e6b9f7103698634cd400f27a154afa67af6166a43fc26417222a79506d34cacc7641946abda1785b7acf9910ad6a0978c91ec84d40b71d2891379af19ffb333e7517e390bd26ac312fe940c340466b4a5d4af1d65c3b5944078f96a1a51a5a53e4bc302818b7c9f63c4a1b07bd7d874cef1c3d4b2f5eb7871" + + +DEFAULT_CONFIG = { + "username": "", + "password": "", + "portal_host": "http://192.168.200.100", + "max_attempts": 60, + "sleep_seconds": 3, + "ping_host": "www.dufe.edu.cn", + "public_key_fallback": "10001&94dd2a8675fb779e6b9f7103698634cd400f27a154afa67af6166a43fc26417222a79506d34cacc7641946abda1785b7acf9910ad6a0978c91ec84d40b71d2891379af19ffb333e7517e390bd26ac312fe940c340466b4a5d4af1d65c3b5944078f96a1a51a5a53e4bc302818b7c9f63c4a1b07bd7d874cef1c3d4b2f5eb7871", +} + + +def load_config(): + def _write_default(): + try: + with open(CONFIG_FILE, "w", encoding="utf-8") as f: + json.dump(DEFAULT_CONFIG, f, ensure_ascii=False, indent=4) + _print(f"已创建默认配置文件: {CONFIG_FILE}") + except Exception: + pass + + if not os.path.exists(CONFIG_FILE): + _write_default() + return dict(DEFAULT_CONFIG) + + # 检测空文件或内容为空 + try: + if os.path.getsize(CONFIG_FILE) == 0: + _print("config.json 为空，重新生成默认配置", "ERROR") + _write_default() + return dict(DEFAULT_CONFIG) + except Exception: + pass + + cfg = dict(DEFAULT_CONFIG) + try: + with open(CONFIG_FILE, "r", encoding="utf-8") as f: + data = json.load(f) + if not isinstance(data, dict): + _print("config.json 格式错误，使用默认配置", "ERROR") + return dict(DEFAULT_CONFIG) + cfg.update(data) + except json.JSONDecodeError as e: + _print(f"config.json 解析失败: {e}，使用默认配置", "ERROR") + except Exception as e: + _print(f"读取 config.json 失败: {e}", "ERROR") + return cfg + + +def _print(msg, level="INFO"): + print(f"[{level}] {msg}", flush=True) + + +_skip_wait = False # Ctrl+C 退出时跳过等待 + + +def _wait_exit(): + global _skip_wait + if _skip_wait: + return + try: + input("按任意键退出...") + except (EOFError, KeyboardInterrupt): + pass + + +def save_state(mac, user_index): + try: + with open(STATE_FILE, "w") as f: + json.dump({"mac": mac, "userIndex": user_index}, f) + except Exception: + pass + + +def load_state(): + try: + if os.path.exists(STATE_FILE): + with open(STATE_FILE, "r") as f: + return json.load(f) + except Exception: + pass + return None + + +def detect_portal(): + try: + requests.get(PORTAL_HOST + "/", timeout=(2, 2), allow_redirects=False) + return None, None + except Exception: + pass + + for url in [ + "http://www.baidu.com", + "http://detectportal.firefox.com/canonical.html", + "http://www.msftconnecttest.com/redirect", + ]: + try: + resp = requests.get(url, timeout=(2, 2), allow_redirects=False) + if resp.status_code in (301, 302, 303, 307, 308): + loc = resp.headers.get("Location", "") + m = re.match(r'(https?://[^/]+(?:/eportal)?)', loc) + if m: + host = re.match(r'https?://[^/]+', m.group(1)).group(0) + base = m.group(1) + return host, base + except Exception: + pass + return None, None + + +_page_info_cache = None + + +def _fetch_page_info(): + global _page_info_cache + if _page_info_cache: + return _page_info_cache + + info = {"public_key": None, "query_string": "", "portal_mac": None} + + def parse_page(resp, extract_qs=True): + html = resp.text + if "?" in resp.url: + info["query_string"] = resp.url.split("?", 1)[1] + m = re.search(r'[?&]mac=([^&]+)', resp.url) + if m: + info["portal_mac"] = urllib.parse.unquote(m.group(1)).replace(':', '').replace('-', '').lower() + # 从 HTML 提取 publicKey + for pid in ['publicKey', 'publicKeyModulus', 'publicKeyExponent']: + m = re.search(rf'(?:id|name)=["\']{pid}["\'][^>]*value=["\']([^"\']+)', html) + if m: + if pid == 'publicKey' and '&' in m.group(1): + info["public_key"] = m.group(1) + break + if not info["public_key"]: + exp_match = re.search(r'(?:id|name)=["\']publicKeyExponent["\'][^>]*value=["\']([^"\']+)', html) + mod_match = re.search(r'(?:id|name)=["\']publicKeyModulus["\'][^>]*value=["\']([^"\']+)', html) + if exp_match and mod_match: + info["public_key"] = exp_match.group(1) + "&" + mod_match.group(1) + if not extract_qs: + return + # 从 HTML 提取 queryString（JS 重定向: location.href='...?wlanuserip=...'） + if not info["query_string"]: + m = re.search(r"location\.href\s*=\s*['\"]([^'\"]*\?[^'\"]+)", html) + if m: + qs_url = m.group(1) + if "?" in qs_url: + info["query_string"] = qs_url.split("?", 1)[1] + if not info["query_string"]: + params = [] + for key in ['wlanuserip', 'wlanacname', 'wlanacip', 'wlanusermac', 'wlanuserfirsturl', + 'ssid', 'nasip', 'snmpagentip', 'mac', 't', 'url']: + m = re.search(rf'(?:id|name)=["\']{key}["\'][^>]*value=["\']([^"\']+)', html) + if m: + params.append(f"{key}={m.group(1)}") + if params: + info["query_string"] = "&".join(params) + + # 方法1: 跟随 portal 根路径重定向链，提取 JS 重定向中的 queryString + try: + resp = requests.get(PORTAL_HOST + "/", timeout=10, allow_redirects=True) + parse_page(resp) + except Exception: + pass + + # 方法2: 访问 eportal 页面获取 publicKey（同时尝试提取 queryString） + try: + resp = requests.get(PORTAL_BASE_URL + "/", timeout=10, allow_redirects=True) + parse_page(resp, extract_qs=not bool(info["query_string"])) + except Exception: + pass + + # 方法3: 对 HTTP 外部 URL 发起请求触发 DNS 劫持，从返回的 HTML 提取 queryString + if not info["query_string"]: + for test_url in ["http://www.baidu.com", "http://detectportal.firefox.com/canonical.html"]: + try: + resp = requests.get(test_url, timeout=5, allow_redirects=False) + if resp.status_code in (301, 302, 303, 307, 308): + loc = resp.headers.get("Location", "") + if "?" in loc: + info["query_string"] = loc.split("?", 1)[1] + m = re.search(r'[?&]mac=([^&]+)', loc) + if m: + info["portal_mac"] = urllib.parse.unquote(m.group(1)).replace(':', '').replace('-', '').lower() + break + elif resp.status_code == 200: + parse_page(resp) + if info["query_string"]: + break + except Exception: + pass + + _page_info_cache = info + return info + + +def get_public_key(): + info = _fetch_page_info() + if info["public_key"]: + return info["public_key"] + return PUBLIC_KEY_FALLBACK + + +def get_query_string(): + return _fetch_page_info()["query_string"] + + +def get_portal_mac(): + return _fetch_page_info()["portal_mac"] + + +def js_rsa_encrypt(data_str, modulus_hex, exponent_hex): + # 模拟 JS RSAUtils.encryptedString：原始RSA，零填充，每2个charCode打包为1个16bit digit + modulus = int(modulus_hex, 16) + exponent = int(exponent_hex, 16) + + modulus_bytes = (modulus.bit_length() + 7) // 8 + js_digits = (modulus_bytes + 1) // 2 + chunk_size = 2 * (js_digits - 1) + + a = [ord(ch) for ch in data_str] + while len(a) % chunk_size != 0: + a.append(0) + + parts = [] + for i in range(0, len(a), chunk_size): + val = 0 + j = 0 + for k in range(i, i + chunk_size, 2): + val += (a[k] + (a[k + 1] << 8)) << (16 * j) + j += 1 + parts.append(hex(pow(val, exponent, modulus))[2:]) + + return " ".join(parts) + + +def double_encode(s): + return urllib.parse.quote(urllib.parse.quote(s, safe=''), safe='') + + +def _ocr_single(img_bytes): + try: + from PIL import Image + import pytesseract + img = Image.open(BytesIO(img_bytes)) + gray = img.convert("L") + binary = gray.point(lambda x: 0 if x < 200 else 255) + return pytesseract.image_to_string( + binary, + config="--psm 7 -c tessedit_char_whitelist=0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZ" + ).strip() + except (ImportError, Exception): + return "" + + +def _fetch_captcha(): + import random + url = f"{PORTAL_BASE_URL}/validcode?rnd={random.random()}" + resp = requests.get(url, timeout=10) + return resp.content, url + + +def recognize_captcha(first_url): + try: + if not first_url.startswith("http"): + first_url = urllib.parse.urljoin(PORTAL_BASE_URL, first_url) + r = requests.get(first_url, timeout=10) + text = _ocr_single(r.content) + if text: + _print(f"验证码 OCR 识别: {text}") + return text + except Exception as e: + _print(f"验证码下载失败: {e}", "ERROR") + + for i in range(2, 4): + _print(f"OCR未识别，刷新验证码重试({i}/3)...") + try: + img_bytes, _ = _fetch_captcha() + text = _ocr_single(img_bytes) + if text: + _print(f"OCR第{i}次: {text}") + return text + except Exception as e: + _print(f"验证码刷新失败: {e}", "ERROR") + + _print("OCR 3次均失败，切换为手动输入...", "ERROR") + try: + from PIL import Image + img_bytes, _ = _fetch_captcha() + Image.open(BytesIO(img_bytes)).show() + except Exception: + pass + return input(" 请手动输入验证码: ").strip() + + +def get_all_macs(): + macs = [] + mac_raw = set() + + def add_mac(m): + if not m: + return + m = m.replace(':', '').replace('-', '').replace('.', '').lower().strip() + if len(m) == 12 and all(c in '0123456789abcdef' for c in m): + if m not in mac_raw and m != '000000000000': + mac_raw.add(m) + macs.append(m) + + # ipconfig /all（Windows 最可靠） + try: + output = subprocess.run( + ['ipconfig', '/all'], + capture_output=True, text=True, timeout=5 + ).stdout + for line in output.split('\n'): + if '物理地址' in line or 'Physical' in line: + parts = line.split(':') + if len(parts) >= 2: + add_mac(parts[-1]) + except Exception: + pass + + # getmac 库 + try: + from getmac import get_mac_address + m = get_mac_address() + if m: + add_mac(m) + except Exception: + pass + + # uuid（不可靠，放最后） + try: + m = format(uuid.getnode(), '012x') + if m != '000000000000': + add_mac(m) + except Exception: + pass + + # PowerShell 备用 + if not macs: + try: + output = subprocess.run( + ['powershell', '-Command', + 'Get-NetAdapter | Select-Object -ExpandProperty MacAddress'], + capture_output=True, text=True, timeout=5 + ).stdout + for line in output.strip().split('\n'): + add_mac(line) + except Exception: + pass + + return macs + + +def is_authenticated(): + test_urls = [ + ("https://detectportal.firefox.com/canonical.html", "success"), + ("https://www.msftconnecttest.com/connecttest.txt", "Microsoft"), + ("https://www.baidu.com", None), + ] + passed = 0 + for url, keyword in test_urls: + try: + r = requests.get(url, allow_redirects=False, timeout=3) + if r.status_code == 200: + if keyword: + if keyword in r.text: + passed += 1 + else: + passed += 1 + except Exception: + continue + return passed >= 2 + + +def do_login(mac, query_string=""): + exponent_hex, modulus_hex = get_public_key().split('&') + plain = f"{PASSWORD}>{mac}" + encrypted = js_rsa_encrypt(plain[::-1], modulus_hex, exponent_hex) + + def build_body(validcode=""): + return "&".join([ + f"userId={double_encode(USERNAME)}", + f"password={double_encode(encrypted)}", + f"service={double_encode('')}", + f"queryString={double_encode(query_string)}", + "operatorPwd=", + "operatorUserId=", + f"validcode={double_encode(validcode)}", + f"passwordEncrypt={double_encode('true')}", + ]) + + def parse_result(result): + ok = result.get("result") == "success" + ui = result.get("userIndex", "") + ki = result.get("keepaliveInterval", 0) + return ok, result.get("message", ""), ui, ki + + try: + resp = requests.post( + f"{PORTAL_BASE_URL}/InterFace.do?method=login", + data=build_body(), + headers={"Content-Type": "application/x-www-form-urlencoded; charset=UTF-8"}, + timeout=10, + ) + resp.encoding = 'utf-8' + result = resp.json() + + if result.get("result") == "fail" and result.get("validCodeUrl"): + captcha_url = result["validCodeUrl"] + if captcha_url and captcha_url.strip(): + _print(f"需要验证码 ({captcha_url})") + code = recognize_captcha(captcha_url) + if code: + _print(f"输入验证码: {code}，重试...") + resp2 = requests.post( + f"{PORTAL_BASE_URL}/InterFace.do?method=login", + data=build_body(code), + headers={"Content-Type": "application/x-www-form-urlencoded; charset=UTF-8"}, + timeout=10, + ) + resp2.encoding = 'utf-8' + result = resp2.json() + + return parse_result(result) + except Exception as e: + return False, str(e), "", 0 + + +def ping_test(host=None, count=3): + if host is None: + host = PING_HOST + try: + cmd = ["ping", "-n", str(count), host] if sys.platform == "win32" else ["ping", "-c", str(count), host] + output = subprocess.run(cmd, capture_output=True, text=True, timeout=10) + if output.returncode == 0: + # 提取平均延迟 + m = re.search(r'平均\s*=\s*(\d+)\s*ms|avg.*?(\d+\.?\d*).*?ms', output.stdout) + if m: + avg = m.group(1) or m.group(2) + _print(f"Ping {host} 平均延迟: {avg}ms") + else: + _print(f"Ping {host} 成功（延迟未解析）") + else: + _print(f"Ping {host} 失败: {output.stderr.strip()}", "ERROR") + except Exception as e: + _print(f"Ping {host} 异常: {e}", "ERROR") + + +def register_mac(mac, user_index): + try: + resp = requests.post( + f"{PORTAL_BASE_URL}/InterFace.do?method=registerMac", + data=f"mac={mac}&userIndex={user_index}", + headers={"Content-Type": "application/x-www-form-urlencoded; charset=UTF-8"}, + timeout=10, + ) + resp.encoding = 'utf-8' + result = resp.json() + if result.get("result") == "success": + _print(f"MAC快速认证已注册: {mac}") + else: + _print(f"MAC注册: {result.get('message', result)}", "ERROR") + except Exception as e: + _print(f"MAC注册异常: {e}", "ERROR") + + +def start_keepalive(user_index, interval_seconds): + interval = max(60, min(600, int(interval_seconds) if interval_seconds else 180)) + + def _loop(): + count = 0 + while True: + time.sleep(interval) + count += 1 + try: + resp = requests.post( + f"{PORTAL_BASE_URL}/InterFace.do?method=keepalive", + data=f"userIndex={user_index}", + headers={"Content-Type": "application/x-www-form-urlencoded; charset=UTF-8"}, + timeout=10, + ) + resp.encoding = 'utf-8' + result = resp.json() + _print(f"保活 #{count} → {result.get('result', '?')}") + except Exception as e: + _print(f"保活 #{count} 异常: {e}", "ERROR") + + t = threading.Thread(target=_loop, daemon=True) + t.start() + _print(f"保活线程已启动（间隔 {interval}s）") + + +def main(): + global PORTAL_HOST, PORTAL_BASE_URL, _page_info_cache + global USERNAME, PASSWORD, MAX_ATTEMPTS, SLEEP_SECONDS + global PING_HOST, PUBLIC_KEY_FALLBACK + + cfg = load_config() + if not cfg["username"] or not cfg["password"]: + _print("请编辑 config.json 填入账号密码后重新运行", "ERROR") + return False + + PORTAL_HOST = cfg["portal_host"] + PORTAL_BASE_URL = PORTAL_HOST + "/eportal" + USERNAME = cfg["username"] + PASSWORD = cfg["password"] + MAX_ATTEMPTS = int(cfg["max_attempts"]) + SLEEP_SECONDS = int(cfg["sleep_seconds"]) + PING_HOST = cfg["ping_host"] + PUBLIC_KEY_FALLBACK = cfg["public_key_fallback"] + + _print(f"校园网认证登录") + _print(f"目标: {PORTAL_HOST} 用户: {USERNAME}") + + print("[INFO] 自动检测 Portal 地址...", end=" ", flush=True) + with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor: + future = executor.submit(detect_portal) + try: + host, base = future.result(timeout=5) + except concurrent.futures.TimeoutError: + host, base = None, None + if host and base: + PORTAL_HOST = host + PORTAL_BASE_URL = base + _page_info_cache = None + print(f"→ {base}") + else: + print("→ 使用默认") + + if is_authenticated(): + _print("已认证，无需登录！") + return True + + saved = load_state() + if saved and saved.get("userIndex"): + try: + resp = requests.post( + f"{PORTAL_BASE_URL}/InterFace.do?method=getOnlineUserInfo", + data=f"userIndex={saved['userIndex']}", + headers={"Content-Type": "application/x-www-form-urlencoded; charset=UTF-8"}, + timeout=10, + ) + resp.encoding = 'utf-8' + info = resp.json() + if info.get("result") == "success": + _print(f"恢复上次会话 userIndex={saved['userIndex']}") + ping_test() + start_keepalive(saved["userIndex"], info.get("keepaliveInterval", 180)) + _print("按 Ctrl+C 退出...") + try: + while True: + time.sleep(60) + except KeyboardInterrupt: + _print("退出保活") + return True + else: + _print("上次会话已失效，重新登录...") + except Exception: + pass + + all_macs = [] + if saved and saved.get("mac"): + all_macs.append(saved["mac"]) + _print(f"[上次MAC] {saved['mac']}") + + portal_mac = get_portal_mac() + if portal_mac: + all_macs.append(portal_mac) + _print(f"[Portal MAC] {portal_mac}") + + local_macs = get_all_macs() + for m in local_macs: + if m not in all_macs: + all_macs.append(m) + + _print(f"[扫描] 发现 {len(local_macs)} 个本地网卡MAC:") + for i, m in enumerate(local_macs): + print(f" {i+1}. {m}") + + qs = get_query_string() + if qs: + _print(f"[queryString] {qs[:80]}...") + + fallbacks = ["111111111111", "000000000000", "ffffffffffff", + "111111111", "123456789abc", "abcdef123456", + "000000000001", "deadbeef0000"] + for m in fallbacks: + if m not in all_macs: + all_macs.append(m) + + _print(f"共 {len(all_macs)} 种MAC方案，最多尝试 {MAX_ATTEMPTS} 次\n") + + last_msg = "" + mac_index = 0 + delay = SLEEP_SECONDS + consecutive_fails = 0 + + for attempt in range(1, MAX_ATTEMPTS + 1): + mac = all_macs[mac_index % len(all_macs)] + mac_index += 1 + + if attempt % (len(all_macs) + 1) == 0: + fresh = get_all_macs() + for m in fresh: + if m not in all_macs: + all_macs.append(m) + _print(f"[新增MAC] {m}") + + print(f"[{attempt:02d}/{MAX_ATTEMPTS}] MAC={mac:14s}", end=" ") + + ok, msg, user_idx, keep_interval = do_login(mac, query_string=qs) + print(f"→ {msg}") + + if ok: + _print(f"登录成功！MAC={mac}") + ping_test() + save_state(mac, user_idx) + if user_idx: + _print(f"userIndex: {user_idx}") + register_mac(mac, user_idx) + start_keepalive(user_idx, keep_interval) + _print("按 Ctrl+C 退出保活...") + try: + while True: + time.sleep(60) + except KeyboardInterrupt: + global _skip_wait + _skip_wait = True + _print("退出保活") + return True + + consecutive_fails += 1 + if msg != last_msg: + delay = SLEEP_SECONDS + consecutive_fails = 0 + else: + delay = min(SLEEP_SECONDS * (1.5 ** (consecutive_fails // 3)), 60) + + last_msg = msg + + if attempt % 5 == 0 and is_authenticated(): + _print("检测到已认证重定向！") + return True + + time.sleep(delay) + + if is_authenticated(): + _print("最终检查：已认证") + return True + + _print(f"{MAX_ATTEMPTS} 次尝试失败。最后: {last_msg}", "ERROR") + return False + + +if __name__ == "__main__": + try: + main() + except KeyboardInterrupt: + pass + except Exception as e: + _print(f"异常退出: {e}", "ERROR") + _wait_exit()