Beats enrollment subcommand (#7182)

This PR implements intial enrollment to Central Management in Kibana. After running the enrollment command, beats will have a valid access token to use when retrieving configurations.

To test this:

- Use the following branches:
  - Elasticsearch: https://github.com/ycombinator/elasticsearch/tree/x-pack/management/beats
  - Kibana: https://github.com/elastic/kibana/tree/feature/x-pack/management/beats
- Retrieve a valid enrollment token:
```
curl  \                             
  -u elastic \           
  -H 'kbn-xsrf: foobar'  \
  -H 'Content-Type: application/json' \
  -X POST \
  http://localhost:5601/api/beats/enrollment_tokens
```
- Use it:
```
<beat> enroll http://localhost:5601 <enrollment_token>
```
- Check agent is enrolled:
```
curl http://localhost:5601/api/beats/agents | jq
```

This is part of #7028, closes #7032

libbeat/kibana/client.go

@@ -140,8 +140,10 @@ func NewClientWithConfig(config *ClientConfig) (*Client, error) {
 		},
 	}
 
-	if err = client.SetVersion(); err != nil {
-		return nil, fmt.Errorf("fail to get the Kibana version: %v", err)
+	if !config.IgnoreVersion {
+		if err = client.SetVersion(); err != nil {
+			return nil, fmt.Errorf("fail to get the Kibana version: %v", err)
+		}
 	}
 
 	return client, nil

libbeat/kibana/client_config.go

@@ -25,13 +25,14 @@ import (
 
 // ClientConfig to connect to Kibana
 type ClientConfig struct {
-	Protocol string            `config:"protocol"`
-	Host     string            `config:"host"`
-	Path     string            `config:"path"`
-	Username string            `config:"username"`
-	Password string            `config:"password"`
-	TLS      *tlscommon.Config `config:"ssl"`
-	Timeout  time.Duration     `config:"timeout"`
+	Protocol      string            `config:"protocol"`
+	Host          string            `config:"host"`
+	Path          string            `config:"path"`
+	Username      string            `config:"username"`
+	Password      string            `config:"password"`
+	TLS           *tlscommon.Config `config:"ssl"`
+	Timeout       time.Duration     `config:"timeout"`
+	IgnoreVersion bool
 }
 
 var (

x-pack/auditbeat/cmd/root.go

@@ -4,11 +4,14 @@
 
 package cmd
 
-import "github.com/elastic/beats/auditbeat/cmd"
+import (
+	"github.com/elastic/beats/auditbeat/cmd"
+	xpackcmd "github.com/elastic/beats/x-pack/libbeat/cmd"
+)
 
 // RootCmd to handle beats cli
 var RootCmd = cmd.RootCmd
 
 func init() {
-	// TODO inject x-pack features
+	xpackcmd.AddXPack(RootCmd, cmd.Name)
 }

x-pack/filebeat/cmd/root.go

@@ -4,11 +4,14 @@
 
 package cmd
 
-import "github.com/elastic/beats/filebeat/cmd"
+import (
+	"github.com/elastic/beats/filebeat/cmd"
+	xpackcmd "github.com/elastic/beats/x-pack/libbeat/cmd"
+)
 
 // RootCmd to handle beats cli
 var RootCmd = cmd.RootCmd
 
 func init() {
-	// TODO inject x-pack features
+	xpackcmd.AddXPack(RootCmd, cmd.Name)
 }

x-pack/heartbeat/cmd/root.go

@@ -4,11 +4,14 @@
 
 package cmd
 
-import "github.com/elastic/beats/heartbeat/cmd"
+import (
+	"github.com/elastic/beats/heartbeat/cmd"
+	xpackcmd "github.com/elastic/beats/x-pack/libbeat/cmd"
+)
 
 // RootCmd to handle beats cli
 var RootCmd = cmd.RootCmd
 
 func init() {
-	// TODO inject x-pack features
+	xpackcmd.AddXPack(RootCmd, cmd.Name)
 }

x-pack/libbeat/cmd/enroll.go

@@ -0,0 +1,55 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+package cmd
+
+import (
+	"fmt"
+
+	"github.com/pkg/errors"
+	"github.com/spf13/cobra"
+
+	"github.com/elastic/beats/libbeat/cmd/instance"
+	"github.com/elastic/beats/libbeat/common/cli"
+	"github.com/elastic/beats/x-pack/libbeat/management"
+)
+
+func getBeat(name, version string) (*instance.Beat, error) {
+	b, err := instance.NewBeat(name, "", version)
+
+	if err != nil {
+		return nil, fmt.Errorf("error creating beat: %s", err)
+	}
+
+	if err = b.Init(); err != nil {
+		return nil, fmt.Errorf("error initializing beat: %s", err)
+	}
+
+	return b, nil
+}
+
+func genEnrollCmd(name, version string) *cobra.Command {
+	enrollCmd := cobra.Command{
+		Use:   "enroll <kibana_url> <enrollment_token>",
+		Short: "Enroll in Kibana for Central Management",
+		Args:  cobra.ExactArgs(2),
+		Run: cli.RunWith(func(cmd *cobra.Command, args []string) error {
+			beat, err := getBeat(name, version)
+			kibanaURL := args[0]
+			enrollmentToken := args[1]
+			if err != nil {
+				return err
+			}
+
+			if err = management.Enroll(beat, kibanaURL, enrollmentToken); err != nil {
+				return errors.Wrap(err, "Error while enrolling")
+			}
+
+			fmt.Println("Enrolled and ready to retrieve settings from Kibana")
+			return nil
+		}),
+	}
+
+	return &enrollCmd
+}

x-pack/libbeat/cmd/inject.go

@@ -0,0 +1,12 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+package cmd
+
+import "github.com/elastic/beats/libbeat/cmd"
+
+// AddXPack extends the given root folder with XPack features
+func AddXPack(root *cmd.BeatsRootCmd, name string) {
+	root.AddCommand(genEnrollCmd(name, ""))
+}

x-pack/libbeat/management/api/client.go

@@ -0,0 +1,95 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+package api
+
+import (
+	"bytes"
+	"encoding/json"
+	"net/http"
+	"net/url"
+	"time"
+
+	"github.com/pkg/errors"
+
+	"github.com/elastic/beats/libbeat/common"
+	"github.com/elastic/beats/libbeat/kibana"
+)
+
+const defaultTimeout = 10 * time.Second
+
+// Client to Central Management API
+type Client struct {
+	client *kibana.Client
+}
+
+// ConfigFromURL generates a full kibana client config from an URL
+func ConfigFromURL(kibanaURL string) (*kibana.ClientConfig, error) {
+	data, err := url.Parse(kibanaURL)
+	if err != nil {
+		return nil, err
+	}
+
+	var username, password string
+	if data.User != nil {
+		username = data.User.Username()
+		password, _ = data.User.Password()
+	}
+
+	return &kibana.ClientConfig{
+		Protocol:      data.Scheme,
+		Host:          data.Host,
+		Path:          data.Path,
+		Username:      username,
+		Password:      password,
+		Timeout:       defaultTimeout,
+		IgnoreVersion: true,
+	}, nil
+}
+
+// NewClient creates and returns a kibana client
+func NewClient(cfg *kibana.ClientConfig) (*Client, error) {
+	client, err := kibana.NewClientWithConfig(cfg)
+	if err != nil {
+		return nil, err
+	}
+	return &Client{
+		client: client,
+	}, nil
+}
+
+// do a request to the API and unmarshall the message, error if anything fails
+func (c *Client) request(method, extraPath string,
+	params common.MapStr, headers http.Header, message interface{}) (int, error) {
+
+	paramsJSON, err := json.Marshal(params)
+	if err != nil {
+		return 400, err
+	}
+
+	statusCode, result, err := c.client.Request(method, extraPath, nil, headers, bytes.NewBuffer(paramsJSON))
+	if err != nil {
+		return statusCode, err
+	}
+
+	if statusCode >= 300 {
+		err = extractError(result)
+	} else {
+		if err = json.Unmarshal(result, message); err != nil {
+			return statusCode, errors.Wrap(err, "error unmarshaling Kibana response")
+		}
+	}
+
+	return statusCode, err
+}
+
+func extractError(result []byte) error {
+	var kibanaResult struct {
+		Message string
+	}
+	if err := json.Unmarshal(result, &kibanaResult); err != nil {
+		return errors.Wrap(err, "parsing Kibana response")
+	}
+	return errors.New(kibanaResult.Message)
+}

x-pack/libbeat/management/api/client_test.go

@@ -0,0 +1,33 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+package api
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"testing"
+)
+
+func newServerClientPair(t *testing.T, handler http.HandlerFunc) (*httptest.Server, *Client) {
+	mux := http.NewServeMux()
+	mux.Handle("/api/status", http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		http.Error(w, "Unauthorized", 401)
+	}))
+	mux.Handle("/", handler)
+
+	server := httptest.NewServer(mux)
+
+	config, err := ConfigFromURL(server.URL)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	client, err := NewClient(config)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	return server, client
+}

