// kernel_process.dart — 桌面端内核子进程管理(tsk_SLCsjNgtmng3 实现版) // // 实现说明: // · 进程管理: dart:io Process.start // · macOS PoC 提权: sudo(开发机需有 sudo 权限 / sudoers 白名单) // 正式版替换为 SMJobBless 特权 Helper + 公证 (BACKLOG-11D-HELPER) // · Clash API: http 包 HTTP 轮询;readiness 探测 GET /connections; // 统计通过 /connections 拿累计字节,差分算瞬时速率 // · 二进制解析优先级(见 _resolveBinaryPath): // 1. 环境变量 PANGOLIN_SINGBOX_BIN // 2. /sing-box (或 .exe) // 3. /../../Resources/sing-box (macOS .app bundle) // 4. /app/kernel/dist/desktop/-/sing-box (开发目录) // 5. /opt/homebrew/bin/sing-box (Apple Silicon brew) // 6. /usr/local/bin/sing-box (Intel brew / Linux) // · 配置 Clash API 端口: 由 spawn 调用方(DesktopVpnBridge)写入配置, // 或从配置文件读 experimental.clash_api.external_controller // // 生命周期: // spawn(configPath) → 起子进程 → 等 Clash API 就绪 → emit connecting→on // kill() → SIGTERM → 等 ≤gracePeriod → SIGKILL → emit off // 意外退出 → emit error(UI 可一键重连) // // ignore_for_file: avoid_print import 'dart:async'; import 'dart:convert'; import 'dart:io'; import 'dart:math' as math; import 'package:http/http.dart' as http; import 'log.dart'; import 'vpn_bridge.dart'; // ══════════════════════════════════════════════════════════════════ // Clash API 客户端 // ══════════════════════════════════════════════════════════════════ /// sing-box Clash 兼容 REST API 客户端。 /// /// baseUrl: http://127.0.0.1: /// secret : experimental.clash_api.secret 的值(Bearer Token) class ClashApiClient { ClashApiClient({ required this.baseUrl, this.secret = '', http.Client? httpClient, }) : _http = httpClient ?? http.Client(); final String baseUrl; final String secret; final http.Client _http; Map get _headers => { if (secret.isNotEmpty) 'Authorization': 'Bearer $secret', 'Content-Type': 'application/json', }; // ── GET /traffic ────────────────────────────────────────────── // Clash /traffic 是 SSE 端点,推 {"up": N, "down": N}(bytes/s)。 // 此实现读首帧后断开,供就绪探测或一次性快照使用。 // 持续统计用 getConnections()(普通 JSON,含 downloadTotal/uploadTotal)。 Future> getTraffic() async { final uri = Uri.parse('$baseUrl/traffic'); final request = http.Request('GET', uri); request.headers.addAll(_headers); request.headers['Accept'] = 'text/event-stream'; late http.StreamedResponse streamed; try { streamed = await _http.send(request).timeout(const Duration(seconds: 3)); } on TimeoutException { throw const HttpException('timeout connecting to /traffic'); } if (streamed.statusCode != 200) { // 读掉 body,释放连接 await streamed.stream.drain(); throw HttpException('traffic: ${streamed.statusCode}'); } final completer = Completer>(); StreamSubscription? sub; sub = streamed.stream .transform(utf8.decoder) .transform(const LineSplitter()) .listen( (line) { if (line.startsWith('data:')) { final payload = line.substring(5).trim(); if (payload.isNotEmpty && !completer.isCompleted) { try { completer.complete(jsonDecode(payload) as Map); } catch (e) { if (!completer.isCompleted) completer.completeError(e); } sub?.cancel(); } } }, onError: (Object e) { if (!completer.isCompleted) completer.completeError(e); }, onDone: () { if (!completer.isCompleted) { completer.completeError( const HttpException('SSE stream ended without data'), ); } }, cancelOnError: true, ); return completer.future.timeout( const Duration(seconds: 5), onTimeout: () { sub?.cancel(); throw TimeoutException('/traffic SSE timeout'); }, ); } // ── GET /proxies ────────────────────────────────────────────── Future> getProxies() async { final response = await _http .get(Uri.parse('$baseUrl/proxies'), headers: _headers) .timeout(const Duration(seconds: 3)); if (response.statusCode != 200) { throw HttpException('proxies: ${response.statusCode}'); } return jsonDecode(response.body) as Map; } // ── PUT /proxies/{group} ────────────────────────────────────── Future selectProxy(String group, String proxy) async { final body = jsonEncode({'name': proxy}); final response = await _http .put(Uri.parse('$baseUrl/proxies/$group'), headers: _headers, body: body) .timeout(const Duration(seconds: 3)); if (response.statusCode != 204 && response.statusCode != 200) { throw HttpException('selectProxy: ${response.statusCode}'); } } // ── GET /connections ────────────────────────────────────────── // 返回 {"downloadTotal": N, "uploadTotal": N, "connections": [...]} // 累计字节;用于统计轮询和就绪探测。 Future> getConnections() async { final response = await _http .get(Uri.parse('$baseUrl/connections'), headers: _headers) .timeout(const Duration(seconds: 3)); if (response.statusCode != 200) { throw HttpException('connections: ${response.statusCode}'); } return jsonDecode(response.body) as Map; } // ── DELETE /connections ─────────────────────────────────────── Future closeAllConnections() async { final response = await _http .delete(Uri.parse('$baseUrl/connections'), headers: _headers) .timeout(const Duration(seconds: 3)); if (response.statusCode != 204 && response.statusCode != 200) { throw HttpException('closeConnections: ${response.statusCode}'); } } // ── GET /group//delay ─────────────────────────────────── // 触发一次 URLTest 延迟测试并返回结果 {tag: delayMs}。 // 注意:此接口会发起真实网络探测,耗时约 timeoutMs。 // 若只需读缓存延迟,使用 getProxies() + extractUrltestResults()。 Future> getGroupDelay( String groupName, { String testUrl = 'http://www.gstatic.com/generate_204', int timeoutMs = 3000, }) async { final uri = Uri.parse('$baseUrl/group/$groupName/delay').replace( queryParameters: { 'url': testUrl, 'timeout': timeoutMs.toString(), }, ); final response = await _http .get(uri, headers: _headers) .timeout(Duration(milliseconds: timeoutMs + 2000)); if (response.statusCode != 200) { throw HttpException('getGroupDelay($groupName): ${response.statusCode}'); } final data = jsonDecode(response.body); if (data is! Map) return {}; return data.cast(); } // ── 工具:从 /proxies 响应提取 URLTest/Fallback 组名 ────────── static List extractUrltestGroups(Map proxiesResponse) { final proxies = proxiesResponse['proxies']; if (proxies is! Map) return const []; final groups = []; for (final entry in proxies.entries) { final p = entry.value; if (p is! Map) continue; final type = (p['type'] as String?) ?? ''; if (type == 'URLTest' || type == 'Fallback') { if (entry.key is String) groups.add(entry.key as String); } } return groups; } // ── 工具:从 /proxies 响应提取 URLTest 延迟列表 ────────────── // 扫描所有类型为 URLTest 的出口组,提取成员节点的最新 history delay。 static List extractUrltestResults( Map proxiesResponse) { final proxies = proxiesResponse['proxies']; if (proxies is! Map) return const []; final results = []; for (final entry in proxies.entries) { final p = entry.value; if (p is! Map) continue; final type = (p['type'] as String?) ?? ''; // URLTest / Fallback 组都有 all 成员列表 if (type != 'URLTest' && type != 'Fallback') continue; final all = p['all']; if (all is! List) continue; for (final memberTag in all) { if (memberTag is! String) continue; final member = proxies[memberTag]; if (member is! Map) continue; final history = member['history']; if (history is List && history.isNotEmpty) { final last = history.last; if (last is Map) { final delay = (last['delay'] as num?)?.toInt() ?? -1; results.add(UrltestResult(tag: memberTag, delayMs: delay)); } } } } return results; } void dispose() => _http.close(); } // ══════════════════════════════════════════════════════════════════ // KernelProcess 接口 // ══════════════════════════════════════════════════════════════════ /// 桌面端内核子进程管理接口。 abstract class KernelProcess { /// 启动内核子进程。[configPath] 为 sing-box JSON 配置文件绝对路径。 /// 返回的 Future 在进程就绪(Clash REST API 可用)后 complete。 Future spawn(String configPath); /// 停止内核子进程。先 SIGTERM,[gracePeriod] 内未退出则 SIGKILL。 Future kill({Duration gracePeriod = const Duration(seconds: 5)}); /// 进程是否正在运行。 bool get isRunning; /// Clash 兼容 REST API 客户端(仅进程运行时有效)。 ClashApiClient get clashApiClient; /// 进程 stdout/stderr 实时日志流(调试面板用)。 Stream get logStream; /// VPN 状态事件流:connecting → on(就绪)/ off(停止)/ error(意外退出)。 Stream get statusStream; /// 每秒流量统计帧(字节累计 + 瞬时速率)。 Stream get statsStream; } // ══════════════════════════════════════════════════════════════════ // DesktopKernelProcess — 真实子进程实现 // ══════════════════════════════════════════════════════════════════ /// 桌面端内核子进程管理(sing-box)。 /// /// 除 [KernelProcess] 接口外,还额外暴露: /// · [statusStream] — VPN 状态事件(connecting / on / off / error) /// · [statsStream] — 每秒流量统计帧(字节数 + 瞬时速率) /// /// 用法(通常由 [DesktopVpnBridge] 驱动,不直接在 UI 使用): /// ```dart /// final kernel = DesktopKernelProcess(); /// await kernel.spawn('/tmp/pangolin/config.json'); /// // 收流量统计 /// kernel.statsStream.listen(print); /// // 停止 /// await kernel.kill(); /// ``` class DesktopKernelProcess implements KernelProcess { DesktopKernelProcess({ Duration readyTimeout = const Duration(seconds: 20), Duration statsPollInterval = const Duration(seconds: 1), this.useSudo = true, }) : _readyTimeout = readyTimeout, _statsPollInterval = statsPollInterval; /// macOS/Linux PoC 时是否用 sudo 启动(TUN 接口需 root)。 /// 正式版改用 SMJobBless Helper;Windows 直接以管理员启动。 final bool useSudo; final Duration _readyTimeout; final Duration _statsPollInterval; Process? _process; ClashApiClient? _clashApi; Timer? _statsTimer; bool _running = false; // 上一帧累计字节(用于计算瞬时速率) int _prevDownTotal = 0; int _prevUpTotal = 0; DateTime _prevPollTime = DateTime.now(); // 最新 URLTest 延迟列表(每次 stats 轮询时更新) List _lastUrltestResults = const []; // URLTest/Fallback 组名(从 /proxies 提取),供主动触发延迟探测。 List _urltestGroups = const []; // 轮询计数:每 _urlTestEvery 拍主动触发一次 group/delay(补偿 urltest 惰性探测)。 int _pollTick = 0; static const int _urlTestEvery = 12; final _statusCtrl = StreamController.broadcast(); final _logCtrl = StreamController.broadcast(); final _statsCtrl = StreamController.broadcast(); // ── 公开属性 ───────────────────────────────────────────────── @override bool get isRunning => _running; @override ClashApiClient get clashApiClient { if (_clashApi == null) throw StateError('kernel not running'); return _clashApi!; } @override Stream get logStream => _logCtrl.stream; /// VPN 状态事件流(connecting / on / off / error)。 @override Stream get statusStream => _statusCtrl.stream; /// 每秒流量统计帧。 @override Stream get statsStream => _statsCtrl.stream; // ── spawn ───────────────────────────────────────────────────── @override Future spawn(String configPath) async { if (_running) throw StateError('DesktopKernelProcess already running'); // 1. 解析二进制路径 final binPath = await _resolveBinaryPath(); _log('binary: $binPath'); // Windows: sing-box loads wintun.dll (from its own directory) to create the // TUN adapter. Warn early if it is missing so a failed connect is diagnosable // rather than surfacing as an opaque sing-box error. if (Platform.isWindows) { final wintun = File('${File(binPath).parent.path}\\wintun.dll'); if (!await wintun.exists()) { _log('[kernel] WARNING: wintun.dll not found next to sing-box.exe ' '(${wintun.path}); TUN will fail to start. Run ' 'app/kernel/fetch-desktop-bin.sh windows amd64.'); } } // 2. 读取配置,提取 Clash API 地址和 secret final clashEndpoint = await _parseClashEndpoint(configPath); final clashUrl = 'http://${clashEndpoint.$1}'; final clashSecret = clashEndpoint.$2; _log('clash_api: $clashUrl (secret=${clashSecret.isNotEmpty})'); // 3. 初始化 ClashApiClient(进程启动前先建好,便于后续就绪检测) _clashApi = ClashApiClient(baseUrl: clashUrl, secret: clashSecret); // 4. 发 connecting 事件 _emitStatus(VpnStatus.connecting); // 5. 构造命令并启动子进程 final cmd = _buildCommand(binPath, configPath); _log('spawn: ${cmd.join(' ')}'); try { _process = await Process.start( cmd.first, cmd.skip(1).toList(), environment: Platform.environment, // 不设 workingDirectory,让 sing-box 自行处理 ); } catch (e) { _clashApi!.dispose(); _clashApi = null; _emitStatus(VpnStatus.error); throw ProcessException(cmd.first, cmd.skip(1).toList(), 'failed to start: $e'); } _running = true; // 6. 挂接 stdout/stderr 日志 _process!.stdout .transform(utf8.decoder) .transform(const LineSplitter()) .listen((l) => _log('[stdout] $l')); _process!.stderr .transform(utf8.decoder) .transform(const LineSplitter()) .listen((l) => _log('[stderr] $l')); // 7. 监测意外退出 _process!.exitCode.then((code) { if (_running) { _running = false; _statsTimer?.cancel(); _statsTimer = null; // 清理 Clash API 客户端,确保下次 spawn() 重建干净的客户端 _clashApi?.dispose(); _clashApi = null; _log('[kernel] unexpected exit: code=$code'); _emitStatus(VpnStatus.error); } }); // 8. 等待 Clash API 就绪 try { await _waitForClashApi(); } catch (e) { // 超时或 API 不可用 await kill(gracePeriod: const Duration(seconds: 2)); rethrow; } // 9. 启动统计轮询 _startStatsPoll(); // 10. 发 on 事件 _emitStatus(VpnStatus.on); _log('[kernel] ready'); } // ── kill ────────────────────────────────────────────────────── @override Future kill({Duration gracePeriod = const Duration(seconds: 5)}) async { if (!_running && _process == null) return; _running = false; _statsTimer?.cancel(); _statsTimer = null; final proc = _process; _process = null; if (proc != null) { _log('[kernel] sending SIGTERM …'); proc.kill(ProcessSignal.sigterm); try { await proc.exitCode.timeout(gracePeriod); _log('[kernel] exited after SIGTERM'); } on TimeoutException { _log('[kernel] SIGTERM timeout, sending SIGKILL …'); proc.kill(ProcessSignal.sigkill); await proc.exitCode.timeout(const Duration(seconds: 3)).catchError((_) => -1); } } _clashApi?.dispose(); _clashApi = null; _resetStats(); _emitStatus(VpnStatus.off); _log('[kernel] stopped'); } // ── 释放资源 ───────────────────────────────────────────────── Future dispose() async { await kill(gracePeriod: const Duration(seconds: 3)); await _statusCtrl.close(); await _logCtrl.close(); await _statsCtrl.close(); } // ── 内部方法 ───────────────────────────────────────────────── void _emitStatus(VpnStatus s) { if (!_statusCtrl.isClosed) _statusCtrl.add(s); } void _log(String msg) { logLine('DesktopKernel', msg); if (!_logCtrl.isClosed) _logCtrl.add(msg); } void _resetStats() { _prevDownTotal = 0; _prevUpTotal = 0; _prevPollTime = DateTime.now(); _lastUrltestResults = const []; } // ── 就绪等待 ───────────────────────────────────────────────── Future _waitForClashApi() async { final deadline = DateTime.now().add(_readyTimeout); while (DateTime.now().isBefore(deadline)) { if (!_running) throw StateError('process exited during startup'); try { await _clashApi!.getConnections(); return; // API 已就绪 } catch (_) { await Future.delayed(const Duration(milliseconds: 250)); } } throw TimeoutException( 'Clash API not ready after ${_readyTimeout.inSeconds}s'); } // ── 统计轮询 ───────────────────────────────────────────────── void _startStatsPoll() { _resetStats(); _prevPollTime = DateTime.now(); _statsTimer = Timer.periodic(_statsPollInterval, (_) async { if (!_running || _statsCtrl.isClosed) return; try { final now = DateTime.now(); // 并行拉取连接统计与 URLTest 延迟(Best-effort:URLTest 失败不影响主统计) final connFuture = _clashApi!.getConnections(); final proxiesFuture = _clashApi!.getProxies().then((p) { _lastUrltestResults = ClashApiClient.extractUrltestResults(p); _urltestGroups = ClashApiClient.extractUrltestGroups(p); }).catchError((_) { // 读取失败静默跳过,保留上次缓存值 }); final data = await connFuture; await proxiesFuture; // 等 URLTest 解析完再组帧 // 周期主动触发 group/delay 探测:sing-box urltest 默认惰性探测, // 仅读 /proxies 缓存会长时间拿不到 history → 延迟空白。每 _urlTestEvery // 拍主动打一次,fire-and-forget(不阻塞主统计帧),下一拍即读到新 history。 if ((_pollTick++ % _urlTestEvery) == 0) { for (final g in _urltestGroups) { unawaited(_clashApi!.getGroupDelay(g).catchError( (_) => {})); } } final downTotal = (data['downloadTotal'] as num?)?.toInt() ?? _prevDownTotal; final upTotal = (data['uploadTotal'] as num?)?.toInt() ?? _prevUpTotal; final deltaSec = now.difference(_prevPollTime).inMilliseconds / 1000.0; final downSpeed = deltaSec > 0 ? (downTotal - _prevDownTotal) / deltaSec : 0.0; final upSpeed = deltaSec > 0 ? (upTotal - _prevUpTotal) / deltaSec : 0.0; _prevDownTotal = downTotal; _prevUpTotal = upTotal; _prevPollTime = now; if (!_statsCtrl.isClosed) { _statsCtrl.add(VpnStatsEvent( uploadBytes: upTotal, downloadBytes: downTotal, uploadSpeed: upSpeed.clamp(0, double.infinity), downloadSpeed: downSpeed.clamp(0, double.infinity), urltestResults: _lastUrltestResults, )); } } catch (_) { // 统计读取失败不影响连接状态;下次重试 } }); } // ── 二进制路径解析 ──────────────────────────────────────────── static Future _resolveBinaryPath() async { final envBin = Platform.environment['PANGOLIN_SINGBOX_BIN']; if (envBin != null && envBin.isNotEmpty) { final f = File(envBin); if (await f.exists()) return envBin; } final binName = Platform.isWindows ? 'sing-box.exe' : 'sing-box'; // 可执行文件同目录 final exeDir = File(Platform.resolvedExecutable).parent; for (final candidate in [ File('${exeDir.path}/$binName'), // macOS .app bundle: Contents/MacOS/../Resources/sing-box File('${exeDir.parent.path}/Resources/$binName'), ]) { if (await candidate.exists()) return candidate.path; } // 开发目录: /app/kernel/dist/desktop/-/sing-box final os = _platformOsName(); final arch = _platformArchName(); if (os != null && arch != null) { // 往上找项目根(含 app/kernel 的目录) Directory dir = exeDir; for (var i = 0; i < 8; i++) { final candidate = File('${dir.path}/app/kernel/dist/desktop/$os-$arch/$binName'); if (await candidate.exists()) return candidate.path; final parent = dir.parent; if (parent.path == dir.path) break; dir = parent; } } // 系统常见安装路径兜底(Apple Silicon brew / Intel·Linux brew) for (final fallback in const [ '/opt/homebrew/bin/sing-box', '/usr/local/bin/sing-box', ]) { if (await File(fallback).exists()) return fallback; } throw FileSystemException( 'sing-box binary not found. ' 'Set PANGOLIN_SINGBOX_BIN or run: ' 'app/kernel/fetch-desktop-bin.sh ${os ?? "darwin"} ${arch ?? "arm64"}', ); } static String? _platformOsName() { if (Platform.isMacOS) return 'darwin'; if (Platform.isLinux) return 'linux'; if (Platform.isWindows) return 'windows'; return null; } static String? _platformArchName() { // Dart 目前没有公开 CPU arch 检测;用环境变量或约定 final envArch = Platform.environment['PROCESSOR_ARCHITECTURE'] ?? Platform.environment['HOSTTYPE'] ?? ''; if (envArch.contains('arm') || envArch.contains('aarch')) return 'arm64'; // Apple Silicon Mac 的 HOSTTYPE 是 arm64,Intel 是 x86_64 // 若 uname 不可用则按照 macOS 默认 arm64(M 系芯片为主) if (Platform.isMacOS) return 'arm64'; return 'amd64'; } // ── 命令构造 ───────────────────────────────────────────────── List _buildCommand(String binPath, String configPath) { if (useSudo && (Platform.isMacOS || Platform.isLinux)) { // PoC: sudo 提权,开发机需配 sudoers 白名单(无密码)或近期有 sudo 缓存。 // 正式版: SMJobBless (macOS) / polkit (Linux) // BACKLOG-11D-HELPER: 实装特权 Helper + 公证后删除此分支 return ['sudo', binPath, 'run', '-c', configPath]; } // Windows: 以管理员权限启动 Flutter 应用时 sing-box 自动获得管理员权 return [binPath, 'run', '-c', configPath]; } // ── 配置文件解析 ────────────────────────────────────────────── /// 读取 configPath,提取 experimental.clash_api.external_controller 和 secret。 /// 返回 (controller_address, secret),如 ('127.0.0.1:51234', 'abc123')。 static Future<(String, String)> _parseClashEndpoint(String configPath) async { final raw = await File(configPath).readAsString(); final Map json; try { json = jsonDecode(raw) as Map; } catch (e) { throw FormatException('invalid sing-box config: $e'); } final exp = json['experimental']; if (exp is! Map) { return ('127.0.0.1:9090', ''); } final api = exp['clash_api']; if (api is! Map) { return ('127.0.0.1:9090', ''); } final controller = (api['external_controller'] as String?) ?? '127.0.0.1:9090'; final secret = (api['secret'] as String?) ?? ''; return (controller, secret); } } // ══════════════════════════════════════════════════════════════════ // 辅助:随机 Clash API 端口 / Secret 生成器 // ══════════════════════════════════════════════════════════════════ /// 生成用于 Clash API 的随机高位端口(49152–65535)。 int generateClashApiPort() { return 49152 + math.Random().nextInt(65535 - 49152 + 1); } /// 生成 32 字节(64 hex 字符)随机 Secret。 String generateClashApiSecret() { final rng = math.Random.secure(); final bytes = List.generate(32, (_) => rng.nextInt(256)); return bytes.map((b) => b.toRadixString(16).padLeft(2, '0')).join(); }