771 lines
28 KiB
Swift
771 lines
28 KiB
Swift
//
|
||
// SDLContext.swift
|
||
// Tun
|
||
//
|
||
// Created by 安礼成 on 2024/2/29.
|
||
//
|
||
|
||
import Foundation
|
||
import NetworkExtension
|
||
import NIOCore
|
||
|
||
private actor SDLWorkerRestartSignal {
|
||
private var generation: UInt64 = 0
|
||
private var waiters: [UUID: CheckedContinuation<UInt64, Never>] = [:]
|
||
|
||
func request() {
|
||
self.generation &+= 1
|
||
let generation = self.generation
|
||
let waiters = self.waiters
|
||
self.waiters.removeAll()
|
||
|
||
for waiter in waiters.values {
|
||
waiter.resume(returning: generation)
|
||
}
|
||
}
|
||
|
||
func currentGeneration() -> UInt64 {
|
||
return self.generation
|
||
}
|
||
|
||
func waitForChange(after observedGeneration: UInt64) async -> UInt64 {
|
||
if self.generation != observedGeneration {
|
||
return self.generation
|
||
}
|
||
|
||
let id = UUID()
|
||
return await withTaskCancellationHandler {
|
||
await withCheckedContinuation { continuation in
|
||
if self.generation != observedGeneration {
|
||
continuation.resume(returning: self.generation)
|
||
} else {
|
||
self.waiters[id] = continuation
|
||
}
|
||
}
|
||
} onCancel: {
|
||
Task {
|
||
await self.cancelWaiter(id: id)
|
||
}
|
||
}
|
||
}
|
||
|
||
private func cancelWaiter(id: UUID) {
|
||
self.waiters.removeValue(forKey: id)
|
||
}
|
||
}
|
||
|
||
// 上下文环境变量，全局共享
|
||
/*
|
||
1. 处理rsa的加解密逻辑
|
||
*/
|
||
actor SDLContextActor {
|
||
|
||
private var config: SDLConfiguration
|
||
// nat的网络类型
|
||
var natType: SDLNATProberActor.NatType = .blocked
|
||
|
||
// AES加密，授权通过后，对象才会被创建
|
||
private var dataCipher: CCDataCipher?
|
||
|
||
// rsa的相关配置, public_key是本地生成的
|
||
// 加密算法相关
|
||
nonisolated let rsaCipher: RSACipher
|
||
|
||
private let dnsCloudService: DNSCloudService
|
||
private let dnsLocalService: DNSLocalService
|
||
private let superService: SDLSuperService
|
||
private let superControlPlane: SDLSuperControlPlane
|
||
private let holeControlPlane: SDLHoleControlPlane
|
||
private let udpHoleService: SDLUDPHoleService
|
||
private let udpHoleV6Service: SDLUDPHoleV6Service
|
||
private let packetOutboundActor: PacketOutboundActor
|
||
private let packetInboundActor: PacketInboundActor
|
||
private let tunNetworkManager: SDLTunNetworkManager
|
||
|
||
private static let publicDnsServers = ["223.5.5.5", "119.29.29.29"]
|
||
|
||
nonisolated private let puncherActor: SDLPuncherActor
|
||
// 网络探测对象
|
||
nonisolated private let proberActor: SDLNATProberActor
|
||
|
||
// 本地ipv6地址信息探测
|
||
private var ipv6AssistClient: SDLIPV6AssistClient?
|
||
private let ipv6AssistEvents: AsyncStream<SDLV6Info?>
|
||
private let ipv6AssistContinuation: AsyncStream<SDLV6Info?>.Continuation
|
||
|
||
private let sessionManager: SessionManager
|
||
nonisolated private let arpResolver: ArpResolver
|
||
|
||
// 内部socket通讯
|
||
// 改为基于 App Group + Darwin Notification 的通知
|
||
|
||
// 流量统计
|
||
nonisolated private let flowTracer: SDLFlowTracer
|
||
|
||
nonisolated private let provider: NEPacketTunnelProvider
|
||
|
||
// 处理权限控制
|
||
private let policyService: PolicyService
|
||
private var rootTask: Task<Void, Error>?
|
||
private var rootTaskID: UUID?
|
||
private var terminalError: Error?
|
||
private let readySignal = AsyncOneShot<Void>()
|
||
private let superRestartSignal = SDLWorkerRestartSignal()
|
||
private let udpHoleRestartSignal = SDLWorkerRestartSignal()
|
||
private let udpHoleV6RestartSignal = SDLWorkerRestartSignal()
|
||
private let dnsCloudRestartSignal = SDLWorkerRestartSignal()
|
||
private let dnsLocalRestartSignal = SDLWorkerRestartSignal()
|
||
|
||
public init(provider: NEPacketTunnelProvider, config: SDLConfiguration, rsaCipher: RSACipher) {
|
||
let puncherActor = SDLPuncherActor()
|
||
let proberActor = SDLNATProberActor(addressArray: config.stunProbeSocketAddressArray)
|
||
let sessionManager = SessionManager()
|
||
let arpResolver = ArpResolver()
|
||
let flowTracer = SDLFlowTracer()
|
||
let policyService = PolicyService(identityId: config.identityId, acl: config.acl)
|
||
let superService = SDLSuperService(serverEndpoint: config.serverEndpoint)
|
||
let udpHoleService = SDLUDPHoleService(proberActor: proberActor)
|
||
let udpHoleV6Service = SDLUDPHoleV6Service()
|
||
let dnsCloudService = DNSCloudService(serverIP: config.serverEndpoint.ip)
|
||
let dnsLocalService = DNSLocalService(publicDnsServers: Self.publicDnsServers)
|
||
let superControlPlane = SDLSuperControlPlane(config: config, rsaCipher: rsaCipher)
|
||
let holeControlPlane = SDLHoleControlPlane(networkAddress: config.networkAddress)
|
||
let tunNetworkManager = SDLTunNetworkManager(provider: provider)
|
||
let ipv6AssistPair = AsyncStream.makeStream(of: Optional<SDLV6Info>.self, bufferingPolicy: .bufferingNewest(1))
|
||
let packetOutboundActor = PacketOutboundActor(
|
||
provider: provider,
|
||
config: config,
|
||
dataCipher: nil,
|
||
sessionManager: sessionManager,
|
||
arpResolver: arpResolver,
|
||
puncherActor: puncherActor,
|
||
policyService: policyService,
|
||
superService: superService,
|
||
udpHoleService: udpHoleService,
|
||
udpHoleV6Service: udpHoleV6Service,
|
||
dnsCloudService: dnsCloudService,
|
||
dnsLocalService: dnsLocalService,
|
||
flowTracer: flowTracer
|
||
)
|
||
let packetInboundActor = PacketInboundActor(
|
||
provider: provider,
|
||
config: config,
|
||
dataCipher: nil,
|
||
policyService: policyService,
|
||
packetOutboundActor: packetOutboundActor,
|
||
arpResolver: arpResolver,
|
||
superService: superService,
|
||
flowTracer: flowTracer
|
||
)
|
||
|
||
self.provider = provider
|
||
self.config = config
|
||
self.rsaCipher = rsaCipher
|
||
|
||
self.puncherActor = puncherActor
|
||
self.proberActor = proberActor
|
||
|
||
self.sessionManager = sessionManager
|
||
self.arpResolver = arpResolver
|
||
self.flowTracer = flowTracer
|
||
|
||
// 权限控制
|
||
self.policyService = policyService
|
||
|
||
self.dnsCloudService = dnsCloudService
|
||
self.dnsLocalService = dnsLocalService
|
||
self.superService = superService
|
||
self.superControlPlane = superControlPlane
|
||
self.holeControlPlane = holeControlPlane
|
||
self.udpHoleService = udpHoleService
|
||
self.udpHoleV6Service = udpHoleV6Service
|
||
self.packetOutboundActor = packetOutboundActor
|
||
self.packetInboundActor = packetInboundActor
|
||
self.tunNetworkManager = tunNetworkManager
|
||
self.ipv6AssistEvents = ipv6AssistPair.stream
|
||
self.ipv6AssistContinuation = ipv6AssistPair.continuation
|
||
}
|
||
|
||
public func start() async throws {
|
||
guard self.rootTask == nil else {
|
||
try await self.readySignal.wait(timeout: .seconds(30))
|
||
return
|
||
}
|
||
|
||
let rootTaskID = UUID()
|
||
let rootTask = Task {
|
||
var result: Result<Void, Error> = .success(())
|
||
|
||
do {
|
||
try await self.runRootBody()
|
||
} catch is CancellationError {
|
||
if let terminalError = self.consumeTerminalError() {
|
||
SDLLogger.fatal("[SDLContext] root task stopped by terminal error: \(terminalError)", category: .context)
|
||
result = .failure(terminalError)
|
||
} else {
|
||
SDLLogger.fatal("[SDLContext] root task cancelled", category: .context)
|
||
result = .failure(CancellationError())
|
||
}
|
||
} catch {
|
||
SDLLogger.fatal("[SDLContext] root task failed: \(error)", category: .context)
|
||
await self.readySignal.fail(error)
|
||
result = .failure(error)
|
||
}
|
||
|
||
await self.cleanupRoot()
|
||
self.finishRootTask(id: rootTaskID)
|
||
try result.get()
|
||
}
|
||
self.rootTaskID = rootTaskID
|
||
self.rootTask = rootTask
|
||
|
||
do {
|
||
try await self.readySignal.wait(timeout: .seconds(30))
|
||
} catch {
|
||
SDLLogger.fatal("[SDLContext] start failed while waiting ready signal: \(error)", category: .context)
|
||
rootTask.cancel()
|
||
_ = try? await rootTask.value
|
||
self.rootTask = nil
|
||
self.rootTaskID = nil
|
||
self.terminalError = nil
|
||
throw error
|
||
}
|
||
}
|
||
|
||
// 处理context的停止问题
|
||
public func stop() async {
|
||
SDLLogger.fatal("[SDLContext] stop requested", category: .context)
|
||
let rootTask = self.rootTask
|
||
|
||
rootTask?.cancel()
|
||
await self.readySignal.fail(CancellationError())
|
||
_ = try? await rootTask?.value
|
||
|
||
self.rootTask = nil
|
||
self.rootTaskID = nil
|
||
self.terminalError = nil
|
||
}
|
||
|
||
public func recoverAfterWake() async throws {
|
||
SDLLogger.log("[SDLContext] recoverAfterWake requested", category: .context)
|
||
|
||
guard self.rootTask != nil else {
|
||
throw TunnelError.invalidContext
|
||
}
|
||
|
||
try await self.readySignal.wait(timeout: .seconds(30))
|
||
|
||
guard let dataCipher = self.dataCipher else {
|
||
throw TunnelError.invalidContext
|
||
}
|
||
|
||
let clearedSessions = await self.sessionManager.clear()
|
||
self.natType = .blocked
|
||
await self.stopCurrentIPv6AssistClient()
|
||
await self.packetOutboundActor.updateRuntime(config: self.config, dataCipher: dataCipher)
|
||
await self.packetInboundActor.updateRuntime(config: self.config, dataCipher: dataCipher)
|
||
try await self.tunNetworkManager.apply(settings: .init(config: self.config), dnsServer: DNSHelper.dnsServer)
|
||
await self.restartVolatileResourcesAfterWake()
|
||
|
||
SDLLogger.log("[SDLContext] recoverAfterWake completed, clearedSessions: \(clearedSessions)", category: .context)
|
||
}
|
||
|
||
private func runRootBody() async throws {
|
||
self.prepareTunnelNotifier()
|
||
|
||
await self.dnsCloudService.updateEventHandler { [weak self] event in
|
||
await self?.handleDNSEvent(event)
|
||
}
|
||
|
||
await self.dnsLocalService.updateEventHandler { [weak self] event in
|
||
await self?.handleDNSEvent(event)
|
||
}
|
||
|
||
await self.superControlPlane.updateDecisionHandler { [weak self] decision in
|
||
await self?.handleSuperDecision(decision)
|
||
}
|
||
|
||
let superControlPlane = self.superControlPlane
|
||
await self.superService.updateMessageHandler { message in
|
||
await superControlPlane.handle(message)
|
||
}
|
||
|
||
let packetInboundActor = self.packetInboundActor
|
||
await self.udpHoleService.updateHandlers(
|
||
onEvent: { [weak self] event in
|
||
await self?.handleUDPHoleControlEvent(event)
|
||
},
|
||
onData: { data in
|
||
await packetInboundActor.handleData(data)
|
||
}
|
||
)
|
||
|
||
await self.udpHoleV6Service.updateHandlers(
|
||
onEvent: { [weak self] event in
|
||
await self?.handleUDPHoleControlEvent(event)
|
||
},
|
||
onData: { data in
|
||
await packetInboundActor.handleData(data)
|
||
}
|
||
)
|
||
|
||
let superService = self.superService
|
||
let dnsCloudService = self.dnsCloudService
|
||
let dnsLocalService = self.dnsLocalService
|
||
let udpHoleService = self.udpHoleService
|
||
let udpHoleV6Service = self.udpHoleV6Service
|
||
let packetOutboundActor = self.packetOutboundActor
|
||
let policyService = self.policyService
|
||
let puncherActor = self.puncherActor
|
||
let arpResolver = self.arpResolver
|
||
let readySignal = self.readySignal
|
||
let superRestartSignal = self.superRestartSignal
|
||
let udpHoleRestartSignal = self.udpHoleRestartSignal
|
||
let udpHoleV6RestartSignal = self.udpHoleV6RestartSignal
|
||
let dnsCloudRestartSignal = self.dnsCloudRestartSignal
|
||
let dnsLocalRestartSignal = self.dnsLocalRestartSignal
|
||
|
||
try await withThrowingTaskGroup(of: Void.self) { group in
|
||
defer {
|
||
group.cancelAll()
|
||
}
|
||
|
||
group.addTask {
|
||
try await Self.runRestarting(name: "superService", restartSignal: superRestartSignal) {
|
||
try await superService.run()
|
||
}
|
||
}
|
||
|
||
group.addTask {
|
||
try await Self.runRestarting(name: "udpHoleService", restartSignal: udpHoleRestartSignal) {
|
||
try await udpHoleService.run()
|
||
}
|
||
}
|
||
|
||
group.addTask {
|
||
try await Self.runRestarting(name: "udpHoleV6Service", restartSignal: udpHoleV6RestartSignal) {
|
||
try await udpHoleV6Service.run()
|
||
}
|
||
}
|
||
|
||
group.addTask {
|
||
try await Self.runRestarting(name: "dnsCloudService", restartSignal: dnsCloudRestartSignal) {
|
||
try await dnsCloudService.run()
|
||
}
|
||
}
|
||
|
||
group.addTask {
|
||
try await Self.runRestarting(name: "dnsLocalService", restartSignal: dnsLocalRestartSignal) {
|
||
try await dnsLocalService.run()
|
||
}
|
||
}
|
||
|
||
group.addTask {
|
||
try await puncherActor.runCleanup()
|
||
}
|
||
|
||
group.addTask {
|
||
try await arpResolver.runCleanup()
|
||
}
|
||
|
||
group.addTask {
|
||
try await self.runIPv6AssistSupervisor()
|
||
}
|
||
|
||
group.addTask(priority: .high) {
|
||
_ = try await readySignal.wait()
|
||
try await packetOutboundActor.runPacketReader()
|
||
}
|
||
|
||
group.addTask {
|
||
_ = try await readySignal.wait()
|
||
try await Self.runPeriodic(name: "updatePolicyTask", interval: .seconds(10)) {
|
||
SDLLogger.log("[SDLContext] updatePolicyTask execute", category: .context)
|
||
await policyService.updatePolicy(superService: superService)
|
||
}
|
||
}
|
||
|
||
group.addTask {
|
||
_ = try await readySignal.wait()
|
||
try await Self.runPeriodic(name: "stunRequestTask", interval: .seconds(8)) {
|
||
try await self.runStunRequestOnce()
|
||
}
|
||
}
|
||
|
||
try await group.waitForAll()
|
||
}
|
||
}
|
||
|
||
private func runIPv6AssistSupervisor() async throws {
|
||
do {
|
||
for await assistInfo in self.ipv6AssistEvents {
|
||
try Task.checkCancellation()
|
||
await self.stopCurrentIPv6AssistClient()
|
||
|
||
guard let assistInfo else {
|
||
continue
|
||
}
|
||
|
||
guard let client = SDLIPV6AssistClient(assistServerInfo: assistInfo) else {
|
||
SDLLogger.log("[SDLContext] invalid ipv6 assist config", category: .context)
|
||
continue
|
||
}
|
||
|
||
self.ipv6AssistClient = client
|
||
|
||
do {
|
||
try await client.run()
|
||
} catch is CancellationError {
|
||
throw CancellationError()
|
||
} catch {
|
||
SDLLogger.log("[SDLContext] ipv6 assist client ended: \(error.localizedDescription)", category: .context)
|
||
}
|
||
|
||
if self.ipv6AssistClient === client {
|
||
self.ipv6AssistClient = nil
|
||
}
|
||
}
|
||
} catch is CancellationError {
|
||
await self.stopCurrentIPv6AssistClient()
|
||
throw CancellationError()
|
||
}
|
||
|
||
await self.stopCurrentIPv6AssistClient()
|
||
}
|
||
|
||
private func stopCurrentIPv6AssistClient() async {
|
||
let client = self.ipv6AssistClient
|
||
self.ipv6AssistClient = nil
|
||
await client?.stop()
|
||
}
|
||
|
||
private func restartVolatileResourcesAfterWake() async {
|
||
SDLLogger.log("[SDLContext] restart volatile resources after wake", category: .context)
|
||
|
||
await self.superRestartSignal.request()
|
||
await self.udpHoleRestartSignal.request()
|
||
await self.udpHoleV6RestartSignal.request()
|
||
await self.dnsCloudRestartSignal.request()
|
||
await self.dnsLocalRestartSignal.request()
|
||
|
||
await self.superService.stop()
|
||
await self.udpHoleService.stop()
|
||
await self.udpHoleV6Service.stop()
|
||
await self.dnsCloudService.stop()
|
||
await self.dnsLocalService.stop()
|
||
}
|
||
|
||
private func cleanupRoot() async {
|
||
await self.puncherActor.stop()
|
||
await self.arpResolver.stop()
|
||
await self.sessionManager.clear()
|
||
|
||
await self.policyService.clear()
|
||
|
||
await self.udpHoleService.stop()
|
||
await self.udpHoleV6Service.stop()
|
||
|
||
await self.dnsCloudService.stop()
|
||
await self.dnsLocalService.stop()
|
||
|
||
await self.superService.stop()
|
||
await self.superControlPlane.reset()
|
||
|
||
self.dataCipher = nil
|
||
self.natType = .blocked
|
||
await self.packetOutboundActor.updateRuntime(config: self.config, dataCipher: nil)
|
||
await self.packetInboundActor.updateRuntime(config: self.config, dataCipher: nil)
|
||
|
||
self.ipv6AssistContinuation.yield(nil)
|
||
await self.stopCurrentIPv6AssistClient()
|
||
}
|
||
|
||
private func requestRootShutdown(error: Error) async {
|
||
self.terminalError = error
|
||
await self.readySignal.fail(error)
|
||
self.rootTask?.cancel()
|
||
}
|
||
|
||
private func consumeTerminalError() -> Error? {
|
||
let error = self.terminalError
|
||
self.terminalError = nil
|
||
return error
|
||
}
|
||
|
||
private func finishRootTask(id: UUID) {
|
||
guard self.rootTaskID == id else {
|
||
return
|
||
}
|
||
|
||
self.rootTask = nil
|
||
self.rootTaskID = nil
|
||
self.terminalError = nil
|
||
}
|
||
|
||
deinit {
|
||
SDLLogger.log("[SDLContext] deinit", category: .context)
|
||
}
|
||
|
||
}
|
||
|
||
extension SDLContextActor {
|
||
|
||
// MARK: probe网络类型
|
||
|
||
private func setNatType(natType: SDLNATProberActor.NatType) {
|
||
self.natType = natType
|
||
}
|
||
|
||
// MARK: Notifier通知机制
|
||
|
||
private func prepareTunnelNotifier() {
|
||
// 启动noticeClient
|
||
// 旧的 UDP NoticeClient 已移除，改为初始化基于 App Group 的通知通道。
|
||
SDLTunnelAppNotifier.shared.clear()
|
||
|
||
SDLLogger.log("[SDLContext] tunnelAppNotifier ready", category: .context)
|
||
}
|
||
|
||
private func publishTunnelEvent(code: Int? = nil, message: String) {
|
||
SDLTunnelAppNotifier.shared.publish(code: code, message: message)
|
||
}
|
||
|
||
// MARK: 数据发送
|
||
|
||
private func sendPacket(type: SDLPacketType, data: Data, remoteAddress: SocketAddress) async {
|
||
switch remoteAddress {
|
||
case .v4:
|
||
await self.udpHoleService.send(type: type, data: data, remoteAddress: remoteAddress)
|
||
case .v6:
|
||
await self.udpHoleV6Service.send(type: type, data: data, remoteAddress: remoteAddress)
|
||
default:
|
||
SDLLogger.log("[SDLContext] unsupported socket family: \(remoteAddress)", category: .context)
|
||
}
|
||
}
|
||
|
||
// MARK: 处理和Super之间的通讯
|
||
|
||
private func handleSuperDecision(_ decision: SuperDecision) async {
|
||
switch decision {
|
||
case .updateIPv6Assist(let assistInfo):
|
||
await self.stopCurrentIPv6AssistClient()
|
||
self.ipv6AssistContinuation.yield(assistInfo)
|
||
case .completeRegistration(let cipher):
|
||
await self.completeSuperRegistration(cipher: cipher)
|
||
case .failTunnel(let error):
|
||
await self.failTunnel(error)
|
||
case .publishTunnelEvent(let code, let message):
|
||
self.publishTunnelEvent(code: code, message: message)
|
||
case .sendSuper(let type, let data):
|
||
await self.superService.send(type: type, data: data)
|
||
case .sendPacket(let type, let data, let remoteAddress):
|
||
await self.sendPacket(type: type, data: data, remoteAddress: remoteAddress)
|
||
case .resolvePeerInfo(let peerInfo):
|
||
let packets = await self.puncherActor.makeRegisterPackets(peerInfo: peerInfo)
|
||
for packet in packets {
|
||
await self.sendPacket(type: .register, data: packet.data, remoteAddress: packet.remoteAddress)
|
||
}
|
||
case .removeSession(let dstMac):
|
||
await self.sessionManager.removeSession(dstMac: dstMac)
|
||
case .requestExposedService:
|
||
await self.requestExposedService()
|
||
case .shutdown(let message):
|
||
SDLLogger.fatal("[SDLContext] Super shutdown received: \(message)", category: .context)
|
||
self.publishTunnelEvent(message: message)
|
||
let error = NSError(domain: "com.jihe.punchnet.tun", code: -2)
|
||
await self.failTunnel(error)
|
||
case .applyPolicyResponse(let policyResponse):
|
||
await self.policyService.applyPolicyResponse(policyResponse)
|
||
case .applyExposedServiceResponse(let response):
|
||
await self.applyExposedServiceResponse(response)
|
||
case .handleARPResponse(let arpResponse):
|
||
await self.arpResolver.handleArpResponse(arpResponse: arpResponse)
|
||
}
|
||
}
|
||
|
||
private func completeSuperRegistration(cipher: CCDataCipher) async {
|
||
self.dataCipher = cipher
|
||
await self.packetOutboundActor.updateRuntime(config: self.config, dataCipher: cipher)
|
||
await self.packetInboundActor.updateRuntime(config: self.config, dataCipher: cipher)
|
||
|
||
do {
|
||
try await self.tunNetworkManager.apply(settings: .init(config: self.config), dnsServer: DNSHelper.dnsServer)
|
||
SDLLogger.log("[SDLContext] setNetworkSettings successed", category: .context)
|
||
await self.readySignal.succeed(())
|
||
} catch {
|
||
SDLLogger.fatal("[SDLContext] apply tunnel network settings failed: \(error)", category: .context)
|
||
SDLLogger.log("[SDLContext] setTunnelNetworkSettings get error: \(error)", category: .context)
|
||
await self.failTunnel(error)
|
||
}
|
||
}
|
||
|
||
private func failTunnel(_ error: Error) async {
|
||
SDLLogger.fatal("[SDLContext] failTunnel: \(error)", category: .context)
|
||
self.provider.cancelTunnelWithError(error)
|
||
await self.requestRootShutdown(error: error)
|
||
}
|
||
|
||
private func requestExposedService() async {
|
||
guard let requestData = await self.policyService.makeExposedServiceRequest() else {
|
||
return
|
||
}
|
||
|
||
await self.superService.send(type: .exposedServiceRequest, data: requestData)
|
||
}
|
||
|
||
private func applyExposedServiceResponse(_ response: SDLExposedServiceResponse) async {
|
||
guard let acl = await self.policyService.applyExposedServiceResponse(response) else {
|
||
return
|
||
}
|
||
|
||
self.config.acl = acl
|
||
}
|
||
|
||
// MARK: DNS service events
|
||
|
||
private func handleDNSEvent(_ event: DNSEvent) async {
|
||
switch event {
|
||
case .packet(let packet):
|
||
let nePacket = NEPacket(data: packet, protocolFamily: 2)
|
||
self.provider.packetFlow.writePacketObjects([nePacket])
|
||
}
|
||
}
|
||
|
||
// MARK: 处理从Hole收到的数据
|
||
|
||
private func handleUDPHoleControlEvent(_ event: SDLUDPHoleService.Event) async {
|
||
let decisions = self.holeControlPlane.handle(event)
|
||
for decision in decisions {
|
||
await self.handleHoleDecision(decision)
|
||
}
|
||
}
|
||
|
||
private func handleHoleDecision(_ decision: HoleDecision) async {
|
||
switch decision {
|
||
case .updateNatType(let natType):
|
||
self.setNatType(natType: natType)
|
||
case .sendPacket(let type, let data, let remoteAddress):
|
||
await self.sendPacket(type: type, data: data, remoteAddress: remoteAddress)
|
||
case .addSession(let session):
|
||
await self.sessionManager.addSession(session: session)
|
||
}
|
||
}
|
||
|
||
// MARK: 和Stun相关的心跳机制
|
||
|
||
private func runStunRequestOnce() async throws {
|
||
let probeReply = try? await self.ipv6AssistClient?.probe(requestTimeout: .seconds(3))
|
||
|
||
if let v6Info = probeReply?.v6Info, let v6Address = SDLUtil.ipv6DataToString(v6Info.v6) {
|
||
SDLLogger.log("[SDLContext] probe ipv6 address: \(v6Address)", category: .context)
|
||
} else {
|
||
SDLLogger.log("[SDLContext] probe ipv6 address: empty", category: .context)
|
||
}
|
||
|
||
await self.superControlPlane.sendStunRequest(natType: self.natType, v6Info: probeReply?.v6Info)
|
||
}
|
||
|
||
// MARK: NEPacketTunnelProvider相关的逻辑
|
||
|
||
// 取消出口节点的时候，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
|
||
}
|
||
await self.packetOutboundActor.updateRuntime(config: self.config, dataCipher: self.dataCipher)
|
||
await self.packetInboundActor.updateRuntime(config: self.config, dataCipher: self.dataCipher)
|
||
|
||
try await self.tunNetworkManager.apply(settings: .init(config: self.config), dnsServer: DNSHelper.dnsServer)
|
||
}
|
||
|
||
}
|
||
|
||
extension SDLContextActor {
|
||
|
||
private static func runRestarting(
|
||
name: String,
|
||
retryDelay: Duration = .seconds(5),
|
||
restartSignal: SDLWorkerRestartSignal? = nil,
|
||
operation: @escaping @Sendable () async throws -> Void
|
||
) async throws {
|
||
var restartGeneration = await restartSignal?.currentGeneration() ?? 0
|
||
|
||
while !Task.isCancelled {
|
||
do {
|
||
try Task.checkCancellation()
|
||
try await operation()
|
||
SDLLogger.log("[SDLContext] worker \(name) ended, will restart", category: .context)
|
||
} catch is CancellationError {
|
||
SDLLogger.log("[SDLContext] worker \(name) cancelled", category: .context)
|
||
throw CancellationError()
|
||
} catch {
|
||
SDLLogger.log("[SDLContext] worker \(name) crashed: \(error.localizedDescription), will restart", category: .context)
|
||
}
|
||
|
||
let nextGeneration = try await Self.waitForRestartSignalOrDelay(
|
||
name: name,
|
||
retryDelay: retryDelay,
|
||
restartSignal: restartSignal,
|
||
observedGeneration: restartGeneration
|
||
)
|
||
restartGeneration = nextGeneration
|
||
}
|
||
}
|
||
|
||
private static func waitForRestartSignalOrDelay(
|
||
name: String,
|
||
retryDelay: Duration,
|
||
restartSignal: SDLWorkerRestartSignal?,
|
||
observedGeneration: UInt64
|
||
) async throws -> UInt64 {
|
||
guard let restartSignal else {
|
||
try await Task.sleep(for: retryDelay)
|
||
return observedGeneration
|
||
}
|
||
|
||
return try await withThrowingTaskGroup(of: UInt64.self) { group in
|
||
group.addTask {
|
||
try await Task.sleep(for: retryDelay)
|
||
return await restartSignal.currentGeneration()
|
||
}
|
||
|
||
group.addTask {
|
||
return await restartSignal.waitForChange(after: observedGeneration)
|
||
}
|
||
|
||
let nextGeneration = try await group.next() ?? observedGeneration
|
||
group.cancelAll()
|
||
|
||
if nextGeneration != observedGeneration {
|
||
SDLLogger.log("[SDLContext] worker \(name) received restart signal", category: .context)
|
||
}
|
||
|
||
return nextGeneration
|
||
}
|
||
}
|
||
|
||
private static func runPeriodic(
|
||
name: String,
|
||
interval: Duration,
|
||
retryDelay: Duration = .seconds(5),
|
||
operation: @escaping @Sendable () async throws -> Void
|
||
) async throws {
|
||
while !Task.isCancelled {
|
||
do {
|
||
try Task.checkCancellation()
|
||
try await operation()
|
||
try await Task.sleep(for: interval)
|
||
} catch is CancellationError {
|
||
SDLLogger.log("[SDLContext] worker \(name) cancelled", category: .context)
|
||
throw CancellationError()
|
||
} catch {
|
||
SDLLogger.log("[SDLContext] worker \(name) crashed: \(error.localizedDescription), will retry", category: .context)
|
||
try await Task.sleep(for: retryDelay)
|
||
}
|
||
}
|
||
}
|
||
|
||
}
|