RoyirinRosmontis/Private-Proxy
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 21 Wiki Activity

feat(lite): 持久化代理模式配置并发布 9999091816

Browse Source
This commit is contained in:
RoyirinRosmontis
2026-09-18 16:23:40 +08:00
parent 8063b6d4e8
commit 71a67d7dca
3 changed files with 438 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lite/main.py
+259 -43
View File
@@ -15,7 +15,8 @@ from rrpp_ui.theme import C_TITLE, C_GREEN, C_YELLOW, C_RED, C_BG, C_CARD, C_BOR
from rrpp_ui.widgets import NodeLabel, StatusDot, ToggleSwitch
from rrpp_ui.dialogs import (
LoginDialog, LoginServices, SettingsDialog, SettingsServices, ThirdPartyDialog, ThirdPartyServices,
WhitelistDialog, WhitelistServices, JumpDialog, JumpServices,
WhitelistDialog, WhitelistServices, FixedRouteDialog, FixedRouteServices,
JumpDialog, JumpServices,
)
from rrpp_core.routing_primitives import (
build_suffix_trie as _build_suffix_trie,
@@ -815,11 +816,15 @@ def _endpoint_label():
return "无可用节点(请登录)"
return f"{domain} ({_endpoint_country()})"
def _proxy_auth_basic():
def _proxy_auth_basic(node=None):
"""正向代理 Basic 认证凭证：methods.http.{username,password}（新格式）
或节点顶层同名字段（旧格式）；用于转发器向上游代理注入 Authorization。"""
_, get = _http_method()
return base64.b64encode(f"{get('username')}:{get('password')}".encode()).decode()
if node is None:
_, get = _http_method()
username, password = get("username"), get("password")
else:
username, password = _node_proxy_credentials(node)
return base64.b64encode(f"{username}:{password}".encode()).decode()
_tls_forwarder = None
_pac_server = None
@@ -1189,7 +1194,7 @@ def _tls_start(remote_host, remote_port=443, use_tls=True, bypass_lan=False, sma
suffix = f" [{route}]" if route else ""
log(f"本地转发器已启动 ({proto} 127.0.0.1:{_TLS_LOCAL_PORT} -> {remote_host}:{remote_port}{suffix})")
def _inject_proxy_auth(data: bytes) -> bytes:
def _inject_proxy_auth(data: bytes, proxy_node=None) -> bytes:
"""在 HTTP 代理请求头中注入 Authorization（如果尚未存在）。
nginx 的 auth_basic 校验 Authorization 而非 Proxy-Authorization，
@@ -1199,12 +1204,12 @@ def _inject_proxy_auth(data: bytes) -> bytes:
Authorization（明文 HTTP 请求）才跳过。注入位置紧跟请求行，
保证上游（取首个 Authorization）读到的是我们的认证。
"""
node, _ = _http_method()
node = proxy_node or _http_method()[0]
if not any(_node_proxy_credentials(node)):
return data
return _inject_proxy_auth_value(data, _proxy_auth_basic())
return _inject_proxy_auth_value(data, _proxy_auth_basic(node))
def _make_request_injector():
def _make_request_injector(proxy_node=None):
"""为明文 HTTP 代理连接生成逐条注入认证的流式处理器。
keep-alive 连接上的后续请求同样需要 Authorization，不能只在首包注入。
@@ -1220,7 +1225,7 @@ def _make_request_injector():
if not _starts_with_method(data):
return data
if b"\r\n\r\n" in data or len(data) >= 65536:
return _inject_proxy_auth(data)
return _inject_proxy_auth(data, proxy_node)
pending = data
return b""
return feed
@@ -1339,38 +1344,187 @@ 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) 键控的匹配缓存失效。
def _use_pac_mode():
"""是否默认启用 PAC 智能分流；缺失时保持原有默认值 True。"""
return bool(_settings_load().get("pac_mode", True))
def _set_use_pac_mode(enabled):
"""保存 PAC 模式偏好，下一次启动时恢复。"""
data = _settings_load()
data["pac_mode"] = bool(enabled)
_settings_save(data)
def _bypass_lan():
"""是否默认绕过内网 IP；缺失时保持原有默认值 True。"""
return bool(_settings_load().get("bypass_lan", True))
def _set_bypass_lan(enabled):
"""保存内网 IP 直连偏好，下一次启动时恢复。"""
data = _settings_load()
data["bypass_lan"] = bool(enabled)
_settings_save(data)
# ── 域名路由（兼容旧代理白名单，settings.json 的 whitelist 键） ────────
# 每条规则 {"value": 域名/IP/CIDR, "enabled": 是否启用,
# "action": "direct" | "node", "node_ref": 节点稳定标识}。
# action=direct 为强制不代理，action=node 为强制走指定节点；旧版只有 value/enabled
# 的白名单自动按 direct 解释。规则保存在内存缓存，修改后立即影响新连接。
_WHITELIST = None # 内存缓存：规范化后的规则列表
_WHITELIST_VERSION = 0 # 规则版本号（保存时 +1，用于匹配缓存失效）
_ROUTE_RULES = None # 全量域名路由规则（含 direct/node）
def _whitelist_rules():
"""白名单规则列表（内存缓存，首次读取 settings.json；保存时同步更新）。"""
global _WHITELIST
if _WHITELIST is None:
def _route_domain_normalize(entry: str) -> str:
"""规范化域名规则，保留 ``*.`` 通配符，兼容旧白名单输入。"""
s = str(entry or "").strip().lower()
if "://" in s:
s = s.split("://", 1)[1]
s = s.split("/", 1)[0].rstrip(".")
wildcard = s.startswith("*.")
if wildcard:
s = s[2:]
return ("*." if wildcard else "") + s
def _route_rule_match(host: str, entry: str) -> bool:
"""匹配域名路由：普通域名兼容旧版的根域+子域语义，``*.`` 仅匹配子域。"""
value = _route_domain_normalize(entry)
h = str(host or "").rstrip(".").lower()
if value.startswith("*."):
base = value[2:]
return bool(base and h.endswith("." + base) and h != base)
return _whitelist_entry_match(h, value)
def _route_node_reference(node):
"""固定节点规则使用稳定引用；跳板引用使用 profile ID，普通节点忽略备注变化。"""
if isinstance(node, dict) and node.get("_rrpp_source_kind") == "jump":
return _node_reference(node)
return _jump_node_reference(node)
def _route_node_label(node):
if node.get("_rrpp_source_kind") == "jump":
return f"{node.get('_jump_name') or '自定义跳板'} ({_jump_route_text(node)})"
return f"{node.get('domain') or '???'} ({node.get('country') or '?'})"
def _route_rules():
"""读取并规范化全量路由规则；旧版 whitelist 自动迁移为强制直连。"""
global _ROUTE_RULES
if _ROUTE_RULES 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
result = []
for item in raw:
if not isinstance(item, dict):
continue
value = _route_domain_normalize(item.get("domain") or item.get("value") or "")
if not value:
continue
action = str(item.get("action") or ("node" if item.get("node_ref") else "direct")).lower()
if action not in ("direct", "node"):
action = "direct"
result.append({
"domain": value,
"enabled": bool(item.get("enabled", True)),
"action": action,
"node_ref": str(item.get("node_ref") or ""),
})
_ROUTE_RULES = result
return _ROUTE_RULES
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 "")]
def _route_rule_for_host(host: str):
"""返回目标域名命中的最高优先级规则；规则变更对新连接立即生效。"""
matches = []
for rule in _route_rules():
if not rule.get("enabled") or not _route_rule_match(host, rule.get("domain", "")):
continue
domain = rule["domain"]
base = domain[2:] if domain.startswith("*.") else domain
# 更长域名优先；同一根域下显式通配符优先于兼容旧语义的普通域名。
matches.append((len(base), int(domain.startswith("*.")), rule))
return max(matches, key=lambda item: (item[0], item[1]))[2] if matches else None
def _fixed_route_nodes():
"""返回可被域名规则选择的当前普通节点与多跳虚拟节点。"""
candidates = [node for node in _load_nodes() if _node_reachable(node)]
counts = {}
for node in candidates:
ref = _route_node_reference(node)
counts[ref] = counts.get(ref, 0) + 1
return [node for node in candidates if counts[_route_node_reference(node)] == 1]
def _fixed_route_node(rule):
"""按规则解析固定节点；节点下线或引用失效时返回 None。"""
ref = str((rule or {}).get("node_ref") or "")
if not ref:
return None
candidates = [node for node in _load_nodes() if _route_node_reference(node) == ref]
return candidates[0] if len(candidates) == 1 and _node_reachable(candidates[0]) else None
def _fixed_route_rules():
"""供设置界面展示规则及节点可用状态，不暴露节点认证信息。"""
nodes = {_route_node_reference(node): node for node in _fixed_route_nodes()}
result = []
for rule in _route_rules():
item = dict(rule)
node = nodes.get(rule.get("node_ref")) if rule.get("action") == "node" else None
item["node_label"] = _route_node_label(node) if node else ("节点不可用" if rule.get("action") == "node" else "强制不代理")
result.append(item)
return result
def _fixed_route_save(rules):
"""保存域名路由规则；保留旧 whitelist 键名以兼容已有用户设置。"""
global _ROUTE_RULES, _WHITELIST, _WHITELIST_VERSION
clean = []
for item in rules or []:
if not isinstance(item, dict):
continue
value = _route_domain_normalize(item.get("domain") or item.get("value") or "")
if not value:
continue
action = str(item.get("action") or "direct").lower()
if action not in ("direct", "node"):
action = "direct"
clean.append({
"value": value,
"enabled": bool(item.get("enabled", True)),
"action": action,
"node_ref": str(item.get("node_ref") or "") if action == "node" else "",
})
data = _settings_load()
data["whitelist"] = clean
_settings_save(data)
_WHITELIST = clean
_ROUTE_RULES = None
_WHITELIST = None
_WHITELIST_VERSION += 1
log(f"代理白名单已保存: {len(clean)} 条（立即生效）")
_is_whitelisted.cache_clear()
log(f"域名路由已保存: {len(clean)} 条（立即生效）")
def _whitelist_rules():
"""兼容旧调用方：只返回强制不代理规则。"""
global _WHITELIST
if _WHITELIST is None:
_WHITELIST = [{"value": r["domain"], "enabled": r["enabled"]}
for r in _route_rules() if r.get("action") == "direct"]
return _WHITELIST
def _whitelist_save(rules):
"""旧白名单保存入口：把传入规则按强制不代理处理。"""
_fixed_route_save([{"domain": r.get("value"), "enabled": r.get("enabled", True), "action": "direct"}
for r in rules or []])
def _whitelist_normalize(entry: str) -> str:
"""条目规范化：去协议头/通配符/末尾点，统一小写。"""
@@ -1407,6 +1561,23 @@ def _whitelist_validate_entry(entry: str):
return "", "IP 格式不正确"
return s, ""
def _fixed_route_validate_domain(entry: str):
"""校验域名路由规则；支持 ``example.com`` 与 ``*.example.com``。"""
value = _route_domain_normalize(entry)
base = value[2:] if value.startswith("*.") else value
if not base:
return "", "域名不能为空"
if any(ch.isspace() for ch in base) or "/" in base or ":" in base:
return "", "域名格式不正确"
labels = base.split(".")
if len(labels) < 2 or any(not label or label.startswith("-") or label.endswith("-")
for label in labels):
return "", "请输入完整域名，例如 example.com"
if any(not all(ch.isalnum() or ch == "-" for ch in label) for label in labels):
return "", "域名只能包含字母、数字、短横线和点"
return value, ""
def _ver_tuple(v):
"""版本号转数字元组：'v0.1.0' -> (0, 1, 0)，无法解析的段按 0 处理。"""
parts = []
@@ -2028,6 +2199,11 @@ def _tls_relay(client, remote_host, remote_port, ctx, use_tls, bypass_lan=False,
injector = None
direct_target = None
allow_direct_fallback = False
forced_rule = None
forced_node = None
route_remote_host = remote_host
route_remote_port = remote_port
route_jump_chain = jump_chain
max_headers = 65536
try:
while len(buf) < max_headers:
@@ -2055,10 +2231,27 @@ def _tls_relay(client, remote_host, remote_port, ctx, use_tls, bypass_lan=False,
else:
target = _parse_http_absolute(first_line)
if target:
direct_target = _direct_connect_target(
target[0], target[1], bypass_lan, smart)
allow_direct_fallback = (direct_target is not None and
_is_smart_china_direct(target[0], bypass_lan, smart))
forced_rule = _route_rule_for_host(target[0])
if forced_rule and forced_rule.get("action") == "direct":
# 显式强制不代理优先于 PAC 的域名表和 IP 判定。
direct_target = (target[0], target[1])
elif forced_rule and forced_rule.get("action") == "node":
forced_node = _fixed_route_node(forced_rule)
if forced_node is None:
raise RuntimeError(
f"域名 {target[0]} 指定的固定节点不可用，请在设置中更新域名路由"
)
terminal = _terminal_node(forced_node)
route_remote_host = str(terminal.get("domain") or "")
route_remote_port = _node_http_port(terminal)
route_jump_chain = _jump_chain(forced_node)
# 固定节点规则必须走指定节点，不允许 PAC 把目标改为直连。
direct_target = None
else:
direct_target = _direct_connect_target(
target[0], target[1], bypass_lan, smart)
allow_direct_fallback = (direct_target is not None and
_is_smart_china_direct(target[0], bypass_lan, smart))
except Exception as e:
log(f"[转发器] 读取首包失败: {e}")
client.close()
@@ -2067,8 +2260,8 @@ def _tls_relay(client, remote_host, remote_port, ctx, use_tls, bypass_lan=False,
remote = None
try:
remote, via_tunnel = _connect_route(
direct_target, allow_direct_fallback, remote_host, remote_port, ctx, use_tls,
jump_chain=jump_chain)
direct_target, allow_direct_fallback, route_remote_host, route_remote_port, ctx, use_tls,
jump_chain=route_jump_chain)
count_traffic = via_tunnel
if not via_tunnel:
if is_connect:
@@ -2079,8 +2272,10 @@ def _tls_relay(client, remote_host, remote_port, ctx, use_tls, bypass_lan=False,
else:
if is_http and b"\r\n\r\n" in buf:
if not is_connect:
injector = _make_request_injector()
buf = _inject_proxy_auth(buf)
injector = _make_request_injector(_terminal_node(forced_node) if forced_node else None)
buf = _inject_proxy_auth(
buf, _terminal_node(forced_node) if forced_node else None
)
if buf:
remote.sendall(buf)
if via_tunnel:
@@ -2239,9 +2434,28 @@ def _login_services():
def _whitelist_services():
return WhitelistServices(
load_icon=load_icon,
load_rules=_whitelist_rules,
validate_rule=_whitelist_validate_entry,
save_rules=_whitelist_save,
load_rules=lambda: [
{"value": item["domain"], "enabled": item["enabled"]}
for item in _fixed_route_rules() if item.get("action") == "direct"
],
validate_rule=_fixed_route_validate_domain,
save_rules=lambda rules: _fixed_route_save([
{"domain": item.get("value"), "enabled": item.get("enabled", True), "action": "direct"}
for item in rules
]),
)
def _fixed_route_services():
return FixedRouteServices(
load_icon=load_icon,
load_rules=_fixed_route_rules,
available_nodes=lambda: [
{"ref": _route_node_reference(node), "label": _route_node_label(node)}
for node in _fixed_route_nodes()
],
validate_domain=_fixed_route_validate_domain,
save_rules=_fixed_route_save,
)
@@ -2287,7 +2501,7 @@ def _settings_services():
set_port=_set_local_port,
set_update_via_proxy=_set_update_via_proxy,
set_skip_update_check=_set_skip_update_check,
open_whitelist=lambda parent: WhitelistDialog(_whitelist_services(), parent).exec_(),
open_whitelist=lambda parent: FixedRouteDialog(_fixed_route_services(), parent).exec_(),
open_jump=lambda parent: JumpDialog(_jump_services(), parent).exec_(),
open_third_party=lambda parent: ThirdPartyDialog(_third_party_services(), parent).exec_(),
has_login=lambda: _auth_load() is not None,
@@ -2423,13 +2637,15 @@ class MainWindow(QWidget):
pac_layout.setSpacing(8)
pac_layout.addStretch()
self.cb_bypass_lan = QCheckBox("不代理内网 IP")
self.cb_bypass_lan.setChecked(True) # 默认勾选：内网 IP 直连，不经过本地转发器
self.cb_bypass_lan.setChecked(_bypass_lan())
self.cb_bypass_lan.setStyleSheet(f"color: {C_SUBTEXT}; font-size: 12px;")
self.cb_bypass_lan.toggled.connect(_set_bypass_lan)
pac_layout.addWidget(self.cb_bypass_lan)
self.cb_pac = QCheckBox("PAC 规则模式")
self.cb_pac.setChecked(True) # 默认开启智能分流（域名表/中国IP判定已足够完善）
self.cb_pac.setChecked(_use_pac_mode())
self.cb_pac.setStyleSheet(f"color: {C_SUBTEXT}; font-size: 12px;")
self.cb_pac.setToolTip("只代理海外服务，使用域名和解析 IP 匹配。")
self.cb_pac.toggled.connect(_set_use_pac_mode)
pac_layout.addWidget(self.cb_pac)
pac_layout.addStretch()
layout.addLayout(pac_layout)