Merge pull request #772 from jetstack/sigv4

Use sigv4 for sending data to s3

Author: SgtCoDFish

internal/cyberark/client.go

@@ -49,7 +49,7 @@ func LoadClientConfigFromEnvironment() (ClientConfig, error) {
 // NewDatauploadClient initializes and returns a new CyberArk Data Upload client.
 // It performs service discovery to find the necessary API endpoints and authenticates
 // using the provided client configuration.
-func NewDatauploadClient(ctx context.Context, httpClient *http.Client, serviceMap *servicediscovery.Services, cfg ClientConfig) (*dataupload.CyberArkClient, error) {
+func NewDatauploadClient(ctx context.Context, httpClient *http.Client, serviceMap *servicediscovery.Services, tenantUUID string, cfg ClientConfig) (*dataupload.CyberArkClient, error) {
 	identityAPI := serviceMap.Identity.API
 	if identityAPI == "" {
 		return nil, errors.New("service discovery returned an empty identity API")
@@ -67,5 +67,5 @@ func NewDatauploadClient(ctx context.Context, httpClient *http.Client, serviceMa
 		return nil, err
 	}
 
-	return dataupload.New(httpClient, discoveryAPI, identityClient.AuthenticateRequest), nil
+	return dataupload.New(httpClient, discoveryAPI, tenantUUID, identityClient.AuthenticateRequest), nil
 }

internal/cyberark/client_test.go

@@ -34,12 +34,12 @@ func TestCyberArkClient_PutSnapshot_MockAPI(t *testing.T) {
 
 	discoveryClient := servicediscovery.New(httpClient)
 
-	serviceMap, err := discoveryClient.DiscoverServices(t.Context(), cfg.Subdomain)
+	serviceMap, tenantUUID, err := discoveryClient.DiscoverServices(t.Context(), cfg.Subdomain)
 	if err != nil {
 		t.Fatalf("failed to discover mock services: %v", err)
 	}
 
-	cl, err := cyberark.NewDatauploadClient(ctx, httpClient, serviceMap, cfg)
+	cl, err := cyberark.NewDatauploadClient(ctx, httpClient, serviceMap, tenantUUID, cfg)
 	require.NoError(t, err)
 
 	err = cl.PutSnapshot(ctx, dataupload.Snapshot{
@@ -78,12 +78,12 @@ func TestCyberArkClient_PutSnapshot_RealAPI(t *testing.T) {
 
 	discoveryClient := servicediscovery.New(httpClient)
 
-	serviceMap, err := discoveryClient.DiscoverServices(t.Context(), cfg.Subdomain)
+	serviceMap, tenantUUID, err := discoveryClient.DiscoverServices(t.Context(), cfg.Subdomain)
 	if err != nil {
 		t.Fatalf("failed to discover services: %v", err)
 	}
 
-	cl, err := cyberark.NewDatauploadClient(ctx, httpClient, serviceMap, cfg)
+	cl, err := cyberark.NewDatauploadClient(ctx, httpClient, serviceMap, tenantUUID, cfg)
 	require.NoError(t, err)
 
 	err = cl.PutSnapshot(ctx, dataupload.Snapshot{

internal/cyberark/dataupload/dataupload.go

@@ -15,6 +15,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 
 	arkapi "github.com/jetstack/preflight/internal/cyberark/api"
+	"github.com/jetstack/preflight/internal/cyberark/identity"
 	"github.com/jetstack/preflight/pkg/version"
 )
 
@@ -33,13 +34,19 @@ type CyberArkClient struct {
 	baseURL    string
 	httpClient *http.Client
 
-	authenticateRequest func(req *http.Request) error
+	tenantUUID string
+
+	authenticateRequest identity.RequestAuthenticator
 }
 
-func New(httpClient *http.Client, baseURL string, authenticateRequest func(req *http.Request) error) *CyberArkClient {
+// New creates a new CyberArkClient. The tenant UUID is best sourced from service discovery along with the base URL.
+func New(httpClient *http.Client, baseURL string, tenantUUID string, authenticateRequest identity.RequestAuthenticator) *CyberArkClient {
 	return &CyberArkClient{
-		baseURL:             baseURL,
-		httpClient:          httpClient,
+		baseURL:    baseURL,
+		httpClient: httpClient,
+
+		tenantUUID: tenantUUID,
+
 		authenticateRequest: authenticateRequest,
 	}
 }
@@ -102,13 +109,6 @@ type Snapshot struct {
 // has been received intact.
 // Read [Checking object integrity for data uploads in Amazon S3](https://docs.aws.amazon.com/AmazonS3/latest/userguide/checking-object-integrity-upload.html),
 // to learn more.
-//
-// TODO(wallrj): There is a bug in the AWS backend:
-// [S3 Presigned PutObjectCommand URLs ignore Sha256 Hash when uploading](https://github.com/aws/aws-sdk/issues/480)
-// ...which means that the `x-amz-checksum-sha256` request header is optional.
-// 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) PutSnapshot(ctx context.Context, snapshot Snapshot) error {
 	if snapshot.ClusterID == "" {
 		return fmt.Errorf("programmer mistake: the snapshot cluster ID cannot be left empty")
@@ -119,10 +119,12 @@ func (c *CyberArkClient) PutSnapshot(ctx context.Context, snapshot Snapshot) err
 	if err := json.NewEncoder(io.MultiWriter(encodedBody, hash)).Encode(snapshot); err != nil {
 		return err
 	}
+
 	checksum := hash.Sum(nil)
 	checksumHex := hex.EncodeToString(checksum)
 	checksumBase64 := base64.StdEncoding.EncodeToString(checksum)
-	presignedUploadURL, err := c.retrievePresignedUploadURL(ctx, checksumHex, snapshot.ClusterID)
+
+	presignedUploadURL, username, err := c.retrievePresignedUploadURL(ctx, checksumHex, snapshot.ClusterID, int64(encodedBody.Len()))
 	if err != nil {
 		return fmt.Errorf("while retrieving snapshot upload URL: %s", err)
 	}
@@ -132,7 +134,21 @@ func (c *CyberArkClient) PutSnapshot(ctx context.Context, snapshot Snapshot) err
 	if err != nil {
 		return err
 	}
+
 	req.Header.Set("X-Amz-Checksum-Sha256", checksumBase64)
+	req.Header.Set("X-Amz-Server-Side-Encryption", "AES256")
+
+	q := url.Values{}
+
+	q.Add("agent_version", snapshot.AgentVersion)
+	q.Add("tenant_id", c.tenantUUID)
+	q.Add("upload_type", "k8s_snapshot")
+	q.Add("uploader_id", snapshot.ClusterID)
+	q.Add("username", username)
+	q.Add("vendor", "k8s")
+
+	req.Header.Set("X-Amz-Tagging", q.Encode())
+
 	version.SetUserAgent(req)
 
 	res, err := c.httpClient.Do(req)
@@ -152,44 +168,57 @@ func (c *CyberArkClient) PutSnapshot(ctx context.Context, snapshot Snapshot) err
 	return nil
 }
 
-func (c *CyberArkClient) retrievePresignedUploadURL(ctx context.Context, checksum string, clusterID string) (string, error) {
+// RetrievePresignedUploadURLRequest is the JSON body sent to the inventory API to request a presigned upload URL.
+type RetrievePresignedUploadURLRequest struct {
+	ClusterID string `json:"cluster_id"`
+	Checksum  string `json:"checksum_sha256"`
+
+	// AgentVersion is the v-prefixed version of the agent uploading the snapshot.
+	// Note that the backend relies on this version being v-prefixed semver.
+	AgentVersion string `json:"agent_version"`
+
+	// FileSize is the size of the data we'll upload in bytes
+	FileSize int64 `json:"file_size"`
+}
+
+func (c *CyberArkClient) retrievePresignedUploadURL(ctx context.Context, checksum string, clusterID string, fileSize int64) (string, string, error) {
 	uploadURL, err := url.JoinPath(c.baseURL, apiPathSnapshotLinks)
 	if err != nil {
-		return "", err
+		return "", "", err
 	}
 
-	request := struct {
-		ClusterID    string `json:"cluster_id"`
-		Checksum     string `json:"checksum_sha256"`
-		AgentVersion string `json:"agent_version"`
-	}{
+	request := RetrievePresignedUploadURLRequest{
 		ClusterID:    clusterID,
 		Checksum:     checksum,
 		AgentVersion: version.PreflightVersion,
+		FileSize:     fileSize,
 	}
 
 	encodedBody := &bytes.Buffer{}
 	if err := json.NewEncoder(encodedBody).Encode(request); err != nil {
-		return "", err
+		return "", "", err
 	}
 
 	req, err := http.NewRequestWithContext(ctx, http.MethodPost, uploadURL, encodedBody)
 	if err != nil {
-		return "", err
+		return "", "", err
 	}
 
 	req.Header.Set("Content-Type", "application/json")
-	if err := c.authenticateRequest(req); err != nil {
-		return "", fmt.Errorf("failed to authenticate request: %s", err)
+
+	username, err := c.authenticateRequest(req)
+	if err != nil {
+		return "", "", fmt.Errorf("failed to authenticate request: %s", err)
 	}
+
 	version.SetUserAgent(req)
 
 	// Add telemetry headers
 	arkapi.SetTelemetryRequestHeader(req)
 
 	res, err := c.httpClient.Do(req)
 	if err != nil {
-		return "", err
+		return "", "", err
 	}
 	defer res.Body.Close()
 
@@ -198,7 +227,7 @@ func (c *CyberArkClient) retrievePresignedUploadURL(ctx context.Context, checksu
 		if len(body) == 0 {
 			body = []byte(`<empty body>`)
 		}
-		return "", fmt.Errorf("received response with status code %d: %s", code, bytes.TrimSpace(body))
+		return "", "", fmt.Errorf("received response with status code %d: %s", code, bytes.TrimSpace(body))
 	}
 
 	response := struct {
@@ -207,11 +236,11 @@ func (c *CyberArkClient) retrievePresignedUploadURL(ctx context.Context, checksu
 
 	if err := json.NewDecoder(io.LimitReader(res.Body, maxRetrievePresignedUploadURLBodySize)).Decode(&response); err != nil {
 		if err == io.ErrUnexpectedEOF {
-			return "", fmt.Errorf("rejecting JSON response from server as it was too large or was truncated")
+			return "", "", fmt.Errorf("rejecting JSON response from server as it was too large or was truncated")
 		}
 
-		return "", fmt.Errorf("failed to parse JSON from otherwise successful request to start data upload: %s", err)
+		return "", "", fmt.Errorf("failed to parse JSON from otherwise successful request to start data upload: %s", err)
 	}
 
-	return response.URL, nil
+	return response.URL, username, nil
 }

internal/cyberark/dataupload/dataupload_test.go

@@ -10,6 +10,7 @@ import (
 	"k8s.io/klog/v2/ktesting"
 
 	"github.com/jetstack/preflight/internal/cyberark/dataupload"
+	"github.com/jetstack/preflight/internal/cyberark/identity"
 	"github.com/jetstack/preflight/pkg/version"
 
 	_ "k8s.io/klog/v2/ktesting/init"
@@ -19,17 +20,17 @@ import (
 // mock API server. The mock server is configured to return different responses
 // based on the cluster ID and bearer token used in the request.
 func TestCyberArkClient_PutSnapshot_MockAPI(t *testing.T) {
-	setToken := func(token string) func(*http.Request) error {
-		return func(req *http.Request) error {
+	setToken := func(token string) identity.RequestAuthenticator {
+		return func(req *http.Request) (string, error) {
 			req.Header.Set("Authorization", fmt.Sprintf("Bearer %s", token))
-			return nil
+			return "foo@example.com", nil // set a dummy username for testing purposes; the actual value is not important for these tests
 		}
 	}
 
 	tests := []struct {
 		name         string
 		snapshot     dataupload.Snapshot
-		authenticate func(req *http.Request) error
+		authenticate identity.RequestAuthenticator
 		requireFn    func(t *testing.T, err error)
 	}{
 		{
@@ -96,7 +97,7 @@ func TestCyberArkClient_PutSnapshot_MockAPI(t *testing.T) {
 
 			datauploadAPIBaseURL, httpClient := dataupload.MockDataUploadServer(t)
 
-			cyberArkClient := dataupload.New(httpClient, datauploadAPIBaseURL, tc.authenticate)
+			cyberArkClient := dataupload.New(httpClient, datauploadAPIBaseURL, "test-tenant-uuid", tc.authenticate)
 
 			err := cyberArkClient.PutSnapshot(ctx, tc.snapshot)
 			tc.requireFn(t, err)