diff --git a/go.mod b/go.mod index a6ff3e5..8d8c875 100644 --- a/go.mod +++ b/go.mod @@ -6,7 +6,9 @@ require ( github.com/Masterminds/semver/v3 v3.5.0 github.com/cenkalti/backoff/v5 v5.0.3 github.com/creack/pty v1.1.24 + github.com/google/uuid v1.6.0 github.com/hashicorp/go-version v1.9.0 + github.com/puzpuzpuz/xsync/v4 v4.5.0 github.com/samber/lo v1.53.0 github.com/spf13/cobra v1.10.2 github.com/stretchr/testify v1.11.1 diff --git a/go.sum b/go.sum index d4f0b53..001c9fb 100644 --- a/go.sum +++ b/go.sum @@ -27,6 +27,8 @@ github.com/inconshreveable/mousetrap v1.1.0/go.mod h1:vpF70FUmC8bwa3OWnCshd2FqLf github.com/jessevdk/go-flags v1.4.0/go.mod h1:4FA24M0QyGHXBuZZK/XkWh8h0e1EYbRYJSGM75WSRxI= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/puzpuzpuz/xsync/v4 v4.5.0 h1:vOSWu6b57/emh+L/Cw0BeQfvxa/cogFywXHeGUxQxAg= +github.com/puzpuzpuz/xsync/v4 v4.5.0/go.mod h1:VJDmTCJMBt8igNxnkQd86r+8KUeN1quSfNKu5bLYFQo= github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM= github.com/samber/lo v1.53.0 h1:t975lj2py4kJPQ6haz1QMgtId2gtmfktACxIXArw3HM= github.com/samber/lo v1.53.0/go.mod h1:4+MXEGsJzbKGaUEQFKBq2xtfuznW9oz/WrgyzMzRoM0= diff --git a/internal/rpc/agent.pb.go b/internal/rpc/agent.pb.go index f1f42cb..f4d3cfb 100644 --- a/internal/rpc/agent.pb.go +++ b/internal/rpc/agent.pb.go @@ -9,7 +9,7 @@ package rpc import ( protoreflect "google.golang.org/protobuf/reflect/protoreflect" protoimpl "google.golang.org/protobuf/runtime/protoimpl" - _ "google.golang.org/protobuf/types/known/emptypb" + emptypb "google.golang.org/protobuf/types/known/emptypb" reflect "reflect" sync "sync" unsafe "unsafe" @@ -22,53 +22,53 @@ const ( _ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20) ) -type ExecRequest_SendSignal_Signal int32 +type SignalRequest_Signal int32 const ( - ExecRequest_SendSignal_SIGNAL_UNSPECIFIED ExecRequest_SendSignal_Signal = 0 - ExecRequest_SendSignal_SIGNAL_SIGTERM ExecRequest_SendSignal_Signal = 1 - ExecRequest_SendSignal_SIGNAL_SIGKILL ExecRequest_SendSignal_Signal = 2 + SignalRequest_SIGNAL_UNSPECIFIED SignalRequest_Signal = 0 + SignalRequest_SIGNAL_SIGTERM SignalRequest_Signal = 1 + SignalRequest_SIGNAL_SIGKILL SignalRequest_Signal = 2 ) -// Enum value maps for ExecRequest_SendSignal_Signal. +// Enum value maps for SignalRequest_Signal. var ( - ExecRequest_SendSignal_Signal_name = map[int32]string{ + SignalRequest_Signal_name = map[int32]string{ 0: "SIGNAL_UNSPECIFIED", 1: "SIGNAL_SIGTERM", 2: "SIGNAL_SIGKILL", } - ExecRequest_SendSignal_Signal_value = map[string]int32{ + SignalRequest_Signal_value = map[string]int32{ "SIGNAL_UNSPECIFIED": 0, "SIGNAL_SIGTERM": 1, "SIGNAL_SIGKILL": 2, } ) -func (x ExecRequest_SendSignal_Signal) Enum() *ExecRequest_SendSignal_Signal { - p := new(ExecRequest_SendSignal_Signal) +func (x SignalRequest_Signal) Enum() *SignalRequest_Signal { + p := new(SignalRequest_Signal) *p = x return p } -func (x ExecRequest_SendSignal_Signal) String() string { +func (x SignalRequest_Signal) String() string { return protoimpl.X.EnumStringOf(x.Descriptor(), protoreflect.EnumNumber(x)) } -func (ExecRequest_SendSignal_Signal) Descriptor() protoreflect.EnumDescriptor { +func (SignalRequest_Signal) Descriptor() protoreflect.EnumDescriptor { return file_rpc_agent_proto_enumTypes[0].Descriptor() } -func (ExecRequest_SendSignal_Signal) Type() protoreflect.EnumType { +func (SignalRequest_Signal) Type() protoreflect.EnumType { return &file_rpc_agent_proto_enumTypes[0] } -func (x ExecRequest_SendSignal_Signal) Number() protoreflect.EnumNumber { +func (x SignalRequest_Signal) Number() protoreflect.EnumNumber { return protoreflect.EnumNumber(x) } -// Deprecated: Use ExecRequest_SendSignal_Signal.Descriptor instead. -func (ExecRequest_SendSignal_Signal) EnumDescriptor() ([]byte, []int) { - return file_rpc_agent_proto_rawDescGZIP(), []int{0, 0, 0} +// Deprecated: Use SignalRequest_Signal.Descriptor instead. +func (SignalRequest_Signal) EnumDescriptor() ([]byte, []int) { + return file_rpc_agent_proto_rawDescGZIP(), []int{6, 0} } type ExecRequest struct { @@ -78,7 +78,6 @@ type ExecRequest struct { // *ExecRequest_Command_ // *ExecRequest_StandardInput // *ExecRequest_TerminalResize - // *ExecRequest_SendSignal_ Type isExecRequest_Type `protobuf_oneof:"type"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache @@ -148,15 +147,6 @@ func (x *ExecRequest) GetTerminalResize() *TerminalSize { return nil } -func (x *ExecRequest) GetSendSignal() *ExecRequest_SendSignal { - if x != nil { - if x, ok := x.Type.(*ExecRequest_SendSignal_); ok { - return x.SendSignal - } - } - return nil -} - type isExecRequest_Type interface { isExecRequest_Type() } @@ -173,18 +163,12 @@ type ExecRequest_TerminalResize struct { TerminalResize *TerminalSize `protobuf:"bytes,3,opt,name=terminal_resize,json=terminalResize,proto3,oneof"` } -type ExecRequest_SendSignal_ struct { - SendSignal *ExecRequest_SendSignal `protobuf:"bytes,4,opt,name=send_signal,json=sendSignal,proto3,oneof"` -} - func (*ExecRequest_Command_) isExecRequest_Type() {} func (*ExecRequest_StandardInput) isExecRequest_Type() {} func (*ExecRequest_TerminalResize) isExecRequest_Type() {} -func (*ExecRequest_SendSignal_) isExecRequest_Type() {} - type ExecResponse struct { state protoimpl.MessageState `protogen:"open.v1"` // Types that are valid to be assigned to Type: @@ -475,27 +459,28 @@ func (x *ResolveIPResponse) GetIp() string { return "" } -type ExecRequest_SendSignal struct { - state protoimpl.MessageState `protogen:"open.v1"` - Signal ExecRequest_SendSignal_Signal `protobuf:"varint,1,opt,name=signal,proto3,enum=ExecRequest_SendSignal_Signal" json:"signal,omitempty"` +type SignalRequest struct { + state protoimpl.MessageState `protogen:"open.v1"` + ExecId string `protobuf:"bytes,1,opt,name=exec_id,json=execId,proto3" json:"exec_id,omitempty"` + Signal SignalRequest_Signal `protobuf:"varint,2,opt,name=signal,proto3,enum=SignalRequest_Signal" json:"signal,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } -func (x *ExecRequest_SendSignal) Reset() { - *x = ExecRequest_SendSignal{} +func (x *SignalRequest) Reset() { + *x = SignalRequest{} mi := &file_rpc_agent_proto_msgTypes[6] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } -func (x *ExecRequest_SendSignal) String() string { +func (x *SignalRequest) String() string { return protoimpl.X.MessageStringOf(x) } -func (*ExecRequest_SendSignal) ProtoMessage() {} +func (*SignalRequest) ProtoMessage() {} -func (x *ExecRequest_SendSignal) ProtoReflect() protoreflect.Message { +func (x *SignalRequest) ProtoReflect() protoreflect.Message { mi := &file_rpc_agent_proto_msgTypes[6] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) @@ -507,16 +492,23 @@ func (x *ExecRequest_SendSignal) ProtoReflect() protoreflect.Message { return mi.MessageOf(x) } -// Deprecated: Use ExecRequest_SendSignal.ProtoReflect.Descriptor instead. -func (*ExecRequest_SendSignal) Descriptor() ([]byte, []int) { - return file_rpc_agent_proto_rawDescGZIP(), []int{0, 0} +// Deprecated: Use SignalRequest.ProtoReflect.Descriptor instead. +func (*SignalRequest) Descriptor() ([]byte, []int) { + return file_rpc_agent_proto_rawDescGZIP(), []int{6} } -func (x *ExecRequest_SendSignal) GetSignal() ExecRequest_SendSignal_Signal { +func (x *SignalRequest) GetExecId() string { + if x != nil { + return x.ExecId + } + return "" +} + +func (x *SignalRequest) GetSignal() SignalRequest_Signal { if x != nil { return x.Signal } - return ExecRequest_SendSignal_SIGNAL_UNSPECIFIED + return SignalRequest_SIGNAL_UNSPECIFIED } type ExecRequest_Command struct { @@ -529,6 +521,7 @@ type ExecRequest_Command struct { Detach bool `protobuf:"varint,6,opt,name=detach,proto3" json:"detach,omitempty"` Env map[string]string `protobuf:"bytes,7,rep,name=env,proto3" json:"env,omitempty" protobuf_key:"bytes,1,opt,name=key" protobuf_val:"bytes,2,opt,name=value"` Workdir string `protobuf:"bytes,8,opt,name=workdir,proto3" json:"workdir,omitempty"` + User string `protobuf:"bytes,9,opt,name=user,proto3" json:"user,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -560,7 +553,7 @@ func (x *ExecRequest_Command) ProtoReflect() protoreflect.Message { // Deprecated: Use ExecRequest_Command.ProtoReflect.Descriptor instead. func (*ExecRequest_Command) Descriptor() ([]byte, []int) { - return file_rpc_agent_proto_rawDescGZIP(), []int{0, 1} + return file_rpc_agent_proto_rawDescGZIP(), []int{0, 0} } func (x *ExecRequest_Command) GetName() string { @@ -619,6 +612,13 @@ func (x *ExecRequest_Command) GetWorkdir() string { return "" } +func (x *ExecRequest_Command) GetUser() string { + if x != nil { + return x.User + } + return "" +} + type ExecResponse_Exit struct { state protoimpl.MessageState `protogen:"open.v1"` Code int32 `protobuf:"varint,1,opt,name=code,proto3" json:"code,omitempty"` @@ -665,6 +665,7 @@ func (x *ExecResponse_Exit) GetCode() int32 { type ExecResponse_Started struct { state protoimpl.MessageState `protogen:"open.v1"` + ExecId string `protobuf:"bytes,1,opt,name=exec_id,json=execId,proto3" json:"exec_id,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -699,24 +700,22 @@ func (*ExecResponse_Started) Descriptor() ([]byte, []int) { return file_rpc_agent_proto_rawDescGZIP(), []int{1, 1} } +func (x *ExecResponse_Started) GetExecId() string { + if x != nil { + return x.ExecId + } + return "" +} + var File_rpc_agent_proto protoreflect.FileDescriptor const file_rpc_agent_proto_rawDesc = "" + "\n" + - "\x0frpc/agent.proto\x1a\x1bgoogle/protobuf/empty.proto\"\xb8\x05\n" + + "\x0frpc/agent.proto\x1a\x1bgoogle/protobuf/empty.proto\"\xff\x03\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\x12:\n" + - "\vsend_signal\x18\x04 \x01(\v2\x17.ExecRequest.SendSignalH\x00R\n" + - "sendSignal\x1a\x8e\x01\n" + - "\n" + - "SendSignal\x126\n" + - "\x06signal\x18\x01 \x01(\x0e2\x1e.ExecRequest.SendSignal.SignalR\x06signal\"H\n" + - "\x06Signal\x12\x16\n" + - "\x12SIGNAL_UNSPECIFIED\x10\x00\x12\x12\n" + - "\x0eSIGNAL_SIGTERM\x10\x01\x12\x12\n" + - "\x0eSIGNAL_SIGKILL\x10\x02\x1a\xb4\x02\n" + + "\x0fterminal_resize\x18\x03 \x01(\v2\r.TerminalSizeH\x00R\x0eterminalResize\x1a\xc8\x02\n" + "\aCommand\x12\x12\n" + "\x04name\x18\x01 \x01(\tR\x04name\x12\x12\n" + "\x04args\x18\x02 \x03(\tR\x04args\x12 \n" + @@ -725,19 +724,21 @@ const file_rpc_agent_proto_rawDesc = "" + "\rterminal_size\x18\x05 \x01(\v2\r.TerminalSizeR\fterminalSize\x12\x16\n" + "\x06detach\x18\x06 \x01(\bR\x06detach\x12/\n" + "\x03env\x18\a \x03(\v2\x1d.ExecRequest.Command.EnvEntryR\x03env\x12\x18\n" + - "\aworkdir\x18\b \x01(\tR\aworkdir\x1a6\n" + + "\aworkdir\x18\b \x01(\tR\aworkdir\x12\x12\n" + + "\x04user\x18\t \x01(\tR\x04user\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\"\x82\x02\n" + + "\x04type\"\x9b\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\x121\n" + "\astarted\x18\x04 \x01(\v2\x15.ExecResponse.StartedH\x00R\astarted\x1a\x1a\n" + "\x04Exit\x12\x12\n" + - "\x04code\x18\x01 \x01(\x05R\x04code\x1a\t\n" + - "\aStartedB\x06\n" + + "\x04code\x18\x01 \x01(\x05R\x04code\x1a\"\n" + + "\aStarted\x12\x17\n" + + "\aexec_id\x18\x01 \x01(\tR\x06execIdB\x06\n" + "\x04type\"6\n" + "\fTerminalSize\x12\x12\n" + "\x04rows\x18\x01 \x01(\rR\x04rows\x12\x12\n" + @@ -746,9 +747,17 @@ const file_rpc_agent_proto_rawDesc = "" + "\x04data\x18\x01 \x01(\fR\x04data\"\x12\n" + "\x10ResolveIPRequest\"#\n" + "\x11ResolveIPResponse\x12\x0e\n" + - "\x02ip\x18\x01 \x01(\tR\x02ip2d\n" + + "\x02ip\x18\x01 \x01(\tR\x02ip\"\xa1\x01\n" + + "\rSignalRequest\x12\x17\n" + + "\aexec_id\x18\x01 \x01(\tR\x06execId\x12-\n" + + "\x06signal\x18\x02 \x01(\x0e2\x15.SignalRequest.SignalR\x06signal\"H\n" + + "\x06Signal\x12\x16\n" + + "\x12SIGNAL_UNSPECIFIED\x10\x00\x12\x12\n" + + "\x0eSIGNAL_SIGTERM\x10\x01\x12\x12\n" + + "\x0eSIGNAL_SIGKILL\x10\x022\x96\x01\n" + "\x05Agent\x12'\n" + - "\x04Exec\x12\f.ExecRequest\x1a\r.ExecResponse(\x010\x01\x122\n" + + "\x04Exec\x12\f.ExecRequest\x1a\r.ExecResponse(\x010\x01\x120\n" + + "\x06Signal\x12\x0e.SignalRequest\x1a\x16.google.protobuf.Empty\x122\n" + "\tResolveIP\x12\x11.ResolveIPRequest\x1a\x12.ResolveIPResponseB5Z3github.com/cirruslabs/tart-guest-agent/internal/rpcb\x06proto3" var ( @@ -766,40 +775,42 @@ func file_rpc_agent_proto_rawDescGZIP() []byte { var file_rpc_agent_proto_enumTypes = make([]protoimpl.EnumInfo, 1) var file_rpc_agent_proto_msgTypes = make([]protoimpl.MessageInfo, 11) var file_rpc_agent_proto_goTypes = []any{ - (ExecRequest_SendSignal_Signal)(0), // 0: ExecRequest.SendSignal.Signal - (*ExecRequest)(nil), // 1: ExecRequest - (*ExecResponse)(nil), // 2: ExecResponse - (*TerminalSize)(nil), // 3: TerminalSize - (*IOChunk)(nil), // 4: IOChunk - (*ResolveIPRequest)(nil), // 5: ResolveIPRequest - (*ResolveIPResponse)(nil), // 6: ResolveIPResponse - (*ExecRequest_SendSignal)(nil), // 7: ExecRequest.SendSignal - (*ExecRequest_Command)(nil), // 8: ExecRequest.Command - nil, // 9: ExecRequest.Command.EnvEntry - (*ExecResponse_Exit)(nil), // 10: ExecResponse.Exit - (*ExecResponse_Started)(nil), // 11: ExecResponse.Started + (SignalRequest_Signal)(0), // 0: SignalRequest.Signal + (*ExecRequest)(nil), // 1: ExecRequest + (*ExecResponse)(nil), // 2: ExecResponse + (*TerminalSize)(nil), // 3: TerminalSize + (*IOChunk)(nil), // 4: IOChunk + (*ResolveIPRequest)(nil), // 5: ResolveIPRequest + (*ResolveIPResponse)(nil), // 6: ResolveIPResponse + (*SignalRequest)(nil), // 7: SignalRequest + (*ExecRequest_Command)(nil), // 8: ExecRequest.Command + nil, // 9: ExecRequest.Command.EnvEntry + (*ExecResponse_Exit)(nil), // 10: ExecResponse.Exit + (*ExecResponse_Started)(nil), // 11: ExecResponse.Started + (*emptypb.Empty)(nil), // 12: google.protobuf.Empty } var file_rpc_agent_proto_depIdxs = []int32{ 8, // 0: ExecRequest.command:type_name -> ExecRequest.Command 4, // 1: ExecRequest.standard_input:type_name -> IOChunk 3, // 2: ExecRequest.terminal_resize:type_name -> TerminalSize - 7, // 3: ExecRequest.send_signal:type_name -> ExecRequest.SendSignal - 10, // 4: ExecResponse.exit:type_name -> ExecResponse.Exit - 4, // 5: ExecResponse.standard_output:type_name -> IOChunk - 4, // 6: ExecResponse.standard_error:type_name -> IOChunk - 11, // 7: ExecResponse.started:type_name -> ExecResponse.Started - 0, // 8: ExecRequest.SendSignal.signal:type_name -> ExecRequest.SendSignal.Signal - 3, // 9: ExecRequest.Command.terminal_size:type_name -> TerminalSize - 9, // 10: ExecRequest.Command.env:type_name -> ExecRequest.Command.EnvEntry - 1, // 11: Agent.Exec:input_type -> ExecRequest + 10, // 3: ExecResponse.exit:type_name -> ExecResponse.Exit + 4, // 4: ExecResponse.standard_output:type_name -> IOChunk + 4, // 5: ExecResponse.standard_error:type_name -> IOChunk + 11, // 6: ExecResponse.started:type_name -> ExecResponse.Started + 0, // 7: SignalRequest.signal:type_name -> SignalRequest.Signal + 3, // 8: ExecRequest.Command.terminal_size:type_name -> TerminalSize + 9, // 9: ExecRequest.Command.env:type_name -> ExecRequest.Command.EnvEntry + 1, // 10: Agent.Exec:input_type -> ExecRequest + 7, // 11: Agent.Signal:input_type -> SignalRequest 5, // 12: Agent.ResolveIP:input_type -> ResolveIPRequest 2, // 13: Agent.Exec:output_type -> ExecResponse - 6, // 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 + 12, // 14: Agent.Signal:output_type -> google.protobuf.Empty + 6, // 15: Agent.ResolveIP:output_type -> ResolveIPResponse + 13, // [13:16] is the sub-list for method output_type + 10, // [10:13] is the sub-list for method input_type + 10, // [10:10] is the sub-list for extension type_name + 10, // [10:10] is the sub-list for extension extendee + 0, // [0:10] is the sub-list for field type_name } func init() { file_rpc_agent_proto_init() } @@ -811,7 +822,6 @@ func file_rpc_agent_proto_init() { (*ExecRequest_Command_)(nil), (*ExecRequest_StandardInput)(nil), (*ExecRequest_TerminalResize)(nil), - (*ExecRequest_SendSignal_)(nil), } file_rpc_agent_proto_msgTypes[1].OneofWrappers = []any{ (*ExecResponse_Exit_)(nil), diff --git a/internal/rpc/agent_grpc.pb.go b/internal/rpc/agent_grpc.pb.go index 4106798..ef82d8a 100644 --- a/internal/rpc/agent_grpc.pb.go +++ b/internal/rpc/agent_grpc.pb.go @@ -1,7 +1,7 @@ // Code generated by protoc-gen-go-grpc. DO NOT EDIT. // versions: // - protoc-gen-go-grpc v1.5.1 -// - protoc (unknown) +// - protoc v7.35.1 // source: rpc/agent.proto package rpc @@ -11,6 +11,7 @@ import ( grpc "google.golang.org/grpc" codes "google.golang.org/grpc/codes" status "google.golang.org/grpc/status" + emptypb "google.golang.org/protobuf/types/known/emptypb" ) // This is a compile-time assertion to ensure that this generated file @@ -20,6 +21,7 @@ const _ = grpc.SupportPackageIsVersion9 const ( Agent_Exec_FullMethodName = "/Agent/Exec" + Agent_Signal_FullMethodName = "/Agent/Signal" Agent_ResolveIP_FullMethodName = "/Agent/ResolveIP" ) @@ -28,6 +30,7 @@ const ( // For semantics around ctx use and closing/ending streaming RPCs, please refer to https://pkg.go.dev/google.golang.org/grpc/?tab=doc#ClientConn.NewStream. type AgentClient interface { Exec(ctx context.Context, opts ...grpc.CallOption) (grpc.BidiStreamingClient[ExecRequest, ExecResponse], error) + Signal(ctx context.Context, in *SignalRequest, opts ...grpc.CallOption) (*emptypb.Empty, error) ResolveIP(ctx context.Context, in *ResolveIPRequest, opts ...grpc.CallOption) (*ResolveIPResponse, error) } @@ -52,6 +55,16 @@ func (c *agentClient) Exec(ctx context.Context, opts ...grpc.CallOption) (grpc.B // This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name. type Agent_ExecClient = grpc.BidiStreamingClient[ExecRequest, ExecResponse] +func (c *agentClient) Signal(ctx context.Context, in *SignalRequest, opts ...grpc.CallOption) (*emptypb.Empty, error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) + out := new(emptypb.Empty) + err := c.cc.Invoke(ctx, Agent_Signal_FullMethodName, in, out, cOpts...) + if err != nil { + return nil, err + } + return out, nil +} + func (c *agentClient) ResolveIP(ctx context.Context, in *ResolveIPRequest, opts ...grpc.CallOption) (*ResolveIPResponse, error) { cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) out := new(ResolveIPResponse) @@ -67,6 +80,7 @@ func (c *agentClient) ResolveIP(ctx context.Context, in *ResolveIPRequest, opts // for forward compatibility. type AgentServer interface { Exec(grpc.BidiStreamingServer[ExecRequest, ExecResponse]) error + Signal(context.Context, *SignalRequest) (*emptypb.Empty, error) ResolveIP(context.Context, *ResolveIPRequest) (*ResolveIPResponse, error) mustEmbedUnimplementedAgentServer() } @@ -81,6 +95,9 @@ type UnimplementedAgentServer struct{} func (UnimplementedAgentServer) Exec(grpc.BidiStreamingServer[ExecRequest, ExecResponse]) error { return status.Errorf(codes.Unimplemented, "method Exec not implemented") } +func (UnimplementedAgentServer) Signal(context.Context, *SignalRequest) (*emptypb.Empty, error) { + return nil, status.Errorf(codes.Unimplemented, "method Signal not implemented") +} func (UnimplementedAgentServer) ResolveIP(context.Context, *ResolveIPRequest) (*ResolveIPResponse, error) { return nil, status.Errorf(codes.Unimplemented, "method ResolveIP not implemented") } @@ -112,6 +129,24 @@ func _Agent_Exec_Handler(srv interface{}, stream grpc.ServerStream) error { // This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name. type Agent_ExecServer = grpc.BidiStreamingServer[ExecRequest, ExecResponse] +func _Agent_Signal_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(SignalRequest) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(AgentServer).Signal(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: Agent_Signal_FullMethodName, + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(AgentServer).Signal(ctx, req.(*SignalRequest)) + } + return interceptor(ctx, in, info, handler) +} + func _Agent_ResolveIP_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { in := new(ResolveIPRequest) if err := dec(in); err != nil { @@ -137,6 +172,10 @@ var Agent_ServiceDesc = grpc.ServiceDesc{ ServiceName: "Agent", HandlerType: (*AgentServer)(nil), Methods: []grpc.MethodDesc{ + { + MethodName: "Signal", + Handler: _Agent_Signal_Handler, + }, { MethodName: "ResolveIP", Handler: _Agent_ResolveIP_Handler, diff --git a/internal/rpc/exec.go b/internal/rpc/exec.go index 2187d46..2ff5483 100644 --- a/internal/rpc/exec.go +++ b/internal/rpc/exec.go @@ -7,16 +7,20 @@ import ( "io" "os" "os/exec" + userpkg "os/user" "slices" + "strconv" "strings" "sync" "syscall" "github.com/creack/pty" + "github.com/google/uuid" "github.com/samber/lo" "go.uber.org/zap" "golang.org/x/sync/errgroup" "google.golang.org/grpc" + "google.golang.org/protobuf/types/known/emptypb" ) const ( @@ -57,12 +61,20 @@ func (rpc *RPC) Exec(stream grpc.BidiStreamingServer[ExecRequest, ExecResponse]) cmd := exec.CommandContext(execCtx, firstExecRequestCommand.Command.Name, firstExecRequestCommand.Command.Args...) - applyExecOverrides(cmd, firstExecRequestCommand.Command) + + cmd.SysProcAttr = &syscall.SysProcAttr{} + + if err := applyExecOverrides(cmd, firstExecRequestCommand.Command); err != nil { + zap.S().Warnf("failed to configure %s: %v", formatCommandAndArgs(firstExecRequestCommand.Command.GetName(), + firstExecRequestCommand.Command.GetArgs()), err) + + return sendStartFailure(stream) + } if firstExecRequestCommand.Command.Detach { cmd.Stdout = io.Discard cmd.Stderr = io.Discard - cmd.SysProcAttr = &syscall.SysProcAttr{Setsid: true} + cmd.SysProcAttr.Setsid = true if err := cmd.Start(); err != nil { zap.S().Warnf("failed to start %s: %v", formatCommandAndArgs(firstExecRequestCommand.Command.GetName(), @@ -76,8 +88,9 @@ func (rpc *RPC) Exec(stream grpc.BidiStreamingServer[ExecRequest, ExecResponse]) return err } - // Explicitly notify the client that the process was started - err = sendStartSuccess(stream) + // Explicitly notify the client that the process was started, + // but don't provide an exec ID since it's a detached process + err = sendStartSuccess(stream, "") if err != nil { return err } @@ -117,7 +130,7 @@ func (rpc *RPC) Exec(stream grpc.BidiStreamingServer[ExecRequest, ExecResponse]) stderr = ptmx } else { // Start the command in its own process group so signals reach all descendants - cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true} + cmd.SysProcAttr.Setpgid = true if firstExecRequestCommand.Command.Interactive { stdin, err = cmd.StdinPipe() @@ -151,8 +164,12 @@ func (rpc *RPC) Exec(stream grpc.BidiStreamingServer[ExecRequest, ExecResponse]) defer ptmx.Close() } + execID := uuid.NewString() + rpc.execs.Store(execID, cmd.Process) + defer rpc.execs.Delete(execID) + // Explicitly notify the client that the process was started - err = sendStartSuccess(stream) + err = sendStartSuccess(stream, execID) if err != nil { // Output readers have not started yet, so cancel and reap directly _ = cmd.Cancel() @@ -169,11 +186,17 @@ func (rpc *RPC) Exec(stream grpc.BidiStreamingServer[ExecRequest, ExecResponse]) } go func() { + var stdinClosed bool + for { request, err := stream.Recv() if err != nil { // Allow the client to close its sending side while continuing to receive responses if errors.Is(err, io.EOF) { + if err := closeStdin(stdin, firstExecRequestCommand.Command.GetTty(), &stdinClosed); err != nil { + reportClientError(err) + } + return } @@ -192,29 +215,18 @@ func (rpc *RPC) Exec(stream grpc.BidiStreamingServer[ExecRequest, ExecResponse]) continue } - dataToWrite := typedAction.StandardInput.Data - // Check if the remote client has received EOF on their standard input if len(typedAction.StandardInput.Data) == 0 { - if firstExecRequestCommand.Command.Tty { - // When using pseudo-terminal, we can't simply close the - // standard input, as the file descriptor is shared for - // standard output and standard error too, so we send - // an EOF character instead - dataToWrite = []byte{eofChar} - } else { - // Close the standard input - if err := stdin.Close(); err != nil { - reportClientError(err) + if err := closeStdin(stdin, firstExecRequestCommand.Command.GetTty(), &stdinClosed); err != nil { + reportClientError(err) - return - } - - continue + return } + + continue } - if _, err := stdin.Write(dataToWrite); err != nil { + if _, err := stdin.Write(typedAction.StandardInput.GetData()); err != nil { reportClientError(err) return @@ -232,25 +244,6 @@ func (rpc *RPC) Exec(stream grpc.BidiStreamingServer[ExecRequest, ExecResponse]) }); err != nil { reportClientError(err) - return - } - case *ExecRequest_SendSignal_: - var signal syscall.Signal - - switch typedAction.SendSignal.GetSignal() { - case ExecRequest_SendSignal_SIGNAL_SIGTERM: - signal = syscall.SIGTERM - case ExecRequest_SendSignal_SIGNAL_SIGKILL: - signal = syscall.SIGKILL - default: - reportClientError(fmt.Errorf("unsupported exec signal %q", typedAction.SendSignal.GetSignal().String())) - - return - } - - if err := signalProcessGroup(cmd.Process, signal); err != nil && !errors.Is(err, os.ErrProcessDone) { - reportClientError(err) - return } } @@ -338,6 +331,9 @@ func (rpc *RPC) Exec(stream grpc.BidiStreamingServer[ExecRequest, ExecResponse]) // Wait for the command to finish err = cmd.Wait() + // Minimize the window in which a finished exec can still be signaled + rpc.execs.Delete(execID) + // Prefer a client error over the command exit result select { case err := <-fromClientErrCh: @@ -383,10 +379,63 @@ func signalProcessGroup(process *os.Process, signal syscall.Signal) error { return nil } -func sendStartSuccess(stream grpc.BidiStreamingServer[ExecRequest, ExecResponse]) error { +func closeStdin(stdin io.WriteCloser, tty bool, closed *bool) error { + if stdin == nil || *closed { + return nil + } + + if tty { + // When using pseudo-terminal, we can't simply close the + // standard input, as the file descriptor is shared for + // standard output and standard error too, so we send + // an EOF character instead + if _, err := stdin.Write([]byte{eofChar}); err != nil { + return err + } + } else if err := stdin.Close(); err != nil { + return err + } + + *closed = true + + return nil +} + +func (rpc *RPC) Signal(_ context.Context, request *SignalRequest) (*emptypb.Empty, error) { + process, ok := rpc.execs.Load(request.GetExecId()) + if !ok { + return nil, fmt.Errorf("exec %q is not running", request.GetExecId()) + } + + var signal syscall.Signal + + switch request.GetSignal() { + case SignalRequest_SIGNAL_SIGTERM: + signal = syscall.SIGTERM + case SignalRequest_SIGNAL_SIGKILL: + signal = syscall.SIGKILL + default: + return nil, fmt.Errorf("unsupported exec signal %q", request.GetSignal().String()) + } + + if err := signalProcessGroup(process, signal); err != nil { + // The process may exit after lookup, so treat the missing process as a no-op + if errors.Is(err, os.ErrProcessDone) { + return &emptypb.Empty{}, nil + } + + return nil, err + } + + return &emptypb.Empty{}, nil +} + +func sendStartSuccess(stream grpc.BidiStreamingServer[ExecRequest, ExecResponse], execID string) error { return stream.Send(&ExecResponse{ Type: &ExecResponse_Started_{ - Started: &ExecResponse_Started{}, + Started: &ExecResponse_Started{ + ExecId: execID, + }, }, }) } @@ -401,7 +450,7 @@ func sendStartFailure(stream grpc.BidiStreamingServer[ExecRequest, ExecResponse] }) } -func applyExecOverrides(cmd *exec.Cmd, command *ExecRequest_Command) { +func applyExecOverrides(cmd *exec.Cmd, command *ExecRequest_Command) error { if command.Workdir != "" { cmd.Dir = command.Workdir } @@ -409,6 +458,37 @@ func applyExecOverrides(cmd *exec.Cmd, command *ExecRequest_Command) { if len(command.Env) > 0 { cmd.Env = mergeEnv(command.Env) } + + if user := command.GetUser(); user != "" { + selectedUser, err := userpkg.Lookup(user) + if err != nil { + return fmt.Errorf("failed to resolve user %q: %w", user, err) + } + + uid, err := strconv.ParseUint(selectedUser.Uid, 10, 32) + if err != nil { + return fmt.Errorf("failed to parse UID %q for user %q: %w", + selectedUser.Uid, user, err) + } + + gid, err := strconv.ParseUint(selectedUser.Gid, 10, 32) + if err != nil { + return fmt.Errorf("failed to parse GID %q for user %q: %w", + selectedUser.Gid, user, err) + } + + if uint32(uid) == uint32(os.Geteuid()) && uint32(gid) == uint32(os.Getegid()) { + return nil + } + + // Avoid changing credentials when the requested user is the same as guest agen't user + cmd.SysProcAttr.Credential = &syscall.Credential{ + Uid: uint32(uid), + Gid: uint32(gid), + } + } + + return nil } func mergeEnv(overrides map[string]string) []string { diff --git a/internal/rpc/exec_test.go b/internal/rpc/exec_test.go index 9a44a6c..eba228d 100644 --- a/internal/rpc/exec_test.go +++ b/internal/rpc/exec_test.go @@ -72,7 +72,7 @@ func (stream *execTestStream) Recv() (*ExecRequest, error) { func (stream *execTestStream) Context() context.Context { return stream.ctx } func TestExecSendsStartedBeforeOutputAndExit(t *testing.T) { - stream, result := startExecTest(t, &ExecRequest_Command{ + _, stream, result := startExecTest(t, &ExecRequest_Command{ Name: execTestShell, Args: []string{"-c", "printf hello"}, }) @@ -97,6 +97,52 @@ func TestExecSendsStartedBeforeOutputAndExit(t *testing.T) { } } +func TestExecClosesStandardInputOnRequestStreamEOF(t *testing.T) { + _, stream, result := startExecTest(t, &ExecRequest_Command{ + Name: "/bin/cat", + Interactive: true, + }) + require.NotNil(t, receiveExecResponse(t, stream).GetStarted()) + + stream.requests <- &ExecRequest{ + Type: &ExecRequest_StandardInput{ + StandardInput: &IOChunk{Data: []byte("hello")}, + }, + } + close(stream.requests) + + response := receiveExecResponse(t, stream) + require.Equal(t, []byte("hello"), response.GetStandardOutput().GetData()) + response = receiveExecResponse(t, stream) + require.EqualValues(t, 0, response.GetExit().GetCode()) + require.NoError(t, receiveExecResult(t, result)) +} + +func TestExecClosesStandardInputOnEmptyChunk(t *testing.T) { + _, stream, result := startExecTest(t, &ExecRequest_Command{ + Name: "/bin/cat", + Interactive: true, + }) + require.NotNil(t, receiveExecResponse(t, stream).GetStarted()) + + stream.requests <- &ExecRequest{ + Type: &ExecRequest_StandardInput{ + StandardInput: &IOChunk{Data: []byte("hello")}, + }, + } + stream.requests <- &ExecRequest{ + Type: &ExecRequest_StandardInput{ + StandardInput: &IOChunk{}, + }, + } + + response := receiveExecResponse(t, stream) + require.Equal(t, []byte("hello"), response.GetStandardOutput().GetData()) + response = receiveExecResponse(t, stream) + require.EqualValues(t, 0, response.GetExit().GetCode()) + require.NoError(t, receiveExecResult(t, result)) +} + func TestExecReportsStartFailureBeforeStarted(t *testing.T) { tests := []struct { name string @@ -119,7 +165,7 @@ func TestExecReportsStartFailureBeforeStarted(t *testing.T) { for _, test := range tests { t.Run(test.name, func(t *testing.T) { - stream, result := startExecTest(t, test.command) + _, stream, result := startExecTest(t, test.command) response := receiveExecResponse(t, stream) require.Nil(t, response.GetStarted()) require.EqualValues(t, execRuntimeFailureExitCode, response.GetExit().GetCode()) @@ -131,45 +177,54 @@ func TestExecReportsStartFailureBeforeStarted(t *testing.T) { func TestExecSignalsProcess(t *testing.T) { tests := []struct { name string - signal ExecRequest_SendSignal_Signal + signal SignalRequest_Signal code int32 err string }{ { name: "SIGTERM", - signal: ExecRequest_SendSignal_SIGNAL_SIGTERM, + signal: SignalRequest_SIGNAL_SIGTERM, code: int32(signalExitCodeOffset + syscall.SIGTERM), }, { name: "SIGKILL", - signal: ExecRequest_SendSignal_SIGNAL_SIGKILL, + signal: SignalRequest_SIGNAL_SIGKILL, code: int32(signalExitCodeOffset + syscall.SIGKILL), }, { name: "unsupported", - signal: ExecRequest_SendSignal_SIGNAL_UNSPECIFIED, + signal: SignalRequest_SIGNAL_UNSPECIFIED, err: `unsupported exec signal "SIGNAL_UNSPECIFIED"`, }, } for _, test := range tests { t.Run(test.name, func(t *testing.T) { - stream, result := startExecTest(t, &ExecRequest_Command{ + rpc, stream, result := startExecTest(t, &ExecRequest_Command{ Name: "/bin/sleep", Args: []string{"30"}, }) - require.NotNil(t, receiveExecResponse(t, stream).GetStarted()) + started := receiveExecResponse(t, stream).GetStarted() + require.NotNil(t, started) - stream.requests <- &ExecRequest{ - Type: &ExecRequest_SendSignal_{ - SendSignal: &ExecRequest_SendSignal{Signal: test.signal}, - }, - } + _, err := rpc.Signal(context.Background(), &SignalRequest{ + ExecId: started.GetExecId(), + Signal: test.signal, + }) if test.err != "" { - require.EqualError(t, receiveExecResult(t, result), test.err) + require.EqualError(t, err, test.err) + _, err = rpc.Signal(context.Background(), &SignalRequest{ + ExecId: started.GetExecId(), + Signal: SignalRequest_SIGNAL_SIGKILL, + }) + require.NoError(t, err) + receiveExecResponse(t, stream) + require.NoError(t, receiveExecResult(t, result)) + return } + require.NoError(t, err) response := receiveExecResponse(t, stream) require.NotNil(t, response.GetExit()) @@ -180,18 +235,19 @@ func TestExecSignalsProcess(t *testing.T) { } func TestExecSignalsProcessGroup(t *testing.T) { - stream, result := startExecTest(t, &ExecRequest_Command{ + rpc, stream, result := startExecTest(t, &ExecRequest_Command{ Name: execTestShell, Args: []string{"-c", "sleep 30 & printf ready; wait"}, }) - require.NotNil(t, receiveExecResponse(t, stream).GetStarted()) + started := receiveExecResponse(t, stream).GetStarted() + require.NotNil(t, started) require.Equal(t, []byte("ready"), receiveExecResponse(t, stream).GetStandardOutput().GetData()) - stream.requests <- &ExecRequest{ - Type: &ExecRequest_SendSignal_{ - SendSignal: &ExecRequest_SendSignal{Signal: ExecRequest_SendSignal_SIGNAL_SIGTERM}, - }, - } + _, err := rpc.Signal(context.Background(), &SignalRequest{ + ExecId: started.GetExecId(), + Signal: SignalRequest_SIGNAL_SIGTERM, + }) + require.NoError(t, err) response := receiveExecResponse(t, stream) require.EqualValues(t, signalExitCodeOffset+syscall.SIGTERM, response.GetExit().GetCode()) @@ -203,7 +259,7 @@ func TestExecReapsProcessWhenStartedCannotBeSent(t *testing.T) { sendErr := errors.New("failed to send Started") var processPID int - _, result := startExecTest(t, &ExecRequest_Command{ + _, _, result := startExecTest(t, &ExecRequest_Command{ Name: execTestShell, Args: []string{"-c", `printf %d "$$" > "$PID_FILE"; exec sleep 30`}, Env: map[string]string{"PID_FILE": pidPath}, @@ -231,7 +287,7 @@ func startExecTest( t *testing.T, command *ExecRequest_Command, configure ...func(*execTestStream), -) (*execTestStream, <-chan error) { +) (*RPC, *execTestStream, <-chan error) { t.Helper() ctx, cancel := context.WithCancel(context.Background()) @@ -240,14 +296,16 @@ func startExecTest( for _, configureStream := range configure { configureStream(stream) } + rpc, err := New(nil) + require.NoError(t, err) result := make(chan error, 1) go func() { - result <- (&RPC{}).Exec(stream) + result <- rpc.Exec(stream) }() stream.requests <- &ExecRequest{ Type: &ExecRequest_Command_{Command: command}, } - return stream, result + return rpc, stream, result } func receiveExecResponse(t *testing.T, stream *execTestStream) *ExecResponse { diff --git a/internal/rpc/rpc.go b/internal/rpc/rpc.go index 798b14e..4f51745 100644 --- a/internal/rpc/rpc.go +++ b/internal/rpc/rpc.go @@ -2,13 +2,17 @@ package rpc import ( "context" - "google.golang.org/grpc" "net" + "os" + + "github.com/puzpuzpuz/xsync/v4" + "google.golang.org/grpc" ) type RPC struct { grpcServer *grpc.Server listener net.Listener + execs *xsync.Map[string, *os.Process] UnimplementedAgentServer } @@ -17,6 +21,7 @@ func New(listener net.Listener) (*RPC, error) { rpc := &RPC{ grpcServer: grpc.NewServer(), listener: listener, + execs: xsync.NewMap[string, *os.Process](), } RegisterAgentServer(rpc.grpcServer, rpc) diff --git a/proto/rpc/agent.proto b/proto/rpc/agent.proto index b015c1c..09942b8 100644 --- a/proto/rpc/agent.proto +++ b/proto/rpc/agent.proto @@ -6,20 +6,11 @@ option go_package = "github.com/cirruslabs/tart-guest-agent/internal/rpc"; service Agent { rpc Exec(stream ExecRequest) returns (stream ExecResponse); + rpc Signal(SignalRequest) returns (google.protobuf.Empty); rpc ResolveIP(ResolveIPRequest) returns (ResolveIPResponse); } message ExecRequest { - message SendSignal { - enum Signal { - SIGNAL_UNSPECIFIED = 0; - SIGNAL_SIGTERM = 1; - SIGNAL_SIGKILL = 2; - } - - Signal signal = 1; - } - message Command { string name = 1; repeated string args = 2; @@ -29,13 +20,13 @@ message ExecRequest { bool detach = 6; map env = 7; string workdir = 8; + string user = 9; } oneof type { Command command = 1; IOChunk standard_input = 2; TerminalSize terminal_resize = 3; - SendSignal send_signal = 4; } } @@ -45,7 +36,7 @@ message ExecResponse { } message Started { - // nothing for now + string exec_id = 1; } oneof type { @@ -72,3 +63,14 @@ message ResolveIPRequest { message ResolveIPResponse { string ip = 1; } + +message SignalRequest { + enum Signal { + SIGNAL_UNSPECIFIED = 0; + SIGNAL_SIGTERM = 1; + SIGNAL_SIGKILL = 2; + } + + string exec_id = 1; + Signal signal = 2; +}