diff --git a/Sources/Containerization/LinuxContainer.swift b/Sources/Containerization/LinuxContainer.swift index 34964fcdc..5860a645d 100644 --- a/Sources/Containerization/LinuxContainer.swift +++ b/Sources/Containerization/LinuxContainer.swift @@ -712,8 +712,11 @@ extension LinuxContainer { fileMountContextHolder.withLock { $0 = ctx } } - // Start up our friendly unix socket relays. - for socket in self.config.sockets { + // Sockets relayed into the container must be staged before the + // container process starts so their bind mounts can be added to + // the runtime spec. Outbound relays are started after the process + // exists, when its mount namespace is available. + for socket in self.config.sockets where socket.direction == .into { try await self.relayUnixSocket( socket: socket, relayManager: relayManager, @@ -834,8 +837,21 @@ extension LinuxContainer { ) try await process.start() + // Resolve outbound sockets through the running container's mount + // namespace. Looking beneath the static rootfs cannot see tmpfs or + // other mounts created by the OCI runtime. + for socket in self.config.sockets where socket.direction == .outOf { + try await self.relayUnixSocket( + socket: socket, + relayManager: createdState.relayManager, + agent: agent, + containerPID: process.pid + ) + } + state = .started(.init(createdState, process: process)) } catch { + try? await createdState.relayManager.stopAll() try? await agent.close() try? await createdState.vm.stop() state.setErrored(error: error) @@ -1143,7 +1159,8 @@ extension LinuxContainer { private func relayUnixSocket( socket: UnixSocketConfiguration, relayManager: UnixSocketRelayManager, - agent: any VirtualMachineAgent + agent: any VirtualMachineAgent, + containerPID: Int32? = nil ) async throws { guard let relayAgent = agent as? SocketRelayAgent else { throw ContainerizationError( @@ -1153,19 +1170,32 @@ extension LinuxContainer { } var socket = socket - let rootInGuest = URL(filePath: self.root) let port: UInt32 if socket.direction == .into { port = self.hostVsockPorts.wrappingAdd(1, ordering: .relaxed).oldValue socket.destination = URL(filePath: Self.guestSocketStagingPath(socket.id)) } else { + guard let containerPID, containerPID > 0 else { + throw ContainerizationError( + .invalidState, + message: "cannot start outbound socket relay before the container process" + ) + } port = self.guestVsockPorts.wrappingAdd(1, ordering: .relaxed).oldValue - socket.source = rootInGuest.appending(path: socket.source.path) } try await relayManager.start(port: port, socket: socket) - try await relayAgent.relaySocket(port: port, configuration: socket) + do { + try await relayAgent.relaySocket( + port: port, + configuration: socket, + containerPID: containerPID + ) + } catch { + try? await relayManager.stop(socket: socket) + throw error + } } /// Default chunk size for file transfers (1MiB). diff --git a/Sources/Containerization/LinuxPod.swift b/Sources/Containerization/LinuxPod.swift index 6275a4d49..b214a20ce 100644 --- a/Sources/Containerization/LinuxPod.swift +++ b/Sources/Containerization/LinuxPod.swift @@ -514,7 +514,7 @@ extension LinuxPod { ) } - for socket in config.sockets { + for socket in config.sockets where socket.direction == .into { try await self.relayUnixSocket( socket: socket, containerID: id, @@ -771,7 +771,7 @@ extension LinuxPod { // Start up unix socket relays for each container for (_, container) in containers { - for socket in container.config.sockets { + for socket in container.config.sockets where socket.direction == .into { try await self.relayUnixSocket( socket: socket, containerID: container.id, @@ -859,6 +859,9 @@ extension LinuxPod { } let agent = try await createdState.vm.dialAgent() + var process: LinuxProcess? + var processStarted = false + var startedRelays: [UnixSocketConfiguration] = [] do { var spec = self.generateRuntimeSpec(containerID: containerID, config: container.config, rootfs: container.rootfs) // We don't need the rootfs, nor do OCI runtimes want it included. @@ -957,7 +960,7 @@ extension LinuxPod { stderr: container.config.process.stderr ) - let process = LinuxProcess( + let newProcess = LinuxProcess( containerID, containerID: containerID, spec: spec, @@ -967,14 +970,68 @@ extension LinuxPod { vm: createdState.vm, logger: self.logger ) - try await process.start() + process = newProcess + try await newProcess.start() + processStarted = true + + for socket in container.config.sockets where socket.direction == .outOf { + try await self.relayUnixSocket( + socket: socket, + containerID: containerID, + relayManager: createdState.relayManager, + agent: agent, + containerPID: newProcess.pid + ) + startedRelays.append(socket) + } - container.process = process + container.process = newProcess container.state = .started state.containers[containerID] = container } catch { + let startError = error + var rollbackError: Error? + for socket in startedRelays { + do { + try await createdState.relayManager.stop(socket: socket, owner: containerID) + } catch { + rollbackError = rollbackError ?? error + } + if let relayAgent = agent as? SocketRelayAgent { + do { + try await relayAgent.stopSocketRelay(configuration: socket) + } catch { + rollbackError = rollbackError ?? error + } + } + } + if let process { + if processStarted { + do { + try await process.kill(.kill) + _ = try await process.wait(timeoutInSeconds: 3) + } catch { + rollbackError = rollbackError ?? error + } + } + do { + try await process.delete() + } catch { + rollbackError = rollbackError ?? error + } + } try? await agent.close() - throw error + container.process = nil + container.state = .errored + state.containers[containerID] = container + if let rollbackError { + throw ContainerizationError( + .internalError, + message: "failed to roll back container \(containerID) after start error: \(startError)", + cause: rollbackError + ) + } + throw startError } } } @@ -998,6 +1055,18 @@ extension LinuxPod { // Handle containers that were hotplugged but never started if container.state == .created { + if createdState.vm.state == .stopped { + try? await createdState.relayManager.stopAll(owner: containerID) + } else { + try await createdState.vm.withAgent { agent in + try await self.stopUnixSocketRelays( + containerID: containerID, + relayManager: createdState.relayManager, + agent: agent + ) + } + } + // Release the hotplug device and virtiofs shares try? await createdState.vm.releaseHotplug(id: containerID) try? await createdState.vm.releaseVirtioFS(id: containerID) @@ -1014,17 +1083,36 @@ extension LinuxPod { ) } + // Check if the vm is even still running + if createdState.vm.state == .stopped { + try? await createdState.relayManager.stopAll(owner: containerID) + container.process = nil + container.state = .stopped + state.containers[containerID] = container + return + } + + var firstError: Error? do { - // Check if the vm is even still running - if createdState.vm.state == .stopped { - container.state = .stopped - state.containers[containerID] = container - return + try await createdState.vm.withAgent { agent in + try await self.stopUnixSocketRelays( + containerID: containerID, + relayManager: createdState.relayManager, + agent: agent + ) } + } catch { + firstError = error + } + do { try await process.kill(.kill) try await process.wait(timeoutInSeconds: 3) + } catch { + firstError = firstError ?? error + } + do { try await createdState.vm.withAgent { agent in // Unmount the rootfs try await agent.umount( @@ -1032,27 +1120,37 @@ extension LinuxPod { flags: 0 ) } + } catch { + firstError = firstError ?? error + } - // Release the hotplug device and virtiofs shares so they can be reused by new containers + // Release the hotplug device and virtiofs shares so they can be reused by new containers + do { try await createdState.vm.releaseHotplug(id: containerID) + } catch { + firstError = firstError ?? error + } + do { try await createdState.vm.releaseVirtioFS(id: containerID) + } catch { + firstError = firstError ?? error + } - // Clean up the process resources + // Clean up the process resources + do { try await process.delete() - - container.process = nil - container.state = .stopped - state.containers[containerID] = container } catch { - // Try to release the hotplug device and virtiofs shares even on error - try? await createdState.vm.releaseHotplug(id: containerID) - try? await createdState.vm.releaseVirtioFS(id: containerID) + firstError = firstError ?? error + } + container.process = nil + if let firstError { container.state = .errored - container.process = nil state.containers[containerID] = container - - throw error + throw firstError + } else { + container.state = .stopped + state.containers[containerID] = container } } } @@ -1309,7 +1407,7 @@ extension LinuxPod { try await self.state.withLock { state in let createdState = try state.phase.createdState("relayUnixSocket") - guard let _ = state.containers[containerID] else { + guard let container = state.containers[containerID] else { throw ContainerizationError( .notFound, message: "container \(containerID) not found in pod" @@ -1321,7 +1419,8 @@ extension LinuxPod { socket: socket, containerID: containerID, relayManager: createdState.relayManager, - agent: agent + agent: agent, + containerPID: container.process?.pid ) } } @@ -1331,7 +1430,8 @@ extension LinuxPod { socket: UnixSocketConfiguration, containerID: String, relayManager: UnixSocketRelayManager, - agent: any VirtualMachineAgent + agent: any VirtualMachineAgent, + containerPID: Int32? = nil ) async throws { guard let relayAgent = agent as? SocketRelayAgent else { throw ContainerizationError( @@ -1342,19 +1442,64 @@ extension LinuxPod { var socket = socket - // Adjust paths to be relative to the container's rootfs - let rootInGuest = URL(filePath: Self.guestRootfsPath(containerID)) - let port: UInt32 if socket.direction == .into { port = self.hostVsockPorts.wrappingAdd(1, ordering: .relaxed).oldValue socket.destination = URL(filePath: Self.guestSocketStagingPath(socket.id)) } else { + guard let containerPID, containerPID > 0 else { + throw ContainerizationError( + .invalidState, + message: "cannot start outbound socket relay before the container process" + ) + } port = self.guestVsockPorts.wrappingAdd(1, ordering: .relaxed).oldValue - socket.source = rootInGuest.appending(path: socket.source.path) } - try await relayManager.start(port: port, socket: socket) - try await relayAgent.relaySocket(port: port, configuration: socket) + try await relayManager.start(port: port, socket: socket, owner: containerID) + do { + try await relayAgent.relaySocket( + port: port, + configuration: socket, + containerPID: containerPID + ) + } catch { + try? await relayManager.stop(socket: socket, owner: containerID) + throw error + } + } + + private func stopUnixSocketRelays( + containerID: String, + relayManager: UnixSocketRelayManager, + agent: any VirtualMachineAgent + ) async throws { + guard let relayAgent = agent as? SocketRelayAgent else { + throw ContainerizationError( + .unsupported, + message: "VirtualMachineAgent does not support relaySocket surface" + ) + } + + let sockets = await relayManager.sockets(owner: containerID) + var firstError: Error? + + do { + try await relayManager.stopAll(owner: containerID) + } catch { + firstError = error + } + + for socket in sockets { + do { + try await relayAgent.stopSocketRelay(configuration: socket) + } catch { + firstError = firstError ?? error + } + } + + if let firstError { + throw firstError + } } } diff --git a/Sources/Containerization/SandboxContext/SandboxContext.grpc.swift b/Sources/Containerization/SandboxContext/SandboxContext.grpc.swift index 6776ed1b3..442320f42 100644 --- a/Sources/Containerization/SandboxContext/SandboxContext.grpc.swift +++ b/Sources/Containerization/SandboxContext/SandboxContext.grpc.swift @@ -6830,4 +6830,4 @@ extension Com_Apple_Containerization_Sandbox_V3_SandboxContext.ClientProtocol { onResponse: handleResponse ) } -} \ No newline at end of file +} diff --git a/Sources/Containerization/SandboxContext/SandboxContext.pb.swift b/Sources/Containerization/SandboxContext/SandboxContext.pb.swift index a091d8b25..2b2ef51b7 100644 --- a/Sources/Containerization/SandboxContext/SandboxContext.pb.swift +++ b/Sources/Containerization/SandboxContext/SandboxContext.pb.swift @@ -239,6 +239,15 @@ public nonisolated struct Com_Apple_Containerization_Sandbox_V3_ProxyVsockReques public var action: Com_Apple_Containerization_Sandbox_V3_ProxyVsockRequest.Action = .into + public var containerPid: Int32 { + get {_containerPid ?? 0} + set {_containerPid = newValue} + } + /// Returns true if `containerPid` has been explicitly set. + public var hasContainerPid: Bool {self._containerPid != nil} + /// Clears the value of `containerPid`. Subsequent reads from it will return its default value. + public mutating func clearContainerPid() {self._containerPid = nil} + public var unknownFields = SwiftProtobuf.UnknownStorage() public nonisolated enum Action: SwiftProtobuf.Enum, Swift.CaseIterable { @@ -278,6 +287,7 @@ public nonisolated struct Com_Apple_Containerization_Sandbox_V3_ProxyVsockReques public init() {} fileprivate var _guestSocketPermissions: UInt32? = nil + fileprivate var _containerPid: Int32? = nil } public nonisolated struct Com_Apple_Containerization_Sandbox_V3_ProxyVsockResponse: Sendable { @@ -1985,7 +1995,7 @@ nonisolated extension Com_Apple_Containerization_Sandbox_V3_SysctlResponse: Swif nonisolated extension Com_Apple_Containerization_Sandbox_V3_ProxyVsockRequest: SwiftProtobuf.Message, SwiftProtobuf._MessageImplementationBase, SwiftProtobuf._ProtoNameProviding { public static let protoMessageName: String = _protobuf_package + ".ProxyVsockRequest" - public static let _protobuf_nameMap = SwiftProtobuf._NameMap(bytecode: "\0\u{1}id\0\u{3}vsock_port\0\u{1}guestPath\0\u{1}guestSocketPermissions\0\u{1}action\0") + public static let _protobuf_nameMap = SwiftProtobuf._NameMap(bytecode: "\0\u{1}id\0\u{3}vsock_port\0\u{1}guestPath\0\u{1}guestSocketPermissions\0\u{1}action\0\u{3}container_pid\0") public mutating func decodeMessage(decoder: inout D) throws { while let fieldNumber = try decoder.nextFieldNumber() { @@ -1998,6 +2008,7 @@ nonisolated extension Com_Apple_Containerization_Sandbox_V3_ProxyVsockRequest: S case 3: try { try decoder.decodeSingularStringField(value: &self.guestPath) }() case 4: try { try decoder.decodeSingularUInt32Field(value: &self._guestSocketPermissions) }() case 5: try { try decoder.decodeSingularEnumField(value: &self.action) }() + case 6: try { try decoder.decodeSingularInt32Field(value: &self._containerPid) }() default: break } } @@ -2023,6 +2034,9 @@ nonisolated extension Com_Apple_Containerization_Sandbox_V3_ProxyVsockRequest: S if self.action != .into { try visitor.visitSingularEnumField(value: self.action, fieldNumber: 5) } + try { if let v = self._containerPid { + try visitor.visitSingularInt32Field(value: v, fieldNumber: 6) + } }() try unknownFields.traverse(visitor: &visitor) } @@ -2032,6 +2046,7 @@ nonisolated extension Com_Apple_Containerization_Sandbox_V3_ProxyVsockRequest: S if lhs.guestPath != rhs.guestPath {return false} if lhs._guestSocketPermissions != rhs._guestSocketPermissions {return false} if lhs.action != rhs.action {return false} + if lhs._containerPid != rhs._containerPid {return false} if lhs.unknownFields != rhs.unknownFields {return false} return true } diff --git a/Sources/Containerization/SandboxContext/SandboxContext.proto b/Sources/Containerization/SandboxContext/SandboxContext.proto index 24fb24256..c10c27909 100644 --- a/Sources/Containerization/SandboxContext/SandboxContext.proto +++ b/Sources/Containerization/SandboxContext/SandboxContext.proto @@ -115,6 +115,7 @@ message ProxyVsockRequest { string guestPath = 3; optional uint32 guestSocketPermissions = 4; Action action = 5; + optional int32 container_pid = 6; } message ProxyVsockResponse {} diff --git a/Sources/Containerization/UnixSocketRelayManager.swift b/Sources/Containerization/UnixSocketRelayManager.swift index f1f9d3d7c..5d72ae9fa 100644 --- a/Sources/Containerization/UnixSocketRelayManager.swift +++ b/Sources/Containerization/UnixSocketRelayManager.swift @@ -19,8 +19,18 @@ import Foundation import Logging package actor UnixSocketRelayManager { + private struct RelayKey: Hashable { + let owner: String? + let socketID: String + } + + private struct ManagedRelay { + let socket: UnixSocketConfiguration + let relay: UnixSocketRelay + } + private let vm: any VirtualMachineInstance - private var relays: [String: UnixSocketRelay] + private var relays: [RelayKey: ManagedRelay] private let log: Logger? init(vm: any VirtualMachineInstance, log: Logger? = nil) { @@ -31,8 +41,9 @@ package actor UnixSocketRelayManager { } extension UnixSocketRelayManager { - func start(port: UInt32, socket: UnixSocketConfiguration) async throws { - guard relays[socket.id] == nil else { + func start(port: UInt32, socket: UnixSocketConfiguration, owner: String? = nil) async throws { + let key = RelayKey(owner: owner, socketID: socket.id) + guard relays[key] == nil else { throw ContainerizationError( .invalidState, message: "socket relay \(socket.id) already started" @@ -47,27 +58,54 @@ extension UnixSocketRelayManager { ) do { - relays[socket.id] = relay + relays[key] = ManagedRelay(socket: socket, relay: relay) try await relay.start() } catch { - relays.removeValue(forKey: socket.id) + relays.removeValue(forKey: key) throw error } } - func stop(socket: UnixSocketConfiguration) async throws { - guard let storedRelay = relays.removeValue(forKey: socket.id) else { + func stop(socket: UnixSocketConfiguration, owner: String? = nil) async throws { + let key = RelayKey(owner: owner, socketID: socket.id) + guard let storedRelay = relays.removeValue(forKey: key) else { throw ContainerizationError( .notFound, message: "failed to stop socket relay" ) } - try storedRelay.stop() + try storedRelay.relay.stop() + } + + func sockets(owner: String) -> [UnixSocketConfiguration] { + relays.compactMap { key, relay in + key.owner == owner ? relay.socket : nil + } + } + + func stopAll(owner: String) async throws { + let keys = relays.keys.filter { $0.owner == owner } + try stop(keys: keys) } func stopAll() async throws { - for (_, relay) in relays { - try relay.stop() + try stop(keys: Array(relays.keys)) + } + + private func stop(keys: [RelayKey]) throws { + let relays = keys.compactMap { self.relays.removeValue(forKey: $0) } + + var firstError: Error? + for relay in relays { + do { + try relay.relay.stop() + } catch { + firstError = firstError ?? error + } + } + + if let firstError { + throw firstError } } } diff --git a/Sources/Containerization/VirtualMachineAgent+Additions.swift b/Sources/Containerization/VirtualMachineAgent+Additions.swift index 2136a569b..08fb2d2ce 100644 --- a/Sources/Containerization/VirtualMachineAgent+Additions.swift +++ b/Sources/Containerization/VirtualMachineAgent+Additions.swift @@ -14,9 +14,28 @@ // limitations under the License. //===----------------------------------------------------------------------===// +import ContainerizationError + /// Protocol to conform to if your agent is capable of relaying unix domain socket /// connections. public protocol SocketRelayAgent { func relaySocket(port: UInt32, configuration: UnixSocketConfiguration) async throws + func relaySocket(port: UInt32, configuration: UnixSocketConfiguration, containerPID: Int32?) async throws func stopSocketRelay(configuration: UnixSocketConfiguration) async throws } + +extension SocketRelayAgent { + public func relaySocket( + port: UInt32, + configuration: UnixSocketConfiguration, + containerPID: Int32? + ) async throws { + guard containerPID == nil else { + throw ContainerizationError( + .unsupported, + message: "agent does not support relaying a socket from a container mount namespace" + ) + } + try await relaySocket(port: port, configuration: configuration) + } +} diff --git a/Sources/Containerization/Vminitd+SocketRelay.swift b/Sources/Containerization/Vminitd+SocketRelay.swift index cd258319c..528988172 100644 --- a/Sources/Containerization/Vminitd+SocketRelay.swift +++ b/Sources/Containerization/Vminitd+SocketRelay.swift @@ -17,6 +17,14 @@ extension Vminitd: SocketRelayAgent { /// Sets up a relay between a host socket to a newly created guest socket, or vice versa. public func relaySocket(port: UInt32, configuration: UnixSocketConfiguration) async throws { + try await relaySocket(port: port, configuration: configuration, containerPID: nil) + } + + public func relaySocket( + port: UInt32, + configuration: UnixSocketConfiguration, + containerPID: Int32? + ) async throws { let request = Com_Apple_Containerization_Sandbox_V3_ProxyVsockRequest.with { $0.id = configuration.id $0.vsockPort = port @@ -32,6 +40,9 @@ extension Vminitd: SocketRelayAgent { case .outOf: $0.guestPath = configuration.source.path $0.action = .outOf + if let containerPID { + $0.containerPid = containerPID + } } } _ = try await client.proxyVsock(request) diff --git a/Sources/Integration/ContainerTests.swift b/Sources/Integration/ContainerTests.swift index 9c8edd880..c6528963d 100644 --- a/Sources/Integration/ContainerTests.swift +++ b/Sources/Integration/ContainerTests.swift @@ -1124,6 +1124,118 @@ extension IntegrationSuite { } } + func testUnixSocketOutOfGuestTmpfs() async throws { + let id = "test-unixsocket-out-of-guest-tmpfs" + let bs = try await bootstrap( + id, + reference: "cgr.dev/chainguard/python:latest-dev@sha256:aa8fd2447b8b52922db57deb3894b622c3229387aaaec5934d64b85dbff6eb17" + ) + let hostSocket = FileManager.default.uniqueTemporaryDirectory(create: true) + .appendingPathComponent("relay.sock") + + let container = try LinuxContainer(id, rootfs: bs.rootfs, vmm: bs.vmm) { config in + config.process.arguments = [ + "/usr/bin/python", + "-c", + """ + import os, socket + path = "/dev/shm/relay.sock" + server = socket.socket(socket.AF_UNIX) + server.bind(path) + os.chmod(path, 0o777) + server.listen() + open("/tmp/relay-ready", "w").close() + connection, _ = server.accept() + connection.sendall(connection.recv(4)) + """, + ] + config.sockets = [ + UnixSocketConfiguration( + source: URL(filePath: "/dev/shm/relay.sock"), + destination: hostSocket, + direction: .outOf + ) + ] + config.bootLog = bs.bootLog + } + + do { + try await container.create() + try await container.start() + + let ready = try await container.exec("wait-for-relay") { config in + config.arguments = ["/bin/sh", "-c", "while [ ! -f /tmp/relay-ready ]; do sleep 0.1; done"] + } + try await ready.start() + let readyStatus = try await ready.wait() + try await ready.delete() + guard readyStatus.exitCode == 0 else { + throw IntegrationError.assert(msg: "socket server did not become ready") + } + + let client = try Socket(type: UnixType(path: hostSocket.path)) + try client.connect() + _ = try client.write(data: Data("PING".utf8)) + + var response = Data(count: 4) + let bytesRead = try client.read(buffer: &response) + guard bytesRead == 4, response == Data("PING".utf8) else { + throw IntegrationError.assert(msg: "unexpected relay response: \(response)") + } + + let status = try await container.wait() + guard status.exitCode == 0 else { + throw IntegrationError.assert(msg: "socket server exited with status \(status)") + } + try await container.stop() + } catch { + try? await container.stop() + throw error + } + } + + func testUnixSocketRelayRollback() async throws { + let id = "test-unixsocket-relay-rollback" + let bs = try await bootstrap(id) + let hostDirectory = FileManager.default.uniqueTemporaryDirectory(create: true) + guard FileManager.default.fileExists(atPath: hostDirectory.path) else { + throw IntegrationError.assert(msg: "failed to create host socket directory") + } + let firstHostSocket = hostDirectory.appendingPathComponent("first.sock") + let invalidHostSocket = hostDirectory.appendingPathComponent(String(repeating: "x", count: 200)) + + let container = try LinuxContainer(id, rootfs: bs.rootfs, vmm: bs.vmm) { config in + config.process.arguments = ["/bin/sleep", "100"] + config.sockets = [ + UnixSocketConfiguration( + source: URL(filePath: "/tmp/first.sock"), + destination: firstHostSocket, + direction: .outOf + ), + UnixSocketConfiguration( + source: URL(filePath: "/tmp/second.sock"), + destination: invalidHostSocket, + direction: .outOf + ), + ] + config.bootLog = bs.bootLog + } + + try await container.create() + do { + try await container.start() + throw IntegrationError.assert(msg: "expected the second relay to fail") + } catch let error as IntegrationError { + throw error + } catch UnixType.Error.nameTooLong { + guard !FileManager.default.fileExists(atPath: firstHostSocket.path) else { + throw IntegrationError.assert(msg: "first relay was not removed after setup failed") + } + } catch { + throw IntegrationError.assert(msg: "unexpected relay setup error: \(error)") + } + } + // NOTE: Once upon a time our guest agent created any proxied unix sockets at // a path that contained the container ID in it. The problem here is if the container // ID is comically long we exceed the max length of a unix domain socket path. diff --git a/Sources/Integration/PodTests.swift b/Sources/Integration/PodTests.swift index ae1caec86..ea09f40ce 100644 --- a/Sources/Integration/PodTests.swift +++ b/Sources/Integration/PodTests.swift @@ -1924,6 +1924,197 @@ extension IntegrationSuite { } } + func testPodUnixSocketOutOfGuestTmpfsStopsWithContainer() async throws { + let id = "test-pod-unixsocket-out-of-guest-tmpfs" + let bs = try await bootstrap( + id, + reference: "cgr.dev/chainguard/python:latest-dev@sha256:aa8fd2447b8b52922db57deb3894b622c3229387aaaec5934d64b85dbff6eb17" + ) + let hostSocket = FileManager.default.uniqueTemporaryDirectory(create: true) + .appendingPathComponent("relay.sock") + let pod = try LinuxPod(id, vmm: bs.vmm) { config in + config.bootLog = bs.bootLog + } + + try await pod.addContainer("server", rootfs: bs.rootfs) { config in + config.process.arguments = [ + "/usr/bin/python", + "-c", + """ + import os, socket + path = "/dev/shm/relay.sock" + server = socket.socket(socket.AF_UNIX) + server.bind(path) + os.chmod(path, 0o777) + server.listen() + open("/tmp/relay-ready", "w").close() + while True: + connection, _ = server.accept() + connection.sendall(connection.recv(4)) + connection.close() + """, + ] + config.sockets = [ + UnixSocketConfiguration( + source: URL(filePath: "/dev/shm/relay.sock"), + destination: hostSocket, + direction: .outOf + ) + ] + } + + do { + try await pod.create() + try await pod.startContainer("server") + + let ready = try await pod.execInContainer("server", processID: "wait-for-relay") { config in + config.arguments = ["/bin/sh", "-c", "while [ ! -f /tmp/relay-ready ]; do sleep 0.1; done"] + } + try await ready.start() + let readyStatus = try await ready.wait() + try await ready.delete() + guard readyStatus.exitCode == 0 else { + throw IntegrationError.assert(msg: "socket server did not become ready") + } + + let client = try Socket(type: UnixType(path: hostSocket.path)) + try client.connect() + _ = try client.write(data: Data("PING".utf8)) + + var response = Data(count: 4) + let bytesRead = try client.read(buffer: &response) + guard bytesRead == 4, response == Data("PING".utf8) else { + throw IntegrationError.assert(msg: "unexpected relay response: \(response)") + } + + try await pod.stopContainer("server") + try await pod.stopContainer("server") + guard !FileManager.default.fileExists(atPath: hostSocket.path) else { + throw IntegrationError.assert(msg: "host relay remained after container stop") + } + try await pod.stop() + } catch { + try? await pod.stop() + throw error + } + } + + func testPodUnixSocketFirstRelayFailureRollsBackProcess() async throws { + try await testPodUnixSocketRelayFailureRollsBackProcess(successfulRelayCount: 0) + } + + func testPodUnixSocketLaterRelayFailureRollsBackProcess() async throws { + try await testPodUnixSocketRelayFailureRollsBackProcess(successfulRelayCount: 1) + } + + func testPodUnixSocketGuestRelayFailureStopsHostRelay() async throws { + let id = "test-pod-unixsocket-guest-relay-failure" + let bs = try await bootstrap(id) + let hostSocket = FileManager.default.uniqueTemporaryDirectory(create: true) + .appendingPathComponent("relay.sock") + let pod = try LinuxPod(id, vmm: bs.vmm) { config in + config.bootLog = bs.bootLog + } + + try await pod.addContainer("container", rootfs: bs.rootfs) { config in + config.process.arguments = ["/bin/sleep", "100"] + config.sockets = [ + UnixSocketConfiguration( + source: URL(filePath: "/tmp/../tmp/relay.sock"), + destination: hostSocket, + direction: .outOf + ) + ] + } + + do { + try await pod.create() + do { + try await pod.startContainer("container") + throw IntegrationError.assert(msg: "expected guest relay setup to fail") + } catch let error as IntegrationError { + throw error + } catch { + } + + guard !FileManager.default.fileExists(atPath: hostSocket.path) else { + throw IntegrationError.assert(msg: "host relay remained after guest setup failed") + } + try await assertPodContainerCannotRestartAfterFailedStart(pod) + + try await pod.stop() + } catch { + try? await pod.stop() + throw error + } + } + + private func testPodUnixSocketRelayFailureRollsBackProcess(successfulRelayCount: Int) async throws { + let id = "test-pod-unixsocket-relay-rollback-\(successfulRelayCount)" + let bs = try await bootstrap(id) + let hostDirectory = FileManager.default.uniqueTemporaryDirectory(create: true) + let validHostSocket = hostDirectory.appendingPathComponent("relay.sock") + let invalidHostSocket = hostDirectory.appendingPathComponent(String(repeating: "x", count: 200)) + let pod = try LinuxPod(id, vmm: bs.vmm) { config in + config.bootLog = bs.bootLog + } + + try await pod.addContainer("container", rootfs: bs.rootfs) { config in + config.process.arguments = ["/bin/sleep", "100"] + var sockets: [UnixSocketConfiguration] = [] + if successfulRelayCount > 0 { + sockets.append( + UnixSocketConfiguration( + source: URL(filePath: "/tmp/relay.sock"), + destination: validHostSocket, + direction: .outOf + )) + } + sockets.append( + UnixSocketConfiguration( + source: URL(filePath: "/tmp/failing.sock"), + destination: invalidHostSocket, + direction: .outOf + )) + config.sockets = sockets + } + + do { + try await pod.create() + do { + try await pod.startContainer("container") + throw IntegrationError.assert(msg: "expected relay setup to fail") + } catch let error as IntegrationError { + throw error + } catch UnixType.Error.nameTooLong { + } catch { + throw IntegrationError.assert(msg: "unexpected relay setup error: \(error)") + } + + guard !FileManager.default.fileExists(atPath: validHostSocket.path) else { + throw IntegrationError.assert(msg: "successful relay was not rolled back") + } + try await assertPodContainerCannotRestartAfterFailedStart(pod) + + try await pod.stop() + } catch { + try? await pod.stop() + throw error + } + } + + private func assertPodContainerCannotRestartAfterFailedStart(_ pod: LinuxPod) async throws { + do { + try await pod.startContainer("container") + throw IntegrationError.assert(msg: "errored container unexpectedly restarted") + } catch let error as IntegrationError { + throw error + } catch let error as ContainerizationError where error.code == .invalidState { + } catch { + throw IntegrationError.assert(msg: "unexpected retry error: \(error)") + } + } + private func createPodHostUnixSocket() throws -> String { let dir = FileManager.default.uniqueTemporaryDirectory(create: true) let socketPath = dir.appendingPathComponent("test.sock").path diff --git a/Sources/Integration/Suite.swift b/Sources/Integration/Suite.swift index bf11e1cef..498136485 100644 --- a/Sources/Integration/Suite.swift +++ b/Sources/Integration/Suite.swift @@ -195,8 +195,10 @@ struct IntegrationSuite: AsyncParsableCommand { static let eventLoop = MultiThreadedEventLoopGroup(numberOfThreads: System.coreCount) - func bootstrap(_ testID: String) async throws -> (rootfs: Containerization.Mount, vmm: VirtualMachineManager, image: Containerization.Image, bootLog: BootLog) { - let reference = "ghcr.io/linuxcontainers/alpine:3.20" + func bootstrap( + _ testID: String, + reference: String = "ghcr.io/linuxcontainers/alpine:3.20" + ) async throws -> (rootfs: Containerization.Mount, vmm: VirtualMachineManager, image: Containerization.Image, bootLog: BootLog) { let store = Self.imageStore let initImage = try await store.getInitImage(reference: Self.initImage) @@ -598,7 +600,13 @@ struct IntegrationSuite: AsyncParsableCommand { Test("unix socket into guest", testUnixSocketIntoGuest), Test("unix socket into guest long container id", testUnixSocketIntoGuestLongContainerID), Test("unix socket into guest symlink", testUnixSocketIntoGuestSymlink), + Test("unix socket out of guest tmpfs", testUnixSocketOutOfGuestTmpfs), + Test("unix socket relay rollback", testUnixSocketRelayRollback), Test("pod unix socket into guest symlink", testPodUnixSocketIntoGuestSymlink), + Test("pod unix socket out of guest tmpfs", testPodUnixSocketOutOfGuestTmpfsStopsWithContainer), + Test("pod unix socket first relay rollback", testPodUnixSocketFirstRelayFailureRollsBackProcess), + Test("pod unix socket later relay rollback", testPodUnixSocketLaterRelayFailureRollsBackProcess), + Test("pod unix socket guest relay rollback", testPodUnixSocketGuestRelayFailureStopsHostRelay), // High-concurrency stdio (exceeds CH's prebound stdio pool size) Test("multiple concurrent processes", testMultipleConcurrentProcesses), diff --git a/vminitd/Sources/VminitdCore/Server+GRPC.swift b/vminitd/Sources/VminitdCore/Server+GRPC.swift index dd07ef54f..b998ad9a9 100644 --- a/vminitd/Sources/VminitdCore/Server+GRPC.swift +++ b/vminitd/Sources/VminitdCore/Server+GRPC.swift @@ -209,20 +209,24 @@ extension Initd: Com_Apple_Containerization_Sandbox_V3_SandboxContext.SimpleServ "action": "\(request.action)", ]) - let proxy = VsockProxy( - id: request.id, - action: request.action == .into ? .dial : .listen, - port: request.vsockPort, - path: URL(fileURLWithPath: request.guestPath), - udsPerms: request.guestSocketPermissions, - log: log - ) - do { - try await proxy.start() - try await state.add(proxy: proxy) + let proxy = try VsockProxy( + id: request.id, + action: request.action == .into ? .dial : .listen, + port: request.vsockPort, + path: URL(fileURLWithPath: request.guestPath), + udsPerms: request.guestSocketPermissions, + containerPID: request.hasContainerPid ? request.containerPid : nil, + log: log + ) + do { + try await proxy.start() + try await state.add(proxy: proxy) + } catch { + try? await proxy.close() + throw error + } } catch { - try? await proxy.close() log.error( "proxyVsock", metadata: [ diff --git a/vminitd/Sources/VminitdCore/VsockProxy.swift b/vminitd/Sources/VminitdCore/VsockProxy.swift index 393fad136..41158b998 100644 --- a/vminitd/Sources/VminitdCore/VsockProxy.swift +++ b/vminitd/Sources/VminitdCore/VsockProxy.swift @@ -22,6 +22,18 @@ import Foundation import LCShim import Logging +#if canImport(Musl) +import Musl +private let osClose = Musl.close +#else +import Glibc +private let osClose = Glibc.close +#endif + +private func osOpen(_ path: String, _ flags: Int32) -> Int32 { + path.withCString { open($0, flags) } +} + actor VsockProxy { enum Action { case listen @@ -40,6 +52,7 @@ actor VsockProxy { private let port: UInt32 private let udsPerms: UInt32? private let log: Logger? + private let containerRoot: Int32? private var listener: Socket? private var task: Task<(), Never>? @@ -51,15 +64,43 @@ actor VsockProxy { port: UInt32, path: URL, udsPerms: UInt32?, + containerPID: Int32? = nil, log: Logger? = nil - ) { + ) throws { self.id = id self.action = action self.port = port - self.path = path + if let containerPID { + guard + action == .listen, + containerPID > 0, + path.path.hasPrefix("/"), + path.standardizedFileURL.path == path.path + else { + throw POSIXError(.EINVAL) + } + + // Keep the container's root mounted and address it through this + // descriptor so a recycled PID cannot redirect later connections. + let root = osOpen("/proc/\(containerPID)/root", O_PATH | O_DIRECTORY | O_CLOEXEC) + guard root >= 0 else { + throw POSIXError.fromErrno() + } + self.containerRoot = root + self.path = URL(filePath: "/proc/self/fd/\(root)\(path.path)") + } else { + self.containerRoot = nil + self.path = path + } self.udsPerms = udsPerms self.log = log } + + deinit { + if let containerRoot { + _ = osClose(containerRoot) + } + } } extension VsockProxy {