Compare commits

...
Author SHA1 Message Date
RoyirinRosmontis 17cdb276ea 修复: - 节点认证方式升级为 SSO:user/password 按 URL-safe Base64 作为查询参数拼接在 config.json 的 nodes_url 后(旧 nginx auth_basic 已废弃),服务端返回的 401/403/429 等中文原因直接透出到登录框,登录框输入占位改为 SSO 邮箱 2026-09-03 22:40:50 +08:00
RoyirinRosmontis f8eba465a9 修复: - 日志不再打印用户数据目录路径:节点配置加载与首启创建提示中的 %LOCALAPPDATA% 路径已移除 2026-08-31 16:36:44 +08:00
RoyirinRosmontis 53e9d1e31a 修复: - 检查更新不再内置回退地址:只读取 config.json 的 update_url,未配置时检查更新直接失败并记录日志 2026-08-31 16:25:48 +08:00
RoyirinRosmontis 59393b9ae9 修复: - 检查更新地址改用 config.json 的 update_url 配置,内置回退地址域名从 git.bu4.ren 更换为 git.rrrosmontis.icu 2026-08-31 16:24:15 +08:00
RoyirinRosmontis 89b5c09e95 修复: - 设置页代理转发端口输入框横向过长:宽度 72px 缩到 54px(够放下 5 位端口号) 2026-08-27 13:19:51 +08:00
RoyirinRosmontis b15beb97b4 修复: - 设置页代理转发端口输入框过大:内边距从 5px 缩到 2px,并固定为 72×22 与右侧开关同高,和其他设置项同一水平 2026-08-27 13:19:33 +08:00
RoyirinRosmontis 567d395d6f 新增功能: - 代理白名单:设置页新增第一个设置项「代理白名单+配置」,点配置打开编辑弹窗(Windows 环境变量风格列表),支持添加/删除强制直连的域名、IP、CIDR(IPv4/IPv6 均支持),每条规则带复选框可单独启用/禁用,保存后立即生效,白名单优先于域名表强制隧道与固定代理模式 2026-08-27 13:18:07 +08:00
+347 -26
View File
@@ -7,6 +7,7 @@ from PyQt5.QtWidgets import (
QApplication, QWidget, QLabel, QPushButton,
QVBoxLayout, QHBoxLayout, QFrame, QMessageBox, QTextEdit,
QCheckBox, QMenu, QDialog, QLineEdit,
QListWidget, QListWidgetItem,
)
from PyQt5.QtCore import Qt, QTimer, QRectF, pyqtSignal
from PyQt5.QtGui import QFont, QIcon, QPixmap, QImage, QColor, QPainter
@@ -91,7 +92,7 @@ def _load_nodes():
_NODES = data
if not _SELECTED_RESTORED:
_restore_selected_node(_NODES) # 恢复上次使用的节点
log(f"已加载节点配置: {len(_NODES)} 个 ({path})")
log(f"已加载节点配置: {len(_NODES)} 个")
if data and not str(data[0].get("domain") or "").strip():
log_red("节点未填写域名:请配置 config.json 的 nodes_url 在线更新,"
"或使用 others/nodes_tool 导入节点配置")
@@ -635,6 +636,74 @@ def _set_skip_update_check(enabled):
_settings_save(data)
log(f"检查更新已{'跳过' if enabled else '恢复'}(PAC 与程序版本;节点更新不受影响)")
# ── 代理白名单(用户可配置,强制直连,settings.json 的 whitelist 键) ─────────
# 每条规则 {"value": 域名/IP/CIDR, "enabled": 是否启用},命中即直连(不经过代理)。
# 规则保存在内存缓存 _WHITELIST,连接建立时直接读取,不落盘回读;
# _WHITELIST_VERSION 在保存时 +1,使按 (host, version) 键控的匹配缓存失效。
_WHITELIST = None # 内存缓存:规范化后的规则列表
_WHITELIST_VERSION = 0 # 规则版本号(保存时 +1,用于匹配缓存失效)
def _whitelist_rules():
"""白名单规则列表(内存缓存,首次读取 settings.json;保存时同步更新)。"""
global _WHITELIST
if _WHITELIST is None:
raw = _settings_load().get("whitelist")
if not isinstance(raw, list):
raw = []
_WHITELIST = [{"value": str(r.get("value") or "").strip().lower(),
"enabled": bool(r.get("enabled", True))}
for r in raw if isinstance(r, dict) and str(r.get("value") or "").strip()]
return _WHITELIST
def _whitelist_save(rules):
"""保存白名单规则(规范化 + 去空白条目),更新内存缓存并记录日志。"""
global _WHITELIST, _WHITELIST_VERSION
clean = [{"value": _whitelist_normalize(r.get("value") or ""),
"enabled": bool(r.get("enabled", True))}
for r in rules if _whitelist_normalize(r.get("value") or "")]
data = _settings_load()
data["whitelist"] = clean
_settings_save(data)
_WHITELIST = clean
_WHITELIST_VERSION += 1
log(f"代理白名单已保存: {len(clean)} 条(立即生效)")
def _whitelist_normalize(entry: str) -> str:
"""条目规范化:去协议头/通配符/末尾点,统一小写。"""
s = str(entry).strip().lower()
if "://" in s:
s = s.split("://", 1)[1]
return s.rstrip("/").rstrip(".").lstrip("*.")
def _whitelist_validate_entry(entry: str):
"""校验并规范化白名单条目;返回 (规范值, 错误),错误非空表示非法。"""
s = _whitelist_normalize(entry)
if not s:
return "", "内容不能为空"
if any(ch.isspace() for ch in s):
return "", "不能包含空格"
import ipaddress
if "/" in s:
# CIDR(IPv4/IPv6 网段或单地址)
try:
ipaddress.ip_network(s, strict=False)
except ValueError:
return "", "IP/CIDR 格式不正确(如 10.0.0.0/8)"
elif s.count(".") == 3 and all(p.isdigit() for p in s.split(".")):
# 形如 x.x.x.x:必须是合法 IPv4,防止 300.1.2.3 这类误当域名
try:
ipaddress.ip_address(s)
except ValueError:
return "", "IP 格式不正确"
elif ":" in s:
# IPv6 字面量(域名不含冒号)
try:
ipaddress.ip_address(s)
except ValueError:
return "", "IP 格式不正确"
return s, ""
def _ver_tuple(v):
"""版本号转数字元组:'v0.1.0' -> (0, 1, 0),无法解析的段按 0 处理。"""
parts = []
@@ -826,16 +895,40 @@ def _nodes_sync_remote():
except Exception as e:
log(f"节点配置在线更新失败,使用本地配置: {e}")
class _AuthError(Exception):
"""订阅接口返回的错误:code 为 HTTP 状态码,str 为服务端返回的中文提示。"""
def __init__(self, msg, code=0):
super().__init__(msg)
self.code = code
def _b64url(s: str) -> str:
"""SSO 认证查询参数的 URL-safe Base64 编码(末尾 padding 省略)。"""
return base64.urlsafe_b64encode(str(s).encode("utf-8")).decode("ascii").rstrip("=")
def _fetch_nodes_auth(url, user, password, timeout=10):
"""带 Basic 认证拉取 nodes.json(参与 nginx 认证;直连,不经过系统代理)。"""
"""用 SSO 账号拉取 nodes.json:user/password 按 URL-safe Base64 作为查询参数
拼接在 nodes_url 后(旧版 nginx auth_basic 已废弃);直连,不经过系统代理。
失败抛 _AuthError(服务端拒绝,str 为服务端中文原因)或 urllib 异常(网络问题)。"""
import json
import urllib.request
import urllib.error
opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))
req = urllib.request.Request(url)
req.add_header("Authorization", "Basic " + base64.b64encode(
f"{user}:{password}".encode()).decode())
with opener.open(req, timeout=timeout) as r:
return json.loads(r.read().decode("utf-8"))
sep = "&" if "?" in url else "?"
req_url = f"{url}{sep}user={_b64url(user)}&password={_b64url(password)}"
try:
with opener.open(req_url, timeout=timeout) as r:
return json.loads(r.read().decode("utf-8"))
except urllib.error.HTTPError as e:
# 401/403/429/5xx:服务端 body 是面向用户的中文原因,直接透出
body = ""
try:
body = e.read().decode("utf-8", "replace").strip()
except Exception:
pass
raise _AuthError(body or f"请求被拒绝 (HTTP {e.code})", e.code) from None
def _user_init_local():
@@ -847,13 +940,13 @@ def _user_init_local():
"username": "", "password": "", "remark": ""}]
(_USER_DIR / "nodes.json").write_bytes(_nodes_encrypt(
json.dumps(template, ensure_ascii=False, indent=2) + "\n"))
log_red(f"已创建 {_USER_DIR / 'nodes.json'}(加密存储),请配置 config.json 的 "
f"nodes_url 在线更新,或使用 others/nodes_tool 导入节点配置")
log_red("已创建节点配置文件(加密存储),请配置 config.json 的 "
"nodes_url 在线更新,或使用 others/nodes_tool 导入节点配置")
def _user_init_remote(base_url=None):
"""检查并更新 PAC 数据(后台线程):
直接拉取两个数据文件(内含 version/date),按版本比对后更新;失败用本地副本。
节点配置的拉取走登录框(nodes_url 需 nginx 认证,见 LoginDialog)。
节点配置的拉取走登录框(nodes_url 需 SSO 认证,见 LoginDialog)。
"""
if _skip_update_check():
return # 设置跳过:不拉取也不打日志
@@ -1003,11 +1096,55 @@ def _is_force_proxy(host: str) -> bool:
h = host.rstrip(".").lower()
return any(h == s or h.endswith("." + s) for s in suffixes)
def _resolve_host_ips(host: str):
"""解析域名的 IPv4/IPv6 地址集合(字面量 IP 返回自身);解析失败返回空集合。"""
import ipaddress
try:
return {ipaddress.ip_address(host)}
except ValueError:
pass
try:
import socket
# 双栈解析(AF_UNSPEC):IPv6 白名单条目也能命中域名解析出的 IPv6 地址
return {ipaddress.ip_address(a[4][0])
for a in socket.getaddrinfo(host, None, socket.AF_UNSPEC, socket.SOCK_STREAM)}
except Exception:
return set()
def _whitelist_entry_match(host: str, entry: str) -> bool:
"""单条白名单规则匹配:IP/CIDR 条目按解析地址比对,域名条目按后缀匹配。"""
h = host.rstrip(".").lower()
entry = entry.rstrip(".").lower()
import ipaddress
net = None
# 只有含点/冒号才按 IP 解析:纯数字串(如 8080)会被 ip_network 误解析为
# 整型地址,应视为域名条目
if "." in entry or ":" in entry:
try:
net = ipaddress.ip_network(entry, strict=False)
except ValueError:
net = None
if net is not None:
return any(addr in net for addr in _resolve_host_ips(h))
# 域名条目:精确匹配或子域后缀匹配(与 pac_domains 同一规则)
return h == entry or h.endswith("." + entry)
@functools.lru_cache(maxsize=8192)
def _is_whitelisted(host: str, version: int) -> bool:
"""host 是否命中已启用的白名单规则(按版本键控缓存,规则变更即失效)。"""
entries = [r["value"] for r in _whitelist_rules() if r.get("enabled")]
if not entries:
return False
return any(_whitelist_entry_match(host, e) for e in entries)
def _decide_direct(host: str, bypass_lan: bool, smart: bool) -> bool:
"""判断连接是否直连。判定顺序(域名表优先于 IP):
域名表命中 -> 隧道;内网 + bypass_lan -> 直连;中国 IP -> 直连,其余 -> 隧道。
仅 PAC(智能分流)模式生效,固定代理模式全部走隧道。
"""判断连接是否直连。判定顺序(白名单最优先):
白名单命中 -> 直连;域名表命中 -> 隧道;内网 + bypass_lan -> 直连;
中国 IP -> 直连,其余 -> 隧道。
白名单在 PAC 与固定代理模式下均生效(固定模式除白名单外全部走隧道)。
"""
if _is_whitelisted(host, _WHITELIST_VERSION):
return True
if smart and _is_force_proxy(host):
return False
if bypass_lan and _is_private_ip(host):
@@ -1284,7 +1421,8 @@ class NodeLabel(QLabel):
# ── 节点配置登录框 ─────────────────────────────────────
class LoginDialog(QDialog):
"""节点配置登录框:凭据参与 nodes_url 的 nginx 认证,成功后拉取并加密保存 nodes.json。"""
"""节点配置登录框:SSO 账号(邮箱+密码,base64url 查询参数)认证 nodes_url,
成功后拉取并加密保存 nodes.json。"""
_done = pyqtSignal(bool, str)
def __init__(self, parent=None):
@@ -1341,7 +1479,7 @@ class LoginDialog(QDialog):
form = QVBoxLayout()
form.setSpacing(6)
self.ed_user = QLineEdit()
self.ed_user.setPlaceholderText("登录名")
self.ed_user.setPlaceholderText("SSO 邮箱")
form.addWidget(self.ed_user)
self.ed_pass = QLineEdit()
@@ -1396,7 +1534,7 @@ class LoginDialog(QDialog):
threading.Thread(target=self._worker, args=(user, password), daemon=True).start()
def _worker(self, user, password):
"""后台登录:Basic 认证拉取 nodes.json,校验后加密写入用户数据目录。"""
"""后台登录:SSO 认证拉取 nodes.json,校验后加密写入用户数据目录。"""
try:
import json
url = str(_config_val("nodes_url", "") or "").strip()
@@ -1410,12 +1548,13 @@ class LoginDialog(QDialog):
text = json.dumps(data, ensure_ascii=False, indent=2) + "\n"
_USER_DIR.mkdir(parents=True, exist_ok=True)
(_USER_DIR / "nodes.json").write_bytes(_nodes_encrypt(text))
_auth_save(user, password) # 记住凭据,之后每次启动自动同步节点
_auth_save(user, password) # 记住 SSO 凭据,之后每次启动自动同步节点
self._done.emit(True, str(len(data)))
except _AuthError as e:
# 服务端拒绝:直接展示其中文原因(认证失败/未开通/限流等)
self._done.emit(False, str(e))
except Exception as e:
msg = ("认证失败,请检查用户名/密码" if getattr(e, "code", None) == 401
else f"登录失败: {e}")
self._done.emit(False, msg)
self._done.emit(False, f"登录失败: {e}")
def _on_done(self, ok, info):
self.btn_ok.setEnabled(True)
@@ -1428,6 +1567,164 @@ class LoginDialog(QDialog):
self.lb_err.setText(info)
self.lb_err.show()
# ── 代理白名单弹窗 ─────────────────────────────────────
class WhitelistDialog(QDialog):
"""代理白名单编辑弹窗:可编辑列表(Windows 环境变量风格),
每条规则带复选框启用/禁用,支持添加/删除,保存后立即生效。"""
def __init__(self, parent=None):
super().__init__(parent)
self.setWindowTitle("代理白名单")
self.setWindowIcon(load_icon())
self.setFixedSize(380, 400)
self.setStyleSheet(f"""
QDialog {{ background-color: {C_BG}; }}
QLabel {{ color: {C_TEXT}; font-family: "Microsoft YaHei", "Segoe UI", sans-serif; font-size: 12px; }}
QLineEdit {{
background-color: {C_CARD};
border: 1px solid {C_BORDER};
border-radius: 6px;
padding: 6px 8px;
color: {C_TEXT};
font-size: 12px;
}}
QListWidget {{
background-color: {C_CARD};
border: 1px solid {C_BORDER};
border-radius: 6px;
padding: 4px;
color: {C_TEXT};
font-size: 12px;
}}
QListWidget::item {{ padding: 3px 6px; border-radius: 4px; }}
QListWidget::item:selected {{ background-color: #e3e8ee; color: {C_TEXT}; }}
QPushButton#btnOk {{
background-color: {C_GREEN}; color: #fff; border: none;
border-radius: 6px; padding: 4px 12px;
font-size: 12px; font-weight: bold;
}}
QPushButton#btnOk:hover {{ background-color: #5ed978; }}
QPushButton#btnCancel {{
background-color: {C_CARD}; color: {C_SUBTEXT};
border: 1px solid {C_BORDER}; border-radius: 6px;
padding: 4px 12px; font-size: 12px;
}}
QPushButton#btnSmall {{
background-color: {C_CARD}; color: {C_SUBTEXT};
border: 1px solid {C_BORDER}; border-radius: 6px;
padding: 3px 10px; font-size: 11px;
}}
QPushButton#btnSmall:hover {{ background-color: #e3e8ee; color: {C_TEXT}; }}
""")
layout = QVBoxLayout(self)
layout.setSpacing(10)
layout.setContentsMargins(20, 16, 20, 16)
title = QLabel("代理白名单")
title.setStyleSheet(f"color: {C_TITLE}; font-size: 15px; font-weight: bold;")
title.setAlignment(Qt.AlignCenter)
layout.addWidget(title)
hint = QLabel("命中以下规则的连接将强制直连(不经过代理)")
hint.setStyleSheet(f"color: {C_SUBTEXT}; font-size: 11px;")
hint.setAlignment(Qt.AlignCenter)
layout.addWidget(hint)
self.lb_err = QLabel("")
self.lb_err.setStyleSheet(f"color: {C_RED}; font-size: 11px;")
self.lb_err.setFixedHeight(14)
self.lb_err.hide() # 无错误时不占空间
layout.addWidget(self.lb_err)
# 规则列表:每条前面带复选框(勾选启用 / 取消禁用)
self.list_widget = QListWidget()
self.list_widget.setToolTip("勾选启用规则;可多选后删除")
for r in _whitelist_rules():
self._add_item(r["value"], r.get("enabled", True))
layout.addWidget(self.list_widget, 1)
# 添加行:输入框 + 添加按钮
add_row = QHBoxLayout()
add_row.setSpacing(8)
self.ed_add = QLineEdit()
self.ed_add.setPlaceholderText("输入域名 / IP / CIDR,如 example.com")
self.ed_add.returnPressed.connect(self._add_rule)
add_row.addWidget(self.ed_add, 1)
self.btn_add = QPushButton("添加")
self.btn_add.setObjectName("btnOk")
self.btn_add.setStyleSheet(f"""
QPushButton {{
background-color: {C_GREEN}; color: #fff; border: none;
border-radius: 6px; padding: 6px 14px;
font-size: 12px; font-weight: bold;
}}
QPushButton:hover {{ background-color: #5ed978; }}
""")
self.btn_add.clicked.connect(self._add_rule)
add_row.addWidget(self.btn_add)
layout.addLayout(add_row)
# 删除选中 + 底部按钮
mid_row = QHBoxLayout()
self.btn_delete = QPushButton("删除选中")
self.btn_delete.setObjectName("btnSmall")
self.btn_delete.clicked.connect(self._delete_selected)
mid_row.addWidget(self.btn_delete)
mid_row.addStretch()
layout.addLayout(mid_row)
btns = QHBoxLayout()
btns.setSpacing(10)
btns.addStretch()
self.btn_cancel = QPushButton("取消")
self.btn_cancel.setObjectName("btnCancel")
self.btn_cancel.clicked.connect(self.reject)
btns.addWidget(self.btn_cancel)
self.btn_ok = QPushButton("保存")
self.btn_ok.setObjectName("btnOk")
self.btn_ok.clicked.connect(self._save)
btns.addWidget(self.btn_ok)
layout.addLayout(btns)
def _add_item(self, value, enabled):
item = QListWidgetItem(value)
item.setFlags(item.flags() | Qt.ItemIsUserCheckable)
item.setCheckState(Qt.Checked if enabled else Qt.Unchecked)
self.list_widget.addItem(item)
def _add_rule(self):
value, err = _whitelist_validate_entry(self.ed_add.text())
if err:
self.lb_err.setText(err)
self.lb_err.show()
return
# 去重(不区分大小写)
for i in range(self.list_widget.count()):
if self.list_widget.item(i).text().lower() == value.lower():
self.lb_err.setText("该规则已存在")
self.lb_err.show()
return
self.lb_err.hide()
self._add_item(value, True)
self.ed_add.clear()
def _delete_selected(self):
for item in self.list_widget.selectedItems():
self.list_widget.takeItem(self.list_widget.row(item))
def _save(self):
"""收集列表内容保存(未点保存前所有改动均可取消)。"""
rules = []
for i in range(self.list_widget.count()):
item = self.list_widget.item(i)
value = str(item.text()).strip()
if value:
rules.append({"value": value, "enabled": item.checkState() == Qt.Checked})
_whitelist_save(rules)
self.accept()
# ── 设置弹窗 ─────────────────────────────────────────
class SettingsDialog(QDialog):
@@ -1437,7 +1734,7 @@ class SettingsDialog(QDialog):
super().__init__(parent)
self.setWindowTitle("设置")
self.setWindowIcon(load_icon())
self.setFixedSize(300, 316)
self.setFixedSize(300, 354)
self.setStyleSheet(f"""
QDialog {{ background-color: {C_BG}; }}
QLabel {{ color: {C_TEXT}; font-family: "Microsoft YaHei", "Segoe UI", sans-serif; font-size: 12px; }}
@@ -1445,7 +1742,7 @@ class SettingsDialog(QDialog):
background-color: {C_CARD};
border: 1px solid {C_BORDER};
border-radius: 6px;
padding: 5px 8px;
padding: 2px 8px;
color: {C_TEXT};
font-size: 12px;
}}
@@ -1485,6 +1782,27 @@ class SettingsDialog(QDialog):
layout.addStretch() # 设置项在空白部分纵向居中(上下均分)
# 代理白名单:第一个设置项,点「配置」打开编辑弹窗
wl_row = QHBoxLayout()
wl_lab = QLabel("代理白名单")
wl_lab.setStyleSheet(f"color: {C_SUBTEXT}; font-size: 12px;")
wl_lab.setToolTip("命中列表的域名 / IP 将强制直连(不经过代理),每条规则可单独启用")
wl_row.addWidget(wl_lab)
wl_row.addStretch()
self.btn_whitelist = QPushButton("配置")
self.btn_whitelist.setCursor(Qt.PointingHandCursor)
self.btn_whitelist.setStyleSheet(f"""
QPushButton {{
background-color: {C_CARD}; color: {C_SUBTEXT};
border: 1px solid {C_BORDER}; border-radius: 6px;
padding: 2px 10px; font-size: 11px;
}}
QPushButton:hover {{ background-color: #e3e8ee; color: {C_TEXT}; }}
""")
self.btn_whitelist.clicked.connect(self._open_whitelist)
wl_row.addWidget(self.btn_whitelist)
layout.addLayout(wl_row)
# 代理转发端口
port_row = QHBoxLayout()
port_lab = QLabel("代理转发端口")
@@ -1492,7 +1810,7 @@ class SettingsDialog(QDialog):
port_row.addWidget(port_lab)
port_row.addStretch()
self.ed_port = QLineEdit(str(_TLS_LOCAL_PORT))
self.ed_port.setFixedWidth(80)
self.ed_port.setFixedSize(54, 22) # 与右侧开关同高,和其他设置项同一水平
self.ed_port.setEnabled(not port_locked)
port_row.addWidget(self.ed_port)
layout.addLayout(port_row)
@@ -1573,6 +1891,10 @@ class SettingsDialog(QDialog):
except Exception:
pass
def _open_whitelist(self):
"""打开代理白名单编辑弹窗(保存后立即生效,无需重启代理)。"""
WhitelistDialog(self).exec_()
def _save_settings(self):
"""保存更改:统一校验并应用全部设置项(后续新增设置项在此扩展)。"""
port = _parse_port(self.ed_port.text())
@@ -2176,11 +2498,10 @@ class MainWindow(QWidget):
if _skip_update_check() and not manual:
return # 设置跳过(仅自动检查):不检查也不打日志
import json, urllib.request
# 基于 Gitea 仓库 releases API 检查更新;config.json 无 update_url 时用默认地址
# 基于 Gitea 仓库 releases API 检查更新;地址只取 config.json 的 update_url
url = str(_config_val("update_url", "") or "").strip()
if not url:
url = ("https://git.bu4.ren/api/v1/repos/RoyirinRosmontis/"
"Private-Proxy/releases/latest")
raise RuntimeError("config.json 未配置 update_url")
log("正在检查更新...")
if _update_via_proxy():
# 走节点隧道(CONNECT,设置「使用代理更新」开启时);显式代理避免残留干扰