Add support for package executors and support for PTY handling (#100)

* define contract for package executors

* introduce npx executor

* add npx and pnpx cmd support

* fix typo

* rm PackageExecutor and depend on PackageManager interface

* add support for PTY to handle parent-child process interaction

* refactor PTY handling in proxy flow

* enforce interactiveSession interface check

* close reader explicitly and clean npm version for pkg executors

* rm interaction from interceptors

* add docs and wait for outputRouter before exit

* add support for non interactive TTY for proxy mode

* add support for CI env var check for non interactive tty proxy mode

* update readme to include npx, pnpx support

* Update internal/flows/proxy_flow.go

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
Signed-off-by: Sahil Bansal <bansalsahil315@gmail.com>

* update ptyx lib

* fix docs typo

---------

Signed-off-by: Sahil Bansal <bansalsahil315@gmail.com>
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

internal/analytics/event.go

@@ -11,6 +11,9 @@ const (
 	eventCommandUv     = "pmg_command_uv"
 	eventCommandPoetry = "pmg_command_poetry"
 
+	eventCommandNpx  = "pmg_command_npx"
+	eventCommandPnpx = "pmg_command_pnpx"
+
 	eventPmgGenerateEnvDocker        = "pmg_command_generate_env_docker"
 	eventPmgGenerateEnvGitHubActions = "pmg_command_generate_env_github_actions"
 	eventPmgGenerateEnvGitLabCI      = "pmg_command_generate_env_gitlab_ci"
@@ -24,6 +27,14 @@ func TrackCommandNpm() {
 	TrackEvent(eventCommandNpm)
 }
 
+func TrackCommandNpx() {
+	TrackEvent(eventCommandNpx)
+}
+
+func TrackCommandPnpx() {
+	TrackEvent(eventCommandPnpx)
+}
+
 func TrackCommandBun() {
 	TrackEvent(eventCommandBun)
 }

internal/flows/proxy_flow.go

@@ -2,16 +2,20 @@ package flows
 
 import (
 	"context"
+	"errors"
 	"fmt"
+	"io"
 	"os"
 	"os/exec"
 	"path/filepath"
+	"sync"
 	"time"
 
 	"github.com/safedep/dry/log"
 	"github.com/safedep/pmg/analyzer"
 	"github.com/safedep/pmg/config"
 	"github.com/safedep/pmg/guard"
+	"github.com/safedep/pmg/internal/pty"
 	"github.com/safedep/pmg/internal/ui"
 	"github.com/safedep/pmg/packagemanager"
 	"github.com/safedep/pmg/proxy"
@@ -34,6 +38,15 @@ func ProxyFlow(pm packagemanager.PackageManager, packageResolver packagemanager.
 
 // Run executes the proxy-based flow
 func (f *proxyFlow) Run(ctx context.Context, args []string, parsedCmd *packagemanager.ParsedCommand) error {
+
+	// Get the ecosystem from the package manager
+	ecosystem := f.pm.Ecosystem()
+
+	// Check if proxy mode is supported for this ecosystem
+	if !interceptors.IsSupported(ecosystem) {
+		return fmt.Errorf("proxy mode is not supported for %s", ecosystem.String())
+	}
+
 	cfg := config.Get()
 
 	// Check if dry-run mode is enabled
@@ -82,24 +95,16 @@ func (f *proxyFlow) Run(ctx context.Context, args []string, parsedCmd *packagema
 	defer close(confirmationChan)
 
 	// Create interaction callbacks for user prompts
-	interaction := guard.PackageManagerGuardInteraction{
-		SetStatus:                ui.SetStatus,
-		ClearStatus:              ui.ClearStatus,
-		ShowWarning:              ui.ShowWarning,
-		GetConfirmationOnMalware: ui.GetConfirmationOnMalware,
-		Block:                    ui.Block,
-	}
-
-	// Get the ecosystem from the package manager
-	ecosystem := f.pm.Ecosystem()
-
-	// Check if proxy mode is supported for this ecosystem
-	if !interceptors.IsSupported(ecosystem) {
-		return fmt.Errorf("proxy mode is not supported for %s", ecosystem.String())
+	// Note: We use a pointer so we can later inject the input reader via SetInput
+	interaction := &guard.PackageManagerGuardInteraction{
+		SetStatus:   ui.SetStatus,
+		ClearStatus: ui.ClearStatus,
+		ShowWarning: ui.ShowWarning,
+		Block:       ui.Block,
 	}
 
 	// Create ecosystem-specific interceptor using factory
-	factory := interceptors.NewInterceptorFactory(malysisAnalyzer, cache, confirmationChan, interaction)
+	factory := interceptors.NewInterceptorFactory(malysisAnalyzer, cache, confirmationChan)
 	interceptor, err := factory.CreateInterceptor(ecosystem)
 	if err != nil {
 		return fmt.Errorf("failed to create interceptor for %s: %w", ecosystem.String(), err)
@@ -127,8 +132,15 @@ func (f *proxyFlow) Run(ctx context.Context, args []string, parsedCmd *packagema
 	log.Infof("Proxy server started on %s", proxyAddr)
 	log.Infof("Running %s with proxy protection enabled", f.pm.Name())
 
+	proxyEnv := f.setupEnvForProxy(proxyAddr, caCertPath)
+
+	if !pty.IsInteractiveTerminal() {
+		// Execute the package manager command with proxy environment variables for non PTY or non-interactive TTY
+		return f.executeWithProxyForNonInteractiveTTY(ctx, parsedCmd, proxyEnv, confirmationChan, interaction)
+	}
+
 	// Execute the package manager command with proxy environment variables
-	return f.executeWithProxy(ctx, parsedCmd, proxyAddr, caCertPath, confirmationChan, interaction)
+	return f.executeWithProxy(ctx, parsedCmd, proxyEnv, confirmationChan, interaction)
 }
 
 // setupCACertificate generates or loads a CA certificate for MITM
@@ -211,21 +223,11 @@ func (f *proxyFlow) createAndStartProxyServer(
 	return proxyServer, proxyAddr, nil
 }
 
-// executeWithProxy executes the package manager command with proxy environment variables
-func (f *proxyFlow) executeWithProxy(ctx context.Context, parsedCmd *packagemanager.ParsedCommand,
-	proxyAddr, caCertPath string, confirmationChan chan *interceptors.ConfirmationRequest,
-	interaction guard.PackageManagerGuardInteraction,
-) error {
-	// Build proxy URL
+func (f *proxyFlow) setupEnvForProxy(proxyAddr, caCertPath string) []string {
 	proxyURL := fmt.Sprintf("http://%s", proxyAddr)
 
-	// Create command
-	cmd := exec.CommandContext(ctx, parsedCmd.Command.Exe, parsedCmd.Command.Args...)
-
-	// Set proxy environment variables. This is what tells the executed command to use the proxy for communication.
-	// However, every package manager has its nuances and may require additional environment variables to be set.
-	cmd.Env = os.Environ()
-	cmd.Env = append(cmd.Env,
+	env := os.Environ()
+	env = append(env,
 		fmt.Sprintf("HTTP_PROXY=%s", proxyURL),
 		fmt.Sprintf("HTTPS_PROXY=%s", proxyURL),
 		fmt.Sprintf("NODE_EXTRA_CA_CERTS=%s", caCertPath),
@@ -237,32 +239,35 @@ func (f *proxyFlow) executeWithProxy(ctx context.Context, parsedCmd *packagemana
 		fmt.Sprintf("PIP_PROXY=%s", proxyURL),
 	)
 
+	return env
+}
+
+// executeWithProxyForNonInteractiveTTY runs the command without PTY (for CI/non-interactive environments)
+func (f *proxyFlow) executeWithProxyForNonInteractiveTTY(
+	ctx context.Context,
+	parsedCmd *packagemanager.ParsedCommand,
+	env []string,
+	confirmationChan chan *interceptors.ConfirmationRequest,
+	interaction *guard.PackageManagerGuardInteraction,
+) error {
+	log.Debugf("Executing proxy for non interactive TTY")
+
+	// For non-interactive terminals, we enforce suspicious packages as malicious
+	interaction.GetConfirmationOnMalware = func(malwarePackages []*analyzer.PackageVersionAnalysisResult) (bool, error) {
+		return false, nil
+	}
+
+	cmd := exec.CommandContext(ctx, parsedCmd.Command.Exe, parsedCmd.Command.Args...)
+	cmd.Env = append(env, "CI=true")
 	cmd.Stdin = os.Stdin
 	cmd.Stdout = os.Stdout
 	cmd.Stderr = os.Stderr
 
-	log.Debugf("Executing command: %s %v", parsedCmd.Command.Exe, parsedCmd.Command.Args)
-	log.Debugf("Proxy environment: HTTP_PROXY=%s, HTTPS_PROXY=%s, NODE_EXTRA_CA_CERTS=%s", proxyURL, proxyURL, caCertPath)
-
-	// Start confirmation handler in goroutine. Use confirmation hooks to pause and resume the executed
-	// process to prevent stdout and stderr from being mixed up. Pause / resume is on a best effort basis.
-	// We do not consider it a critical error if pause / resume fails.
-	go interceptors.HandleConfirmationRequests(confirmationChan, interaction, &interceptors.ConfirmationHook{
-		BeforeInteraction: func([]*analyzer.PackageVersionAnalysisResult) error {
-			if err := platformPauseProcess(cmd); err != nil {
-				log.Warnf("Failed to pause process for user interaction: %v", err)
-			}
-
-			return nil
-		},
-		AfterInteraction: func([]*analyzer.PackageVersionAnalysisResult, bool) error {
-			if err := platformResumeProcess(cmd); err != nil {
-				log.Warnf("Failed to resume process after user interaction: %v", err)
-			}
-
-			return nil
-		},
-	})
+	go interceptors.HandleConfirmationRequests(
+		confirmationChan,
+		interaction,
+		nil,
+	)
 
 	err := cmd.Run()
 	if err != nil {
@@ -276,3 +281,122 @@ func (f *proxyFlow) executeWithProxy(ctx context.Context, parsedCmd *packagemana
 	log.Debugf("Command completed successfully")
 	return nil
 }
+
+// executeWithProxy executes the package manager command with proxy environment variables.
+func (f *proxyFlow) executeWithProxy(
+	ctx context.Context,
+	parsedCmd *packagemanager.ParsedCommand,
+	env []string,
+	confirmationChan chan *interceptors.ConfirmationRequest,
+	interaction *guard.PackageManagerGuardInteraction,
+) error {
+	log.Debugf("Executing proxy for interactive TTY")
+
+	// Set the confirmation handler to use the interaction's reader
+	// This allows PTY input routing during proxy mode
+	interaction.GetConfirmationOnMalware = func(malwarePackages []*analyzer.PackageVersionAnalysisResult) (bool, error) {
+		return ui.GetConfirmationOnMalwareWithReader(malwarePackages, interaction.Reader())
+	}
+
+	sessionConfig := pty.NewSessionConfig(parsedCmd.Command.Exe, parsedCmd.Command.Args, env)
+
+	sess, err := pty.NewSession(ctx, sessionConfig)
+	if err != nil {
+		return fmt.Errorf("failed to create pty session: %w", err)
+	}
+	defer sess.Close()
+
+	outputRouter, err := pty.NewOutputRouter(os.Stdout)
+	if err != nil {
+		return fmt.Errorf("failed to create output router: %w", err)
+	}
+
+	var wg sync.WaitGroup
+	wg.Go(func() {
+		io.Copy(outputRouter, sess.PtyReader())
+	})
+
+	inputRouter, err := pty.NewInputRouter(sess.PtyWriter())
+	if err != nil {
+		return fmt.Errorf("failed to create input router: %w", err)
+	}
+
+	promptReader, promptWriter := io.Pipe()
+	defer func() {
+		promptWriter.Close()
+		promptReader.Close()
+	}()
+
+	// Note: This goroutine cannot be cleanly cancelled because os.Stdin.Read() is
+	// a blocking syscall that doesn't support timeouts or cancellation. This is a
+	// known limitation. The goroutine will exit when the process terminates, which
+	// is acceptable for a CLI tool. For long-running servers, stdin reading should
+	// be handled differently.
+	go inputRouter.ReadLoop(os.Stdin)
+
+	go interceptors.HandleConfirmationRequests(
+		confirmationChan,
+		interaction,
+		&interceptors.ConfirmationHook{
+			BeforeInteraction: func(_ []*analyzer.PackageVersionAnalysisResult) error {
+				// Pause printing the child output
+				outputRouter.Pause()
+
+				// Restore "Cooked" mode so user can type normally with echo
+				if err := sess.SetCookedMode(); err != nil {
+					return fmt.Errorf("failed to set cooked mode: %w", err)
+				}
+
+				// Force cursor visible (ANSI escape sequence)
+				fmt.Fprint(os.Stdout, "\033[?25h")
+
+				// Switch Input: Route keystrokes to the Prompt Pipe
+				inputRouter.RouteToPrompt(promptWriter)
+
+				// Inject the Reader into the Interaction for the confirmation prompt
+				interaction.SetInput(promptReader)
+
+				return nil
+			},
+			AfterInteraction: func(_ []*analyzer.PackageVersionAnalysisResult, _ bool) error {
+				// Switch input back to PTY
+				inputRouter.RouteToPTY()
+
+				// Restore "Raw" mode for the PTY
+				if err := sess.SetRawMode(); err != nil {
+					return fmt.Errorf("failed to set raw mode: %w", err)
+				}
+
+				// Clear the interaction input (back to default)
+				interaction.SetInput(nil)
+
+				// Flush buffered output and resume live output
+				outputRouter.Resume()
+
+				return nil
+			},
+		},
+	)
+
+	err = sess.Wait()
+
+	// Wait for the routers to copy all the remaining data
+	wg.Wait()
+
+	if err != nil {
+		var exitErr *pty.ExitError
+		if errors.As(err, &exitErr) {
+			// Close writer and reader
+			promptWriter.Close()
+			promptReader.Close()
+
+			// Close the session
+			sess.Close()
+
+			os.Exit(exitErr.Code)
+		}
+		return err
+	}
+
+	return nil
+}

internal/pty/router.go

@@ -0,0 +1,115 @@
+package pty
+
+import (
+	"bytes"
+	"io"
+	"sync"
+	"sync/atomic"
+)
+
+// OutputRouter manages buffered vs live output.
+type OutputRouter struct {
+	mu        sync.Mutex
+	stdout    io.Writer
+	buffer    bytes.Buffer
+	buffering bool
+}
+
+func NewOutputRouter(out io.Writer) (*OutputRouter, error) {
+	return &OutputRouter{
+		stdout: out,
+	}, nil
+}
+
+func (r *OutputRouter) Write(p []byte) (n int, err error) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+
+	if r.buffering {
+		// We are in "Prompt Mode", so save this output for later.
+		// If we printed it now, it would mess up the confirmation prompt.
+		return r.buffer.Write(p)
+	}
+
+	// Normal mode: just print it to stdout.
+	return r.stdout.Write(p)
+}
+
+// Pause starts buffering output. Call this before showing a confirmation prompt.
+func (r *OutputRouter) Pause() {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	r.buffering = true
+}
+
+// Resume stops buffering, flushes any buffered output, and resumes live output.
+// Call this after the confirmation prompt is complete.
+func (r *OutputRouter) Resume() {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+
+	// Flush any buffered output
+	if r.buffer.Len() > 0 {
+		_, _ = io.Copy(r.stdout, &r.buffer)
+		r.buffer.Reset()
+	}
+
+	r.buffering = false
+}
+
+// writerDest wraps io.Writer for use with atomic.Pointer
+// (atomic.Value panics on nil interface stores)
+type writerDest struct {
+	w io.Writer
+}
+
+// InputRouter manages routing stdin to either PTY or a prompt pipe.
+// Only ONE goroutine should call ReadLoop().
+type InputRouter struct {
+	dest       atomic.Pointer[writerDest]
+	defaultDst io.Writer // PTY writer
+}
+
+func NewInputRouter(ptyWriter io.Writer) (*InputRouter, error) {
+	return &InputRouter{
+		defaultDst: ptyWriter,
+	}, nil
+}
+
+// ReadLoop continuously reads from src and routes data to the current destination.
+//
+// IMPORTANT: Only ONE goroutine should call ReadLoop() because:
+//  1. Multiple readers on the same source (e.g., stdin) cause data splitting -
+//     one goroutine might read "hel" while another reads "lo\n"
+//  2. Concurrent routing decisions create race conditions on the destination
+//  3. User input becomes unpredictably interleaved between readers
+//
+// This function blocks until src returns an error (e.g., EOF).
+func (r *InputRouter) ReadLoop(src io.Reader) {
+	buf := make([]byte, 1024)
+	for {
+		nr, err := src.Read(buf)
+		if err != nil {
+			return
+		}
+
+		// Check where to route the data
+		if dest := r.dest.Load(); dest != nil {
+			// Send confirmation prompt response to the pipe. (PMG)
+			_, _ = dest.w.Write(buf[:nr])
+		} else {
+			// Send response to the child PTY.
+			_, _ = r.defaultDst.Write(buf[:nr])
+		}
+	}
+}
+
+// RouteToPrompt switches input to go to the given writer (prompt pipe)
+func (r *InputRouter) RouteToPrompt(w io.Writer) {
+	r.dest.Store(&writerDest{w: w})
+}
+
+// RouteToPTY switches input back to the PTY (default)
+func (r *InputRouter) RouteToPTY() {
+	r.dest.Store(nil)
+}

internal/pty/session.go

@@ -0,0 +1,175 @@
+package pty
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"os"
+	"strings"
+
+	"github.com/KennethanCeyer/ptyx"
+	"golang.org/x/term"
+)
+
+// InteractiveSession manages a PTY-based command execution with
+// support for input/output routing and terminal mode switching.
+type InteractiveSession interface {
+	// PtyWriter returns the writer to send input to the child process
+	PtyWriter() io.Writer
+
+	// PtyReader returns the reader to receive output from the child process
+	PtyReader() io.Reader
+
+	// SetRawMode puts terminal in raw mode (for PTY passthrough)
+	SetRawMode() error
+
+	// SetCookedMode restores normal terminal mode (for prompts)
+	SetCookedMode() error
+
+	// Wait blocks until the child process exits
+	// Returns ExitError if process exited with non-zero code
+	Wait() error
+
+	// Close cleans up resources (PTY, terminal state)
+	Close() error
+}
+
+// IsInteractiveTerminal returns true if stdin is a real terminal (TTY).
+// Returns false in CI environments (when the "CI" env var set to "true"),
+// when input is piped, or in non-interactive shells.
+func IsInteractiveTerminal() bool {
+	if ci := os.Getenv("CI"); ci != "" && strings.ToLower(ci) == "true" {
+		return false
+	}
+
+	return term.IsTerminal(int(os.Stdin.Fd()))
+}
+
+var _ InteractiveSession = &session{}
+
+type session struct {
+	console  ptyx.Console
+	spawn    ptyx.Session
+	oldState ptyx.RawState // Saved terminal state for restoration
+}
+
+// SessionConfig holds options for creating a session
+type SessionConfig struct {
+	Command string
+	Args    []string
+	Env     []string
+}
+
+func NewSessionConfig(cmd string, args, env []string) SessionConfig {
+	return SessionConfig{
+		Command: cmd,
+		Args:    args,
+		Env:     env,
+	}
+}
+
+// NewSession creates a new interactive PTY session.
+// The terminal is put into raw mode automatically.
+func NewSession(ctx context.Context, cfg SessionConfig) (InteractiveSession, error) {
+	if cfg.Command == "" {
+		return nil, fmt.Errorf("pty session requires command")
+	}
+
+	// 1. Create console
+	c, err := ptyx.NewConsole()
+	if err != nil {
+		return nil, fmt.Errorf("failed to create console: %w", err)
+	}
+	c.EnableVT()
+
+	// 2. Set raw mode, save old state
+	oldState, err := c.MakeRaw()
+	if err != nil {
+		c.Close()
+		return nil, fmt.Errorf("failed to set raw mode: %w", err)
+	}
+
+	// 3. Get terminal size
+	cols, rows := c.Size()
+
+	// 4. Spawn the process
+	s, err := ptyx.Spawn(ctx, ptyx.SpawnOpts{
+		Prog: cfg.Command,
+		Args: cfg.Args,
+		Cols: cols,
+		Rows: rows,
+		Env:  cfg.Env,
+	})
+	if err != nil {
+		c.Restore(oldState)
+		c.Close()
+		return nil, fmt.Errorf("failed to spawn: %w", err)
+	}
+
+	return &session{
+		console:  c,
+		spawn:    s,
+		oldState: oldState,
+	}, nil
+}
+
+func (s *session) PtyWriter() io.Writer { return s.spawn.PtyWriter() }
+func (s *session) PtyReader() io.Reader { return s.spawn.PtyReader() }
+
+func (s *session) SetRawMode() error {
+	_, err := s.console.MakeRaw()
+	return err
+}
+
+func (s *session) SetCookedMode() error {
+	return s.console.Restore(s.oldState)
+}
+
+func (s *session) Wait() error {
+	err := s.spawn.Wait()
+	if err != nil {
+		if exitErr, ok := err.(*ptyx.ExitError); ok {
+			return &ExitError{Code: exitErr.ExitCode, Err: err}
+		}
+		return &ExitError{Code: -1, Err: err}
+	}
+	return nil
+}
+
+func (s *session) Close() error {
+	// Always restore terminal state
+	if s.oldState != nil {
+		_ = s.console.Restore(s.oldState)
+	}
+
+	if s.spawn != nil {
+		_ = s.spawn.Close()
+	}
+
+	if s.console != nil {
+		_ = s.console.Close()
+	}
+
+	return nil
+}
+
+// ExitError is returned when the child process exits with non-zero code
+type ExitError struct {
+	Code int
+	Err  error // Underlying error from ptyx
+}
+
+func (e *ExitError) Error() string {
+	if e.Code != 0 {
+		return fmt.Sprintf("process exited with code %d", e.Code)
+	}
+	if e.Err != nil {
+		return e.Err.Error()
+	}
+	return "unknown process error"
+}
+
+// Unwrap allows errors.Is and errors.As to work
+func (e *ExitError) Unwrap() error {
+	return e.Err
+}

internal/ui/ui.go

@@ -1,7 +1,9 @@
 package ui
 
 import (
+	"bufio"
 	"fmt"
+	"io"
 	"os"
 	"strings"
 
@@ -76,7 +78,15 @@ func SetStatus(status string) {
 	StartSpinnerWithColor(fmt.Sprintf("ℹ️ %s", status), Colors.Green)
 }
 
+// GetConfirmationOnMalware prompts the user to confirm installation of suspicious packages.
+// It reads from os.Stdin. Use GetConfirmationOnMalwareWithReader for custom input sources.
 func GetConfirmationOnMalware(malwarePackages []*analyzer.PackageVersionAnalysisResult) (bool, error) {
+	return GetConfirmationOnMalwareWithReader(malwarePackages, os.Stdin)
+}
+
+// GetConfirmationOnMalwareWithReader prompts the user to confirm installation of suspicious packages.
+// It reads from the provided reader, allowing for PTY input routing during proxy mode.
+func GetConfirmationOnMalwareWithReader(malwarePackages []*analyzer.PackageVersionAnalysisResult, reader io.Reader) (bool, error) {
 	StopSpinner()
 
 	fmt.Println()
@@ -87,19 +97,19 @@ func GetConfirmationOnMalware(malwarePackages []*analyzer.PackageVersionAnalysis
 	fmt.Println()
 	fmt.Print(Colors.Yellow("Do you want to continue with the installation? (y/N) "))
 
-	var response string
-
-	// We don't care about the error here because we will return false
-	// if the user doesn't provide a valid response
-	_, _ = fmt.Scanln(&response)
-
-	if len(response) == 0 {
-		return false, nil
+	// Use Scanner on the provided reader to support PTY input routing
+	scanner := bufio.NewScanner(reader)
+	if scanner.Scan() {
+		response := strings.ToLower(strings.TrimSpace(scanner.Text()))
+		if response == "y" || response == "yes" || (len(response) > 0 && response[0] == 'y') {
+			return true, nil
+		}
 	}
 
-	response = strings.ToLower(response)
-	if response == "y" || response == "yes" || response[0] == 'y' {
-		return true, nil
+	// Check for scanner errors, but don't treat them as fatal
+	if err := scanner.Err(); err != nil {
+		// On EOF or interrupted read, just return false (deny)
+		return false, nil
 	}
 
 	return false, nil