vm: add the proxyapp support (#3269)

* vm: add pool.Close() support
* vm: add proxyapp client implementation
* vm/proxyapp: autogenerate mocks
* vm/proxyapp: add proxyapp tests
* pkg/mgrconfig: add proxyapp type tests

1 files changed, 503 insertions, 0 deletions

diff --git a/vm/proxyapp/proxyappclient_test.go b/vm/proxyapp/proxyappclient_test.go
new file mode 100644
index 000000000..db9649672
--- /dev/null
+++ b/vm/proxyapp/proxyappclient_test.go
@@ -0,0 +1,503 @@
+// Copyright 2022 syzkaller project authors. All rights reserved.
+// Use of this source code is governed by Apache 2 LICENSE that can be found in the LICENSE file.
+
+package proxyapp
+
+import (
+	"bytes"
+	"fmt"
+	"io"
+	"net/rpc"
+	"net/rpc/jsonrpc"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/google/syzkaller/pkg/report"
+	"github.com/google/syzkaller/vm/proxyapp/mocks"
+	"github.com/google/syzkaller/vm/proxyapp/proxyrpc"
+	"github.com/google/syzkaller/vm/vmimpl"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
+)
+
+var testEnv = &vmimpl.Env{
+	Config: []byte(`
+{
+    "cmd": "/path/to/proxyapp_binary",
+    "config": {
+			"internal_values": 123
+    }
+  }
+`)}
+
+func makeTestParams() *proxyAppParams {
+	return &proxyAppParams{
+		CommandRunner:  osutilCommandContext,
+		InitRetryDelay: 0,
+	}
+}
+
+func makeMockProxyAppProcess(t *testing.T) (
+	*mock.Mock, io.WriteCloser, io.ReadCloser, io.ReadCloser) {
+	rStdin, wStdin := io.Pipe()
+	rStdout, wStdout := io.Pipe()
+	rStderr, wStderr := io.Pipe()
+	wStderr.Close()
+
+	server := rpc.NewServer()
+	handler := mocks.NewProxyAppInterface(t)
+	server.RegisterName("ProxyVM", struct{ proxyrpc.ProxyAppInterface }{handler})
+
+	go server.ServeCodec(jsonrpc.NewServerCodec(stdInOutCloser{
+		rStdin,
+		wStdout,
+	}))
+
+	return &handler.Mock, wStdin, rStdout, rStderr
+}
+
+type nopWriteCloser struct {
+	io.Writer
+}
+
+func (nopWriteCloser) Close() error {
+	return nil
+}
+
+func TestCtor_Ok(t *testing.T) {
+	_, mCmdRunner, params := proxyAppServerFixture(t)
+	p, err := ctor(params, testEnv)
+
+	assert.Nil(t, err)
+	assert.Equal(t, 2, p.Count())
+
+	<-mCmdRunner.onWaitCalled
+}
+
+func TestCtor_ReadBadConfig(t *testing.T) {
+	pool, err := ctor(makeTestParams(), &vmimpl.Env{
+		Config: []byte(`{"wrong_key": 1}`),
+	})
+	assert.NotNil(t, err)
+	assert.Nil(t, pool)
+}
+
+func TestCtor_FailedPipes(t *testing.T) {
+	mCmdRunner, params := makeMockCommandRunner(t)
+	mCmdRunner.
+		On("StdinPipe").
+		Return(nil, fmt.Errorf("stdinpipe error")).
+		Once().
+		On("StdinPipe").
+		Return(nopWriteCloser{&bytes.Buffer{}}, nil).
+		On("StdoutPipe").
+		Return(nil, fmt.Errorf("stdoutpipe error")).
+		Once().
+		On("StdoutPipe").
+		Return(io.NopCloser(strings.NewReader("")), nil).
+		On("StderrPipe").
+		Return(nil, fmt.Errorf("stderrpipe error"))
+
+	for i := 0; i < 3; i++ {
+		p, err := ctor(params, testEnv)
+		assert.NotNil(t, err)
+		assert.Nil(t, p)
+	}
+}
+
+func TestClose_waitDone(t *testing.T) {
+	_, mCmdRunner, params := proxyAppServerFixture(t)
+	mCmdRunner.
+		On("waitDone").
+		Return(nil)
+
+	p, _ := ctor(params, testEnv)
+	p.(io.Closer).Close()
+}
+
+func TestCtor_FailedStartProxyApp(t *testing.T) {
+	mCmdRunner, params := makeMockCommandRunner(t)
+	mCmdRunner.
+		On("StdinPipe").
+		Return(nopWriteCloser{&bytes.Buffer{}}, nil).
+		On("StdoutPipe").
+		Return(io.NopCloser(strings.NewReader("")), nil).
+		On("StderrPipe").
+		Return(io.NopCloser(strings.NewReader("")), nil).
+		On("Start").
+		Return(fmt.Errorf("failed to start program"))
+
+	p, err := ctor(params, testEnv)
+	assert.NotNil(t, err)
+	assert.Nil(t, p)
+}
+
+// TODO: reuse proxyAppServerFixture() code: func could be called here once Mock.Unset() error
+//  fixed https://github.com/stretchr/testify/issues/1236
+//  nolint: dupl
+func TestCtor_FailedConstructPool(t *testing.T) {
+	mProxyAppServer, stdin, stdout, stderr :=
+		makeMockProxyAppProcess(t)
+
+	mProxyAppServer.
+		On("CreatePool", mock.Anything, mock.Anything).
+		Return(fmt.Errorf("failed to construct pool"))
+
+	mCmdRunner, params := makeMockCommandRunner(t)
+	mCmdRunner.
+		On("StdinPipe").
+		Return(stdin, nil).
+		On("StdoutPipe").
+		Return(stdout, nil).
+		On("StderrPipe").
+		Return(stderr, nil).
+		On("Start").
+		Return(nil).
+		On("Wait").
+		Run(func(args mock.Arguments) {
+			<-mCmdRunner.ctx.Done()
+		}).
+		Return(nil)
+
+	p, err := ctor(params, testEnv)
+	assert.NotNil(t, err)
+	assert.Nil(t, p)
+}
+
+// TODO: to remove duplicate see TestCtor_FailedConstructPool() comment
+//  nolint: dupl
+func proxyAppServerFixture(t *testing.T) (*mock.Mock, *mockCommandRunner, *proxyAppParams) {
+	mProxyAppServer, stdin, stdout, stderr :=
+		makeMockProxyAppProcess(t)
+
+	mProxyAppServer.
+		On("CreatePool", mock.Anything, mock.Anything).
+		Run(func(args mock.Arguments) {
+			out := args.Get(1).(*proxyrpc.CreatePoolResult)
+			out.Count = 2
+		}).
+		Return(nil)
+
+	mCmdRunner, params := makeMockCommandRunner(t)
+	mCmdRunner.
+		On("StdinPipe").
+		Return(stdin, nil).
+		On("StdoutPipe").
+		Return(stdout, nil).
+		On("StderrPipe").
+		Return(stderr, nil).
+		On("Start").
+		Return(nil).
+		On("Wait").
+		Run(func(args mock.Arguments) {
+			<-mCmdRunner.ctx.Done()
+			mCmdRunner.MethodCalled("waitDone")
+		}).
+		Return(nil).
+		Maybe()
+
+	return mProxyAppServer, mCmdRunner, params
+}
+
+func poolFixture(t *testing.T) (*mock.Mock, *mockCommandRunner, vmimpl.Pool) {
+	mProxyAppServer, mCmdRunner, params := proxyAppServerFixture(t)
+	p, _ := ctor(params, testEnv)
+	return mProxyAppServer, mCmdRunner, p
+}
+
+func TestPool_Create_Ok(t *testing.T) {
+	mockServer, _, p := poolFixture(t)
+	mockServer.
+		On("CreateInstance", mock.Anything, mock.Anything).
+		Return(nil)
+
+	inst, err := p.Create("workdir", 0)
+	assert.NotNil(t, inst)
+	assert.Nil(t, err)
+}
+
+func TestPool_Create_ProxyNilError(t *testing.T) {
+	_, mCmdRunner, p := poolFixture(t)
+	mCmdRunner.
+		On("waitDone").
+		Return(nil)
+
+	p.(io.Closer).Close()
+
+	inst, err := p.Create("workdir", 0)
+	assert.Nil(t, inst)
+	assert.NotNil(t, err)
+}
+
+func TestPool_Create_OutOfPoolError(t *testing.T) {
+	mockServer, _, p := poolFixture(t)
+	mockServer.
+		On("CreateInstance", mock.Anything, mock.Anything).
+		Run(func(args mock.Arguments) {
+			in := args.Get(0).(proxyrpc.CreateInstanceParams)
+			assert.Equal(t, p.Count(), in.Index)
+		}).
+		Return(fmt.Errorf("out of pool size"))
+
+	inst, err := p.Create("workdir", p.Count())
+	assert.Nil(t, inst)
+	assert.NotNil(t, err)
+}
+
+func TestPool_Create_ProxyFailure(t *testing.T) {
+	mockServer, _, p := poolFixture(t)
+	mockServer.
+		On("CreateInstance", mock.Anything, mock.Anything).
+		Return(fmt.Errorf("create instance failure"))
+
+	inst, err := p.Create("workdir", 0)
+	assert.Nil(t, inst)
+	assert.NotNil(t, err)
+}
+
+// nolint: dupl
+func createInstanceFixture(t *testing.T) (*mock.Mock, vmimpl.Instance) {
+	mockServer, _, p := poolFixture(t)
+	mockServer.
+		On("CreateInstance", mock.Anything, mock.Anything).
+		Run(func(args mock.Arguments) {
+			in := args.Get(0).(proxyrpc.CreateInstanceParams)
+			out := args.Get(1).(*proxyrpc.CreateInstanceResult)
+			out.ID = fmt.Sprintf("instance_id_%v", in.Index)
+		}).
+		Return(nil)
+
+	inst, err := p.Create("workdir", 0)
+	assert.Nil(t, err)
+	assert.NotNil(t, inst)
+
+	return mockServer, inst
+}
+
+func TestInstance_Close(t *testing.T) {
+	mockInstance, inst := createInstanceFixture(t)
+	mockInstance.
+		On("Close", mock.Anything, mock.Anything).
+		Return(fmt.Errorf("mock error"))
+
+	inst.Close()
+}
+
+func TestInstance_Diagnose_Ok(t *testing.T) {
+	mockInstance, inst := createInstanceFixture(t)
+	mockInstance.
+		On("Diagnose", mock.Anything, mock.Anything).
+		Run(func(args mock.Arguments) {
+			out := args.Get(1).(*proxyrpc.DiagnoseReply)
+			out.Diagnosis = "diagnostic result"
+		}).
+		Return(nil)
+
+	diagnosis, wait := inst.Diagnose(nil)
+	assert.NotNil(t, diagnosis)
+	assert.Equal(t, wait, false)
+
+	diagnosis, wait = inst.Diagnose(&report.Report{})
+	assert.NotNil(t, diagnosis)
+	assert.Equal(t, wait, false)
+}
+
+func TestInstance_Diagnose_Failure(t *testing.T) {
+	mockInstance, inst := createInstanceFixture(t)
+	mockInstance.
+		On("Diagnose", mock.Anything, mock.Anything).
+		Return(fmt.Errorf("diagnose failed"))
+
+	diagnosis, wait := inst.Diagnose(&report.Report{})
+	assert.Nil(t, diagnosis)
+	assert.Equal(t, wait, false)
+}
+
+func TestInstance_Copy_OK(t *testing.T) {
+	mockInstance, inst := createInstanceFixture(t)
+	mockInstance.
+		On("Copy", mock.Anything, mock.Anything).
+		Run(func(args mock.Arguments) {
+			out := args.Get(1).(*proxyrpc.CopyResult)
+			out.VMFileName = "remote_file_path"
+		}).
+		Return(nil)
+
+	remotePath, err := inst.Copy("host/path")
+	assert.Nil(t, err)
+	assert.NotEmpty(t, remotePath)
+}
+
+// nolint: dupl
+func TestInstance_Copy_Failure(t *testing.T) {
+	mockInstance, inst := createInstanceFixture(t)
+	mockInstance.
+		On("Copy", mock.Anything, mock.Anything).
+		Return(fmt.Errorf("copy failure"))
+
+	remotePath, err := inst.Copy("host/path")
+	assert.NotNil(t, err)
+	assert.Empty(t, remotePath)
+}
+
+// nolint: dupl
+func TestInstance_Forward_OK(t *testing.T) {
+	mockInstance, inst := createInstanceFixture(t)
+	mockInstance.
+		On("Forward", mock.Anything, mock.Anything).
+		Run(func(args mock.Arguments) {
+			in := args.Get(0).(proxyrpc.ForwardParams)
+			out := args.Get(1).(*proxyrpc.ForwardResult)
+			out.ManagerAddress = fmt.Sprintf("manager_address:%v", in.Port)
+		}).
+		Return(nil)
+
+	remoteAddressToUse, err := inst.Forward(12345)
+	assert.Nil(t, err)
+	assert.Equal(t, "manager_address:12345", remoteAddressToUse)
+}
+
+// nolint: dupl
+func TestInstance_Forward_Failure(t *testing.T) {
+	mockInstance, inst := createInstanceFixture(t)
+	mockInstance.
+		On("Forward", mock.Anything, mock.Anything).
+		Return(fmt.Errorf("forward failure"))
+
+	remoteAddressToUse, err := inst.Forward(12345)
+	assert.NotNil(t, err)
+	assert.Empty(t, remoteAddressToUse)
+}
+
+func TestInstance_Run_SimpleOk(t *testing.T) {
+	mockInstance, inst := createInstanceFixture(t)
+	mockInstance.
+		On("RunStart", mock.Anything, mock.Anything).
+		Return(nil).
+		On("RunReadProgress", mock.Anything, mock.Anything).
+		Return(nil).
+		Maybe()
+
+	outc, errc, err := inst.Run(10*time.Second, make(chan bool), "command")
+	assert.NotNil(t, outc)
+	assert.NotNil(t, errc)
+	assert.Nil(t, err)
+}
+
+func TestInstance_Run_Failure(t *testing.T) {
+	mockInstance, inst := createInstanceFixture(t)
+	mockInstance.
+		On("RunStart", mock.Anything, mock.Anything).
+		Return(fmt.Errorf("run start error"))
+
+	outc, errc, err := inst.Run(10*time.Second, make(chan bool), "command")
+	assert.Nil(t, outc)
+	assert.Nil(t, errc)
+	assert.NotEmpty(t, err)
+}
+
+func TestInstance_Run_OnTimeout(t *testing.T) {
+	mockInstance, inst := createInstanceFixture(t)
+	mockInstance.
+		On("RunStart", mock.Anything, mock.Anything).
+		Return(nil).
+		On("RunReadProgress", mock.Anything, mock.Anything).
+		Return(nil).Maybe().
+		On("RunStop", mock.Anything, mock.Anything).
+		Return(nil)
+
+	_, errc, _ := inst.Run(time.Second, make(chan bool), "command")
+	err := <-errc
+
+	assert.Equal(t, err, vmimpl.ErrTimeout)
+}
+
+func TestInstance_Run_OnStop(t *testing.T) {
+	mockInstance, inst := createInstanceFixture(t)
+	mockInstance.
+		On("RunStart", mock.Anything, mock.Anything).
+		Return(nil).
+		On("RunReadProgress", mock.Anything, mock.Anything).
+		Return(nil).
+		Maybe().
+		On("RunStop", mock.Anything, mock.Anything).
+		Return(nil)
+
+	stop := make(chan bool)
+	_, errc, _ := inst.Run(10*time.Second, stop, "command")
+	stop <- true
+	err := <-errc
+	assert.Equal(t, err, vmimpl.ErrTimeout)
+}
+
+func TestInstance_RunReadProgress_OnErrorReceived(t *testing.T) {
+	mockInstance, inst := createInstanceFixture(t)
+	mockInstance.
+		On("RunStart", mock.Anything, mock.Anything).
+		Return(nil).
+		On("RunReadProgress", mock.Anything, mock.Anything).
+		Return(nil).
+		Times(100).
+		On("RunReadProgress", mock.Anything, mock.Anything).
+		Run(func(args mock.Arguments) {
+			out := args.Get(1).(*proxyrpc.RunReadProgressReply)
+			out.Error = "mock error"
+		}).
+		Return(nil).
+		Once()
+
+	outc, _, _ := inst.Run(10*time.Second, make(chan bool), "command")
+	output := string(<-outc)
+
+	assert.Equal(t, "mock error\nSYZFAIL: proxy app plugin error\n", output)
+}
+
+// nolint: dupl
+func TestInstance_RunReadProgress_OnFinished(t *testing.T) {
+	mockInstance, inst := createInstanceFixture(t)
+	mockInstance.
+		On("RunStart", mock.Anything, mock.Anything).
+		Return(nil).
+		On("RunReadProgress", mock.Anything, mock.Anything).
+		Return(nil).Times(100).
+		On("RunReadProgress", mock.Anything, mock.Anything).
+		Run(func(args mock.Arguments) {
+			out := args.Get(1).(*proxyrpc.RunReadProgressReply)
+			out.Finished = true
+		}).
+		Return(nil).
+		Once()
+
+	_, errc, _ := inst.Run(10*time.Second, make(chan bool), "command")
+	err := <-errc
+
+	assert.Equal(t, err, nil)
+}
+
+func TestInstance_RunReadProgress_Failed(t *testing.T) {
+	mockInstance, inst := createInstanceFixture(t)
+	mockInstance.
+		On("RunStart", mock.Anything, mock.Anything).
+		Run(func(args mock.Arguments) {
+			out := args.Get(1).(*proxyrpc.RunStartReply)
+			out.RunID = "test_run_id"
+		}).
+		Return(nil).
+		On("RunReadProgress", mock.Anything, mock.Anything).
+		Return(fmt.Errorf("runreadprogresserror")).
+		Once()
+
+	outc, _, _ := inst.Run(10*time.Second, make(chan bool), "command")
+	output := string(<-outc)
+
+	assert.Equal(t,
+		"error reading progress from instance_id_0:test_run_id: runreadprogresserror\nSYZFAIL: proxy app plugin error\n",
+		output,
+	)
+}
+
+// TODO: test for periodical proxyapp subprocess crashes handling.
+//  [option] check pool size was changed
+
+// TODO: test pool.Close() calls plugin API and return error.