//===----------------------------------------------------------------------===// // Copyright © 2025 Apple Inc. and the container project authors. All rights reserved. // // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. // You may obtain a copy of the License at // // https://www.apache.org/licenses/LICENSE-2.0 // // Unless required by applicable law or agreed to in writing, software // distributed under the License is distributed on an "AS IS" BASIS, // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. // See the License for the specific language governing permissions and // limitations under the License. //===----------------------------------------------------------------------===// import Collections import Foundation import Logging import NIO import NIOFoundationCompat import Synchronization // Proxy backend for a single client address (clientIP, clientPort). private final class UDPProxyBackend: ChannelInboundHandler, Sendable { typealias InboundIn = AddressedEnvelope typealias OutboundOut = AddressedEnvelope private struct State { var queuedPayloads: Deque var channel: (any Channel)? } private let clientAddress: SocketAddress private let serverAddress: SocketAddress private let frontendChannel: any Channel private let log: Logger? private let state: Mutex init(clientAddress: SocketAddress, serverAddress: SocketAddress, frontendChannel: any Channel, log: Logger? = nil) { self.clientAddress = clientAddress self.serverAddress = serverAddress self.frontendChannel = frontendChannel self.log = log let initialState = State(queuedPayloads: Deque(), channel: nil) self.state = Mutex(initialState) } func channelRead(context: ChannelHandlerContext, data: NIOAny) { // relay data from server to client. let inbound = self.unwrapInboundIn(data) let outbound = OutboundOut(remoteAddress: self.clientAddress, data: inbound.data) self.log?.trace("backend - writing datagram to client") _ = self.frontendChannel.writeAndFlush(outbound) } func channelActive(context: ChannelHandlerContext) { state.withLock { if !$0.queuedPayloads.isEmpty { self.log?.trace("backend - writing \($0.queuedPayloads.count) queued datagrams to server") while let queuedData = $0.queuedPayloads.popFirst() { let outbound: UDPProxyBackend.OutboundOut = OutboundOut(remoteAddress: self.serverAddress, data: queuedData) _ = context.channel.writeAndFlush(outbound) } } $0.channel = context.channel } } func write(data: ByteBuffer) { // change package remote address from proxy server to real server state.withLock { if let channel = $0.channel { // channel has been initialized, so relay any queued packets, along with this one to outbound self.log?.trace("backend - writing datagram to server") let outbound: UDPProxyBackend.OutboundOut = OutboundOut(remoteAddress: self.serverAddress, data: data) _ = channel.writeAndFlush(outbound) } else { // channel is initializing, queue self.log?.trace("backend - queuing datagram") $0.queuedPayloads.append(data) } } } func close() { state.withLock { guard let channel = $0.channel else { self.log?.warning("backend - close on inactive channel") return } _ = channel.close() } } } private struct ProxyContext { public let proxy: UDPProxyBackend public let closeFuture: EventLoopFuture } private final class UDPProxyFrontend: ChannelInboundHandler, Sendable { typealias InboundIn = AddressedEnvelope typealias OutboundOut = AddressedEnvelope private let maxProxies = UInt(256) private let proxyAddress: SocketAddress private let serverAddress: SocketAddress private let eventLoopGroup: any EventLoopGroup private let log: Logger? private let proxies: Mutex> init(proxyAddress: SocketAddress, serverAddress: SocketAddress, eventLoopGroup: any EventLoopGroup, log: Logger? = nil) { self.proxyAddress = proxyAddress self.serverAddress = serverAddress self.eventLoopGroup = eventLoopGroup self.proxies = Mutex(LRUCache(size: maxProxies)) self.log = log } func channelRead(context: ChannelHandlerContext, data: NIOAny) { let inbound = self.unwrapInboundIn(data) guard let clientIP = inbound.remoteAddress.ipAddress else { log?.error("frontend - no client IP address in inbound payload") return } guard let clientPort = inbound.remoteAddress.port else { log?.error("frontend - no client port in inbound payload") return } let key = "\(clientIP):\(clientPort)" do { try proxies.withLock { if let context = $0.get(key) { context.proxy.write(data: inbound.data) } else { self.log?.trace("frontend - creating backend") let proxy = UDPProxyBackend( clientAddress: inbound.remoteAddress, serverAddress: self.serverAddress, frontendChannel: context.channel, log: log ) let proxyAddress = try SocketAddress(ipAddress: "0.0.0.0", port: 0) let proxyToServerFuture = DatagramBootstrap(group: self.eventLoopGroup) .channelInitializer { self.log?.trace("frontend - initializing backend") return $0.pipeline.addHandler(proxy) } .bind(to: proxyAddress) .flatMap { $0.closeFuture } let context = ProxyContext(proxy: proxy, closeFuture: proxyToServerFuture) if let (_, evictedContext) = $0.put(key: key, value: context) { self.log?.trace("frontend - closing evicted backend") evictedContext.proxy.close() } proxy.write(data: inbound.data) } } } catch { log?.error("server handler - backend channel creation failed with error: \(error)") return } } } public struct UDPForwarder: SocketForwarder { private let proxyAddress: SocketAddress private let serverAddress: SocketAddress private let eventLoopGroup: any EventLoopGroup private let log: Logger? public init( proxyAddress: SocketAddress, serverAddress: SocketAddress, eventLoopGroup: any EventLoopGroup, log: Logger? = nil ) throws { self.proxyAddress = proxyAddress self.serverAddress = serverAddress self.eventLoopGroup = eventLoopGroup self.log = log } public func run() throws -> EventLoopFuture { self.log?.trace("frontend - creating channel") let proxyToServerHandler = UDPProxyFrontend( proxyAddress: proxyAddress, serverAddress: serverAddress, eventLoopGroup: self.eventLoopGroup, log: log ) let bootstrap = DatagramBootstrap(group: self.eventLoopGroup) .channelInitializer { serverChannel in self.log?.trace("frontend - initializing channel") return serverChannel.pipeline.addHandler(proxyToServerHandler) } return bootstrap .bind(to: proxyAddress) .flatMap { $0.eventLoop.makeSucceededFuture(SocketForwarderResult(channel: $0)) } } }