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

上传文件至「lite」

Browse Source
This commit is contained in:
RoyirinRosmontis
2026-08-23 02:05:03 +08:00
parent 1a891a1ad2
commit ab7c6cc87a
3 changed files with 3958 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lite/proxy_lite.py
+130 -65
View File
@@ -1,12 +1,8 @@
# -*- coding: utf-8 -*-
import sys, os, subprocess, time, ctypes, threading, functools, ssl as _ssl, base64, bisect
import sys, os, subprocess, time, ctypes, threading, functools, base64, bisect
from pathlib import Path
# 私有 HTTPS 端点通常使用自签名/内部证书，创建免校验 SSL 上下文
# 用于：更新检查等程序内部 HTTP 请求
_SSL_NO_VERIFY = _ssl._create_unverified_context()
from PyQt5.QtWidgets import (
QApplication, QWidget, QLabel, QPushButton,
QVBoxLayout, QHBoxLayout, QFrame, QMessageBox, QTextEdit,
@@ -15,13 +11,6 @@ from PyQt5.QtWidgets import (
from PyQt5.QtCore import Qt, QTimer, pyqtSignal
from PyQt5.QtGui import QFont, QIcon, QPixmap, QImage, QColor
VERSION = "999908162319"
UPDATE_USER = "user"
UPDATE_PASSWORD = "ftA406u@all#2309"
# 与 nginx 正向代理使用同一套 Basic 认证凭证
_PROXY_AUTH_BASIC = base64.b64encode(f"{UPDATE_USER}:{UPDATE_PASSWORD}".encode()).decode()
# GUI 窗口在 96 DPI / 100% 缩放下的逻辑尺寸；启用 Qt 高分屏缩放后，
# Qt 会自动按 devicePixelRatio 放大/缩小，因此代码里保持逻辑尺寸不变即可。
_BASE_WINDOW_WIDTH = 340
@@ -51,22 +40,93 @@ else:
def _endpoint_domain():
return "iam.aeolianchi.me"
def _endpoint_port(default=443):
return 443
# ── 普通设置（config.json）────────────────────────────
# 版本号、更新链接等设置，缺失时用默认值兜底
_CONFIG = None
def _load_config():
"""加载 config.json；失败或缺失返回空字典（各设置用默认值兜底）。"""
global _CONFIG
if _CONFIG is not None:
return _CONFIG
_CONFIG = {}
try:
import json
path = _data_path("config.json")
if path is None:
log("config.json 缺失，使用默认设置")
return _CONFIG
data = json.loads(Path(path).read_text("utf-8"))
if not isinstance(data, dict):
raise ValueError("config.json 顶层应为对象")
_CONFIG = data
log(f"已加载普通设置: {len(_CONFIG)} 项 (config.json)")
except Exception as e:
log_red(f"config.json 加载失败: {e}")
return _CONFIG
def _config_val(key, default=""):
return _load_config().get(key, default)
# ── 节点配置（nodes.json）────────────────────────────
# 节点列表，当前固定取第一个（后续可扩展选择）
_NODES = None
def _load_nodes():
"""加载 nodes.json 节点列表；失败或缺失时返回空列表。"""
global _NODES
if _NODES is not None:
return _NODES
_NODES = []
try:
import json
path = _data_path("nodes.json")
if path is None:
log_red("nodes.json 缺失，无法加载节点配置")
return _NODES
data = json.loads(Path(path).read_text("utf-8"))
if not isinstance(data, list):
raise ValueError("nodes.json 顶层应为节点数组")
_NODES = data
log(f"已加载节点配置: {len(_NODES)} 个 (nodes.json)")
except Exception as e:
log_red(f"nodes.json 加载失败: {e}")
return _NODES
def _current_node():
"""当前使用的节点：默认取列表第一个；无节点时返回 None。"""
nodes = _load_nodes()
return nodes[0] if nodes else None
def _node_val(key, default=""):
node = _current_node()
return node.get(key, default) if node else default
def _endpoint_domain():
return _node_val("domain", "???")
def _endpoint_port():
return int(_node_val("port", 443))
def _endpoint_country():
return _node_val("country", "")
def _proxy_auth_basic():
"""与 nginx 正向代理使用同一套 Basic 认证凭证（取自 nodes.json）。"""
node = _current_node()
if not node:
return ""
return base64.b64encode(f"{node.get('username', '')}:{node.get('password', '')}".encode()).decode()
_TLS_LOCAL_PORT = 1080
_tls_forwarder = None
_pac_server = None
PROXY_HOST = "127.0.0.1"
PROXY_PORT = _TLS_LOCAL_PORT
PROXY_REG = r"HKCU\Software\Microsoft\Windows\CurrentVersion\Internet Settings"
# _TLS_LOCAL_PORT / PROXY_PORT 定义在 _data_path 之后（取自 config.json local_port）
# PAC 模式使用全量转发 PAC：所有流量送进本地转发器，
# 由转发器按 IP 智能分流（中国 IP 直连、其余走加密隧道）。
# 域名表已提取到 pac_domains.json（强制走隧道的补充规则），不再内置。
# PAC 模式使用全量转发 PAC：所有流量送进本地转发器按 IP 智能分流
# （中国 IP 直连、其余走加密隧道）；域名表在 pac_domains.json。
_BUILT_IN_PAC = r'''function FindProxyForURL(url, host) {
var proxy = "PROXY 127.0.0.1:1080";
return proxy;
@@ -239,11 +299,11 @@ def proxy_off():
total = up + down
log(f"本次转发流量共 {_fmt_bytes(total)}")
# ── 本地认证转发器（直连 / WG 模式均使用） ─────────────────
# ── 本地认证转发器 ─────────────────────────────────────
# ── 流量统计 ──────────────────────────────────────────
# 所有代理流量都经过本地 127.0.0.1:1080 转发器（直连 / WG / PAC 均如此），
# 在转发器里计数即可覆盖全部模式。每次"开启代理"时清零，即"本次连接"流量。
# 所有代理流量都经过本地转发器（直连 / PAC 均如此），转发器里计数即可覆盖全部模式；
# 每次"开启代理"时清零，即"本次连接"流量。
_traffic_lock = threading.Lock()
_traffic = {"up": 0, "down": 0}
_traffic_prev = {"up": 0, "down": 0, "ts": 0.0}
@@ -290,17 +350,15 @@ def _traffic_snapshot():
return _traffic["up"], _traffic["down"]
def _tls_start(remote_host, remote_port=443, use_tls=True, bypass_lan=False, smart=False):
"""启动本地认证转发器：127.0.0.1:1080 -> remote_host:remote_port（可选 TLS 包裹）。
"""启动本地认证转发器：127.0.0.1:<local_port> -> remote_host:remote_port（可选 TLS 包裹）。
smart=True 时启用 IP 智能分流：pac_domains.json 命中或非中国 IP 走隧道，
其余（中国 IP）直连。
smart=True 时启用 IP 智能分流（域名表/非中国 IP 走隧道，中国 IP 直连）。
"""
global _tls_forwarder
_tls_stop()
import socket, ssl
# 服务器证书为 Let's Encrypt 正式证书，默认上下文完整校验
ctx = ssl.create_default_context()
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
svr = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
svr.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
@@ -345,7 +403,7 @@ def _inject_proxy_auth(data: bytes) -> bytes:
headers_lower = data[:end].lower()
if b"authorization:" in headers_lower:
return data
auth_line = f"Authorization: Basic {_PROXY_AUTH_BASIC}\r\n".encode()
auth_line = f"Authorization: Basic {_proxy_auth_basic()}\r\n".encode()
return data[:end + 2] + auth_line + data[end + 2:]
def _make_request_injector():
@@ -409,12 +467,14 @@ _FORCE_PROXY_SUFFIXES = None
_FORCE_PROXY_LOADED = False
def _data_path(name: str):
"""数据文件路径：EXE 同目录优先（可覆盖更新），其次打包内置目录。"""
for d in (PERSIST_DIR, BASE_DIR):
p = Path(d) / name
if p.is_file():
return p
return None
"""数据文件路径：只取打包内置目录（单文件 EXE，不依赖外部文件）。"""
p = Path(BASE_DIR) / name
return p if p.is_file() else None
# 以下模块级取值需在 _data_path 定义之后执行
VERSION = _config_val("version", "999908162319")
_TLS_LOCAL_PORT = int(_config_val("local_port", 1080)) # 本地转发器监听端口
PROXY_PORT = _TLS_LOCAL_PORT
def _load_cn_ranges():
"""加载 pac_ip_cn.json 中的中国 IPv4 段，合并为整数区间供二分查找。"""
@@ -528,12 +588,8 @@ def _is_force_proxy(host: str) -> bool:
def _decide_direct(host: str, bypass_lan: bool, smart: bool) -> bool:
"""判断连接是否直连。判定顺序（域名表优先于 IP）：
1. 智能分流：pac_domains.json 命中 -> 强制隧道；
2. 内网 IP 且勾选 bypass_lan -> 直连；
3. 智能分流：中国 IP -> 直连，其余 -> 隧道。
域名表与 IP 判定仅在智能分流（PAC）模式下生效；
固定代理模式下全部走隧道，也不触发数据文件加载。
域名表命中 -> 隧道；内网 + bypass_lan -> 直连；中国 IP -> 直连，其余 -> 隧道。
仅 PAC（智能分流）模式生效，固定代理模式全部走隧道。
"""
if smart and _is_force_proxy(host):
return False
@@ -728,7 +784,7 @@ def _cleanup_stale_proxy():
"""清理上次异常退出残留的系统代理设置。
TLS 转发器 / PAC 服务只活在进程内，程序启动时若注册表仍指向
127.0.0.1:1080 或本地 PAC 地址，说明是残留——不清理会导致全网不通。
本地转发端口或本地 PAC 地址，说明是残留——不清理会导致全网不通。
"""
try:
if not is_proxy_on():
@@ -800,8 +856,6 @@ class StatusDot(QLabel):
# ── 主窗口 ────────────────────────────────────────────
# ── 主窗口 ────────────────────────────────────────────
class MainWindow(QWidget):
log_signal = pyqtSignal(str)
_op_done = pyqtSignal(str, bool)
@@ -836,7 +890,7 @@ class MainWindow(QWidget):
def init_ui(self):
"""初始化主界面布局与样式。"""
self.setWindowTitle("Royirin Rosmontis Private Proxy")
self.setWindowTitle("R.R.P.P.")
self.setWindowFlags(self.windowFlags() & ~Qt.WindowContextHelpButtonHint)
self.setWindowIcon(load_icon())
self.setFixedSize(_BASE_WINDOW_WIDTH, _BASE_WINDOW_HEIGHT)
@@ -861,7 +915,7 @@ class MainWindow(QWidget):
layout.addWidget(title)
# 节点副标题
subtitle = QLabel(f"节点: {_endpoint_domain()} (JP)")
subtitle = QLabel(f"节点: {_endpoint_domain()} ({_endpoint_country()})")
subtitle.setStyleSheet(f"color: {C_TITLE}; font-size: 12px;")
subtitle.setAlignment(Qt.AlignCenter)
layout.addWidget(subtitle)
@@ -966,7 +1020,6 @@ class MainWindow(QWidget):
""")
layout.addWidget(self.log_text)
# 底部提示小字：独立容器收紧两行间距（主布局 spacing=14 太宽）
hints = QWidget()
hints_layout = QVBoxLayout(hints)
hints_layout.setContentsMargins(0, 0, 0, 0)
@@ -977,8 +1030,7 @@ class MainWindow(QWidget):
warn_label.setAlignment(Qt.AlignCenter)
hints_layout.addWidget(warn_label)
# 主背景是白色，纯白字不可见，用接近白的浅灰达到同样的弱化效果
privacy_label = QLabel("使用此软件将会收集并上报您的IP地址及部分设备信息")
privacy_label = QLabel("使用此软件将会收集并上报您的 IP 地址及部分设备信息")
privacy_label.setStyleSheet("color: #c8ccd4; font-size: 9px;")
privacy_label.setAlignment(Qt.AlignCenter)
hints_layout.addWidget(privacy_label)
@@ -1038,11 +1090,11 @@ class MainWindow(QWidget):
log("TLS 直连模式")
ep = _endpoint_domain()
log(f" 目标域名: {ep}")
# PAC 模式 = 智能分流：PAC 把全部流量送进转发器，由转发器按域名表/解析 IP 决定直连或隧道
# PAC 模式：全部流量进转发器，按域名表/解析 IP 智能分流
if use_pac:
_load_cn_ranges()
_load_force_proxy_suffixes()
_tls_start(ep, 443, use_tls=True, bypass_lan=use_bypass_lan, smart=use_pac)
_tls_start(ep, _endpoint_port(), use_tls=True, bypass_lan=use_bypass_lan, smart=use_pac)
if use_pac:
# 本地生成全量转发 PAC（不再从服务器拉取），全部流量交给转发器做智能分流
pac_content = _build_pac(f"{PROXY_HOST}:{PROXY_PORT}")
@@ -1082,7 +1134,7 @@ class MainWindow(QWidget):
threading.Thread(target=self._stop_thread, daemon=True).start()
def _stop_thread(self):
"""后台关闭代理：清理系统代理、停止 TLS/PAC 服务、断开 WG。"""
"""后台关闭代理：清理系统代理、停止 TLS/PAC 服务。"""
try:
log("正在关闭代理...")
proxy_off()
@@ -1104,8 +1156,18 @@ class MainWindow(QWidget):
self.cb_pac.setEnabled(is_off)
self.cb_bypass_lan.setEnabled(is_off)
# ── 日志 ──
_early_logs = [] # 窗口初始化期间（log_text 创建前）到达的日志缓存
def _append_log(self, msg: str):
"""追加日志并自动滚动，最多保留 200 行。"""
"""追加日志并自动滚动，最多保留 200 行；log_text 未创建时先缓存。"""
if not hasattr(self, "log_text"):
self._early_logs.append(msg)
return
for m in self._early_logs + [msg]:
self._append_log_line(m)
self._early_logs.clear()
def _append_log_line(self, msg: str):
self.log_text.append(msg)
scrollbar = self.log_text.verticalScrollBar()
scrollbar.setValue(scrollbar.maximum())
@@ -1122,7 +1184,7 @@ class MainWindow(QWidget):
if self._status_busy:
return
self._status_busy = True
# GUI 线程读取模式选择，传给后台线程；直连模式下不检测 WG，避免无谓的 wg 子进程
# GUI 线程读取模式选择，传给后台线程
threading.Thread(target=self._poll_status, daemon=True).start()
def _poll_status(self):
@@ -1161,16 +1223,19 @@ class MainWindow(QWidget):
try:
import json, urllib.request, base64
domain = _endpoint_domain()
if domain == "???":
log("更新检查: 无法获取服务器域名")
self._update_done.emit()
return
# 更新链接优先取 config.json 的 update_url，缺失时回退到域名拼接
url = _config_val("update_url", "")
if not url:
if domain == "???":
log("更新检查: 无法获取服务器域名")
self._update_done.emit()
return
url = f"https://{domain}/latest_version.json"
log("正在检查更新...")
url = f"https://{domain}/latest_version.json"
req = urllib.request.Request(url)
auth = base64.b64encode(f"{UPDATE_USER}:{UPDATE_PASSWORD}".encode()).decode()
auth = _proxy_auth_basic()
req.add_header("Authorization", f"Basic {auth}")
data = urllib.request.urlopen(req, context=_SSL_NO_VERIFY, timeout=10).read().decode("utf-8-sig")
data = urllib.request.urlopen(req, timeout=10).read().decode("utf-8-sig")
info = json.loads(data)
log("客户端认证成功")
self._latest_version = info.get("latest_version", "")
@@ -1214,10 +1279,10 @@ class MainWindow(QWidget):
log(f"保存到: {save_path}")
req = urllib.request.Request(url)
auth = base64.b64encode(f"{UPDATE_USER}:{UPDATE_PASSWORD}".encode()).decode()
auth = _proxy_auth_basic()
req.add_header("Authorization", f"Basic {auth}")
with urllib.request.urlopen(req, context=_SSL_NO_VERIFY, timeout=120) as resp:
with urllib.request.urlopen(req, timeout=120) as resp:
data = resp.read()
if len(data) < 1024:
@@ -1300,7 +1365,7 @@ def main():
QMessageBox.information(None, "提示", "程序已在运行中。")
sys.exit(0)
# 上次异常退出可能残留指向 127.0.0.1:1080 的系统代理（转发器已随进程消失），
# 上次异常退出可能残留指向本地转发端口的系统代理（转发器已随进程消失），
# 不清理会导致全网不通，启动时先清掉
_cleanup_stale_proxy()