Add a snapshot JSON format for uploads and convert DataReadings to Snapshot format

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

Author: wallrj

api/datareading.go

@@ -1,8 +1,11 @@
 package api
 
 import (
+	"bytes"
 	"encoding/json"
 	"time"
+
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 )
 
 // DataReadingsPost is the payload in the upload request.
@@ -28,8 +31,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 +51,20 @@ 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
+}

pkg/client/client_cyberark.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 
+	"k8s.io/apimachinery/pkg/version"
 	"k8s.io/client-go/discovery"
 
 	"github.com/jetstack/preflight/pkg/datagatherer"
@@ -57,17 +58,18 @@ func (g *DataGathererDiscovery) WaitForCacheSync(ctx context.Context) error {
 	return nil
 }
 
+type DiscoveryData struct {
+	ServerVersion *version.Info `json:"server_version"`
+}
+
 // 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 &DiscoveryData{
+		ServerVersion: serverVersion,
+	}, 1, nil
 }

pkg/datagatherer/k8s/discovery.go

@@ -307,14 +307,17 @@ func (g *DataGathererDynamic) WaitForCacheSync(ctx context.Context) error {
 	return nil
 }
 
+type DynamicData struct {
+	Items []*api.GatheredResource `json:"items"`
+}
+
 // Fetch will fetch the requested data from the apiserver, or return an error
 // if fetching the data fails.
 func (g *DataGathererDynamic) Fetch() (interface{}, int, error) {
 	if g.groupVersionResource.String() == "" {
 		return nil, -1, fmt.Errorf("resource type must be specified")
 	}
 
-	var list = map[string]interface{}{}
 	var items = []*api.GatheredResource{}
 
 	fetchNamespaces := g.namespaces
@@ -344,10 +347,9 @@ func (g *DataGathererDynamic) Fetch() (interface{}, int, error) {
 		return nil, -1, err
 	}
 
-	// add gathered resources to items
-	list["items"] = items
-
-	return list, len(items), nil
+	return &DynamicData{
+		Items: items,
+	}, len(items), nil
 }
 
 func redactList(list []*api.GatheredResource, excludeAnnotKeys, excludeLabelKeys []*regexp.Regexp) error {

pkg/datagatherer/k8s/dynamic.go

@@ -730,15 +730,12 @@ func TestDynamicGatherer_Fetch(t *testing.T) {
 			}
 
 			if tc.expected != nil {
-				items, ok := res.(map[string]interface{})
+				data, ok := res.(*DynamicData)
 				if !ok {
-					t.Errorf("expected result be an map[string]interface{} but wasn't")
+					t.Errorf("expected result be *DynamicData but wasn't")
 				}
 
-				list, ok := items["items"].([]*api.GatheredResource)
-				if !ok {
-					t.Errorf("expected result be an []*api.GatheredResource but wasn't")
-				}
+				list := data.Items
 				// sorting list of results by name
 				sortGatheredResources(list)
 				// sorting list of expected results by name
@@ -1045,10 +1042,9 @@ func TestDynamicGathererNativeResources_Fetch(t *testing.T) {
 			}
 
 			if tc.expected != nil {
-				res, ok := rawRes.(map[string]interface{})
-				require.Truef(t, ok, "expected result be an map[string]interface{} but wasn't")
-				actual := res["items"].([]*api.GatheredResource)
-				require.Truef(t, ok, "expected result be an []*api.GatheredResource but wasn't")
+				res, ok := rawRes.(*DynamicData)
+				require.Truef(t, ok, "expected result be an *DynamicData but wasn't")
+				actual := res.Items
 
 				// sorting list of results by name
 				sortGatheredResources(actual)

pkg/datagatherer/k8s/dynamic_test.go

@@ -16,6 +16,9 @@ var SecretSelectedFields = []FieldPath{
 	{"metadata", "ownerReferences"},
 	{"metadata", "selfLink"},
 	{"metadata", "uid"},
+	{"metadata", "creationTimestamp"},
+	{"metadata", "deletionTimestamp"},
+	{"metadata", "resourceVersion"},
 
 	{"type"},
 	{"data", "tls.crt"},

pkg/datagatherer/k8s/fieldfilter.go

@@ -12,9 +12,11 @@ import (
 	"net/http"
 	"net/url"
 
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/client-go/transport"
 
 	"github.com/jetstack/preflight/api"
+	"github.com/jetstack/preflight/pkg/datagatherer/k8s"
 	"github.com/jetstack/preflight/pkg/version"
 )
 
@@ -29,6 +31,92 @@ const (
 	apiPathSnapshotLinks = "/api/ingestions/kubernetes/snapshot-links"
 )
 
+type ResourceData map[string][]*unstructured.Unstructured
+
+// Snapshot is the JSON that the CyberArk Discovery and Context API expects to
+// be uploaded to the AWS presigned URL.
+type Snapshot struct {
+	AgentVersion    string                       `json:"agent_version"`
+	ClusterID       string                       `json:"cluster_id"`
+	K8SVersion      string                       `json:"k8s_version"`
+	Secrets         []*unstructured.Unstructured `json:"secrets"`
+	ServiceAccounts []*unstructured.Unstructured `json:"service_accounts"`
+	Roles           []*unstructured.Unstructured `json:"roles"`
+	RoleBindings    []*unstructured.Unstructured `json:"role_bindings"`
+}
+
+// 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":        "roles",
+	"ark/rolebindings":        "rolebindings",
+	"ark/clusterrolebindings": "rolebindings",
+}
+
+func extractResourceListFromReading(reading *api.DataReading) ([]*unstructured.Unstructured, error) {
+	data, ok := reading.Data.(*k8s.DynamicData)
+	if !ok {
+		return nil, fmt.Errorf("failed to convert data: %s", reading.DataGatherer)
+	}
+	items := data.Items
+	resources := make([]*unstructured.Unstructured, len(items))
+	for i, item := range items {
+		if resource, ok := item.Resource.(*unstructured.Unstructured); ok {
+			resources[i] = resource
+		} else {
+			return nil, fmt.Errorf("failed to convert resource: %#v", item)
+		}
+	}
+	return resources, nil
+}
+
+func extractServerVersionFromReading(reading *api.DataReading) (string, error) {
+	data, ok := reading.Data.(*k8s.DiscoveryData)
+	if !ok {
+		return "", fmt.Errorf("failed to convert data: %s", reading.DataGatherer)
+	}
+	if data.ServerVersion == nil {
+		return "unknown", nil
+	}
+	return data.ServerVersion.GitVersion, nil
+}
+
+// ConvertDataReadingsToCyberarkSnapshot converts jetstack-secure DataReadings into Cyberark Snapshot format.
+func ConvertDataReadingsToCyberarkSnapshot(
+	payload api.DataReadingsPost,
+) (_ *Snapshot, err error) {
+	k8sVersion := ""
+	resourceData := ResourceData{}
+	for _, reading := range payload.DataReadings {
+		if reading.DataGatherer == "ark/discovery" {
+			k8sVersion, err = extractServerVersionFromReading(reading)
+			if err != nil {
+				return nil, fmt.Errorf("while extracting server version from data-reading: %s", err)
+			}
+		}
+		if key, found := gathererNameToresourceDataKeyMap[reading.DataGatherer]; found {
+			var resources []*unstructured.Unstructured
+			resources, err = extractResourceListFromReading(reading)
+			if err != nil {
+				return nil, fmt.Errorf("while extracting resource list from data-reading: %s", err)
+			}
+			resourceData[key] = append(resourceData[key], resources...)
+		}
+	}
+
+	return &Snapshot{
+		AgentVersion:    payload.AgentMetadata.Version,
+		ClusterID:       payload.AgentMetadata.ClusterID,
+		K8SVersion:      k8sVersion,
+		Secrets:         resourceData["secrets"],
+		ServiceAccounts: resourceData["serviceaccounts"],
+		Roles:           resourceData["roles"],
+		RoleBindings:    resourceData["rolebindings"],
+	}, nil
+}
+
 type CyberArkClient struct {
 	baseURL string
 	client  *http.Client
@@ -63,9 +151,14 @@ func (c *CyberArkClient) PostDataReadingsWithOptions(ctx context.Context, payloa
 		return fmt.Errorf("programmer mistake: the cluster name (aka `cluster_id` in the config file) cannot be left empty")
 	}
 
+	snapshot, err := ConvertDataReadingsToCyberarkSnapshot(payload)
+	if err != nil {
+		return fmt.Errorf("while converting datareadings to Cyberark snapshot format: %s", err)
+	}
+
 	encodedBody := &bytes.Buffer{}
 	checksum := sha256.New()
-	if err := json.NewEncoder(io.MultiWriter(encodedBody, checksum)).Encode(payload); err != nil {
+	if err := json.NewEncoder(io.MultiWriter(encodedBody, checksum)).Encode(snapshot); err != nil {
 		return err
 	}