feat(agent): add MachineHub output mode with CyberArk client integration

- Introduced a new `MachineHub` output mode in the agent configuration.
- Implemented `CyberArkClient` to handle data uploads using CyberArk APIs.
- Added environment-based configuration loading for CyberArk integration.
- Updated `dataupload` package to support `ClusterID` instead of `ClusterName`.
- Enhanced test coverage for `MachineHub` mode and CyberArk client functionality.
- Refactored mock servers to support CyberArk service discovery and identity APIs.

Author: wallrj-cyberark

pkg/agent/config.go

@@ -1,6 +1,7 @@
 package agent
 
 import (
+	"crypto/x509"
 	"fmt"
 	"io"
 	"net/url"
@@ -10,9 +11,11 @@ import (
 
 	"github.com/go-logr/logr"
 	"github.com/hashicorp/go-multierror"
+	"github.com/jetstack/venafi-connection-lib/http_client"
 	"github.com/spf13/cobra"
 	"gopkg.in/yaml.v3"
 	"k8s.io/client-go/rest"
+	"k8s.io/client-go/transport"
 
 	"github.com/jetstack/preflight/api"
 	"github.com/jetstack/preflight/pkg/client"
@@ -334,6 +337,7 @@ const (
 	VenafiCloudKeypair          OutputMode = "Venafi Cloud Key Pair Service Account"
 	VenafiCloudVenafiConnection OutputMode = "Venafi Cloud VenafiConnection"
 	LocalFile                   OutputMode = "Local File"
+	MachineHub                  OutputMode = "MachineHub"
 )
 
 // The command-line flags and the config file are combined into this struct by
@@ -420,6 +424,9 @@ func ValidateAndCombineConfig(log logr.Logger, cfg Config, flags AgentCmdFlags)
 		case !flags.VenafiCloudMode && flags.CredentialsPath != "":
 			mode = JetstackSecureOAuth
 			reason = "--credentials-file was specified without --venafi-cloud"
+		case flags.MachineHubMode:
+			mode = MachineHub
+			reason = "--machine-hub was specified"
 		case flags.OutputPath != "":
 			mode = LocalFile
 			reason = "--output-path was specified"
@@ -433,6 +440,7 @@ func ValidateAndCombineConfig(log logr.Logger, cfg Config, flags AgentCmdFlags)
 				" - Use --venafi-connection for the " + string(VenafiCloudVenafiConnection) + " mode.\n" +
 				" - Use --credentials-file alone if you want to use the " + string(JetstackSecureOAuth) + " mode.\n" +
 				" - Use --api-token if you want to use the " + string(JetstackSecureAPIToken) + " mode.\n" +
+				" - Use --machine-hub if you want to use the " + string(MachineHub) + " mode.\n" +
 				" - Use --output-path or output-path in the config file for " + string(LocalFile) + " mode.")
 		}
 
@@ -762,6 +770,17 @@ func validateCredsAndCreateClient(log logr.Logger, flagCredentialsPath, flagClie
 		}
 	case LocalFile:
 		outputClient = client.NewFileClient(cfg.OutputPath)
+	case MachineHub:
+		var (
+			err     error
+			rootCAs *x509.CertPool
+		)
+		httpClient := http_client.NewDefaultClient(version.UserAgent(), rootCAs)
+		httpClient.Transport = transport.NewDebuggingRoundTripper(httpClient.Transport, transport.DebugByContext)
+		outputClient, err = client.NewCyberArk(httpClient)
+		if err != nil {
+			errs = multierror.Append(errs, err)
+		}
 	default:
 		panic(fmt.Errorf("programmer mistake: output mode not implemented: %s", cfg.OutputMode))
 	}

pkg/agent/config_test.go

