From 7fba1fec072d5dd6a051435a5ebc890ef8f4face Mon Sep 17 00:00:00 2001 From: anlicheng <244108715@qq.com> Date: Wed, 6 May 2026 21:30:58 +0800 Subject: [PATCH] =?UTF-8?q?fix=20context=20=E7=BB=93=E6=9E=84?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../{Actors => Context}/SDLContextActor.swift | 1223 +++++++++-------- 1 file changed, 635 insertions(+), 588 deletions(-) rename Tun/Punchnet/{Actors => Context}/SDLContextActor.swift (90%) diff --git a/Tun/Punchnet/Actors/SDLContextActor.swift b/Tun/Punchnet/Context/SDLContextActor.swift similarity index 90% rename from Tun/Punchnet/Actors/SDLContextActor.swift rename to Tun/Punchnet/Context/SDLContextActor.swift index eb17e6d..e65b6fe 100644 --- a/Tun/Punchnet/Actors/SDLContextActor.swift +++ b/Tun/Punchnet/Context/SDLContextActor.swift @@ -14,6 +14,34 @@ import NIOCore 1. 处理rsa的加解密逻辑 */ +private func startMonitorTask(name: String, _ body: @escaping () async throws -> Void, retryDelay: Duration = .seconds(5)) -> Task { + return Task(name: name) { + while true { + do { + try Task.checkCancellation() + try await body() + } catch is CancellationError { + SDLLogger.log("[SDLContext] worker \(name) cancelled", for: .debug) + break + } catch let err { + SDLLogger.log("[SDLContext] worker \(name) crashed: \(err.localizedDescription), will restart", for: .debug) + do { + try await Task.sleep(for: retryDelay) + } catch is CancellationError { + break + } catch { + break + } + } + } + } +} + +// 转换成ip地址 +private func asIpAddress(_ ipNum: UInt32) -> String { + return SDLUtil.int32ToIp(ipNum) +} + enum SDLContextError: Error { case udpHoleClosed @@ -61,7 +89,8 @@ actor SDLContextActor { private var udpHoleLocalAddress: SocketAddress? private var udpHoleV6: SDLUDPHoleV6? - + private var udpHoleV6MonitorTask: Task? + // dns的client对象 private var dnsClient: DNSCloudClient? private var dnsMonitorTask: Task? @@ -136,85 +165,228 @@ actor SDLContextActor { self.startDnsLocalMonitor() self.startUDPHoleMonitor() + + // self.startUDPHoleV6Monitor() + self.startSuperMonitor() } - // 取消出口节点的时候，ip地址为: 0.0.0.0 - public func updateExitNode(exitNodeIp: String) async throws { - if let ip = SDLUtil.ipv4StrToInt32(exitNodeIp), ip > 0 { - self.config.exitNode = .init(exitNodeIp: ip) - } else { - self.config.exitNode = nil - } - try await self.setNetworkSettings(config: config, dnsServer: DNSHelper.dnsServer) + // 处理context的停止问题 + public func stop() async { + await self.puncherActor.stop() + await self.arpServer.clear() + await self.sessionManager.clear() + + self.flowSessionManager.clear() + + let udpHole = self.udpHole + self.udpHole = nil + self.udpHoleLocalAddress = nil + self.udpHoleMonitorTask?.cancel() + self.udpHoleMonitorTask = nil + self.natProbeTask?.cancel() + self.natProbeTask = nil + await udpHole?.stop() + + let udpHoleV6 = self.udpHoleV6 + self.udpHoleV6 = nil + self.udpHoleV6MonitorTask?.cancel() + self.udpHoleV6MonitorTask = nil + udpHoleV6?.stop() + + let dnsClient = self.dnsClient + self.dnsClient = nil + let dnsLocalClient = self.dnsLocalClient + self.dnsLocalClient = nil + self.dnsMonitorTask?.cancel() + self.dnsMonitorTask = nil + self.dnsLocalMonitorTask?.cancel() + self.dnsLocalMonitorTask = nil + dnsClient?.stop() + await dnsLocalClient?.stop() + + let superClient = self.superClient + self.superClient = nil + self.superMonitorTask?.cancel() + self.superMonitorTask = nil + await superClient?.stop() + + self.readTask?.cancel() + self.readTask = nil + + self.registerTask?.cancel() + self.registerTask = nil + + self.stunRequestTask?.cancel() + self.stunRequestTask = nil + + self.updatePolicyTask?.cancel() + self.updatePolicyTask = nil + + self.sessionToken = nil + self.dataCipher = nil + self.natType = .blocked + + await self.ipv6AssistClient?.stop() + self.ipv6AssistClient = nil } + + deinit { + self.udpHole = nil + self.udpHoleLocalAddress = nil + self.udpHoleV6 = nil + self.dnsClient = nil + SDLLogger.log("[SDLContext] deinit", for: .debug) + } + +} + +// MARK: probe网络类型 +extension SDLContextActor { + + private func setNatType(natType: SDLNATProberActor.NatType) { + self.natType = natType + } + + // 探测当前网络的类型 + private func probeNatType() async { + guard let udpHole = self.udpHole else { + return + } + // 开始探测nat的类型 + self.natType = await self.proberActor.probeNatType(using: udpHole) + SDLLogger.log("[SDLContext] nat_type is: \(natType)") + } +} + +// MARK: Notifier通知机制 +extension SDLContextActor { + + private func prepareTunnelNotifier() { + // 启动noticeClient + // 旧的 UDP NoticeClient 已移除，改为初始化基于 App Group 的通知通道。 + SDLTunnelAppNotifier.shared.clear() + + SDLLogger.log("[SDLContext] tunnelAppNotifier ready") + } + + private func publishTunnelEvent(code: Int? = nil, message: String) { + SDLTunnelAppNotifier.shared.publish(code: code, message: message) + } +} + +// MARK: 数据发送 +extension SDLContextActor { + // 发送给super/stun节点的数据 + private func sendSuperPacket(type: SDLPacketType, data: Data) async { + await self.sendPacket(type: type, data: data, remoteAddress: self.config.stunSocketAddress) + } + + // 发送给peer的数据 + private func sendPeerPacket(type: SDLPacketType, data: Data, remoteAddress: SocketAddress) async { + await self.sendPacket(type: type, data: data, remoteAddress: remoteAddress) + } + + private func sendPacket(type: SDLPacketType, data: Data, remoteAddress: SocketAddress) async { + switch remoteAddress { + case .v4: + guard let udpHole = self.udpHole else { + SDLLogger.log("[SDLContext] udpHole is nil for remoteAddress: \(remoteAddress)", for: .debug) + return + } + await udpHole.send(type: type, data: data, remoteAddress: remoteAddress) + case .v6: + guard let udpHoleV6 = self.udpHoleV6 else { + SDLLogger.log("[SDLContext] udpHoleV6 is nil for remoteAddress: \(remoteAddress)", for: .debug) + return + } + udpHoleV6.send(type: type, data: data, remoteAddress: remoteAddress) + default: + SDLLogger.log("[SDLContext] unsupported socket family: \(remoteAddress)", for: .debug) + } + } +} + +// MARK: 处理和Super之间的通讯 +extension SDLContextActor { private func startSuperMonitor() { guard self.superMonitorTask == nil else { return } - self.superMonitorTask = self.startMonitorTask(name: "superMonitorTask") { + self.superMonitorTask = startMonitorTask(name: "superMonitorTask") { try await self.startSuperClient() } } private func startSuperClient() async throws { - // 启动monitor let superClient = SDLSuperClient(host: self.config.serverHost, port: 1443) self.superClient = superClient + await superClient.start() - - defer { - if self.superClient === superClient { - self.superClient = nil - } - SDLLogger.log("[SDLContext] startSuperClient defer") - } - + do { try await withTaskCancellationHandler { - // 这里必须等待quic的协商完成 - try await Task.sleep(for: .seconds(0.5)) - SDLLogger.log("[SDLContext] start super client: \(self.config.serverHost)") - - try await withThrowingTaskGroup { group in - defer { - group.cancelAll() - } - - group.addTask { - for try await message in await superClient.messageStream { - try Task.checkCancellation() - await self.handleQUICMessage(message: message) - } - } - - group.addTask { - while true { - try await Task.sleep(for: .seconds(5)) - try Task.checkCancellation() - await superClient.send(type: .ping, data: Data()) - } - } - - try await group.next() - } + try await runSuperClient(superClient) } onCancel: { - SDLLogger.log("[SDLSuperClient] startSuperClient taskGroup cancel", for: .debug) + SDLLogger.log("[SDLContext] startSuperClient onCancel", for: .debug) + Task { - SDLLogger.log("[SDLContext] startSuperClient onCancel") await superClient.stop() } } - } catch let err { - await superClient.stop() - SDLLogger.log("[SDLContext] startSuperClient catch err: \(err)") - throw err + + await cleanupSuperClient(superClient) + + } catch { + await cleanupSuperClient(superClient) + SDLLogger.log("[SDLContext] startSuperClient catch err: \(error)") + throw error } } + + private func runSuperClient(_ superClient: SDLSuperClient) async throws { + try await Task.sleep(for: .seconds(0.5)) + try Task.checkCancellation() + + SDLLogger.log("[SDLContext] start super client: \(self.config.serverHost)") + + try await withThrowingTaskGroup(of: Void.self) { group in + defer { + group.cancelAll() + } + + group.addTask { + for try await message in await superClient.messageStream { + try Task.checkCancellation() + await self.handleSuperMessage(message: message) + } + } + + group.addTask { + while true { + try await Task.sleep(for: .seconds(5)) + try Task.checkCancellation() + await superClient.send(type: .ping, data: Data()) + } + } + + _ = try await group.next() + } + } + + private func cleanupSuperClient(_ superClient: SDLSuperClient) async { + await superClient.stop() + + if self.superClient === superClient { + self.superClient = nil + } + + SDLLogger.log("[SDLContext] cleanupSuperClient") + } - private func handleQUICMessage(message: SDLQUICInboundMessage) async { + private func handleSuperMessage(message: SDLQUICInboundMessage) async { switch message { case .welcome(let welcome): SDLLogger.log("[SDLContext] quic welcome: \(welcome)") @@ -269,531 +441,6 @@ actor SDLContextActor { } } - private func prepareTunnelNotifier() { - // 启动noticeClient - // 旧的 UDP NoticeClient 已移除，改为初始化基于 App Group 的通知通道。 - SDLTunnelAppNotifier.shared.clear() - - SDLLogger.log("[SDLContext] tunnelAppNotifier ready") - } - - private func startDnsMonitor() { - guard self.dnsMonitorTask == nil else { - return - } - - self.dnsMonitorTask = self.startMonitorTask(name: "dnsMonitorTask") { - try await self.startDnsClient() - } - } - - private func startDnsClient() async throws { - // 启动dns服务 - let dnsClient = DNSCloudClient(host: self.config.serverHost, port: 15353) - self.dnsClient = dnsClient - dnsClient.start() - - defer { - dnsClient.stop() - self.dnsClient = nil - } - - try await withTaskCancellationHandler { - for try await packet in dnsClient.packetFlow { - try Task.checkCancellation() - - let nePacket = NEPacket(data: packet, protocolFamily: 2) - self.provider.packetFlow.writePacketObjects([nePacket]) - } - } onCancel: { - dnsClient.stop() - } - } - - private func startDnsLocalMonitor() { - guard self.dnsLocalMonitorTask == nil else { - return - } - - self.dnsLocalMonitorTask = self.startMonitorTask(name: "dnsLocalMonitorTask") { - try await self.startDnsLocalClient() - } - } - - private func startDnsLocalClient() async throws { - let dnsServer = self.publicDnsServers.randomElement() ?? self.publicDnsServers[0] - // 启动dns服务 - let dnsLocalClient = DNSLocalClient(host: dnsServer) - await dnsLocalClient.start() - SDLLogger.log("[SDLContext] dnsLocalClient started") - self.dnsLocalClient = dnsLocalClient - - defer { - self.dnsLocalClient = nil - } - - do { - try await withTaskCancellationHandler { - // 处理事件流 - for try await packet in dnsLocalClient.packetFlow { - try Task.checkCancellation() - // 要想办法构造一个完整的Ip包 - let nePacket = NEPacket(data: packet, protocolFamily: 2) - self.provider.packetFlow.writePacketObjects([nePacket]) - } - } onCancel: { - Task { - await dnsLocalClient.stop() - } - } - } catch let err { - await dnsLocalClient.stop() - throw err - } - } - - private func startUDPHoleMonitor() { - guard self.udpHoleMonitorTask == nil else { - return - } - - self.udpHoleMonitorTask = self.startMonitorTask(name: "udpHoleMonitorTask") { - try await self.startUDPHole() - } - } - - private func startUDPHole() async throws { - // 启动udp服务器 - let udpHole = try SDLUDPHole() - let localAddress = try await udpHole.start() - SDLLogger.log("[SDLContext] udpHole started, on address: \(localAddress)") - self.udpHole = udpHole - self.udpHoleLocalAddress = localAddress - - defer { - if self.udpHole === udpHole { - self.udpHole = nil - self.udpHoleLocalAddress = nil - } - } - - // 开始探测nat的类型 - self.natProbeTask?.cancel() - let proberActor = self.proberActor - self.natProbeTask = Task { [weak self] in - SDLLogger.log("[SDLContext] start probeNatType") - let natType = await proberActor.probeNatType(using: udpHole) - if Task.isCancelled { - return - } - await self?.setNatType(natType: natType) - } - - SDLLogger.log("[SDLContext] udpHole ready") - - do { - try await withTaskCancellationHandler { - for try await (remoteAddress, message) in await udpHole.messageStream() { - try Task.checkCancellation() - - switch message.inboundMessage { - case .control(let message): - switch message { - case .stunReply(_): - SDLLogger.log("[SDLContext] get a stunReply", for: .debug) - case .stunProbeReply(let probeReply): - await self.proberActor.handleProbeReply(localAddress: localAddress, reply: probeReply) - case .register(let register): - try? await self.handleRegister(remoteAddress: remoteAddress, register: register, source: .v4) - case .registerAck(let registerAck): - await self.handleRegisterAck(remoteAddress: remoteAddress, registerAck: registerAck, source: .v6) - } - case .data(let data): - try? await self.handleHoleData(data: data) - } - } - } onCancel: { - Task { - await udpHole.stop() - } - } - } catch let err { - await udpHole.stop() - throw err - } - } - - private func startUDPHoleV6() async throws { - // 启动udp服务器 - let udpHoleV6 = try SDLUDPHoleV6() - let localAddress = try udpHoleV6.start() - self.udpHoleV6 = udpHoleV6 - - if let localAddress { - SDLLogger.log("[SDLContext] udpHoleV6 started, on address: \(localAddress)") - } else { - SDLLogger.log("[SDLContext] udpHoleV6 started, no local address") - } - - defer { - if self.udpHoleV6 === udpHoleV6 { - udpHoleV6.stop() - self.udpHoleV6 = nil - } - } - - try await withThrowingTaskGroup { group in - defer { - group.cancelAll() - } - - // 处理消息流 - group.addTask { - for await (remoteAddress, message) in udpHoleV6.messageStream { - try Task.checkCancellation() - - switch message.inboundMessage { - case .control(let message): - switch message { - case .register(let register): - try? await self.handleRegister(remoteAddress: remoteAddress, register: register, source: .v6) - case .registerAck(let registerAck): - await self.handleRegisterAck(remoteAddress: remoteAddress, registerAck: registerAck, source: .v6) - default: - () - } - case .data(let data): - try? await self.handleHoleData(data: data) - } - } - } - - group.addTask { - for await event in udpHoleV6.eventStream { - try Task.checkCancellation() - - switch event { - case .ready: - SDLLogger.log("[SDLContext] udpHoleV6 ready") - case .closed, .errorCaught: - throw SDLContextError.udpHoleClosed - } - } - } - - try await group.next() - } - } - - // 处理context的停止问题 - public func stop() async { - await self.puncherActor.stop() - await self.arpServer.clear() - await self.sessionManager.clear() - - self.flowSessionManager.clear() - - let udpHole = self.udpHole - self.udpHole = nil - self.udpHoleLocalAddress = nil - - let udpHoleV6 = self.udpHoleV6 - self.udpHoleV6 = nil - - self.udpHoleMonitorTask?.cancel() - self.udpHoleMonitorTask = nil - self.natProbeTask?.cancel() - self.natProbeTask = nil - udpHoleV6?.stop() - await udpHole?.stop() - - let dnsClient = self.dnsClient - self.dnsClient = nil - let dnsLocalClient = self.dnsLocalClient - self.dnsLocalClient = nil - self.dnsMonitorTask?.cancel() - self.dnsMonitorTask = nil - self.dnsLocalMonitorTask?.cancel() - self.dnsLocalMonitorTask = nil - dnsClient?.stop() - await dnsLocalClient?.stop() - - let superClient = self.superClient - self.superClient = nil - self.superMonitorTask?.cancel() - self.superMonitorTask = nil - await superClient?.stop() - - self.readTask?.cancel() - self.readTask = nil - - self.registerTask?.cancel() - self.registerTask = nil - - self.stunRequestTask?.cancel() - self.stunRequestTask = nil - - self.updatePolicyTask?.cancel() - self.updatePolicyTask = nil - - self.sessionToken = nil - self.dataCipher = nil - self.natType = .blocked - - await self.ipv6AssistClient?.stop() - self.ipv6AssistClient = nil - } - - private func publishTunnelEvent(code: Int? = nil, message: String) { - SDLTunnelAppNotifier.shared.publish(code: code, message: message) - } - - private func setNatType(natType: SDLNATProberActor.NatType) { - self.natType = natType - } - - // 注册成功super的回调函数 - private func whenRegistedSuper() async { - self.updatePolicyTask?.cancel() - - self.updatePolicyTask = Task { - do { - while true { - try await Task.sleep(for: .seconds(300)) - SDLLogger.log("[SDLContext] updatePolicyTask execute") - await self.identifyStore.batUpdatePolicy(using: self.superClient, dstIdentityID: self.config.identityId) - } - } catch let err { - SDLLogger.log("[SDLContext] updatePolicyTask stop with err: \(err)") - } - } - } - - // MARK: -- StunRequestTask - private func startStunRequestTask(welcome: SDLWelcome) async { - self.stunRequestTask?.cancel() - self.stunRequestTask = nil - - await self.ipv6AssistClient?.stop() - self.ipv6AssistClient = SDLIPV6AssistClient(assistServerInfo: welcome.ipv6Assist) - await self.ipv6AssistClient?.start() - let ipv6AssistClient = self.ipv6AssistClient - - // 通过 welcome信息拿到当前可用使用的ipv6地址 - // 处理心跳逻辑 - self.stunRequestTask = Task.detached { [weak self] in - let timerStream = SDLAsyncTimerStream() - timerStream.start(interval: .seconds(8)) - - for await _ in timerStream.stream { - if Task.isCancelled { - break - } - - let probeReply = try? await ipv6AssistClient?.probe(requestTimeout: .seconds(3)) - if let v6Info = probeReply?.v6Info, let v6Address = SDLUtil.ipv6DataToString(v6Info.v6) { - SDLLogger.log("[SDLContext] probe ipv6 address: \(v6Address)") - } else { - SDLLogger.log("[SDLContext] probe ipv6 address: empty") - } - await self?.sendStunRequest(probeReply: probeReply) - } - - SDLLogger.log("[SDLContext] udp stunRequestTask cancel") - } - } - - private func sendStunRequest(probeReply: SDLV6AssistProbeReply?) async { - guard let sessionToken = self.sessionToken else { - return - } - - var stunRequest = SDLStunRequest() - stunRequest.clientID = self.config.clientId - stunRequest.networkID = self.config.networkAddress.networkId - stunRequest.ip = self.config.networkAddress.ip - stunRequest.mac = self.config.networkAddress.mac - stunRequest.natType = UInt32(self.natType.rawValue) - stunRequest.sessionToken = sessionToken - - if let v6Info = probeReply?.v6Info { - stunRequest.v6Info = v6Info - } - - if let stunData = try? stunRequest.serializedData() { - await self.sendSuperPacket(type: .stunRequest, data: stunData) - } - } - - // 开始读取数据, 用单独的线程处理packetFlow - private func startReader() { - // 停止之前的任务 - self.readTask?.cancel() - - // 开启新的任务 - let provider = self.provider - self.readTask = Task.detached(priority: .high) { [weak self] in - repeat { - if Task.isCancelled { - return - } - - let (packets, numbers) = await provider.packetFlow.readPackets() - if Task.isCancelled { - return - } - - for (data, number) in zip(packets, numbers) where number == 2 { - if let ipPacket = IPPacket(data) { - await self?.dealTunPacket(packet: ipPacket) - } - } - } while true - } - } - - // MARK: 网络改变时需要重新配置网络信息 - private func setNetworkSettings(config: SDLConfiguration, dnsServer: String) async throws { - let networkAddress = config.networkAddress - - // 配置路由规则 - var routes: [NEIPv4Route] = [ - NEIPv4Route(destinationAddress: networkAddress.netAddress, subnetMask: networkAddress.maskAddress), - NEIPv4Route(destinationAddress: dnsServer, subnetMask: "255.255.255.255"), - ] - - // 如果存在出口节点配置，则接管系统默认留有 - if config.exitNode != nil { - routes.append(.default()) - } - - // Add code here to start the process of connecting the tunnel. - let networkSettings = NEPacketTunnelNetworkSettings(tunnelRemoteAddress: "8.8.8.8") - networkSettings.mtu = 1250 - - // 设置网卡的DNS解析 - let networkDomain = networkAddress.networkDomain - let dnsSettings = NEDNSSettings(servers: [dnsServer]) - - dnsSettings.searchDomains = [networkDomain] - dnsSettings.matchDomains = [networkDomain, ""] - // 设置为 false 允许系统在补全 Search Domain 时也能匹配到此设置 - dnsSettings.matchDomainsNoSearch = false - - networkSettings.dnsSettings = dnsSettings - - let ipv4Settings = NEIPv4Settings(addresses: [networkAddress.ipAddress], subnetMasks: [networkAddress.maskAddress]) - // 设置路由表 - ipv4Settings.includedRoutes = routes - - // 配置要排除的路由 - ipv4Settings.excludedRoutes = self.getIpv4ExcludeRoutes() - - networkSettings.ipv4Settings = ipv4Settings - // 网卡配置设置必须成功 - try await self.provider.setTunnelNetworkSettings(networkSettings) - } - - // 探测当前网络的类型 - private func probeNatType() async { - guard let udpHole = self.udpHole else { - return - } - // 开始探测nat的类型 - self.natType = await self.proberActor.probeNatType(using: udpHole) - SDLLogger.log("[SDLContext] nat_type is: \(natType)") - } - - // 发送给super/stun节点的数据 - private func sendSuperPacket(type: SDLPacketType, data: Data) async { - await self.sendPacket(type: type, data: data, remoteAddress: self.config.stunSocketAddress) - } - - // 发送给peer的数据 - private func sendPeerPacket(type: SDLPacketType, data: Data, remoteAddress: SocketAddress) async { - await self.sendPacket(type: type, data: data, remoteAddress: remoteAddress) - } - - private func sendPacket(type: SDLPacketType, data: Data, remoteAddress: SocketAddress) async { - switch remoteAddress { - case .v4: - guard let udpHole = self.udpHole else { - SDLLogger.log("[SDLContext] udpHole is nil for remoteAddress: \(remoteAddress)", for: .debug) - return - } - await udpHole.send(type: type, data: data, remoteAddress: remoteAddress) - case .v6: - guard let udpHoleV6 = self.udpHoleV6 else { - SDLLogger.log("[SDLContext] udpHoleV6 is nil for remoteAddress: \(remoteAddress)", for: .debug) - return - } - udpHoleV6.send(type: type, data: data, remoteAddress: remoteAddress) - default: - SDLLogger.log("[SDLContext] unsupported socket family: \(remoteAddress)", for: .debug) - } - } - - private func getIpv4ExcludeRoutes() -> [NEIPv4Route] { - // 要排除的路由表 - let dnsServers = SDLUtil.getMacOSSystemDnsServers() - var ipv4DnsServers = dnsServers.filter {!$0.contains(":")} - - // 增加常见的dns服务 - let commonDnsServers = [ - "8.8.8.8", - "8.8.4.4", - "223.5.5.5", - "223.6.6.6", - "114.114.114.114" - ] - for ip in commonDnsServers { - if !ipv4DnsServers.contains(ip) { - ipv4DnsServers.append(ip) - } - } - - return ipv4DnsServers.map { NEIPv4Route(destinationAddress: $0, subnetMask: "255.255.255.255") } - } - - deinit { - self.udpHole = nil - self.udpHoleLocalAddress = nil - self.udpHoleV6 = nil - self.dnsClient = nil - SDLLogger.log("[SDLContext] deinit", for: .debug) - } -} - -extension SDLContextActor { - - func startMonitorTask(name: String, _ body: @escaping () async throws -> Void, retryDelay: Duration = .seconds(5)) -> Task { - return Task(name: name) { - while true { - do { - try Task.checkCancellation() - try await body() - } catch is CancellationError { - SDLLogger.log("[SDLContext] worker \(name) cancelled", for: .debug) - break - } catch let err { - SDLLogger.log("[SDLContext] worker \(name) crashed: \(err.localizedDescription), will restart", for: .debug) - do { - try await Task.sleep(for: retryDelay) - } catch is CancellationError { - break - } catch { - break - } - } - } - } - } - -} - -// 处理和Super之间的通讯 -extension SDLContextActor { - private func makeSuperEventProcessor() -> SDLSuperEventProcessor { return .init(networkAddress: self.config.networkAddress) } @@ -837,6 +484,23 @@ extension SDLContextActor { } } + // 注册成功super的回调函数 + private func whenRegistedSuper() async { + self.updatePolicyTask?.cancel() + + self.updatePolicyTask = Task { + do { + while true { + try await Task.sleep(for: .seconds(300)) + SDLLogger.log("[SDLContext] updatePolicyTask execute") + await self.identifyStore.batUpdatePolicy(using: self.superClient, dstIdentityID: self.config.identityId) + } + } catch let err { + SDLLogger.log("[SDLContext] updatePolicyTask stop with err: \(err)") + } + } + } + private func handleRegisterSuperNak(nakPacket: SDLRegisterSuperNak) { let errorMessage = nakPacket.errorMessage guard let errorCode = SDLNAKErrorCode(rawValue: UInt8(nakPacket.errorCode)) else { @@ -901,9 +565,163 @@ extension SDLContextActor { } -// 处理从Hole收到的数据 +// MARK: 处理DnsLocal extension SDLContextActor { + private func startDnsLocalMonitor() { + guard self.dnsLocalMonitorTask == nil else { + return + } + + self.dnsLocalMonitorTask = startMonitorTask(name: "dnsLocalMonitorTask") { + try await self.startDnsLocalClient() + } + } + + private func startDnsLocalClient() async throws { + let dnsServer = self.publicDnsServers.randomElement() ?? self.publicDnsServers[0] + // 启动dns服务 + let dnsLocalClient = DNSLocalClient(host: dnsServer) + await dnsLocalClient.start() + SDLLogger.log("[SDLContext] dnsLocalClient started") + self.dnsLocalClient = dnsLocalClient + + defer { + self.dnsLocalClient = nil + } + + do { + try await withTaskCancellationHandler { + // 处理事件流 + for try await packet in dnsLocalClient.packetFlow { + try Task.checkCancellation() + // 要想办法构造一个完整的Ip包 + let nePacket = NEPacket(data: packet, protocolFamily: 2) + self.provider.packetFlow.writePacketObjects([nePacket]) + } + } onCancel: { + Task { + await dnsLocalClient.stop() + } + } + } catch let err { + await dnsLocalClient.stop() + throw err + } + } +} + +// MARK: 处理DnsCloud +extension SDLContextActor { + + private func startDnsMonitor() { + guard self.dnsMonitorTask == nil else { + return + } + + self.dnsMonitorTask = startMonitorTask(name: "dnsMonitorTask") { + try await self.startDnsClient() + } + } + + private func startDnsClient() async throws { + // 启动dns服务 + let dnsClient = DNSCloudClient(host: self.config.serverHost, port: 15353) + self.dnsClient = dnsClient + dnsClient.start() + + defer { + dnsClient.stop() + self.dnsClient = nil + } + + try await withTaskCancellationHandler { + for try await packet in dnsClient.packetFlow { + try Task.checkCancellation() + + let nePacket = NEPacket(data: packet, protocolFamily: 2) + self.provider.packetFlow.writePacketObjects([nePacket]) + } + } onCancel: { + dnsClient.stop() + } + } +} + +// MARK: 处理从Hole收到的数据 +extension SDLContextActor { + + private func startUDPHoleMonitor() { + guard self.udpHoleMonitorTask == nil else { + return + } + + self.udpHoleMonitorTask = startMonitorTask(name: "udpHoleMonitorTask") { + try await self.startUDPHole() + } + } + + private func startUDPHole() async throws { + // 启动udp服务器 + let udpHole = try SDLUDPHole() + let localAddress = try await udpHole.start() + SDLLogger.log("[SDLContext] udpHole started, on address: \(localAddress)") + self.udpHole = udpHole + self.udpHoleLocalAddress = localAddress + + defer { + if self.udpHole === udpHole { + self.udpHole = nil + self.udpHoleLocalAddress = nil + } + } + + // 开始探测nat的类型 + self.natProbeTask?.cancel() + let proberActor = self.proberActor + self.natProbeTask = Task { [weak self] in + SDLLogger.log("[SDLContext] start probeNatType") + let natType = await proberActor.probeNatType(using: udpHole) + if Task.isCancelled { + return + } + await self?.setNatType(natType: natType) + } + + SDLLogger.log("[SDLContext] udpHole ready") + + do { + try await withTaskCancellationHandler { + for try await (remoteAddress, message) in await udpHole.messageStream() { + try Task.checkCancellation() + + switch message.inboundMessage { + case .control(let message): + switch message { + case .stunReply(_): + SDLLogger.log("[SDLContext] get a stunReply", for: .debug) + case .stunProbeReply(let probeReply): + await self.proberActor.handleProbeReply(localAddress: localAddress, reply: probeReply) + case .register(let register): + try? await self.handleRegister(remoteAddress: remoteAddress, register: register, source: .v4) + case .registerAck(let registerAck): + await self.handleRegisterAck(remoteAddress: remoteAddress, registerAck: registerAck, source: .v6) + } + case .data(let data): + try? await self.handleHoleData(data: data) + } + } + } onCancel: { + Task { + await udpHole.stop() + } + } + } catch let err { + await udpHole.stop() + throw err + } + } + private func handleRegister(remoteAddress: SocketAddress, register: SDLRegister, source: UDPHoleKind) async throws { let networkAddr = config.networkAddress SDLLogger.log("[SDLContext] register packet: \(register), network_address: \(networkAddr)") @@ -981,17 +799,246 @@ extension SDLContextActor { } -private extension UInt32 { - // 转换成ip地址 - func asIpAddress() -> String { - return SDLUtil.int32ToIp(self) +// MARK: 处理从HoleV6收到的数据 +extension SDLContextActor { + + private func startUDPHoleV6Monitor() { + guard self.udpHoleV6MonitorTask == nil else { + return + } + + self.udpHoleV6MonitorTask = startMonitorTask(name: "udpHoleV6MonitorTask") { + try await self.startUDPHoleV6() + } + } + + private func startUDPHoleV6() async throws { + // 启动udp服务器 + let udpHoleV6 = try SDLUDPHoleV6() + let localAddress = try udpHoleV6.start() + self.udpHoleV6 = udpHoleV6 + + if let localAddress { + SDLLogger.log("[SDLContext] udpHoleV6 started, on address: \(localAddress)") + } else { + SDLLogger.log("[SDLContext] udpHoleV6 started, no local address") + } + + defer { + if self.udpHoleV6 === udpHoleV6 { + udpHoleV6.stop() + self.udpHoleV6 = nil + } + } + + try await withThrowingTaskGroup { group in + defer { + group.cancelAll() + } + + // 处理消息流 + group.addTask { + for await (remoteAddress, message) in udpHoleV6.messageStream { + try Task.checkCancellation() + + switch message.inboundMessage { + case .control(let message): + switch message { + case .register(let register): + try? await self.handleRegister(remoteAddress: remoteAddress, register: register, source: .v6) + case .registerAck(let registerAck): + await self.handleRegisterAck(remoteAddress: remoteAddress, registerAck: registerAck, source: .v6) + default: + () + } + case .data(let data): + try? await self.handleHoleData(data: data) + } + } + } + + group.addTask { + for await event in udpHoleV6.eventStream { + try Task.checkCancellation() + + switch event { + case .ready: + SDLLogger.log("[SDLContext] udpHoleV6 ready") + case .closed, .errorCaught: + throw SDLContextError.udpHoleClosed + } + } + } + + try await group.next() + } + } + +} + +// MARK: 和Stun相关的心跳机制 +extension SDLContextActor { + // MARK: -- StunRequestTask + private func startStunRequestTask(welcome: SDLWelcome) async { + self.stunRequestTask?.cancel() + self.stunRequestTask = nil + + await self.ipv6AssistClient?.stop() + self.ipv6AssistClient = SDLIPV6AssistClient(assistServerInfo: welcome.ipv6Assist) + await self.ipv6AssistClient?.start() + let ipv6AssistClient = self.ipv6AssistClient + + // 通过 welcome信息拿到当前可用使用的ipv6地址 + // 处理心跳逻辑 + self.stunRequestTask = Task.detached { [weak self] in + let timerStream = SDLAsyncTimerStream() + timerStream.start(interval: .seconds(8)) + + for await _ in timerStream.stream { + if Task.isCancelled { + break + } + + let probeReply = try? await ipv6AssistClient?.probe(requestTimeout: .seconds(3)) + if let v6Info = probeReply?.v6Info, let v6Address = SDLUtil.ipv6DataToString(v6Info.v6) { + SDLLogger.log("[SDLContext] probe ipv6 address: \(v6Address)") + } else { + SDLLogger.log("[SDLContext] probe ipv6 address: empty") + } + await self?.sendStunRequest(probeReply: probeReply) + } + + SDLLogger.log("[SDLContext] udp stunRequestTask cancel") + } + } + + private func sendStunRequest(probeReply: SDLV6AssistProbeReply?) async { + guard let sessionToken = self.sessionToken else { + return + } + + var stunRequest = SDLStunRequest() + stunRequest.clientID = self.config.clientId + stunRequest.networkID = self.config.networkAddress.networkId + stunRequest.ip = self.config.networkAddress.ip + stunRequest.mac = self.config.networkAddress.mac + stunRequest.natType = UInt32(self.natType.rawValue) + stunRequest.sessionToken = sessionToken + + if let v6Info = probeReply?.v6Info { + stunRequest.v6Info = v6Info + } + + if let stunData = try? stunRequest.serializedData() { + await self.sendSuperPacket(type: .stunRequest, data: stunData) + } } } - -// 处理从Tun收到的数据 +// MARK: 处理从Tun相关的逻辑 extension SDLContextActor { + // 开始读取数据, 用单独的线程处理packetFlow + private func startReader() { + // 停止之前的任务 + self.readTask?.cancel() + + // 开启新的任务 + let provider = self.provider + self.readTask = Task.detached(priority: .high) { [weak self] in + repeat { + if Task.isCancelled { + return + } + + let (packets, numbers) = await provider.packetFlow.readPackets() + if Task.isCancelled { + return + } + + for (data, number) in zip(packets, numbers) where number == 2 { + if let ipPacket = IPPacket(data) { + await self?.dealTunPacket(packet: ipPacket) + } + } + } while true + } + } + + // 取消出口节点的时候，ip地址为: 0.0.0.0 + public func updateExitNode(exitNodeIp: String) async throws { + if let ip = SDLUtil.ipv4StrToInt32(exitNodeIp), ip > 0 { + self.config.exitNode = .init(exitNodeIp: ip) + } else { + self.config.exitNode = nil + } + try await self.setNetworkSettings(config: config, dnsServer: DNSHelper.dnsServer) + } + + // MARK: 网络改变时需要重新配置网络信息 + private func setNetworkSettings(config: SDLConfiguration, dnsServer: String) async throws { + let networkAddress = config.networkAddress + + // 配置路由规则 + var routes: [NEIPv4Route] = [ + NEIPv4Route(destinationAddress: networkAddress.netAddress, subnetMask: networkAddress.maskAddress), + NEIPv4Route(destinationAddress: dnsServer, subnetMask: "255.255.255.255"), + ] + + // 如果存在出口节点配置，则接管系统默认留有 + if config.exitNode != nil { + routes.append(.default()) + } + + // Add code here to start the process of connecting the tunnel. + let networkSettings = NEPacketTunnelNetworkSettings(tunnelRemoteAddress: "8.8.8.8") + networkSettings.mtu = 1250 + + // 设置网卡的DNS解析 + let networkDomain = networkAddress.networkDomain + let dnsSettings = NEDNSSettings(servers: [dnsServer]) + + dnsSettings.searchDomains = [networkDomain] + dnsSettings.matchDomains = [networkDomain, ""] + // 设置为 false 允许系统在补全 Search Domain 时也能匹配到此设置 + dnsSettings.matchDomainsNoSearch = false + + networkSettings.dnsSettings = dnsSettings + + let ipv4Settings = NEIPv4Settings(addresses: [networkAddress.ipAddress], subnetMasks: [networkAddress.maskAddress]) + // 设置路由表 + ipv4Settings.includedRoutes = routes + + // 配置要排除的路由 + ipv4Settings.excludedRoutes = self.getIpv4ExcludeRoutes() + + networkSettings.ipv4Settings = ipv4Settings + // 网卡配置设置必须成功 + try await self.provider.setTunnelNetworkSettings(networkSettings) + } + + private func getIpv4ExcludeRoutes() -> [NEIPv4Route] { + // 要排除的路由表 + let dnsServers = SDLUtil.getMacOSSystemDnsServers() + var ipv4DnsServers = dnsServers.filter {!$0.contains(":")} + + // 增加常见的dns服务 + let commonDnsServers = [ + "8.8.8.8", + "8.8.4.4", + "223.5.5.5", + "223.6.6.6", + "114.114.114.114" + ] + for ip in commonDnsServers { + if !ipv4DnsServers.contains(ip) { + ipv4DnsServers.append(ip) + } + } + + return ipv4DnsServers.map { NEIPv4Route(destinationAddress: $0, subnetMask: "255.255.255.255") } + } + // 处理读取的每个数据包, Tun收到的包的一级路由 private func dealTunPacket(packet: IPPacket) async { let router = SDLTunPacketRouter(networkAddress: self.config.networkAddress, exitNode: self.config.exitNode) @@ -1029,18 +1076,18 @@ extension SDLContextActor { private func forwardPacketToNextHop(ip: UInt32, type: LayerPacket.PacketType, data: Data, kind: SDLTunPacketRouter.ForwardKind) async { switch kind { case .sameNetwork: - SDLLogger.log("[SDLContext] dstIp: \(ip.asIpAddress()) same network", for: .trace) + SDLLogger.log("[SDLContext] dstIp: \(asIpAddress(ip)) same network", for: .trace) case .exitNode, .dnsExitNode: - SDLLogger.log("[SDLContext] use exit_node: \(ip.asIpAddress())", for: .trace) + SDLLogger.log("[SDLContext] use exit_node: \(asIpAddress(ip))", for: .trace) } // 查找arp缓存中是否有目标mac地址 if let dstMac = await self.arpServer.query(ip: ip) { - SDLLogger.log("[SDLContext] dstIp: \(ip.asIpAddress()), dst_mac is: \(SDLUtil.formatMacAddress(mac: dstMac))", for: .trace) + SDLLogger.log("[SDLContext] dstIp: \(asIpAddress(ip)), dst_mac is: \(SDLUtil.formatMacAddress(mac: dstMac))", for: .trace) await self.routeLayerPacket(dstMac: dstMac, type: type, data: data) } else { - SDLLogger.log("[SDLContext] dstIp: \(ip.asIpAddress()) arp query not found, broadcast", for: .trace) + SDLLogger.log("[SDLContext] dstIp: \(asIpAddress(ip)) arp query not found, broadcast", for: .trace) // // 构造arp广播 // let arpReqeust = ARPPacket.arpRequest(senderIP: networkAddr.ip, senderMAC: networkAddr.mac, targetIP: dstIp) // await self.routeLayerPacket(dstMac: ARPPacket.broadcastMac , type: .arp, data: arpReqeust.marshal())