<commit_before>// Copyright 2016 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 gce allows to use Google Compute Engine (GCE) virtual machines as VMs.
// It is assumed that syz-manager also runs on GCE as VMs are created in the current project/zone.
//
// See https://cloud.google.com/compute/docs for details.
// In particular, how to build GCE-compatible images:
// https://cloud.google.com/compute/docs/tutorials/building-images
// Working with serial console:
// https://cloud.google.com/compute/docs/instances/interacting-with-serial-console
package gce

import (
	"archive/tar"
	"bytes"
	"compress/gzip"
	"fmt"
	"io"
	"io/ioutil"
	"os"
	"os/exec"
	"path/filepath"
	"time"

	"github.com/google/syzkaller/pkg/config"
	"github.com/google/syzkaller/pkg/gce"
	"github.com/google/syzkaller/pkg/gcs"
	. "github.com/google/syzkaller/pkg/log"
	"github.com/google/syzkaller/pkg/osutil"
	"github.com/google/syzkaller/vm/vmimpl"
)

func init() {
	vmimpl.Register("gce", ctor)
}

type Config struct {
	Count        int    // number of VMs to use
	Machine_Type string // GCE machine type (e.g. "n1-highcpu-2")
	GCS_Path     string // GCS path to upload image
}

type Pool struct {
	env      *vmimpl.Env
	cfg      *Config
	GCE      *gce.Context
	gceImage string
}

type instance struct {
	cfg     *Config
	GCE     *gce.Context
	debug   bool
	name    string
	ip      string
	offset  int64
	gceKey  string // per-instance private ssh key associated with the instance
	sshKey  string // ssh key
	sshUser string
	workdir string
	closed  chan bool
}

func ctor(env *vmimpl.Env) (vmimpl.Pool, error) {
	if env.Name == "" {
		return nil, fmt.Errorf("config param name is empty (required for GCE)")
	}
	if env.Image == "" {
		return nil, fmt.Errorf("config param image is empty (required for GCE)")
	}
	cfg := &Config{
		Count: 1,
	}
	if err := config.LoadData(env.Config, cfg); err != nil {
		return nil, fmt.Errorf("failed to parse gce vm config: %v", err)
	}
	if cfg.Count < 1 || cfg.Count > 1000 {
		return nil, fmt.Errorf("invalid config param count: %v, want [1, 1000]", cfg.Count)
	}
	if env.Debug {
		cfg.Count = 1
	}
	if cfg.Machine_Type == "" {
		return nil, fmt.Errorf("machine_type parameter is empty")
	}
	if cfg.GCS_Path == "" {
		return nil, fmt.Errorf("gcs_path parameter is empty")
	}

	GCE, err := gce.NewContext()
	if err != nil {
		return nil, fmt.Errorf("failed to init gce: %v", err)
	}
	Logf(0, "GCE initialized: running on %v, internal IP %v, project %v, zone %v",
		GCE.Instance, GCE.InternalIP, GCE.ProjectID, GCE.ZoneID)

	gcsImage := filepath.Join(cfg.GCS_Path, env.Name+"-image.tar.gz")
	Logf(0, "uploading image to %v...", gcsImage)
	if err := uploadImageToGCS(env.Image, gcsImage); err != nil {
		return nil, err
	}
	gceImage := env.Name
	Logf(0, "creating GCE image %v...", gceImage)
	if err := GCE.DeleteImage(gceImage); err != nil {
		return nil, fmt.Errorf("failed to delete GCE image: %v", err)
	}
	if err := GCE.CreateImage(gceImage, gcsImage); err != nil {
		return nil, fmt.Errorf("failed to create GCE image: %v", err)
	}
	pool := &Pool{
		cfg:      cfg,
		env:      env,
		GCE:      GCE,
		gceImage: gceImage,
	}
	return pool, nil
}

func (pool *Pool) Count() int {
	return pool.cfg.Count
}