@@ -199,6 +199,7 @@ func Test_ValidateAndCombineConfig(t *testing.T) {
 			 - Use --venafi-connection for the Venafi Cloud VenafiConnection mode.
 			 - Use --credentials-file alone if you want to use the Jetstack Secure OAuth mode.
 			 - Use --api-token if you want to use the Jetstack Secure API Token mode.
+			 - Use --machine-hub if you want to use the MachineHub mode.
 			 - Use --output-path or output-path in the config file for Local File mode.`))
 		assert.Nil(t, cl)
 	})
@@ -617,6 +618,38 @@ func Test_ValidateAndCombineConfig(t *testing.T) {
 		assert.Equal(t, VenafiCloudVenafiConnection, got.OutputMode)
 	})
 
+	t.Run("--machine-hub selects MachineHub mode", func(t *testing.T) {
+		t.Setenv("POD_NAMESPACE", "venafi")
+		t.Setenv("KUBECONFIG", withFile(t, fakeKubeconfig))
+		t.Setenv("ARK_SUBDOMAIN", "tlspk")
+		t.Setenv("ARK_USERNAME", "[email protected]")
+		t.Setenv("ARK_SECRET", "test-secret")
+		got, cl, err := ValidateAndCombineConfig(discardLogs(),
+			withConfig(""),
+			withCmdLineFlags("--period", "1m", "--machine-hub"))
+		require.NoError(t, err)
+		assert.Equal(t, MachineHub, got.OutputMode)
+		assert.IsType(t, &client.CyberArkClient{}, cl)
+	})
+
+	t.Run("--machine-hub without required environment variables", func(t *testing.T) {
+		t.Setenv("POD_NAMESPACE", "venafi")
+		t.Setenv("KUBECONFIG", withFile(t, fakeKubeconfig))
+		t.Setenv("ARK_SUBDOMAIN", "")
+		t.Setenv("ARK_USERNAME", "")
+		t.Setenv("ARK_SECRET", "")
+		got, cl, err := ValidateAndCombineConfig(discardLogs(),
+			withConfig(""),
+			withCmdLineFlags("--period", "1m", "--machine-hub"))
+		assert.Equal(t, CombinedConfig{}, got)
+		assert.Nil(t, cl)
+		assert.EqualError(t, err, testutil.Undent(`
+			validating creds: failed loading config using the MachineHub mode: 1 error occurred:
+				* missing environment variables: ARK_SUBDOMAIN, ARK_USERNAME, ARK_SECRET
+
+	   `))
+	})
+
 	t.Run("argument: --output-file selects local file mode", func(t *testing.T) {
 		log, gotLog := recordLogs(t)
 		got, outputClient, err := ValidateAndCombineConfig(log,

pkg/client/client_cyberark.go

@@ -1,9 +1,51 @@
 package client
 
 import (
+	"context"
+	"fmt"
+	"net/http"
+
+	"github.com/jetstack/preflight/api"
+	"github.com/jetstack/preflight/pkg/internal/cyberark"
 	"github.com/jetstack/preflight/pkg/internal/cyberark/dataupload"
 )
 
-type CyberArkClient = dataupload.CyberArkClient
+type CyberArkClient struct {
+	configLoader cyberark.ClientConfigLoader
+	httpClient   *http.Client
+}
+
+var _ Client = &CyberArkClient{}
+
+func NewCyberArk(httpClient *http.Client) (*CyberArkClient, error) {
+	configLoader := cyberark.LoadClientConfigFromEnvironment
+	_, err := configLoader()
+	if err != nil {
+		return nil, err
+	}
+	return &CyberArkClient{
+		configLoader: configLoader,
+		httpClient:   httpClient,
+	}, nil
+}
+
+// An API token is obtained by authenticating with the ARK_USERNAME and ARK_SECRET from the environment.
+// ARK_SUBDOMAIN should be your tenant subdomain.
+func (o *CyberArkClient) PostDataReadingsWithOptions(ctx context.Context, readings []*api.DataReading, options Options) error {
+	cfg, err := o.configLoader()
+	if err != nil {
+		return err
+	}
+	datauploadClient, err := cyberark.NewDatauploadClient(ctx, o.httpClient, cfg)
+	if err != nil {
+		return fmt.Errorf("while initializing data upload client: %s", err)
+	}
 
-var NewCyberArkClient = dataupload.New
+	err = datauploadClient.PutSnapshot(ctx, api.DataReadingsPost{}, dataupload.Options{
+		ClusterID: options.ClusterID,
+	})
+	if err != nil {
+		return fmt.Errorf("while posting snapshot: %s", err)
+	}
+	return nil
+}

pkg/client/client_cyberark_test.go

@@ -0,0 +1,73 @@
+package client_test
+
+import (
+	"net/http"
+	"os"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"k8s.io/client-go/transport"
+	"k8s.io/klog/v2"
+	"k8s.io/klog/v2/ktesting"
+
+	"github.com/jetstack/preflight/api"
+	"github.com/jetstack/preflight/pkg/client"
+	"github.com/jetstack/preflight/pkg/internal/cyberark/servicediscovery"
+	"github.com/jetstack/preflight/pkg/testutil"
+
+	_ "k8s.io/klog/v2/ktesting/init"
+)
+
+func TestCyberArkClient_PostDataReadingsWithOptions_MockAPI(t *testing.T) {
+	t.Setenv("ARK_SUBDOMAIN", servicediscovery.MockDiscoverySubdomain)
+	t.Setenv("ARK_USERNAME", "[email protected]")
+	t.Setenv("ARK_SECRET", "somepassword")
+	t.Run("success", func(t *testing.T) {
+		logger := ktesting.NewLogger(t, ktesting.DefaultConfig)
+		ctx := klog.NewContext(t.Context(), logger)
+
+		httpClient := testutil.FakeCyberArk(t)
+
+		c, err := client.NewCyberArk(httpClient)
+		require.NoError(t, err)
+
+		var readings []*api.DataReading
+		err = c.PostDataReadingsWithOptions(ctx, readings, client.Options{
+			ClusterID: "success-cluster-id",
+		})
+		require.NoError(t, err)
+	})
+}
+
+// TestCyberArkClient_PostDataReadingsWithOptions_RealAPI demonstrates that the
+// dataupload code works with the real inventory API.
+//
+// To enable verbose request logging:
+//
+//	go test ./pkg/internal/cyberark/dataupload/... \
+//	  -v -count 1 -run TestCyberArkClient_PostDataReadingsWithOptions_RealAPI -args -testing.v 6
+func TestCyberArkClient_PostDataReadingsWithOptions_RealAPI(t *testing.T) {
+	t.Run("success", func(t *testing.T) {
+		subdomain := os.Getenv("ARK_SUBDOMAIN")
+		username := os.Getenv("ARK_USERNAME")
+		secret := os.Getenv("ARK_SECRET")
+
+		if subdomain == "" || username == "" || secret == "" {
+			t.Skip("Skipping because one of the following environment variables is unset or empty: ARK_SUBDOMAIN, ARK_USERNAME, ARK_SECRET")
+			return
+		}
+
+		logger := ktesting.NewLogger(t, ktesting.DefaultConfig)
+		ctx := klog.NewContext(t.Context(), logger)
+
+		c, err := client.NewCyberArk(&http.Client{
+			Transport: transport.NewDebuggingRoundTripper(http.DefaultTransport, transport.DebugByContext),
+		})
+		require.NoError(t, err)
+		var readings []*api.DataReading
+		err = c.PostDataReadingsWithOptions(ctx, readings, client.Options{
+			ClusterID: "success-cluster-id",
+		})
+		require.NoError(t, err)
+	})
+}

pkg/internal/cyberark/client.go

@@ -0,0 +1,60 @@
+package cyberark
+
+import (
+	"context"
+	"errors"
+	"net/http"
+	"os"
+
+	"github.com/jetstack/preflight/pkg/internal/cyberark/dataupload"
+	"github.com/jetstack/preflight/pkg/internal/cyberark/identity"
+	"github.com/jetstack/preflight/pkg/internal/cyberark/servicediscovery"
+)
+
+type ClientConfig struct {
+	subdomain string
+	username  string
+	secret    string
+}
+
+type ClientConfigLoader func() (ClientConfig, error)
+
+func LoadClientConfigFromEnvironment() (ClientConfig, error) {
+	subdomain := os.Getenv("ARK_SUBDOMAIN")
+	username := os.Getenv("ARK_USERNAME")
+	secret := os.Getenv("ARK_SECRET")
+
+	if subdomain == "" || username == "" || secret == "" {
+		return ClientConfig{}, errors.New(
+			"missing environment variables: ARK_SUBDOMAIN, ARK_USERNAME, ARK_SECRET")
+	}
+
+	return ClientConfig{
+		subdomain: subdomain,
+		username:  username,
+		secret:    secret,
+	}, nil
+
+}
+
+func NewDatauploadClient(ctx context.Context, httpClient *http.Client, cfg ClientConfig) (*dataupload.CyberArkClient, error) {
+	discoveryClient := servicediscovery.New(httpClient)
+	serviceMap, err := discoveryClient.DiscoverServices(ctx, cfg.subdomain)
+	if err != nil {
+		return nil, err
+	}
+	identityAPI := serviceMap.Identity.API
+	if identityAPI == "" {
+		return nil, errors.New("service discovery returned an empty identity API")
+	}
+	identityClient := identity.New(httpClient, identityAPI, cfg.subdomain)
+	err = identityClient.LoginUsernamePassword(ctx, cfg.username, []byte(cfg.secret))
+	if err != nil {
+		return nil, err
+	}
+	discoveryAPI := serviceMap.DiscoveryContext.API
+	if discoveryAPI == "" {
+		return nil, errors.New("service discovery returned an empty discovery API")
+	}
+	return dataupload.New(httpClient, discoveryAPI, identityClient.AuthenticateRequest), nil
+}

pkg/internal/cyberark/dataupload/dataupload.go

@@ -35,18 +35,18 @@ type CyberArkClient struct {
 }
 
 type Options struct {
-	ClusterName string
+	ClusterID string
 }
 
-func New(httpClient *http.Client, baseURL string, authenticateRequest func(req *http.Request) error) (*CyberArkClient, error) {
+func New(httpClient *http.Client, baseURL string, authenticateRequest func(req *http.Request) error) *CyberArkClient {
 	return &CyberArkClient{
 		baseURL:             baseURL,
 		httpClient:          httpClient,
 		authenticateRequest: authenticateRequest,
-	}, nil
+	}
 }
 
-// PostDataReadingsWithOptions PUTs the supplied payload to an [AWS presigned URL] which it obtains via the CyberArk inventory API.
+// PutSnapshot PUTs the supplied payload to an [AWS presigned URL] which it obtains via the CyberArk inventory API.
 // [AWS presigned URL]: https://docs.aws.amazon.com/AmazonS3/latest/API/sigv4-query-string-auth.html
 //
 // A SHA256 checksum header is included in the request, to verify that the payload
@@ -60,11 +60,7 @@ func New(httpClient *http.Client, baseURL string, authenticateRequest func(req *
 // If you omit that header, it is possible to PUT any data.
 // There is a work around listed in that issue which we have shared with the
 // CyberArk API team.
-func (c *CyberArkClient) PostDataReadingsWithOptions(ctx context.Context, payload api.DataReadingsPost, opts Options) error {
-	if opts.ClusterName == "" {
-		return fmt.Errorf("programmer mistake: the cluster name (aka `cluster_id` in the config file) cannot be left empty")
-	}
-
+func (c *CyberArkClient) PutSnapshot(ctx context.Context, payload api.DataReadingsPost, opts Options) error {
 	encodedBody := &bytes.Buffer{}
 	hash := sha256.New()
 	if err := json.NewEncoder(io.MultiWriter(encodedBody, hash)).Encode(payload); err != nil {
@@ -114,7 +110,7 @@ func (c *CyberArkClient) retrievePresignedUploadURL(ctx context.Context, checksu
 		Checksum     string `json:"checksum_sha256"`
 		AgentVersion string `json:"agent_version"`
 	}{
-		ClusterID:    opts.ClusterName,
+		ClusterID:    opts.ClusterID,
 		Checksum:     checksum,
 		AgentVersion: version.PreflightVersion,
 	}