fix policy

Tun/Punchnet/Context/SDLContextActor.swift

@@ -94,12 +94,8 @@ actor SDLContextActor {
     nonisolated private let provider: NEPacketTunnelProvider
     
     // 处理权限控制
-    private let identifyStore: IdentityStore
     private var updatePolicyTask: Task<Void, Never>?
-    private let snapshotPublisher: SnapshotPublisher<IdentitySnapshot>
+    private let policyService: PolicyService
-    
-    // Flow流会话管理， 过期时间为: 180秒
-    private let flowSessionManager = SDLFlowSessionManager(sessionTimeout: 180)
     
     // stunRequest任务
     private var stunRequestTask: Task<Void, Never>?
@@ -115,9 +111,7 @@ actor SDLContextActor {
         self.arpServer = ArpServer()
 
         // 权限控制
-        let snapshotPublisher = SnapshotPublisher(initial: IdentitySnapshot.empty())
+        self.policyService = PolicyService(identityId: config.identityId)
-        self.identifyStore = IdentityStore(publisher: snapshotPublisher)
-        self.snapshotPublisher = snapshotPublisher
     }
     
     public func start() async {
@@ -152,13 +146,12 @@ actor SDLContextActor {
         await self.arpServer.stop()
         await self.sessionManager.clear()
 
-        self.flowSessionManager.clear()
-
         self.stunRequestTask?.cancel()
         self.stunRequestTask = nil
 
         self.updatePolicyTask?.cancel()
         self.updatePolicyTask = nil
+        await self.policyService.clear()
         
         let packetReaderService = self.packetReaderService
         self.packetReaderService = nil
@@ -268,7 +261,7 @@ extension SDLContextActor {
             await self.handleEvent(event: event)
         case .policyReponse(let policyResponse):
             // 处理权限的请求问题
-            await self.identifyStore.applyPolicyResponse(policyResponse)
+            await self.policyService.identifyStore.applyPolicyResponse(policyResponse)
         case .arpResponse(let arpResponse):
             SDLLogger.log("[SDLContext] get arp response: \(arpResponse)")
             await self.arpServer.handleArpResponse(arpResponse: arpResponse)
@@ -327,10 +320,7 @@ extension SDLContextActor {
                 while true {
                     try await Task.sleep(for: .seconds(300))
                     SDLLogger.log("[SDLContext] updatePolicyTask execute")
-                    let requests = await self.identifyStore.makeBatchPolicyRequests(dstIdentityID: self.config.identityId)
+                    await self.policyService.updatePolicy(superService: self.superService)
-                    for request in requests {
-                        await self.superService?.send(type: .policyRequest, data: request)
-                    }
                 }
             } catch let err {
                 SDLLogger.log("[SDLContext] updatePolicyTask stop with err: \(err)")
@@ -490,14 +480,12 @@ extension SDLContextActor {
         return .init(
             networkAddress: self.config.networkAddress,
             dataCipher: self.dataCipher,
-            snapshotPublisher: self.snapshotPublisher,
+            policyService: self.policyService)
-            flowSessionManager: self.flowSessionManager
-        )
     }
     
     private func handleHoleData(data: SDLData) async throws {
         let processor = self.makeHoleDataProcessor()
-        guard let plan = try processor.makeProcessingPlan(data: data) else {
+        guard let plan = try await processor.makeProcessingPlan(data: data) else {
             return
         }
         
@@ -516,7 +504,7 @@ extension SDLContextActor {
             SDLLogger.log("[SDLContext] hole identity: \(identityID), allow, data count: \(packetData.count)", for: .trace)
         case .requestPolicy(let srcIdentityID):
             SDLLogger.log("[SDLContext] not found identity: \(srcIdentityID) ruleMap", for: .debug)
-            if let queryData = await self.identifyStore.makePolicyRequest(srcIdentityId: srcIdentityID, dstIdentityId: self.config.identityId) {
+            if let queryData = await self.policyService.identifyStore.makePolicyRequest(srcIdentityId: srcIdentityID, dstIdentityId: self.config.identityId) {
                 await self.superService?.send(type: .policyRequest, data: queryData)
             }
         case .none:
@@ -692,7 +680,7 @@ extension SDLContextActor {
         // 外部出去的数据，需要建立FlowSession
         // 外部数据进来的时候需要查找
         if decision.shouldTrackFlow, let flowSession = packet.flowSession() {
-            self.flowSessionManager.updateSession(flowSession)
+            await self.policyService.flowSessionManager.updateSession(flowSession)
             //SDLLogger.shared.log("[SDLContext] flow_session: \(flowSession)", level: .debug)
         }
         

Tun/Punchnet/Context/SDLHoleDataProcessor.swift

@@ -23,20 +23,17 @@ final class SDLHoleDataProcessor {
     
     private let networkAddress: SDLConfiguration.NetworkAddress
     private let dataCipher: CCDataCipher?
-    private let snapshotPublisher: SnapshotPublisher<IdentitySnapshot>
+    private let policyService: PolicyService
-    private let flowSessionManager: SDLFlowSessionManager
     
     init(networkAddress: SDLConfiguration.NetworkAddress,
          dataCipher: CCDataCipher?,
-         snapshotPublisher: SnapshotPublisher<IdentitySnapshot>,
+         policyService: PolicyService) {
-         flowSessionManager: SDLFlowSessionManager) {
         self.networkAddress = networkAddress
         self.dataCipher = dataCipher
-        self.snapshotPublisher = snapshotPublisher
+        self.policyService = policyService
-        self.flowSessionManager = flowSessionManager
     }
     
-    func makeProcessingPlan(data: SDLData) throws -> ProcessingPlan? {
+    func makeProcessingPlan(data: SDLData) async throws -> ProcessingPlan? {
         guard let dataCipher = self.dataCipher else {
             return nil
         }
@@ -55,7 +52,7 @@ final class SDLHoleDataProcessor {
         case .arp:
             return self.makeARPPlan(layerData: layerPacket.data, inboundBytes: inboundBytes)
         case .ipv4:
-            return self.makeIPv4Plan(layerData: layerPacket.data, identityID: data.identityID, inboundBytes: inboundBytes)
+            return await 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)
@@ -89,17 +86,14 @@ final class SDLHoleDataProcessor {
         return .init(inboundBytes: inboundBytes, action: .none)
     }
     
-    private func makeIPv4Plan(layerData: Data, identityID: UInt32, inboundBytes: Int) -> ProcessingPlan {
+    private func makeIPv4Plan(layerData: Data, identityID: UInt32, inboundBytes: Int) async -> ProcessingPlan {
         // 有数据是通过出口网关转发的，所有只判断是合法的ip包
         guard let ipPacket = IPPacket(layerData) else {
             return .init(inboundBytes: inboundBytes, action: .none)
         }
         
         // 检查权限逻辑
-        let identitySnapshot = self.snapshotPublisher.current()
+        if await self.policyService.checkPolicy(srcIdentityID: identityID, ipPacket: ipPacket) {
-        let ruleMap = identitySnapshot.lookup(identityID)
-        
-        if true || self.checkPolicy(ipPacket: ipPacket, ruleMap: ruleMap) {
             return .init(
                 inboundBytes: inboundBytes,
                 action: .writeToTun(packetData: ipPacket.data, identityID: identityID)
@@ -111,33 +105,4 @@ final class SDLHoleDataProcessor {
             action: .requestPolicy(srcIdentityID: identityID)
         )
     }
-    
-    private func checkPolicy(ipPacket: IPPacket, ruleMap: IdentityRuleMap?) -> Bool {
-        // 进来的数据反转一下，然后再处理
-        if let reverseFlowSession = ipPacket.flowSession()?.reverse(),
-           self.flowSessionManager.hasSession(reverseFlowSession) {
-            self.flowSessionManager.updateSession(reverseFlowSession)
-            return true
-        }
-        
-        // 检查权限逻辑
-        let proto = ipPacket.header.proto
-        // 优先判断访问规则
-        switch ipPacket.transportPacket {
-        case .tcp(let tcpPacket):
-            if let ruleMap, ruleMap.isAllow(proto: proto, port: tcpPacket.header.dstPort) {
-                return true
-            }
-        case .udp(let udpPacket):
-            if let ruleMap, ruleMap.isAllow(proto: proto, port: udpPacket.dstPort) {
-                return true
-            }
-        case .icmp(_):
-            return true
-        default:
-            return false
-        }
-        
-        return false
-    }
 }

Tun/Punchnet/Policy/PolicyService.swift

@@ -0,0 +1,80 @@
+//
+//  PolicyService.swift
+//  punchnet
+//
+//  Created by 安礼成 on 2026/5/19.
+//
+
+import Foundation
+
+actor PolicyService {
+    // 处理权限控制
+    let identifyStore: IdentityStore
+    private let snapshotPublisher: SnapshotPublisher<IdentitySnapshot>
+    
+    // Flow流会话管理， 过期时间为: 180秒
+    let flowSessionManager = SDLFlowSessionManager(sessionTimeout: 180)
+    
+    // 当前节点的identityId值
+    let identityId: UInt32
+    
+    init(identityId: UInt32) {
+        self.identityId = identityId
+        // 权限控制
+        let snapshotPublisher = SnapshotPublisher(initial: IdentitySnapshot.empty())
+        self.identifyStore = IdentityStore(publisher: snapshotPublisher)
+        self.snapshotPublisher = snapshotPublisher
+    }
+    
+    func checkPolicy(srcIdentityID: UInt32, ipPacket: IPPacket) -> Bool {
+        // 进来的数据反转一下，然后再处理
+        if let reverseFlowSession = ipPacket.flowSession()?.reverse(),
+           self.flowSessionManager.hasSession(reverseFlowSession) {
+            self.flowSessionManager.updateSession(reverseFlowSession)
+            return true
+        }
+        
+        // 检查权限逻辑
+        let identitySnapshot = self.snapshotPublisher.current()
+        let ruleMap = identitySnapshot.lookup(srcIdentityID)
+        // 检查权限逻辑
+        let proto = ipPacket.header.proto
+        // 优先判断访问规则
+        switch ipPacket.transportPacket {
+        case .tcp(let tcpPacket):
+            if let ruleMap, ruleMap.isAllow(proto: proto, port: tcpPacket.header.dstPort) {
+                return true
+            }
+        case .udp(let udpPacket):
+            if let ruleMap, ruleMap.isAllow(proto: proto, port: udpPacket.dstPort) {
+                return true
+            }
+        case .icmp(_):
+            return true
+        default:
+            return false
+        }
+        
+        return false
+    }
+    
+    func updatePolicy(superService: SDLSuperService?) async {
+        guard let superService else {
+            return
+        }
+        
+        let requests = await self.identifyStore.makeBatchPolicyRequests(dstIdentityID: self.identityId)
+        for request in requests {
+            await superService.send(type: .policyRequest, data: request)
+        }
+    }
+    
+    func clear() {
+        self.flowSessionManager.clear()
+    }
+    
+    deinit {
+        SDLLogger.log("[PolicyService] deinit")
+    }
+    
+}