LinuxContainer: Add ability to close stdin (#201)

This allows the user the ability to raise an EOF for the containers
stdin. Today there's no way to close stdin so something as simple as
"cat" and relying on EOF to move the process forward doesn't work

Sources/Containerization/Agent/Vminitd.swift

@@ -198,6 +198,16 @@ extension Vminitd: VirtualMachineAgent {
         _ = try await client.deleteProcess(request)
     }
 
+    public func closeProcessStdin(id: String, containerID: String?) async throws {
+        let request = Com_Apple_Containerization_Sandbox_V3_CloseProcessStdinRequest.with {
+            $0.id = id
+            if let containerID {
+                $0.containerID = containerID
+            }
+        }
+        _ = try await client.closeProcessStdin(request)
+    }
+
     public func up(name: String, mtu: UInt32? = nil) async throws {
         let request = Com_Apple_Containerization_Sandbox_V3_IpLinkSetRequest.with {
             $0.interface = name

Sources/Containerization/LinuxContainer.swift

@@ -715,6 +715,13 @@ extension LinuxContainer {
         return try await state.vm.dial(port)
     }
 
+    /// Close the containers standard input to signal no more input is
+    /// arriving.
+    public func closeStdin() async throws {
+        let state = try self.state.startedState("closeStdin")
+        return try await state.process.closeStdin()
+    }
+
     /// Relay a unix socket from in the container to the host, or from the host
     /// to inside the container.
     public func relayUnixSocket(socket: UnixSocketConfiguration) async throws {

Sources/Containerization/LinuxProcess.swift

@@ -211,8 +211,30 @@ extension LinuxProcess {
                                 try handle.write(contentsOf: data)
                             } catch {
                                 self.logger?.error("failed to write to stdin: \(error)")
+                                break
+                            }
+                        }
+
+                        do {
+                            self.logger?.debug("stdin relay finished, closing")
+
+                            // There's two ways we can wind up here:
+                            //
+                            // 1. The stream finished on its own (e.g. we wrote all the
+                            // data) and we will close the underlying stdin in the guest below.
+                            //
+                            // 2. The client explicitly called closeStdin() themselves
+                            // which will cancel this relay task AFTER actually closing
+                            // the fds. If the client did that, then this task will be
+                            // cancelled, and the fds are already gone so there's nothing
+                            // for us to do.
+                            if Task.isCancelled {
                                 return
                             }
+
+                            try await self._closeStdin()
+                        } catch {
+                            self.logger?.error("failed to close stdin: \(error)")
                         }
                     }
                 }
@@ -361,6 +383,28 @@ extension LinuxProcess {
         }
     }
 
+    public func closeStdin() async throws {
+        do {
+            try await self._closeStdin()
+            self.state.withLock {
+                $0.stdinRelay?.cancel()
+            }
+        } catch {
+            throw ContainerizationError(
+                .internalError,
+                message: "failed to close stdin",
+                cause: error,
+            )
+        }
+    }
+
+    func _closeStdin() async throws {
+        try await self.agent.closeProcessStdin(
+            id: self.id,
+            containerID: self.owningContainer
+        )
+    }
+
     /// Wait on the process to exit with an optional timeout. Returns the exit code of the process.
     @discardableResult
     public func wait(timeoutInSeconds: Int64? = nil) async throws -> Int32 {

Sources/Containerization/SandboxContext/SandboxContext.grpc.swift

@@ -104,6 +104,11 @@ public protocol Com_Apple_Containerization_Sandbox_V3_SandboxContextClientProtoc
     callOptions: CallOptions?
   ) -> UnaryCall<Com_Apple_Containerization_Sandbox_V3_ResizeProcessRequest, Com_Apple_Containerization_Sandbox_V3_ResizeProcessResponse>
 
+  func closeProcessStdin(
+    _ request: Com_Apple_Containerization_Sandbox_V3_CloseProcessStdinRequest,
+    callOptions: CallOptions?
+  ) -> UnaryCall<Com_Apple_Containerization_Sandbox_V3_CloseProcessStdinRequest, Com_Apple_Containerization_Sandbox_V3_CloseProcessStdinResponse>
+
   func proxyVsock(
     _ request: Com_Apple_Containerization_Sandbox_V3_ProxyVsockRequest,
     callOptions: CallOptions?
@@ -408,6 +413,24 @@ extension Com_Apple_Containerization_Sandbox_V3_SandboxContextClientProtocol {
     )
   }
 
