WIP: Proof of concept / demo

Signed-off-by: Richard Wall <[email protected]>

Author: wallrj

agent.yaml

@@ -1,17 +1,3 @@
-server: "https://platform.jetstack.io"
-organization_id: "my-organization"
-cluster_id: "my_cluster"
-period: "0h1m0s"
-data-gatherers:
-  - kind: "dummy"
-    name: "dummy"
-    config:
-      failed-attempts: 5
-  - kind: "dummy"
-    name: "dummy-fail"
-    config:
-      always-fail: true
-venafi-cloud:
-  uploader_id: "example-id"
-  upload_path: "/example/endpoint/path"
-  
+machineHub:
+  subdomain: adb-static
+  credentialsSecretName: machinehub-credentials

pkg/agent/config.go

@@ -702,7 +702,6 @@ func ValidateAndCombineConfig(log logr.Logger, cfg Config, flags AgentCmdFlags)
 	if err != nil {
 		return CombinedConfig{}, nil, multierror.Prefix(err, "validating creds:")
 	}
-
 	return res, preflightClient, nil
 }
 

pkg/agent/run.go

@@ -30,6 +30,7 @@ import (
 	"k8s.io/klog/v2"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
 
+	"github.com/jetstack/preflight/pkg/internal/cyberark/identity"
 	"github.com/jetstack/preflight/api"
 	"github.com/jetstack/preflight/pkg/client"
 	"github.com/jetstack/preflight/pkg/clusteruid"
@@ -79,6 +80,23 @@ func Run(cmd *cobra.Command, args []string) (returnErr error) {
 		return fmt.Errorf("While evaluating configuration: %v", err)
 	}
 
+	var caClient *client.CyberArkClient
+	{
+		identityClient, err := identity.New(ctx, cfg.MachineHub.Subdomain)
+		if err != nil {
+			return fmt.Errorf("while creating the CyberArk identity client: %v", err)
+		}
+		username := os.Getenv("ARK_USERNAME")
+		password := []byte(os.Getenv("ARK_SECRET"))
+		if err := identityClient.LoginUsernamePassword(ctx, username, password); err != nil {
+			return fmt.Errorf("while logging in: %v", err)
+		}
+		caClient, err = client.NewCyberArkClient(nil, cfg.MachineHub.Subdomain, identityClient.AuthenticateRequest)
+		if err != nil {
+			return fmt.Errorf("while creating the CyberArk dataupload client: %v", err)
+		}
+	}
+
 	// We need the cluster UID before we progress further so it can be sent along with other data readings
 
 	{
@@ -262,7 +280,7 @@ func Run(cmd *cobra.Command, args []string) (returnErr error) {
 	// be cancelled, which will cause this blocking loop to exit
 	// instead of waiting for the time period.
 	for {
-		if err := gatherAndOutputData(klog.NewContext(ctx, log), eventf, config, preflightClient, dataGatherers); err != nil {
+		if err := gatherAndOutputData(klog.NewContext(ctx, log), eventf, config, preflightClient, caClient, dataGatherers); err != nil {
 			return err
 		}
 
@@ -316,7 +334,7 @@ func newEventf(log logr.Logger, installNS string) (Eventf, error) {
 // Like Printf but for sending events to the agent's Pod object.
 type Eventf func(eventType, reason, msg string, args ...interface{})
 
-func gatherAndOutputData(ctx context.Context, eventf Eventf, config CombinedConfig, preflightClient client.Client, dataGatherers map[string]datagatherer.DataGatherer) error {
+func gatherAndOutputData(ctx context.Context, eventf Eventf, config CombinedConfig, preflightClient client.Client, caClient *client.CyberArkClient, dataGatherers map[string]datagatherer.DataGatherer) error {
 	log := klog.FromContext(ctx).WithName("gatherAndOutputData")
 	var readings []*api.DataReading
 
@@ -359,11 +377,19 @@ func gatherAndOutputData(ctx context.Context, eventf Eventf, config CombinedConf
 			eventf("Warning", "PushingErr", "retrying in %v after error: %s", t, err)
 			log.Info("Warning: PushingErr: retrying", "in", t, "reason", err)
 		})
-
 		if config.MachineHubMode {
+			clusterID := clusteruid.ClusterUIDFromContext(ctx)
+			payload := api.DataReadingsPost{
+				AgentMetadata: &api.AgentMetadata{
+					Version: "",
+					ClusterID: clusterID,
+				},
+				DataReadings: readings,
+			}
 			post := func() (any, error) {
-				log.Info("machine hub mode not yet implemented")
-				return struct{}{}, nil
+				return struct{}{}, caClient.PostDataReadingsWithOptions(ctx, payload, client.CyberArkClientOptions{
+					ClusterName: clusterID,
+				})
 			}
 
 			group.Go(func() error {

pkg/client/client_cyberark.go

@@ -5,5 +5,6 @@ import (
 )
 
 type CyberArkClient = dataupload.CyberArkClient
+type CyberArkClientOptions = dataupload.Options
 
 var NewCyberArkClient = dataupload.NewCyberArkClient

pkg/internal/cyberark/dataupload/dataupload.go

@@ -64,7 +64,7 @@ func (c *CyberArkClient) PostDataReadingsWithOptions(ctx context.Context, payloa
 
 	presignedUploadURL, err := c.retrievePresignedUploadURL(ctx, hex.EncodeToString(checksum.Sum(nil)), opts)
 	if err != nil {
-		return err
+		return fmt.Errorf("while retrieving presigned upload URL: %v", err)
 	}
 
 	req, err := http.NewRequestWithContext(ctx, http.MethodPost, presignedUploadURL, encodedBody)
@@ -120,7 +120,7 @@ func (c *CyberArkClient) retrievePresignedUploadURL(ctx context.Context, checksu
 
 	req.Header.Set("Content-Type", "application/json")
 	if err := c.authenticateRequest(req); err != nil {
-		return "", fmt.Errorf("failed to authenticate request")
+		return "", fmt.Errorf("failed to authenticate request: %v", err)
 	}
 	version.SetUserAgent(req)
 

pkg/internal/cyberark/identity/identity.go

@@ -13,6 +13,7 @@ import (
 
 	"github.com/cenkalti/backoff/v5"
 	"k8s.io/klog/v2"
+	"k8s.io/client-go/transport"
 
 	"github.com/jetstack/preflight/pkg/internal/cyberark/servicediscovery"
 	"github.com/jetstack/preflight/pkg/logs"
@@ -209,11 +210,13 @@ func NewWithDiscoveryClient(ctx context.Context, discoveryClient *servicediscove
 	if err != nil {
 		return nil, err
 	}
+	httpClient := &http.Client{
+		Timeout: 10 * time.Second,
+	}
+	httpClient.Transport = transport.DebugWrappers(http.DefaultTransport)
 
 	return &Client{
-		client: &http.Client{
-			Timeout: 10 * time.Second,
-		},
+		client: httpClient,
 
 		endpoint:  endpoint,
 		subdomain: subdomain,
@@ -277,23 +280,27 @@ func (c *Client) doStartAuthentication(ctx context.Context, username string) (ad
 
 	bodyJSON, err := json.Marshal(body)
 	if err != nil {
+		logger.Error(err, "")
 		return response, fmt.Errorf("failed to marshal JSON for request to StartAuthentication endpoint: %s", err)
 	}
 
 	endpoint, err := url.JoinPath(c.endpoint, "Security", "StartAuthentication")
 	if err != nil {
+		logger.Error(err, "")
 		return response, fmt.Errorf("failed to create URL for request to CyberArk Identity StartAuthentication: %s", err)
 	}
 
 	request, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, bytes.NewReader(bodyJSON))
 	if err != nil {
+		logger.Error(err, "")
 		return response, fmt.Errorf("failed to initialise request to Identity endpoint %s: %s", endpoint, err)
 	}
 
 	setIdentityHeaders(request)
 
 	httpResponse, err := c.client.Do(request)
 	if err != nil {
+		logger.Error(err, "")
 		return response, fmt.Errorf("failed to perform HTTP request to start authentication: %s", err)
 	}
 
@@ -302,6 +309,7 @@ func (c *Client) doStartAuthentication(ctx context.Context, username string) (ad
 	if httpResponse.StatusCode != http.StatusOK {
 		err := fmt.Errorf("got unexpected status code %s from request to start authentication in CyberArk Identity API", httpResponse.Status)
 		if httpResponse.StatusCode >= 500 || httpResponse.StatusCode < 400 {
+			logger.Error(err, "")
 			return response, err
 		}
 
@@ -315,55 +323,69 @@ func (c *Client) doStartAuthentication(ctx context.Context, username string) (ad
 	err = json.NewDecoder(io.LimitReader(httpResponse.Body, maxStartAuthenticationBodySize)).Decode(&startAuthResponse)
 	if err != nil {
 		if err == io.ErrUnexpectedEOF {
+			logger.Error(err, "")
 			return response, fmt.Errorf("rejecting JSON response from server as it was too large or was truncated")
 		}
 
+		logger.Error(err, "")
 		return response, fmt.Errorf("failed to parse JSON from otherwise successful request to start authentication: %s", err)
 	}
 
 	if !startAuthResponse.Success {
-		return response, fmt.Errorf("got a failure response from request to start authentication: message=%q, error=%q", startAuthResponse.Message, startAuthResponse.ErrorID)
+		err := fmt.Errorf("got a failure response from request to start authentication: message=%q, error=%q", startAuthResponse.Message, startAuthResponse.ErrorID)
+		logger.Error(err, "")
+		return response, err
 	}
 
 	logger.V(logs.Debug).Info("made successful request to StartAuthentication", "summary", startAuthResponse.Result.Summary)
 
 	if startAuthResponse.Result.Summary != SummaryNewPackage {
 		// This means we can't respond to whatever summary the server sent.
 		// The best thing to do is try and find a challenge we can solve anyway.
-		klog.FromContext(ctx).Info("got an unexpected Summary from StartAuthentication response; will attempt to complete a login challenge anyway", "summary", startAuthResponse.Result.Summary)
+		logger.Info("got an unexpected Summary from StartAuthentication response; will attempt to complete a login challenge anyway", "summary", startAuthResponse.Result.Summary)
 	}
 
 	// We can only handle a UP type challenge, and if there are any other challenges, we'll have to fail because we can't handle them.
 	// https://github.com/cyberark/ark-sdk-python/blob/3be12c3f2d3a2d0407025028943e584b6edc5996/ark_sdk_python/auth/identity/ark_identity.py#L405
 	switch len(startAuthResponse.Result.Challenges) {
 	case 0:
-		return response, fmt.Errorf("got no valid challenges in response to start authentication; unable to log in")
+		err := fmt.Errorf("got no valid challenges in response to start authentication; unable to log in")
+		logger.Error(err, "")
+		return response, err
 
 	case 1:
 		// do nothing, this is ideal
 
 	default:
-		return response, fmt.Errorf("got %d challenges in response to start authentication, which means MFA may be enabled; unable to log in", len(startAuthResponse.Result.Challenges))
+		err := fmt.Errorf("got %d challenges in response to start authentication, which means MFA may be enabled; unable to log in", len(startAuthResponse.Result.Challenges))
+		logger.Error(err, "")
+		return response, err
 	}
 
 	challenge := startAuthResponse.Result.Challenges[0]
 
 	switch len(challenge.Mechanisms) {
 	case 0:
 		// presumably this shouldn't happen, but handle the case anyway
-		return response, fmt.Errorf("got no mechanisms for challenge from Identity server")
+		err := fmt.Errorf("got no mechanisms for challenge from Identity server")
+		logger.Error(err, "")
+		return response, err
 
 	case 1:
 		// do nothing, this is ideal
 
 	default:
-		return response, fmt.Errorf("got %d mechanisms in response to start authentication, which means MFA may be enabled; unable to log in", len(challenge.Mechanisms))
+		err := fmt.Errorf("got %d mechanisms in response to start authentication, which means MFA may be enabled; unable to log in", len(challenge.Mechanisms))
+		logger.Error(err, "")
+		return response, err
 	}
 
 	mechanism := challenge.Mechanisms[0]
 
 	if !mechanism.Enrolled || mechanism.Name != MechanismUsernamePassword {
-		return response, errNoUPMechanism
+		err := errNoUPMechanism
+		logger.Error(err, "")
+		return response, err
 	}
 
 	response.Action = ActionAnswer
@@ -378,6 +400,10 @@ func (c *Client) doStartAuthentication(ctx context.Context, username string) (ad
 // doAdvanceAuthentication performs the second step of the login process, sending the password to the server
 // and receiving a token in response.
 func (c *Client) doAdvanceAuthentication(ctx context.Context, username string, password *[]byte, requestBody advanceAuthenticationRequestBody) error {
+
+	logger := klog.FromContext(ctx)
+
+
 	if password == nil {
 		return backoff.Permanent(fmt.Errorf("password must not be nil; this is a programming error"))
 	}
@@ -396,13 +422,15 @@ func (c *Client) doAdvanceAuthentication(ctx context.Context, username string, p
 
 	request, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, bytes.NewReader(bodyJSON))
 	if err != nil {
+		logger.Error(err, "")
 		return fmt.Errorf("failed to initialise request to Identity endpoint %s: %s", endpoint, err)
 	}
 
 	setIdentityHeaders(request)
 
 	httpResponse, err := c.client.Do(request)
 	if err != nil {
+		logger.Error(err, "")
 		return fmt.Errorf("failed to perform HTTP request to advance authentication: %s", err)
 	}
 
@@ -413,6 +441,7 @@ func (c *Client) doAdvanceAuthentication(ctx context.Context, username string, p
 	if httpResponse.StatusCode != http.StatusOK {
 		err := fmt.Errorf("got unexpected status code %s from request to advance authentication in CyberArk Identity API", httpResponse.Status)
 		if httpResponse.StatusCode >= 500 || httpResponse.StatusCode < 400 {
+			logger.Error(err, "")
 			return err
 		}
 
@@ -425,14 +454,19 @@ func (c *Client) doAdvanceAuthentication(ctx context.Context, username string, p
 	err = json.NewDecoder(io.LimitReader(httpResponse.Body, maxAdvanceAuthenticationBodySize)).Decode(&advanceAuthResponse)
 	if err != nil {
 		if err == io.ErrUnexpectedEOF {
+			logger.Error(err, "")
 			return fmt.Errorf("rejecting JSON response from server as it was too large or was truncated")
 		}
 
+		logger.Error(err, "")
 		return fmt.Errorf("failed to parse JSON from otherwise successful request to advance authentication: %s", err)
 	}
 
 	if !advanceAuthResponse.Success {
-		return fmt.Errorf("got a failure response from request to advance authentication: message=%q, error=%q", advanceAuthResponse.Message, advanceAuthResponse.ErrorID)
+		// TODO: Permanent error?
+		err := fmt.Errorf("got a failure response from request to advance authentication: message=%q, error=%q", advanceAuthResponse.Message, advanceAuthResponse.ErrorID)
+		logger.Error(err, "")
+		return err
 	}
 
 	if advanceAuthResponse.Result.Summary != SummaryLoginSuccess {
@@ -441,7 +475,7 @@ func (c *Client) doAdvanceAuthentication(ctx context.Context, username string, p
 		return backoff.Permanent(fmt.Errorf("got a %s response from AdvanceAuthentication; this implies that the user account %s requires MFA, which is not supported. Try unlocking MFA for this user", advanceAuthResponse.Result.Summary, username))
 	}
 
-	klog.FromContext(ctx).Info("successfully completed AdvanceAuthentication request to CyberArk Identity; login complete", "username", username)
+	logger.Info("successfully completed AdvanceAuthentication request to CyberArk Identity; login complete", "username", username)
 
 	c.tokenCachedMutex.Lock()
 

pkg/internal/cyberark/servicediscovery/discovery.go

@@ -9,6 +9,8 @@ import (
 	"net/url"
 	"time"
 
+	"k8s.io/client-go/transport"
+
 	"github.com/jetstack/preflight/pkg/version"
 )
 
@@ -63,6 +65,7 @@ func New(clientOpts ...ClientOpt) *Client {
 	client := &Client{
 		client: &http.Client{
 			Timeout: 10 * time.Second,
+			Transport: transport.DebugWrappers(http.DefaultTransport),
 		},
 		endpoint: prodDiscoveryEndpoint,
 	}