CyberArk(client): add CyberArk snapshot conversion and test utilities

- Introduced `ConvertDataReadingsToCyberarkSnapshot` to transform data readings into CyberArk snapshot format.
- Enhanced `PostDataReadingsWithOptions` to utilize the new snapshot conversion.
- Added `DynamicData` and `DiscoveryData` types for structured data handling.
- Updated `DataGathererDynamic` and `DataGathererDiscovery` to return strongly typed data.
- Implemented `ParseDataReadings` in `testutil` for decoding and testing data readings.
- Added test data and golden file support for snapshot conversion validation.

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

Author: wallrj-cyberark

api/datareading.go

@@ -1,8 +1,12 @@
 package api
 
 import (
+	"bytes"
 	"encoding/json"
 	"time"
+
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/version"
 )
 
 // DataReadingsPost is the payload in the upload request.
@@ -28,8 +32,8 @@ type DataReading struct {
 type GatheredResource struct {
 	// Resource is a reference to a k8s object that was found by the informer
 	// should be of type unstructured.Unstructured, raw Object
-	Resource  interface{}
-	DeletedAt Time
+	Resource  interface{} `json:"resource"`
+	DeletedAt Time        `json:"deleted_at,omitempty"`
 }
 
 func (v GatheredResource) MarshalJSON() ([]byte, error) {
@@ -48,3 +52,32 @@ func (v GatheredResource) MarshalJSON() ([]byte, error) {
 
 	return json.Marshal(data)
 }
+
+func (v *GatheredResource) UnmarshalJSON(data []byte) error {
+	var tmpResource struct {
+		Resource  *unstructured.Unstructured `json:"resource"`
+		DeletedAt Time                       `json:"deleted_at,omitempty"`
+	}
+
+	d := json.NewDecoder(bytes.NewReader(data))
+	d.DisallowUnknownFields()
+
+	if err := d.Decode(&tmpResource); err != nil {
+		return err
+	}
+	v.Resource = tmpResource.Resource
+	v.DeletedAt = tmpResource.DeletedAt
+	return nil
+}
+
+// DynamicData is the DataReading.Data returned by the k8s.DataGathererDynamic
+// gatherer
+type DynamicData struct {
+	Items []*GatheredResource `json:"items"`
+}
+
+// DiscoveryData is the DataReading.Data returned by the k8s.ConfigDiscovery
+// gatherer
+type DiscoveryData struct {
+	ServerVersion *version.Info `json:"server_version"`
+}

pkg/agent/run.go

@@ -305,6 +305,14 @@ func gatherAndOutputData(ctx context.Context, eventf Eventf, config CombinedConf
 	var readings []*api.DataReading
 
 	if config.InputPath != "" {
+		// TODO(wallrj): The datareadings read from disk can not yet be pushed
+		// to the CyberArk Discovery and Context API. Why? Because they have
+		// simple data types such as map[string]interface{}. In contrast, the
+		// data from data gatherers can be cast to rich types like DynamicData
+		// or DiscoveryData The CyberArk dataupload client requires the data to
+		// have rich types to convert it to the Discovery and Context snapshots
+		// format. Consider refactoring testutil.ParseDataReadings so that it
+		// can be used here.
 		log.V(logs.Debug).Info("Reading data from local file", "inputPath", config.InputPath)
 		data, err := os.ReadFile(config.InputPath)
 		if err != nil {

pkg/client/client_cyberark.go

@@ -5,6 +5,8 @@ import (
 	"fmt"
 	"net/http"
 
+	"k8s.io/apimachinery/pkg/runtime"
+
 	"github.com/jetstack/preflight/api"
 	"github.com/jetstack/preflight/pkg/internal/cyberark"
 	"github.com/jetstack/preflight/pkg/internal/cyberark/dataupload"
@@ -49,14 +51,118 @@ func (o *CyberArkClient) PostDataReadingsWithOptions(ctx context.Context, readin
 		return fmt.Errorf("while initializing data upload client: %s", err)
 	}
 
-	err = datauploadClient.PutSnapshot(ctx, dataupload.Snapshot{
-		// Temporary hard coded cluster ID.
-		// TODO(wallrj): The clusterID will eventually be extracted from the supplied readings.
-		ClusterID:    "success-cluster-id",
-		AgentVersion: version.PreflightVersion,
-	})
+	snapshot, err := ConvertDataReadingsToCyberarkSnapshot(readings)
+	if err != nil {
+		return fmt.Errorf("while converting data readings: %s", err)
+	}
+	// Temporary hard coded cluster ID.
+	// TODO(wallrj): The clusterID will eventually be extracted from the supplied readings.
+	snapshot.ClusterID = "success-cluster-id"
+
+	err = datauploadClient.PutSnapshot(ctx, snapshot)
 	if err != nil {
 		return fmt.Errorf("while uploading snapshot: %s", err)
 	}
 	return nil
 }
+
+type resourceData map[string][]runtime.Object
+
+// The names of Datagatherers which have the data to populate the Cyberark
+// Snapshot mapped to the key in the Cyberark snapshot.
+var gathererNameToResourceDataKeyMap = map[string]string{
+	"ark/secrets":             "secrets",
+	"ark/serviceaccounts":     "serviceaccounts",
+	"ark/roles":               "roles",
+	"ark/clusterroles":        "clusterroles",
+	"ark/rolebindings":        "rolebindings",
+	"ark/clusterrolebindings": "clusterrolebindings",
+	"ark/jobs":                "jobs",
+	"ark/cronjobs":            "cronjobs",
+	"ark/deployments":         "deployments",
+	"ark/statefulsets":        "statefulsets",
+	"ark/daemonsets":          "daemonsets",
+	"ark/pods":                "pods",
+}
+
+// extractServerVersionFromReading converts the opaque data from a DiscoveryData
+// data reading to allow access to the Kubernetes version fields within.
+func extractServerVersionFromReading(reading *api.DataReading) (string, error) {
+	data, ok := reading.Data.(*api.DiscoveryData)
+	if !ok {
+		return "", fmt.Errorf(
+			"programmer mistake: the DataReading must have data type *api.DiscoveryData. "+
+				"This DataReading (%s) has data type %T", reading.DataGatherer, reading.Data)
+	}
+	if data.ServerVersion == nil {
+		return "unknown", nil
+	}
+	return data.ServerVersion.GitVersion, nil
+}
+
+// extractResourceListFromReading converts the opaque data from a DynamicData
+// data reading to runtime.Object resources, to allow access to the metadata and
+// other kubernetes API fields.
+func extractResourceListFromReading(reading *api.DataReading) ([]runtime.Object, error) {
+	data, ok := reading.Data.(*api.DynamicData)
+	if !ok {
+		return nil, fmt.Errorf(
+			"programmer mistake: the DataReading must have data type *api.DynamicData. "+
+				"This DataReading (%s) has data type %T", reading.DataGatherer, reading.Data)
+	}
+	resources := make([]runtime.Object, len(data.Items))
+	for i, item := range data.Items {
+		if resource, ok := item.Resource.(runtime.Object); ok {
+			resources[i] = resource
+		} else {
+			return nil, fmt.Errorf(
+				"programmer mistake: the DynamicData items must have Resource type runtime.Object. "+
+					"This item (%d) has Resource type %T", i, item.Resource)
+		}
+	}
+	return resources, nil
+}
+
+// ConvertDataReadingsToCyberarkSnapshot converts a list of DataReadings to a CyberArk Snapshot.
+// It extracts the Kubernetes version from the "ark/discovery" DataReading and
+// collects resources from other DataReadings based on their DataGatherer names.
+// If any required data is missing or cannot be converted, an error is returned.
+func ConvertDataReadingsToCyberarkSnapshot(
+	readings []*api.DataReading,
+) (s dataupload.Snapshot, _ error) {
+	k8sVersion := ""
+	resourceData := resourceData{}
+	for _, reading := range readings {
+		if reading.DataGatherer == "ark/discovery" {
+			var err error
+			k8sVersion, err = extractServerVersionFromReading(reading)
+			if err != nil {
+				return s, fmt.Errorf("while extracting server version from data-reading: %s", err)
+			}
+		}
+		if key, found := gathererNameToResourceDataKeyMap[reading.DataGatherer]; found {
+			resources, err := extractResourceListFromReading(reading)
+			if err != nil {
+				return s, fmt.Errorf("while extracting resource list from data-reading: %s", err)
+			}
+			resourceData[key] = append(resourceData[key], resources...)
+		}
+	}
+
+	return dataupload.Snapshot{
+		AgentVersion:        version.PreflightVersion,
+		K8SVersion:          k8sVersion,
+		Secrets:             resourceData["secrets"],
+		ServiceAccounts:     resourceData["serviceaccounts"],
+		Roles:               resourceData["roles"],
+		ClusterRoles:        resourceData["clusterroles"],
+		RoleBindings:        resourceData["rolebindings"],
+		ClusterRoleBindings: resourceData["clusterrolebindings"],
+		Jobs:                resourceData["jobs"],
+		CronJobs:            resourceData["cronjobs"],
+		Deployments:         resourceData["deployments"],
+		Statefulsets:        resourceData["statefulsets"],
+		Daemonsets:          resourceData["daemonsets"],
+		Pods:                resourceData["pods"],
+	}, nil
+}

pkg/client/client_cyberark_test.go

@@ -2,10 +2,13 @@ package client_test
 
 import (
 	"crypto/x509"
+	"encoding/json"
 	"errors"
+	"os"
 	"testing"
 
 	"github.com/jetstack/venafi-connection-lib/http_client"
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"k8s.io/client-go/transport"
 	"k8s.io/klog/v2"
@@ -67,8 +70,25 @@ func TestCyberArkClient_PostDataReadingsWithOptions_RealAPI(t *testing.T) {
 			}
 			require.NoError(t, err)
 		}
-		var readings []*api.DataReading
+		readings := testutil.ParseDataReadings(t, testutil.ReadGZIP(t, "testdata/example-1/datareadings.json.gz"))
 		err = c.PostDataReadingsWithOptions(ctx, readings, client.Options{})
 		require.NoError(t, err)
 	})
 }
+
+func TestConvertDataReadingsToCyberarkSnapshot(t *testing.T) {
+	dataReadings := testutil.ParseDataReadings(t, testutil.ReadGZIP(t, "testdata/example-1/datareadings.json.gz"))
+	snapshot, err := client.ConvertDataReadingsToCyberarkSnapshot(dataReadings)
+	require.NoError(t, err)
+
+	actualSnapshotBytes, err := json.MarshalIndent(snapshot, "", "  ")
+	require.NoError(t, err)
+
+	goldenFilePath := "testdata/example-1/snapshot.json.gz"
+	if _, update := os.LookupEnv("UPDATE_GOLDEN_FILES"); update {
+		testutil.WriteGZIP(t, goldenFilePath, actualSnapshotBytes)
+	} else {
+		expectedSnapshotBytes := testutil.ReadGZIP(t, goldenFilePath)
+		assert.JSONEq(t, string(expectedSnapshotBytes), string(actualSnapshotBytes))
+	}
+}

pkg/client/testdata/example-1/README.md

@@ -0,0 +1,25 @@
+# README
+
+Data captured from a cert-manager E2E test cluster.
+
+```bash
+cd cert-manager
+make e2e-setup
+```
+
+```bash
+cd jetstack-secure
+go run . agent \
+    --log-level 6 \
+    --one-shot \
+    --agent-config-file pkg/client/testdata/example-1/agent.yaml \
+    --output-path pkg/client/testdata/example-1/datareadings.json
+gzip pkg/internal/cyberark/dataupload/testdata/example-1/datareadings.json
+```
+
+
+To recreate the golden output file:
+
+```bash
+UPDATE_GOLDEN_FILES=true go test ./pkg/client/... -run TestConvertDataReadingsToCyberarkSnapshot
+```

pkg/client/testdata/example-1/agent.yaml

@@ -0,0 +1,93 @@
+data-gatherers:
+# gather k8s apiserver version information
+- kind: k8s-discovery
+  name: ark/discovery
+- kind: k8s-dynamic
+  name: ark/secrets
+  config:
+    resource-type:
+      version: v1
+      resource: secrets
+    field-selectors:
+    - type!=kubernetes.io/service-account-token
+    - type!=kubernetes.io/dockercfg
+    - type!=kubernetes.io/dockerconfigjson
+    - type!=kubernetes.io/basic-auth
+    - type!=kubernetes.io/ssh-auth
+    - type!=bootstrap.kubernetes.io/token
+    - type!=helm.sh/release.v1
+- kind: k8s-dynamic
+  name: ark/serviceaccounts
+  config:
+    resource-type:
+      resource: serviceaccounts
+      version: v1
+- kind: k8s-dynamic
+  name: ark/roles
+  config:
+    resource-type:
+      version: v1
+      group: rbac.authorization.k8s.io
+      resource: roles
+- kind: k8s-dynamic
+  name: ark/clusterroles
+  config:
+    resource-type:
+      version: v1
+      group: rbac.authorization.k8s.io
+      resource: clusterroles
+- kind: k8s-dynamic
+  name: ark/rolebindings
+  config:
+    resource-type:
+      version: v1
+      group: rbac.authorization.k8s.io
+      resource: rolebindings
+- kind: k8s-dynamic
+  name: ark/clusterrolebindings
+  config:
+    resource-type:
+      version: v1
+      group: rbac.authorization.k8s.io
+      resource: clusterrolebindings
+- kind: k8s-dynamic
+  name: ark/jobs
+  config:
+    resource-type:
+      version: v1
+      group: batch
+      resource: jobs
+- kind: k8s-dynamic
+  name: ark/cronjobs
+  config:
+    resource-type:
+      version: v1
+      group: batch
+      resource: cronjobs
+- kind: k8s-dynamic
+  name: ark/deployments
+  config:
+    resource-type:
+      version: v1
+      group: apps
+      resource: deployments
+- kind: k8s-dynamic
+  name: ark/statefulsets
+  config:
+    resource-type:
+      version: v1
+      group: apps
+      resource: statefulsets
+- kind: k8s-dynamic
+  name: ark/daemonsets
+  config:
+    resource-type:
+      version: v1
+      group: apps
+      resource: daemonsets
+- kind: k8s-dynamic
+  name: ark/pods
+  config:
+    resource-type:
+      version: v1
+      resource: pods

pkg/client/testdata/example-1/datareadings.json.gz

@@ -6,6 +6,7 @@ import (
 
 	"k8s.io/client-go/discovery"
 
+	"github.com/jetstack/preflight/api"
 	"github.com/jetstack/preflight/pkg/datagatherer"
 )
 
@@ -59,15 +60,12 @@ func (g *DataGathererDiscovery) WaitForCacheSync(ctx context.Context) error {
 
 // Fetch will fetch discovery data from the apiserver, or return an error
 func (g *DataGathererDiscovery) Fetch() (interface{}, int, error) {
-	data, err := g.cl.ServerVersion()
+	serverVersion, err := g.cl.ServerVersion()
 	if err != nil {
 		return nil, -1, fmt.Errorf("failed to get server version: %v", err)
 	}
 
-	response := map[string]interface{}{
-		// data has type Info: https://godoc.org/k8s.io/apimachinery/pkg/version#Info
-		"server_version": data,
-	}
-
-	return response, len(response), nil
+	return &api.DiscoveryData{
+		ServerVersion: serverVersion,
+	}, 1, nil
 }