+  /// Close IO for a given process.
+  ///
+  /// - Parameters:
+  ///   - request: Request to send to CloseProcessStdin.
+  ///   - callOptions: Call options.
+  /// - Returns: A `UnaryCall` with futures for the metadata, status and response.
+  public func closeProcessStdin(
+    _ request: Com_Apple_Containerization_Sandbox_V3_CloseProcessStdinRequest,
+    callOptions: CallOptions? = nil
+  ) -> UnaryCall<Com_Apple_Containerization_Sandbox_V3_CloseProcessStdinRequest, Com_Apple_Containerization_Sandbox_V3_CloseProcessStdinResponse> {
+    return self.makeUnaryCall(
+      path: Com_Apple_Containerization_Sandbox_V3_SandboxContextClientMetadata.Methods.closeProcessStdin.path,
+      request: request,
+      callOptions: callOptions ?? self.defaultCallOptions,
+      interceptors: self.interceptors?.makeCloseProcessStdinInterceptors() ?? []
+    )
+  }
+
   /// Proxy a vsock port to a unix domain socket in the guest, or vice versa.
   ///
   /// - Parameters:
@@ -704,6 +727,11 @@ public protocol Com_Apple_Containerization_Sandbox_V3_SandboxContextAsyncClientP
     callOptions: CallOptions?
   ) -> GRPCAsyncUnaryCall<Com_Apple_Containerization_Sandbox_V3_ResizeProcessRequest, Com_Apple_Containerization_Sandbox_V3_ResizeProcessResponse>
 
+  func makeCloseProcessStdinCall(
+    _ request: Com_Apple_Containerization_Sandbox_V3_CloseProcessStdinRequest,
+    callOptions: CallOptions?
+  ) -> GRPCAsyncUnaryCall<Com_Apple_Containerization_Sandbox_V3_CloseProcessStdinRequest, Com_Apple_Containerization_Sandbox_V3_CloseProcessStdinResponse>
+
   func makeProxyVsockCall(
     _ request: Com_Apple_Containerization_Sandbox_V3_ProxyVsockRequest,
     callOptions: CallOptions?
@@ -928,6 +956,18 @@ extension Com_Apple_Containerization_Sandbox_V3_SandboxContextAsyncClientProtoco
     )
   }
 
+  public func makeCloseProcessStdinCall(
+    _ request: Com_Apple_Containerization_Sandbox_V3_CloseProcessStdinRequest,
+    callOptions: CallOptions? = nil
+  ) -> GRPCAsyncUnaryCall<Com_Apple_Containerization_Sandbox_V3_CloseProcessStdinRequest, Com_Apple_Containerization_Sandbox_V3_CloseProcessStdinResponse> {
+    return self.makeAsyncUnaryCall(
+      path: Com_Apple_Containerization_Sandbox_V3_SandboxContextClientMetadata.Methods.closeProcessStdin.path,
+      request: request,
+      callOptions: callOptions ?? self.defaultCallOptions,
+      interceptors: self.interceptors?.makeCloseProcessStdinInterceptors() ?? []
+    )
+  }
+
   public func makeProxyVsockCall(
     _ request: Com_Apple_Containerization_Sandbox_V3_ProxyVsockRequest,
     callOptions: CallOptions? = nil
@@ -1207,6 +1247,18 @@ extension Com_Apple_Containerization_Sandbox_V3_SandboxContextAsyncClientProtoco
     )
   }
 
