diff --git a/Tun/Punchnet/Inbound/PacketInboundActor.swift b/Tun/Punchnet/Inbound/PacketInboundActor.swift index 702a5c1..96847ce 100644 --- a/Tun/Punchnet/Inbound/PacketInboundActor.swift +++ b/Tun/Punchnet/Inbound/PacketInboundActor.swift @@ -49,10 +49,11 @@ actor PacketInboundActor { let processor = SDLHoleDataProcessor( networkAddress: self.networkAddress, dataCipher: self.dataCipher, - policyService: self.policyService + policySnapshot: self.policyService.policySnapshot(), + flowSessionManager: self.policyService.flowSessionManager ) - guard let plan = try? await processor.makeProcessingPlan(data: data) else { + guard let plan = try? processor.makeProcessingPlan(data: data) else { return } diff --git a/Tun/Punchnet/Inbound/SDLHoleDataProcessor.swift b/Tun/Punchnet/Inbound/SDLHoleDataProcessor.swift index 44202c9..293f0fc 100644 --- a/Tun/Punchnet/Inbound/SDLHoleDataProcessor.swift +++ b/Tun/Punchnet/Inbound/SDLHoleDataProcessor.swift @@ -23,17 +23,20 @@ final class SDLHoleDataProcessor { private let networkAddress: SDLConfiguration.NetworkAddress private let dataCipher: CCDataCipher? - private let policyService: PolicyService + private let policySnapshot: PolicySnapshot + private let flowSessionManager: SDLFlowSessionManager init(networkAddress: SDLConfiguration.NetworkAddress, dataCipher: CCDataCipher?, - policyService: PolicyService) { + policySnapshot: PolicySnapshot, + flowSessionManager: SDLFlowSessionManager) { self.networkAddress = networkAddress self.dataCipher = dataCipher - self.policyService = policyService + self.policySnapshot = policySnapshot + self.flowSessionManager = flowSessionManager } - func makeProcessingPlan(data: SDLData) async throws -> ProcessingPlan? { + func makeProcessingPlan(data: SDLData) throws -> ProcessingPlan? { guard let dataCipher = self.dataCipher else { return nil } @@ -52,7 +55,7 @@ final class SDLHoleDataProcessor { case .arp: return self.makeARPPlan(layerData: layerPacket.data, inboundBytes: inboundBytes) case .ipv4: - return await self.makeIPv4Plan(layerData: layerPacket.data, identityID: data.identityID, inboundBytes: inboundBytes) + return self.makeIPv4Plan(layerData: layerPacket.data, identityID: data.identityID, inboundBytes: inboundBytes) default: SDLLogger.log("[SDLContext] get invalid packet", for: .debug) return .init(inboundBytes: inboundBytes, action: .none) @@ -86,14 +89,23 @@ final class SDLHoleDataProcessor { return .init(inboundBytes: inboundBytes, action: .none) } - private func makeIPv4Plan(layerData: Data, identityID: UInt32, inboundBytes: Int) async -> ProcessingPlan { + private func makeIPv4Plan(layerData: Data, identityID: UInt32, inboundBytes: Int) -> ProcessingPlan { // 有数据是通过出口网关转发的,所有只判断是合法的ip包 guard let ipPacket = IPPacket(layerData) else { return .init(inboundBytes: inboundBytes, action: .none) } + if let reverseFlowSession = ipPacket.flowSession()?.reverse(), + self.flowSessionManager.hasSession(reverseFlowSession) { + self.flowSessionManager.updateSession(reverseFlowSession) + return .init( + inboundBytes: inboundBytes, + action: .writeToTun(packetData: ipPacket.data, identityID: identityID) + ) + } + // 检查权限逻辑 - if await self.policyService.checkPolicy(srcIdentityID: identityID, ipPacket: ipPacket) { + if self.policySnapshot.allows(srcIdentityID: identityID, ipPacket: ipPacket) { return .init( inboundBytes: inboundBytes, action: .writeToTun(packetData: ipPacket.data, identityID: identityID) diff --git a/Tun/Punchnet/Outbound/PacketOutboundActor.swift b/Tun/Punchnet/Outbound/PacketOutboundActor.swift index 7c8eb72..b5d7f86 100644 --- a/Tun/Punchnet/Outbound/PacketOutboundActor.swift +++ b/Tun/Punchnet/Outbound/PacketOutboundActor.swift @@ -116,7 +116,7 @@ actor PacketOutboundActor { let decision = router.route(packet: packet) if decision.shouldTrackFlow, let flowSession = packet.flowSession() { - await self.policyService.flowSessionManager.updateSession(flowSession) + self.policyService.flowSessionManager.updateSession(flowSession) } await self.handleTunRouteDecision(decision) @@ -147,7 +147,7 @@ actor PacketOutboundActor { func routeLayerPacket(dstMac: Data, type: LayerPacket.PacketType, data: Data) async { let forwarder = self.makeLayerPacketForwarder() - guard let plan = try? await forwarder.makeDeliveryPlan(dstMac: dstMac, type: type, data: data) else { + guard let plan = try? forwarder.makeDeliveryPlan(dstMac: dstMac, type: type, data: data) else { return } @@ -211,7 +211,7 @@ actor PacketOutboundActor { networkAddress: self.networkAddress, identityID: self.identityId, dataCipher: self.dataCipher, - sessionManager: self.sessionManager + sessionSnapshot: self.sessionManager.snapshot() ) } diff --git a/Tun/Punchnet/Outbound/SDLLayerPacketForwarder.swift b/Tun/Punchnet/Outbound/SDLLayerPacketForwarder.swift index 7922214..e6267b8 100644 --- a/Tun/Punchnet/Outbound/SDLLayerPacketForwarder.swift +++ b/Tun/Punchnet/Outbound/SDLLayerPacketForwarder.swift @@ -18,9 +18,9 @@ struct SDLLayerPacketForwarder { let networkAddress: SDLConfiguration.NetworkAddress let identityID: UInt32 let dataCipher: CCDataCipher? - let sessionManager: SessionManager + let sessionSnapshot: SessionSnapshot - func makeDeliveryPlan(dstMac: Data, type: LayerPacket.PacketType, data: Data) async throws -> DeliveryPlan? { + func makeDeliveryPlan(dstMac: Data, type: LayerPacket.PacketType, data: Data) throws -> DeliveryPlan? { guard let payload = try self.makePayload(dstMac: dstMac, type: type, data: data) else { return nil } @@ -29,7 +29,7 @@ struct SDLLayerPacketForwarder { return .superNode(payload: payload) } - if let session = await self.sessionManager.getSession(toAddress: dstMac) { + if let session = self.sessionSnapshot.getSession(toAddress: dstMac) { return .peer(payload: payload, session: session) } diff --git a/Tun/Punchnet/Policy/PolicyService.swift b/Tun/Punchnet/Policy/PolicyService.swift index b50b83a..1477f73 100644 --- a/Tun/Punchnet/Policy/PolicyService.swift +++ b/Tun/Punchnet/Policy/PolicyService.swift @@ -10,10 +10,10 @@ import Foundation actor PolicyService { // 处理权限控制 let identifyStore: IdentityStore - private let snapshotPublisher: SnapshotPublisher + nonisolated private let snapshotPublisher: SnapshotPublisher // Flow流会话管理, 过期时间为: 180秒 - let flowSessionManager = SDLFlowSessionManager(sessionTimeout: 180) + nonisolated let flowSessionManager = SDLFlowSessionManager(sessionTimeout: 180) // 当前节点的identityId值 let identityId: UInt32 @@ -57,6 +57,10 @@ actor PolicyService { return false } + + nonisolated func policySnapshot() -> PolicySnapshot { + return PolicySnapshot(identitySnapshot: self.snapshotPublisher.current()) + } func updatePolicy(superServiceProxy: SDLSuperServiceProxy) async { let requests = await self.identifyStore.makeBatchPolicyRequests(dstIdentityID: self.identityId) diff --git a/Tun/Punchnet/Policy/PolicySnapshot.swift b/Tun/Punchnet/Policy/PolicySnapshot.swift new file mode 100644 index 0000000..b97ffa7 --- /dev/null +++ b/Tun/Punchnet/Policy/PolicySnapshot.swift @@ -0,0 +1,36 @@ +// +// PolicySnapshot.swift +// Tun +// +// Created by Codex on 2026/5/20. +// + +import Foundation + +final class PolicySnapshot: Snapshot { + private let identitySnapshot: IdentitySnapshot + + init(identitySnapshot: IdentitySnapshot) { + self.identitySnapshot = identitySnapshot + } + + func allows(srcIdentityID: UInt32, ipPacket: IPPacket) -> Bool { + let ruleMap = self.identitySnapshot.lookup(srcIdentityID) + let proto = ipPacket.header.proto + + switch ipPacket.transportPacket { + case .tcp(let tcpPacket): + return ruleMap?.isAllow(proto: proto, port: tcpPacket.header.dstPort) ?? false + case .udp(let udpPacket): + return ruleMap?.isAllow(proto: proto, port: udpPacket.dstPort) ?? false + case .icmp: + return true + default: + return false + } + } + + static func empty() -> PolicySnapshot { + return PolicySnapshot(identitySnapshot: .empty()) + } +} diff --git a/Tun/Punchnet/Policy/SDLFlowSessionManager.swift b/Tun/Punchnet/Policy/SDLFlowSessionManager.swift index 2eec422..113b2aa 100644 --- a/Tun/Punchnet/Policy/SDLFlowSessionManager.swift +++ b/Tun/Punchnet/Policy/SDLFlowSessionManager.swift @@ -43,7 +43,7 @@ struct FlowSession: Hashable { } // MARK: - 会话管理器 -final class SDLFlowSessionManager { +final class SDLFlowSessionManager: @unchecked Sendable { private var sessions: [FlowSession: TimeInterval] = [:] private let lock = NSLock() private let sessionTimeout: TimeInterval diff --git a/Tun/Punchnet/SessionManager.swift b/Tun/Punchnet/SessionManager.swift index b25e08b..2d19a36 100644 --- a/Tun/Punchnet/SessionManager.swift +++ b/Tun/Punchnet/SessionManager.swift @@ -36,11 +36,38 @@ struct Session { } } +final class SessionSnapshot: Snapshot { + private let sessions: [Data: [Session.AddressType: Session]] + + init(sessions: [Data: [Session.AddressType: Session]]) { + self.sessions = sessions + } + + func getSession(toAddress: Data) -> Session? { + guard let peerSessions = self.sessions[toAddress] else { + return nil + } + + return peerSessions.values.max(by: { $0.lastTimestamp < $1.lastTimestamp }) + } + + static func empty() -> SessionSnapshot { + return SessionSnapshot(sessions: [:]) + } +} + actor SessionManager { private var sessions: [Data: [Session.AddressType: Session]] = [:] // session的有效时间 - private let ttl: Int32 = 10 + private let ttl: Int32 + nonisolated private let snapshotPublisher: SnapshotPublisher + + init() { + let ttl: Int32 = 10 + self.ttl = ttl + self.snapshotPublisher = SnapshotPublisher(initial: SessionSnapshot.empty()) + } func getSession(toAddress: Data) -> Session? { let timestamp = Int32(Date().timeIntervalSince1970) @@ -52,11 +79,13 @@ actor SessionManager { peerSessions = peerSessions.filter { $0.value.lastTimestamp + ttl >= timestamp } guard !peerSessions.isEmpty else { self.sessions.removeValue(forKey: toAddress) + self.publishSnapshot() return nil } guard var session = self.selectSession(in: peerSessions) else { self.sessions[toAddress] = peerSessions + self.publishSnapshot() return nil } @@ -64,6 +93,7 @@ actor SessionManager { peerSessions[session.addressType] = session self.sessions[toAddress] = peerSessions + self.publishSnapshot() return session } @@ -78,18 +108,38 @@ actor SessionManager { sessions[session.addressType] = session self.sessions[session.dstMac] = sessions + self.publishSnapshot() } func removeSession(dstMac: Data) { self.sessions.removeValue(forKey: dstMac) + self.publishSnapshot() } func clear() { self.sessions.removeAll() + self.publishSnapshot() + } + + nonisolated func snapshot() -> SessionSnapshot { + return self.snapshotPublisher.current() } private func selectSession(in sessions: [Session.AddressType: Session]) -> Session? { return sessions.values.max(by: { $0.lastTimestamp < $1.lastTimestamp }) } + + private func publishSnapshot() { + self.snapshotPublisher.publish(self.compileSnapshot()) + } + + private func compileSnapshot() -> SessionSnapshot { + let timestamp = Int32(Date().timeIntervalSince1970) + let sessions = self.sessions.compactMapValues { peerSessions in + let validSessions = peerSessions.filter { $0.value.lastTimestamp + self.ttl >= timestamp } + return validSessions.isEmpty ? nil : validSessions + } + return SessionSnapshot(sessions: sessions) + } } diff --git a/Tun/Punchnet/Snapshot/SnapshotPublisher.swift b/Tun/Punchnet/Snapshot/SnapshotPublisher.swift index 2714761..fdaee29 100644 --- a/Tun/Punchnet/Snapshot/SnapshotPublisher.swift +++ b/Tun/Punchnet/Snapshot/SnapshotPublisher.swift @@ -6,7 +6,7 @@ // import Atomics -final class SnapshotPublisher { +final class SnapshotPublisher: @unchecked Sendable { private let atomic: ManagedAtomic> init(initial snapshot: S) { @@ -15,17 +15,17 @@ final class SnapshotPublisher { func publish(_ snapshot: S) { let newRef = Unmanaged.passRetained(snapshot) - let oldRef = atomic.exchange(newRef, ordering: .acquiring) + let oldRef = atomic.exchange(newRef, ordering: .releasing) oldRef.release() } @inline(__always) func current() -> S { - atomic.load(ordering: .relaxed).takeUnretainedValue() + atomic.load(ordering: .acquiring).takeUnretainedValue() } deinit { - let ref = atomic.load(ordering: .relaxed) + let ref = atomic.load(ordering: .acquiring) ref.release() }