Merge remote-tracking branch 'origin/main' into api-key-support

internal/fleetserver/client.go

@@ -6,64 +6,123 @@ package fleetserver
 
 import (
 	"context"
-	"encoding/json"
+	"crypto/tls"
 	"fmt"
 	"io"
 	"net/http"
 	"net/url"
 
+	"github.com/elastic/elastic-package/internal/certs"
 	"github.com/elastic/elastic-package/internal/logger"
 )
 
+// Client is a client for Fleet Server API. This API only supports authentication with API
+// keys, though some endpoints are also available without any authentication.
 type Client struct {
 	address string
-	http    *http.Client
+	apiKey  string
+
+	certificateAuthority string
+	tlSkipVerify         bool
+
+	http            *http.Client
+	httpClientSetup func(*http.Client) *http.Client
 }
 
-func NewClient(address string) *Client {
-	return &Client{
+type ClientOption func(*Client)
+
+func NewClient(address string, opts ...ClientOption) (*Client, error) {
+	client := Client{
 		address: address,
-		http:    http.DefaultClient,
 	}
+
+	for _, opt := range opts {
+		opt(&client)
+	}
+
+	httpClient, err := client.httpClient()
+	if err != nil {
+		return nil, fmt.Errorf("cannot create HTTP client: %w", err)
+	}
+	client.http = httpClient
+	return &client, nil
 }
 
-type Status struct {
-	Name   string `json:"name"`
-	Status string `json:"status"`
+// APIKey option sets the API key to be used by the client for authentication.
+func APIKey(apiKey string) ClientOption {
+	return func(c *Client) {
+		c.apiKey = apiKey
+	}
+}
 
-	// Version is only present if client is authenticated.
-	Version struct {
-		Number string `json:"number"`
-	} `json:"version"`
+// TLSSkipVerify option disables TLS verification.
+func TLSSkipVerify() ClientOption {
+	return func(c *Client) {
+		c.tlSkipVerify = true
+	}
 }
 
-func (c *Client) Status(ctx context.Context) (*Status, error) {
-	statusURL, err := url.JoinPath(c.address, "/api/status")
-	if err != nil {
-		return nil, fmt.Errorf("could not build URL: %w", err)
+// CertificateAuthority sets the certificate authority to be used by the client.
+func CertificateAuthority(certificateAuthority string) ClientOption {
+	return func(c *Client) {
+		c.certificateAuthority = certificateAuthority
 	}
-	logger.Debugf("GET %s", statusURL)
-	req, err := http.NewRequestWithContext(ctx, "GET", statusURL, nil)
-	if err != nil {
-		return nil, err
+}
+
+// HTTPClientSetup adds an initializing function for the http client.
+func HTTPClientSetup(setup func(*http.Client) *http.Client) ClientOption {
+	return func(c *Client) {
+		c.httpClientSetup = setup
 	}
-	resp, err := c.http.Do(req)
-	if err != nil {
-		return nil, fmt.Errorf("request failed (url: %s): %w", statusURL, err)
+}
+
+func (c *Client) httpClient() (*http.Client, error) {
+	client := &http.Client{}
+	if c.tlSkipVerify {
+		client.Transport = &http.Transport{
+			TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
+		}
+	} else if c.certificateAuthority != "" {
+		rootCAs, err := certs.SystemPoolWithCACertificate(c.certificateAuthority)
+		if err != nil {
+			return nil, fmt.Errorf("reading CA certificate: %w", err)
+		}
+		client.Transport = &http.Transport{
+			TLSClientConfig: &tls.Config{RootCAs: rootCAs},
+		}
 	}
-	defer resp.Body.Close()
-	if resp.StatusCode >= 300 {
-		return nil, fmt.Errorf("unexpected status code %v", resp.StatusCode)
+
+	if c.httpClientSetup != nil {
+		client = c.httpClientSetup(client)
 	}
-	body, err := io.ReadAll(resp.Body)
+
+	return client, nil
+}
+
+func (c *Client) httpRequest(ctx context.Context, method, resourcePath string, reqBody io.Reader) (*http.Request, error) {
+	base, err := url.Parse(c.address)
 	if err != nil {
-		return nil, fmt.Errorf("failed to read response body: %w", err)
+		return nil, fmt.Errorf("could not create base URL from host: %v: %w", c.address, err)
 	}
-	var status Status
-	err = json.Unmarshal(body, &status)
+
+	rel, err := url.Parse(resourcePath)
 	if err != nil {
-		return nil, fmt.Errorf("failed to parse response body: %w", err)
+		return nil, fmt.Errorf("could not create relative URL from resource path: %v: %w", resourcePath, err)
+	}
+
+	u := base.JoinPath(rel.EscapedPath())
+	u.RawQuery = rel.RawQuery
+
+	logger.Debugf("%s %s", method, u)
+
+	req, err := http.NewRequestWithContext(ctx, method, u.String(), reqBody)
+	if err != nil {
+		return nil, fmt.Errorf("could not create %v request to Fleet Server API resource: %s: %w", method, resourcePath, err)
+	}
+
+	if c.apiKey != "" {
+		req.Header.Set("Authorization", "ApiKey "+c.apiKey)
 	}
 
-	return &status, nil
+	return req, nil
 }

internal/kibana/client.go

@@ -77,10 +77,8 @@ func NewClient(opts ...ClientOption) (*Client, error) {
 		}
 		c.versionInfo = v.Version
 
-		if c.versionInfo.Number == "" {
-			// Version info may not contain any version if this is a managed Kibana.
-			c.semver = semver.MustParse("9.0.0")
-		} else {
+		// Version info may not contain any version if this is a managed Kibana.
+		if c.versionInfo.Number != "" {
 			c.semver, err = semver.NewVersion(c.versionInfo.Number)
 			if err != nil {
 				return nil, fmt.Errorf("failed to parse Kibana version (%s): %w", c.versionInfo.Number, err)

internal/kibana/enrollmenttokens.go

@@ -69,7 +69,7 @@ func (c *Client) getEnrollmentTokens(ctx context.Context, kuery string) ([]Enrol
 		}
 
 		if err := json.Unmarshal(respBody, &resp); err != nil {
-			return nil, fmt.Errorf("could not decode policies response: %w", err)
+			return nil, fmt.Errorf("could not decode response to get enrollment tokens: %w", err)
 		}
 
 		// Tokens are listed twice, at least on some versions, get only one copy of them.

internal/kibana/fleet.go

@@ -113,24 +113,6 @@ func (c *Client) AddFleetOutput(ctx context.Context, fo FleetOutput) error {
 	return nil
 }
 
-// RemoveFleetOutput removes an output from Fleet
-func (c *Client) RemoveFleetOutput(ctx context.Context, outputID string) error {
-	statusCode, respBody, err := c.delete(ctx, fmt.Sprintf("%s/outputs/%s", FleetAPI, outputID))
-	if err != nil {
-		return fmt.Errorf("could not delete fleet output: %w", err)
-	}
-
-	if statusCode == http.StatusNotFound {
-		// Already removed, ignore error.
-		return nil
-	}
-	if statusCode != http.StatusOK {
-		return fmt.Errorf("could not add fleet output; API status code = %d; response body = %s", statusCode, respBody)
-	}
-
-	return nil
-}
-
 // RemoveFleetOutput removes an output from Fleet
 func (c *Client) RemoveFleetOutput(ctx context.Context, outputID string) error {
 	statusCode, respBody, err := c.delete(ctx, fmt.Sprintf("%s/outputs/%s", FleetAPI, outputID))

internal/serverless/project.go

@@ -140,7 +140,10 @@ func (p *Project) getKibanaHealth(ctx context.Context, kibanaClient *kibana.Clie
 }
 
 func (p *Project) getFleetHealth(ctx context.Context) error {
-	client := fleetserver.NewClient(p.Endpoints.Fleet)
+	client, err := fleetserver.NewClient(p.Endpoints.Fleet)
+	if err != nil {
+		return fmt.Errorf("could not create Fleet Server client: %w", err)
+	}
 	status, err := client.Status(ctx)
 	if err != nil {
 		return err

internal/stack/environment.go

@@ -148,11 +148,11 @@ func (p *environmentProvider) initClients() error {
 	return nil
 }
 
-func (p environmentProvider) setupFleet(ctx context.Context, config Config, options Options) (Config, error) {
+func (p *environmentProvider) setupFleet(ctx context.Context, config Config, options Options) (Config, error) {
 	const localFleetServerURL = "https://fleet-server:8220"
 
 	fleetServerURL, err := p.kibana.DefaultFleetServerURL(ctx)
-	if errors.Is(err, kibana.ErrFleetServerNotFound) || !isFleetServerReachable(ctx, fleetServerURL) {
+	if errors.Is(err, kibana.ErrFleetServerNotFound) || !isFleetServerReachable(ctx, fleetServerURL, config) {
 		// We need to setup a local Fleet Server
 		fleetServerURL = localFleetServerURL
 		config.Parameters[paramFleetServerManaged] = "true"
@@ -195,8 +195,12 @@ func fleetServerHostID(namespace string) string {
 	return "elastic-package-" + namespace
 }
 
-func isFleetServerReachable(ctx context.Context, address string) bool {
-	status, err := fleetserver.NewClient(address).Status(ctx)
+func isFleetServerReachable(ctx context.Context, address string, config Config) bool {
+	client, err := fleetserver.NewClient(address, fleetserver.APIKey(config.ElasticsearchAPIKey))
+	if err != nil {
+		return false
+	}
+	status, err := client.Status(ctx)
 	return err == nil && strings.ToLower(status.Status) == "healthy"
 }
 
@@ -374,8 +378,14 @@ func (p *environmentProvider) fleetStatus(ctx context.Context, options Options,
 		return status
 	}
 
-	// TODO: Add authentication for fleet server client.
-	client := fleetserver.NewClient(address)
+	client, err := fleetserver.NewClient(address,
+		fleetserver.APIKey(config.ElasticsearchAPIKey),
+		fleetserver.CertificateAuthority(config.CACertFile),
+	)
+	if err != nil {
+		status.Status = "unknown: " + err.Error()
+	}
+
 	fleetServerStatus, err := client.Status(ctx)
 	if err != nil {
 		status.Status = "unknown: " + err.Error()