+  public func closeProcessStdin(
+    _ request: Com_Apple_Containerization_Sandbox_V3_CloseProcessStdinRequest,
+    callOptions: CallOptions? = nil
+  ) async throws -> Com_Apple_Containerization_Sandbox_V3_CloseProcessStdinResponse {
+    return try await self.performAsyncUnaryCall(
+      path: Com_Apple_Containerization_Sandbox_V3_SandboxContextClientMetadata.Methods.closeProcessStdin.path,
+      request: request,
+      callOptions: callOptions ?? self.defaultCallOptions,
+      interceptors: self.interceptors?.makeCloseProcessStdinInterceptors() ?? []
+    )
+  }
+
   public func proxyVsock(
     _ request: Com_Apple_Containerization_Sandbox_V3_ProxyVsockRequest,
     callOptions: CallOptions? = nil
@@ -1377,6 +1429,9 @@ public protocol Com_Apple_Containerization_Sandbox_V3_SandboxContextClientInterc
   /// - Returns: Interceptors to use when invoking 'resizeProcess'.
   func makeResizeProcessInterceptors() -> [ClientInterceptor<Com_Apple_Containerization_Sandbox_V3_ResizeProcessRequest, Com_Apple_Containerization_Sandbox_V3_ResizeProcessResponse>]
 
+  /// - Returns: Interceptors to use when invoking 'closeProcessStdin'.
+  func makeCloseProcessStdinInterceptors() -> [ClientInterceptor<Com_Apple_Containerization_Sandbox_V3_CloseProcessStdinRequest, Com_Apple_Containerization_Sandbox_V3_CloseProcessStdinResponse>]
+
   /// - Returns: Interceptors to use when invoking 'proxyVsock'.
   func makeProxyVsockInterceptors() -> [ClientInterceptor<Com_Apple_Containerization_Sandbox_V3_ProxyVsockRequest, Com_Apple_Containerization_Sandbox_V3_ProxyVsockResponse>]
 
@@ -1424,6 +1479,7 @@ public enum Com_Apple_Containerization_Sandbox_V3_SandboxContextClientMetadata {
       Com_Apple_Containerization_Sandbox_V3_SandboxContextClientMetadata.Methods.killProcess,
       Com_Apple_Containerization_Sandbox_V3_SandboxContextClientMetadata.Methods.waitProcess,
       Com_Apple_Containerization_Sandbox_V3_SandboxContextClientMetadata.Methods.resizeProcess,
+      Com_Apple_Containerization_Sandbox_V3_SandboxContextClientMetadata.Methods.closeProcessStdin,
       Com_Apple_Containerization_Sandbox_V3_SandboxContextClientMetadata.Methods.proxyVsock,
       Com_Apple_Containerization_Sandbox_V3_SandboxContextClientMetadata.Methods.stopVsockProxy,
       Com_Apple_Containerization_Sandbox_V3_SandboxContextClientMetadata.Methods.ipLinkSet,
@@ -1521,6 +1577,12 @@ public enum Com_Apple_Containerization_Sandbox_V3_SandboxContextClientMetadata {
       type: GRPCCallType.unary
     )
 
+    public static let closeProcessStdin = GRPCMethodDescriptor(
+      name: "CloseProcessStdin",
+      path: "/com.apple.containerization.sandbox.v3.SandboxContext/CloseProcessStdin",
+      type: GRPCCallType.unary
+    )
+
     public static let proxyVsock = GRPCMethodDescriptor(
       name: "ProxyVsock",
       path: "/com.apple.containerization.sandbox.v3.SandboxContext/ProxyVsock",
@@ -1626,6 +1688,9 @@ public protocol Com_Apple_Containerization_Sandbox_V3_SandboxContextProvider: Ca
   /// not have a pty allocated.
   func resizeProcess(request: Com_Apple_Containerization_Sandbox_V3_ResizeProcessRequest, context: StatusOnlyCallContext) -> EventLoopFuture<Com_Apple_Containerization_Sandbox_V3_ResizeProcessResponse>
 
+  /// Close IO for a given process.
+  func closeProcessStdin(request: Com_Apple_Containerization_Sandbox_V3_CloseProcessStdinRequest, context: StatusOnlyCallContext) -> EventLoopFuture<Com_Apple_Containerization_Sandbox_V3_CloseProcessStdinResponse>
+
   /// Proxy a vsock port to a unix domain socket in the guest, or vice versa.
   func proxyVsock(request: Com_Apple_Containerization_Sandbox_V3_ProxyVsockRequest, context: StatusOnlyCallContext) -> EventLoopFuture<Com_Apple_Containerization_Sandbox_V3_ProxyVsockResponse>
 
@@ -1792,6 +1857,15 @@ extension Com_Apple_Containerization_Sandbox_V3_SandboxContextProvider {
         userFunction: self.resizeProcess(request:context:)
       )
 
+    case "CloseProcessStdin":
+      return UnaryServerHandler(
+        context: context,
+        requestDeserializer: ProtobufDeserializer<Com_Apple_Containerization_Sandbox_V3_CloseProcessStdinRequest>(),
+        responseSerializer: ProtobufSerializer<Com_Apple_Containerization_Sandbox_V3_CloseProcessStdinResponse>(),
+        interceptors: self.interceptors?.makeCloseProcessStdinInterceptors() ?? [],
+        userFunction: self.closeProcessStdin(request:context:)
+      )
+
     case "ProxyVsock":
       return UnaryServerHandler(
         context: context,
@@ -1972,6 +2046,12 @@ public protocol Com_Apple_Containerization_Sandbox_V3_SandboxContextAsyncProvide
     context: GRPCAsyncServerCallContext
   ) async throws -> Com_Apple_Containerization_Sandbox_V3_ResizeProcessResponse
 
+  /// Close IO for a given process.
+  func closeProcessStdin(
+    request: Com_Apple_Containerization_Sandbox_V3_CloseProcessStdinRequest,
+    context: GRPCAsyncServerCallContext
+  ) async throws -> Com_Apple_Containerization_Sandbox_V3_CloseProcessStdinResponse
+
   /// Proxy a vsock port to a unix domain socket in the guest, or vice versa.
   func proxyVsock(
     request: Com_Apple_Containerization_Sandbox_V3_ProxyVsockRequest,
@@ -2172,6 +2252,15 @@ extension Com_Apple_Containerization_Sandbox_V3_SandboxContextAsyncProvider {
         wrapping: { try await self.resizeProcess(request: $0, context: $1) }
       )
 
+    case "CloseProcessStdin":
+      return GRPCAsyncServerHandler(
+        context: context,
+        requestDeserializer: ProtobufDeserializer<Com_Apple_Containerization_Sandbox_V3_CloseProcessStdinRequest>(),
+        responseSerializer: ProtobufSerializer<Com_Apple_Containerization_Sandbox_V3_CloseProcessStdinResponse>(),
+        interceptors: self.interceptors?.makeCloseProcessStdinInterceptors() ?? [],
+        wrapping: { try await self.closeProcessStdin(request: $0, context: $1) }
+      )
+
     case "ProxyVsock":
       return GRPCAsyncServerHandler(
         context: context,
@@ -2317,6 +2406,10 @@ public protocol Com_Apple_Containerization_Sandbox_V3_SandboxContextServerInterc
   ///   Defaults to calling `self.makeInterceptors()`.
   func makeResizeProcessInterceptors() -> [ServerInterceptor<Com_Apple_Containerization_Sandbox_V3_ResizeProcessRequest, Com_Apple_Containerization_Sandbox_V3_ResizeProcessResponse>]
 
+  /// - Returns: Interceptors to use when handling 'closeProcessStdin'.
+  ///   Defaults to calling `self.makeInterceptors()`.
+  func makeCloseProcessStdinInterceptors() -> [ServerInterceptor<Com_Apple_Containerization_Sandbox_V3_CloseProcessStdinRequest, Com_Apple_Containerization_Sandbox_V3_CloseProcessStdinResponse>]
+
   /// - Returns: Interceptors to use when handling 'proxyVsock'.
   ///   Defaults to calling `self.makeInterceptors()`.
   func makeProxyVsockInterceptors() -> [ServerInterceptor<Com_Apple_Containerization_Sandbox_V3_ProxyVsockRequest, Com_Apple_Containerization_Sandbox_V3_ProxyVsockResponse>]
@@ -2373,6 +2466,7 @@ public enum Com_Apple_Containerization_Sandbox_V3_SandboxContextServerMetadata {
       Com_Apple_Containerization_Sandbox_V3_SandboxContextServerMetadata.Methods.killProcess,
       Com_Apple_Containerization_Sandbox_V3_SandboxContextServerMetadata.Methods.waitProcess,
       Com_Apple_Containerization_Sandbox_V3_SandboxContextServerMetadata.Methods.resizeProcess,
+      Com_Apple_Containerization_Sandbox_V3_SandboxContextServerMetadata.Methods.closeProcessStdin,
       Com_Apple_Containerization_Sandbox_V3_SandboxContextServerMetadata.Methods.proxyVsock,
       Com_Apple_Containerization_Sandbox_V3_SandboxContextServerMetadata.Methods.stopVsockProxy,
       Com_Apple_Containerization_Sandbox_V3_SandboxContextServerMetadata.Methods.ipLinkSet,
@@ -2470,6 +2564,12 @@ public enum Com_Apple_Containerization_Sandbox_V3_SandboxContextServerMetadata {
       type: GRPCCallType.unary
     )
 
+    public static let closeProcessStdin = GRPCMethodDescriptor(
+      name: "CloseProcessStdin",
+      path: "/com.apple.containerization.sandbox.v3.SandboxContext/CloseProcessStdin",
+      type: GRPCCallType.unary
+    )
+
     public static let proxyVsock = GRPCMethodDescriptor(
       name: "ProxyVsock",
       path: "/com.apple.containerization.sandbox.v3.SandboxContext/ProxyVsock",

Sources/Containerization/SandboxContext/SandboxContext.pb.swift

@@ -623,6 +623,39 @@ public struct Com_Apple_Containerization_Sandbox_V3_KillProcessResponse: Sendabl
   public init() {}
 }
 
+public struct Com_Apple_Containerization_Sandbox_V3_CloseProcessStdinRequest: Sendable {
+  // SwiftProtobuf.Message conformance is added in an extension below. See the
+  // `Message` and `Message+*Additions` files in the SwiftProtobuf library for
+  // methods supported on all messages.
+
+  public var id: String = String()
+
+  public var containerID: String {
+    get {return _containerID ?? String()}
+    set {_containerID = newValue}
+  }
+  /// Returns true if `containerID` has been explicitly set.
+  public var hasContainerID: Bool {return self._containerID != nil}
+  /// Clears the value of `containerID`. Subsequent reads from it will return its default value.
+  public mutating func clearContainerID() {self._containerID = nil}
+
+  public var unknownFields = SwiftProtobuf.UnknownStorage()
+
+  public init() {}
+
+  fileprivate var _containerID: String? = nil
+}
+
+public struct Com_Apple_Containerization_Sandbox_V3_CloseProcessStdinResponse: Sendable {
+  // SwiftProtobuf.Message conformance is added in an extension below. See the
+  // `Message` and `Message+*Additions` files in the SwiftProtobuf library for
+  // methods supported on all messages.
+
+  public var unknownFields = SwiftProtobuf.UnknownStorage()
+
+  public init() {}
+}
+
 public struct Com_Apple_Containerization_Sandbox_V3_MkdirRequest: Sendable {
   // SwiftProtobuf.Message conformance is added in an extension below. See the
   // `Message` and `Message+*Additions` files in the SwiftProtobuf library for
@@ -1935,6 +1968,67 @@ extension Com_Apple_Containerization_Sandbox_V3_KillProcessResponse: SwiftProtob
   }
 }
 
+extension Com_Apple_Containerization_Sandbox_V3_CloseProcessStdinRequest: SwiftProtobuf.Message, SwiftProtobuf._MessageImplementationBase, SwiftProtobuf._ProtoNameProviding {
+  public static let protoMessageName: String = _protobuf_package + ".CloseProcessStdinRequest"
+  public static let _protobuf_nameMap: SwiftProtobuf._NameMap = [
+    1: .same(proto: "id"),
+    2: .same(proto: "containerID"),
+  ]
+
+  public mutating func decodeMessage<D: SwiftProtobuf.Decoder>(decoder: inout D) throws {
+    while let fieldNumber = try decoder.nextFieldNumber() {
+      // The use of inline closures is to circumvent an issue where the compiler
+      // allocates stack space for every case branch when no optimizations are
+      // enabled. https://github.com/apple/swift-protobuf/issues/1034
+      switch fieldNumber {
+      case 1: try { try decoder.decodeSingularStringField(value: &self.id) }()
+      case 2: try { try decoder.decodeSingularStringField(value: &self._containerID) }()
+      default: break
+      }
+    }
+  }
+
+  public func traverse<V: SwiftProtobuf.Visitor>(visitor: inout V) throws {
+    // The use of inline closures is to circumvent an issue where the compiler
+    // allocates stack space for every if/case branch local when no optimizations
+    // are enabled. https://github.com/apple/swift-protobuf/issues/1034 and
+    // https://github.com/apple/swift-protobuf/issues/1182
+    if !self.id.isEmpty {
+      try visitor.visitSingularStringField(value: self.id, fieldNumber: 1)
+    }
+    try { if let v = self._containerID {
+      try visitor.visitSingularStringField(value: v, fieldNumber: 2)
+    } }()
+    try unknownFields.traverse(visitor: &visitor)
+  }
+
+  public static func ==(lhs: Com_Apple_Containerization_Sandbox_V3_CloseProcessStdinRequest, rhs: Com_Apple_Containerization_Sandbox_V3_CloseProcessStdinRequest) -> Bool {
+    if lhs.id != rhs.id {return false}
+    if lhs._containerID != rhs._containerID {return false}
+    if lhs.unknownFields != rhs.unknownFields {return false}
+    return true
+  }
+}
+
+extension Com_Apple_Containerization_Sandbox_V3_CloseProcessStdinResponse: SwiftProtobuf.Message, SwiftProtobuf._MessageImplementationBase, SwiftProtobuf._ProtoNameProviding {
+  public static let protoMessageName: String = _protobuf_package + ".CloseProcessStdinResponse"
+  public static let _protobuf_nameMap = SwiftProtobuf._NameMap()
+
+  public mutating func decodeMessage<D: SwiftProtobuf.Decoder>(decoder: inout D) throws {
+    // Load everything into unknown fields
+    while try decoder.nextFieldNumber() != nil {}
+  }
+
+  public func traverse<V: SwiftProtobuf.Visitor>(visitor: inout V) throws {
+    try unknownFields.traverse(visitor: &visitor)
+  }
+
+  public static func ==(lhs: Com_Apple_Containerization_Sandbox_V3_CloseProcessStdinResponse, rhs: Com_Apple_Containerization_Sandbox_V3_CloseProcessStdinResponse) -> Bool {
+    if lhs.unknownFields != rhs.unknownFields {return false}
+    return true
+  }
+}
+
 extension Com_Apple_Containerization_Sandbox_V3_MkdirRequest: SwiftProtobuf.Message, SwiftProtobuf._MessageImplementationBase, SwiftProtobuf._ProtoNameProviding {
   public static let protoMessageName: String = _protobuf_package + ".MkdirRequest"
   public static let _protobuf_nameMap: SwiftProtobuf._NameMap = [

Sources/Containerization/SandboxContext/SandboxContext.proto

@@ -34,6 +34,8 @@ service SandboxContext {
   // Resize the tty of a given process. This will error if the process does
   // not have a pty allocated.
   rpc ResizeProcess(ResizeProcessRequest) returns (ResizeProcessResponse);
+  // Close IO for a given process.
+  rpc CloseProcessStdin(CloseProcessStdinRequest) returns (CloseProcessStdinResponse);
 
   // Proxy a vsock port to a unix domain socket in the guest, or vice versa.
   rpc ProxyVsock(ProxyVsockRequest) returns (ProxyVsockResponse);
@@ -182,6 +184,13 @@ message KillProcessRequest {
 
 message KillProcessResponse { int32 result = 1; }
 
+message CloseProcessStdinRequest {
+  string id = 1;
+  optional string containerID = 2;
+}
+
+message CloseProcessStdinResponse {}
+
 message MkdirRequest {
   string path = 1;
   bool all = 2;

Sources/Containerization/VirtualMachineAgent.swift

@@ -14,6 +14,7 @@
 // limitations under the License.
 //===----------------------------------------------------------------------===//
 
+import ContainerizationError
 import ContainerizationOCI
 import Foundation
 
@@ -51,6 +52,7 @@ public protocol VirtualMachineAgent: Sendable {
     func resizeProcess(id: String, containerID: String?, columns: UInt32, rows: UInt32) async throws
     func waitProcess(id: String, containerID: String?, timeoutInSeconds: Int64?) async throws -> Int32
     func deleteProcess(id: String, containerID: String?) async throws
+    func closeProcessStdin(id: String, containerID: String?) async throws
 
     // Networking
     func up(name: String, mtu: UInt32?) async throws
@@ -59,3 +61,9 @@ public protocol VirtualMachineAgent: Sendable {
     func routeAddDefault(name: String, gateway: String) async throws
     func configureDNS(config: DNS, location: String) async throws
 }
+
+extension VirtualMachineAgent {
+    public func closeProcessStdin(id: String, containerID: String?) async throws {
+        throw ContainerizationError(.unsupported, message: "closeProcessStdin")
+    }
+}

Sources/Integration/ProcessTests.swift

@@ -74,6 +74,21 @@ extension IntegrationSuite {
         }
     }
 
+    final class StdinBuffer: ReaderStream {
+        let data: Data
+
+        init(data: Data) {
+            self.data = data
+        }
+
+        func stream() -> AsyncStream<Data> {
+            let (stream, cont) = AsyncStream<Data>.makeStream()
+            cont.yield(self.data)
+            cont.finish()
+            return stream
+        }
+    }
+
     func testProcessEchoHi() async throws {
         let id = "test-process-echo-hi"
         let bs = try await bootstrap()
@@ -402,4 +417,36 @@ extension IntegrationSuite {
                 msg: "process should have returned on stdout '\(expected)' != '\(String(data: buffer.data, encoding: .utf8)!)'")
         }
     }
+
+    func testProcessStdin() async throws {
+        let id = "test-container-stdin"
+
+        let bs = try await bootstrap()
+        let container = LinuxContainer(
+            id,
+            rootfs: bs.rootfs,
+            vmm: bs.vmm
+        )
+        container.arguments = ["cat"]
+        container.stdin = StdinBuffer(data: "Hello from test".data(using: .utf8)!)
+
+        let buffer = BufferWriter()
+        container.stdout = buffer
+
+        try await container.create()
+        try await container.start()
+
+        let status = try await container.wait()
+        try await container.stop()
+
+        guard status == 0 else {
+            throw IntegrationError.assert(msg: "process status \(status) != 0")
+        }
+        let expected = "Hello from test"
+
+        guard String(data: buffer.data, encoding: .utf8) == "\(expected)" else {
+            throw IntegrationError.assert(
+                msg: "process should have returned on stdout '\(expected)' != '\(String(data: buffer.data, encoding: .utf8)!)'")
+        }
+    }
 }

Sources/Integration/Suite.swift

@@ -199,6 +199,7 @@ struct IntegrationSuite: AsyncParsableCommand {
             "process false": testProcessFalse,
             "process echo hi": testProcessEchoHi,
             "process user": testProcessUser,
+            "process stdin": testProcessStdin,
             "process home envvar": testProcessHomeEnvvar,
             "process custom home envvar": testProcessCustomHomeEnvvar,
             "process tty ensure TERM": testProcessTtyEnvvar,

vminitd/Sources/vminitd/ManagedContainer.swift

@@ -111,6 +111,11 @@ extension ManagedContainer {
         try proc.resize(size: size)
     }
 
+    func closeStdin(execID: String) throws {
+        let proc = try self.getExecOrInit(execID: execID)
+        try proc.closeStdin()
+    }
+
     func deleteExec(id: String) throws {
         try ensureExecExists(id)
         do {

vminitd/Sources/vminitd/ManagedProcess.swift

@@ -216,4 +216,10 @@ extension ManagedProcess {
             try $0.io.resize(size: size)
         }
     }
+
+    func closeStdin() throws {
+        try self.lock.withLock {
+            try $0.io.closeStdin()
+        }
+    }
 }

vminitd/Sources/vminitd/Server+GRPC.swift

@@ -609,6 +609,41 @@ extension Initd: Com_Apple_Containerization_Sandbox_V3_SandboxContextAsyncProvid
         }
     }
 
+    func closeProcessStdin(
+        request: Com_Apple_Containerization_Sandbox_V3_CloseProcessStdinRequest, context: GRPCAsyncServerCallContext
+    ) async throws -> Com_Apple_Containerization_Sandbox_V3_CloseProcessStdinResponse {
+        log.debug(
+            "closeProcessStdin",
+            metadata: [
+                "id": "\(request.id)",
+                "containerID": "\(request.containerID)",
+            ])
+
+        if !request.hasContainerID {
+            fatalError("processes in the root of the vm not implemented")
+        }
+
+        do {
+            let ctr = try await self.state.get(container: request.containerID)
+
+            try await ctr.closeStdin(execID: request.id)
+
+            return .init()
+        } catch {
+            log.error(
+                "closeProcessStdin",
+                metadata: [
+                    "id": "\(request.id)",
+                    "containerID": "\(request.containerID)",
+                    "error": "\(error)",
+                ])
+            throw GRPCStatus(
+                code: .internalError,
+                message: "closeProcessStdin: failed to close process stdin: \(error)"
+            )
+        }
+    }
+
     func ipLinkSet(
         request: Com_Apple_Containerization_Sandbox_V3_IpLinkSetRequest, context: GRPC.GRPCAsyncServerCallContext
     ) async throws -> Com_Apple_Containerization_Sandbox_V3_IpLinkSetResponse {