public_git
/
tart-guest-agent
mirror of https://github.com/cirruslabs/tart-guest-agent.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Exec as user, out-of-band signalling, and stdin closure when the request stream ends (#58) 

* Support running exec commands as a selected user

* Move exec signaling out of band

* Close standard input when the exec request stream ends
This commit is contained in:
edi-oai 2026-08-10 23:23:07 +01:00 committed by GitHub
parent d4ebcbde41
commit 721b111c0c
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
8 changed files with 374 additions and 176 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
go.mod
View File

@ -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

2
go.sum
View File

@ -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=

194
internal/rpc/agent.pb.go
View File

@ -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),

41
internal/rpc/agent_grpc.pb.go
View File

@ -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,

170
internal/rpc/exec.go
View File

@ -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 {

108
internal/rpc/exec_test.go
View File

@ -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 {

7
internal/rpc/rpc.go
View File

@ -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)

26
proto/rpc/agent.proto
View File

@ -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<string, string> 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;
}