Worker storage bucket credential state (#46)
* feat: parse generic aws api errors * feat: parse s3 api errors * feat: return credential state for all status errors * fix status codes on certain http error codes and re-enable two tests * switch code to FailedPrecondition * update certain error codes * update go.mod for new sdk version --------- Co-authored-by: Damian Debkowski <damian.debkowski@hashicorp.com>
This commit is contained in:
1 parent
88798fb76b
commit
5add0a1b53
16 files changed
+2090
-394
No files matched your search
@@ -132,7 +132,7 @@ func TestPluginOnCreateCatalogErr(t *testing.T) {
|
||||
),
|
||||
}),
|
||||
},
|
||||
expectedErrContains: fmt.Sprintf("error during credential rotation: error rotating credentials: error calling CreateAccessKey: error calling iam.GetUser: %s", testGetUserErr),
|
||||
expectedErrContains: "aws service unknown: unknown error: error rotating credentials",
|
||||
expectedErrCode: codes.InvalidArgument,
|
||||
},
|
||||
{
|
||||
@@ -163,7 +163,7 @@ func TestPluginOnCreateCatalogErr(t *testing.T) {
|
||||
),
|
||||
}),
|
||||
},
|
||||
expectedErrContains: fmt.Sprintf("error during credential validation: error validating credentials: %s", testGetCallerIdentityErr),
|
||||
expectedErrContains: "aws service unknown: unknown error: validating credentials",
|
||||
expectedErrCode: codes.InvalidArgument,
|
||||
},
|
||||
}
|
||||
@@ -342,7 +342,7 @@ func TestPluginOnUpdateCatalogErr(t *testing.T) {
|
||||
),
|
||||
}),
|
||||
},
|
||||
expectedErrContains: fmt.Sprintf("error attempting to replace credentials: error deleting old access key: %s", testDeleteAccessKeyErr),
|
||||
expectedErrContains: "aws service unknown: unknown error: failed to delete access key",
|
||||
expectedErrCode: codes.InvalidArgument,
|
||||
},
|
||||
{
|
||||
@@ -375,7 +375,7 @@ func TestPluginOnUpdateCatalogErr(t *testing.T) {
|
||||
),
|
||||
}),
|
||||
},
|
||||
expectedErrContains: fmt.Sprintf("error during credential rotation: error rotating credentials: error calling CreateAccessKey: error calling iam.GetUser: %s", testGetUserErr),
|
||||
expectedErrContains: "aws service unknown: unknown error: error rotating credentials",
|
||||
expectedErrCode: codes.InvalidArgument,
|
||||
},
|
||||
}
|
||||
@@ -466,7 +466,7 @@ func TestPluginOnDeleteCatalogErr(t *testing.T) {
|
||||
),
|
||||
}),
|
||||
},
|
||||
expectedErrContains: fmt.Sprintf("error removing rotated credentials during catalog deletion: error deleting old access key: %s", testDeleteAccessKeyErr),
|
||||
expectedErrContains: "aws service unknown: unknown error: failed to delete access key",
|
||||
expectedErrCode: codes.Aborted,
|
||||
},
|
||||
}
|
||||
|
||||
@@ -7,6 +7,8 @@ import (
|
||||
"testing"
|
||||
|
||||
cred "github.com/hashicorp/boundary-plugin-aws/internal/credential"
|
||||
pb "github.com/hashicorp/boundary/sdk/pbs/plugin"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
"google.golang.org/grpc/codes"
|
||||
"google.golang.org/grpc/status"
|
||||
@@ -19,6 +21,7 @@ func TestGetStorageAttributes(t *testing.T) {
|
||||
in *structpb.Struct
|
||||
expected *StorageAttributes
|
||||
expectedErrContains string
|
||||
expectedDetails *pb.StorageBucketCredentialState
|
||||
}{
|
||||
{
|
||||
name: "missing region",
|
||||
@@ -100,13 +103,17 @@ func TestGetStorageAttributes(t *testing.T) {
|
||||
for _, tc := range cases {
|
||||
tc := tc
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
require := require.New(t)
|
||||
require, assert := require.New(t), assert.New(t)
|
||||
|
||||
actual, err := getStorageAttributes(tc.in)
|
||||
if tc.expectedErrContains != "" {
|
||||
require.Error(err)
|
||||
require.Contains(err.Error(), tc.expectedErrContains)
|
||||
require.Equal(status.Code(err), codes.InvalidArgument)
|
||||
require.Equal(codes.InvalidArgument, status.Code(err))
|
||||
|
||||
st, ok := status.FromError(err)
|
||||
require.True(ok)
|
||||
assert.Len(st.Details(), 0)
|
||||
return
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,68 @@
|
||||
package storage
|
||||
|
||||
import (
|
||||
"github.com/hashicorp/boundary-plugin-aws/internal/errors"
|
||||
pb "github.com/hashicorp/boundary/sdk/pbs/plugin"
|
||||
"google.golang.org/grpc/codes"
|
||||
"google.golang.org/grpc/status"
|
||||
)
|
||||
|
||||
const (
|
||||
s3StatusErrMsg = "aws s3 error: %s"
|
||||
|
||||
// s3ErrorBadDigest is returned when there is a checksum
|
||||
// mismatch between the request and the s3 service. This
|
||||
// error code is returned for the PutObject request.
|
||||
s3ErrorBadDigest = "BadDigest"
|
||||
)
|
||||
|
||||
func setStatePermission(req any, permission *pb.Permission) *pb.StorageBucketCredentialState {
|
||||
state := &pb.StorageBucketCredentialState{
|
||||
State: &pb.Permissions{},
|
||||
}
|
||||
switch req.(type) {
|
||||
case pb.HeadObjectRequest:
|
||||
state.State.Read = permission
|
||||
case *pb.HeadObjectRequest:
|
||||
state.State.Read = permission
|
||||
case pb.GetObjectRequest:
|
||||
state.State.Read = permission
|
||||
case *pb.GetObjectRequest:
|
||||
state.State.Read = permission
|
||||
case pb.PutObjectRequest:
|
||||
state.State.Write = permission
|
||||
case *pb.PutObjectRequest:
|
||||
state.State.Write = permission
|
||||
case pb.DeleteObjectsRequest:
|
||||
state.State.Delete = permission
|
||||
case *pb.DeleteObjectsRequest:
|
||||
state.State.Delete = permission
|
||||
}
|
||||
return state
|
||||
}
|
||||
|
||||
// parseS3Error will convert an S3 api error into a RPC status. If the
|
||||
// error does not match against an s3 error type, it will fallback to
|
||||
// evaluating against generic aws service errors. This includes throttling,
|
||||
// credentials, and connectivity error types.
|
||||
//
|
||||
// https://docs.aws.amazon.com/AmazonS3/latest/API/ErrorResponses.html
|
||||
func parseS3Error(err error, msg string, req any) *status.Status {
|
||||
if err == nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
st, p := errors.ParseAWSError(err, msg)
|
||||
if p != nil {
|
||||
state := setStatePermission(req, p)
|
||||
if st, err = st.WithDetails(state); err != nil {
|
||||
st = status.New(codes.Internal, err.Error())
|
||||
}
|
||||
}
|
||||
|
||||
return st
|
||||
}
|
||||
|
||||
func checksumMistmatchStatus() error {
|
||||
return status.New(codes.Aborted, "mismatched checksum").Err()
|
||||
}
|
||||
@@ -0,0 +1,180 @@
|
||||
package storage
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"github.com/hashicorp/boundary-plugin-aws/internal/errors"
|
||||
pb "github.com/hashicorp/boundary/sdk/pbs/plugin"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
"google.golang.org/grpc/codes"
|
||||
"google.golang.org/protobuf/types/known/timestamppb"
|
||||
)
|
||||
|
||||
func Test_ParseS3Error(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
cases := []struct {
|
||||
name string
|
||||
req any
|
||||
err error
|
||||
expectedStatusCode codes.Code
|
||||
expectedStatusMsg string
|
||||
expectedDetails *pb.StorageBucketCredentialState
|
||||
}{
|
||||
{
|
||||
name: "nil-error",
|
||||
expectedStatusCode: codes.OK,
|
||||
},
|
||||
{
|
||||
name: "no-such-bucket",
|
||||
req: &pb.GetObjectRequest{},
|
||||
err: TestAwsS3Error("NoSuchBucket", "GetObject", "no such bucket"),
|
||||
expectedStatusCode: codes.NotFound,
|
||||
expectedStatusMsg: "aws s3 error: test",
|
||||
expectedDetails: &pb.StorageBucketCredentialState{
|
||||
State: &pb.Permissions{
|
||||
Read: &pb.Permission{
|
||||
State: pb.StateType_STATE_TYPE_ERROR,
|
||||
ErrorDetails: "no such bucket",
|
||||
CheckedAt: timestamppb.Now(),
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "no-such-key",
|
||||
req: &pb.GetObjectRequest{},
|
||||
err: TestAwsS3Error("NoSuchKey", "GetObject", "no such key"),
|
||||
expectedStatusCode: codes.NotFound,
|
||||
expectedStatusMsg: "aws s3 error: test",
|
||||
expectedDetails: &pb.StorageBucketCredentialState{
|
||||
State: &pb.Permissions{
|
||||
Read: &pb.Permission{
|
||||
State: pb.StateType_STATE_TYPE_UNKNOWN,
|
||||
CheckedAt: timestamppb.Now(),
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "not-found",
|
||||
req: &pb.GetObjectRequest{},
|
||||
err: errors.TestAwsHttpResponseError(404, "not found"),
|
||||
expectedStatusCode: codes.NotFound,
|
||||
// cannot embed services within mock http errors, therefore unknown service
|
||||
expectedStatusMsg: "aws service unknown: resource not found error: test",
|
||||
expectedDetails: &pb.StorageBucketCredentialState{
|
||||
State: &pb.Permissions{
|
||||
Read: &pb.Permission{
|
||||
State: pb.StateType_STATE_TYPE_UNKNOWN,
|
||||
CheckedAt: timestamppb.Now(),
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "invalid-object-state",
|
||||
req: &pb.GetObjectRequest{},
|
||||
err: TestAwsS3Error("InvalidObjectState", "GetObject", "invalid object state"),
|
||||
expectedStatusCode: codes.NotFound,
|
||||
expectedStatusMsg: "aws s3 error: test",
|
||||
expectedDetails: &pb.StorageBucketCredentialState{
|
||||
State: &pb.Permissions{
|
||||
Read: &pb.Permission{
|
||||
State: pb.StateType_STATE_TYPE_UNKNOWN,
|
||||
CheckedAt: timestamppb.Now(),
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "bad-digest",
|
||||
req: &pb.PutObjectRequest{},
|
||||
err: TestAwsS3Error(s3ErrorBadDigest, "PutObject", "checksum mismatch"),
|
||||
expectedStatusCode: codes.Aborted,
|
||||
expectedStatusMsg: "aws s3 error: test",
|
||||
expectedDetails: &pb.StorageBucketCredentialState{
|
||||
State: &pb.Permissions{
|
||||
Write: &pb.Permission{
|
||||
State: pb.StateType_STATE_TYPE_UNKNOWN,
|
||||
CheckedAt: timestamppb.Now(),
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "fallback-throttle-error",
|
||||
req: &pb.GetObjectRequest{},
|
||||
err: TestAwsS3Error("Throttling", "action", "throttling exception"),
|
||||
expectedStatusCode: codes.Unavailable,
|
||||
expectedStatusMsg: "aws service s3: throttling error: test",
|
||||
expectedDetails: &pb.StorageBucketCredentialState{
|
||||
State: &pb.Permissions{
|
||||
Read: &pb.Permission{
|
||||
State: pb.StateType_STATE_TYPE_UNKNOWN,
|
||||
CheckedAt: timestamppb.Now(),
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "fallback-connectivity-error",
|
||||
req: &pb.GetObjectRequest{},
|
||||
err: TestAwsS3Error("RequestTimeout", "action", "request timeout"),
|
||||
expectedStatusCode: codes.DeadlineExceeded,
|
||||
expectedStatusMsg: "aws s3 error: test",
|
||||
expectedDetails: &pb.StorageBucketCredentialState{
|
||||
State: &pb.Permissions{
|
||||
Read: &pb.Permission{
|
||||
State: pb.StateType_STATE_TYPE_UNKNOWN,
|
||||
CheckedAt: timestamppb.Now(),
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "fallback-credentials-error",
|
||||
req: &pb.GetObjectRequest{},
|
||||
err: TestAwsS3Error("AccessDenied", "action", "access denied"),
|
||||
expectedStatusCode: codes.PermissionDenied,
|
||||
expectedStatusMsg: "aws service s3: invalid credentials error: test",
|
||||
expectedDetails: &pb.StorageBucketCredentialState{
|
||||
State: &pb.Permissions{
|
||||
Read: &pb.Permission{
|
||||
State: pb.StateType_STATE_TYPE_ERROR,
|
||||
ErrorDetails: "access denied",
|
||||
CheckedAt: timestamppb.Now(),
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
for _, tt := range cases {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
require, assert := require.New(t), assert.New(t)
|
||||
|
||||
actualStatus := parseS3Error(tt.err, "test", tt.req)
|
||||
if tt.expectedStatusCode == codes.OK {
|
||||
require.Nil(actualStatus)
|
||||
return
|
||||
}
|
||||
|
||||
require.Equal(tt.expectedStatusCode, actualStatus.Code())
|
||||
require.Equal(tt.expectedStatusMsg, actualStatus.Message())
|
||||
require.Len(actualStatus.Details(), 1)
|
||||
|
||||
foundDetail := false
|
||||
for _, detail := range actualStatus.Details() {
|
||||
switch errDetail := detail.(type) {
|
||||
case *pb.StorageBucketCredentialState:
|
||||
foundDetail = true
|
||||
errors.CheckSimilarPermission(assert, tt.expectedDetails.State.Read, errDetail.State.Read, false)
|
||||
errors.CheckSimilarPermission(assert, tt.expectedDetails.State.Write, errDetail.State.Write, false)
|
||||
errors.CheckSimilarPermission(assert, tt.expectedDetails.State.Delete, errDetail.State.Delete, false)
|
||||
}
|
||||
}
|
||||
require.True(foundDetail, "did not find error details with status")
|
||||
})
|
||||
}
|
||||
}
|
||||
+128
-111
@@ -9,19 +9,21 @@ import (
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/base64"
|
||||
"fmt"
|
||||
"io"
|
||||
"os"
|
||||
"path"
|
||||
"strings"
|
||||
|
||||
"github.com/aws/aws-sdk-go-v2/aws"
|
||||
"github.com/aws/aws-sdk-go-v2/aws/retry"
|
||||
"github.com/aws/aws-sdk-go-v2/service/s3"
|
||||
"github.com/aws/aws-sdk-go-v2/service/s3/types"
|
||||
"github.com/google/uuid"
|
||||
cred "github.com/hashicorp/boundary-plugin-aws/internal/credential"
|
||||
"github.com/hashicorp/boundary-plugin-aws/internal/errors"
|
||||
"github.com/hashicorp/boundary/sdk/pbs/controller/api/resources/storagebuckets"
|
||||
pb "github.com/hashicorp/boundary/sdk/pbs/plugin"
|
||||
"github.com/hashicorp/go-multierror"
|
||||
"google.golang.org/grpc/codes"
|
||||
"google.golang.org/grpc/status"
|
||||
"google.golang.org/protobuf/types/known/timestamppb"
|
||||
@@ -46,15 +48,15 @@ type StoragePlugin struct {
|
||||
func (p *StoragePlugin) OnCreateStorageBucket(ctx context.Context, req *pb.OnCreateStorageBucketRequest) (*pb.OnCreateStorageBucketResponse, error) {
|
||||
bucket := req.GetBucket()
|
||||
if bucket == nil {
|
||||
return nil, status.Error(codes.InvalidArgument, "bucket is required")
|
||||
return nil, errors.BadRequestStatus("bucket is required")
|
||||
}
|
||||
if bucket.BucketName == "" {
|
||||
return nil, status.Error(codes.InvalidArgument, "bucketName is required")
|
||||
return nil, errors.BadRequestStatus("bucketName is required")
|
||||
}
|
||||
|
||||
attrs := bucket.GetAttributes()
|
||||
if attrs == nil {
|
||||
return nil, status.Error(codes.InvalidArgument, "attributes is required")
|
||||
return nil, errors.BadRequestStatus("attributes is required")
|
||||
}
|
||||
|
||||
storageAttributes, err := getStorageAttributes(attrs)
|
||||
@@ -74,13 +76,13 @@ func (p *StoragePlugin) OnCreateStorageBucket(ctx context.Context, req *pb.OnCre
|
||||
}, p.testCredStateOpts...)...,
|
||||
)
|
||||
if err != nil {
|
||||
return nil, status.Errorf(codes.InvalidArgument, "error setting up persisted state: %s", err)
|
||||
return nil, errors.BadRequestStatus("error setting up persisted state: %s", err)
|
||||
}
|
||||
|
||||
// Try to rotate AWS static credentials
|
||||
if cred.GetCredentialType(credConfig) == cred.StaticAWS && !storageAttributes.DisableCredentialRotation {
|
||||
if err := credState.RotateCreds(ctx); err != nil {
|
||||
return nil, status.Errorf(codes.InvalidArgument, "error during credential rotation: %s", err)
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
|
||||
@@ -90,17 +92,17 @@ func (p *StoragePlugin) OnCreateStorageBucket(ctx context.Context, req *pb.OnCre
|
||||
}, p.testStorageStateOpts...)...,
|
||||
)
|
||||
if err != nil {
|
||||
return nil, status.Errorf(codes.InvalidArgument, "error setting up persisted state: %s", err)
|
||||
return nil, errors.BadRequestStatus("error setting up persisted state: %s", err)
|
||||
}
|
||||
|
||||
// perform dry run to ensure we can interact with the bucket as expected.
|
||||
if err := dryRunValidation(ctx, storageState, storageAttributes, bucket); err != nil {
|
||||
return nil, err
|
||||
if st := dryRunValidation(ctx, storageState, storageAttributes, bucket); st != nil {
|
||||
return nil, st.Err()
|
||||
}
|
||||
|
||||
persistedProto, err := storageState.toProto()
|
||||
if err != nil {
|
||||
return nil, status.Error(codes.InvalidArgument, err.Error())
|
||||
return nil, errors.BadRequestStatus(err.Error())
|
||||
}
|
||||
|
||||
return &pb.OnCreateStorageBucketResponse{
|
||||
@@ -115,19 +117,19 @@ func (p *StoragePlugin) OnUpdateStorageBucket(ctx context.Context, req *pb.OnUpd
|
||||
// fetching the new copy to keep things a little less messy.
|
||||
newBucket := req.GetNewBucket()
|
||||
if newBucket == nil {
|
||||
return nil, status.Error(codes.InvalidArgument, "new bucket is required")
|
||||
return nil, errors.BadRequestStatus("new bucket is required")
|
||||
}
|
||||
if newBucket.BucketName == "" {
|
||||
return nil, status.Error(codes.InvalidArgument, "new bucketName is required")
|
||||
return nil, errors.BadRequestStatus("new bucketName is required")
|
||||
}
|
||||
currentBucket := req.GetCurrentBucket()
|
||||
if currentBucket == nil {
|
||||
return nil, status.Error(codes.InvalidArgument, "current bucket is required")
|
||||
return nil, errors.BadRequestStatus("current bucket is required")
|
||||
}
|
||||
|
||||
oldAttrs := currentBucket.GetAttributes()
|
||||
if oldAttrs == nil {
|
||||
return nil, status.Error(codes.InvalidArgument, "current bucket attributes is required")
|
||||
return nil, errors.BadRequestStatus("current bucket attributes is required")
|
||||
}
|
||||
oldStorageAttributes, err := getStorageAttributes(oldAttrs)
|
||||
if err != nil {
|
||||
@@ -135,7 +137,7 @@ func (p *StoragePlugin) OnUpdateStorageBucket(ctx context.Context, req *pb.OnUpd
|
||||
}
|
||||
newAttrs := newBucket.GetAttributes()
|
||||
if newAttrs == nil {
|
||||
return nil, status.Error(codes.InvalidArgument, "new bucket attributes is required")
|
||||
return nil, errors.BadRequestStatus("new bucket attributes is required")
|
||||
}
|
||||
newStorageAttributes, err := getStorageAttributes(newAttrs)
|
||||
if err != nil {
|
||||
@@ -147,7 +149,7 @@ func (p *StoragePlugin) OnUpdateStorageBucket(ctx context.Context, req *pb.OnUpd
|
||||
oldStorageAttributes.CredentialAttributes,
|
||||
p.testCredStateOpts...)
|
||||
if err != nil {
|
||||
return nil, status.Errorf(codes.InvalidArgument, "error loading persisted state: %s", err)
|
||||
return nil, errors.BadRequestStatus("error loading persisted state: %s", err)
|
||||
}
|
||||
|
||||
// Verify the incoming credentials are valid and return any errors to the
|
||||
@@ -170,7 +172,7 @@ func (p *StoragePlugin) OnUpdateStorageBucket(ctx context.Context, req *pb.OnUpd
|
||||
}, p.testCredStateOpts...)...,
|
||||
)
|
||||
if err != nil {
|
||||
return nil, status.Errorf(codes.InvalidArgument, "error setting up new credential persisted state: %s", err)
|
||||
return nil, errors.BadRequestStatus("error setting up new credential persisted state: %s", err)
|
||||
}
|
||||
newStorageState, err := newAwsStoragePersistedState(
|
||||
append([]awsStoragePersistedStateOption{
|
||||
@@ -178,10 +180,10 @@ func (p *StoragePlugin) OnUpdateStorageBucket(ctx context.Context, req *pb.OnUpd
|
||||
}, p.testStorageStateOpts...)...,
|
||||
)
|
||||
if err != nil {
|
||||
return nil, status.Errorf(codes.InvalidArgument, "error loading persisted state: %s", err)
|
||||
return nil, errors.BadRequestStatus("error loading persisted state: %s", err)
|
||||
}
|
||||
if err := dryRunValidation(ctx, newStorageState, newStorageAttributes, newBucket); err != nil {
|
||||
return nil, status.Errorf(codes.InvalidArgument, "failed to validate incoming credentials: %s", err)
|
||||
if st := dryRunValidation(ctx, newStorageState, newStorageAttributes, newBucket); st != nil {
|
||||
return nil, st.Err()
|
||||
}
|
||||
|
||||
// Replace the existing credential state.
|
||||
@@ -189,7 +191,7 @@ func (p *StoragePlugin) OnUpdateStorageBucket(ctx context.Context, req *pb.OnUpd
|
||||
// and deletes the credentials if we are managing them
|
||||
// (ie: if we've rotated them before).
|
||||
if err := credState.ReplaceCreds(ctx, updatedCredentials); err != nil {
|
||||
return nil, status.Errorf(codes.InvalidArgument, "error attempting to replace credentials: %s", err)
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
|
||||
@@ -199,7 +201,7 @@ func (p *StoragePlugin) OnUpdateStorageBucket(ctx context.Context, req *pb.OnUpd
|
||||
// This is not allowed.
|
||||
if newStorageAttributes.DisableCredentialRotation && newBucket.GetSecrets() == nil {
|
||||
if !credState.CredsLastRotatedTime.IsZero() {
|
||||
return nil, status.Error(codes.FailedPrecondition, "cannot disable rotation for already-rotated credentials")
|
||||
return nil, errors.BadRequestStatus("cannot disable rotation for already-rotated credentials")
|
||||
}
|
||||
}
|
||||
|
||||
@@ -207,7 +209,7 @@ func (p *StoragePlugin) OnUpdateStorageBucket(ctx context.Context, req *pb.OnUpd
|
||||
// freshly replaced credentials, we can rotate here.
|
||||
if !newStorageAttributes.DisableCredentialRotation && credState.CredsLastRotatedTime.IsZero() {
|
||||
if err := credState.RotateCreds(ctx); err != nil {
|
||||
return nil, status.Errorf(codes.InvalidArgument, "error during credential rotation: %s", err)
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -218,17 +220,17 @@ func (p *StoragePlugin) OnUpdateStorageBucket(ctx context.Context, req *pb.OnUpd
|
||||
}, p.testStorageStateOpts...)...,
|
||||
)
|
||||
if err != nil {
|
||||
return nil, status.Errorf(codes.InvalidArgument, "error loading persisted state: %s", err)
|
||||
return nil, errors.BadRequestStatus("error loading persisted state: %s", err)
|
||||
}
|
||||
|
||||
// perform dry run to ensure we can interact with the bucket as expected.
|
||||
if err := dryRunValidation(ctx, storageState, newStorageAttributes, newBucket); err != nil {
|
||||
return nil, err
|
||||
if st := dryRunValidation(ctx, storageState, newStorageAttributes, newBucket); st != nil {
|
||||
return nil, st.Err()
|
||||
}
|
||||
|
||||
persistedProto, err := storageState.toProto()
|
||||
if err != nil {
|
||||
return nil, status.Error(codes.InvalidArgument, err.Error())
|
||||
return nil, errors.BadRequestStatus(err.Error())
|
||||
}
|
||||
|
||||
return &pb.OnUpdateStorageBucketResponse{
|
||||
@@ -240,12 +242,12 @@ func (p *StoragePlugin) OnUpdateStorageBucket(ctx context.Context, req *pb.OnUpd
|
||||
func (p *StoragePlugin) OnDeleteStorageBucket(ctx context.Context, req *pb.OnDeleteStorageBucketRequest) (*pb.OnDeleteStorageBucketResponse, error) {
|
||||
bucket := req.GetBucket()
|
||||
if bucket == nil {
|
||||
return nil, status.Error(codes.InvalidArgument, "bucket is required")
|
||||
return nil, errors.BadRequestStatus("bucket is required")
|
||||
}
|
||||
|
||||
attrs := bucket.GetAttributes()
|
||||
if attrs == nil {
|
||||
return nil, status.Error(codes.InvalidArgument, "attributes is required")
|
||||
return nil, errors.BadRequestStatus("attributes is required")
|
||||
}
|
||||
|
||||
storageAttributes, err := getStorageAttributes(attrs)
|
||||
@@ -264,7 +266,7 @@ func (p *StoragePlugin) OnDeleteStorageBucket(ctx context.Context, req *pb.OnDel
|
||||
storageAttributes.CredentialAttributes,
|
||||
p.testCredStateOpts...)
|
||||
if err != nil {
|
||||
return nil, status.Errorf(codes.InvalidArgument, "error loading persisted state: %s", err)
|
||||
return nil, errors.BadRequestStatus("error loading persisted state: %s", err)
|
||||
}
|
||||
|
||||
_, err = newAwsStoragePersistedState(
|
||||
@@ -273,7 +275,7 @@ func (p *StoragePlugin) OnDeleteStorageBucket(ctx context.Context, req *pb.OnDel
|
||||
}, p.testStorageStateOpts...)...,
|
||||
)
|
||||
if err != nil {
|
||||
return nil, status.Errorf(codes.InvalidArgument, "error loading persisted state: %s", err)
|
||||
return nil, errors.BadRequestStatus("error loading persisted state: %s", err)
|
||||
}
|
||||
|
||||
// try to delete static credentials
|
||||
@@ -283,7 +285,7 @@ func (p *StoragePlugin) OnDeleteStorageBucket(ctx context.Context, req *pb.OnDel
|
||||
// credentials to ensure that it has the proper permissions to do
|
||||
// it.
|
||||
if err := credState.DeleteCreds(ctx); err != nil {
|
||||
return nil, status.Errorf(codes.Aborted, "error removing rotated credentials during storage bucket deletion: %s", err)
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -294,21 +296,21 @@ func (p *StoragePlugin) OnDeleteStorageBucket(ctx context.Context, req *pb.OnDel
|
||||
// HeadObject is called to get the metadata of an object.
|
||||
func (p *StoragePlugin) HeadObject(ctx context.Context, req *pb.HeadObjectRequest) (*pb.HeadObjectResponse, error) {
|
||||
if req.GetKey() == "" {
|
||||
return nil, status.Error(codes.InvalidArgument, "key is required")
|
||||
return nil, errors.BadRequestStatus("key is required")
|
||||
}
|
||||
|
||||
bucket := req.GetBucket()
|
||||
if bucket == nil {
|
||||
return nil, status.Error(codes.InvalidArgument, "bucket is required")
|
||||
return nil, errors.BadRequestStatus("bucket is required")
|
||||
}
|
||||
|
||||
if bucket.BucketName == "" {
|
||||
return nil, status.Error(codes.InvalidArgument, "bucketName is required")
|
||||
return nil, errors.BadRequestStatus("bucketName is required")
|
||||
}
|
||||
|
||||
attrs := bucket.GetAttributes()
|
||||
if attrs == nil {
|
||||
return nil, status.Error(codes.InvalidArgument, "attributes is required")
|
||||
return nil, errors.BadRequestStatus("attributes is required")
|
||||
}
|
||||
|
||||
storageAttributes, err := getStorageAttributes(attrs)
|
||||
@@ -328,7 +330,7 @@ func (p *StoragePlugin) HeadObject(ctx context.Context, req *pb.HeadObjectReques
|
||||
}, p.testCredStateOpts...)...,
|
||||
)
|
||||
if err != nil {
|
||||
return nil, status.Errorf(codes.InvalidArgument, "error setting up persisted state: %s", err)
|
||||
return nil, errors.BadRequestStatus("error setting up persisted state: %s", err)
|
||||
}
|
||||
|
||||
storageState, err := newAwsStoragePersistedState(
|
||||
@@ -337,7 +339,7 @@ func (p *StoragePlugin) HeadObject(ctx context.Context, req *pb.HeadObjectReques
|
||||
}, p.testStorageStateOpts...)...,
|
||||
)
|
||||
if err != nil {
|
||||
return nil, status.Errorf(codes.InvalidArgument, "error setting up persisted state: %s", err)
|
||||
return nil, errors.BadRequestStatus("error setting up persisted state: %s", err)
|
||||
}
|
||||
|
||||
opts := []s3Option{}
|
||||
@@ -346,7 +348,7 @@ func (p *StoragePlugin) HeadObject(ctx context.Context, req *pb.HeadObjectReques
|
||||
}
|
||||
s3Client, err := storageState.S3Client(ctx, opts...)
|
||||
if err != nil {
|
||||
return nil, status.Errorf(codes.InvalidArgument, "error getting S3 client: %s", err)
|
||||
return nil, errors.BadRequestStatus("error getting S3 client: %s", err)
|
||||
}
|
||||
|
||||
objectKey := path.Join(bucket.GetBucketPrefix(), req.GetKey())
|
||||
@@ -355,7 +357,7 @@ func (p *StoragePlugin) HeadObject(ctx context.Context, req *pb.HeadObjectReques
|
||||
Key: aws.String(objectKey),
|
||||
})
|
||||
if err != nil {
|
||||
return nil, parseAWSErrCode("error getting head object from s3", err)
|
||||
return nil, parseS3Error(err, "failed to head object", req).Err()
|
||||
}
|
||||
return &pb.HeadObjectResponse{
|
||||
ContentLength: resp.ContentLength,
|
||||
@@ -367,16 +369,16 @@ func (p *StoragePlugin) HeadObject(ctx context.Context, req *pb.HeadObjectReques
|
||||
func (p *StoragePlugin) ValidatePermissions(ctx context.Context, req *pb.ValidatePermissionsRequest) (*pb.ValidatePermissionsResponse, error) {
|
||||
bucket := req.GetBucket()
|
||||
if bucket == nil {
|
||||
return nil, status.Error(codes.InvalidArgument, "bucket is required")
|
||||
return nil, errors.BadRequestStatus("bucket is required")
|
||||
}
|
||||
|
||||
if bucket.BucketName == "" {
|
||||
return nil, status.Error(codes.InvalidArgument, "bucketName is required")
|
||||
return nil, errors.BadRequestStatus("bucketName is required")
|
||||
}
|
||||
|
||||
attrs := bucket.GetAttributes()
|
||||
if attrs == nil {
|
||||
return nil, status.Error(codes.InvalidArgument, "attributes is required")
|
||||
return nil, errors.BadRequestStatus("attributes is required")
|
||||
}
|
||||
|
||||
storageAttributes, err := getStorageAttributes(attrs)
|
||||
@@ -396,7 +398,7 @@ func (p *StoragePlugin) ValidatePermissions(ctx context.Context, req *pb.Validat
|
||||
}, p.testCredStateOpts...)...,
|
||||
)
|
||||
if err != nil {
|
||||
return nil, status.Errorf(codes.InvalidArgument, "error setting up persisted state: %s", err)
|
||||
return nil, errors.BadRequestStatus("error setting up persisted state: %s", err)
|
||||
}
|
||||
|
||||
storageState, err := newAwsStoragePersistedState(
|
||||
@@ -405,12 +407,12 @@ func (p *StoragePlugin) ValidatePermissions(ctx context.Context, req *pb.Validat
|
||||
}, p.testStorageStateOpts...)...,
|
||||
)
|
||||
if err != nil {
|
||||
return nil, status.Errorf(codes.InvalidArgument, "error setting up persisted state: %s", err)
|
||||
return nil, errors.BadRequestStatus("error setting up persisted state: %s", err)
|
||||
}
|
||||
|
||||
// perform dry run to ensure we can interact with the bucket as expected.
|
||||
if err := dryRunValidation(ctx, storageState, storageAttributes, bucket); err != nil {
|
||||
return nil, err
|
||||
if st := dryRunValidation(ctx, storageState, storageAttributes, bucket); st != nil {
|
||||
return nil, st.Err()
|
||||
}
|
||||
|
||||
return &pb.ValidatePermissionsResponse{}, nil
|
||||
@@ -420,21 +422,21 @@ func (p *StoragePlugin) ValidatePermissions(ctx context.Context, req *pb.Validat
|
||||
// GetObject is a blocking call until the stream has been recieved in full.
|
||||
func (p *StoragePlugin) GetObject(req *pb.GetObjectRequest, stream pb.StoragePluginService_GetObjectServer) error {
|
||||
if req.GetKey() == "" {
|
||||
return status.Error(codes.InvalidArgument, "key is required")
|
||||
return errors.BadRequestStatus("key is required")
|
||||
}
|
||||
|
||||
bucket := req.GetBucket()
|
||||
if bucket == nil {
|
||||
return status.Error(codes.InvalidArgument, "bucket is required")
|
||||
return errors.BadRequestStatus("bucket is required")
|
||||
}
|
||||
|
||||
if bucket.GetBucketName() == "" {
|
||||
return status.Error(codes.InvalidArgument, "bucketName is required")
|
||||
return errors.BadRequestStatus("bucketName is required")
|
||||
}
|
||||
|
||||
attrs := bucket.GetAttributes()
|
||||
if attrs == nil {
|
||||
return status.Error(codes.InvalidArgument, "attributes is required")
|
||||
return errors.BadRequestStatus("attributes is required")
|
||||
}
|
||||
|
||||
storageAttributes, err := getStorageAttributes(attrs)
|
||||
@@ -454,7 +456,7 @@ func (p *StoragePlugin) GetObject(req *pb.GetObjectRequest, stream pb.StoragePlu
|
||||
}, p.testCredStateOpts...)...,
|
||||
)
|
||||
if err != nil {
|
||||
return status.Errorf(codes.InvalidArgument, "error setting up persisted state: %s", err)
|
||||
return errors.BadRequestStatus("error setting up persisted state: %s", err)
|
||||
}
|
||||
|
||||
storageState, err := newAwsStoragePersistedState(
|
||||
@@ -463,7 +465,7 @@ func (p *StoragePlugin) GetObject(req *pb.GetObjectRequest, stream pb.StoragePlu
|
||||
}, p.testStorageStateOpts...)...,
|
||||
)
|
||||
if err != nil {
|
||||
return status.Errorf(codes.InvalidArgument, "error setting up persisted state: %s", err)
|
||||
return errors.BadRequestStatus("error setting up persisted state: %s", err)
|
||||
}
|
||||
|
||||
opts := []s3Option{}
|
||||
@@ -472,7 +474,7 @@ func (p *StoragePlugin) GetObject(req *pb.GetObjectRequest, stream pb.StoragePlu
|
||||
}
|
||||
s3Client, err := storageState.S3Client(stream.Context(), opts...)
|
||||
if err != nil {
|
||||
return status.Errorf(codes.InvalidArgument, "error getting S3 client: %s", err)
|
||||
return errors.BadRequestStatus("error getting S3 client: %s", err)
|
||||
}
|
||||
|
||||
objectKey := path.Join(bucket.GetBucketPrefix(), req.GetKey())
|
||||
@@ -481,7 +483,7 @@ func (p *StoragePlugin) GetObject(req *pb.GetObjectRequest, stream pb.StoragePlu
|
||||
Key: aws.String(objectKey),
|
||||
})
|
||||
if err != nil {
|
||||
return parseAWSErrCode("error getting object from s3", err)
|
||||
return parseS3Error(err, "failed to get object", req).Err()
|
||||
}
|
||||
|
||||
defer resp.Body.Close()
|
||||
@@ -494,13 +496,13 @@ func (p *StoragePlugin) GetObject(req *pb.GetObjectRequest, stream pb.StoragePlu
|
||||
buffer := make([]byte, chunkSize)
|
||||
n, err := reader.Read(buffer)
|
||||
if err != nil && err != io.EOF {
|
||||
return status.Errorf(codes.Internal, "error reading chunk from s3: %s", err)
|
||||
return errors.UnknownStatus("error reading chunk from s3: %s", err)
|
||||
}
|
||||
if n > 0 {
|
||||
if err := stream.Send(&pb.GetObjectResponse{
|
||||
FileChunk: buffer[:n],
|
||||
}); err != nil {
|
||||
return status.Errorf(codes.Internal, "error sending chunk to client: %s", err)
|
||||
return errors.UnknownStatus("error sending chunk to client: %s", err)
|
||||
}
|
||||
}
|
||||
if err == io.EOF {
|
||||
@@ -514,38 +516,38 @@ func (p *StoragePlugin) GetObject(req *pb.GetObjectRequest, stream pb.StoragePlu
|
||||
// PutObject is called when putting objects into an s3 bucket.
|
||||
func (p *StoragePlugin) PutObject(ctx context.Context, req *pb.PutObjectRequest) (*pb.PutObjectResponse, error) {
|
||||
if req == nil {
|
||||
return nil, status.Errorf(codes.InvalidArgument, "request is required")
|
||||
return nil, errors.BadRequestStatus("request is required")
|
||||
}
|
||||
if req.GetKey() == "" {
|
||||
return nil, status.Errorf(codes.InvalidArgument, "key is required")
|
||||
return nil, errors.BadRequestStatus("key is required")
|
||||
}
|
||||
bucket := req.GetBucket()
|
||||
if bucket == nil {
|
||||
return nil, status.Error(codes.InvalidArgument, "bucket is required")
|
||||
return nil, errors.BadRequestStatus("bucket is required")
|
||||
}
|
||||
if bucket.GetBucketName() == "" {
|
||||
return nil, status.Error(codes.InvalidArgument, "bucketName is required")
|
||||
return nil, errors.BadRequestStatus("bucketName is required")
|
||||
}
|
||||
attrs := bucket.GetAttributes()
|
||||
if attrs == nil {
|
||||
return nil, status.Error(codes.InvalidArgument, "attributes is required")
|
||||
return nil, errors.BadRequestStatus("attributes is required")
|
||||
}
|
||||
if req.GetPath() == "" {
|
||||
return nil, status.Errorf(codes.InvalidArgument, "path is required")
|
||||
return nil, errors.BadRequestStatus("path is required")
|
||||
}
|
||||
file, err := os.Open(req.GetPath())
|
||||
if err != nil {
|
||||
return nil, status.Errorf(codes.Internal, "failed to open file")
|
||||
return nil, errors.BadRequestStatus("failed to open file")
|
||||
}
|
||||
info, err := file.Stat()
|
||||
if err != nil {
|
||||
return nil, status.Errorf(codes.Internal, "failed to read file info")
|
||||
return nil, errors.BadRequestStatus("failed to read file info")
|
||||
}
|
||||
if info.IsDir() {
|
||||
return nil, status.Errorf(codes.InvalidArgument, "path is not a file")
|
||||
return nil, errors.BadRequestStatus("path is not a file")
|
||||
}
|
||||
if info.Size() == 0 {
|
||||
return nil, status.Errorf(codes.InvalidArgument, "file is empty")
|
||||
return nil, errors.BadRequestStatus("file is empty")
|
||||
}
|
||||
|
||||
storageAttributes, err := getStorageAttributes(attrs)
|
||||
@@ -565,7 +567,7 @@ func (p *StoragePlugin) PutObject(ctx context.Context, req *pb.PutObjectRequest)
|
||||
}, p.testCredStateOpts...)...,
|
||||
)
|
||||
if err != nil {
|
||||
return nil, status.Errorf(codes.InvalidArgument, "error setting up persisted state: %s", err)
|
||||
return nil, errors.BadRequestStatus("error setting up persisted state: %s", err)
|
||||
}
|
||||
|
||||
storageState, err := newAwsStoragePersistedState(
|
||||
@@ -574,7 +576,7 @@ func (p *StoragePlugin) PutObject(ctx context.Context, req *pb.PutObjectRequest)
|
||||
}, p.testStorageStateOpts...)...,
|
||||
)
|
||||
if err != nil {
|
||||
return nil, status.Errorf(codes.InvalidArgument, "error setting up persisted state: %s", err)
|
||||
return nil, errors.BadRequestStatus("error setting up persisted state: %s", err)
|
||||
}
|
||||
|
||||
opts := []s3Option{}
|
||||
@@ -583,18 +585,18 @@ func (p *StoragePlugin) PutObject(ctx context.Context, req *pb.PutObjectRequest)
|
||||
}
|
||||
s3Client, err := storageState.S3Client(ctx, opts...)
|
||||
if err != nil {
|
||||
return nil, status.Errorf(codes.InvalidArgument, "error getting S3 client: %s", err)
|
||||
return nil, errors.BadRequestStatus("error getting S3 client: %s", err)
|
||||
}
|
||||
|
||||
hash := sha256.New()
|
||||
if _, err := io.Copy(hash, file); err != nil {
|
||||
return nil, status.Errorf(codes.Internal, "failed to calcualte hash")
|
||||
return nil, errors.UnknownStatus("failed to calcualte hash")
|
||||
}
|
||||
checksum := base64.StdEncoding.EncodeToString(hash.Sum(nil))
|
||||
|
||||
_, err = file.Seek(0, io.SeekStart)
|
||||
if err != nil {
|
||||
return nil, status.Errorf(codes.Internal, "failed to rewind file pointer")
|
||||
return nil, errors.UnknownStatus("failed to rewind file pointer")
|
||||
}
|
||||
|
||||
objectKey := path.Join(bucket.GetBucketPrefix(), req.GetKey())
|
||||
@@ -606,17 +608,17 @@ func (p *StoragePlugin) PutObject(ctx context.Context, req *pb.PutObjectRequest)
|
||||
ChecksumSHA256: aws.String(checksum),
|
||||
})
|
||||
if err != nil {
|
||||
return nil, parseAWSErrCode("error putting object into s3", err)
|
||||
return nil, parseS3Error(err, "failed to put object", req).Err()
|
||||
}
|
||||
if resp.ChecksumSHA256 == nil {
|
||||
return nil, status.Errorf(codes.Internal, "missing checksum response from aws")
|
||||
return nil, errors.UnknownStatus("missing checksum response from aws")
|
||||
}
|
||||
if checksum != *resp.ChecksumSHA256 {
|
||||
return nil, status.Errorf(codes.Internal, "mismatched checksum")
|
||||
return nil, checksumMistmatchStatus()
|
||||
}
|
||||
decodedChecksum, err := base64.StdEncoding.DecodeString(*resp.ChecksumSHA256)
|
||||
if err != nil {
|
||||
return nil, status.Errorf(codes.Internal, "failed to decode checksum value from aws")
|
||||
return nil, errors.UnknownStatus("failed to decode checksum value from aws")
|
||||
}
|
||||
return &pb.PutObjectResponse{
|
||||
ChecksumSha_256: decodedChecksum,
|
||||
@@ -626,21 +628,21 @@ func (p *StoragePlugin) PutObject(ctx context.Context, req *pb.PutObjectRequest)
|
||||
// DeleteObjects is used to delete one or many objects from an s3 bucket.
|
||||
func (p *StoragePlugin) DeleteObjects(ctx context.Context, req *pb.DeleteObjectsRequest) (*pb.DeleteObjectsResponse, error) {
|
||||
if req.GetKeyPrefix() == "" {
|
||||
return nil, status.Error(codes.InvalidArgument, "key prefix is required")
|
||||
return nil, errors.BadRequestStatus("key prefix is required")
|
||||
}
|
||||
|
||||
bucket := req.GetBucket()
|
||||
if bucket == nil {
|
||||
return nil, status.Error(codes.InvalidArgument, "bucket is required")
|
||||
return nil, errors.BadRequestStatus("bucket is required")
|
||||
}
|
||||
|
||||
if bucket.GetBucketName() == "" {
|
||||
return nil, status.Error(codes.InvalidArgument, "bucketName is required")
|
||||
return nil, errors.BadRequestStatus("bucketName is required")
|
||||
}
|
||||
|
||||
attrs := bucket.GetAttributes()
|
||||
if attrs == nil {
|
||||
return nil, status.Error(codes.InvalidArgument, "attributes is required")
|
||||
return nil, errors.BadRequestStatus("attributes is required")
|
||||
}
|
||||
|
||||
storageAttributes, err := getStorageAttributes(attrs)
|
||||
@@ -660,7 +662,7 @@ func (p *StoragePlugin) DeleteObjects(ctx context.Context, req *pb.DeleteObjects
|
||||
}, p.testCredStateOpts...)...,
|
||||
)
|
||||
if err != nil {
|
||||
return nil, status.Errorf(codes.InvalidArgument, "error setting up persisted state: %s", err)
|
||||
return nil, errors.BadRequestStatus("error setting up persisted state: %s", err)
|
||||
}
|
||||
|
||||
storageState, err := newAwsStoragePersistedState(
|
||||
@@ -669,7 +671,7 @@ func (p *StoragePlugin) DeleteObjects(ctx context.Context, req *pb.DeleteObjects
|
||||
}, p.testStorageStateOpts...)...,
|
||||
)
|
||||
if err != nil {
|
||||
return nil, status.Errorf(codes.InvalidArgument, "error setting up persisted state: %s", err)
|
||||
return nil, errors.BadRequestStatus("error setting up persisted state: %s", err)
|
||||
}
|
||||
|
||||
opts := []s3Option{}
|
||||
@@ -678,7 +680,7 @@ func (p *StoragePlugin) DeleteObjects(ctx context.Context, req *pb.DeleteObjects
|
||||
}
|
||||
client, err := storageState.S3Client(ctx, opts...)
|
||||
if err != nil {
|
||||
return nil, status.Errorf(codes.InvalidArgument, "error getting S3 client: %s", err)
|
||||
return nil, errors.BadRequestStatus("error getting S3 client: %s", err)
|
||||
}
|
||||
|
||||
prefix := path.Join(bucket.GetBucketPrefix(), req.GetKeyPrefix())
|
||||
@@ -694,7 +696,7 @@ func (p *StoragePlugin) DeleteObjects(ctx context.Context, req *pb.DeleteObjects
|
||||
Key: aws.String(prefix),
|
||||
})
|
||||
if err != nil {
|
||||
return nil, parseAWSErrCode("error deleting S3 object", err)
|
||||
return nil, parseS3Error(err, "failed to delete object", req).Err()
|
||||
}
|
||||
return &pb.DeleteObjectsResponse{
|
||||
ObjectsDeleted: uint32(1),
|
||||
@@ -714,7 +716,7 @@ func (p *StoragePlugin) DeleteObjects(ctx context.Context, req *pb.DeleteObjects
|
||||
ContinuationToken: conToken,
|
||||
})
|
||||
if err != nil {
|
||||
return nil, parseAWSErrCode("error iterating S3 bucket contents", err)
|
||||
return nil, parseS3Error(err, "failed to list objects", req).Err()
|
||||
}
|
||||
truncated = res.IsTruncated
|
||||
conToken = res.NextContinuationToken
|
||||
@@ -736,7 +738,7 @@ func (p *StoragePlugin) DeleteObjects(ctx context.Context, req *pb.DeleteObjects
|
||||
},
|
||||
})
|
||||
if err != nil {
|
||||
return nil, parseAWSErrCode("error deleting S3 object(s)", err)
|
||||
return nil, parseS3Error(err, "failed to delete objects", req).Err()
|
||||
}
|
||||
deleted += len(res.Deleted)
|
||||
}
|
||||
@@ -752,15 +754,15 @@ func (p *StoragePlugin) DeleteObjects(ctx context.Context, req *pb.DeleteObjects
|
||||
// PutObject permissions. This function will read the same file it had uploaded to validate
|
||||
// it has GetObject permissions. This function will read the metadata of the same file it
|
||||
// had uploaded to validate it has GetObjectAttributes permissions.
|
||||
func dryRunValidation(ctx context.Context, state *awsStoragePersistedState, attrs *StorageAttributes, bucket *storagebuckets.StorageBucket) error {
|
||||
func dryRunValidation(ctx context.Context, state *awsStoragePersistedState, attrs *StorageAttributes, bucket *storagebuckets.StorageBucket) *status.Status {
|
||||
if state == nil {
|
||||
return status.Error(codes.InvalidArgument, "persisted state is required")
|
||||
return status.New(codes.InvalidArgument, "persisted state is required")
|
||||
}
|
||||
if attrs == nil {
|
||||
return status.Error(codes.InvalidArgument, "attributes is required")
|
||||
return status.New(codes.InvalidArgument, "attributes is required")
|
||||
}
|
||||
if bucket == nil {
|
||||
return status.Error(codes.InvalidArgument, "bucket is required")
|
||||
return status.New(codes.InvalidArgument, "bucket is required")
|
||||
}
|
||||
|
||||
opts := []s3Option{}
|
||||
@@ -770,61 +772,76 @@ func dryRunValidation(ctx context.Context, state *awsStoragePersistedState, attr
|
||||
|
||||
client, err := state.S3Client(ctx, opts...)
|
||||
if err != nil {
|
||||
return status.Errorf(codes.InvalidArgument, "error getting S3 client: %s", err)
|
||||
return status.New(codes.InvalidArgument, fmt.Sprintf("error getting S3 client: %s", err))
|
||||
}
|
||||
|
||||
var errs *multierror.Error
|
||||
permissions := &pb.Permissions{}
|
||||
// we track the codes returned by ParseAWSError and pick the most severe one by using the max func
|
||||
|
||||
objectKey := path.Join(bucket.GetBucketPrefix(), uuid.New().String())
|
||||
if _, err := client.PutObject(ctx, &s3.PutObjectInput{
|
||||
Bucket: aws.String(bucket.GetBucketName()),
|
||||
Key: aws.String(objectKey),
|
||||
Body: bytes.NewReader([]byte("hashicorp boundary aws plugin access test")),
|
||||
}); err != nil {
|
||||
return parseAWSErrCode("error failed to put object", err)
|
||||
var st *status.Status
|
||||
st, permissions.Write = errors.ParseAWSError(err, "failed to put object")
|
||||
errs = multierror.Append(errs, st.Err())
|
||||
} else {
|
||||
permissions.Write = &pb.Permission{State: pb.StateType_STATE_TYPE_OK, CheckedAt: timestamppb.Now()}
|
||||
}
|
||||
|
||||
if _, err := client.GetObject(ctx, &s3.GetObjectInput{
|
||||
Bucket: aws.String(bucket.GetBucketName()),
|
||||
Key: aws.String(objectKey),
|
||||
}); err != nil {
|
||||
return parseAWSErrCode("error failed to get object", err)
|
||||
var st *status.Status
|
||||
st, permissions.Read = errors.ParseAWSError(err, "failed to get object")
|
||||
errs = multierror.Append(errs, st.Err())
|
||||
} else {
|
||||
permissions.Read = &pb.Permission{State: pb.StateType_STATE_TYPE_OK, CheckedAt: timestamppb.Now()}
|
||||
}
|
||||
|
||||
if _, err := client.HeadObject(ctx, &s3.HeadObjectInput{
|
||||
Bucket: aws.String(bucket.GetBucketName()),
|
||||
Key: aws.String(objectKey),
|
||||
}); err != nil {
|
||||
return parseAWSErrCode("error failed to get head object", err)
|
||||
var st *status.Status
|
||||
st, permissions.Read = errors.ParseAWSError(err, "failed to head object")
|
||||
errs = multierror.Append(errs, st.Err())
|
||||
}
|
||||
|
||||
if res, err := client.ListObjectsV2(ctx, &s3.ListObjectsV2Input{
|
||||
Bucket: aws.String(bucket.GetBucketName()),
|
||||
Prefix: aws.String(objectKey),
|
||||
}); err != nil {
|
||||
return parseAWSErrCode("error failed to list objects", err)
|
||||
var st *status.Status
|
||||
st, permissions.Read = errors.ParseAWSError(err, "failed to list object")
|
||||
errs = multierror.Append(errs, st.Err())
|
||||
} else if res == nil || len(res.Contents) != 1 || *res.Contents[0].Key != objectKey {
|
||||
return status.Errorf(codes.InvalidArgument, "list response did not contain the expected key: %+v", res)
|
||||
return status.New(codes.Aborted, fmt.Sprintf("list response did not contain the expected key: %+v", res))
|
||||
}
|
||||
|
||||
if _, err := client.DeleteObject(ctx, &s3.DeleteObjectInput{
|
||||
Bucket: aws.String(bucket.GetBucketName()),
|
||||
Key: aws.String(objectKey),
|
||||
}); err != nil {
|
||||
return parseAWSErrCode("error failed to delete object", err)
|
||||
var st *status.Status
|
||||
st, permissions.Delete = errors.ParseAWSError(err, "failed to delete object")
|
||||
errs = multierror.Append(errs, st.Err())
|
||||
} else {
|
||||
permissions.Delete = &pb.Permission{State: pb.StateType_STATE_TYPE_OK, CheckedAt: timestamppb.Now()}
|
||||
}
|
||||
|
||||
if errs != nil {
|
||||
st := status.New(codes.FailedPrecondition, fmt.Sprintf("failed to validate provided aws credentials: %v", errs.Unwrap()))
|
||||
state := &pb.StorageBucketCredentialState{State: permissions}
|
||||
if st, err = st.WithDetails(state); err != nil {
|
||||
st = status.New(codes.Internal, err.Error())
|
||||
}
|
||||
return st
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// parseAWSErrCode detects if an aws error is a throttle error.
|
||||
// If the error is a throttle error, then an Unavailable status code
|
||||
// is returned. Otherwise the default status code is an Internal error.
|
||||
func parseAWSErrCode(msg string, err error) error {
|
||||
errCode := codes.Internal
|
||||
retryErr := retry.ThrottleErrorCode{
|
||||
Codes: retry.DefaultThrottleErrorCodes,
|
||||
}
|
||||
if retryErr.IsErrorThrottle(err).Bool() {
|
||||
errCode = codes.Unavailable
|
||||
}
|
||||
return status.Errorf(errCode, "%s: %s", msg, err)
|
||||
}
|
||||
File diff suppressed because it is too large.
Load diff
@@ -84,7 +84,6 @@ func newAwsStoragePersistedState(opts ...awsStoragePersistedStateOption) (*awsSt
|
||||
}
|
||||
|
||||
func (s *awsStoragePersistedState) toProto() (*storagebuckets.StorageBucketPersisted, error) {
|
||||
|
||||
data, err := structpb.NewStruct(s.ToMap())
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("error converting state to structpb.Struct: %w", err)
|
||||
|
||||
@@ -14,6 +14,7 @@ import (
|
||||
"github.com/aws/aws-sdk-go-v2/aws"
|
||||
"github.com/aws/aws-sdk-go-v2/service/s3"
|
||||
"github.com/aws/aws-sdk-go-v2/service/sts"
|
||||
"github.com/aws/smithy-go"
|
||||
"github.com/hashicorp/boundary-plugin-aws/internal/credential"
|
||||
"github.com/hashicorp/boundary/sdk/pbs/controller/api/resources/storagebuckets"
|
||||
pb "github.com/hashicorp/boundary/sdk/pbs/plugin"
|
||||
@@ -28,25 +29,8 @@ const (
|
||||
testGetUserErr = "test error for GetUser"
|
||||
testGetCallerIdentityErr = "test error for GetCallerIdentity"
|
||||
testDeleteAccessKeyErr = "test error for DeleteAccessKey"
|
||||
testGetObjectErr = "test error for GetObject"
|
||||
testPutObjectErr = "test error for PutObject"
|
||||
testHeadObjectErr = "test error for HeadObject"
|
||||
testListObjectV2Err = "test error for ListObjectV2"
|
||||
testDeleteObjectErr = "test error for DeleteObject"
|
||||
testDeleteObjectsErr = "test error for DeleteObjects"
|
||||
)
|
||||
|
||||
// throttleErr is a mocked error used for testing the aws s3 client
|
||||
type throttleErr struct{}
|
||||
|
||||
func (e *throttleErr) ErrorCode() string {
|
||||
return "ThrottlingException"
|
||||
}
|
||||
|
||||
func (e *throttleErr) Error() string {
|
||||
return "ThrottlingException"
|
||||
}
|
||||
|
||||
type testMockS3State struct {
|
||||
GetObjectCalled bool
|
||||
GetObjectInputParams *s3.GetObjectInput
|
||||
@@ -414,3 +398,16 @@ func validSTSMock() []credential.AwsCredentialPersistedStateOption {
|
||||
}),
|
||||
}
|
||||
}
|
||||
|
||||
// TestAwsS3Error returns an s3 api error
|
||||
func TestAwsS3Error(code, op, msg string) error {
|
||||
return &smithy.OperationError{
|
||||
ServiceID: "s3",
|
||||
OperationName: op,
|
||||
Err: &smithy.GenericAPIError{
|
||||
Code: code,
|
||||
Message: msg,
|
||||
Fault: smithy.FaultServer,
|
||||
},
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user