func (pool *Pool) Create(workdir string, index int) (vmimpl.Instance, error) {
	name := fmt.Sprintf("%v-%v", pool.env.Name, index)
	// Create SSH key for the instance.
	gceKey := filepath.Join(workdir, "key")
	keygen := exec.Command("ssh-keygen", "-t", "rsa", "-b", "2048", "-N", "", "-C", "syzkaller", "-f", gceKey)
	if out, err := keygen.CombinedOutput(); err != nil {
		return nil, fmt.Errorf("failed to execute ssh-keygen: %v\n%s", err, out)
	}
	gceKeyPub, err := ioutil.ReadFile(gceKey + ".pub")
	if err != nil {
		return nil, fmt.Errorf("failed to read file: %v", err)
	}

	Logf(0, "deleting instance: %v", name)
	if err := pool.GCE.DeleteInstance(name, true); err != nil {
		return nil, err
	}
	Logf(0, "creating instance: %v", name)
	ip, err := pool.GCE.CreateInstance(name, pool.cfg.Machine_Type, pool.gceImage, string(gceKeyPub))
	if err != nil {
		return nil, err
	}

	ok := false
	defer func() {
		if !ok {
			pool.GCE.DeleteInstance(name, true)
		}
	}()
	sshKey := pool.env.Sshkey
	sshUser := "root"
	if sshKey == "" {
		// Assuming image supports GCE ssh fanciness.
		sshKey = gceKey
		sshUser = "syzkaller"
	}
	Logf(0, "wait instance to boot: %v (%v)", name, ip)
	if err := waitInstanceBoot(pool.env.Debug, ip, sshKey, sshUser); err != nil {
		return nil, err
	}
	ok = true
	inst := &instance{
		cfg:     pool.cfg,
		debug:   pool.env.Debug,
		GCE:     pool.GCE,
		name:    name,
		ip:      ip,
		gceKey:  gceKey,
		sshKey:  sshKey,
		sshUser: sshUser,
		closed:  make(chan bool),
	}
	return inst, nil
}

func (inst *instance) Close() {
	close(inst.closed)
	inst.GCE.DeleteInstance(inst.name, false)
}

func (inst *instance) Forward(port int) (string, error) {
	return fmt.Sprintf("%v:%v", inst.GCE.InternalIP, port), nil
}

func (inst *instance) Copy(hostSrc string) (string, error) {
	vmDst := "./" + filepath.Base(hostSrc)
	args := append(sshArgs(inst.debug, inst.sshKey, "-P", 22), hostSrc, inst.sshUser+"@"+inst.name+":"+vmDst)
	if _, err := runCmd(inst.debug, "scp", args...); err != nil {
		return "", err
	}
	return vmDst, nil
}

