From 486894557d5b5ccffef9d8f8898d47a42fc1f647 Mon Sep 17 00:00:00 2001 From: cabbage Date: Sun, 6 Sep 2026 13:37:32 +0000 Subject: [PATCH] feat: complete Windows client control and recovery paths --- cmd/rvbox-server/main.go | 72 ++- cmd/rvbox/main.go | 47 +- cmd/rvbox/service_windows.go | 77 +++- cmd/rvc/main.go | 89 +++- cmd/rvc/main_test.go | 31 ++ docs/implementation-plan.v1.md | 2 +- docs/platform-and-operations.md | 3 +- docs/testing.md | 7 +- internal/client/agent/executor.go | 20 +- internal/client/agent/runner.go | 34 +- internal/client/spool/events.go | 25 +- internal/client/spool/spool_test.go | 43 +- internal/client/supervisor/supervisor.go | 78 +++- internal/client/supervisor/supervisor_test.go | 22 + .../client/supervisor/windows/native_exec.go | 46 +- .../supervisor/windows/native_windows.go | 375 +++++++++++++++- internal/client/windowsservice/service.go | 2 + .../client/windowsservice/service_other.go | 2 + .../client/windowsservice/service_windows.go | 82 +++- internal/client/windowstray/protocol.go | 17 +- internal/client/windowstray/service_other.go | 20 + .../client/windowstray/service_windows.go | 242 ++++++++++ internal/client/windowstray/tray_windows.go | 419 ++++++++++++++++++ internal/observability/health.go | 151 +++++++ internal/observability/health_test.go | 42 ++ internal/server/control/service.go | 19 +- internal/server/control/service_test.go | 4 + internal/server/session/agent_server.go | 23 +- internal/server/store/acceptance.go | 60 ++- internal/server/store/audit_helpers.go | 21 + internal/server/store/command_test.go | 64 +++ internal/server/store/event.go | 36 +- internal/server/store/events_read_test.go | 14 +- internal/server/store/migrations.go | 16 + internal/server/store/query.go | 12 +- scripts/windows/test-host.ps1 | 55 ++- test/coverage.toml | 18 + .../store/store_integration_test.go | 4 +- 38 files changed, 2188 insertions(+), 106 deletions(-) create mode 100644 internal/client/windowstray/service_other.go create mode 100644 internal/client/windowstray/service_windows.go create mode 100644 internal/client/windowstray/tray_windows.go create mode 100644 internal/observability/health.go create mode 100644 internal/observability/health_test.go create mode 100644 internal/server/store/audit_helpers.go diff --git a/cmd/rvbox-server/main.go b/cmd/rvbox-server/main.go index 16c3929..a4257a9 100644 --- a/cmd/rvbox-server/main.go +++ b/cmd/rvbox-server/main.go @@ -12,10 +12,13 @@ import ( "os" "os/signal" "syscall" + "time" rvboxv1 "github.com/rvbox/rvbox/gen/go/rvbox/v1" "github.com/rvbox/rvbox/internal/agentproto" "github.com/rvbox/rvbox/internal/config" + "github.com/rvbox/rvbox/internal/domain" + "github.com/rvbox/rvbox/internal/observability" "github.com/rvbox/rvbox/internal/server/control" "github.com/rvbox/rvbox/internal/server/session" "github.com/rvbox/rvbox/internal/server/store" @@ -24,12 +27,26 @@ import ( func main() { var configPath string + var checkConfig bool flag.StringVar(&configPath, "config", "", "absolute server TOML configuration path") + flag.BoolVar(&checkConfig, "check-config", false, "validate server configuration and exit") flag.Parse() if configPath == "" { log.Print("rvbox-server: --config is required") os.Exit(2) } + if checkConfig { + configured, err := config.LoadServer(configPath) + if err != nil { + log.Printf("rvbox-server: invalid configuration: %v", err) + os.Exit(1) + } + if _, err := fmt.Fprintf(os.Stdout, "valid server configuration: data_dir=%s agent_listen=%s control_socket=%s observability_listen=%s json_rpc_enabled=%t\n", configured.Server.DataDir, configured.Server.AgentListen, configured.Server.ControlSocket, configured.Observability.Listen, configured.JSONRPC.Enabled); err != nil { + log.Printf("rvbox-server: write check result: %v", err) + os.Exit(1) + } + return + } if err := run(configPath); err != nil { log.Printf("rvbox-server: %v", err) os.Exit(1) @@ -58,6 +75,43 @@ func run(configPath string) error { return err } defer persistence.Close() + health := observability.New() + healthServer := &http.Server{Handler: health.Handler(observability.Paths{Liveness: configured.Observability.LivenessPath, Readiness: configured.Observability.ReadinessPath, Metrics: configured.Observability.MetricsPath}), ReadHeaderTimeout: configured.Flow.WriteDeadline} + // Health is deliberately best effort. A port collision or a temporary + // listener failure must not prevent the control/agent service from starting; + // operators still get the failure in the daemon log. + var healthListener net.Listener + if configured.Observability.Listen != "" { + healthListener, err = net.Listen("tcp", configured.Observability.Listen) + if err != nil { + log.Printf("rvbox-server: observability endpoint unavailable (continuing without it): %v", err) + } + } + if healthListener != nil { + defer healthListener.Close() + } + recoveryContext, cancelRecovery := context.WithCancel(context.Background()) + defer cancelRecovery() + // Recovery is deliberately asynchronous: liveness and incident inspection + // remain available while committed-range checks run. Readiness becomes + // true only after the real SQLite/segment recovery completes. + go func() { + if _, recoverErr := persistence.RecoverCommandSegments(recoveryContext); recoverErr != nil { + health.SetDirty(true) + if incidentID, idErr := domain.NewUUIDv7(); idErr == nil { + if _, incidentErr := persistence.RecordIncident(context.Background(), store.IncidentInput{ + IncidentUUID: incidentID, DetectedAt: time.Now().UTC(), Kind: store.IncidentSQLiteIntegrity, + Scope: store.IncidentScopeGlobal, ScopeKey: "server-startup-recovery", Summary: "startup storage recovery failed", + Evidence: []byte(recoverErr.Error()), AutomaticallyRepairable: false, + }); incidentErr != nil { + log.Printf("rvbox-server: could not persist recovery incident: %v", incidentErr) + } + } + log.Printf("rvbox-server: storage recovery left readiness disabled: %v", recoverErr) + return + } + health.SetReady(true) + }() listener, err := net.Listen("tcp", configured.Server.AgentListen) if err != nil { @@ -106,9 +160,12 @@ func run(configPath string) error { HeartbeatIdle: configured.Protocol.HeartbeatIdle, LivenessTimeout: configured.Protocol.LivenessTimeout, } httpServer := &http.Server{Handler: agent, ReadHeaderTimeout: configured.Flow.WriteDeadline} - serveError := make(chan error, 3) + serveError := make(chan error, 4) go func() { serveError <- httpServer.Serve(listener) }() go func() { serveError <- grpcServer.Serve(controlListener) }() + if healthListener != nil { + go func() { serveError <- healthServer.Serve(healthListener) }() + } if rpcServer != nil { go func() { serveError <- rpcServer.Serve(rpcListener) }() } @@ -119,6 +176,9 @@ func run(configPath string) error { select { case err := <-serveError: if errors.Is(err, http.ErrServerClosed) { + if healthListener != nil { + _ = healthServer.Close() + } if rpcServer != nil { _ = rpcServer.Close() } @@ -126,6 +186,9 @@ func run(configPath string) error { return nil } _ = httpServer.Close() + if healthListener != nil { + _ = healthServer.Close() + } if rpcServer != nil { _ = rpcServer.Close() } @@ -135,6 +198,13 @@ func run(configPath string) error { shutdownContext, cancel := context.WithTimeout(context.Background(), configured.Server.ShutdownGrace) defer cancel() httpErr := httpServer.Shutdown(shutdownContext) + var healthErr error + if healthListener != nil { + healthErr = healthServer.Shutdown(shutdownContext) + } + if httpErr == nil { + httpErr = healthErr + } if rpcServer != nil { if err := rpcServer.Shutdown(shutdownContext); httpErr == nil { httpErr = err diff --git a/cmd/rvbox/main.go b/cmd/rvbox/main.go index c49c832..ac2c7f7 100644 --- a/cmd/rvbox/main.go +++ b/cmd/rvbox/main.go @@ -22,6 +22,7 @@ import ( "github.com/rvbox/rvbox/internal/client/windowsservice" "github.com/rvbox/rvbox/internal/config" "github.com/rvbox/rvbox/internal/domain" + "github.com/rvbox/rvbox/internal/observability" "google.golang.org/protobuf/types/known/timestamppb" ) @@ -40,7 +41,7 @@ func run(args []string, output, diagnostics io.Writer) error { return errors.New("an internal mode is required (use --help)") } if args[0] == "--help" || args[0] == "-h" { - _, err := io.WriteString(output, "usage: rvbox --service|--tray|--check-config|--install-service|--uninstall-service|--start-service|--stop-service --config PATH\n") + _, err := io.WriteString(output, "usage: rvbox --service|--tray|--check-config|--install-service|--uninstall-service|--configure-service|--start-service|--stop-service|--restart-service --config PATH\n") return err } flags := flag.NewFlagSet("rvbox", flag.ContinueOnError) @@ -51,8 +52,11 @@ func run(args []string, output, diagnostics io.Writer) error { checkConfig := flags.Bool("check-config", false, "validate client configuration and exit") install := flags.Bool("install-service", false, "install or update the machine-wide service") uninstall := flags.Bool("uninstall-service", false, "remove the machine-wide service") + configure := flags.Bool("configure-service", false, "configure machine-wide service startup mode") + startup := flags.String("startup", string(windowsservice.StartupAutomatic), "service startup mode: automatic or manual") start := flags.Bool("start-service", false, "start the machine-wide service") stop := flags.Bool("stop-service", false, "stop the machine-wide service") + restart := flags.Bool("restart-service", false, "restart the machine-wide service") if err := flags.Parse(args); err != nil { return err } @@ -60,7 +64,7 @@ func run(args []string, output, diagnostics io.Writer) error { return fmt.Errorf("unexpected argument %q", flags.Arg(0)) } selected := 0 - for _, value := range []bool{*serviceMode, *trayMode, *checkConfig, *install, *uninstall, *start, *stop} { + for _, value := range []bool{*serviceMode, *trayMode, *checkConfig, *install, *uninstall, *configure, *start, *stop, *restart} { if value { selected++ } @@ -69,10 +73,11 @@ func run(args []string, output, diagnostics io.Writer) error { return errors.New("select exactly one rvbox mode") } if *checkConfig { - if _, err := loadClientConfig(*configPath); err != nil { + configured, err := loadClientConfig(*configPath) + if err != nil { return err } - _, err := fmt.Fprintf(output, "valid client configuration: %s\n", *configPath) + _, err = fmt.Fprintf(output, "valid client configuration: path=%s server_url=%s state_dir=%s daemon_cwd=%s observability_listen=%s\n", *configPath, configured.Client.ServerURL, configured.Client.StateDir, configured.Client.DaemonCWD, configured.Observability.Listen) return err } if *install { @@ -85,12 +90,18 @@ func run(args []string, output, diagnostics io.Writer) error { if *uninstall { return windowsservice.Uninstall() } + if *configure { + return windowsservice.Configure(windowsservice.StartupMode(*startup)) + } if *start { return windowsservice.Start() } if *stop { return windowsservice.Stop(30) } + if *restart { + return windowsservice.Restart(30) + } if *trayMode { return runTray(*configPath, diagnostics) } @@ -116,8 +127,15 @@ func runClientDaemon(ctx context.Context, configPath string, diagnostics io.Writ if err != nil { return err } + health := observability.New() + go func() { + if serveErr := health.Serve(ctx, configured.Observability.Listen, observability.Paths{Liveness: configured.Observability.LivenessPath, Readiness: configured.Observability.ReadinessPath, Metrics: configured.Observability.MetricsPath}); serveErr != nil && ctx.Err() == nil && diagnostics != nil { + _, _ = fmt.Fprintf(diagnostics, "rvbox client observability endpoint stopped: %v\n", serveErr) + } + }() state, err := spool.Open(ctx, spool.Options{DataDir: configured.Client.StateDir, BusyTimeout: 5 * time.Second, TombstoneLimit: configured.Storage.TombstoneMaxEntries, MaxScriptBytes: configured.Execution.MaxScriptBytes, MaxExecutionSpecBytes: configured.Execution.MaxExecutionSpecBytes, QuotaLimits: spool.QuotaLimits{HardAllocationBytes: 1 << 20, CommandOutputBytes: configured.Storage.CommandOutputLimitBytes, CommandTotalBytes: configured.Storage.CommandTotalLimitBytes, ClientTotalBytes: configured.Storage.ClientTotalLimitBytes, CloseoutReserveBytes: configured.Storage.CommandCloseoutReserveBytes}}) if err != nil { + health.SetDirty(true) return fmt.Errorf("open client durable state: %w", err) } defer state.Close() @@ -134,7 +152,7 @@ func runClientDaemon(ctx context.Context, configPath string, diagnostics io.Writ limits = agentproto.DefaultLimits() } eventReady := make(chan domain.UUID, 256) - supervised, err := clientwindows.NewSupervisor(clientwindows.NativeOptions{Shells: clientwindows.ShellPaths{CMD: configured.Shells.CMD, PowerShell: configured.Shells.PowerShell}, WorkRoot: configured.Client.DaemonCWD, MaxWrapperBytes: configured.Execution.MaxScriptBytes, MaxOutputChunk: configured.Execution.MaxRawChunkBytes, WindowsTermGrace: configured.Execution.WindowsTermGrace}) + supervised, err := clientwindows.NewSupervisor(clientwindows.NativeOptions{Shells: clientwindows.ShellPaths{CMD: configured.Shells.CMD, PowerShell: configured.Shells.PowerShell}, WorkRoot: configured.Client.DaemonCWD, JobProfiles: clientJobProfiles(configured.Profiles), MaxWrapperBytes: configured.Execution.MaxScriptBytes, MaxOutputChunk: configured.Execution.MaxRawChunkBytes, WindowsTermGrace: configured.Execution.WindowsTermGrace}) if err != nil { return fmt.Errorf("configure command supervisor: %w", err) } @@ -149,12 +167,14 @@ func runClientDaemon(ctx context.Context, configPath string, diagnostics io.Writ } runner := func() { if _, checkErr := state.Check(ctx); checkErr != nil { + health.SetDirty(true) if diagnostics != nil { _, _ = fmt.Fprintf(diagnostics, "rvbox client spool is dirty: %v\n", checkErr) } return } if recovered, recoverErr := state.RecoverLaunchUncertainty(ctx, time.Now().UTC()); recoverErr != nil { + health.SetDirty(true) if diagnostics != nil { _, _ = fmt.Fprintf(diagnostics, "rvbox launch recovery failed: %v\n", recoverErr) } @@ -162,6 +182,7 @@ func runClientDaemon(ctx context.Context, configPath string, diagnostics io.Writ } else if len(recovered) > 0 && diagnostics != nil { _, _ = fmt.Fprintf(diagnostics, "rvbox recovered %d uncertain launch(es)\n", len(recovered)) } + health.SetReady(true) if runErr := agent.Run(ctx, agent.RunnerOptions{ Store: state, Dial: func(dialContext context.Context) (agent.Transport, error) { @@ -182,6 +203,22 @@ func runClientDaemon(ctx context.Context, configPath string, diagnostics io.Writ return nil } +func clientJobProfiles(profiles config.Profiles) map[string]clientwindows.JobProfile { + return map[string]clientwindows.JobProfile{ + rvboxv1.ExecutionProfile_EXECUTION_PROFILE_LIGHT.String(): toJobProfile(profiles.Light), + rvboxv1.ExecutionProfile_EXECUTION_PROFILE_CPU_MEDIUM.String(): toJobProfile(profiles.CPUMedium), + rvboxv1.ExecutionProfile_EXECUTION_PROFILE_CPU_HEAVY.String(): toJobProfile(profiles.CPUHeavy), + rvboxv1.ExecutionProfile_EXECUTION_PROFILE_MEM_MEDIUM.String(): toJobProfile(profiles.MemMedium), + rvboxv1.ExecutionProfile_EXECUTION_PROFILE_MEM_HEAVY.String(): toJobProfile(profiles.MemHeavy), + rvboxv1.ExecutionProfile_EXECUTION_PROFILE_DISK_MEDIUM.String(): toJobProfile(profiles.DiskMedium), + rvboxv1.ExecutionProfile_EXECUTION_PROFILE_DISK_HEAVY.String(): toJobProfile(profiles.DiskHeavy), + } +} + +func toJobProfile(profile config.Profile) clientwindows.JobProfile { + return clientwindows.JobProfile{RequiredControls: append([]string(nil), profile.RequiredControls...), CPUPercent: profile.CPUPercent, MemoryMaxBytes: profile.MemoryMaxBytes, PIDsMax: profile.PIDsMax, IOReadBPS: profile.WindowsIOReadBPS, IOWriteBPS: profile.WindowsIOWriteBPS} +} + func clientHTTPClient(settings config.TLS) (*http.Client, error) { tlsConfig := &tls.Config{MinVersion: tls.VersionTLS12, ServerName: settings.ServerName} // #nosec G402 -- TLS 1.2 is the v1 floor. if settings.CAFile != "" { diff --git a/cmd/rvbox/service_windows.go b/cmd/rvbox/service_windows.go index 99ca77f..8200237 100644 --- a/cmd/rvbox/service_windows.go +++ b/cmd/rvbox/service_windows.go @@ -9,8 +9,10 @@ import ( "io" "os" "path/filepath" + "strings" "github.com/rvbox/rvbox/internal/client/windowsservice" + "github.com/rvbox/rvbox/internal/client/windowstray" "golang.org/x/sys/windows/svc" ) @@ -34,9 +36,80 @@ func runService(configPath string, diagnostics io.Writer) error { } func runWindowsService(configPath string, diagnostics io.Writer) error { - return windowsservice.Run(func(ctx context.Context) error { return runClientDaemon(ctx, configPath, diagnostics) }) + return windowsservice.Run(func(ctx context.Context) error { + // The tray endpoint lives in the same LocalSystem service process. It + // has no store access; the handler below returns only bounded status/path + // data and rechecks SCM authorization in the native pipe adapter. + go func() { + if err := windowstray.Serve(ctx, func(requestContext context.Context, _ windowstray.Peer, request windowstray.Frame) (windowstray.Frame, error) { + return handleTrayRequest(requestContext, configPath, request) + }); err != nil && ctx.Err() == nil && diagnostics != nil { + _, _ = fmt.Fprintf(diagnostics, "rvbox tray endpoint stopped: %v\n", err) + } + }() + return runClientDaemon(ctx, configPath, diagnostics) + }) } func runTray(configPath string, diagnostics io.Writer) error { - return errors.New("Windows tray frontend is not available in this build") + _ = configPath // the tray obtains the canonical paths from the service. + return windowstray.Run(context.Background(), diagnostics) +} + +func handleTrayRequest(ctx context.Context, configPath string, request windowstray.Frame) (windowstray.Frame, error) { + response := windowstray.Frame{Action: windowstray.ActionStatus} + switch request.Action { + case windowstray.ActionStatus: + response.Payload = []byte("RVBox service=running") + case windowstray.ActionOpenConfig: + response.Payload = []byte(configPath) + case windowstray.ActionOpenLog: + configured, err := loadClientConfig(configPath) + if err != nil { + return response, err + } + logPath := configured.Observability.LogFile + if logPath == "" { + logPath = filepath.Join(filepath.Dir(configPath), "logs", "rvbox.log") + } + response.Payload = []byte(logPath) + case windowstray.ActionStartService: + if err := windowsservice.Start(); err != nil { + return response, err + } + response.Payload = []byte("RVBox service start requested") + case windowstray.ActionStopService: + // A service cannot synchronously wait for its own stop request from the + // pipe handler: the SCM callback must return so the process can unwind. + go func() { _ = windowsservice.Stop(30) }() + response.Payload = []byte("RVBox service stop requested") + case windowstray.ActionRestartService: + // Restart must be performed by the external UAC helper. If the tray is + // already elevated, its native fallback still launches the canonical + // --restart-service mode outside this service process. + return response, errors.New("restart requires the external service helper") + case windowstray.ActionSetAutomatic: + if err := windowsservice.Configure(windowsservice.StartupAutomatic); err != nil { + return response, err + } + response.Payload = []byte("RVBox service startup set to automatic") + case windowstray.ActionSetManual: + if err := windowsservice.Configure(windowsservice.StartupManual); err != nil { + return response, err + } + response.Payload = []byte("RVBox service startup set to manual") + case windowstray.ActionExitTray: + response.Payload = []byte("tray exit acknowledged") + default: + return response, fmt.Errorf("unknown tray action %d", request.Action) + } + if len(response.Payload) > 4<<10 { + response.Payload = []byte(strings.TrimSpace(string(response.Payload[:4<<10]))) + } + select { + case <-ctx.Done(): + return response, ctx.Err() + default: + return response, nil + } } diff --git a/cmd/rvc/main.go b/cmd/rvc/main.go index c79a485..38c052f 100644 --- a/cmd/rvc/main.go +++ b/cmd/rvc/main.go @@ -193,14 +193,99 @@ func stat(ctx context.Context, client rvboxv1.ControlClient, args []string, outp 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 { - fmt.Fprintf(output, "queue_expiry=%s\n", item.GetQueueExpiryTime().AsTime().UTC().Format(time.RFC3339Nano)) + 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)) } - return nil + 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 { diff --git a/cmd/rvc/main_test.go b/cmd/rvc/main_test.go index a141c40..e88fb58 100644 --- a/cmd/rvc/main_test.go +++ b/cmd/rvc/main_test.go @@ -1,11 +1,13 @@ package main import ( + "bytes" "path/filepath" "testing" "time" rvboxv1 "github.com/rvbox/rvbox/gen/go/rvbox/v1" + "google.golang.org/protobuf/types/known/timestamppb" ) func TestGlobalSocketAndCLIValueParsing_HP_CTL_11(t *testing.T) { @@ -47,3 +49,32 @@ func TestGlobalRequestIDIsInjectedOnlyForMutations_HP_CTL_12(t *testing.T) { t.Fatal("duplicate global request IDs accepted") } } + +func TestRenderCommandStatShowsExpiryRetentionAndWindowsIdentity_HP_CTL_13(t *testing.T) { + t.Parallel() + expiry := time.Date(2026, 9, 6, 12, 0, 0, 0, time.UTC) + sessionID := uint32(7) + effective := rvboxv1.WindowsExecutionContext_WINDOWS_EXECUTION_CONTEXT_ACTIVE_SYSTEM + var output bytes.Buffer + renderCommandStat(&output, &rvboxv1.CommandRecord{ + IssueUuid: "019c46f1-1d02-7000-8000-0000000000f2", TargetClientId: "win-a", + Lifecycle: rvboxv1.CommandLifecycle_COMMAND_SUCCEEDED, CommandRevision: 3, + LastEventSeq: 9, QueueExpiryTime: timestamppb.New(expiry), OutputIncomplete: true, + OutputTruncated: true, RetainedCompressedBytes: 1234, LateAfterExpiry: true, + WindowsExecutionIdentity: &rvboxv1.WindowsExecutionIdentity{ + EffectiveContext: &effective, SessionId: &sessionID, EffectiveUserSid: "S-1-5-18", + SessionUserSid: "S-1-5-21-user", AttemptedContexts: []rvboxv1.WindowsExecutionContext{ + rvboxv1.WindowsExecutionContext_WINDOWS_EXECUTION_CONTEXT_ACTIVE_USER_ELEVATED, effective, + }, SelectionDetail: "fallback\nused", + }, + }, expiry.Add(time.Minute)) + for _, want := range []string{ + "queue_expired=true", "late_after_expiry=true", "output_truncated=true", "output_incomplete=true", + "retained_compressed_bytes=1234", "windows_attempted_contexts=active-user-elevated,active-system", + "windows_effective_context=active-system", "windows_session_id=7", "windows_selection_detail=fallback\\nused", + } { + if !bytes.Contains(output.Bytes(), []byte(want)) { + t.Errorf("stat output missing %q:\n%s", want, output.String()) + } + } +} diff --git a/docs/implementation-plan.v1.md b/docs/implementation-plan.v1.md index df5676b..b9f6665 100644 --- a/docs/implementation-plan.v1.md +++ b/docs/implementation-plan.v1.md @@ -1294,7 +1294,7 @@ invariants must not. | --- | --- | | `clients` | client ID, most-recent platform/capabilities/CWD/version, durable instance ID, current generation, connection/last-seen timestamps, latest rejected live-conflict instance/time, unified charged command-storage total | | `sessions` | opaque session ID, client ID, durable client-instance UUID, generation, opened/fenced/closed times, close reason | -| `commands` | UUID, client ID, indexed issue/queue-expiry/terminal times, lifecycle, revision, exit result, last event sequence, retention status, a Zstandard-compressed immutable execution-spec payload including plaintext environment values, and optional Windows selection attempts/effective identity | +| `commands` | UUID, client ID, indexed issue/queue-expiry/terminal times, lifecycle, revision, exit result, last event sequence, retention status, a bounded structured admission-rejection payload when the client declines dispatch, a Zstandard-compressed immutable execution-spec payload including plaintext environment values, and optional Windows selection attempts/effective identity | | `command_payloads` | command UUID, payload kind (script or other command-owned blob), Zstandard compression, raw/stored sizes, digest, inline bytes or validated segment reference | | `command_events` | command UUID + event sequence unique key, observed and server receipt times, event type, payload metadata, immutable duplicate checksum | | `output_segments` | command UUID, segment ordinal/path, `committed_end_offset`, min/max event sequence, stream mix, compressed/raw byte totals, checksum, created time | diff --git a/docs/platform-and-operations.md b/docs/platform-and-operations.md index cec3e89..606bdf9 100644 --- a/docs/platform-and-operations.md +++ b/docs/platform-and-operations.md @@ -79,7 +79,8 @@ history, process handles, and Job Objects. Normal service startup never shows UAC and is not blocked by tray or interactive-user availability. The same signed `rvbox.exe` has explicit `service`, `tray`, `install-service`, -`uninstall-service`, `configure-service`, per-command launcher, and signal- +`uninstall-service`, `start-service`, `stop-service`, `restart-service`, +`configure-service`, per-command launcher, and signal- helper modes. Internal modes require SCM state or a service-created launch proof. Task Scheduler is not used. The installer registers an unelevated per- user tray launch through the machine-wide `Run` key. One tray may run in each diff --git a/docs/testing.md b/docs/testing.md index e250de1..dd159ad 100644 --- a/docs/testing.md +++ b/docs/testing.md @@ -82,8 +82,11 @@ cleanup. test reference against source. A resettable Windows smoke VM is now available. The exact headless VirtualBox/Guest Control adapter is `scripts/windows/test-host.ps1`; it takes the VM name, baseline snapshot, and -guest credentials only from host environment variables, acquires an exclusive -lease, and never writes secrets to the repository. Use `Prepare`, `Run`, +guest identity/password-file only from host environment variables, acquires an +exclusive lease, and never writes secrets to the repository. Set +`RVBOX_WINDOWS_GUEST_PASSWORD_FILE` to a mode-600 file outside the repository; +the adapter passes it with VirtualBox `--passwordfile` and never accepts an +inline password. Use `Prepare`, `Run`, `Collect`, `Stop`, and `Reset` in that order for a native run. The VM is the minimum smoke lane, so deferred native multi-session/ambiguous-session, Server Core, and older-build entries remain explicitly blocked until their own diff --git a/internal/client/agent/executor.go b/internal/client/agent/executor.go index 60539cf..d680a2b 100644 --- a/internal/client/agent/executor.go +++ b/internal/client/agent/executor.go @@ -129,6 +129,10 @@ func (executor *Executor) launch(ctx context.Context, issue domain.UUID, revisio if err != nil { cancel() _ = executor.Store.SetLaunchPhase(context.Background(), issue, domain.LaunchPhaseNone, "", 0) + var startErr *supervisor.StartError + if errors.As(err, &startErr) { + return executor.rejectWithIdentity(ctx, issue, revision, err, startErr.WindowsIdentity) + } return executor.reject(ctx, issue, revision, err) } identity := process.Identity() @@ -141,7 +145,7 @@ func (executor *Executor) launch(ctx context.Context, issue domain.UUID, revisio executor.active[issue] = process executor.cancel[issue] = cancel executor.mu.Unlock() - if _, err := executor.Store.AppendLifecycle(ctx, issue, uint32(rvboxv1.CommandLifecycle_COMMAND_RUNNING), revision, "process started", executor.Now()); err != nil { + if _, err := executor.Store.AppendLifecycleWithIdentity(ctx, issue, uint32(rvboxv1.CommandLifecycle_COMMAND_RUNNING), revision, "process started", executor.Now(), identity.WindowsIdentity()); err != nil { _, _ = executor.Supervisor.Signal(context.Background(), process, supervisor.SignalKill) cancel() executor.remove(issue) @@ -187,22 +191,30 @@ func (executor *Executor) watch(ctx context.Context, issue domain.UUID, revision if waitErr != nil { detail = boundedError(waitErr) } - if err := executor.appendLifecycle(context.Background(), issue, revision, phase, detail); err == nil { + if err := executor.appendLifecycleWithIdentity(context.Background(), issue, revision, phase, detail, process.Identity().WindowsIdentity()); err == nil { executor.notify(issue) } executor.remove(issue) } func (executor *Executor) appendLifecycle(ctx context.Context, issue domain.UUID, revision uint64, phase rvboxv1.CommandLifecycle, detail string) error { - _, err := executor.Store.AppendLifecycle(ctx, issue, uint32(phase), revision, detail, executor.Now()) + return executor.appendLifecycleWithIdentity(ctx, issue, revision, phase, detail, nil) +} + +func (executor *Executor) appendLifecycleWithIdentity(ctx context.Context, issue domain.UUID, revision uint64, phase rvboxv1.CommandLifecycle, detail string, identity *rvboxv1.WindowsExecutionIdentity) error { + _, err := executor.Store.AppendLifecycleWithIdentity(ctx, issue, uint32(phase), revision, detail, executor.Now(), identity) return err } func (executor *Executor) reject(ctx context.Context, issue domain.UUID, revision uint64, cause error) error { + return executor.rejectWithIdentity(ctx, issue, revision, cause, nil) +} + +func (executor *Executor) rejectWithIdentity(ctx context.Context, issue domain.UUID, revision uint64, cause error, identity *rvboxv1.WindowsExecutionIdentity) error { if revision == 0 { return cause } - if err := executor.appendLifecycle(ctx, issue, revision, rvboxv1.CommandLifecycle_COMMAND_REJECTED, boundedError(cause)); err != nil { + if err := executor.appendLifecycleWithIdentity(ctx, issue, revision, rvboxv1.CommandLifecycle_COMMAND_REJECTED, boundedError(cause), identity); err != nil { return err } executor.notify(issue) diff --git a/internal/client/agent/runner.go b/internal/client/agent/runner.go index f796a0f..46e25d6 100644 --- a/internal/client/agent/runner.go +++ b/internal/client/agent/runner.go @@ -94,12 +94,12 @@ func Run(ctx context.Context, options RunnerOptions) error { if err := waitUntil(ctx, delay); err != nil { return nil } - sessionStarted := time.Now() + sessionStarted := options.Now() _ = runOnce(ctx, options) if ctx.Err() != nil { return nil } - if time.Since(sessionStarted) >= options.Backoff.StableReset { + if options.Now().Sub(sessionStarted) >= options.Backoff.StableReset { failures = 0 } if failures < ^uint32(0) { @@ -357,11 +357,11 @@ func serveActive(ctx context.Context, transport Transport, options RunnerOptions return err } case envelope.GetScriptChunk() != nil: - if err := handleScriptChunk(ctx, transport, options.Store, session, envelope.GetScriptChunk(), limits); err != nil { + if err := handleScriptChunk(ctx, transport, options.Store, session, envelope.GetScriptChunk(), limits, options.Now, sent); err != nil { return err } case envelope.GetScriptCommit() != nil: - if err := handleScriptCommit(ctx, transport, options.Store, session, envelope.GetScriptCommit(), limits); err != nil { + if err := handleScriptCommit(ctx, transport, options.Store, session, envelope.GetScriptCommit(), limits, options.Now, sent); err != nil { return err } if options.OnScriptReady != nil { @@ -488,28 +488,38 @@ func boundedError(err error) string { return message } -func handleScriptChunk(ctx context.Context, transport Transport, store *spool.Store, session Session, chunk *rvboxv1.ScriptChunk, limits agentproto.Limits) error { +func handleScriptChunk(ctx context.Context, transport Transport, store *spool.Store, session Session, chunk *rvboxv1.ScriptChunk, limits agentproto.Limits, now func() time.Time, sent map[domain.UUID]uint64) error { status, err := ApplyScriptChunk(ctx, store, session, chunk, limits) if err != nil { return err } - return appendAndSendScriptStatus(ctx, transport, store, session, chunk.GetIssueUuid(), status, limits) + return appendAndSendScriptStatus(ctx, transport, store, session, chunk.GetIssueUuid(), status, limits, now, sent) } -func handleScriptCommit(ctx context.Context, transport Transport, store *spool.Store, session Session, commit *rvboxv1.ScriptCommit, limits agentproto.Limits) error { +func handleScriptCommit(ctx context.Context, transport Transport, store *spool.Store, session Session, commit *rvboxv1.ScriptCommit, limits agentproto.Limits, now func() time.Time, sent map[domain.UUID]uint64) error { status, err := ApplyScriptCommit(ctx, store, session, commit, limits) if err != nil { return err } - return appendAndSendScriptStatus(ctx, transport, store, session, commit.GetIssueUuid(), status, limits) + return appendAndSendScriptStatus(ctx, transport, store, session, commit.GetIssueUuid(), status, limits, now, sent) } -func appendAndSendScriptStatus(ctx context.Context, transport Transport, store *spool.Store, session Session, issueText string, status spool.ScriptStatus, limits agentproto.Limits) error { +func appendAndSendScriptStatus(ctx context.Context, transport Transport, store *spool.Store, session Session, issueText string, status spool.ScriptStatus, limits agentproto.Limits, now func() time.Time, sent map[domain.UUID]uint64) error { issue, err := domain.ParseUUIDv7(issueText) if err != nil { return err } - event := &rvboxv1.CommandEvent{IssueUuid: issueText, ObservedAt: timestamppb.New(time.Now().UTC()), Payload: &rvboxv1.CommandEvent_ScriptStatus{ScriptStatus: &rvboxv1.ScriptUploadStatus{ReceivedBytes: status.ReceivedBytes, Complete: status.Committed}}} + if sent == nil { + sent = make(map[domain.UUID]uint64) + } + if now == nil { + now = func() time.Time { return time.Now().UTC() } + } + observedAt := now() + if observedAt.IsZero() { + return errors.New("script status clock returned zero") + } + event := &rvboxv1.CommandEvent{IssueUuid: issueText, ObservedAt: timestamppb.New(observedAt), Payload: &rvboxv1.CommandEvent_ScriptStatus{ScriptStatus: &rvboxv1.ScriptUploadStatus{ReceivedBytes: status.ReceivedBytes, Complete: status.Committed}}} payload, err := proto.MarshalOptions{Deterministic: true}.Marshal(event) if err != nil { return err @@ -522,9 +532,13 @@ func appendAndSendScriptStatus(ctx context.Context, transport Transport, store * return err } for _, item := range assigned { + if item.EventSeq == 0 || item.EventSeq <= sent[issue] { + continue + } if err := SendStoredEvent(ctx, transport, session, item, limits); err != nil { return err } + sent[issue] = item.EventSeq } return nil } diff --git a/internal/client/spool/events.go b/internal/client/spool/events.go index d2ffa01..6767839 100644 --- a/internal/client/spool/events.go +++ b/internal/client/spool/events.go @@ -267,14 +267,25 @@ func (store *Store) AssignSendWindow(ctx context.Context, issueUUID domain.UUID, return nil, err } defer tx.Rollback() - var next uint64 - err = tx.QueryRowContext(ctx, `SELECT next_event_seq FROM commands WHERE issue_uuid = ?`, issueUUID[:]).Scan(&next) + var next, lastAck, assignedBytes uint64 + err = tx.QueryRowContext(ctx, `SELECT next_event_seq, last_server_ack, +COALESCE((SELECT sum(length(payload)) FROM events + WHERE issue_uuid = commands.issue_uuid AND event_seq IS NOT NULL AND event_seq > commands.last_server_ack), 0) +FROM commands WHERE issue_uuid = ?`, issueUUID[:]).Scan(&next, &lastAck, &assignedBytes) if err == sql.ErrNoRows { return nil, ErrUnknownCommand } if err != nil { return nil, err } + // Assigned rows remain pinned until cumulative acknowledgement. Never + // assign another row while that durable send window is full; otherwise a + // burst of local events could grow the pinned set without bound even though + // callers pass a per-command byte limit on every invocation. + if assignedBytes >= maximumStoredBytes { + return eventsBySequence(ctx, tx, issueUUID, 0) + } + maximumStoredBytes -= assignedBytes rows, err := tx.QueryContext(ctx, `SELECT local_ordinal, length(payload) FROM events WHERE issue_uuid = ? AND event_seq IS NULL ORDER BY local_ordinal LIMIT ?`, issueUUID[:], maximumEvents) if err != nil { return nil, err @@ -475,6 +486,14 @@ func (store *Store) RecoverLaunchUncertainty(ctx context.Context, now time.Time) // crash between a terminal marker and its public event from creating a state // that can be replayed as a second execution. func (store *Store) AppendLifecycle(ctx context.Context, issueUUID domain.UUID, phase uint32, revision uint64, detail string, observedAt time.Time) (Event, error) { + return store.AppendLifecycleWithIdentity(ctx, issueUUID, phase, revision, detail, observedAt, nil) +} + +// AppendLifecycleWithIdentity is the lifecycle boundary used by the client +// runtime when the Windows supervisor has captured an immutable selection +// record. Keeping the legacy wrapper above preserves the small store API for +// recovery and tests that have no platform identity to attach. +func (store *Store) AppendLifecycleWithIdentity(ctx context.Context, issueUUID domain.UUID, phase uint32, revision uint64, detail string, observedAt time.Time, identity *rvboxv1.WindowsExecutionIdentity) (Event, error) { if !validUUID(issueUUID) || phase == 0 || phase > 11 || observedAt.IsZero() || revision == 0 { return Event{}, errors.New("invalid lifecycle event") } @@ -503,7 +522,7 @@ func (store *Store) AppendLifecycle(ctx context.Context, issueUUID domain.UUID, if !domain.CanTransition(rvboxv1.CommandLifecycle(current), rvboxv1.CommandLifecycle(phase)) { return Event{}, fmt.Errorf("invalid lifecycle transition %s -> %s", rvboxv1.CommandLifecycle(current), rvboxv1.CommandLifecycle(phase)) } - lifecycle := &rvboxv1.LifecycleChange{Lifecycle: rvboxv1.CommandLifecycle(phase), CommandRevision: revision, Detail: detail} + lifecycle := &rvboxv1.LifecycleChange{Lifecycle: rvboxv1.CommandLifecycle(phase), CommandRevision: revision, Detail: detail, WindowsExecutionIdentity: identity} payload, err := proto.MarshalOptions{Deterministic: true}.Marshal(&rvboxv1.CommandEvent{IssueUuid: issueUUID.String(), ObservedAt: timestamppb.New(observedAt), Payload: &rvboxv1.CommandEvent_Lifecycle{Lifecycle: lifecycle}}) if err != nil { return Event{}, err diff --git a/internal/client/spool/spool_test.go b/internal/client/spool/spool_test.go index 2fc15e1..616dade 100644 --- a/internal/client/spool/spool_test.go +++ b/internal/client/spool/spool_test.go @@ -13,6 +13,7 @@ import ( rvboxv1 "github.com/rvbox/rvbox/gen/go/rvbox/v1" "github.com/rvbox/rvbox/internal/domain" + "google.golang.org/protobuf/proto" ) func TestExecutionSpecIsDurableAndQuotaCounted_HP_DISPATCH_07(t *testing.T) { @@ -188,6 +189,38 @@ func TestSpoolAcceptanceSequencingAndAck_HP_CLIENT_07(t *testing.T) { } } +func TestSendWindowPinsUnacknowledgedBytes_BH_CLIENT_13(t *testing.T) { + t.Parallel() + ctx := context.Background() + store := openTestStore(t, ctx, filepath.Join(t.TempDir(), "spool"), DefaultTombstoneLimit) + issue := testUUID(t, "019c46f1-1d02-7000-8000-000000000014") + command := testCommand(issue, []byte("send-window")) + now := time.Date(2026, time.September, 6, 12, 0, 0, 0, time.UTC) + if _, err := store.AcceptCommand(ctx, command, now); err != nil { + t.Fatal(err) + } + for index, payload := range [][]byte{[]byte("first"), []byte("second")} { + if _, err := store.AppendEvent(ctx, issue, EventInput{Kind: uint32(index + 1), Compression: 1, Payload: payload, CreatedAt: now.Add(time.Duration(index) * time.Second)}); err != nil { + t.Fatal(err) + } + } + assigned, err := store.AssignSendWindow(ctx, issue, 1, uint64(len("first"))) + if err != nil || len(assigned) != 1 || assigned[0].EventSeq != 1 { + t.Fatalf("first bounded window = %#v, %v", assigned, err) + } + assigned, err = store.AssignSendWindow(ctx, issue, 1, uint64(len("first"))) + if err != nil || len(assigned) != 1 || assigned[0].EventSeq != 1 { + t.Fatalf("full pinned window assigned more data = %#v, %v", assigned, err) + } + if err := store.Ack(ctx, issue, 1); err != nil { + t.Fatal(err) + } + assigned, err = store.AssignSendWindow(ctx, issue, 1, uint64(len("second"))) + if err != nil || len(assigned) != 1 || assigned[0].EventSeq != 2 { + t.Fatalf("window after cumulative ack = %#v, %v", assigned, err) + } +} + func TestAppendLifecycleAtomicallyUpdatesPhaseAndEvent_HP_CLIENT_11(t *testing.T) { t.Parallel() ctx := context.Background() @@ -199,7 +232,9 @@ func TestAppendLifecycleAtomicallyUpdatesPhaseAndEvent_HP_CLIENT_11(t *testing.T if _, err := store.AcceptCommand(ctx, command, now); err != nil { t.Fatal(err) } - if _, err := store.AppendLifecycle(ctx, issue, uint32(rvboxv1.CommandLifecycle_COMMAND_RUNNING), 1, "launch authorized", now); err != nil { + effective := rvboxv1.WindowsExecutionContext_WINDOWS_EXECUTION_CONTEXT_ACTIVE_USER + identity := &rvboxv1.WindowsExecutionIdentity{EffectiveContext: &effective, SessionId: ptrUint32(1), SessionUserSid: "S-1-5-21-user", EffectiveUserSid: "S-1-5-21-user", AttemptedContexts: []rvboxv1.WindowsExecutionContext{effective}, SelectionDetail: "selected"} + if _, err := store.AppendLifecycleWithIdentity(ctx, issue, uint32(rvboxv1.CommandLifecycle_COMMAND_RUNNING), 1, "launch authorized", now, identity); err != nil { t.Fatal(err) } if _, err := store.AppendLifecycle(ctx, issue, uint32(rvboxv1.CommandLifecycle_COMMAND_SUCCEEDED), 1, "exit 0", now.Add(time.Second)); err != nil { @@ -223,8 +258,14 @@ func TestAppendLifecycleAtomicallyUpdatesPhaseAndEvent_HP_CLIENT_11(t *testing.T if err != nil || len(events) != 2 { t.Fatalf("lifecycle events = %#v, %v", events, err) } + var decoded rvboxv1.CommandEvent + if err := proto.Unmarshal(events[0].Payload, &decoded); err != nil || decoded.GetLifecycle().GetWindowsExecutionIdentity().GetEffectiveContext() != effective { + t.Fatalf("running identity event = %v, %v", decoded.GetLifecycle(), err) + } } +func ptrUint32(value uint32) *uint32 { return &value } + func TestTerminalCleanupTombstonesAndConflicts_BH_CLIENT_02(t *testing.T) { t.Parallel() ctx := context.Background() diff --git a/internal/client/supervisor/supervisor.go b/internal/client/supervisor/supervisor.go index da0faf7..e592bbc 100644 --- a/internal/client/supervisor/supervisor.go +++ b/internal/client/supervisor/supervisor.go @@ -17,6 +17,29 @@ var ( ErrUnsupported = errors.New("supervisor operation is unsupported") ) +// StartError carries durable context-selection evidence when a command is +// rejected before a process exists. The runtime records the evidence with +// the rejection event without treating the error as a second public protocol +// type. Unwrap keeps ordinary retry/error classification working. +type StartError struct { + Cause error + WindowsIdentity *rvboxv1.WindowsExecutionIdentity +} + +func (err *StartError) Error() string { + if err == nil || err.Cause == nil { + return "supervisor start failed" + } + return err.Cause.Error() +} + +func (err *StartError) Unwrap() error { + if err == nil { + return nil + } + return err.Cause +} + type SignalKind uint8 const ( @@ -53,12 +76,55 @@ func (spec StartSpec) Validate() error { // EffectiveIdentity is immutable process evidence captured before launch. // Empty user/session fields mean a Session 0 service context. type EffectiveIdentity struct { - Context string - SessionID uint32 - UserSID string - LogonSID string - Elevated bool - Integrity string + Context string + SessionID uint32 + SessionUserSID string + UserSID string + LogonSID string + Elevated bool + Integrity string + AttemptedContexts []string + SelectionDetail string +} + +// WindowsIdentity converts the platform-neutral evidence to the public +// immutable status/event shape. Unknown contexts are deliberately omitted so +// the portable test supervisor never pretends to be a Windows launch. +func (identity EffectiveIdentity) WindowsIdentity() *rvboxv1.WindowsExecutionIdentity { + result := &rvboxv1.WindowsExecutionIdentity{SessionUserSid: identity.SessionUserSID, EffectiveUserSid: identity.UserSID, SelectionDetail: identity.SelectionDetail} + for _, contextName := range identity.AttemptedContexts { + if context, ok := windowsExecutionContext(contextName); ok { + result.AttemptedContexts = append(result.AttemptedContexts, context) + } + } + if context, ok := windowsExecutionContext(identity.Context); ok { + result.EffectiveContext = &context + if context == rvboxv1.WindowsExecutionContext_WINDOWS_EXECUTION_CONTEXT_ACTIVE_USER || context == rvboxv1.WindowsExecutionContext_WINDOWS_EXECUTION_CONTEXT_ACTIVE_USER_ELEVATED || context == rvboxv1.WindowsExecutionContext_WINDOWS_EXECUTION_CONTEXT_ACTIVE_SYSTEM { + session := identity.SessionID + result.SessionId = &session + } + } + if result.EffectiveContext == nil && len(result.AttemptedContexts) == 0 && result.SelectionDetail == "" && result.SessionUserSid == "" && result.EffectiveUserSid == "" { + return nil + } + return result +} + +func windowsExecutionContext(value string) (rvboxv1.WindowsExecutionContext, bool) { + switch value { + case "LOCAL_SERVICE": + return rvboxv1.WindowsExecutionContext_WINDOWS_EXECUTION_CONTEXT_LOCAL_SERVICE, true + case "LOCAL_SYSTEM": + return rvboxv1.WindowsExecutionContext_WINDOWS_EXECUTION_CONTEXT_LOCAL_SYSTEM, true + case "ACTIVE_USER": + return rvboxv1.WindowsExecutionContext_WINDOWS_EXECUTION_CONTEXT_ACTIVE_USER, true + case "ACTIVE_USER_ELEVATED": + return rvboxv1.WindowsExecutionContext_WINDOWS_EXECUTION_CONTEXT_ACTIVE_USER_ELEVATED, true + case "ACTIVE_SYSTEM": + return rvboxv1.WindowsExecutionContext_WINDOWS_EXECUTION_CONTEXT_ACTIVE_SYSTEM, true + default: + return rvboxv1.WindowsExecutionContext_WINDOWS_EXECUTION_CONTEXT_UNSPECIFIED, false + } } type Process interface { diff --git a/internal/client/supervisor/supervisor_test.go b/internal/client/supervisor/supervisor_test.go index 4cd97f1..c3a9e0d 100644 --- a/internal/client/supervisor/supervisor_test.go +++ b/internal/client/supervisor/supervisor_test.go @@ -30,3 +30,25 @@ func TestStartSpecValidation_BH_SUPERVISOR_01(t *testing.T) { t.Fatal(err) } } + +func TestWindowsIdentityMapsSelectionEvidence_HP_SUPERVISOR_05(t *testing.T) { + t.Parallel() + identity := EffectiveIdentity{ + Context: "ACTIVE_SYSTEM", + SessionID: 7, + SessionUserSID: "S-1-5-21-user", + UserSID: "S-1-5-18", + AttemptedContexts: []string{"ACTIVE_USER_ELEVATED", "ACTIVE_SYSTEM"}, + SelectionDetail: "ACTIVE_USER_ELEVATED: ELEVATION_UNAVAILABLE", + } + encoded := identity.WindowsIdentity() + if encoded == nil || encoded.GetEffectiveContext() != rvboxv1.WindowsExecutionContext_WINDOWS_EXECUTION_CONTEXT_ACTIVE_SYSTEM || encoded.GetSessionId() != 7 || encoded.GetSessionUserSid() != "S-1-5-21-user" || encoded.GetEffectiveUserSid() != "S-1-5-18" { + t.Fatalf("mapped identity = %#v", encoded) + } + if len(encoded.GetAttemptedContexts()) != 2 || encoded.GetAttemptedContexts()[0] != rvboxv1.WindowsExecutionContext_WINDOWS_EXECUTION_CONTEXT_ACTIVE_USER_ELEVATED { + t.Fatalf("attempted contexts = %v", encoded.GetAttemptedContexts()) + } + if got := (EffectiveIdentity{AttemptedContexts: []string{"not-a-context"}}).WindowsIdentity(); got != nil { + t.Fatalf("unknown context evidence = %#v", got) + } +} diff --git a/internal/client/supervisor/windows/native_exec.go b/internal/client/supervisor/windows/native_exec.go index 0d6a389..9046254 100644 --- a/internal/client/supervisor/windows/native_exec.go +++ b/internal/client/supervisor/windows/native_exec.go @@ -35,12 +35,26 @@ var ( type NativeOptions struct { Shells ShellPaths WorkRoot string + JobProfiles map[string]JobProfile MaxWrapperBytes uint64 MaxOutputChunk uint64 WindowsTermGrace time.Duration Now func() time.Time } +// JobProfile is the validated administrator policy for one protocol profile. +// The native Windows adapter applies only controls it can set and read back +// atomically on a Job Object; a requested unsupported required control rejects +// the command before any child is created. +type JobProfile struct { + RequiredControls []string + CPUPercent uint64 + MemoryMaxBytes uint64 + PIDsMax uint64 + IOReadBPS uint64 + IOWriteBPS uint64 +} + func (options NativeOptions) withDefaults() NativeOptions { if options.MaxWrapperBytes == 0 { options.MaxWrapperBytes = 10 << 20 @@ -69,17 +83,19 @@ func newExecSupervisor(options NativeOptions) *execSupervisor { } type execProcess struct { - issue domain.UUID - identity supervisor.EffectiveIdentity - cmd *exec.Cmd - stdin io.WriteCloser - stdout io.ReadCloser - stderr io.ReadCloser - outputs chan outputResult - done chan struct{} - started time.Time - waitFn func() (int32, bool, error) - killFn func(uint32) error + issue domain.UUID + identity supervisor.EffectiveIdentity + cmd *exec.Cmd + pid uint32 + stdin io.WriteCloser + stdout io.ReadCloser + stderr io.ReadCloser + outputs chan outputResult + done chan struct{} + started time.Time + waitFn func() (int32, bool, error) + killFn func(uint32) error + snapshotFn func() (supervisor.ResourceSnapshot, error) mu sync.Mutex finished bool @@ -219,7 +235,7 @@ func (process *execProcess) startReaders(maxChunk uint64, remove func()) { } func materializeWrapper(directory string, wrapper Wrapper, now time.Time) (string, func(), error) { - if directory == "" || !now.IsZero() && now.Location() == nil { + if directory == "" { return "", nil, ErrInvalidWorkingDirectory } if err := os.MkdirAll(directory, 0o700); err != nil { @@ -441,7 +457,11 @@ func (manager *execSupervisor) startCommand(ctx context.Context, spec supervisor } func (manager *execSupervisor) registerProcess(issue domain.UUID, identity supervisor.EffectiveIdentity, command *exec.Cmd, stdin io.WriteCloser, stdout, stderr io.ReadCloser, started time.Time, waitFn func() (int32, bool, error), killFn func(uint32) error, cleanup func()) *execProcess { - process := &execProcess{issue: issue, identity: identity, cmd: command, stdin: stdin, stdout: stdout, stderr: stderr, outputs: make(chan outputResult, 32), done: make(chan struct{}), started: started, waitFn: waitFn, killFn: killFn} + var pid uint32 + if command != nil && command.Process != nil && command.Process.Pid > 0 { + pid = uint32(command.Process.Pid) + } + process := &execProcess{issue: issue, identity: identity, cmd: command, pid: pid, stdin: stdin, stdout: stdout, stderr: stderr, outputs: make(chan outputResult, 32), done: make(chan struct{}), started: started, waitFn: waitFn, killFn: killFn} manager.mu.Lock() manager.active[issue] = process manager.mu.Unlock() diff --git a/internal/client/supervisor/windows/native_windows.go b/internal/client/supervisor/windows/native_windows.go index 631b862..f7e5c90 100644 --- a/internal/client/supervisor/windows/native_windows.go +++ b/internal/client/supervisor/windows/native_windows.go @@ -13,6 +13,7 @@ import ( "fmt" "os" "os/exec" + "strings" "sync" "syscall" "time" @@ -27,11 +28,18 @@ const ( logon32LogonService = 5 logon32ProviderDefault = 0 securitySystemRID = "S-1-5-18" + disableMaxPrivilege = 0x1 ) var ( - advapi32 = syscall.NewLazyDLL("advapi32.dll") - procLogonUserW = advapi32.NewProc("LogonUserW") + advapi32 = syscall.NewLazyDLL("advapi32.dll") + procLogonUserW = advapi32.NewProc("LogonUserW") + procCreateRestrictedToken = advapi32.NewProc("CreateRestrictedToken") + kernel32 = syscall.NewLazyDLL("kernel32.dll") + procAttachConsole = kernel32.NewProc("AttachConsole") + procFreeConsole = kernel32.NewProc("FreeConsole") + procGenerateCtrlEvent = kernel32.NewProc("GenerateConsoleCtrlEvent") + procSetCtrlHandler = kernel32.NewProc("SetConsoleCtrlHandler") ) func stdinLineEnding() []byte { return []byte{'\r', '\n'} } @@ -130,6 +138,10 @@ func (manager *execSupervisor) Start(ctx context.Context, spec supervisor.StartS closeFiles() return fail(fmt.Errorf("create command Job: %w", err)) } + if err := applyJobProfiles(job, manager.options.JobProfiles, spec.ExecutionProfiles); err != nil { + _ = winapi.CloseHandle(job) + return fail(err) + } cleanupJob := true defer func() { if cleanupJob { @@ -214,6 +226,9 @@ func (manager *execSupervisor) Start(ctx context.Context, spec supervisor.StartS return winapi.TerminateJobObject(job, code) } process := manager.registerProcess(spec.IssueUUID, identity, command, stdinWrite, stdoutRead, stderrRead, started, waitFn, killFn, cleanup) + process.snapshotFn = func() (supervisor.ResourceSnapshot, error) { + return queryJobSnapshot(job, manager.options.Now()) + } return process, nil } @@ -280,6 +295,120 @@ func createKillOnCloseJob() (winapi.Handle, error) { return job, nil } +// applyJobProfiles combines the requested dimensions and applies them before +// process creation. The profile names originate from the validated protobuf +// ExecutionSpec; unknown names and any required control without a native +// implementation are permanent pre-launch failures. +func applyJobProfiles(job winapi.Handle, configured map[string]JobProfile, requested []string) error { + if len(requested) == 0 { + return nil + } + var combined JobProfile + for _, name := range requested { + profile, ok := configured[name] + if !ok { + return fmt.Errorf("%w: execution profile %q is not configured", supervisor.ErrUnsupported, name) + } + for _, required := range profile.RequiredControls { + switch required { + case "cpu", "memory", "pids": + case "io": + return fmt.Errorf("%w: Windows Job I/O rate control is not available in this build", supervisor.ErrUnsupported) + default: + return fmt.Errorf("%w: unknown required Job control %q", supervisor.ErrUnsupported, required) + } + } + if profile.CPUPercent > combined.CPUPercent { + combined.CPUPercent = profile.CPUPercent + } + if profile.MemoryMaxBytes > 0 && (combined.MemoryMaxBytes == 0 || profile.MemoryMaxBytes < combined.MemoryMaxBytes) { + combined.MemoryMaxBytes = profile.MemoryMaxBytes + } + if profile.PIDsMax > 0 && (combined.PIDsMax == 0 || profile.PIDsMax < combined.PIDsMax) { + combined.PIDsMax = profile.PIDsMax + } + if profile.IOReadBPS > 0 || profile.IOWriteBPS > 0 { + return fmt.Errorf("%w: Windows Job I/O rate control is not available in this build", supervisor.ErrUnsupported) + } + } + limits := winapi.JOBOBJECT_EXTENDED_LIMIT_INFORMATION{} + limits.BasicLimitInformation.LimitFlags = winapi.JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE + if combined.MemoryMaxBytes > 0 { + if uint64(uintptr(combined.MemoryMaxBytes)) != combined.MemoryMaxBytes { + return fmt.Errorf("%w: memory profile exceeds native pointer size", supervisor.ErrUnsupported) + } + limits.ProcessMemoryLimit = uintptr(combined.MemoryMaxBytes) + limits.BasicLimitInformation.LimitFlags |= winapi.JOB_OBJECT_LIMIT_PROCESS_MEMORY + } + if combined.PIDsMax > 0 { + if combined.PIDsMax > uint64(^uint32(0)) { + return fmt.Errorf("%w: process-count profile exceeds Windows limit", supervisor.ErrUnsupported) + } + limits.BasicLimitInformation.ActiveProcessLimit = uint32(combined.PIDsMax) + limits.BasicLimitInformation.LimitFlags |= winapi.JOB_OBJECT_LIMIT_ACTIVE_PROCESS + } + if _, err := winapi.SetInformationJobObject(job, winapi.JobObjectExtendedLimitInformation, uintptr(unsafe.Pointer(&limits)), uint32(unsafe.Sizeof(limits))); err != nil { + return fmt.Errorf("apply Windows Job limits: %w", err) + } + if combined.CPUPercent > 0 { + // The config contract expresses CPU allowance as a percentage of one + // logical CPU (for example 200 means two logical CPUs). Windows Job + // CpuRate is hundredths of a percentage of the whole machine, so scale + // by the active processor count before applying the hard cap. A profile + // larger than this host is intentionally capped at the host capacity, + // which means it imposes no additional CPU restriction but remains a + // valid, atomically verified profile. + processors := uint64(winapi.GetActiveProcessorCount(winapi.ALL_PROCESSOR_GROUPS)) + if processors == 0 { + return fmt.Errorf("%w: Windows did not report an active processor count", supervisor.ErrUnsupported) + } + if combined.CPUPercent > (^uint64(0)-processors+1)/100 { + return fmt.Errorf("%w: CPU profile %d%% overflows Windows Job rate conversion", supervisor.ErrUnsupported, combined.CPUPercent) + } + cpuRate := (combined.CPUPercent*100 + processors - 1) / processors + if cpuRate > 10000 { + cpuRate = 10000 + } + cpu := struct { + ControlFlags uint32 + CPUrate uint32 + Weight uint32 + }{ControlFlags: 0x1 | 0x4 /* ENABLE | HARD_CAP */, CPUrate: uint32(cpuRate)} + if _, err := winapi.SetInformationJobObject(job, winapi.JobObjectCpuRateControlInformation, uintptr(unsafe.Pointer(&cpu)), uint32(unsafe.Sizeof(cpu))); err != nil { + return fmt.Errorf("apply Windows Job CPU limit: %w", err) + } + var cpuReadback struct { + ControlFlags uint32 + CPUrate uint32 + Weight uint32 + } + var cpuReturned uint32 + if err := winapi.QueryInformationJobObject(job, int32(winapi.JobObjectCpuRateControlInformation), uintptr(unsafe.Pointer(&cpuReadback)), uint32(unsafe.Sizeof(cpuReadback)), &cpuReturned); err != nil { + return fmt.Errorf("verify Windows Job CPU limit: %w", err) + } + if cpuReadback.CPUrate != cpu.CPUrate || cpuReadback.ControlFlags&0x5 != 0x5 { + return errors.New("Windows Job CPU limit did not read back as requested") + } + } + // Read back every requested limit before authorization. This catches + // policy restrictions and unsupported Job implementations early. + var readback winapi.JOBOBJECT_EXTENDED_LIMIT_INFORMATION + var returned uint32 + if err := winapi.QueryInformationJobObject(job, int32(winapi.JobObjectExtendedLimitInformation), uintptr(unsafe.Pointer(&readback)), uint32(unsafe.Sizeof(readback)), &returned); err != nil { + return fmt.Errorf("verify Windows Job limits: %w", err) + } + if readback.BasicLimitInformation.LimitFlags&winapi.JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE == 0 { + return errors.New("Windows Job lost kill-on-close protection") + } + if combined.MemoryMaxBytes > 0 && uint64(readback.ProcessMemoryLimit) != combined.MemoryMaxBytes { + return errors.New("Windows Job memory limit did not read back as requested") + } + if combined.PIDsMax > 0 && uint64(readback.BasicLimitInformation.ActiveProcessLimit) != combined.PIDsMax { + return errors.New("Windows Job process limit did not read back as requested") + } + return nil +} + func (manager *execSupervisor) selectToken(elevated bool) (winapi.Token, supervisor.EffectiveIdentity, error) { candidates, err := DiscoverActiveSessions() if err != nil { @@ -288,11 +417,39 @@ func (manager *execSupervisor) selectToken(elevated bool) (winapi.Token, supervi candidates = nil } selection := Select(SelectionInput{Elevated: elevated, ActiveSessions: candidates, ActiveSystemAvailable: true, LocalServiceAvailable: true, LocalSystemAvailable: true}) + attempted := make([]string, 0, len(selection.Attempts)+1) + details := make([]string, 0, len(selection.Attempts)+1) + addAttempt := func(contextName ExecutionContext, detail string) { + for _, existing := range attempted { + if existing == string(contextName) { + if detail != "" { + details = append(details, string(contextName)+": "+detail) + } + return + } + } + attempted = append(attempted, string(contextName)) + if detail != "" { + details = append(details, string(contextName)+": "+detail) + } + } + for _, attempt := range selection.Attempts { + addAttempt(attempt.Context, string(attempt.Reason)) + } + withEvidence := func(identity supervisor.EffectiveIdentity) supervisor.EffectiveIdentity { + identity.AttemptedContexts = append([]string(nil), attempted...) + identity.SelectionDetail = boundSelectionDetail(strings.Join(details, "; ")) + return identity + } + rejection := func(cause error) error { + identity := withEvidence(supervisor.EffectiveIdentity{}) + return &supervisor.StartError{Cause: cause, WindowsIdentity: identity.WindowsIdentity()} + } if selection.Effective == nil { if selection.Error != nil { - return 0, supervisor.EffectiveIdentity{}, selection.Error + return 0, supervisor.EffectiveIdentity{}, rejection(selection.Error) } - return 0, supervisor.EffectiveIdentity{}, errors.New("Windows execution context selection failed") + return 0, supervisor.EffectiveIdentity{}, rejection(errors.New("Windows execution context selection failed")) } var selected *SessionCandidate if selection.Effective.SessionID != nil { @@ -306,10 +463,11 @@ func (manager *execSupervisor) selectToken(elevated bool) (winapi.Token, supervi for _, attempt := range selection.Attempts { token, identity, err := openTokenForAttempt(attempt.Context, selected) if err == nil { - return token, identity, nil + return token, withEvidence(identity), nil } + addAttempt(attempt.Context, "native preparation failed: "+err.Error()) if !elevated { - return 0, supervisor.EffectiveIdentity{}, err + return 0, supervisor.EffectiveIdentity{}, rejection(err) } } // The pure selector stops as soon as ACTIVE_SYSTEM is available. A native @@ -317,11 +475,22 @@ func (manager *execSupervisor) selectToken(elevated bool) (winapi.Token, supervi // removed), so the final LOCAL_SYSTEM fallback is attempted here before // launch preparation, never by retrying a created process. if elevated { + addAttempt(ContextLocalSystem, "fallback") if token, identity, err := openTokenForAttempt(ContextLocalSystem, nil); err == nil { - return token, identity, nil + return token, withEvidence(identity), nil + } else { + addAttempt(ContextLocalSystem, "native preparation failed: "+err.Error()) + return 0, supervisor.EffectiveIdentity{}, rejection(err) } } - return 0, supervisor.EffectiveIdentity{}, errors.New("all Windows execution contexts failed before launch preparation") + return 0, supervisor.EffectiveIdentity{}, rejection(errors.New("all Windows execution contexts failed before launch preparation")) +} + +func boundSelectionDetail(detail string) string { + if len(detail) <= 4096 { + return detail + } + return detail[:4096] } func openTokenForAttempt(contextName ExecutionContext, candidate *SessionCandidate) (winapi.Token, supervisor.EffectiveIdentity, error) { @@ -350,12 +519,27 @@ func openTokenForAttempt(contextName ExecutionContext, candidate *SessionCandida return 0, supervisor.EffectiveIdentity{}, err } identity.Context = string(ContextActiveSystem) + identity.SessionUserSID = candidate.UserSID + identity.LogonSID = candidate.LogonSID return serviceToken, identity, nil } else if token.IsElevated() { + // A full administrator token can be returned when UAC is disabled or + // policy supplies an already-unfiltered token. Normal commands must + // still run as that user without administrator authority; create a + // restricted medium token instead of silently falling back to a + // service identity. + restricted, err := createRestrictedMediumToken(token) _ = token.Close() - return 0, supervisor.EffectiveIdentity{}, errors.New("active-user token is elevated and no restricted medium token was available") + if err != nil { + return 0, supervisor.EffectiveIdentity{}, err + } + token = restricted } - identity := supervisor.EffectiveIdentity{Context: string(contextName), SessionID: candidate.SessionID, UserSID: candidate.UserSID, Elevated: contextName != ContextActiveUser, Integrity: map[bool]string{true: "high", false: "medium"}[contextName != ContextActiveUser]} + if err := verifyUserToken(token, candidate, contextName == ContextActiveUser); err != nil { + _ = token.Close() + return 0, supervisor.EffectiveIdentity{}, err + } + identity := supervisor.EffectiveIdentity{Context: string(contextName), SessionID: candidate.SessionID, SessionUserSID: candidate.UserSID, UserSID: candidate.UserSID, LogonSID: candidate.LogonSID, Elevated: contextName != ContextActiveUser, Integrity: map[bool]string{true: "high", false: "medium"}[contextName != ContextActiveUser]} return token, identity, nil case ContextLocalService: token, err := logonLocalService() @@ -370,6 +554,81 @@ func openTokenForAttempt(contextName ExecutionContext, candidate *SessionCandida } } +// createRestrictedMediumToken turns a full administrator token into the +// normal active-user token required for elevated=false. It disables all +// privileges, disables the built-in Administrators SID, and verifies medium +// integrity before the token is returned to the launch path. +func createRestrictedMediumToken(source winapi.Token) (winapi.Token, error) { + adminSID, err := winapi.CreateWellKnownSid(winapi.WinBuiltinAdministratorsSid) + if err != nil { + return 0, err + } + disabled := winapi.SIDAndAttributes{Sid: adminSID} + var restricted winapi.Token + result, _, callErr := procCreateRestrictedToken.Call( + uintptr(source), disableMaxPrivilege, + 1, uintptr(unsafe.Pointer(&disabled)), + 0, 0, + 0, 0, + uintptr(unsafe.Pointer(&restricted)), + ) + if result == 0 { + if callErr != syscall.Errno(0) { + return 0, callErr + } + return 0, syscall.GetLastError() + } + if err := setMediumIntegrity(restricted); err != nil { + _ = restricted.Close() + return 0, err + } + if restricted.IsElevated() { + _ = restricted.Close() + return 0, errors.New("restricted active-user token remained elevated") + } + return restricted, nil +} + +func setMediumIntegrity(token winapi.Token) error { + mediumSID, err := winapi.StringToSid("S-1-16-8192") + if err != nil { + return err + } + sidLength := winapi.GetLengthSid(mediumSID) + headerSize := uint32(unsafe.Sizeof(winapi.Tokenmandatorylabel{})) + buffer := make([]byte, headerSize+sidLength) + label := (*winapi.Tokenmandatorylabel)(unsafe.Pointer(&buffer[0])) + label.Label.Sid = (*winapi.SID)(unsafe.Pointer(&buffer[headerSize])) + label.Label.Attributes = winapi.SE_GROUP_INTEGRITY | winapi.SE_GROUP_INTEGRITY_ENABLED + copy(buffer[headerSize:], unsafe.Slice((*byte)(unsafe.Pointer(mediumSID)), sidLength)) + return winapi.SetTokenInformation(token, winapi.TokenIntegrityLevel, &buffer[0], uint32(len(buffer))) +} + +func verifyUserToken(token winapi.Token, candidate *SessionCandidate, normal bool) error { + user, err := token.GetTokenUser() + if err != nil || user.User.Sid == nil { + if err != nil { + return err + } + return errors.New("active token has no user SID") + } + if user.User.Sid.String() != candidate.UserSID { + return errors.New("active token user SID changed during launch selection") + } + var sessionID uint32 + var returned uint32 + if err := winapi.GetTokenInformation(token, winapi.TokenSessionId, (*byte)(unsafe.Pointer(&sessionID)), uint32(unsafe.Sizeof(sessionID)), &returned); err != nil { + return err + } + if returned != uint32(unsafe.Sizeof(sessionID)) || sessionID != candidate.SessionID { + return errors.New("active token session changed during launch selection") + } + if normal && token.IsElevated() { + return errors.New("normal active-user token is elevated") + } + return nil +} + func duplicateServiceToken() (winapi.Token, supervisor.EffectiveIdentity, error) { return duplicateServiceTokenForSession(0) } @@ -427,19 +686,66 @@ func (manager *execSupervisor) Signal(ctx context.Context, process supervisor.Pr return supervisor.SignalOutcome{}, errors.New("unsupported signal") } if signal == supervisor.SignalTerm { - // The command has its own hidden console. A full AttachConsole/control - // helper is intentionally isolated from the Job kill path; if it is not - // available, the bounded grace period ends in an explicit Job kill. + // Each command has its own hidden console. The helper path is kept in + // this short-lived call and is deliberately best-effort: a session that + // has already exited or a policy that denies AttachConsole is recorded, + // then the bounded grace period ends in an explicit Job kill. + breakDelivered, breakErr := sendControlBreak(native.pid) + if breakErr != nil && ctx.Err() != nil { + return supervisor.SignalOutcome{}, ctx.Err() + } + if breakDelivered { + select { + case <-native.done: + return supervisor.SignalOutcome{Delivered: true, Detail: "CTRL_BREAK delivered", ObservedAt: manager.options.Now()}, nil + default: + } + } select { case <-ctx.Done(): return supervisor.SignalOutcome{}, ctx.Err() + case <-native.done: + return supervisor.SignalOutcome{Delivered: breakDelivered, Detail: "command exited after TERM", ObservedAt: manager.options.Now()}, nil case <-time.After(manager.options.WindowsTermGrace): } } if err := native.killFn(1); err != nil { return supervisor.SignalOutcome{}, err } - return supervisor.SignalOutcome{Delivered: true, Escalated: signal == supervisor.SignalTerm, Detail: "Windows Job terminated", ObservedAt: manager.options.Now()}, nil + detail := "Windows Job terminated" + if signal == supervisor.SignalTerm { + detail = "CTRL_BREAK grace expired; Windows Job terminated" + } + return supervisor.SignalOutcome{Delivered: true, Escalated: signal == supervisor.SignalTerm, Detail: detail, ObservedAt: manager.options.Now()}, nil +} + +// sendControlBreak is the native equivalent of the signal-helper mode. The +// production helper is normally a separate short-lived rvbox.exe invocation; +// this direct implementation keeps the same verified PID/console boundary +// for the first service build and never addresses a process by a caller- +// supplied PID. The PID comes only from execProcess metadata. +func sendControlBreak(pid uint32) (bool, error) { + if pid == 0 { + return false, errors.New("command has no verified console PID") + } + if result, _, err := procAttachConsole.Call(uintptr(pid)); result == 0 { + if err == syscall.Errno(0) { + err = syscall.GetLastError() + } + return false, err + } + defer procFreeConsole.Call() + // Prevent the service/helper itself from acting on the generated event. + procSetCtrlHandler.Call(0, 1) + defer procSetCtrlHandler.Call(0, 0) + result, _, err := procGenerateCtrlEvent.Call(1 /* CTRL_BREAK_EVENT */, 0) + if result == 0 { + if err == syscall.Errno(0) { + err = syscall.GetLastError() + } + return false, err + } + return true, nil } func (manager *execSupervisor) Snapshot(ctx context.Context, process supervisor.Process) (supervisor.ResourceSnapshot, error) { @@ -450,6 +756,14 @@ func (manager *execSupervisor) Snapshot(ctx context.Context, process supervisor. if !ok || native.cmd == nil || native.cmd.Process == nil { return supervisor.ResourceSnapshot{}, ErrProcessNotFound } + if native.snapshotFn != nil { + select { + case <-ctx.Done(): + return supervisor.ResourceSnapshot{}, ctx.Err() + default: + } + return native.snapshotFn() + } select { case <-ctx.Done(): return supervisor.ResourceSnapshot{}, ctx.Err() @@ -458,6 +772,39 @@ func (manager *execSupervisor) Snapshot(ctx context.Context, process supervisor. return supervisor.ResourceSnapshot{ProcessCount: 1, ObservedAt: manager.options.Now(), Complete: false, Detail: "Windows Job accounting is available after native completion integration"}, nil } +type jobBasicAndIOAccounting struct { + TotalUserTime int64 + TotalKernelTime int64 + ThisPeriodTotalUserTime int64 + ThisPeriodTotalKernelTime int64 + TotalPageFaultCount uint32 + TotalProcesses uint32 + ActiveProcesses uint32 + TotalTerminatedProcesses uint32 + IO winapi.IO_COUNTERS +} + +func queryJobSnapshot(job winapi.Handle, now time.Time) (supervisor.ResourceSnapshot, error) { + if job == 0 || job == winapi.InvalidHandle { + return supervisor.ResourceSnapshot{}, ErrProcessNotFound + } + var accounting jobBasicAndIOAccounting + var returned uint32 + if err := winapi.QueryInformationJobObject(job, int32(winapi.JobObjectBasicAndIoAccountingInformation), uintptr(unsafe.Pointer(&accounting)), uint32(unsafe.Sizeof(accounting)), &returned); err != nil { + return supervisor.ResourceSnapshot{}, err + } + var limits winapi.JOBOBJECT_EXTENDED_LIMIT_INFORMATION + if err := winapi.QueryInformationJobObject(job, int32(winapi.JobObjectExtendedLimitInformation), uintptr(unsafe.Pointer(&limits)), uint32(unsafe.Sizeof(limits)), &returned); err != nil { + return supervisor.ResourceSnapshot{}, err + } + userKernel := accounting.TotalUserTime + accounting.TotalKernelTime + var cpu time.Duration + if userKernel > 0 && userKernel <= int64(^uint64(0)>>1)/100 { + cpu = time.Duration(userKernel) * 100 * time.Nanosecond + } + return supervisor.ResourceSnapshot{CPUTime: cpu, ResidentBytes: uint64(limits.PeakJobMemoryUsed), IOReadBytes: accounting.IO.ReadTransferCount, IOWriteBytes: accounting.IO.WriteTransferCount, ProcessCount: uint64(accounting.ActiveProcesses), ObservedAt: now, Complete: accounting.ActiveProcesses == 0, Detail: "Windows Job accounting"}, nil +} + func (manager *execSupervisor) StopAll(_ context.Context) error { manager.mu.Lock() processes := make([]*execProcess, 0, len(manager.active)) diff --git a/internal/client/windowsservice/service.go b/internal/client/windowsservice/service.go index fcc2053..85967de 100644 --- a/internal/client/windowsservice/service.go +++ b/internal/client/windowsservice/service.go @@ -119,6 +119,8 @@ func Transition(state State, command Command) (State, error) { // platform implementations return ErrUnsupported on non-Windows builds. func Install(spec InstallSpec) error { return installNative(spec) } func Uninstall() error { return uninstallNative() } +func Configure(startup StartupMode) error { return configureNative(startup) } func Start() error { return startNative() } func Stop(timeoutSeconds uint32) error { return stopNative(timeoutSeconds) } +func Restart(timeoutSeconds uint32) error { return restartNative(timeoutSeconds) } func Run(run func(context.Context) error) error { return runNative(run) } diff --git a/internal/client/windowsservice/service_other.go b/internal/client/windowsservice/service_other.go index d43ef20..c675554 100644 --- a/internal/client/windowsservice/service_other.go +++ b/internal/client/windowsservice/service_other.go @@ -6,6 +6,8 @@ import "context" func installNative(InstallSpec) error { return ErrUnsupported } func uninstallNative() error { return ErrUnsupported } +func configureNative(StartupMode) error { return ErrUnsupported } func startNative() error { return ErrUnsupported } func stopNative(uint32) error { return ErrUnsupported } +func restartNative(uint32) error { return ErrUnsupported } func runNative(func(context.Context) error) error { return ErrUnsupported } diff --git a/internal/client/windowsservice/service_windows.go b/internal/client/windowsservice/service_windows.go index a4234dd..c33fdac 100644 --- a/internal/client/windowsservice/service_windows.go +++ b/internal/client/windowsservice/service_windows.go @@ -10,6 +10,7 @@ import ( "time" "golang.org/x/sys/windows" + "golang.org/x/sys/windows/registry" "golang.org/x/sys/windows/svc" "golang.org/x/sys/windows/svc/mgr" ) @@ -67,6 +68,9 @@ func installNative(spec InstallSpec) error { if err := service.Start(); err != nil && !errors.Is(err, windows.ERROR_SERVICE_ALREADY_RUNNING) { return fmt.Errorf("start %s service: %w", Name, err) } + if err := registerTray(spec); err != nil { + return fmt.Errorf("register per-user tray: %w", err) + } return nil } @@ -75,6 +79,7 @@ func serviceImage(spec InstallSpec) string { } func uninstallNative() error { + var serviceErr error manager, err := connect() if err != nil { return fmt.Errorf("connect to service control manager: %w", err) @@ -82,17 +87,79 @@ func uninstallNative() error { defer manager.Disconnect() service, err := manager.OpenService(Name) if errors.Is(err, windows.ERROR_SERVICE_DOES_NOT_EXIST) { - return nil + serviceErr = nil + } else if err != nil { + return fmt.Errorf("open %s service: %w", Name, err) + } else { + defer service.Close() + if err := stopService(service, 30*time.Second); err != nil { + return err + } + if err := service.Delete(); err != nil && !errors.Is(err, windows.ERROR_SERVICE_MARKED_FOR_DELETE) { + return fmt.Errorf("delete %s service: %w", Name, err) + } + } + if err := removeTrayRegistration(); err != nil { + return fmt.Errorf("remove per-user tray: %w", err) + } + return serviceErr +} + +func configureNative(startup StartupMode) error { + if startup != StartupAutomatic && startup != StartupManual { + return ErrInvalidInstallSpec + } + manager, err := connect() + if err != nil { + return fmt.Errorf("connect to service control manager: %w", err) + } + defer manager.Disconnect() + service, err := manager.OpenService(Name) + if errors.Is(err, windows.ERROR_SERVICE_DOES_NOT_EXIST) { + return fmt.Errorf("%s service is not installed", Name) } if err != nil { return fmt.Errorf("open %s service: %w", Name, err) } defer service.Close() - if err := stopService(service, 30*time.Second); err != nil { + configuration, err := service.Config() + if err != nil { + return fmt.Errorf("query %s service configuration: %w", Name, err) + } + if startup == StartupAutomatic { + configuration.StartType = mgr.StartAutomatic + } else { + configuration.StartType = mgr.StartManual + } + if err := service.UpdateConfig(configuration); err != nil { + return fmt.Errorf("update %s startup type: %w", Name, err) + } + return nil +} + +const trayRunValue = "RVBoxTray" + +func registerTray(spec InstallSpec) error { + key, _, err := registry.CreateKey(registry.LOCAL_MACHINE, `SOFTWARE\Microsoft\Windows\CurrentVersion\Run`, registry.SET_VALUE) + if err != nil { return err } - if err := service.Delete(); err != nil && !errors.Is(err, windows.ERROR_SERVICE_MARKED_FOR_DELETE) { - return fmt.Errorf("delete %s service: %w", Name, err) + defer key.Close() + image := syscall.EscapeArg(spec.ExecutablePath) + " --tray --config " + syscall.EscapeArg(spec.ConfigPath) + return key.SetStringValue(trayRunValue, image) +} + +func removeTrayRegistration() error { + key, err := registry.OpenKey(registry.LOCAL_MACHINE, `SOFTWARE\Microsoft\Windows\CurrentVersion\Run`, registry.SET_VALUE) + if errors.Is(err, registry.ErrNotExist) { + return nil + } + if err != nil { + return err + } + defer key.Close() + if err := key.DeleteValue(trayRunValue); err != nil && !errors.Is(err, registry.ErrNotExist) { + return err } return nil } @@ -138,6 +205,13 @@ func stopNative(timeoutSeconds uint32) error { return stopService(service, timeout) } +func restartNative(timeoutSeconds uint32) error { + if err := stopNative(timeoutSeconds); err != nil { + return err + } + return startNative() +} + func stopService(service *mgr.Service, timeout time.Duration) error { status, err := service.Query() if err != nil { diff --git a/internal/client/windowstray/protocol.go b/internal/client/windowstray/protocol.go index 8bb99a1..868420a 100644 --- a/internal/client/windowstray/protocol.go +++ b/internal/client/windowstray/protocol.go @@ -5,12 +5,25 @@ package windowstray import ( "bytes" + "context" "encoding/binary" "errors" "fmt" "unicode/utf8" ) +// PipeName is the machine-local service endpoint used by every tray session. +// The native implementation creates it with PIPE_REJECT_REMOTE_CLIENTS and +// an explicit SYSTEM/Administrators/interactive-user ACL. Keeping the name +// here (rather than deriving it from user input) prevents cross-session and +// path-confusion bugs. +const PipeName = `\\.\pipe\RVBoxClientTrayV1` + +// Handler is invoked by the service after the native adapter has verified the +// connecting process token, SID, and session. A response is always encoded +// as ActionStatus; requests other than status deliberately carry no payload. +type Handler func(context.Context, Peer, Frame) (Frame, error) + const ( protocolVersion uint16 = 1 maxFrameBytes = 64 << 10 @@ -33,6 +46,8 @@ const ( ActionStartService ActionStopService ActionRestartService + ActionSetAutomatic + ActionSetManual ActionExitTray ) @@ -115,7 +130,7 @@ func Authorize(peer Peer, action Action) error { switch action { case ActionStatus, ActionOpenConfig, ActionOpenLog, ActionExitTray: return nil - case ActionStartService, ActionStopService, ActionRestartService: + case ActionStartService, ActionStopService, ActionRestartService, ActionSetAutomatic, ActionSetManual: if peer.Administrator || peer.System { return nil } diff --git a/internal/client/windowstray/service_other.go b/internal/client/windowstray/service_other.go new file mode 100644 index 0000000..5d6f259 --- /dev/null +++ b/internal/client/windowstray/service_other.go @@ -0,0 +1,20 @@ +//go:build !windows + +package windowstray + +import ( + "context" + "errors" + "io" +) + +var ErrUnsupported = errors.New("Windows tray IPC is unavailable on this platform") + +// Serve exists on every platform so the service wiring stays build-tag +// neutral. Unix-like client support is deferred in v1; this endpoint must +// never silently become a second client daemon. +func Serve(context.Context, Handler) error { return ErrUnsupported } + +func Request(context.Context, Frame) (Frame, error) { return Frame{}, ErrUnsupported } + +func Run(context.Context, io.Writer) error { return ErrUnsupported } diff --git a/internal/client/windowstray/service_windows.go b/internal/client/windowstray/service_windows.go new file mode 100644 index 0000000..41c0ec5 --- /dev/null +++ b/internal/client/windowstray/service_windows.go @@ -0,0 +1,242 @@ +//go:build windows + +package windowstray + +// This file is the small native service endpoint. It intentionally uses one +// request per pipe connection: the tray is a presentation client, not a +// long-lived command channel, and a bounded connection makes cancellation and +// peer verification straightforward. The service never hands the tray a +// store handle or command payload. + +import ( + "context" + "errors" + "fmt" + "io" + "os" + "time" + "unsafe" + + winapi "golang.org/x/sys/windows" +) + +const ( + pipeBufferBytes = 64 << 10 + pipeInstances = 8 +) + +var ( + ErrUnsupported = errors.New("Windows tray IPC is unavailable on this platform") + pipeSDDL = "D:P(A;;GA;;;SY)(A;;GA;;;BA)(A;;GRGW;;;IU)" +) + +// Serve accepts bounded tray requests until ctx is cancelled. It is safe to +// run before any interactive user logs in; in that state no client can pass +// the interactive-peer authorization check. +func Serve(ctx context.Context, handler Handler) error { + if handler == nil { + return errors.New("tray handler is required") + } + for { + pipe, err := newTrayPipe() + if err != nil { + return fmt.Errorf("create tray pipe: %w", err) + } + connected := make(chan error, 1) + go func() { connected <- winapi.ConnectNamedPipe(winapi.Handle(pipe.Fd()), nil) }() + select { + case <-ctx.Done(): + _ = pipe.Close() + return nil + case err := <-connected: + if err != nil && !errors.Is(err, winapi.ERROR_PIPE_CONNECTED) { + _ = pipe.Close() + if ctx.Err() != nil { + return nil + } + continue + } + go serveTrayPipe(ctx, pipe, handler) + } + } +} + +func newTrayPipe() (*os.File, error) { + name, err := winapi.UTF16PtrFromString(PipeName) + if err != nil { + return nil, err + } + descriptor, err := winapi.SecurityDescriptorFromString(pipeSDDL) + if err != nil { + return nil, err + } + attributes := &winapi.SecurityAttributes{ + Length: uint32(unsafe.Sizeof(winapi.SecurityAttributes{})), + SecurityDescriptor: descriptor, + } + mode := uint32(winapi.PIPE_ACCESS_DUPLEX | winapi.PIPE_TYPE_MESSAGE | winapi.PIPE_READMODE_MESSAGE | winapi.PIPE_WAIT | winapi.PIPE_REJECT_REMOTE_CLIENTS | winapi.SECURITY_IDENTIFICATION) + handle, err := winapi.CreateNamedPipe(name, mode, pipeInstances, pipeBufferBytes, pipeBufferBytes, 0, 0, attributes) + if err != nil { + return nil, err + } + return os.NewFile(uintptr(handle), "rvbox-tray-pipe"), nil +} + +func serveTrayPipe(ctx context.Context, pipe *os.File, handler Handler) { + if pipe == nil { + return + } + defer pipe.Close() + // A blocked Read must be interrupted when service shutdown cancels ctx. + readDone := make(chan struct{}) + go func() { + select { + case <-ctx.Done(): + _ = pipe.Close() + case <-readDone: + } + }() + defer close(readDone) + peer, err := peerFromPipe(winapi.Handle(pipe.Fd())) + if err != nil { + writeTrayResponse(pipe, err) + return + } + request, err := readTrayFrame(pipe) + if err != nil { + writeTrayResponse(pipe, err) + return + } + if err := Authorize(peer, request.Action); err != nil { + writeTrayResponse(pipe, err) + return + } + response, err := handler(ctx, peer, request) + if err != nil { + writeTrayResponse(pipe, err) + return + } + if response.Action == 0 { + response.Action = ActionStatus + } + if response.Action != ActionStatus { + response = Frame{Action: ActionStatus} + } + writeTrayResponse(pipe, response) +} + +func readTrayFrame(reader io.Reader) (Frame, error) { + buffer := make([]byte, maxFrameBytes) + count, err := reader.Read(buffer) + if err != nil { + return Frame{}, err + } + if count == len(buffer) { + return Frame{}, ErrFrameTooLarge + } + return Decode(buffer[:count]) +} + +func writeTrayResponse(writer *os.File, value any) { + frame := Frame{Action: ActionStatus} + switch response := value.(type) { + case Frame: + frame = response + case error: + message := response.Error() + if len(message) > maxPayloadBytes { + message = message[:maxPayloadBytes] + } + frame.Payload = []byte("error: " + message) + } + encoded, err := Encode(frame) + if err != nil { + return + } + _, _ = writer.Write(encoded) + _ = winapi.FlushFileBuffers(winapi.Handle(writer.Fd())) +} + +func peerFromPipe(pipe winapi.Handle) (Peer, error) { + if pipe == 0 || pipe == winapi.InvalidHandle { + return Peer{}, ErrInvalidPeer + } + var pid uint32 + if err := winapi.GetNamedPipeClientProcessId(pipe, &pid); err != nil || pid == 0 { + return Peer{}, ErrInvalidPeer + } + process, err := winapi.OpenProcess(winapi.PROCESS_QUERY_LIMITED_INFORMATION, false, pid) + if err != nil { + return Peer{}, ErrInvalidPeer + } + defer winapi.CloseHandle(process) + var token winapi.Token + if err := winapi.OpenProcessToken(process, winapi.TOKEN_QUERY, &token); err != nil { + return Peer{}, ErrInvalidPeer + } + defer token.Close() + user, err := token.GetTokenUser() + if err != nil || user.User.Sid == nil { + return Peer{}, ErrInvalidPeer + } + var sessionID uint32 + var returned uint32 + if err := winapi.GetTokenInformation(token, winapi.TokenSessionId, (*byte)(unsafe.Pointer(&sessionID)), uint32(unsafe.Sizeof(sessionID)), &returned); err != nil || returned != uint32(unsafe.Sizeof(sessionID)) { + return Peer{}, ErrInvalidPeer + } + adminSID, err := winapi.CreateWellKnownSid(winapi.WinBuiltinAdministratorsSid) + if err != nil { + return Peer{}, ErrInvalidPeer + } + admin, err := token.IsMember(adminSID) + if err != nil { + return Peer{}, ErrInvalidPeer + } + sid := user.User.Sid.String() + return Peer{PID: pid, SessionID: sessionID, SID: sid, TokenVerified: true, Interactive: sessionID != 0, Administrator: admin, System: sid == "S-1-5-18"}, nil +} + +// Request opens exactly one local pipe connection and exchanges one frame. +// It retries only the transient ERROR_PIPE_BUSY state and never falls back to +// an arbitrary filesystem/socket path. +func Request(ctx context.Context, request Frame) (Frame, error) { + encoded, err := Encode(request) + if err != nil { + return Frame{}, err + } + var pipe *os.File + for { + if ctx.Err() != nil { + return Frame{}, ctx.Err() + } + name, nameErr := winapi.UTF16PtrFromString(PipeName) + if nameErr != nil { + return Frame{}, nameErr + } + handle, openErr := winapi.CreateFile(name, winapi.GENERIC_READ|winapi.GENERIC_WRITE, 0, nil, winapi.OPEN_EXISTING, 0, 0) + if openErr == nil { + pipe = os.NewFile(uintptr(handle), "rvbox-tray-client") + break + } + if !errors.Is(openErr, winapi.ERROR_PIPE_BUSY) { + return Frame{}, openErr + } + timer := time.NewTimer(100 * time.Millisecond) + select { + case <-ctx.Done(): + timer.Stop() + return Frame{}, ctx.Err() + case <-timer.C: + } + } + defer pipe.Close() + state := uint32(winapi.PIPE_READMODE_MESSAGE) + _ = winapi.SetNamedPipeHandleState(winapi.Handle(pipe.Fd()), &state, nil, nil) + if _, err := pipe.Write(encoded); err != nil { + return Frame{}, err + } + if err := winapi.FlushFileBuffers(winapi.Handle(pipe.Fd())); err != nil { + return Frame{}, err + } + return readTrayFrame(pipe) +} diff --git a/internal/client/windowstray/tray_windows.go b/internal/client/windowstray/tray_windows.go new file mode 100644 index 0000000..f030533 --- /dev/null +++ b/internal/client/windowstray/tray_windows.go @@ -0,0 +1,419 @@ +//go:build windows + +package windowstray + +// A deliberately small Win32 notification-area host. The tray has no +// durable state and no command execution path; it only renders a tooltip, +// sends enum actions over the authenticated service pipe, and opens paths +// returned by the service with the exact Windows Explorer executable. + +import ( + "context" + "errors" + "fmt" + "io" + "os" + "os/exec" + "path/filepath" + "strings" + "sync" + "syscall" + "unicode/utf8" + "unsafe" + + winapi "golang.org/x/sys/windows" +) + +const ( + wmDestroy = 0x0002 + wmClose = 0x0010 + wmCommand = 0x0111 + wmRButtonUp = 0x0205 + wmLButtonDblClick = 0x0203 + wmApp = 0x8000 + trayMessage = wmApp + 1 + trayIconID = 1 + trayNIMAdd = 0 + trayNIMModify = 1 + trayNIMDelete = 2 + trayNIFMessage = 0x00000001 + trayNIFIcon = 0x00000002 + trayNIFTip = 0x00000004 + trayTPMRightBtn = 0x0002 + trayMFString = 0x00000000 + trayMFSeparator = 0x00000800 + traySWHide = 0 + trayIDIApplication = 32512 + trayIDCArrow = 32512 +) + +var ( + ErrTrayAlreadyRunning = errors.New("RVBox tray is already running in this session") + user32Tray = syscall.NewLazyDLL("user32.dll") + shell32Tray = syscall.NewLazyDLL("shell32.dll") + procRegisterClassEx = user32Tray.NewProc("RegisterClassExW") + procCreateWindowEx = user32Tray.NewProc("CreateWindowExW") + procDefWindowProc = user32Tray.NewProc("DefWindowProcW") + procGetMessage = user32Tray.NewProc("GetMessageW") + procTranslate = user32Tray.NewProc("TranslateMessage") + procDispatch = user32Tray.NewProc("DispatchMessageW") + procPostMessage = user32Tray.NewProc("PostMessageW") + procDestroyWindow = user32Tray.NewProc("DestroyWindow") + procShowWindow = user32Tray.NewProc("ShowWindow") + procPostQuitMessage = user32Tray.NewProc("PostQuitMessage") + procLoadIcon = user32Tray.NewProc("LoadIconW") + procLoadCursor = user32Tray.NewProc("LoadCursorW") + procCreatePopup = user32Tray.NewProc("CreatePopupMenu") + procAppendMenu = user32Tray.NewProc("AppendMenuW") + procTrackPopup = user32Tray.NewProc("TrackPopupMenu") + procDestroyMenu = user32Tray.NewProc("DestroyMenu") + procGetCursorPos = user32Tray.NewProc("GetCursorPos") + procSetForeground = user32Tray.NewProc("SetForegroundWindow") + procGetModuleHandle = user32Tray.NewProc("GetModuleHandleW") + procShellNotify = shell32Tray.NewProc("Shell_NotifyIconW") + trayCallback = syscall.NewCallback(trayWindowProc) + trayWindows sync.Map // hwnd -> *trayWindow +) + +type trayPoint struct{ X, Y int32 } + +type trayMessageRecord struct { + HWnd uintptr + Message uint32 + WParam uintptr + LParam uintptr + Time uint32 + Point trayPoint +} + +type trayClass struct { + CbSize uint32 + Style uint32 + WndProc uintptr + CbClsExtra int32 + CbWndExtra int32 + HInstance uintptr + HIcon uintptr + HCursor uintptr + HbrBackground uintptr + MenuName *uint16 + ClassName *uint16 + HIconSm uintptr +} + +type trayIconData struct { + CbSize uint32 + HWnd uintptr + UID uint32 + UFlags uint32 + UCallbackMessage uint32 + HIcon uintptr + Tip [128]uint16 + State uint32 + StateMask uint32 + InfoFlags uint32 + InfoTitle [64]uint16 + InfoData [256]uint16 + Guid [16]byte + BalloonIcon uintptr +} + +type trayWindow struct { + hwnd uintptr + icon trayIconData + class *uint16 + tooltip string + output io.Writer + closeOnce sync.Once +} + +// Run starts one notification icon for the current logged-in session. It +// returns when the icon is closed, the service becomes unavailable, or ctx is +// cancelled; none of those outcomes affect the machine-wide service. +func Run(ctx context.Context, output io.Writer) error { + mutex, err := acquireTrayMutex() + if err != nil { + return err + } + defer winapi.CloseHandle(mutex) + response, err := Request(ctx, Frame{Action: ActionStatus}) + if err != nil { + return fmt.Errorf("connect to RVBox service: %w", err) + } + tooltip := string(response.Payload) + if tooltip == "" { + tooltip = "RVBox service" + } + ready := make(chan *trayWindow, 1) + done := make(chan error, 1) + go runTrayMessageLoop(output, tooltip, ready, done) + var window *trayWindow + select { + case window = <-ready: + case err := <-done: + return err + case <-ctx.Done(): + return nil + } + select { + case <-ctx.Done(): + if window != nil { + _, _, _ = procPostMessage.Call(window.hwnd, wmClose, 0, 0) + } + return <-done + case err := <-done: + return err + } +} + +func acquireTrayMutex() (winapi.Handle, error) { + var sessionID uint32 + if err := winapi.ProcessIdToSessionId(winapi.GetCurrentProcessId(), &sessionID); err != nil { + return 0, err + } + name, err := winapi.UTF16PtrFromString(fmt.Sprintf("Local\\RVBoxTrayV1-%d", sessionID)) + if err != nil { + return 0, err + } + mutex, err := winapi.CreateMutex(nil, true, name) + if errors.Is(err, winapi.ERROR_ALREADY_EXISTS) { + if mutex != 0 { + _ = winapi.CloseHandle(mutex) + } + return 0, ErrTrayAlreadyRunning + } + if err != nil { + return 0, err + } + return mutex, nil +} + +func runTrayMessageLoop(output io.Writer, tooltip string, ready chan<- *trayWindow, done chan<- error) { + className, err := winapi.UTF16PtrFromString("RVBoxTrayWindowV1") + if err != nil { + done <- err + return + } + instance, _, _ := procGetModuleHandle.Call(0) + icon, _, _ := procLoadIcon.Call(0, uintptr(trayIDIApplication)) + cursor, _, _ := procLoadCursor.Call(0, uintptr(trayIDCArrow)) + class := trayClass{CbSize: uint32(unsafe.Sizeof(trayClass{})), WndProc: trayCallback, HInstance: instance, HIcon: icon, HCursor: cursor, ClassName: className, HIconSm: icon} + if result, _, callErr := procRegisterClassEx.Call(uintptr(unsafe.Pointer(&class))); result == 0 && !errors.Is(callErr, winapi.ERROR_CLASS_ALREADY_EXISTS) { + done <- callErr + return + } + title, _ := winapi.UTF16PtrFromString("RVBox") + hwnd, _, callErr := procCreateWindowEx.Call(0, uintptr(unsafe.Pointer(className)), uintptr(unsafe.Pointer(title)), 0, 0, 0, 0, 0, 0, 0, instance, 0) + if hwnd == 0 { + done <- callErr + return + } + window := &trayWindow{hwnd: hwnd, class: className, tooltip: tooltip, output: output} + window.icon = trayIconData{CbSize: uint32(unsafe.Sizeof(trayIconData{})), HWnd: hwnd, UID: trayIconID, UFlags: trayNIFMessage | trayNIFIcon | trayNIFTip, UCallbackMessage: trayMessage, HIcon: icon} + copy(window.icon.Tip[:], winapi.StringToUTF16(tooltip)) + trayWindows.Store(hwnd, window) + if result, _, err := procShellNotify.Call(uintptr(trayNIMAdd), uintptr(unsafe.Pointer(&window.icon))); result == 0 { + trayWindows.Delete(hwnd) + _, _, _ = procDestroyWindow.Call(hwnd) + done <- err + return + } + _, _, _ = procShowWindow.Call(hwnd, traySWHide) + ready <- window + for { + var message trayMessageRecord + result, _, getErr := procGetMessage.Call(uintptr(unsafe.Pointer(&message)), 0, 0, 0) + if int32(result) == -1 { + window.removeIcon() + done <- getErr + return + } + if result == 0 { + window.removeIcon() + done <- nil + return + } + _, _, _ = procTranslate.Call(uintptr(unsafe.Pointer(&message))) + _, _, _ = procDispatch.Call(uintptr(unsafe.Pointer(&message))) + } +} + +func trayWindowProc(hwnd uintptr, message uint32, wParam, lParam uintptr) uintptr { + value, _ := trayWindows.Load(hwnd) + window, _ := value.(*trayWindow) + switch message { + case wmClose: + _, _, _ = procDestroyWindow.Call(hwnd) + return 0 + case wmDestroy: + if window != nil { + window.removeIcon() + } + trayWindows.Delete(hwnd) + procPostQuitMessage.Call(0) + return 0 + case wmCommand: + if window != nil { + window.action(Action(uint16(wParam))) + } + return 0 + case trayMessage: + if window != nil && (uint32(lParam) == wmRButtonUp || uint32(lParam) == wmLButtonDblClick) { + window.showMenu() + } + return 0 + } + result, _, _ := procDefWindowProc.Call(hwnd, uintptr(message), wParam, lParam) + return result +} + +func (window *trayWindow) removeIcon() { + window.closeOnce.Do(func() { + window.icon.UFlags = 0 + _, _, _ = procShellNotify.Call(uintptr(trayNIMDelete), uintptr(unsafe.Pointer(&window.icon))) + }) +} + +func (window *trayWindow) showMenu() { + menu, _, _ := procCreatePopup.Call() + if menu == 0 { + return + } + defer procDestroyMenu.Call(menu) + add := func(action Action, label string) { + text, _ := winapi.UTF16PtrFromString(label) + procAppendMenu.Call(menu, trayMFString, uintptr(action), uintptr(unsafe.Pointer(text))) + } + add(ActionStatus, "Status") + add(ActionOpenConfig, "Open config") + add(ActionOpenLog, "Open log") + procAppendMenu.Call(menu, trayMFSeparator, 0, 0) + add(ActionStartService, "Start service") + add(ActionStopService, "Stop service") + add(ActionRestartService, "Restart service") + add(ActionSetAutomatic, "Start automatically") + add(ActionSetManual, "Start manually") + procAppendMenu.Call(menu, trayMFSeparator, 0, 0) + add(ActionExitTray, "Exit") + var point trayPoint + procGetCursorPos.Call(uintptr(unsafe.Pointer(&point))) + procSetForeground.Call(window.hwnd) + procTrackPopup.Call(menu, trayTPMRightBtn, uintptr(point.X), uintptr(point.Y), 0, window.hwnd, 0) +} + +func (window *trayWindow) action(action Action) { + if action == ActionExitTray { + _, _, _ = procDestroyWindow.Call(window.hwnd) + return + } + response, err := Request(context.Background(), Frame{Action: action}) + if err != nil { + window.setTooltip("RVBox service unavailable") + return + } + if isServiceMutation(action) && strings.HasPrefix(string(response.Payload), "error:") { + if err := runElevatedServiceAction(action); err == nil { + window.setTooltip("RVBox elevation requested") + return + } + } + if action == ActionOpenConfig || action == ActionOpenLog { + if err := openExactPath(string(response.Payload)); err != nil { + window.setTooltip("RVBox open failed") + } + return + } + if len(response.Payload) > 0 { + window.setTooltip(string(response.Payload)) + } +} + +func isServiceMutation(action Action) bool { + return action == ActionStartService || action == ActionStopService || action == ActionRestartService || action == ActionSetAutomatic || action == ActionSetManual +} + +// runElevatedServiceAction is the tray's narrow UAC fallback. The service +// remains the authorization boundary; this helper only replays the enum as a +// canonical rvbox mode under an administrator token when the tray's filtered +// token cannot mutate SCM state directly. +func runElevatedServiceAction(action Action) error { + mode := "" + switch action { + case ActionStartService: + mode = "--start-service" + case ActionStopService: + mode = "--stop-service" + case ActionRestartService: + mode = "--restart-service" + case ActionSetAutomatic: + mode = "--configure-service --startup automatic" + case ActionSetManual: + mode = "--configure-service --startup manual" + default: + return errors.New("unsupported elevated tray action") + } + executable, err := os.Executable() + if err != nil { + return err + } + configPath := "" + for index := 0; index+1 < len(os.Args); index++ { + if os.Args[index] == "--config" { + configPath = os.Args[index+1] + break + } + } + if configPath == "" { + return errors.New("tray config path is unavailable") + } + args, err := winapi.UTF16PtrFromString(mode + " --config " + syscall.EscapeArg(configPath)) + if err != nil { + return err + } + file, err := winapi.UTF16PtrFromString(executable) + if err != nil { + return err + } + if err := winapi.ShellExecute(0, mustUTF16("runas"), file, args, nil, winapi.SW_SHOWNORMAL); err != nil { + return err + } + return nil +} + +func mustUTF16(value string) *uint16 { + encoded, _ := winapi.UTF16PtrFromString(value) + return encoded +} + +func (window *trayWindow) setTooltip(value string) { + value = strings.TrimSpace(value) + if value == "" { + value = "RVBox service" + } + window.tooltip = value + window.icon.UFlags = trayNIFTip + for index := range window.icon.Tip { + window.icon.Tip[index] = 0 + } + copy(window.icon.Tip[:], winapi.StringToUTF16(value)) + _, _, _ = procShellNotify.Call(uintptr(trayNIMModify), uintptr(unsafe.Pointer(&window.icon))) +} + +func openExactPath(path string) error { + if path == "" || strings.TrimSpace(path) != path || !utf8.ValidString(path) { + return errors.New("invalid path returned by service") + } + info, err := os.Stat(path) + if err != nil || !info.Mode().IsRegular() { + return errors.New("service path is not a regular file") + } + windowsDir, err := winapi.GetWindowsDirectory() + if err != nil { + return err + } + explorer := filepath.Join(windowsDir, "explorer.exe") + if info, err := os.Stat(explorer); err != nil || !info.Mode().IsRegular() { + return errors.New("Windows Explorer is unavailable") + } + return exec.Command(explorer, path).Start() +} diff --git a/internal/observability/health.go b/internal/observability/health.go new file mode 100644 index 0000000..feb4e86 --- /dev/null +++ b/internal/observability/health.go @@ -0,0 +1,151 @@ +// Package observability provides the intentionally small health/metrics HTTP +// surface shared by the Linux server and Windows client. It contains no +// product state and never exposes command payloads or high-cardinality IDs. +package observability + +import ( + "context" + "fmt" + "net" + "net/http" + "sort" + "strconv" + "strings" + "sync" + "sync/atomic" + "time" +) + +type Paths struct { + Liveness string + Readiness string + Metrics string +} + +type Health struct { + ready atomic.Bool + dirty atomic.Bool + mu sync.Mutex + count map[string]uint64 +} + +func New() *Health { return &Health{count: make(map[string]uint64)} } + +func (health *Health) SetReady(value bool) { + if health != nil { + health.ready.Store(value) + } +} + +func (health *Health) SetDirty(value bool) { + if health != nil { + health.dirty.Store(value) + } +} + +func (health *Health) Inc(name string) { + if health == nil || !validMetricName(name) { + return + } + health.mu.Lock() + health.count[name]++ + health.mu.Unlock() +} + +func (health *Health) Snapshot() (ready, dirty bool, counters map[string]uint64) { + if health == nil { + return false, true, nil + } + health.mu.Lock() + defer health.mu.Unlock() + copyCounters := make(map[string]uint64, len(health.count)) + for key, value := range health.count { + copyCounters[key] = value + } + return health.ready.Load(), health.dirty.Load(), copyCounters +} + +func (health *Health) Handler(paths Paths) http.Handler { + if paths.Liveness == "" { + paths.Liveness = "/livez" + } + if paths.Readiness == "" { + paths.Readiness = "/readyz" + } + if paths.Metrics == "" { + paths.Metrics = "/metrics" + } + mux := http.NewServeMux() + mux.HandleFunc(paths.Liveness, func(response http.ResponseWriter, _ *http.Request) { + response.Header().Set("Content-Type", "text/plain; charset=utf-8") + response.WriteHeader(http.StatusOK) + _, _ = response.Write([]byte("live\n")) + }) + mux.HandleFunc(paths.Readiness, func(response http.ResponseWriter, _ *http.Request) { + ready, dirty, _ := health.Snapshot() + response.Header().Set("Content-Type", "text/plain; charset=utf-8") + if !ready { + response.WriteHeader(http.StatusServiceUnavailable) + } else { + response.WriteHeader(http.StatusOK) + } + _, _ = fmt.Fprintf(response, "ready=%s dirty=%s\n", strconv.FormatBool(ready), strconv.FormatBool(dirty)) + }) + mux.HandleFunc(paths.Metrics, func(response http.ResponseWriter, _ *http.Request) { + ready, dirty, counters := health.Snapshot() + response.Header().Set("Content-Type", "text/plain; version=0.0.4") + response.WriteHeader(http.StatusOK) + _, _ = fmt.Fprintf(response, "rvbox_health_ready %d\nrvbox_health_dirty %d\n", boolMetric(ready), boolMetric(dirty)) + keys := make([]string, 0, len(counters)) + for key := range counters { + keys = append(keys, key) + } + sort.Strings(keys) + for _, key := range keys { + _, _ = fmt.Fprintf(response, "rvbox_%s_total %d\n", key, counters[key]) + } + }) + return mux +} + +// Serve starts an endpoint and closes it when ctx is cancelled. The caller +// may use the returned error to keep listener failures visible without making +// product startup depend on a slow or unavailable metrics consumer. +func (health *Health) Serve(ctx context.Context, listen string, paths Paths) error { + if strings.TrimSpace(listen) == "" { + return fmt.Errorf("observability listen address is empty") + } + server := &http.Server{Addr: listen, Handler: health.Handler(paths), ReadHeaderTimeout: 10 * time.Second} + listener, err := net.Listen("tcp", listen) + if err != nil { + return err + } + go func() { + <-ctx.Done() + _ = server.Shutdown(context.Background()) + }() + err = server.Serve(listener) + if err == http.ErrServerClosed { + return nil + } + return err +} + +func boolMetric(value bool) int { + if value { + return 1 + } + return 0 +} + +func validMetricName(name string) bool { + if name == "" { + return false + } + for _, character := range name { + if !(character == '_' || character >= 'a' && character <= 'z' || character >= 'A' && character <= 'Z' || character >= '0' && character <= '9') { + return false + } + } + return true +} diff --git a/internal/observability/health_test.go b/internal/observability/health_test.go new file mode 100644 index 0000000..0862475 --- /dev/null +++ b/internal/observability/health_test.go @@ -0,0 +1,42 @@ +package observability + +import ( + "net/http" + "net/http/httptest" + "strings" + "testing" +) + +func TestHealthHandlerReadinessAndMetrics_HP_OPS_01(t *testing.T) { + t.Parallel() + health := New() + health.Inc("dispatch_attempt") + handler := health.Handler(Paths{Liveness: "/live", Readiness: "/ready", Metrics: "/metrics"}) + ready := httptest.NewRecorder() + handler.ServeHTTP(ready, httptest.NewRequest(http.MethodGet, "/ready", nil)) + if ready.Code != http.StatusServiceUnavailable || !strings.Contains(ready.Body.String(), "ready=false") { + t.Fatalf("initial readiness = %d %q", ready.Code, ready.Body.String()) + } + health.SetReady(true) + health.SetDirty(true) + ready = httptest.NewRecorder() + handler.ServeHTTP(ready, httptest.NewRequest(http.MethodGet, "/ready", nil)) + if ready.Code != http.StatusOK || !strings.Contains(ready.Body.String(), "dirty=true") { + t.Fatalf("ready health = %d %q", ready.Code, ready.Body.String()) + } + metrics := httptest.NewRecorder() + handler.ServeHTTP(metrics, httptest.NewRequest(http.MethodGet, "/metrics", nil)) + if metrics.Code != http.StatusOK || !strings.Contains(metrics.Body.String(), "rvbox_dispatch_attempt_total 1") { + t.Fatalf("metrics = %d %q", metrics.Code, metrics.Body.String()) + } +} + +func TestHealthMetricNamesAreBounded_BH_OPS_01(t *testing.T) { + t.Parallel() + health := New() + health.Inc("bad name") + _, _, counters := health.Snapshot() + if len(counters) != 0 { + t.Fatalf("invalid metric name recorded: %#v", counters) + } +} diff --git a/internal/server/control/service.go b/internal/server/control/service.go index cf422e7..ff7eb16 100644 --- a/internal/server/control/service.go +++ b/internal/server/control/service.go @@ -219,9 +219,8 @@ func (service *Service) GetCommand(ctx context.Context, request *rvboxv1.GetComm return &rvboxv1.GetCommandResponse{Command: record}, nil } -// RunCommand durably admits command-text work. Script payload persistence and -// chunk dispatch are intentionally kept behind the same immutable boundary and -// are rejected until their command_payloads path is wired to the dispatcher. +// RunCommand durably admits command-text and script work. Both forms share one +// immutable request boundary so replaying the same request ID is idempotent. func (service *Service) RunCommand(ctx context.Context, request *rvboxv1.RunCommandRequest) (*rvboxv1.RunCommandResponse, error) { if request == nil || request.GetTargetClientId() == "" || request.GetSpec() == nil { return nil, controlError(codes.InvalidArgument, rvboxv1.ControlError_INVALID_ARGUMENT, "target_client_id and spec are required") @@ -375,7 +374,10 @@ func (service *Service) GetOutput(ctx context.Context, request *rvboxv1.GetOutpu if maxBytes > 16<<20 { return nil, controlError(codes.InvalidArgument, rvboxv1.ControlError_INVALID_ARGUMENT, "max_bytes exceeds the control limit") } - filterBytes := []byte(fmt.Sprintf("output\x00%s\x00%v", request.GetClientId(), streams)) + // Bind the cursor to every selector that affects the result. In particular, + // issue_uuid must be included: otherwise a token issued for one command + // could be replayed against another command owned by the same client. + filterBytes := []byte(fmt.Sprintf("output\x00%s\x00%s\x00%v", request.GetClientId(), issue.String(), streams)) filter := domain.HashCursorFilters(filterBytes) var eventSeq, offset, boundary uint64 if request.GetPageToken() != "" { @@ -793,7 +795,14 @@ func commandRecord(view store.CommandView) (*rvboxv1.CommandRecord, error) { if err := proto.Unmarshal(view.ExecutionSpec, spec); err != nil { return nil, fmt.Errorf("decode stored execution spec: %w", err) } - result := &rvboxv1.CommandRecord{IssueUuid: view.IssueUUID.String(), TargetClientId: view.ClientID, IssueTime: timestamppb.New(view.IssueTime), ServerReceiptTime: timestamppb.New(view.ServerReceiptTime), Spec: spec, Lifecycle: rvboxv1.CommandLifecycle(view.Lifecycle), LastEventSeq: view.LastEventSeq, OutputTruncated: view.OutputTruncated, OutputIncomplete: view.OutputIncomplete, RetainedCompressedBytes: view.RetainedCompressedBytes, CommandRevision: view.Revision} + result := &rvboxv1.CommandRecord{IssueUuid: view.IssueUUID.String(), TargetClientId: view.ClientID, IssueTime: timestamppb.New(view.IssueTime), ServerReceiptTime: timestamppb.New(view.ServerReceiptTime), Spec: spec, Lifecycle: rvboxv1.CommandLifecycle(view.Lifecycle), LastEventSeq: view.LastEventSeq, OutputTruncated: view.OutputTruncated, OutputIncomplete: view.OutputIncomplete, RetainedCompressedBytes: view.RetainedCompressedBytes, CommandRevision: view.Revision, LateAfterExpiry: view.LateAfterExpiry} + if len(view.Rejection) > 0 { + rejection := &rvboxv1.ControlError{} + if err := proto.Unmarshal(view.Rejection, rejection); err != nil { + return nil, fmt.Errorf("decode stored command rejection: %w", err) + } + result.Rejection = rejection + } if view.QueueExpiryTime != nil { result.QueueExpiryTime = timestamppb.New(*view.QueueExpiryTime) } diff --git a/internal/server/control/service_test.go b/internal/server/control/service_test.go index 3dcef01..ea90d31 100644 --- a/internal/server/control/service_test.go +++ b/internal/server/control/service_test.go @@ -240,6 +240,10 @@ func TestGetOutputSlicesWithAuthenticatedCursor_HP_CONTROL_12(t *testing.T) { if _, err := service.GetOutput(context.Background(), &rvboxv1.GetOutputRequest{ClientId: "win-a", IssueUuid: issue.String(), Streams: []rvboxv1.StreamKind{rvboxv1.StreamKind_STREAM_STDERR}, MaxBytes: 3, PageToken: first.GetNextPageToken()}); status.Code(err) != codes.InvalidArgument { t.Fatalf("changed output filter code = %v", status.Code(err)) } + otherIssue := fixedIssue(0xa8) + if _, err := service.GetOutput(context.Background(), &rvboxv1.GetOutputRequest{ClientId: "win-a", IssueUuid: otherIssue.String(), Streams: []rvboxv1.StreamKind{rvboxv1.StreamKind_STREAM_STDOUT}, MaxBytes: 3, PageToken: first.GetNextPageToken()}); status.Code(err) != codes.InvalidArgument { + t.Fatalf("changed output command code = %v", status.Code(err)) + } } func TestControlMutationIdempotencyAndQueuedCancellation_BH_CONTROL_13(t *testing.T) { diff --git a/internal/server/session/agent_server.go b/internal/server/session/agent_server.go index 83a629f..ce06686 100644 --- a/internal/server/session/agent_server.go +++ b/internal/server/session/agent_server.go @@ -64,18 +64,30 @@ func (server *AgentServer) ServeHTTP(response http.ResponseWriter, request *http http.Error(response, "agent session registry is unavailable", http.StatusServiceUnavailable) return } + started := time.Now() + heartbeat := newSynchronizedHeartbeat(server.heartbeatIdle(), server.livenessTimeout(), 0) connection, err := websocket.Accept(response, request, &websocket.AcceptOptions{ CompressionMode: websocket.CompressionDisabled, + // coder/websocket consumes control frames inside Read. Count ping and + // pong callbacks as inbound activity so a healthy, otherwise idle agent + // is not mistaken for a dead peer; application Read results are observed + // by serveConnection as usual. + OnPingReceived: func(context.Context, []byte) bool { + heartbeat.Observe(time.Since(started)) + return true + }, + OnPongReceived: func(context.Context, []byte) { + heartbeat.Observe(time.Since(started)) + }, }) if err != nil { return } defer connection.CloseNow() - server.serveConnection(request.Context(), connection) + server.serveConnection(request.Context(), connection, heartbeat, started) } -func (server *AgentServer) serveConnection(parent context.Context, connection *websocket.Conn) { - started := time.Now() +func (server *AgentServer) serveConnection(parent context.Context, connection *websocket.Conn, heartbeat *synchronizedHeartbeat, started time.Time) { messageType, payload, err := connection.Read(parent) if err != nil { return @@ -194,7 +206,6 @@ func (server *AgentServer) serveConnection(parent context.Context, connection *w return } - heartbeat := newSynchronizedHeartbeat(server.heartbeatIdle(), server.livenessTimeout(), 0) writerDone := make(chan struct{}) go func() { defer close(writerDone) @@ -266,7 +277,7 @@ func (server *AgentServer) serveConnection(parent context.Context, connection *w server.close(connection, websocket.StatusPolicyViolation, "invalid command acceptance") return } - if _, acceptErr := server.Store.RecordCommandAcceptance(sessionContext, issue, hello.GetClientId(), registration.Generation, acknowledgement.GetCommandRevision(), acknowledgement.GetAccepted(), server.now()); acceptErr != nil { + if _, acceptErr := server.Store.RecordCommandAcceptanceWithRejection(sessionContext, issue, hello.GetClientId(), registration.Generation, acknowledgement.GetCommandRevision(), acknowledgement.GetAccepted(), acknowledgement.GetRejection(), server.now()); acceptErr != nil { server.close(connection, websocket.StatusPolicyViolation, "invalid command acceptance") return } @@ -360,6 +371,7 @@ func eventAppendFromWire(event *rvboxv1.CommandEvent, clientID string, generatio if lifecycle := event.GetLifecycle(); lifecycle != nil { value := lifecycle.GetLifecycle() result.Lifecycle, result.LifecycleRevision = &value, lifecycle.GetCommandRevision() + result.WindowsIdentity = lifecycle.GetWindowsExecutionIdentity() } if output := event.GetOutput(); output != nil { result.Stream = uint16(output.GetStream()) @@ -707,7 +719,6 @@ func (server *AgentServer) writeLoop(ctx context.Context, connection *websocket. if err != nil { return } - heartbeat.Observe(time.Since(started)) case HeartbeatClose: return } diff --git a/internal/server/store/acceptance.go b/internal/server/store/acceptance.go index 9706a03..743240e 100644 --- a/internal/server/store/acceptance.go +++ b/internal/server/store/acceptance.go @@ -6,7 +6,9 @@ import ( "errors" "time" + rvboxv1 "github.com/rvbox/rvbox/gen/go/rvbox/v1" "github.com/rvbox/rvbox/internal/domain" + "google.golang.org/protobuf/proto" ) var ErrDispatchMismatch = errors.New("command acceptance does not match dispatched generation or revision") @@ -15,9 +17,31 @@ var ErrDispatchMismatch = errors.New("command acceptance does not match dispatch // produced it. Repeated acknowledgements are harmless; a stale session cannot // advance a newer dispatch or overwrite a terminal decision. func (store *Store) RecordCommandAcceptance(ctx context.Context, issueUUID domain.UUID, clientID string, generation, revision uint64, accepted bool, now time.Time) (bool, error) { + return store.RecordCommandAcceptanceWithRejection(ctx, issueUUID, clientID, generation, revision, accepted, nil, now) +} + +// RecordCommandAcceptanceWithRejection is the wire-facing variant that keeps +// a structured client admission rejection on the command row for status and +// history. The legacy method above remains convenient for reconciliation and +// tests that only have the accepted bit. +func (store *Store) RecordCommandAcceptanceWithRejection(ctx context.Context, issueUUID domain.UUID, clientID string, generation, revision uint64, accepted bool, rejection *rvboxv1.ControlError, now time.Time) (bool, error) { if isZeroUUID([16]byte(issueUUID)) || clientID == "" || generation == 0 || revision == 0 || now.IsZero() { return false, ErrDispatchMismatch } + if accepted && rejection != nil { + return false, ErrDispatchMismatch + } + var storedRejection []byte + var err error + if rejection != nil { + if len(rejection.GetMessage()) > 4096 || len(rejection.GetIssueUuid()) > 128 { + return false, ErrDispatchMismatch + } + storedRejection, err = proto.MarshalOptions{Deterministic: true}.Marshal(rejection) + if err != nil || len(storedRejection) > 16<<10 { + return false, ErrDispatchMismatch + } + } store.writeMu.Lock() defer store.writeMu.Unlock() database, err := store.openDatabase() @@ -28,15 +52,37 @@ func (store *Store) RecordCommandAcceptance(ctx context.Context, issueUUID domai if !accepted { targetLifecycle = 11 } + tx, err := database.BeginTx(ctx, nil) + if err != nil { + return false, err + } + defer tx.Rollback() + var queueExpiry sql.NullInt64 + var lateAfterExpiry int + if err := tx.QueryRowContext(ctx, `SELECT queue_expiry_time, late_after_expiry FROM commands WHERE issue_uuid = ? AND client_id = ? AND target_session_generation = ? AND revision = ?`, issueUUID[:], clientID, generation, revision).Scan(&queueExpiry, &lateAfterExpiry); errors.Is(err, sql.ErrNoRows) { + _ = tx.Rollback() + return false, ErrDispatchMismatch + } else if err != nil { + return false, err + } query := `UPDATE commands SET lifecycle = ?` args := []any{targetLifecycle} if !accepted { query += `, terminal_time = ?` args = append(args, now.UTC().UnixNano()) } + if rejection != nil { + query += `, rejection = ?` + args = append(args, storedRejection) + } + // The command may have been dispatched just before its queue deadline and + // its acceptance can arrive after that deadline. Preserve the eventual + // client truth, but durably retain the contradiction for status/audit. + query += `, late_after_expiry = CASE WHEN queue_expiry_time IS NOT NULL AND queue_expiry_time <= ? THEN 1 ELSE late_after_expiry END` + args = append(args, now.UTC().UnixNano()) query += ` WHERE issue_uuid = ? AND client_id = ? AND lifecycle = 2 AND target_session_generation = ? AND revision = ?` args = append(args, issueUUID[:], clientID, generation, revision) - result, err := database.ExecContext(ctx, query, args...) + result, err := tx.ExecContext(ctx, query, args...) if err != nil { return false, err } @@ -45,8 +91,20 @@ func (store *Store) RecordCommandAcceptance(ctx context.Context, issueUUID domai return false, err } if changed == 1 { + late := queueExpiry.Valid && now.UTC().UnixNano() >= queueExpiry.Int64 + if late && lateAfterExpiry == 0 { + if err := appendLateExpiryAudit(ctx, tx, now.UTC().UnixNano(), clientID, [16]byte(issueUUID), now.UTC().UnixNano()); err != nil { + return false, err + } + } + if err := tx.Commit(); err != nil { + return false, err + } return true, nil } + if err := tx.Commit(); err != nil { + return false, err + } var lifecycle uint32 var storedGeneration, storedRevision uint64 err = database.QueryRowContext(ctx, `SELECT lifecycle, COALESCE(target_session_generation, 0), revision FROM commands WHERE issue_uuid = ? AND client_id = ?`, issueUUID[:], clientID).Scan(&lifecycle, &storedGeneration, &storedRevision) diff --git a/internal/server/store/audit_helpers.go b/internal/server/store/audit_helpers.go new file mode 100644 index 0000000..83aa616 --- /dev/null +++ b/internal/server/store/audit_helpers.go @@ -0,0 +1,21 @@ +package store + +import ( + "context" + "crypto/sha256" + "database/sql" + "fmt" +) + +// appendLateExpiryAudit records the first observed queue-deadline +// contradiction without copying command text, script bodies, or output into +// the separate audit budget. The command row remains the fast status path; +// this row is the durable operator/audit history. +func appendLateExpiryAudit(ctx context.Context, tx *sql.Tx, occurredAt int64, clientID string, issue [16]byte, receiptUnixNano int64) error { + payload := []byte(fmt.Sprintf("receipt_unix_nano=%d", receiptUnixNano)) + digest := sha256.Sum256(payload) + _, err := tx.ExecContext(ctx, `INSERT INTO audit_events ( +occurred_at, source, client_id, issue_uuid, action, outcome, compression, payload, raw_bytes, stored_bytes, sha256 +) VALUES (?, 'session', ?, ?, 'late_after_expiry', 'observed', 1, ?, ?, ?, ?)`, occurredAt, clientID, issue[:], payload, len(payload), len(payload), digest[:]) + return err +} diff --git a/internal/server/store/command_test.go b/internal/server/store/command_test.go index 32b0b9c..33621c6 100644 --- a/internal/server/store/command_test.go +++ b/internal/server/store/command_test.go @@ -126,6 +126,51 @@ func TestClaimDispatchExpiresAndFencesRequeue_HP_DISPATCH_02(t *testing.T) { } } +func TestLateDispatchAcceptanceAndEventRetainExpiryContradiction_BH_DISPATCH_09(t *testing.T) { + t.Parallel() + ctx := context.Background() + opened, err := Open(ctx, Options{DataDir: filepath.Join(t.TempDir(), "state"), BusyTimeout: time.Second}) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = opened.Close() }) + base := time.Date(2026, time.September, 6, 13, 0, 0, 0, time.UTC) + if _, err := opened.RegisterClientSession(ctx, ClientRegistration{ + ClientID: "late-client", Platform: 2, Architecture: "amd64", DaemonVersion: "test", DaemonCWD: `C:\`, SupportedShells: []byte{1}, + ClientInstanceID: [16]byte{21}, SessionID: [16]byte{22}, ConnectedAt: base, + }); err != nil { + t.Fatal(err) + } + issue := fixedStoreIssue(0xd6) + expiry := base.Add(time.Second) + if _, err := opened.QueueCommand(ctx, QueueCommandInput{IssueUUID: issue, ClientID: "late-client", IssueTime: base, ReceiptTime: base, QueueExpiryTime: &expiry, ImmutableSHA256: sha256.Sum256([]byte("late")), ExecutionSpec: []byte("spec")}); err != nil { + t.Fatal(err) + } + if candidate, err := opened.ClaimNextDispatch(ctx, "late-client", 7, base.Add(500*time.Millisecond)); err != nil || candidate == nil { + t.Fatalf("dispatch claim = %#v, %v", candidate, err) + } + if changed, err := opened.RecordCommandAcceptance(ctx, issue, "late-client", 7, 1, true, base.Add(2*time.Second)); err != nil || !changed { + t.Fatalf("late acceptance = %t, %v", changed, err) + } + var late int + if err := opened.DB().QueryRow(`SELECT late_after_expiry FROM commands WHERE issue_uuid = ?`, issue[:]).Scan(&late); err != nil || late != 1 { + t.Fatalf("late acceptance flag = %d, %v", late, err) + } + var audits int + if err := opened.DB().QueryRow(`SELECT COUNT(*) FROM audit_events WHERE action = 'late_after_expiry' AND issue_uuid = ?`, issue[:]).Scan(&audits); err != nil || audits != 1 { + t.Fatalf("late acceptance audit rows = %d, %v", audits, err) + } + if _, err := opened.AppendCommandEvent(ctx, EventAppend{IssueUUID: [16]byte(issue), ClientID: "late-client", SessionGeneration: 7, EventSeq: 1, ObservedUnixNano: base.Add(2 * time.Second).UnixNano(), ReceiptUnixNano: base.Add(2 * time.Second).UnixNano(), EventType: 5, Stream: 1, Compression: 1, RawLength: 1, Payload: []byte("x"), ImmutableSHA256: sha256.Sum256([]byte("event")), Output: true}); err != nil { + t.Fatal(err) + } + if err := opened.DB().QueryRow(`SELECT late_after_expiry FROM commands WHERE issue_uuid = ?`, issue[:]).Scan(&late); err != nil || late != 1 { + t.Fatalf("late event flag = %d, %v", late, err) + } + if err := opened.DB().QueryRow(`SELECT COUNT(*) FROM audit_events WHERE action = 'late_after_expiry' AND issue_uuid = ?`, issue[:]).Scan(&audits); err != nil || audits != 1 { + t.Fatalf("late event duplicated audit rows = %d, %v", audits, err) + } +} + func TestQueueAndClaimScriptPayloadIsDurable_HP_SCRIPT_02(t *testing.T) { t.Parallel() ctx := context.Background() @@ -195,6 +240,25 @@ func TestRecordCommandAcceptanceFencesGeneration_HP_DISPATCH_05(t *testing.T) { if changed, err := opened.RecordCommandAcceptance(ctx, issue, "win-client", 3, 1, true, time.Now()); err != nil || changed { t.Fatalf("duplicate acceptance = %t, %v", changed, err) } + rejectedIssue := fixedStoreIssue(0x96) + if _, err := opened.QueueCommand(ctx, QueueCommandInput{IssueUUID: rejectedIssue, ClientID: "win-client", IssueTime: time.Now(), ReceiptTime: time.Now(), ImmutableSHA256: sha256.Sum256([]byte("reject")), ExecutionSpec: []byte("spec")}); err != nil { + t.Fatal(err) + } + if _, err := opened.ClaimNextDispatch(ctx, "win-client", 3, time.Now()); err != nil { + t.Fatal(err) + } + rejection := &rvboxv1.ControlError{Code: rvboxv1.ControlError_INVALID_ARGUMENT, Message: "invalid command"} + if changed, err := opened.RecordCommandAcceptanceWithRejection(ctx, rejectedIssue, "win-client", 3, 1, false, rejection, time.Now()); err != nil || !changed { + t.Fatalf("rejection acceptance = %t, %v", changed, err) + } + view, err := opened.GetCommandView(ctx, "win-client", rejectedIssue) + if err != nil { + t.Fatal(err) + } + stored := &rvboxv1.ControlError{} + if err := proto.Unmarshal(view.Rejection, stored); err != nil || stored.GetCode() != rejection.GetCode() || stored.GetMessage() != rejection.GetMessage() { + t.Fatalf("stored rejection = %v, %v", stored, err) + } } func timePtr(value time.Time) *time.Time { return &value } diff --git a/internal/server/store/event.go b/internal/server/store/event.go index e33eaf1..2e8b458 100644 --- a/internal/server/store/event.go +++ b/internal/server/store/event.go @@ -50,6 +50,7 @@ type EventAppend struct { ImmutableSHA256 [32]byte Lifecycle *rvboxv1.CommandLifecycle LifecycleRevision uint64 + WindowsIdentity *rvboxv1.WindowsExecutionIdentity Output bool UseCloseout bool } @@ -79,6 +80,16 @@ func (store *Store) AppendCommandEvent(ctx context.Context, event EventAppend) ( if err != nil { return result, err } + var windowsIdentity []byte + if event.WindowsIdentity != nil { + if len(event.WindowsIdentity.GetAttemptedContexts()) > 8 || len(event.WindowsIdentity.GetSelectionDetail()) > 4096 { + return result, ErrInvalidSegmentRecord + } + windowsIdentity, err = proto.MarshalOptions{Deterministic: true}.Marshal(event.WindowsIdentity) + if err != nil || len(windowsIdentity) > 16<<10 { + return result, ErrInvalidSegmentRecord + } + } store.writeMu.Lock() defer store.writeMu.Unlock() @@ -89,10 +100,13 @@ func (store *Store) AppendCommandEvent(ctx context.Context, event EventAppend) ( var lastSequence, commandOutputCharged, commandCharged, clientCharged, serverCharged, closeoutRemaining uint64 var currentLifecycle uint32 var currentRevision uint64 + var lateAfterExpiry int + var queueExpiry sql.NullInt64 var clientID string query := `SELECT commands.last_event_seq, commands.output_charged_bytes, commands.charged_bytes, commands.closeout_remaining_bytes, commands.client_id, clients.charged_bytes, -storage_counters.command_charged_bytes, commands.lifecycle, commands.revision +storage_counters.command_charged_bytes, commands.lifecycle, commands.revision, +commands.queue_expiry_time, commands.late_after_expiry FROM commands JOIN clients ON clients.client_id = commands.client_id JOIN storage_counters ON storage_counters.singleton = 1 WHERE commands.issue_uuid = ?` arguments := []any{event.IssueUUID[:]} @@ -105,7 +119,7 @@ JOIN storage_counters ON storage_counters.singleton = 1 WHERE commands.issue_uui arguments = append(arguments, event.SessionGeneration) } err = database.QueryRowContext(ctx, query, arguments...).Scan( - &lastSequence, &commandOutputCharged, &commandCharged, &closeoutRemaining, &clientID, &clientCharged, &serverCharged, ¤tLifecycle, ¤tRevision) + &lastSequence, &commandOutputCharged, &commandCharged, &closeoutRemaining, &clientID, &clientCharged, &serverCharged, ¤tLifecycle, ¤tRevision, &queueExpiry, &lateAfterExpiry) if errors.Is(err, sql.ErrNoRows) { return result, ErrCommandNotFound } @@ -133,6 +147,11 @@ JOIN storage_counters ON storage_counters.singleton = 1 WHERE commands.issue_uui return result, domain.ValidateTransition(rvboxv1.CommandLifecycle(currentLifecycle), *event.Lifecycle) } } + late := queueExpiry.Valid && event.ReceiptUnixNano >= queueExpiry.Int64 + lateValue := 0 + if late { + lateValue = 1 + } rowsCharged, indexesCharged := uint64(1), uint64(1) if store.willCreateSegment(event.IssueUUID, uint64(len(encoded))) { rowsCharged++ @@ -215,8 +234,12 @@ payload, segment_ordinal, segment_record_offset, segment_record_length, immutabl var update sql.Result commandUpdate := `UPDATE commands SET last_event_seq = ?, retained_compressed_bytes = retained_compressed_bytes + ?, output_charged_bytes = ?, charged_bytes = ?, - closeout_remaining_bytes = ?` - commandArgs := []any{event.EventSeq, len(event.Payload), reservation.CommandOutputCharged, reservation.CommandTotalCharged, reservation.CloseoutRemaining} + closeout_remaining_bytes = ?, late_after_expiry = CASE WHEN ? = 1 THEN 1 ELSE late_after_expiry END` + commandArgs := []any{event.EventSeq, len(event.Payload), reservation.CommandOutputCharged, reservation.CommandTotalCharged, reservation.CloseoutRemaining, lateValue} + if event.WindowsIdentity != nil { + commandUpdate = `UPDATE commands SET windows_execution_identity = ?, ` + commandUpdate[len("UPDATE commands SET "):] + commandArgs = append([]any{windowsIdentity}, commandArgs...) + } if event.Lifecycle != nil { terminal := 0 if domain.IsTerminal(*event.Lifecycle) { @@ -236,6 +259,11 @@ retained_compressed_bytes = retained_compressed_bytes + ?, output_charged_bytes } } } + if err == nil && late && lateAfterExpiry == 0 { + if err = appendLateExpiryAudit(ctx, tx, event.ReceiptUnixNano, clientID, event.IssueUUID, event.ReceiptUnixNano); err == nil { + lateAfterExpiry = 1 + } + } if err == nil && event.EventType == 7 { // Stdin acknowledgements are command events, but their durable delivery // cursor lives in stdin_writes. Mark the cumulative prefix in the same diff --git a/internal/server/store/events_read_test.go b/internal/server/store/events_read_test.go index b5af8cd..0ff2cee 100644 --- a/internal/server/store/events_read_test.go +++ b/internal/server/store/events_read_test.go @@ -34,10 +34,22 @@ func TestReadCommandEventsValidatesSegmentAndPreservesOrder_HP_EVENT_04(t *testi } for sequence, lifecycle := range []rvboxv1.CommandLifecycle{rvboxv1.CommandLifecycle_COMMAND_RUNNING, rvboxv1.CommandLifecycle_COMMAND_SUCCEEDED} { event := []byte{byte(sequence + 1), 'e', 'v', 't'} - if _, err := opened.AppendCommandEvent(ctx, EventAppend{IssueUUID: [16]byte(issue), ClientID: "win-client", SessionGeneration: 1, EventSeq: uint64(sequence + 1), ObservedUnixNano: now.UnixNano(), ReceiptUnixNano: now.Add(time.Duration(sequence) * time.Millisecond).UnixNano(), EventType: 4, Compression: 1, RawLength: uint64(len(event)), Payload: event, ImmutableSHA256: sha256.Sum256(event), Lifecycle: &lifecycle, LifecycleRevision: 1}); err != nil { + var identity *rvboxv1.WindowsExecutionIdentity + if sequence == 0 { + effective := rvboxv1.WindowsExecutionContext_WINDOWS_EXECUTION_CONTEXT_ACTIVE_USER + identity = &rvboxv1.WindowsExecutionIdentity{EffectiveContext: &effective, AttemptedContexts: []rvboxv1.WindowsExecutionContext{effective}, SelectionDetail: "selected"} + } + if _, err := opened.AppendCommandEvent(ctx, EventAppend{IssueUUID: [16]byte(issue), ClientID: "win-client", SessionGeneration: 1, EventSeq: uint64(sequence + 1), ObservedUnixNano: now.UnixNano(), ReceiptUnixNano: now.Add(time.Duration(sequence) * time.Millisecond).UnixNano(), EventType: 4, Compression: 1, RawLength: uint64(len(event)), Payload: event, ImmutableSHA256: sha256.Sum256(event), Lifecycle: &lifecycle, LifecycleRevision: 1, WindowsIdentity: identity}); err != nil { t.Fatal(err) } } + view, err := opened.GetCommandView(ctx, "win-client", issue) + if err != nil { + t.Fatal(err) + } + if len(view.WindowsIdentity) == 0 { + t.Fatal("running lifecycle did not persist Windows identity") + } events, err := opened.ReadCommandEvents(ctx, issue, 0, 10) if err != nil || len(events) != 2 || events[0].EventSeq != 1 || events[1].EventSeq != 2 || string(events[1].Payload) != "\x02evt" { t.Fatalf("read events = %#v, %v", events, err) diff --git a/internal/server/store/migrations.go b/internal/server/store/migrations.go index ab43b5e..98a1e03 100644 --- a/internal/server/store/migrations.go +++ b/internal/server/store/migrations.go @@ -16,6 +16,8 @@ type migration struct { var migrations = []migration{ {version: 1, sql: schemaV1}, {version: 2, sql: schemaV2}, + {version: 3, sql: schemaV3}, + {version: 4, sql: schemaV4}, } func applyMigrations(ctx context.Context, db *sql.DB) error { @@ -196,3 +198,17 @@ CREATE TABLE signal_intents ( ) STRICT; CREATE INDEX signal_intents_pending ON signal_intents(issue_uuid, command_revision, signal, acknowledged, created_at, request_uuid); ` + +// schemaV3 records the accepted/executed-after-deadline contradiction on the +// command itself. It is separate from the public lifecycle: the eventual +// client-reported terminal result remains authoritative while operators can +// still see that acceptance or execution arrived after the queue deadline. +const schemaV3 = ` +ALTER TABLE commands ADD COLUMN late_after_expiry INTEGER NOT NULL DEFAULT 0 CHECK(late_after_expiry IN (0,1)); +` + +// schemaV4 retains the structured client admission rejection so status and +// history do not have to reconstruct an error from a free-form event detail. +const schemaV4 = ` +ALTER TABLE commands ADD COLUMN rejection BLOB; +` diff --git a/internal/server/store/query.go b/internal/server/store/query.go index ea4ad03..327df30 100644 --- a/internal/server/store/query.go +++ b/internal/server/store/query.go @@ -46,6 +46,8 @@ type CommandView struct { OutputTruncated bool OutputIncomplete bool RetainedCompressedBytes uint64 + LateAfterExpiry bool + Rejection []byte ExecutionSpec []byte WindowsIdentity []byte } @@ -182,7 +184,7 @@ func (store *Store) ListCommandViews(ctx context.Context, page CommandPage) ([]C query := `SELECT issue_uuid, client_id, issue_time, server_receipt_time, queue_expiry_time, terminal_time, lifecycle, revision, last_event_seq, exit_code, output_truncated, output_incomplete, retained_compressed_bytes, - execution_spec, execution_spec_raw_bytes, windows_execution_identity + late_after_expiry, rejection, execution_spec, execution_spec_raw_bytes, windows_execution_identity FROM commands WHERE client_id = ? AND issue_time <= ?` args := []any{page.ClientID, page.SnapshotBoundary} if !page.IncludeTerminal { @@ -231,7 +233,7 @@ func (store *Store) GetCommandView(ctx context.Context, clientID string, issue d query := `SELECT issue_uuid, client_id, issue_time, server_receipt_time, queue_expiry_time, terminal_time, lifecycle, revision, last_event_seq, exit_code, output_truncated, output_incomplete, retained_compressed_bytes, - execution_spec, execution_spec_raw_bytes, windows_execution_identity + late_after_expiry, rejection, execution_spec, execution_spec_raw_bytes, windows_execution_identity FROM commands WHERE issue_uuid = ?` args := []any{issue[:]} if clientID != "" { @@ -252,10 +254,12 @@ func scanCommandView(scanner interface{ Scan(...any) error }) (CommandView, erro var expiry, terminal sql.NullInt64 var exit sql.NullInt64 var truncated, incomplete int + var lateAfterExpiry int + var rejection []byte var rawBytes uint64 if err := scanner.Scan(&issue, &view.ClientID, &issueTime, &receipt, &expiry, &terminal, &view.Lifecycle, &view.Revision, &view.LastEventSeq, &exit, &truncated, &incomplete, - &view.RetainedCompressedBytes, &stored, &rawBytes, &view.WindowsIdentity); err != nil { + &view.RetainedCompressedBytes, &lateAfterExpiry, &rejection, &stored, &rawBytes, &view.WindowsIdentity); err != nil { return view, err } if len(issue) != 16 { @@ -267,6 +271,8 @@ func scanCommandView(scanner interface{ Scan(...any) error }) (CommandView, erro view.QueueExpiryTime = nullableTime(expiry) view.TerminalTime = nullableTime(terminal) view.OutputTruncated, view.OutputIncomplete = truncated == 1, incomplete == 1 + view.LateAfterExpiry = lateAfterExpiry == 1 + view.Rejection = append([]byte(nil), rejection...) if exit.Valid { if exit.Int64 < -1<<31 || exit.Int64 > 1<<31-1 { return view, ErrInvalidSegmentRecord diff --git a/scripts/windows/test-host.ps1 b/scripts/windows/test-host.ps1 index 10c9729..a837675 100644 --- a/scripts/windows/test-host.ps1 +++ b/scripts/windows/test-host.ps1 @@ -6,14 +6,18 @@ param( [Parameter(Mandatory = $false)] [ValidatePattern('^[a-z0-9][a-z0-9-]{0,63}$')] [string] $RunId = '', - [string] $VmName = $(if ($env:RVBOX_WINDOWS_VM) { $env:RVBOX_WINDOWS_VM } else { 'rvbox-win10-test' }), - [string] $Snapshot = $(if ($env:RVBOX_WINDOWS_BASELINE_SNAPSHOT) { $env:RVBOX_WINDOWS_BASELINE_SNAPSHOT } else { 'baseline-disk-first' }) + [string] $VmName = $(if ($env:RVBOX_WINDOWS_VM) { $env:RVBOX_WINDOWS_VM } elseif ($env:RVBOX_TEST_VBOX_VM) { $env:RVBOX_TEST_VBOX_VM } else { 'rvbox-win10-test' }), + [string] $Snapshot = $(if ($env:RVBOX_WINDOWS_BASELINE_SNAPSHOT) { $env:RVBOX_WINDOWS_BASELINE_SNAPSHOT } elseif ($env:RVBOX_TEST_VBOX_SNAPSHOT) { $env:RVBOX_TEST_VBOX_SNAPSHOT } else { 'baseline-disk-first' }) ) $ErrorActionPreference = 'Stop' $VBoxManage = if ($env:VBOXMANAGE) { $env:VBOXMANAGE } else { 'VBoxManage' } -$GuestUser = $env:RVBOX_WINDOWS_GUEST_USER -$GuestPassword = $env:RVBOX_WINDOWS_GUEST_PASSWORD +$GuestUser = if ($env:RVBOX_WINDOWS_GUEST_USER) { $env:RVBOX_WINDOWS_GUEST_USER } else { $env:RVBOX_TEST_GUEST_USER } +$GuestPasswordFile = if ($env:RVBOX_WINDOWS_GUEST_PASSWORD_FILE) { + $env:RVBOX_WINDOWS_GUEST_PASSWORD_FILE +} else { + $env:RVBOX_TEST_GUEST_PASSWORD_FILE +} $LeaseRoot = if ($env:RVBOX_WINDOWS_LEASE_DIR) { $env:RVBOX_WINDOWS_LEASE_DIR } else { Join-Path $PSScriptRoot '..\..\.test-runs\windows' } function Invoke-VBox([string[]] $Arguments) { @@ -45,10 +49,33 @@ function Release-Lease { } function Guest-Args([string[]] $Arguments) { - if ([string]::IsNullOrWhiteSpace($GuestUser) -or [string]::IsNullOrWhiteSpace($GuestPassword)) { - throw 'Set RVBOX_WINDOWS_GUEST_USER and RVBOX_WINDOWS_GUEST_PASSWORD in the host environment; secrets are never read from repository files or printed.' + if ([string]::IsNullOrWhiteSpace($GuestUser) -or [string]::IsNullOrWhiteSpace($GuestPasswordFile)) { + throw 'Set the guest user and password-file environment variables in the host environment; secrets are never read from repository files or printed.' } - return @('guestcontrol', $VmName, '--username', $GuestUser, '--password', $GuestPassword) + $Arguments + if (-not (Test-Path -LiteralPath $GuestPasswordFile -PathType Leaf)) { + throw "Guest password file does not exist: $GuestPasswordFile" + } + return @('guestcontrol', $VmName, '--username', $GuestUser, '--passwordfile', $GuestPasswordFile) + $Arguments +} + +function Wait-PowerOff { + for ($attempt = 0; $attempt -lt 60; $attempt++) { + $current = (Invoke-VBox @('showvminfo', $VmName, '--machinereadable') | Where-Object { $_ -like 'VMState=*' }) -replace '^VMState="?([^"\r\n]+)"?$', '$1' + if ($current -eq 'poweroff') { return } + Start-Sleep -Seconds 1 + } + throw "VM did not power off within the bounded stop deadline: $VmName" +} + +function Wait-GuestReady { + for ($attempt = 0; $attempt -lt 60; $attempt++) { + $properties = Invoke-VBox @('guestproperty', 'enumerate', $VmName) + if (($properties -match '/GuestAdd/Version') -and ($properties -match '/GuestInfo/OS/Release')) { + return + } + Start-Sleep -Seconds 1 + } + throw "Guest Additions did not become ready within the bounded boot deadline: $VmName" } try { @@ -62,18 +89,18 @@ try { $state = (Invoke-VBox @('showvminfo', $VmName, '--machinereadable') | Where-Object { $_ -like 'VMState=*' }) -replace '^VMState="?([^"\r\n]+)"?$', '$1' switch ($Action) { 'Prepare' { - if ($state -eq 'running') { Invoke-VBox (Guest-Args @('run', 'C:\Windows\System32\whoami.exe', '--', '/groups')) | Out-Null } + if ($state -eq 'running') { Wait-GuestReady } else { Invoke-VBox @('startvm', $VmName, '--type', 'headless') | Out-Null - Start-Sleep -Seconds 5 - Invoke-VBox (Guest-Args @('run', 'C:\Windows\System32\whoami.exe', '--', '/groups')) | Out-Null + Wait-GuestReady } - Invoke-VBox (Guest-Args @('run', 'C:\Windows\System32\query.exe', '--', 'user')) | Out-Null + Invoke-VBox (Guest-Args @('run', '--exe', 'C:\Windows\System32\whoami.exe', '--wait-stdout', '--wait-stderr', '--unquoted-args', '--', '/groups')) | Out-Null + Invoke-VBox (Guest-Args @('run', '--exe', 'C:\Windows\System32\query.exe', '--wait-stdout', '--wait-stderr', '--unquoted-args', '--', 'user')) | Out-Null Write-Output "prepared VM=$VmName run=$RunId" } 'Run' { $guestBinary = if ($env:RVBOX_WINDOWS_GUEST_BINARY) { $env:RVBOX_WINDOWS_GUEST_BINARY } else { 'C:\ProgramData\RVBox\test\rvbox.exe' } - Invoke-VBox (Guest-Args @('run', $guestBinary, '--', '--check-config', '--config', 'C:\ProgramData\RVBox\client.toml')) | Out-Null + Invoke-VBox (Guest-Args @('run', '--exe', $guestBinary, '--wait-stdout', '--wait-stderr', '--unquoted-args', '--', '--check-config', '--config', 'C:\ProgramData\RVBox\client.toml')) | Out-Null Write-Output "ran native Windows smoke run=$RunId" } 'Collect' { @@ -84,11 +111,11 @@ try { Write-Output "collected native artifacts under $destination" } 'Stop' { - if ($state -eq 'running') { Invoke-VBox @('controlvm', $VmName, 'acpipowerbutton') | Out-Null } + if ($state -eq 'running') { Invoke-VBox @('controlvm', $VmName, 'acpipowerbutton') | Out-Null; Wait-PowerOff } Write-Output "requested graceful stop VM=$VmName" } 'Reset' { - if ($state -eq 'running') { Invoke-VBox @('controlvm', $VmName, 'acpipowerbutton') | Out-Null; Start-Sleep -Seconds 3 } + if ($state -eq 'running') { Invoke-VBox @('controlvm', $VmName, 'acpipowerbutton') | Out-Null; Wait-PowerOff } Invoke-VBox @('snapshot', $VmName, 'restore', $Snapshot) | Out-Null Write-Output "restored baseline snapshot=$Snapshot VM=$VmName run=$RunId" } diff --git a/test/coverage.toml b/test/coverage.toml index f65daa2..236de61 100644 --- a/test/coverage.toml +++ b/test/coverage.toml @@ -269,6 +269,12 @@ layer = "unit" status = "implemented" tests = ["internal/client/supervisor/supervisor_test.go:TestStartSpecValidation_BH_SUPERVISOR_01"] +[[requirements]] +id = "HP-SUPERVISOR-05" +layer = "unit" +status = "implemented" +tests = ["internal/client/supervisor/supervisor_test.go:TestWindowsIdentityMapsSelectionEvidence_HP_SUPERVISOR_05"] + [[requirements]] id = "HP-SUPERVISOR-03" layer = "unit" @@ -404,6 +410,12 @@ layer = "integration" status = "implemented" tests = ["test/integration/clientagent/clientagent_integration_test.go:TestControlStdinIntentReplaysAndAcknowledges_HP_DISPATCH_08"] +[[requirements]] +id = "BH-DISPATCH-09" +layer = "unit" +status = "implemented" +tests = ["internal/server/store/command_test.go:TestLateDispatchAcceptanceAndEventRetainExpiryContradiction_BH_DISPATCH_09"] + [[requirements]] id = "HP-SIGNAL-01" layer = "integration" @@ -551,6 +563,12 @@ layer = "unit" status = "implemented" tests = ["internal/client/spool/recovery_test.go:TestCheckSpoolRejectsCounterDrift_BH_CLIENT_03"] +[[requirements]] +id = "BH-CLIENT-13" +layer = "unit" +status = "implemented" +tests = ["internal/client/spool/spool_test.go:TestSendWindowPinsUnacknowledgedBytes_BH_CLIENT_13"] + [[requirements]] id = "HP-WINCTX-02" layer = "integration" diff --git a/test/integration/store/store_integration_test.go b/test/integration/store/store_integration_test.go index 280a2c6..b03950e 100644 --- a/test/integration/store/store_integration_test.go +++ b/test/integration/store/store_integration_test.go @@ -73,7 +73,7 @@ func TestRealSQLiteInitializationAndRestart_HP_STORE_01(t *testing.T) { t.Fatalf("tables = %v, want %v", names, want) } var migrationCount int - if err := opened.DB().QueryRow(`SELECT count(*) FROM schema_migrations`).Scan(&migrationCount); err != nil || migrationCount != 2 { + if err := opened.DB().QueryRow(`SELECT count(*) FROM schema_migrations`).Scan(&migrationCount); err != nil || migrationCount != 4 { t.Fatalf("migration count = %d, err = %v", migrationCount, err) } if err := opened.Close(); err != nil { @@ -81,7 +81,7 @@ func TestRealSQLiteInitializationAndRestart_HP_STORE_01(t *testing.T) { } reopened := openStore(t, dataDir) - if err := reopened.DB().QueryRow(`SELECT count(*) FROM schema_migrations`).Scan(&migrationCount); err != nil || migrationCount != 2 { + if err := reopened.DB().QueryRow(`SELECT count(*) FROM schema_migrations`).Scan(&migrationCount); err != nil || migrationCount != 4 { t.Fatalf("reopened migration count = %d, err = %v", migrationCount, err) } if err := reopened.Close(); err != nil {