diff --git a/README.md b/README.md index e58ec0e..035c8b4 100644 --- a/README.md +++ b/README.md @@ -11,6 +11,9 @@ Currently implemented features: * `tart exec` support (`--run-rpc`) * it's recommended to invoke it as a launchd [global agent](https://launchd.info/) because fewer privileges will be available to commands started via `tart exec` * however, you can also invoke it as a launchd [global daemon](https://launchd.info/) if running commands started via `tart exec` as `root` is desired + * attached `/Agent/Exec` commands report `Started` with the guest process ID before standard output, standard error, or exit + * clients can send a `Signal` with a unique, nonzero `request_id` and a guest-native signal number; `all = false` signals only that command, while `all = true` signals only its dedicated process group + * a matching `SignalAck` is sent only after successful delivery; existing clients can ignore the additive events, and detached commands retain their original exit-only behavior * `tart ip --resolver=agent` support (`--run-rpc`) * allows resolving VM's IP address without relying on DHCP leases and/or an ARP table diff --git a/internal/rpc/agent.pb.go b/internal/rpc/agent.pb.go index c1b27fa..8c5ed1c 100644 --- a/internal/rpc/agent.pb.go +++ b/internal/rpc/agent.pb.go @@ -29,6 +29,7 @@ type ExecRequest struct { // *ExecRequest_Command_ // *ExecRequest_StandardInput // *ExecRequest_TerminalResize + // *ExecRequest_Signal_ Type isExecRequest_Type `protobuf_oneof:"type"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache @@ -98,6 +99,15 @@ func (x *ExecRequest) GetTerminalResize() *TerminalSize { return nil } +func (x *ExecRequest) GetSignal() *ExecRequest_Signal { + if x != nil { + if x, ok := x.Type.(*ExecRequest_Signal_); ok { + return x.Signal + } + } + return nil +} + type isExecRequest_Type interface { isExecRequest_Type() } @@ -114,12 +124,18 @@ type ExecRequest_TerminalResize struct { TerminalResize *TerminalSize `protobuf:"bytes,3,opt,name=terminal_resize,json=terminalResize,proto3,oneof"` } +type ExecRequest_Signal_ struct { + Signal *ExecRequest_Signal `protobuf:"bytes,4,opt,name=signal,proto3,oneof"` +} + func (*ExecRequest_Command_) isExecRequest_Type() {} func (*ExecRequest_StandardInput) isExecRequest_Type() {} func (*ExecRequest_TerminalResize) isExecRequest_Type() {} +func (*ExecRequest_Signal_) isExecRequest_Type() {} + type ExecResponse struct { state protoimpl.MessageState `protogen:"open.v1"` // Types that are valid to be assigned to Type: @@ -127,6 +143,8 @@ type ExecResponse struct { // *ExecResponse_Exit_ // *ExecResponse_StandardOutput // *ExecResponse_StandardError + // *ExecResponse_Started_ + // *ExecResponse_SignalAck_ Type isExecResponse_Type `protobuf_oneof:"type"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache @@ -196,6 +214,24 @@ func (x *ExecResponse) GetStandardError() *IOChunk { return nil } +func (x *ExecResponse) GetStarted() *ExecResponse_Started { + if x != nil { + if x, ok := x.Type.(*ExecResponse_Started_); ok { + return x.Started + } + } + return nil +} + +func (x *ExecResponse) GetSignalAck() *ExecResponse_SignalAck { + if x != nil { + if x, ok := x.Type.(*ExecResponse_SignalAck_); ok { + return x.SignalAck + } + } + return nil +} + type isExecResponse_Type interface { isExecResponse_Type() } @@ -212,12 +248,24 @@ type ExecResponse_StandardError struct { StandardError *IOChunk `protobuf:"bytes,3,opt,name=standard_error,json=standardError,proto3,oneof"` } +type ExecResponse_Started_ struct { + Started *ExecResponse_Started `protobuf:"bytes,4,opt,name=started,proto3,oneof"` +} + +type ExecResponse_SignalAck_ struct { + SignalAck *ExecResponse_SignalAck `protobuf:"bytes,5,opt,name=signal_ack,json=signalAck,proto3,oneof"` +} + func (*ExecResponse_Exit_) isExecResponse_Type() {} func (*ExecResponse_StandardOutput) isExecResponse_Type() {} func (*ExecResponse_StandardError) isExecResponse_Type() {} +func (*ExecResponse_Started_) isExecResponse_Type() {} + +func (*ExecResponse_SignalAck_) isExecResponse_Type() {} + type TerminalSize struct { state protoimpl.MessageState `protogen:"open.v1"` Rows uint32 `protobuf:"varint,1,opt,name=rows,proto3" json:"rows,omitempty"` @@ -494,6 +542,66 @@ func (x *ExecRequest_Command) GetWorkdir() string { return "" } +type ExecRequest_Signal struct { + state protoimpl.MessageState `protogen:"open.v1"` + RequestId uint64 `protobuf:"varint,1,opt,name=request_id,json=requestId,proto3" json:"request_id,omitempty"` + Signal uint32 `protobuf:"varint,2,opt,name=signal,proto3" json:"signal,omitempty"` + All bool `protobuf:"varint,3,opt,name=all,proto3" json:"all,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *ExecRequest_Signal) Reset() { + *x = ExecRequest_Signal{} + mi := &file_rpc_agent_proto_msgTypes[7] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *ExecRequest_Signal) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*ExecRequest_Signal) ProtoMessage() {} + +func (x *ExecRequest_Signal) ProtoReflect() protoreflect.Message { + mi := &file_rpc_agent_proto_msgTypes[7] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use ExecRequest_Signal.ProtoReflect.Descriptor instead. +func (*ExecRequest_Signal) Descriptor() ([]byte, []int) { + return file_rpc_agent_proto_rawDescGZIP(), []int{0, 1} +} + +func (x *ExecRequest_Signal) GetRequestId() uint64 { + if x != nil { + return x.RequestId + } + return 0 +} + +func (x *ExecRequest_Signal) GetSignal() uint32 { + if x != nil { + return x.Signal + } + return 0 +} + +func (x *ExecRequest_Signal) GetAll() bool { + if x != nil { + return x.All + } + return false +} + type ExecResponse_Exit struct { state protoimpl.MessageState `protogen:"open.v1"` Code int32 `protobuf:"varint,1,opt,name=code,proto3" json:"code,omitempty"` @@ -503,7 +611,7 @@ type ExecResponse_Exit struct { func (x *ExecResponse_Exit) Reset() { *x = ExecResponse_Exit{} - mi := &file_rpc_agent_proto_msgTypes[8] + mi := &file_rpc_agent_proto_msgTypes[9] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -515,7 +623,7 @@ func (x *ExecResponse_Exit) String() string { func (*ExecResponse_Exit) ProtoMessage() {} func (x *ExecResponse_Exit) ProtoReflect() protoreflect.Message { - mi := &file_rpc_agent_proto_msgTypes[8] + mi := &file_rpc_agent_proto_msgTypes[9] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -538,15 +646,104 @@ func (x *ExecResponse_Exit) GetCode() int32 { return 0 } +type ExecResponse_Started struct { + state protoimpl.MessageState `protogen:"open.v1"` + Pid uint32 `protobuf:"varint,1,opt,name=pid,proto3" json:"pid,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *ExecResponse_Started) Reset() { + *x = ExecResponse_Started{} + mi := &file_rpc_agent_proto_msgTypes[10] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *ExecResponse_Started) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*ExecResponse_Started) ProtoMessage() {} + +func (x *ExecResponse_Started) ProtoReflect() protoreflect.Message { + mi := &file_rpc_agent_proto_msgTypes[10] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use ExecResponse_Started.ProtoReflect.Descriptor instead. +func (*ExecResponse_Started) Descriptor() ([]byte, []int) { + return file_rpc_agent_proto_rawDescGZIP(), []int{1, 1} +} + +func (x *ExecResponse_Started) GetPid() uint32 { + if x != nil { + return x.Pid + } + return 0 +} + +type ExecResponse_SignalAck struct { + state protoimpl.MessageState `protogen:"open.v1"` + RequestId uint64 `protobuf:"varint,1,opt,name=request_id,json=requestId,proto3" json:"request_id,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *ExecResponse_SignalAck) Reset() { + *x = ExecResponse_SignalAck{} + mi := &file_rpc_agent_proto_msgTypes[11] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *ExecResponse_SignalAck) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*ExecResponse_SignalAck) ProtoMessage() {} + +func (x *ExecResponse_SignalAck) ProtoReflect() protoreflect.Message { + mi := &file_rpc_agent_proto_msgTypes[11] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use ExecResponse_SignalAck.ProtoReflect.Descriptor instead. +func (*ExecResponse_SignalAck) Descriptor() ([]byte, []int) { + return file_rpc_agent_proto_rawDescGZIP(), []int{1, 2} +} + +func (x *ExecResponse_SignalAck) GetRequestId() uint64 { + if x != nil { + return x.RequestId + } + return 0 +} + var File_rpc_agent_proto protoreflect.FileDescriptor const file_rpc_agent_proto_rawDesc = "" + "\n" + - "\x0frpc/agent.proto\x1a\x1bgoogle/protobuf/empty.proto\"\xeb\x03\n" + + "\x0frpc/agent.proto\x1a\x1bgoogle/protobuf/empty.proto\"\xed\x04\n" + "\vExecRequest\x120\n" + "\acommand\x18\x01 \x01(\v2\x14.ExecRequest.CommandH\x00R\acommand\x121\n" + "\x0estandard_input\x18\x02 \x01(\v2\b.IOChunkH\x00R\rstandardInput\x128\n" + - "\x0fterminal_resize\x18\x03 \x01(\v2\r.TerminalSizeH\x00R\x0eterminalResize\x1a\xb4\x02\n" + + "\x0fterminal_resize\x18\x03 \x01(\v2\r.TerminalSizeH\x00R\x0eterminalResize\x12-\n" + + "\x06signal\x18\x04 \x01(\v2\x13.ExecRequest.SignalH\x00R\x06signal\x1a\xb4\x02\n" + "\aCommand\x12\x12\n" + "\x04name\x18\x01 \x01(\tR\x04name\x12\x12\n" + "\x04args\x18\x02 \x03(\tR\x04args\x12 \n" + @@ -558,14 +755,27 @@ const file_rpc_agent_proto_rawDesc = "" + "\aworkdir\x18\b \x01(\tR\aworkdir\x1a6\n" + "\bEnvEntry\x12\x10\n" + "\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" + - "\x05value\x18\x02 \x01(\tR\x05value:\x028\x01B\x06\n" + - "\x04type\"\xc4\x01\n" + + "\x05value\x18\x02 \x01(\tR\x05value:\x028\x01\x1aQ\n" + + "\x06Signal\x12\x1d\n" + + "\n" + + "request_id\x18\x01 \x01(\x04R\trequestId\x12\x16\n" + + "\x06signal\x18\x02 \x01(\rR\x06signal\x12\x10\n" + + "\x03all\x18\x03 \x01(\bR\x03allB\x06\n" + + "\x04type\"\xfa\x02\n" + "\fExecResponse\x12(\n" + "\x04exit\x18\x01 \x01(\v2\x12.ExecResponse.ExitH\x00R\x04exit\x123\n" + "\x0fstandard_output\x18\x02 \x01(\v2\b.IOChunkH\x00R\x0estandardOutput\x121\n" + - "\x0estandard_error\x18\x03 \x01(\v2\b.IOChunkH\x00R\rstandardError\x1a\x1a\n" + + "\x0estandard_error\x18\x03 \x01(\v2\b.IOChunkH\x00R\rstandardError\x121\n" + + "\astarted\x18\x04 \x01(\v2\x15.ExecResponse.StartedH\x00R\astarted\x128\n" + + "\n" + + "signal_ack\x18\x05 \x01(\v2\x17.ExecResponse.SignalAckH\x00R\tsignalAck\x1a\x1a\n" + "\x04Exit\x12\x12\n" + - "\x04code\x18\x01 \x01(\x05R\x04codeB\x06\n" + + "\x04code\x18\x01 \x01(\x05R\x04code\x1a\x1b\n" + + "\aStarted\x12\x10\n" + + "\x03pid\x18\x01 \x01(\rR\x03pid\x1a*\n" + + "\tSignalAck\x12\x1d\n" + + "\n" + + "request_id\x18\x01 \x01(\x04R\trequestIdB\x06\n" + "\x04type\"6\n" + "\fTerminalSize\x12\x12\n" + "\x04rows\x18\x01 \x01(\rR\x04rows\x12\x12\n" + @@ -591,36 +801,42 @@ func file_rpc_agent_proto_rawDescGZIP() []byte { return file_rpc_agent_proto_rawDescData } -var file_rpc_agent_proto_msgTypes = make([]protoimpl.MessageInfo, 9) +var file_rpc_agent_proto_msgTypes = make([]protoimpl.MessageInfo, 12) var file_rpc_agent_proto_goTypes = []any{ - (*ExecRequest)(nil), // 0: ExecRequest - (*ExecResponse)(nil), // 1: ExecResponse - (*TerminalSize)(nil), // 2: TerminalSize - (*IOChunk)(nil), // 3: IOChunk - (*ResolveIPRequest)(nil), // 4: ResolveIPRequest - (*ResolveIPResponse)(nil), // 5: ResolveIPResponse - (*ExecRequest_Command)(nil), // 6: ExecRequest.Command - nil, // 7: ExecRequest.Command.EnvEntry - (*ExecResponse_Exit)(nil), // 8: ExecResponse.Exit + (*ExecRequest)(nil), // 0: ExecRequest + (*ExecResponse)(nil), // 1: ExecResponse + (*TerminalSize)(nil), // 2: TerminalSize + (*IOChunk)(nil), // 3: IOChunk + (*ResolveIPRequest)(nil), // 4: ResolveIPRequest + (*ResolveIPResponse)(nil), // 5: ResolveIPResponse + (*ExecRequest_Command)(nil), // 6: ExecRequest.Command + (*ExecRequest_Signal)(nil), // 7: ExecRequest.Signal + nil, // 8: ExecRequest.Command.EnvEntry + (*ExecResponse_Exit)(nil), // 9: ExecResponse.Exit + (*ExecResponse_Started)(nil), // 10: ExecResponse.Started + (*ExecResponse_SignalAck)(nil), // 11: ExecResponse.SignalAck } var file_rpc_agent_proto_depIdxs = []int32{ 6, // 0: ExecRequest.command:type_name -> ExecRequest.Command 3, // 1: ExecRequest.standard_input:type_name -> IOChunk 2, // 2: ExecRequest.terminal_resize:type_name -> TerminalSize - 8, // 3: ExecResponse.exit:type_name -> ExecResponse.Exit - 3, // 4: ExecResponse.standard_output:type_name -> IOChunk - 3, // 5: ExecResponse.standard_error:type_name -> IOChunk - 2, // 6: ExecRequest.Command.terminal_size:type_name -> TerminalSize - 7, // 7: ExecRequest.Command.env:type_name -> ExecRequest.Command.EnvEntry - 0, // 8: Agent.Exec:input_type -> ExecRequest - 4, // 9: Agent.ResolveIP:input_type -> ResolveIPRequest - 1, // 10: Agent.Exec:output_type -> ExecResponse - 5, // 11: Agent.ResolveIP:output_type -> ResolveIPResponse - 10, // [10:12] is the sub-list for method output_type - 8, // [8:10] is the sub-list for method input_type - 8, // [8:8] is the sub-list for extension type_name - 8, // [8:8] is the sub-list for extension extendee - 0, // [0:8] is the sub-list for field type_name + 7, // 3: ExecRequest.signal:type_name -> ExecRequest.Signal + 9, // 4: ExecResponse.exit:type_name -> ExecResponse.Exit + 3, // 5: ExecResponse.standard_output:type_name -> IOChunk + 3, // 6: ExecResponse.standard_error:type_name -> IOChunk + 10, // 7: ExecResponse.started:type_name -> ExecResponse.Started + 11, // 8: ExecResponse.signal_ack:type_name -> ExecResponse.SignalAck + 2, // 9: ExecRequest.Command.terminal_size:type_name -> TerminalSize + 8, // 10: ExecRequest.Command.env:type_name -> ExecRequest.Command.EnvEntry + 0, // 11: Agent.Exec:input_type -> ExecRequest + 4, // 12: Agent.ResolveIP:input_type -> ResolveIPRequest + 1, // 13: Agent.Exec:output_type -> ExecResponse + 5, // 14: Agent.ResolveIP:output_type -> ResolveIPResponse + 13, // [13:15] is the sub-list for method output_type + 11, // [11:13] is the sub-list for method input_type + 11, // [11:11] is the sub-list for extension type_name + 11, // [11:11] is the sub-list for extension extendee + 0, // [0:11] is the sub-list for field type_name } func init() { file_rpc_agent_proto_init() } @@ -632,11 +848,14 @@ func file_rpc_agent_proto_init() { (*ExecRequest_Command_)(nil), (*ExecRequest_StandardInput)(nil), (*ExecRequest_TerminalResize)(nil), + (*ExecRequest_Signal_)(nil), } file_rpc_agent_proto_msgTypes[1].OneofWrappers = []any{ (*ExecResponse_Exit_)(nil), (*ExecResponse_StandardOutput)(nil), (*ExecResponse_StandardError)(nil), + (*ExecResponse_Started_)(nil), + (*ExecResponse_SignalAck_)(nil), } type x struct{} out := protoimpl.TypeBuilder{ @@ -644,7 +863,7 @@ func file_rpc_agent_proto_init() { GoPackagePath: reflect.TypeOf(x{}).PkgPath(), RawDescriptor: unsafe.Slice(unsafe.StringData(file_rpc_agent_proto_rawDesc), len(file_rpc_agent_proto_rawDesc)), NumEnums: 0, - NumMessages: 9, + NumMessages: 12, NumExtensions: 0, NumServices: 1, }, diff --git a/internal/rpc/exec.go b/internal/rpc/exec.go index 0e934ab..5418656 100644 --- a/internal/rpc/exec.go +++ b/internal/rpc/exec.go @@ -4,17 +4,21 @@ import ( "context" "errors" "fmt" - "github.com/creack/pty" - "github.com/samber/lo" - "go.uber.org/zap" - "golang.org/x/sync/errgroup" - "google.golang.org/grpc" "io" "os" "os/exec" "slices" "strings" + "sync" "syscall" + + "github.com/creack/pty" + "github.com/samber/lo" + "go.uber.org/zap" + "golang.org/x/sync/errgroup" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" ) const ( @@ -23,6 +27,18 @@ const ( eofChar = 0x04 ) +type execResponseSender struct { + stream grpc.BidiStreamingServer[ExecRequest, ExecResponse] + mu sync.Mutex +} + +func (sender *execResponseSender) send(response *ExecResponse) error { + sender.mu.Lock() + defer sender.mu.Unlock() + + return sender.stream.Send(response) +} + func (rpc *RPC) Exec(stream grpc.BidiStreamingServer[ExecRequest, ExecResponse]) error { // Read the first exec request, it should describe a command to execute firstExecRequest, err := stream.Recv() @@ -43,7 +59,9 @@ func (rpc *RPC) Exec(stream grpc.BidiStreamingServer[ExecRequest, ExecResponse]) } // Execute the command - execCtx := stream.Context() + execCtx, cancelExec := context.WithCancel(stream.Context()) + defer cancelExec() + if firstExecRequestCommand.Command.Detach { execCtx = context.Background() } @@ -51,6 +69,7 @@ func (rpc *RPC) Exec(stream grpc.BidiStreamingServer[ExecRequest, ExecResponse]) cmd := exec.CommandContext(execCtx, firstExecRequestCommand.Command.Name, firstExecRequestCommand.Command.Args...) applyExecOverrides(cmd, firstExecRequestCommand.Command) + responseSender := &execResponseSender{stream: stream} if firstExecRequestCommand.Command.Detach { cmd.Stdout = io.Discard @@ -66,7 +85,7 @@ func (rpc *RPC) Exec(stream grpc.BidiStreamingServer[ExecRequest, ExecResponse]) } } - if err := stream.Send(&ExecResponse{ + if err := responseSender.send(&ExecResponse{ Type: &ExecResponse_Exit_{ Exit: &ExecResponse_Exit{ Code: 0, @@ -112,6 +131,10 @@ func (rpc *RPC) Exec(stream grpc.BidiStreamingServer[ExecRequest, ExecResponse]) return err } + // Give each attached command its own process group. PTY commands already + // receive a dedicated session and process group from pty.StartWithSize. + cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true} + err = cmd.Start() } if err != nil { @@ -121,15 +144,39 @@ func (rpc *RPC) Exec(stream grpc.BidiStreamingServer[ExecRequest, ExecResponse]) defer ptmx.Close() } - // Handle standard input and terminal resize events from the client + // Send the managed guest PID before starting any output readers, so even + // commands that finish immediately cannot produce output before Started. + if err := responseSender.send(&ExecResponse{ + Type: &ExecResponse_Started_{ + Started: &ExecResponse_Started{Pid: uint32(cmd.Process.Pid)}, + }, + }); err != nil { + cancelExec() + _ = cmd.Wait() + return err + } + + // Handle standard input, terminal resize, and signals from this stream only. fromClientErrCh := make(chan error, 1) + reportClientError := func(err error) { + select { + case fromClientErrCh <- err: + default: + } + cancelExec() + } + + var signalMu sync.Mutex + processExited := false + seenSignalRequests := make(map[uint64]struct{}) go func() { for { request, err := stream.Recv() if err != nil { - if !errors.Is(err, context.Canceled) { - fromClientErrCh <- err + if !errors.Is(err, io.EOF) && !errors.Is(err, context.Canceled) && + status.Code(err) != codes.Canceled { + reportClientError(err) } return @@ -156,7 +203,7 @@ func (rpc *RPC) Exec(stream grpc.BidiStreamingServer[ExecRequest, ExecResponse]) } else { // Close the standard input if err := stdin.Close(); err != nil { - fromClientErrCh <- err + reportClientError(err) return } @@ -166,7 +213,7 @@ func (rpc *RPC) Exec(stream grpc.BidiStreamingServer[ExecRequest, ExecResponse]) } if _, err := stdin.Write(dataToWrite); err != nil { - fromClientErrCh <- err + reportClientError(err) return } @@ -181,8 +228,45 @@ func (rpc *RPC) Exec(stream grpc.BidiStreamingServer[ExecRequest, ExecResponse]) Rows: uint16(typedAction.TerminalResize.GetRows()), Cols: uint16(typedAction.TerminalResize.GetCols()), }); err != nil { - fromClientErrCh <- err + reportClientError(err) + return + } + case *ExecRequest_Signal_: + signalRequest := typedAction.Signal + if signalRequest == nil || signalRequest.RequestId == 0 { + reportClientError(status.Error(codes.InvalidArgument, + "signal request_id must be nonzero")) + return + } + if _, seen := seenSignalRequests[signalRequest.RequestId]; seen { + reportClientError(status.Errorf(codes.InvalidArgument, + "signal request_id %d has already been used", signalRequest.RequestId)) + return + } + seenSignalRequests[signalRequest.RequestId] = struct{}{} + + if err := func() error { + signalMu.Lock() + defer signalMu.Unlock() + + if processExited { + return status.Error(codes.FailedPrecondition, + "managed process has already exited") + } + if err := deliverExecSignal(cmd.Process, signalRequest); err != nil { + return err + } + + return responseSender.send(&ExecResponse{ + Type: &ExecResponse_SignalAck_{ + SignalAck: &ExecResponse_SignalAck{ + RequestId: signalRequest.RequestId, + }, + }, + }) + }(); err != nil { + reportClientError(err) return } } @@ -210,7 +294,7 @@ func (rpc *RPC) Exec(stream grpc.BidiStreamingServer[ExecRequest, ExecResponse]) return err } - if err := stream.Send(&ExecResponse{ + if err := responseSender.send(&ExecResponse{ Type: &ExecResponse_StandardOutput{ StandardOutput: &IOChunk{ Data: slices.Clone(buf[:n]), @@ -240,7 +324,7 @@ func (rpc *RPC) Exec(stream grpc.BidiStreamingServer[ExecRequest, ExecResponse]) return err } - if err := stream.Send(&ExecResponse{ + if err := responseSender.send(&ExecResponse{ Type: &ExecResponse_StandardError{ StandardError: &IOChunk{ Data: slices.Clone(buf[:n]), @@ -257,19 +341,29 @@ func (rpc *RPC) Exec(stream grpc.BidiStreamingServer[ExecRequest, ExecResponse]) zap.S().Warnf("%v", err) } - // Wait for the command to finish - exitCode := 0 + // Wait for the command to finish before allowing the final exit response. + waitErr := cmd.Wait() + signalMu.Lock() + defer signalMu.Unlock() + processExited = true - if err := cmd.Wait(); err != nil { + select { + case err := <-fromClientErrCh: + return err + default: + } + + exitCode := 0 + if waitErr != nil { var exitError *exec.ExitError - if errors.As(err, &exitError) { + if errors.As(waitErr, &exitError) { exitCode = exitError.ExitCode() } else { - return err + return waitErr } } - return stream.Send(&ExecResponse{ + return responseSender.send(&ExecResponse{ Type: &ExecResponse_Exit_{ Exit: &ExecResponse_Exit{ Code: int32(exitCode), @@ -278,6 +372,43 @@ func (rpc *RPC) Exec(stream grpc.BidiStreamingServer[ExecRequest, ExecResponse]) }) } +func deliverExecSignal(process *os.Process, request *ExecRequest_Signal) error { + var signal syscall.Signal + switch request.Signal { + case uint32(syscall.SIGHUP), uint32(syscall.SIGINT), uint32(syscall.SIGQUIT), + uint32(syscall.SIGKILL), uint32(syscall.SIGTERM), uint32(syscall.SIGUSR1), + uint32(syscall.SIGUSR2), uint32(syscall.SIGCONT), uint32(syscall.SIGSTOP), + uint32(syscall.SIGTSTP), uint32(syscall.SIGWINCH): + signal = syscall.Signal(request.Signal) + default: + return status.Errorf(codes.InvalidArgument, + "unsupported signal %d", request.Signal) + } + + var err error + if request.All { + // Every attached command leads its own process group, so a negative + // managed PID cannot signal the agent or another execution. + err = syscall.Kill(-process.Pid, signal) + } else { + err = process.Signal(signal) + } + if err != nil { + if errors.Is(err, os.ErrProcessDone) || errors.Is(err, syscall.ESRCH) { + return status.Errorf(codes.FailedPrecondition, + "managed process is no longer running: %v", err) + } + if errors.Is(err, syscall.EPERM) { + return status.Errorf(codes.PermissionDenied, + "cannot signal managed process: %v", err) + } + return status.Errorf(codes.Internal, + "cannot deliver signal %d to managed process: %v", request.Signal, err) + } + + return nil +} + func applyExecOverrides(cmd *exec.Cmd, command *ExecRequest_Command) { if command.Workdir != "" { cmd.Dir = command.Workdir diff --git a/internal/rpc/exec_test.go b/internal/rpc/exec_test.go new file mode 100644 index 0000000..2d55099 --- /dev/null +++ b/internal/rpc/exec_test.go @@ -0,0 +1,740 @@ +package rpc + +import ( + "context" + "errors" + "fmt" + "io" + "net" + "os" + "path/filepath" + "strconv" + "strings" + "syscall" + "testing" + "time" + + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/credentials/insecure" + "google.golang.org/grpc/status" + "google.golang.org/grpc/test/bufconn" + "google.golang.org/protobuf/proto" + "google.golang.org/protobuf/reflect/protodesc" + "google.golang.org/protobuf/reflect/protoreflect" + "google.golang.org/protobuf/reflect/protoregistry" + "google.golang.org/protobuf/types/descriptorpb" + "google.golang.org/protobuf/types/dynamicpb" +) + +const execTestTimeout = 15 * time.Second + +type execTestResult struct { + stdout strings.Builder + stderr strings.Builder + acks []uint64 + exit int32 +} + +func newExecTestClient(t *testing.T) (*grpc.ClientConn, AgentClient) { + t.Helper() + + listener := bufconn.Listen(1024 * 1024) + server, err := New(listener) + if err != nil { + t.Fatalf("create RPC server: %v", err) + } + + serverContext, stopServer := context.WithCancel(context.Background()) + go func() { + _ = server.Run(serverContext) + }() + + connection, err := grpc.NewClient("passthrough:///bufnet", + grpc.WithContextDialer(func(context.Context, string) (net.Conn, error) { + return listener.Dial() + }), + grpc.WithTransportCredentials(insecure.NewCredentials()), + ) + if err != nil { + stopServer() + _ = listener.Close() + t.Fatalf("create in-memory gRPC client: %v", err) + } + + t.Cleanup(func() { + stopServer() + _ = connection.Close() + _ = listener.Close() + }) + + return connection, NewAgentClient(connection) +} + +func startExecTest(t *testing.T, client AgentClient, command *ExecRequest_Command) ( + grpc.BidiStreamingClient[ExecRequest, ExecResponse], *ExecResponse_Started, +) { + t.Helper() + + ctx, cancel := context.WithTimeout(context.Background(), execTestTimeout) + t.Cleanup(cancel) + + stream, err := client.Exec(ctx) + if err != nil { + t.Fatalf("open exec stream: %v", err) + } + if err := stream.Send(&ExecRequest{ + Type: &ExecRequest_Command_{Command: command}, + }); err != nil { + t.Fatalf("send exec command: %v", err) + } + + response, err := stream.Recv() + if err != nil { + t.Fatalf("receive first exec response: %v", err) + } + started := response.GetStarted() + if started == nil { + t.Fatalf("first exec response = %T, want Started", response.GetType()) + } + if started.Pid == 0 { + t.Fatal("Started contains a zero process ID") + } + + return stream, started +} + +func addExecTestResponse(t *testing.T, result *execTestResult, response *ExecResponse) bool { + t.Helper() + + switch event := response.GetType().(type) { + case *ExecResponse_StandardOutput: + _, _ = result.stdout.Write(event.StandardOutput.Data) + case *ExecResponse_StandardError: + _, _ = result.stderr.Write(event.StandardError.Data) + case *ExecResponse_SignalAck_: + result.acks = append(result.acks, event.SignalAck.RequestId) + case *ExecResponse_Exit_: + result.exit = event.Exit.Code + return true + default: + t.Fatalf("unexpected exec response: %T", event) + } + + return false +} + +func finishExecTest(t *testing.T, stream grpc.BidiStreamingClient[ExecRequest, ExecResponse], + result *execTestResult, +) { + t.Helper() + + for { + response, err := stream.Recv() + if err != nil { + t.Fatalf("receive exec response: %v", err) + } + if addExecTestResponse(t, result, response) { + break + } + } + + if response, err := stream.Recv(); !errors.Is(err, io.EOF) { + t.Fatalf("response after Exit = %v, %v; want EOF", response, err) + } +} + +func waitForExecOutput(t *testing.T, stream grpc.BidiStreamingClient[ExecRequest, ExecResponse], + result *execTestResult, text string, +) { + t.Helper() + + for !strings.Contains(result.stdout.String(), text) { + response, err := stream.Recv() + if err != nil { + t.Fatalf("wait for output %q: %v", text, err) + } + if addExecTestResponse(t, result, response) { + t.Fatalf("process exited before producing %q; stdout = %q", text, result.stdout.String()) + } + } +} + +func sendExecTestSignal(t *testing.T, stream grpc.BidiStreamingClient[ExecRequest, ExecResponse], + requestID uint64, signal syscall.Signal, all bool, +) { + t.Helper() + + if err := stream.Send(&ExecRequest{ + Type: &ExecRequest_Signal_{ + Signal: &ExecRequest_Signal{ + RequestId: requestID, + Signal: uint32(signal), + All: all, + }, + }, + }); err != nil { + t.Fatalf("send signal request %d: %v", requestID, err) + } +} + +func TestExecStartedContainsRealPIDAndPrecedesOutput(t *testing.T) { + _, client := newExecTestClient(t) + stream, started := startExecTest(t, client, &ExecRequest_Command{ + Name: "sh", + Args: []string{"-c", `printf '%s\n' "$$"; printf 'standard-error\n' >&2`}, + }) + + var result execTestResult + finishExecTest(t, stream, &result) + + actualPID, err := strconv.ParseUint(strings.TrimSpace(result.stdout.String()), 10, 32) + if err != nil { + t.Fatalf("parse managed shell PID %q: %v", result.stdout.String(), err) + } + if started.Pid != uint32(actualPID) { + t.Fatalf("Started PID = %d, actual guest process PID = %d", started.Pid, actualPID) + } + if got := result.stderr.String(); got != "standard-error\n" { + t.Fatalf("standard error = %q, want %q", got, "standard-error\n") + } + if result.exit != 0 { + t.Fatalf("exit code = %d, want 0", result.exit) + } +} + +func TestExecStartedPrecedesFastCommandOutput(t *testing.T) { + _, client := newExecTestClient(t) + + for iteration := range 24 { + t.Run(fmt.Sprintf("command-%02d", iteration), func(t *testing.T) { + stream, _ := startExecTest(t, client, &ExecRequest_Command{ + Name: "sh", + Args: []string{"-c", "printf fast"}, + }) + + var result execTestResult + finishExecTest(t, stream, &result) + if got := result.stdout.String(); got != "fast" { + t.Fatalf("standard output = %q, want %q", got, "fast") + } + if result.exit != 0 { + t.Fatalf("exit code = %d, want 0", result.exit) + } + }) + } +} + +func TestExecSerializesConcurrentStandardStreams(t *testing.T) { + _, client := newExecTestClient(t) + stream, _ := startExecTest(t, client, &ExecRequest_Command{ + Name: "sh", + Args: []string{"-c", `i=0; while [ "$i" -lt 128 ]; do printf 'out-%s\n' "$i"; printf 'err-%s\n' "$i" >&2; i=$((i + 1)); done`}, + }) + + var result execTestResult + finishExecTest(t, stream, &result) + + if got := strings.Count(result.stdout.String(), "\n"); got != 128 { + t.Fatalf("standard output line count = %d, want 128", got) + } + if got := strings.Count(result.stderr.String(), "\n"); got != 128 { + t.Fatalf("standard error line count = %d, want 128", got) + } + if result.exit != 0 { + t.Fatalf("exit code = %d, want 0", result.exit) + } +} + +func TestExecPreservesEnvironmentAndWorkingDirectory(t *testing.T) { + _, client := newExecTestClient(t) + workdir := t.TempDir() + resolvedWorkdir, err := filepath.EvalSymlinks(workdir) + if err != nil { + t.Fatalf("resolve test working directory: %v", err) + } + stream, _ := startExecTest(t, client, &ExecRequest_Command{ + Name: "sh", + Args: []string{"-c", `printf '%s:%s' "$TART_EXEC_LIFECYCLE_TEST" "$PWD"`}, + Env: map[string]string{"TART_EXEC_LIFECYCLE_TEST": "preserved"}, + Workdir: workdir, + }) + + var result execTestResult + finishExecTest(t, stream, &result) + + if got, want := result.stdout.String(), "preserved:"+resolvedWorkdir; got != want { + t.Fatalf("environment and workdir = %q, want %q", got, want) + } + if result.exit != 0 { + t.Fatalf("exit code = %d, want 0", result.exit) + } +} + +func TestExecInteractiveStandardInputAndEOF(t *testing.T) { + _, client := newExecTestClient(t) + stream, _ := startExecTest(t, client, &ExecRequest_Command{ + Name: "sh", + Args: []string{"-c", "cat"}, + Interactive: true, + }) + + for _, data := range [][]byte{[]byte("interactive input\n"), {}} { + if err := stream.Send(&ExecRequest{ + Type: &ExecRequest_StandardInput{StandardInput: &IOChunk{Data: data}}, + }); err != nil { + t.Fatalf("send standard input %q: %v", data, err) + } + } + if err := stream.CloseSend(); err != nil { + t.Fatalf("half-close client stream: %v", err) + } + + var result execTestResult + finishExecTest(t, stream, &result) + + if got := result.stdout.String(); got != "interactive input\n" { + t.Fatalf("standard output = %q, want %q", got, "interactive input\n") + } + if result.exit != 0 { + t.Fatalf("exit code = %d, want 0", result.exit) + } +} + +func TestExecInteractivePTYAndResize(t *testing.T) { + _, client := newExecTestClient(t) + stream, _ := startExecTest(t, client, &ExecRequest_Command{ + Name: "sh", + Args: []string{"-c", `stty size; IFS= read -r line; stty size; printf 'input:%s\n' "$line"`}, + Interactive: true, + Tty: true, + TerminalSize: &TerminalSize{ + Rows: 24, + Cols: 80, + }, + }) + + var result execTestResult + waitForExecOutput(t, stream, &result, "24 80") + + if err := stream.Send(&ExecRequest{ + Type: &ExecRequest_TerminalResize{ + TerminalResize: &TerminalSize{Rows: 41, Cols: 101}, + }, + }); err != nil { + t.Fatalf("resize pseudo-terminal: %v", err) + } + if err := stream.Send(&ExecRequest{ + Type: &ExecRequest_StandardInput{ + StandardInput: &IOChunk{Data: []byte("hello from a tty\n")}, + }, + }); err != nil { + t.Fatalf("send pseudo-terminal input: %v", err) + } + + finishExecTest(t, stream, &result) + for _, want := range []string{"24 80", "41 101", "input:hello from a tty"} { + if !strings.Contains(result.stdout.String(), want) { + t.Fatalf("pseudo-terminal output %q does not contain %q", result.stdout.String(), want) + } + } + if result.exit != 0 { + t.Fatalf("exit code = %d, want 0", result.exit) + } +} + +func TestExecSignalAcknowledgesSuccessfulDelivery(t *testing.T) { + tests := []struct { + name string + signal syscall.Signal + all bool + tty bool + }{ + {name: "process-SIGTERM", signal: syscall.SIGTERM}, + {name: "process-SIGKILL", signal: syscall.SIGKILL}, + {name: "group-SIGTERM", signal: syscall.SIGTERM, all: true}, + {name: "group-SIGKILL", signal: syscall.SIGKILL, all: true}, + {name: "pty-group-SIGTERM", signal: syscall.SIGTERM, all: true, tty: true}, + {name: "pty-group-SIGKILL", signal: syscall.SIGKILL, all: true, tty: true}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + _, client := newExecTestClient(t) + stream, started := startExecTest(t, client, &ExecRequest_Command{ + Name: "sh", + Args: []string{"-c", "exec sleep 30"}, + Tty: test.tty, + }) + + groupID, err := syscall.Getpgid(int(started.Pid)) + if err != nil { + t.Fatalf("read managed process group: %v", err) + } + if groupID != int(started.Pid) { + t.Fatalf("managed process group = %d, want managed PID %d", groupID, started.Pid) + } + if groupID == syscall.Getpgrp() { + t.Fatal("managed process shares the agent's process group") + } + + const requestID = 73 + sendExecTestSignal(t, stream, requestID, test.signal, test.all) + + var result execTestResult + finishExecTest(t, stream, &result) + if len(result.acks) != 1 || result.acks[0] != requestID { + t.Fatalf("signal acknowledgments = %v, want [%d]", result.acks, requestID) + } + if result.exit != -1 { + t.Fatalf("signaled exit code = %d, want -1", result.exit) + } + }) + } +} + +func TestExecSignalRequestsAreCorrelated(t *testing.T) { + _, client := newExecTestClient(t) + stream, _ := startExecTest(t, client, &ExecRequest_Command{ + Name: "sh", + Args: []string{"-c", "exec sleep 30"}, + }) + + for _, request := range []struct { + id uint64 + signal syscall.Signal + }{ + {id: 11, signal: syscall.SIGSTOP}, + {id: 29, signal: syscall.SIGCONT}, + } { + sendExecTestSignal(t, stream, request.id, request.signal, false) + response, err := stream.Recv() + if err != nil { + t.Fatalf("receive acknowledgment for request %d: %v", request.id, err) + } + if ack := response.GetSignalAck(); ack == nil || ack.RequestId != request.id { + t.Fatalf("acknowledgment = %v, want request_id %d", response, request.id) + } + } + + const finalRequestID = 47 + sendExecTestSignal(t, stream, finalRequestID, syscall.SIGTERM, false) + + var result execTestResult + finishExecTest(t, stream, &result) + if len(result.acks) != 1 || result.acks[0] != finalRequestID { + t.Fatalf("final acknowledgments = %v, want [%d]", result.acks, finalRequestID) + } + if result.exit != -1 { + t.Fatalf("signaled exit code = %d, want -1", result.exit) + } +} + +func TestExecSignalAllIsolatesSiblingExecutions(t *testing.T) { + _, client := newExecTestClient(t) + marker := filepath.Join(t.TempDir(), "group-terminated") + groupScript := `sh -c 'trap '"'"'printf terminated > "$1"; exit 0'"'"' TERM; printf "group-ready\n"; while :; do sleep 1; done' _ "$1" & wait` + + groupStream, _ := startExecTest(t, client, &ExecRequest_Command{ + Name: "sh", + Args: []string{"-c", groupScript, "group", marker}, + }) + var groupResult execTestResult + waitForExecOutput(t, groupStream, &groupResult, "group-ready") + + siblingStream, _ := startExecTest(t, client, &ExecRequest_Command{ + Name: "sh", + Args: []string{"-c", "cat"}, + Interactive: true, + }) + + const groupRequestID = 101 + sendExecTestSignal(t, groupStream, groupRequestID, syscall.SIGTERM, true) + finishExecTest(t, groupStream, &groupResult) + if len(groupResult.acks) != 1 || groupResult.acks[0] != groupRequestID { + t.Fatalf("group acknowledgments = %v, want [%d]", groupResult.acks, groupRequestID) + } + + deadline := time.NewTimer(3 * time.Second) + ticker := time.NewTicker(10 * time.Millisecond) + defer deadline.Stop() + defer ticker.Stop() + for { + contents, err := os.ReadFile(marker) + if err == nil { + if string(contents) != "terminated" { + t.Fatalf("group child marker = %q, want %q", contents, "terminated") + } + break + } + if !errors.Is(err, os.ErrNotExist) { + t.Fatalf("read group child marker: %v", err) + } + + select { + case <-deadline.C: + t.Fatal("signal-all did not reach the managed process group's child") + case <-ticker.C: + } + } + + for _, data := range [][]byte{[]byte("sibling still alive\n"), {}} { + if err := siblingStream.Send(&ExecRequest{ + Type: &ExecRequest_StandardInput{StandardInput: &IOChunk{Data: data}}, + }); err != nil { + t.Fatalf("write to isolated sibling execution: %v", err) + } + } + + var siblingResult execTestResult + finishExecTest(t, siblingStream, &siblingResult) + if got := siblingResult.stdout.String(); got != "sibling still alive\n" { + t.Fatalf("sibling output = %q, want %q", got, "sibling still alive\n") + } + if siblingResult.exit != 0 { + t.Fatalf("sibling exit code = %d, want 0", siblingResult.exit) + } +} + +func TestExecRejectsInvalidSignalRequestsWithoutAcknowledging(t *testing.T) { + tests := []struct { + name string + request *ExecRequest_Signal + }{ + { + name: "missing-request-id", + request: &ExecRequest_Signal{Signal: uint32(syscall.SIGTERM)}, + }, + { + name: "zero-signal", + request: &ExecRequest_Signal{RequestId: 1}, + }, + { + name: "unsupported-signal", + request: &ExecRequest_Signal{RequestId: 1, Signal: ^uint32(0)}, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + _, client := newExecTestClient(t) + stream, _ := startExecTest(t, client, &ExecRequest_Command{ + Name: "sh", + Args: []string{"-c", "exec sleep 30"}, + }) + + if err := stream.Send(&ExecRequest{ + Type: &ExecRequest_Signal_{Signal: test.request}, + }); err != nil { + t.Fatalf("send invalid signal request: %v", err) + } + + for { + response, err := stream.Recv() + if err != nil { + if got := status.Code(err); got != codes.InvalidArgument { + t.Fatalf("invalid signal status = %v (%v), want InvalidArgument", got, err) + } + break + } + if ack := response.GetSignalAck(); ack != nil { + t.Fatalf("invalid request unexpectedly acknowledged: %v", ack) + } + if exit := response.GetExit(); exit != nil { + t.Fatalf("invalid request produced a successful exit event: %v", exit) + } + } + }) + } +} + +func TestExecRejectsReusedSignalRequestID(t *testing.T) { + _, client := newExecTestClient(t) + stream, _ := startExecTest(t, client, &ExecRequest_Command{ + Name: "sh", + Args: []string{"-c", `trap '' USR1; printf ready; exec sleep 30`}, + }) + var result execTestResult + waitForExecOutput(t, stream, &result, "ready") + + const requestID = 19 + sendExecTestSignal(t, stream, requestID, syscall.SIGUSR1, false) + response, err := stream.Recv() + if err != nil { + t.Fatalf("receive original signal acknowledgment: %v", err) + } + if ack := response.GetSignalAck(); ack == nil || ack.RequestId != requestID { + t.Fatalf("original acknowledgment = %v, want request_id %d", response, requestID) + } + + sendExecTestSignal(t, stream, requestID, syscall.SIGUSR1, false) + for { + response, err = stream.Recv() + if err != nil { + if got := status.Code(err); got != codes.InvalidArgument { + t.Fatalf("reused request status = %v (%v), want InvalidArgument", got, err) + } + break + } + if ack := response.GetSignalAck(); ack != nil { + t.Fatalf("reused request unexpectedly acknowledged: %v", ack) + } + } +} + +func TestExecDetachedRetainsLegacyExitOnlyResponse(t *testing.T) { + _, client := newExecTestClient(t) + ctx, cancel := context.WithTimeout(context.Background(), execTestTimeout) + defer cancel() + + stream, err := client.Exec(ctx) + if err != nil { + t.Fatalf("open detached exec stream: %v", err) + } + if err := stream.Send(&ExecRequest{ + Type: &ExecRequest_Command_{ + Command: &ExecRequest_Command{ + Name: "sh", + Args: []string{"-c", "exit 0"}, + Detach: true, + }, + }, + }); err != nil { + t.Fatalf("send detached command: %v", err) + } + + response, err := stream.Recv() + if err != nil { + t.Fatalf("receive detached exit: %v", err) + } + if exit := response.GetExit(); exit == nil || exit.Code != 0 { + t.Fatalf("first detached response = %v, want legacy exit code 0", response) + } + if response, err := stream.Recv(); !errors.Is(err, io.EOF) { + t.Fatalf("response after detached Exit = %v, %v; want EOF", response, err) + } +} + +func legacyExecDescriptors(t *testing.T) (protoreflect.MessageDescriptor, protoreflect.MessageDescriptor) { + t.Helper() + + file := proto.Clone(protodesc.ToFileDescriptorProto(File_rpc_agent_proto)).(*descriptorpb.FileDescriptorProto) + for _, message := range file.MessageType { + switch message.GetName() { + case "ExecRequest": + fields := message.Field[:0] + for _, field := range message.Field { + if field.GetNumber() <= 3 { + fields = append(fields, field) + } + } + message.Field = fields + + nested := message.NestedType[:0] + for _, child := range message.NestedType { + if child.GetName() != "Signal" { + nested = append(nested, child) + } + } + message.NestedType = nested + case "ExecResponse": + fields := message.Field[:0] + for _, field := range message.Field { + if field.GetNumber() <= 3 { + fields = append(fields, field) + } + } + message.Field = fields + + nested := message.NestedType[:0] + for _, child := range message.NestedType { + if child.GetName() == "Exit" { + nested = append(nested, child) + } + } + message.NestedType = nested + } + } + + legacy, err := protodesc.NewFile(file, protoregistry.GlobalFiles) + if err != nil { + t.Fatalf("build original Exec protocol descriptors: %v", err) + } + + return legacy.Messages().ByName("ExecRequest"), legacy.Messages().ByName("ExecResponse") +} + +func TestExecLegacyClientIgnoresStartedAndReceivesOriginalEvents(t *testing.T) { + connection, _ := newExecTestClient(t) + requestDescriptor, responseDescriptor := legacyExecDescriptors(t) + ctx, cancel := context.WithTimeout(context.Background(), execTestTimeout) + defer cancel() + + stream, err := connection.NewStream(ctx, &grpc.StreamDesc{ + StreamName: "Exec", + ServerStreams: true, + ClientStreams: true, + }, Agent_Exec_FullMethodName) + if err != nil { + t.Fatalf("open legacy exec stream: %v", err) + } + + request := dynamicpb.NewMessage(requestDescriptor) + commandField := requestDescriptor.Fields().ByName("command") + command := dynamicpb.NewMessage(commandField.Message()) + command.Set(command.Descriptor().Fields().ByName("name"), protoreflect.ValueOfString("sh")) + args := command.Mutable(command.Descriptor().Fields().ByName("args")).List() + args.Append(protoreflect.ValueOfString("-c")) + args.Append(protoreflect.ValueOfString("printf legacy-output; printf legacy-error >&2")) + request.Set(commandField, protoreflect.ValueOfMessage(command)) + if err := stream.SendMsg(request); err != nil { + t.Fatalf("send legacy-format command: %v", err) + } + + first := dynamicpb.NewMessage(responseDescriptor) + if err := stream.RecvMsg(first); err != nil { + t.Fatalf("receive legacy-format Started: %v", err) + } + if field := first.WhichOneof(responseDescriptor.Oneofs().ByName("type")); field != nil { + t.Fatalf("legacy client recognized new Started event as %s", field.FullName()) + } + if len(first.GetUnknown()) == 0 { + t.Fatal("legacy client did not retain the additive unknown Started field") + } + + var stdout, stderr strings.Builder + for { + response := dynamicpb.NewMessage(responseDescriptor) + if err := stream.RecvMsg(response); err != nil { + t.Fatalf("receive legacy-format response: %v", err) + } + field := response.WhichOneof(responseDescriptor.Oneofs().ByName("type")) + if field == nil { + continue + } + + event := response.Get(field).Message() + switch field.Name() { + case "standard_output": + _, _ = stdout.Write(event.Get(event.Descriptor().Fields().ByName("data")).Bytes()) + case "standard_error": + _, _ = stderr.Write(event.Get(event.Descriptor().Fields().ByName("data")).Bytes()) + case "exit": + if code := event.Get(event.Descriptor().Fields().ByName("code")).Int(); code != 0 { + t.Fatalf("legacy exit code = %d, want 0", code) + } + if got := stdout.String(); got != "legacy-output" { + t.Fatalf("legacy standard output = %q, want %q", got, "legacy-output") + } + if got := stderr.String(); got != "legacy-error" { + t.Fatalf("legacy standard error = %q, want %q", got, "legacy-error") + } + if err := stream.RecvMsg(dynamicpb.NewMessage(responseDescriptor)); !errors.Is(err, io.EOF) { + t.Fatalf("legacy response after Exit = %v, want EOF", err) + } + return + default: + t.Fatalf("unexpected legacy event %s", field.FullName()) + } + } +} diff --git a/proto/rpc/agent.proto b/proto/rpc/agent.proto index 638fdd7..401f5b0 100644 --- a/proto/rpc/agent.proto +++ b/proto/rpc/agent.proto @@ -21,10 +21,17 @@ message ExecRequest { string workdir = 8; } + message Signal { + uint64 request_id = 1; + uint32 signal = 2; + bool all = 3; + } + oneof type { Command command = 1; IOChunk standard_input = 2; TerminalSize terminal_resize = 3; + Signal signal = 4; } } @@ -33,10 +40,20 @@ message ExecResponse { int32 code = 1; } + message Started { + uint32 pid = 1; + } + + message SignalAck { + uint64 request_id = 1; + } + oneof type { Exit exit = 1; IOChunk standard_output = 2; IOChunk standard_error = 3; + Started started = 4; + SignalAck signal_ack = 5; } }