func (inst *instance) Run(timeout time.Duration, stop <-chan bool, command string) (<-chan []byte, <-chan error, error) {
	conRpipe, conWpipe, err := osutil.LongPipe()
	if err != nil {
		return nil, nil, err
	}

	conAddr := fmt.Sprintf("%v.%v.%v.syzkaller.port=1@ssh-serialport.googleapis.com",
		inst.GCE.ProjectID, inst.GCE.ZoneID, inst.name)
	conArgs := append(sshArgs(inst.debug, inst.gceKey, "-p", 9600), conAddr)
	con := exec.Command("ssh", conArgs...)
	con.Env = []string{}
	con.Stdout = conWpipe
	con.Stderr = conWpipe
	if _, err := con.StdinPipe(); err != nil { // SSH would close connection on stdin EOF
		conRpipe.Close()
		conWpipe.Close()
		return nil, nil, err
	}
	if err := con.Start(); err != nil {
		conRpipe.Close()
		conWpipe.Close()
		return nil, nil, fmt.Errorf("failed to connect to console server: %v", err)

	}
	conWpipe.Close()

	var tee io.Writer
	if inst.debug {
		tee = os.Stdout
	}
	merger := vmimpl.NewOutputMerger(tee)
	merger.Add("console", conRpipe)

	// We've started the console reading ssh command, but it has not necessary connected yet.
	// If we proceed to running the target command right away, we can miss part
	// of console output. During repro we can crash machines very quickly and
	// would miss beginning of a crash. Before ssh starts piping console output,
	// it usually prints:
	// "serialport: Connected to ... port 1 (session ID: ..., active connections: 1)"
	// So we wait for this line, or at least a minute and at least some output.
	{
		var output []byte
		timeout := time.NewTimer(time.Minute)
	loop:
		for {
			select {
			case out := <-merger.Output:
				output = append(output, out...)
				if bytes.Contains(output, []byte("serialport")) {
					break loop
				}
			case <-timeout.C:
				break loop
			}
		}
		timeout.Stop()
		if len(output) == 0 {
			con.Process.Kill()
			merger.Wait()
			return nil, nil, fmt.Errorf("no output from console")
		}
	}

	sshRpipe, sshWpipe, err := osutil.LongPipe()
	if err != nil {
		con.Process.Kill()
		merger.Wait()
		sshRpipe.Close()
		return nil, nil, err
	}
	if inst.sshUser != "root" {
		command = fmt.Sprintf("sudo bash -c '%v'", command)
	}
	args := append(sshArgs(inst.debug, inst.sshKey, "-p", 22), inst.sshUser+"@"+inst.name, command)<commit_msg>vm/gce: another attempt to fix trimmed console output<commit_after>// Copyright 2016 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 gce allows to use Google Compute Engine (GCE) virtual machines as VMs.
// It is assumed that syz-manager also runs on GCE as VMs are created in the current project/zone.
//
// See https://cloud.google.com/compute/docs for details.
// In particular, how to build GCE-compatible images:
// https://cloud.google.com/compute/docs/tutorials/building-images
// Working with serial console:
// https://cloud.google.com/compute/docs/instances/interacting-with-serial-console
package gce

import (
	"archive/tar"
	"bytes"
	"compress/gzip"
	"fmt"
	"io"
	"io/ioutil"
	"os"
	"os/exec"
	"path/filepath"
	"time"

	"github.com/google/syzkaller/pkg/config"
	"github.com/google/syzkaller/pkg/gce"
	"github.com/google/syzkaller/pkg/gcs"
	. "github.com/google/syzkaller/pkg/log"
	"github.com/google/syzkaller/pkg/osutil"
	"github.com/google/syzkaller/vm/vmimpl"
)

func init() {
	vmimpl.Register("gce", ctor)
}

type Config struct {
	Count        int    // number of VMs to use
	Machine_Type string // GCE machine type (e.g. "n1-highcpu-2")
	GCS_Path     string // GCS path to upload image
}

type Pool struct {
	env      *vmimpl.Env
	cfg      *Config
	GCE      *gce.Context
	gceImage string
}

type instance struct {
	cfg     *Config
	GCE     *gce.Context
	debug   bool
	name    string
	ip      string
	offset  int64
	gceKey  string // per-instance private ssh key associated with the instance
	sshKey  string // ssh key
	sshUser string
	workdir string
	closed  chan bool
}

func ctor(env *vmimpl.Env) (vmimpl.Pool, error) {
	if env.Name == "" {
		return nil, fmt.Errorf("config param name is empty (required for GCE)")
	}
	if env.Image == "" {
		return nil, fmt.Errorf("config param image is empty (required for GCE)")
	}
	cfg := &Config{
		Count: 1,
	}
	if err := config.LoadData(env.Config, cfg); err != nil {
		return nil, fmt.Errorf("failed to parse gce vm config: %v", err)
	}
	if cfg.Count < 1 || cfg.Count > 1000 {
		return nil, fmt.Errorf("invalid config param count: %v, want [1, 1000]", cfg.Count)
	}
	if env.Debug {
		cfg.Count = 1
	}
	if cfg.Machine_Type == "" {
		return nil, fmt.Errorf("machine_type parameter is empty")
	}
	if cfg.GCS_Path == "" {
		return nil, fmt.Errorf("gcs_path parameter is empty")
	}

	GCE, err := gce.NewContext()
	if err != nil {
		return nil, fmt.Errorf("failed to init gce: %v", err)
	}
	Logf(0, "GCE initialized: running on %v, internal IP %v, project %v, zone %v",
		GCE.Instance, GCE.InternalIP, GCE.ProjectID, GCE.ZoneID)

	gcsImage := filepath.Join(cfg.GCS_Path, env.Name+"-image.tar.gz")
	Logf(0, "uploading image to %v...", gcsImage)
	if err := uploadImageToGCS(env.Image, gcsImage); err != nil {
		return nil, err
	}
	gceImage := env.Name
	Logf(0, "creating GCE image %v...", gceImage)
	if err := GCE.DeleteImage(gceImage); err != nil {
		return nil, fmt.Errorf("failed to delete GCE image: %v", err)
	}
	if err := GCE.CreateImage(gceImage, gcsImage); err != nil {
		return nil, fmt.Errorf("failed to create GCE image: %v", err)
	}
	pool := &Pool{
		cfg:      cfg,
		env:      env,
		GCE:      GCE,
		gceImage: gceImage,
	}
	return pool, nil
}

func (pool *Pool) Count() int {
	return pool.cfg.Count
}

func (pool *Pool) Create(workdir string, index int) (vmimpl.Instance, error) {
	name := fmt.Sprintf("%v-%v", pool.env.Name, index)
	// Create SSH key for the instance.
	gceKey := filepath.Join(workdir, "key")
	keygen := exec.Command("ssh-keygen", "-t", "rsa", "-b", "2048", "-N", "", "-C", "syzkaller", "-f", gceKey)
	if out, err := keygen.CombinedOutput(); err != nil {
		return nil, fmt.Errorf("failed to execute ssh-keygen: %v\n%s", err, out)
	}
	gceKeyPub, err := ioutil.ReadFile(gceKey + ".pub")
	if err != nil {
		return nil, fmt.Errorf("failed to read file: %v", err)
	}

	Logf(0, "deleting instance: %v", name)
	if err := pool.GCE.DeleteInstance(name, true); err != nil {
		return nil, err
	}
	Logf(0, "creating instance: %v", name)
	ip, err := pool.GCE.CreateInstance(name, pool.cfg.Machine_Type, pool.gceImage, string(gceKeyPub))
	if err != nil {
		return nil, err
	}

	ok := false
	defer func() {
		if !ok {
			pool.GCE.DeleteInstance(name, true)
		}
	}()
	sshKey := pool.env.Sshkey
	sshUser := "root"
	if sshKey == "" {
		// Assuming image supports GCE ssh fanciness.
		sshKey = gceKey
		sshUser = "syzkaller"
	}
	Logf(0, "wait instance to boot: %v (%v)", name, ip)
	if err := waitInstanceBoot(pool.env.Debug, ip, sshKey, sshUser); err != nil {
		return nil, err
	}
	ok = true
	inst := &instance{
		cfg:     pool.cfg,
		debug:   pool.env.Debug,
		GCE:     pool.GCE,
		name:    name,
		ip:      ip,
		gceKey:  gceKey,
		sshKey:  sshKey,
		sshUser: sshUser,
		closed:  make(chan bool),
	}
	return inst, nil
}

func (inst *instance) Close() {
	close(inst.closed)
	inst.GCE.DeleteInstance(inst.name, false)
}

func (inst *instance) Forward(port int) (string, error) {
	return fmt.Sprintf("%v:%v", inst.GCE.InternalIP, port), nil
}

func (inst *instance) Copy(hostSrc string) (string, error) {
	vmDst := "./" + filepath.Base(hostSrc)
	args := append(sshArgs(inst.debug, inst.sshKey, "-P", 22), hostSrc, inst.sshUser+"@"+inst.name+":"+vmDst)
	if _, err := runCmd(inst.debug, "scp", args...); err != nil {
		return "", err
	}
	return vmDst, nil
}

func (inst *instance) Run(timeout time.Duration, stop <-chan bool, command string) (<-chan []byte, <-chan error, error) {
	conRpipe, conWpipe, err := osutil.LongPipe()
	if err != nil {
		return nil, nil, err
	}

	conAddr := fmt.Sprintf("%v.%v.%v.syzkaller.port=1@ssh-serialport.googleapis.com",
		inst.GCE.ProjectID, inst.GCE.ZoneID, inst.name)
	conArgs := append(sshArgs(inst.debug, inst.gceKey, "-p", 9600), conAddr)
	con := exec.Command("ssh", conArgs...)
	con.Env = []string{}
	con.Stdout = conWpipe
	con.Stderr = conWpipe
	if _, err := con.StdinPipe(); err != nil { // SSH would close connection on stdin EOF
		conRpipe.Close()
		conWpipe.Close()
		return nil, nil, err
	}
	if err := con.Start(); err != nil {
		conRpipe.Close()
		conWpipe.Close()
		return nil, nil, fmt.Errorf("failed to connect to console server: %v", err)

	}
	conWpipe.Close()

	var tee io.Writer
	if inst.debug {
		tee = os.Stdout
	}
	merger := vmimpl.NewOutputMerger(tee)
	merger.Add("console", conRpipe)

	// We've started the console reading ssh command, but it has not necessary connected yet.
	// If we proceed to running the target command right away, we can miss part
	// of console output. During repro we can crash machines very quickly and
	// would miss beginning of a crash. Before ssh starts piping console output,
	// it usually prints:
	// "serialport: Connected to ... port 1 (session ID: ..., active connections: 1)"
	// So we wait for this line, or at least a minute and at least some output.
	{
		var output []byte
		timeout := time.NewTimer(time.Minute)
	loop:
		for {
			select {
			case out := <-merger.Output:
				output = append(output, out...)
				if bytes.Contains(output, []byte("serialport")) {
					// Just to make sure (otherwise we still see trimmed reports).
					time.Sleep(5 * time.Second)
					break loop
				}
			case <-timeout.C:
				break loop
			}
		}
		timeout.Stop()
		if len(output) == 0 {
			con.Process.Kill()
			merger.Wait()
			return nil, nil, fmt.Errorf("no output from console")
		}
	}

	sshRpipe, sshWpipe, err := osutil.LongPipe()
	if err != nil {
		con.Process.Kill()
		merger.Wait()
		sshRpipe.Close()
		return nil, nil, err
	}
	if inst.sshUser != "root" {
		command = fmt.Sprintf("sudo bash -c '%v'", command)
	}
	args := append(sshArgs(inst.debug, inst.sshKey, "-p", 22), inst.sshUser+"@"+inst.name, command)
{'fraction_non_alphanumeric': 0.15165876777251186, 'fraction_numerical': 0.005648250340842693, 'mean_word_length': 3.396118721461187, 'pattern_counts': {'":': 2, '<': 15, '<?xml version=': 0, '>': 5, 'https://': 6, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 2, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1492138', 'n_tokens_mistral': 5997, 'n_tokens_neox': 5629, 'n_words': 1642}