// Command rvc is the RVBox local control-plane CLI. It never opens the server // database; all state changes and reads go through the Unix gRPC socket. package main import ( "context" "crypto/sha256" "errors" "flag" "fmt" "io" "net" "os" "path/filepath" "strings" "time" rvboxv1 "github.com/rvbox/rvbox/gen/go/rvbox/v1" "github.com/rvbox/rvbox/internal/agentproto" "github.com/rvbox/rvbox/internal/domain" "google.golang.org/grpc" "google.golang.org/grpc/credentials/insecure" "google.golang.org/protobuf/types/known/durationpb" ) const defaultControlSocket = "/run/rvbox/server.sock" func main() { if err := run(os.Args[1:], os.Stdout, os.Stderr); err != nil { fmt.Fprintln(os.Stderr, "rvc:", err) os.Exit(1) } } func run(args []string, output, diagnostics io.Writer) error { if len(args) == 0 { return errors.New("a command is required (stat or run)") } socket, args, err := globalSocket(args) if err != nil { return err } requestID, args, err := globalRequestID(args) if err != nil { return err } if len(args) == 0 { return errors.New("a command is required (stat or run)") } if requestID != "" && !readOnlyCommand(args) { args = append([]string{args[0], "--request-id", requestID}, args[1:]...) } ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) defer cancel() connection, err := dial(ctx, socket) if err != nil { return err } defer connection.Close() client := rvboxv1.NewControlClient(connection) switch args[0] { case "stat": return stat(ctx, client, args[1:], output) case "run": return runCommand(ctx, client, args[1:], output, diagnostics) case "append": return appendStdin(ctx, client, args[1:], output) case "close-stdin": return closeStdin(ctx, client, args[1:], output) case "kill": return killCommand(ctx, client, args[1:], output) case "storage": return storage(ctx, client, args[1:], output) case "client": return clientCommand(ctx, client, args[1:], output) default: return fmt.Errorf("unknown command %q", args[0]) } } func globalSocket(args []string) (string, []string, error) { socket := defaultControlSocket remaining := make([]string, 0, len(args)) for index := 0; index < len(args); index++ { if args[index] == "--socket" { if index+1 >= len(args) { return "", nil, errors.New("--socket requires a path") } socket = args[index+1] index++ continue } if strings.HasPrefix(args[index], "--socket=") { socket = strings.TrimPrefix(args[index], "--socket=") continue } remaining = append(remaining, args[index]) } if socket == "" || !filepath.IsAbs(socket) { return "", nil, errors.New("--socket must be an absolute path") } return socket, remaining, nil } // globalRequestID removes the optional request ID wherever the CLI's global // option parser finds it. It is injected before the subcommand's positional // arguments so Go's flag package cannot mistake it for command text. Read-only // commands intentionally discard the value in run(). func globalRequestID(args []string) (string, []string, error) { var requestID string remaining := make([]string, 0, len(args)) for index := 0; index < len(args); index++ { if args[index] == "--request-id" { if index+1 >= len(args) { return "", nil, errors.New("--request-id requires a UUIDv7") } if requestID != "" { return "", nil, errors.New("--request-id may be supplied only once") } requestID = args[index+1] index++ continue } if strings.HasPrefix(args[index], "--request-id=") { if requestID != "" { return "", nil, errors.New("--request-id may be supplied only once") } requestID = strings.TrimPrefix(args[index], "--request-id=") continue } remaining = append(remaining, args[index]) } return requestID, remaining, nil } func readOnlyCommand(args []string) bool { if len(args) == 0 || args[0] == "stat" { return true } return args[0] == "storage" && len(args) > 1 && args[1] == "incidents" } func dial(ctx context.Context, socket string) (*grpc.ClientConn, error) { dialer := func(ctx context.Context, _ string) (net.Conn, error) { return (&net.Dialer{}).DialContext(ctx, "unix", socket) } return grpc.DialContext(ctx, "passthrough:///rvbox-control", grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithContextDialer(dialer), grpc.WithBlock()) } func stat(ctx context.Context, client rvboxv1.ControlClient, args []string, output io.Writer) error { flags := flag.NewFlagSet("stat", flag.ContinueOnError) flags.SetOutput(io.Discard) all := flags.Bool("all", false, "show all command pages") pageSize := flags.Uint("per-page", 100, "number of records per page") if err := flags.Parse(args); err != nil { return err } positionals := flags.Args() if len(positionals) == 0 { var token string for { response, err := client.ListClients(ctx, &rvboxv1.ListClientsRequest{PageSize: uint32(*pageSize), PageToken: token}) if err != nil { return err } for _, item := range response.GetClients() { fmt.Fprintf(output, "client %s connected=%t platform=%s running=%d queued=%d\n", item.GetClientId(), item.GetConnected(), item.GetPlatform(), item.GetRunningCommands(), item.GetQueuedCommands()) } if !*all || response.GetNextPageToken() == "" { return nil } token = response.GetNextPageToken() } } if len(positionals) > 2 { return errors.New("stat accepts at most CLIENT and ISSUE_UUID") } clientID := positionals[0] if len(positionals) == 1 { response, err := client.GetClient(ctx, &rvboxv1.GetClientRequest{ClientId: clientID}) if err != nil { return err } item := response.GetClient() fmt.Fprintf(output, "client %s connected=%t platform=%s running=%d queued=%d instance=%s\n", item.GetClientId(), item.GetConnected(), item.GetPlatform(), item.GetRunningCommands(), item.GetQueuedCommands(), item.GetClientInstanceId()) if item.GetPendingInstanceId() != "" { fmt.Fprintf(output, "pending_instance=%s\n", item.GetPendingInstanceId()) } return nil } response, err := client.GetCommand(ctx, &rvboxv1.GetCommandRequest{ClientId: clientID, IssueUuid: positionals[1]}) if err != nil { return err } item := response.GetCommand() renderCommandStat(output, item, time.Now().UTC()) return nil } // renderCommandStat keeps the human CLI useful when a caller does not have a // protobuf-aware inspection tool. Durable lifecycle, expiry, retention, and // Windows identity are rendered independently: a late terminal result must // not erase the fact that the queue deadline was crossed. func renderCommandStat(output io.Writer, item *rvboxv1.CommandRecord, now time.Time) { if item == nil { return } fmt.Fprintf(output, "command %s client=%s lifecycle=%s revision=%d events=%d\n", item.GetIssueUuid(), item.GetTargetClientId(), item.GetLifecycle(), item.GetCommandRevision(), item.GetLastEventSeq()) if item.GetQueueExpiryTime() != nil { expiry := item.GetQueueExpiryTime().AsTime().UTC() fmt.Fprintf(output, "queue_expiry=%s\n", expiry.Format(time.RFC3339Nano)) if !now.IsZero() && !now.Before(expiry) { fmt.Fprintln(output, "queue_expired=true") } } if item.GetLateAfterExpiry() { fmt.Fprintln(output, "late_after_expiry=true (terminal result arrived after queue expiry)") } if item.GetTerminalTime() != nil { fmt.Fprintf(output, "terminal_time=%s\n", item.GetTerminalTime().AsTime().UTC().Format(time.RFC3339Nano)) } if item.GetExitCode() != 0 || item.ExitCode != nil { fmt.Fprintf(output, "exit_code=%d\n", item.GetExitCode()) } if item.GetOutputTruncated() { fmt.Fprintln(output, "output_truncated=true") } if item.GetOutputIncomplete() { fmt.Fprintln(output, "output_incomplete=true") } if item.GetRetainedCompressedBytes() > 0 { fmt.Fprintf(output, "retained_compressed_bytes=%d\n", item.GetRetainedCompressedBytes()) } if rejection := item.GetRejection(); rejection != nil { fmt.Fprintf(output, "rejection_code=%s rejection=%s\n", rejection.GetCode(), singleLine(rejection.GetMessage())) } if identity := item.GetWindowsExecutionIdentity(); identity != nil { renderWindowsIdentity(output, item.GetSpec().GetElevated(), identity) } } func renderWindowsIdentity(output io.Writer, requestedElevated bool, identity *rvboxv1.WindowsExecutionIdentity) { fmt.Fprintf(output, "windows_requested_elevated=%t\n", requestedElevated) if contexts := identity.GetAttemptedContexts(); len(contexts) > 0 { names := make([]string, 0, len(contexts)) for _, context := range contexts { names = append(names, windowsContextName(context)) } fmt.Fprintf(output, "windows_attempted_contexts=%s\n", strings.Join(names, ",")) } if identity.EffectiveContext != nil { fmt.Fprintf(output, "windows_effective_context=%s\n", windowsContextName(identity.GetEffectiveContext())) } else { fmt.Fprintln(output, "windows_effective_context=none") } if identity.GetEffectiveUserSid() != "" { fmt.Fprintf(output, "windows_effective_user_sid=%s\n", singleLine(identity.GetEffectiveUserSid())) } if identity.SessionId != nil { fmt.Fprintf(output, "windows_session_id=%d\n", identity.GetSessionId()) } if identity.GetSessionUserSid() != "" { fmt.Fprintf(output, "windows_session_user_sid=%s\n", singleLine(identity.GetSessionUserSid())) } if identity.GetSelectionDetail() != "" { fmt.Fprintf(output, "windows_selection_detail=%s\n", singleLine(identity.GetSelectionDetail())) } } func windowsContextName(value rvboxv1.WindowsExecutionContext) string { switch value { case rvboxv1.WindowsExecutionContext_WINDOWS_EXECUTION_CONTEXT_LOCAL_SERVICE: return "local-service" case rvboxv1.WindowsExecutionContext_WINDOWS_EXECUTION_CONTEXT_LOCAL_SYSTEM: return "local-system" case rvboxv1.WindowsExecutionContext_WINDOWS_EXECUTION_CONTEXT_ACTIVE_USER: return "active-user" case rvboxv1.WindowsExecutionContext_WINDOWS_EXECUTION_CONTEXT_ACTIVE_USER_ELEVATED: return "active-user-elevated" case rvboxv1.WindowsExecutionContext_WINDOWS_EXECUTION_CONTEXT_ACTIVE_SYSTEM: return "active-system" default: return "unspecified" } } func singleLine(value string) string { return strings.NewReplacer("\r", "\\r", "\n", "\\n", "\t", "\\t").Replace(value) } func runCommand(ctx context.Context, client rvboxv1.ControlClient, args []string, output, diagnostics io.Writer) error { flags := flag.NewFlagSet("run", flag.ContinueOnError) flags.SetOutput(io.Discard) background := flags.Bool("background", false, "return after durable admission") cwd := flags.String("cwd", "", "command working directory") shell := flags.String("shell", "", "shell (sh, bash, cmd, powershell)") requestID := flags.String("request-id", "", "canonical UUIDv7 used for idempotent admission") elevated := flags.Bool("elevated", false, "request the Windows elevated execution policy") queueTTL := flags.Duration("queue-ttl", -1, "queue TTL; zero means no expiry") scriptPath := flags.String("script", "", "script file path") envValues := repeatedFlag{} profileValues := repeatedFlag{} flags.Var(&envValues, "env", "environment override KEY=VALUE") flags.Var(&profileValues, "profile", "execution profile") if err := flags.Parse(args); err != nil { return err } positionals := flags.Args() if len(positionals) == 0 { return errors.New("run requires CLIENT and COMMAND (or --script PATH CLIENT)") } if *requestID == "" { generated, err := domain.NewUUIDv7() if err != nil { return err } *requestID = generated.String() } if _, err := domain.ParseUUIDv7(*requestID); err != nil { return fmt.Errorf("--request-id: %w", err) } spec := &rvboxv1.ExecutionSpec{Cwd: *cwd, Elevated: *elevated} spec.ShellType, _ = parseShell(*shell) for _, value := range envValues { parts := strings.SplitN(value, "=", 2) if len(parts) != 2 || parts[0] == "" { return fmt.Errorf("--env must be KEY=VALUE: %q", value) } if spec.EnvOverrides == nil { spec.EnvOverrides = make(map[string]string) } spec.EnvOverrides[parts[0]] = parts[1] } for _, value := range profileValues { profile, err := parseProfile(value) if err != nil { return err } spec.ExecutionProfiles = append(spec.ExecutionProfiles, profile) } var clientID string if *scriptPath != "" { if len(positionals) != 1 { return errors.New("script form accepts exactly CLIENT") } clientID = positionals[0] data, err := os.ReadFile(*scriptPath) if err != nil { return err } if len(data) > 10<<20 { return errors.New("script exceeds the 10 MiB limit") } digest := sha256.Sum256(data) spec.Source = &rvboxv1.ExecutionSpec_Script{Script: &rvboxv1.ScriptDescriptor{Filename: filepath.Base(*scriptPath), SizeBytes: uint64(len(data)), Sha256: digest[:]}} request := &rvboxv1.RunCommandRequest{TargetClientId: clientID, Spec: spec, ScriptContent: data, RequestId: *requestID, QueueTtl: queueDuration(*queueTTL)} return printRunResponse(ctx, client, request, output, *background) } if len(positionals) < 2 { return errors.New("run requires CLIENT and COMMAND") } clientID = positionals[0] spec.Source = &rvboxv1.ExecutionSpec_CommandText{CommandText: strings.Join(positionals[1:], " ")} request := &rvboxv1.RunCommandRequest{TargetClientId: clientID, Spec: spec, RequestId: *requestID, QueueTtl: queueDuration(*queueTTL)} _ = diagnostics return printRunResponse(ctx, client, request, output, *background) } func printRunResponse(ctx context.Context, client rvboxv1.ControlClient, request *rvboxv1.RunCommandRequest, output io.Writer, background bool) error { response, err := client.RunCommand(ctx, request) if err != nil { return err } fmt.Fprintf(output, "%s %s\n", response.GetIssueUuid(), response.GetLifecycle()) if background { return nil } follow, err := client.FollowCommand(ctx, &rvboxv1.FollowCommandRequest{ClientId: request.GetTargetClientId(), IssueUuid: response.GetIssueUuid(), IncludeExisting: true}) if err != nil { return err } for { item, err := follow.Recv() if errors.Is(err, io.EOF) { return nil } if err != nil { return err } event := item.GetEvent() if event == nil { continue } switch payload := event.Payload.(type) { case *rvboxv1.CommandEvent_Output: data, decodeErr := agentproto.DecodeOutputChunk(payload.Output, 64<<10) if decodeErr != nil { return decodeErr } if _, writeErr := output.Write(data); writeErr != nil { return writeErr } case *rvboxv1.CommandEvent_Lifecycle: fmt.Fprintf(output, "\n[%s] %s\n", payload.Lifecycle.GetLifecycle(), payload.Lifecycle.GetDetail()) } } } func appendStdin(ctx context.Context, client rvboxv1.ControlClient, args []string, output io.Writer) error { flags := flag.NewFlagSet("append", flag.ContinueOnError) flags.SetOutput(io.Discard) filePath := flags.String("file", "", "read stdin data from a file") raw := flags.Bool("raw", false, "do not append a newline") requestID := flags.String("request-id", "", "canonical UUIDv7") if err := flags.Parse(args); err != nil { return err } positionals := flags.Args() if len(positionals) < 2 || len(positionals) > 3 || (*filePath != "" && len(positionals) == 3) { return errors.New("append requires CLIENT UUID TEXT, or --file PATH CLIENT UUID") } data := []byte{} var err error if *filePath != "" { data, err = os.ReadFile(*filePath) } else { data = []byte(positionals[2]) } if err != nil { return err } if *requestID == "" { *requestID, err = generatedRequestID() if err != nil { return err } } issue := positionals[1] if _, err := domain.ParseUUIDv7(issue); err != nil { return err } response, err := client.AppendStdin(ctx, &rvboxv1.AppendStdinRequest{ClientId: positionals[0], IssueUuid: issue, Data: data, AppendNewline: !*raw, RequestId: *requestID}) if err != nil { return err } fmt.Fprintf(output, "%d\n", response.GetWriteSeq()) return nil } func closeStdin(ctx context.Context, client rvboxv1.ControlClient, args []string, output io.Writer) error { flags := flag.NewFlagSet("close-stdin", flag.ContinueOnError) flags.SetOutput(io.Discard) requestID := flags.String("request-id", "", "canonical UUIDv7") if err := flags.Parse(args); err != nil { return err } if len(flags.Args()) != 2 { return errors.New("close-stdin requires CLIENT UUID") } if _, err := domain.ParseUUIDv7(flags.Args()[1]); err != nil { return err } if *requestID == "" { var err error *requestID, err = generatedRequestID() if err != nil { return err } } response, err := client.CloseStdin(ctx, &rvboxv1.CloseStdinRequest{ClientId: flags.Args()[0], IssueUuid: flags.Args()[1], RequestId: *requestID}) if err != nil { return err } fmt.Fprintf(output, "%d\n", response.GetWriteSeq()) return nil } func killCommand(ctx context.Context, client rvboxv1.ControlClient, args []string, output io.Writer) error { flags := flag.NewFlagSet("kill", flag.ContinueOnError) flags.SetOutput(io.Discard) requestID := flags.String("request-id", "", "canonical UUIDv7") if err := flags.Parse(args); err != nil { return err } positionals := flags.Args() signal := rvboxv1.SignalKind_SIGNAL_TERM if len(positionals) == 3 { parsed, err := parseSignal(positionals[0]) if err != nil { return err } signal = parsed positionals = positionals[1:] } if len(positionals) != 2 { return errors.New("kill requires [SIGNAL] CLIENT UUID") } if _, err := domain.ParseUUIDv7(positionals[1]); err != nil { return err } if *requestID == "" { var err error *requestID, err = generatedRequestID() if err != nil { return err } } response, err := client.SignalCommand(ctx, &rvboxv1.ControlSignalCommandRequest{ClientId: positionals[0], IssueUuid: positionals[1], Signal: signal, RequestId: *requestID}) if err != nil { return err } fmt.Fprintf(output, "%d\n", response.GetCommandRevision()) return nil } func storage(ctx context.Context, client rvboxv1.ControlClient, args []string, output io.Writer) error { flags := flag.NewFlagSet("storage", flag.ContinueOnError) flags.SetOutput(io.Discard) requestID := flags.String("request-id", "", "canonical UUIDv7") includeResolved := flags.Bool("all", false, "include resolved incidents") if err := flags.Parse(args); err != nil { return err } args = flags.Args() if len(args) == 0 { return errors.New("storage requires incidents, repair, or acknowledge") } switch args[0] { case "incidents": response, err := client.ListStorageIncidents(ctx, &rvboxv1.ListStorageIncidentsRequest{IncludeResolved: *includeResolved}) if err != nil { return err } for _, incident := range response.GetIncidents() { fmt.Fprintf(output, "%s state=%s scope=%s summary=%s\n", incident.GetIncidentId(), incident.GetState(), incident.GetScope(), incident.GetSummary()) } return nil case "repair", "acknowledge": if len(args) < 2 { return errors.New("storage mutation requires INCIDENT_ID") } if *requestID == "" { var err error *requestID, err = generatedRequestID() if err != nil { return err } } if args[0] == "repair" { response, callErr := client.RepairStorageIncident(ctx, &rvboxv1.RepairStorageIncidentRequest{IncidentId: args[1], RequestId: *requestID}) if callErr != nil { return callErr } fmt.Fprintf(output, "%s\n", response.GetIncident().GetState()) return nil } note := strings.Join(args[2:], " ") if note == "" { return errors.New("storage acknowledge requires a note") } response, callErr := client.AcknowledgeStorageIncident(ctx, &rvboxv1.AcknowledgeStorageIncidentRequest{IncidentId: args[1], RequestId: *requestID, Note: note}) if callErr != nil { return callErr } fmt.Fprintf(output, "%s\n", response.GetIncident().GetState()) return nil default: return fmt.Errorf("unknown storage command %q", args[0]) } } func clientCommand(ctx context.Context, client rvboxv1.ControlClient, args []string, output io.Writer) error { flags := flag.NewFlagSet("client", flag.ContinueOnError) flags.SetOutput(io.Discard) requestID := flags.String("request-id", "", "canonical UUIDv7") if err := flags.Parse(args); err != nil { return err } args = flags.Args() if len(args) != 3 || args[0] != "takeover" { return errors.New("client takeover requires CLIENT INSTANCE_ID") } if _, err := domain.ParseUUIDv7(args[2]); err != nil { return err } if *requestID == "" { var err error *requestID, err = generatedRequestID() if err != nil { return err } } response, err := client.AuthorizeClientTakeover(ctx, &rvboxv1.AuthorizeClientTakeoverRequest{ClientId: args[1], ClientInstanceId: args[2], RequestId: *requestID}) if err != nil { return err } fmt.Fprintln(output, response.GetExpiresAt().AsTime().UTC().Format(time.RFC3339Nano)) return nil } func generatedRequestID() (string, error) { value, err := domain.NewUUIDv7() if err != nil { return "", err } return value.String(), nil } func parseSignal(value string) (rvboxv1.SignalKind, error) { value = strings.ToUpper(strings.TrimPrefix(value, "SIG")) switch value { case "HUP": return rvboxv1.SignalKind_SIGNAL_HUP, nil case "INT": return rvboxv1.SignalKind_SIGNAL_INT, nil case "TERM": return rvboxv1.SignalKind_SIGNAL_TERM, nil case "KILL": return rvboxv1.SignalKind_SIGNAL_KILL, nil case "USR1": return rvboxv1.SignalKind_SIGNAL_USR1, nil case "USR2": return rvboxv1.SignalKind_SIGNAL_USR2, nil default: return rvboxv1.SignalKind_SIGNAL_KIND_UNSPECIFIED, fmt.Errorf("unsupported signal %q", value) } } func contains(values []string, target string) bool { for _, value := range values { if value == target { return true } } return false } func queueDuration(value time.Duration) *durationpb.Duration { if value < 0 { return nil } return durationpb.New(value) } func parseShell(value string) (rvboxv1.ShellType, error) { switch strings.ToLower(value) { case "", "default": return rvboxv1.ShellType_SHELL_TYPE_UNSPECIFIED, nil case "sh": return rvboxv1.ShellType_SHELL_SH, nil case "bash": return rvboxv1.ShellType_SHELL_BASH, nil case "cmd": return rvboxv1.ShellType_SHELL_CMD, nil case "powershell", "pwsh": return rvboxv1.ShellType_SHELL_POWERSHELL, nil default: return rvboxv1.ShellType_SHELL_TYPE_UNSPECIFIED, fmt.Errorf("unknown shell %q", value) } } func parseProfile(value string) (rvboxv1.ExecutionProfile, error) { switch strings.ToLower(value) { case "light": return rvboxv1.ExecutionProfile_EXECUTION_PROFILE_LIGHT, nil case "cpu-medium": return rvboxv1.ExecutionProfile_EXECUTION_PROFILE_CPU_MEDIUM, nil case "cpu-heavy": return rvboxv1.ExecutionProfile_EXECUTION_PROFILE_CPU_HEAVY, nil case "mem-medium": return rvboxv1.ExecutionProfile_EXECUTION_PROFILE_MEM_MEDIUM, nil case "mem-heavy": return rvboxv1.ExecutionProfile_EXECUTION_PROFILE_MEM_HEAVY, nil case "disk-medium": return rvboxv1.ExecutionProfile_EXECUTION_PROFILE_DISK_MEDIUM, nil case "disk-heavy": return rvboxv1.ExecutionProfile_EXECUTION_PROFILE_DISK_HEAVY, nil default: return rvboxv1.ExecutionProfile_EXECUTION_PROFILE_UNSPECIFIED, fmt.Errorf("unknown profile %q", value) } } type repeatedFlag []string func (flag *repeatedFlag) String() string { return strings.Join(*flag, ",") } func (flag *repeatedFlag) Set(value string) error { *flag = append(*flag, value) return nil }