anlicheng/punchnet-macos
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

修复热路径

Browse Source
This commit is contained in:
anlicheng 2026-05-20 21:55:58 +08:00
parent fbeb8d1a09
commit 76b5d3212f
9 changed files with 126 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
Tun/Punchnet/Inbound/PacketInboundActor.swift
View File

@ -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
}

26
Tun/Punchnet/Inbound/SDLHoleDataProcessor.swift
View File

@ -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)

6
Tun/Punchnet/Outbound/PacketOutboundActor.swift
View File

@ -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()
)
}

6
Tun/Punchnet/Outbound/SDLLayerPacketForwarder.swift
View File

@ -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)
}

8
Tun/Punchnet/Policy/PolicyService.swift
View File

@ -10,10 +10,10 @@ import Foundation
actor PolicyService {
// 处理权限控制
let identifyStore: IdentityStore
private let snapshotPublisher: SnapshotPublisher<IdentitySnapshot>
nonisolated private let snapshotPublisher: SnapshotPublisher<IdentitySnapshot>
// 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)

36
Tun/Punchnet/Policy/PolicySnapshot.swift Normal file
View File

@ -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())
}
}

2
Tun/Punchnet/Policy/SDLFlowSessionManager.swift
View File

@ -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

52
Tun/Punchnet/SessionManager.swift
View File

@ -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<SessionSnapshot>
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)
}
}

8
Tun/Punchnet/Snapshot/SnapshotPublisher.swift
View File

@ -6,7 +6,7 @@
//
import Atomics
final class SnapshotPublisher<S: Snapshot> {
final class SnapshotPublisher<S: Snapshot>: @unchecked Sendable {
private let atomic: ManagedAtomic<Unmanaged<S>>
init(initial snapshot: S) {
@ -15,17 +15,17 @@ final class SnapshotPublisher<S: Snapshot> {
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()
}