Use named pipe to decrease local Windows runtime (#220)

* Use named pipe to decrease local Windows runtime

This uses PowerShell to spawn a process that listens on a named pipe for
Base64 encoded commands and executes them. This drastically decreases
the total runtime when running locally on Windows by using a single
PowerShell session to execute commands instead of spawning a new session
for each command.

Huge thanks to @sdelano for the initial idea!

* Modify named pipe creation to use unique name
* Remove `$ProgressPreference = 'SilentlyContinue'`
* Modify code to persist pipe usage
* Removed pipe argument (default is In/Out)
* Privatize pipe related functions
* Add fallback to not used pipe if creation fails
* Move Windows local command logic to separate class
* Revert file code format
* Change `@platform` to `@os` for consistency
* Split Windows runners and create a `GenericRunner`
* Add test for Windows named pipe usage
* Tweak `GenericRunner` and default `run_command`
* Remove `Local::` from `CommandWrapper`
* Fix test for Windows named pipe
* Make changes to integrate @jquick's cacheing magic
* Remove unneeded logic for opening an existing pipe

A pipe is opened only once per object, multiple processes shouldn't
share the same pipe because it could be terminated by the other process.

* Add comments explaining pipe check tests
* Change `Shellout` to `ShellOut`
* Change `stubs` to `expects` where appropriate
* Modify pipe tests
* Move/Modify Unit tests and add Integration tests
* Fix local tests when running with others.
* Minor style changes
* Remove extra newline between `let`s
* Modify TransportHelper to accept platform options

Signed-off-by: Jerry Aldrich <jerryaldrichiii@gmail.com>
Signed-off-by: Jared Quick <jquick@chef.io>

lib/train/extras/command_wrapper.rb

@@ -121,43 +121,6 @@ def safe_string(str)
     end
   end
 
-  # this is required if you run locally on windows,
-  # winrm connections provide a PowerShell shell automatically
-  # TODO: only activate in local mode
-  class PowerShellCommand < CommandWrapperBase
-    Train::Options.attach(self)
-
-    def initialize(backend, options)
-      @backend = backend
-      validate_options(options)
-    end
-
-    def run(script)
-      # wrap the script to ensure we always run it via powershell
-      # especially in local mode, we cannot be sure that we get a Powershell
-      # we may just get a `cmd`.
-      # TODO: we may want to opt for powershell.exe -command instead of `encodeCommand`
-      "powershell -NoProfile -encodedCommand #{encoded(safe_script(script))}"
-    end
-
-    # suppress the progress stream from leaking to stderr
-    def safe_script(script)
-      "$ProgressPreference='SilentlyContinue';" + script
-    end
-
-    # Encodes the script so that it can be passed to the PowerShell
-    # --EncodedCommand argument.
-    # @return [String] The UTF-16LE base64 encoded script
-    def encoded(script)
-      encoded_script = safe_script(script).encode('UTF-16LE', 'UTF-8')
-      Base64.strict_encode64(encoded_script)
-    end
-
-    def to_s
-      'PowerShell CommandWrapper'
-    end
-  end
-
   class CommandWrapper
     include_options LinuxCommand
 
@@ -168,10 +131,6 @@ def self.load(transport, options)
         msg = res.verify
         fail Train::UserError, "Sudo failed: #{msg}" unless msg.nil?
         res
-      # only use powershell command for local transport. winrm transport
-      # uses powershell as default
-      elsif transport.os.windows? && transport.class == Train::Transports::Local::Connection
-        PowerShellCommand.new(transport, options)
       end
     end
   end

lib/train/transports/local.rb

@@ -12,15 +12,32 @@ class Local < Train.plugin(1)
 
     include_options Train::Extras::CommandWrapper
 
+    class PipeError < ::StandardError; end
+
     def connection(_ = nil)
       @connection ||= Connection.new(@options)
     end
 
     class Connection < BaseConnection
       def initialize(options)
         super(options)
-        @cmd_wrapper = nil
-        @cmd_wrapper = CommandWrapper.load(self, options)
+
+        # While OS is being discovered, use the GenericRunner
+        @runner = GenericRunner.new
+        @runner.cmd_wrapper = CommandWrapper.load(self, options)
+
+        if os.windows?
+          # Attempt to use a named pipe but fallback to ShellOut if that fails
+          begin
+            @runner = WindowsPipeRunner.new
+          rescue PipeError
+            @runner = WindowsShellRunner.new
+          end
+        end
+      end
+
+      def local?
+        true
       end
 
       def login_command
@@ -31,17 +48,10 @@ def uri
         'local://'
       end
 
-      def local?
-        true
-      end
-
       private
 
       def run_command_via_connection(cmd)
-        cmd = @cmd_wrapper.run(cmd) unless @cmd_wrapper.nil?
-        res = Mixlib::ShellOut.new(cmd)
-        res.run_command
-        CommandResult.new(res.stdout, res.stderr, res.exitstatus)
+        @runner.run_command(cmd)
       rescue Errno::ENOENT => _
         CommandResult.new('', '', 1)
       end
@@ -53,6 +63,128 @@ def file_via_connection(path)
           Train::File::Local::Unix.new(self, path)
         end
       end
+
+      class GenericRunner
+        attr_writer :cmd_wrapper
+
+        def run_command(cmd)
+          if defined?(@cmd_wrapper) && !@cmd_wrapper.nil?
+            cmd = @cmd_wrapper.run(cmd)
+          end
+
+          res = Mixlib::ShellOut.new(cmd)
+          res.run_command
+          Local::CommandResult.new(res.stdout, res.stderr, res.exitstatus)
+        end
+      end
+
+      class WindowsShellRunner
+        require 'json'
+        require 'base64'
+
+        def run_command(script)
+          # Prevent progress stream from leaking into stderr
+          script = "$ProgressPreference='SilentlyContinue';" + script
+
+          # Encode script so PowerShell can use it
+          script = script.encode('UTF-16LE', 'UTF-8')
+          base64_script = Base64.strict_encode64(script)
+
+          cmd = "powershell -NoProfile -EncodedCommand #{base64_script}"
+
+          res = Mixlib::ShellOut.new(cmd)
+          res.run_command
+          Local::CommandResult.new(res.stdout, res.stderr, res.exitstatus)
+        end
+      end
+
+      class WindowsPipeRunner
+        require 'json'
+        require 'base64'
+        require 'securerandom'
+
+        def initialize
+          @pipe = acquire_pipe
+          fail PipeError if @pipe.nil?
+        end
+
+        def run_command(cmd)
+          script = "$ProgressPreference='SilentlyContinue';" + cmd
+          encoded_script = Base64.strict_encode64(script)
+          @pipe.puts(encoded_script)
+          @pipe.flush
+          res = OpenStruct.new(JSON.parse(Base64.decode64(@pipe.readline)))
+          Local::CommandResult.new(res.stdout, res.stderr, res.exitstatus)
+        end
+
+        private
+
+        def acquire_pipe
+          pipe_name = "inspec_#{SecureRandom.hex}"
+
+          start_pipe_server(pipe_name)
+
+          pipe = nil
+
+          # PowerShell needs time to create pipe.
+          100.times do
+            begin
+              pipe = open("//./pipe/#{pipe_name}", 'r+')
+              break
+            rescue
+              sleep 0.1
+            end
+          end
+
+          pipe
+        end
+
+        def start_pipe_server(pipe_name)
+          require 'win32/process'
+
+          script = <<-EOF
+            $ErrorActionPreference = 'Stop'
+
+            $pipeServer = New-Object System.IO.Pipes.NamedPipeServerStream('#{pipe_name}')
+            $pipeReader = New-Object System.IO.StreamReader($pipeServer)
+            $pipeWriter = New-Object System.IO.StreamWriter($pipeServer)
+
+            $pipeServer.WaitForConnection()
+
+            # Create loop to receive and process user commands/scripts
+            $clientConnected = $true
+            while($clientConnected) {
+              $input = $pipeReader.ReadLine()
+              $command = [System.Text.Encoding]::UTF8.GetString([System.Convert]::FromBase64String($input))
+
+              # Execute user command/script and convert result to JSON
+              $scriptBlock = $ExecutionContext.InvokeCommand.NewScriptBlock($command)
+              try {
+                $stdout = & $scriptBlock | Out-String
+                $result = @{ 'stdout' = $stdout ; 'stderr' = ''; 'exitstatus' = 0 }
+              } catch {
+                $stderr = $_ | Out-String
+                $result = @{ 'stdout' = ''; 'stderr' = $_; 'exitstatus' = 1 }
+              }
+              $resultJSON = $result | ConvertTo-JSON
+
+              # Encode JSON in Base64 and write to pipe
+              $encodedResult = [System.Convert]::ToBase64String([System.Text.Encoding]::UTF8.GetBytes($resultJSON))
+              $pipeWriter.WriteLine($encodedResult)
+              $pipeWriter.Flush()
+            }
+          EOF
+
+          utf8_script = script.encode('UTF-16LE', 'UTF-8')
+          base64_script = Base64.strict_encode64(utf8_script)
+          cmd = "powershell -NoProfile -ExecutionPolicy bypass -NonInteractive -EncodedCommand #{base64_script}"
+
+          server_pid = Process.create(command_line: cmd).process_id
+
+          # Ensure process is killed when the Train process exits
+          at_exit { Process.kill('KILL', server_pid) }
+        end
+      end
     end
   end
 end

test/unit/transports/local_test.rb

@@ -6,9 +6,11 @@
 class TransportHelper
   attr_accessor :transport
 
-  def initialize
+  def initialize(user_opts = {})
+    opts = {platform_name: 'mock', family_hierarchy: ['mock']}.merge(user_opts)
     Train::Platforms::Detect::Specifications::OS.load
-    plat = Train::Platforms.name('mock').in_family('linux')
+    plat = Train::Platforms.name(opts[:platform_name])
+    plat.family_hierarchy = opts[:family_hierarchy]
     plat.add_platform_methods
     Train::Platforms::Detect.stubs(:scan).returns(plat)
     @transport = Train::Transports::Local.new
@@ -93,4 +95,37 @@ def mock_run_cmd(cmd, &block)
       end
     end
   end
+
+  describe 'when running on Windows' do
+    let(:connection) do
+      TransportHelper.new(family_hierarchy: ['windows']).transport.connection
+    end
+    let(:runner) { mock }
+
+    it 'uses `WindowsPipeRunner` by default' do
+      Train::Transports::Local::Connection::WindowsPipeRunner
+        .expects(:new)
+        .returns(runner)
+
+      Train::Transports::Local::Connection::WindowsShellRunner
+        .expects(:new)
+        .never
+
+      runner.expects(:run_command).with('not actually executed')
+      connection.run_command('not actually executed')
+    end
+
+    it 'uses `WindowsShellRunner` when a named pipe is not available' do
+      Train::Transports::Local::Connection::WindowsPipeRunner
+        .expects(:new)
+        .raises(Train::Transports::Local::PipeError)
+
+      Train::Transports::Local::Connection::WindowsShellRunner
+        .expects(:new)
+        .returns(runner)
+
+      runner.expects(:run_command).with('not actually executed')
+      connection.run_command('not actually executed')
+    end
+  end
 end

test/windows/local_test.rb

@@ -8,6 +8,9 @@
 require 'train'
 require 'tempfile'
 
+# Loading here to ensure methods exist to be stubbed
+require 'train/transports/local'
+
 describe 'windows local command' do
   let(:conn) {
     # get final config
@@ -40,6 +43,38 @@
     cmd.stderr.must_equal ''
   end
 
+  it 'can execute a command using a named pipe' do
+    SecureRandom.expects(:hex).returns('via_pipe')
+
+    Train::Transports::Local::Connection::WindowsShellRunner
+      .any_instance
+      .expects(:new)
+      .never
+
+    cmd = conn.run_command('Write-Output "Create pipe"')
+    File.exist?('//./pipe/inspec_via_pipe').must_equal true
+    cmd.stdout.must_equal "Create pipe\r\n"
+    cmd.stderr.must_equal ''
+  end
+
+  it 'can execute a command via ShellRunner if pipe creation fails' do
+    # By forcing `acquire_pipe` to fail to return a pipe, any attempts to create
+    # a `WindowsPipeRunner` object should fail. If we can still run a command,
+    # then we know that it was successfully executed by `Mixlib::ShellOut`.
+    Train::Transports::Local::Connection::WindowsPipeRunner
+      .any_instance
+      .expects(:acquire_pipe)
+      .at_least_once
+      .returns(nil)
+
+    proc { Train::Transports::Local::Connection::WindowsPipeRunner.new }
+      .must_raise(Train::Transports::Local::PipeError)
+
+    cmd = conn.run_command('Write-Output "test"')
+    cmd.stdout.must_equal "test\r\n"
+    cmd.stderr.must_equal ''
+  end
+
   describe 'file' do
     before do
       @temp = Tempfile.new('foo')