x-pack/libbeat/management/api/enroll.go

@@ -0,0 +1,36 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+package api
+
+import (
+	"net/http"
+
+	uuid "github.com/satori/go.uuid"
+
+	"github.com/elastic/beats/libbeat/common"
+)
+
+// Enroll a beat in central management, this call returns a valid access token to retrieve configurations
+func (c *Client) Enroll(beatType, beatVersion, hostname string, beatUUID uuid.UUID, enrollmentToken string) (string, error) {
+	params := common.MapStr{
+		"type":      beatType,
+		"host_name": hostname,
+		"version":   beatVersion,
+	}
+
+	resp := struct {
+		AccessToken string `json:"access_token"`
+	}{}
+
+	headers := http.Header{}
+	headers.Set("kbn-beats-enrollment-token", enrollmentToken)
+
+	_, err := c.request("POST", "/api/beats/agent/"+beatUUID.String(), params, headers, &resp)
+	if err != nil {
+		return "", err
+	}
+
+	return resp.AccessToken, err
+}

x-pack/libbeat/management/api/enroll_test.go

@@ -0,0 +1,71 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+package api
+
+import (
+	"encoding/json"
+	"fmt"
+	"io/ioutil"
+	"net/http"
+	"testing"
+
+	uuid "github.com/satori/go.uuid"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestEnrollValid(t *testing.T) {
+	beatUUID := uuid.NewV4()
+
+	server, client := newServerClientPair(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		defer r.Body.Close()
+		body, err := ioutil.ReadAll(r.Body)
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		// Check correct path is used
+		assert.Equal(t, "/api/beats/agent/"+beatUUID.String(), r.URL.Path)
+
+		// Check enrollment token is correct
+		assert.Equal(t, "thisismyenrollmenttoken", r.Header.Get("kbn-beats-enrollment-token"))
+
+		request := struct {
+			Hostname string `json:"host_name"`
+			Type     string `json:"type"`
+			Version  string `json:"version"`
+		}{}
+		if err := json.Unmarshal(body, &request); err != nil {
+			t.Fatal(err)
+		}
+
+		assert.Equal(t, "myhostname.lan", request.Hostname)
+		assert.Equal(t, "metricbeat", request.Type)
+		assert.Equal(t, "6.3.0", request.Version)
+
+		fmt.Fprintf(w, `{"access_token": "fooo"}`)
+	}))
+	defer server.Close()
+
+	accessToken, err := client.Enroll("metricbeat", "6.3.0", "myhostname.lan", beatUUID, "thisismyenrollmenttoken")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	assert.Equal(t, "fooo", accessToken)
+}
+
+func TestEnrollError(t *testing.T) {
+	beatUUID := uuid.NewV4()
+
+	server, client := newServerClientPair(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		http.Error(w, `{"message": "Invalid enrollment token"}`, 400)
+	}))
+	defer server.Close()
+
+	accessToken, err := client.Enroll("metricbeat", "6.3.0", "myhostname.lan", beatUUID, "thisismyenrollmenttoken")
+
+	assert.NotNil(t, err)
+	assert.Equal(t, "", accessToken)
+}