Skip to content

Commit

Permalink
Feature/aws kms support (#281)
Browse files Browse the repository at this point in the history 
* feat: new s3 configuration with kms support

Signed-off-by: Roko Romic <rokoromic@gmail.com>
  • Loading branch information
@rromic
rromic authored Mar 25, 2023
1 parent 4e459a9 commit 61d883d
Show file tree
Hide file tree
Showing 10 changed files with 149 additions and 53 deletions.
3 changes: 3 additions & 0 deletions charts/policy-reporter/config.yaml
View file
Original file line number Diff line number Diff line change
Expand Up @@ -161,6 +161,9 @@ s3:
region: {{ .Values.target.s3.region }}
endpoint: {{ .Values.target.s3.endpoint }}
bucket: {{ .Values.target.s3.bucket }}
bucketKeyEnabled: {{ .Values.target.s3.bucketKeyEnabled }}
sseKmsKeyId: {{ .Values.target.s3.sseKmsKeyId }}
serverSideEncryption: { .Values.target.s3.serverSideEncryption }}
pathStyle: {{ .Values.target.s3.pathStyle }}
prefix: {{ .Values.target.s3.prefix }}
minimumPriority: {{ .Values.target.s3.minimumPriority | quote }}
Expand Down
6 changes: 6 additions & 0 deletions charts/policy-reporter/values.yaml
View file
Original file line number Diff line number Diff line change
Expand Up @@ -459,6 +459,12 @@ target:
endpoint: ""
# S3 storage, bucket name
bucket: ""
# S3 storage to use an S3 Bucket Key for object encryption with SSE-KMS
bucketKeyEnabled: false
# S3 storage KMS Key ID for object encryption with SSE-KMS
sseKmsKeyId: ""
# S3 storage server-side encryption algorithm used when storing this object in Amazon S3, AES256, aws:kms
serverSideEncryption: ""
# S3 storage, force path style configuration
pathStyle: false
# name of prefix, keys will have format: s3://<bucket>/<prefix>/YYYY-MM-DD/YYYY-MM-DDTHH:mm:ss.s+01:00.json
Expand Down
33 changes: 18 additions & 15 deletions pkg/config/config.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -130,21 +130,24 @@ type Webhook struct {

// S3 configuration
type S3 struct {
Name string `mapstructure:"name"`
AccessKeyID string `mapstructure:"accessKeyID"`
SecretAccessKey string `mapstructure:"secretAccessKey"`
Region string `mapstructure:"region"`
Endpoint string `mapstructure:"endpoint"`
Prefix string `mapstructure:"prefix"`
Bucket string `mapstructure:"bucket"`
PathStyle bool `mapstructure:"pathStyle"`
SecretRef string `mapstructure:"secretRef"`
CustomFields map[string]string `mapstructure:"customFields"`
SkipExisting bool `mapstructure:"skipExistingOnStartup"`
MinimumPriority string `mapstructure:"minimumPriority"`
Filter TargetFilter `mapstructure:"filter"`
Sources []string `mapstructure:"sources"`
Channels []S3 `mapstructure:"channels"`
Name string `mapstructure:"name"`
AccessKeyID string `mapstructure:"accessKeyID"`
SecretAccessKey string `mapstructure:"secretAccessKey"`
Region string `mapstructure:"region"`
Endpoint string `mapstructure:"endpoint"`
Prefix string `mapstructure:"prefix"`
Bucket string `mapstructure:"bucket"`
BucketKeyEnabled bool `mapstructure:"bucketKeyEnabled"`
SseKmsKeyId string `mapstructure:"sseKmsKeyId"`
ServerSideEncryption string `mapstructure:"serverSideEncryption"`
PathStyle bool `mapstructure:"pathStyle"`
SecretRef string `mapstructure:"secretRef"`
CustomFields map[string]string `mapstructure:"customFields"`
SkipExisting bool `mapstructure:"skipExistingOnStartup"`
MinimumPriority string `mapstructure:"minimumPriority"`
Filter TargetFilter `mapstructure:"filter"`
Sources []string `mapstructure:"sources"`
Channels []S3 `mapstructure:"channels"`
}

// Kinesis configuration
Expand Down
29 changes: 16 additions & 13 deletions pkg/config/resolver_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -81,17 +81,20 @@ var testConfig = &config.Config{
}},
},
S3: config.S3{
AccessKeyID: "AccessKey",
SecretAccessKey: "SecretAccessKey",
Bucket: "test",
SkipExisting: true,
MinimumPriority: "debug",
Endpoint: "https://storage.yandexcloud.net",
PathStyle: true,
Region: "ru-central1",
Prefix: "prefix",
CustomFields: map[string]string{"field": "value"},
Channels: []config.S3{{}},
AccessKeyID: "AccessKey",
SecretAccessKey: "SecretAccessKey",
Bucket: "test",
BucketKeyEnabled: false,
SseKmsKeyId: "",
ServerSideEncryption: "",
SkipExisting: true,
MinimumPriority: "debug",
Endpoint: "https://storage.yandexcloud.net",
PathStyle: true,
Region: "ru-central1",
Prefix: "prefix",
CustomFields: map[string]string{"field": "value"},
Channels: []config.S3{{}},
},
Kinesis: config.Kinesis{
AccessKeyID: "AccessKey",
Expand All @@ -105,7 +108,7 @@ var testConfig = &config.Config{
Channels: []config.Kinesis{{}},
},
GCS: config.GCS{
Credentials: "Credentials",
Credentials: `{"token": "token", "type": "authorized_user"}`,
Bucket: "test",
SkipExisting: true,
MinimumPriority: "debug",
Expand All @@ -132,7 +135,7 @@ func Test_ResolveTargets(t *testing.T) {
resolver := config.NewResolver(testConfig, &rest.Config{})

if count := len(resolver.TargetClients()); count != 19 {
t.Errorf("Expected 17 Clients, got %d", count)
t.Errorf("Expected 19 Clients, got %d", count)
}
}

Expand Down
38 changes: 27 additions & 11 deletions pkg/config/target_factory.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -236,7 +236,7 @@ func (f *TargetFactory) KinesisClients(config Kinesis) []target.Client {
return clients
}

// S3Clients resolver method
// GCSClients resolver method
func (f *TargetFactory) GCSClients(config GCS) []target.Client {
clients := make([]target.Client, 0)
if config.Name == "" {
Expand Down Expand Up @@ -283,7 +283,7 @@ func (f *TargetFactory) createSlackClient(config Slack, parent Slack) target.Cli
Name: config.Name,
SkipExistingOnStartup: config.SkipExisting,
ResultFilter: createResultFilter(config.Filter, config.MinimumPriority, config.Sources),
ReportFilter: createReprotFilter(config.Filter),
ReportFilter: createReportFilter(config.Filter),
},
Webhook: config.Webhook,
CustomFields: config.CustomFields,
Expand Down Expand Up @@ -329,7 +329,7 @@ func (f *TargetFactory) createLokiClient(config Loki, parent Loki) target.Client
Name: config.Name,
SkipExistingOnStartup: config.SkipExisting,
ResultFilter: createResultFilter(config.Filter, config.MinimumPriority, config.Sources),
ReportFilter: createReprotFilter(config.Filter),
ReportFilter: createReportFilter(config.Filter),
},
Host: config.Host + config.Path,
CustomLabels: config.CustomLabels,
Expand Down Expand Up @@ -391,7 +391,7 @@ func (f *TargetFactory) createElasticsearchClient(config Elasticsearch, parent E
Name: config.Name,
SkipExistingOnStartup: config.SkipExisting,
ResultFilter: createResultFilter(config.Filter, config.MinimumPriority, config.Sources),
ReportFilter: createReprotFilter(config.Filter),
ReportFilter: createReportFilter(config.Filter),
},
Host: config.Host,
Username: config.Username,
Expand Down Expand Up @@ -427,7 +427,7 @@ func (f *TargetFactory) createDiscordClient(config Discord, parent Discord) targ
Name: config.Name,
SkipExistingOnStartup: config.SkipExisting,
ResultFilter: createResultFilter(config.Filter, config.MinimumPriority, config.Sources),
ReportFilter: createReprotFilter(config.Filter),
ReportFilter: createReportFilter(config.Filter),
},
Webhook: config.Webhook,
CustomFields: config.CustomFields,
Expand Down Expand Up @@ -471,7 +471,7 @@ func (f *TargetFactory) createTeamsClient(config Teams, parent Teams) target.Cli
Name: config.Name,
SkipExistingOnStartup: config.SkipExisting,
ResultFilter: createResultFilter(config.Filter, config.MinimumPriority, config.Sources),
ReportFilter: createReprotFilter(config.Filter),
ReportFilter: createReportFilter(config.Filter),
},
Webhook: config.Webhook,
CustomFields: config.CustomFields,
Expand Down Expand Up @@ -523,7 +523,7 @@ func (f *TargetFactory) createWebhookClient(config Webhook, parent Webhook) targ
Name: config.Name,
SkipExistingOnStartup: config.SkipExisting,
ResultFilter: createResultFilter(config.Filter, config.MinimumPriority, config.Sources),
ReportFilter: createReprotFilter(config.Filter),
ReportFilter: createReportFilter(config.Filter),
},
Host: config.Host,
Headers: config.Headers,
Expand Down Expand Up @@ -587,13 +587,26 @@ func (f *TargetFactory) createS3Client(config S3, parent S3) target.Client {
config.SkipExisting = parent.SkipExisting
}

if !config.BucketKeyEnabled {
config.BucketKeyEnabled = parent.BucketKeyEnabled
}

if config.SseKmsKeyId == "" {
config.SseKmsKeyId = parent.SseKmsKeyId
}

if config.ServerSideEncryption == "" {
config.ServerSideEncryption = parent.ServerSideEncryption
}

s3Client := helper.NewS3Client(
config.AccessKeyID,
config.SecretAccessKey,
config.Region,
config.Endpoint,
config.Bucket,
config.PathStyle,
helper.WithKMS(&config.BucketKeyEnabled, &config.SseKmsKeyId, &config.ServerSideEncryption),
)

sugar.Infof("%s configured", config.Name)
Expand All @@ -603,7 +616,7 @@ func (f *TargetFactory) createS3Client(config S3, parent S3) target.Client {
Name: config.Name,
SkipExistingOnStartup: config.SkipExisting,
ResultFilter: createResultFilter(config.Filter, config.MinimumPriority, config.Sources),
ReportFilter: createReprotFilter(config.Filter),
ReportFilter: createReportFilter(config.Filter),
},
S3: s3Client,
CustomFields: config.CustomFields,
Expand Down Expand Up @@ -675,7 +688,7 @@ func (f *TargetFactory) createKinesisClient(config Kinesis, parent Kinesis) targ
Name: config.Name,
SkipExistingOnStartup: config.SkipExisting,
ResultFilter: createResultFilter(config.Filter, config.MinimumPriority, config.Sources),
ReportFilter: createReprotFilter(config.Filter),
ReportFilter: createReportFilter(config.Filter),
},
CustomFields: config.CustomFields,
Kinesis: kinesisClient,
Expand Down Expand Up @@ -732,7 +745,7 @@ func (f *TargetFactory) createGCSClient(config GCS, parent GCS) target.Client {
Name: config.Name,
SkipExistingOnStartup: config.SkipExisting,
ResultFilter: createResultFilter(config.Filter, config.MinimumPriority, config.Sources),
ReportFilter: createReprotFilter(config.Filter),
ReportFilter: createReportFilter(config.Filter),
},
Client: gcsClient,
CustomFields: config.CustomFields,
Expand Down Expand Up @@ -786,6 +799,9 @@ func (f *TargetFactory) mapSecretValues(config any, ref string) {
if values.SecretAccessKey != "" {
c.SecretAccessKey = values.SecretAccessKey
}
if values.KmsKeyId != "" {
c.SseKmsKeyId = values.KmsKeyId
}

case *Kinesis:
if values.AccessKeyID != "" {
Expand Down Expand Up @@ -824,7 +840,7 @@ func createResultFilter(filter TargetFilter, minimumPriority string, sources []s
)
}

func createReprotFilter(filter TargetFilter) *report.ReportFilter {
func createReportFilter(filter TargetFilter) *report.ReportFilter {
return target.NewReportFilter(
ToRuleSet(filter.ReportLabels),
)
Expand Down
32 changes: 30 additions & 2 deletions pkg/config/target_factory_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -29,8 +29,9 @@ func newFakeClient() v1.SecretInterface {
"webhook": []byte("http://localhost:9200/webhook"),
"accessKeyID": []byte("accessKeyID"),
"secretAccessKey": []byte("secretAccessKey"),
"kmsKeyId": []byte("kmsKeyId"),
"token": []byte("token"),
"credentials": []byte("credentials"),
"credentials": []byte(`{"token": "token", "type": "authorized_user"}`),
},
}).CoreV1().Secrets("default")
}
Expand Down Expand Up @@ -154,6 +155,26 @@ func Test_ResolveTargetWithoutHost(t *testing.T) {
t.Error("Expected Client to be nil if no bucket is configured")
}
})
t.Run("S3.SSE-S3", func(t *testing.T) {
if len(factory.S3Clients(config.S3{Endpoint: "https://storage.yandexcloud.net", AccessKeyID: "access", SecretAccessKey: "secret", Region: "ru-central1", ServerSideEncryption: "AES256"})) != 0 {
t.Error("Expected Client to be nil if server side encryption is not configured")
}
})
t.Run("S3.SSE-KMS", func(t *testing.T) {
if len(factory.S3Clients(config.S3{Endpoint: "https://storage.yandexcloud.net", AccessKeyID: "access", SecretAccessKey: "secret", Region: "ru-central1", ServerSideEncryption: "aws:kms"})) != 0 {
t.Error("Expected Client to be nil if server side encryption is not configured")
}
})
t.Run("S3.SSE-KMS-S3-KEY", func(t *testing.T) {
if len(factory.S3Clients(config.S3{Endpoint: "https://storage.yandexcloud.net", AccessKeyID: "access", SecretAccessKey: "secret", Region: "ru-central1", BucketKeyEnabled: true, ServerSideEncryption: "aws:kms"})) != 0 {
t.Error("Expected Client to be nil if server side encryption is not configured")
}
})
t.Run("S3.SSE-KMS-KEY-ID", func(t *testing.T) {
if len(factory.S3Clients(config.S3{Endpoint: "https://storage.yandexcloud.net", AccessKeyID: "access", SecretAccessKey: "secret", Region: "ru-central1", ServerSideEncryption: "aws:kms", SseKmsKeyId: "SseKmsKeyId"})) != 0 {
t.Error("Expected Client to be nil if server side encryption is not configured")
}
})
t.Run("Kinesis.Endoint", func(t *testing.T) {
if len(factory.KinesisClients(config.Kinesis{})) != 0 {
t.Error("Expected Client to be nil if no endpoint is configured")
Expand All @@ -176,7 +197,7 @@ func Test_ResolveTargetWithoutHost(t *testing.T) {
})
t.Run("Kinesis.StreamName", func(t *testing.T) {
if len(factory.KinesisClients(config.Kinesis{Endpoint: "https://yds.serverless.yandexcloud.net", AccessKeyID: "access", SecretAccessKey: "secret", Region: "ru-central1"})) != 0 {
t.Error("Expected Client to be nil if no bucket is configured")
t.Error("Expected Client to be nil if no stream name is configured")
}
})
t.Run("GCS.Bucket", func(t *testing.T) {
Expand Down Expand Up @@ -293,6 +314,13 @@ func Test_GetValuesFromSecret(t *testing.T) {
}
})

t.Run("Get S3 values from Secret with KMS", func(t *testing.T) {
clients := factory.S3Clients(config.S3{SecretRef: secretName, Endpoint: "endoint", Bucket: "bucket", Region: "region", BucketKeyEnabled: true, ServerSideEncryption: "aws:kms"})
if len(clients) != 1 {
t.Error("Expected one client created")
}
})

t.Run("Get Kinesis values from Secret", func(t *testing.T) {
clients := factory.KinesisClients(config.Kinesis{SecretRef: secretName, Endpoint: "endpoint", StreamName: "stream", Region: "region"})
if len(clients) != 1 {
Expand Down
45 changes: 36 additions & 9 deletions pkg/helper/aws.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -18,21 +18,42 @@ type AWSClient interface {
}

type s3Client struct {
bucket string
uploader *s3manager.Uploader
bucket string
uploader *s3manager.Uploader
bucketKeyEnabled *bool
sseKmsKeyId *string
serverSideEncryption *string
}

type Options func(s *s3Client)

func WithKMS(bucketKeyEnabled *bool, sseKmsKeyId, serverSideEncryption *string) Options {
return func(s *s3Client) {
s.bucketKeyEnabled = bucketKeyEnabled
if *sseKmsKeyId != "" {
s.sseKmsKeyId = sseKmsKeyId
}

if *serverSideEncryption != "" {
s.serverSideEncryption = serverSideEncryption
}
}
}

func (s *s3Client) Upload(body *bytes.Buffer, key string) error {
_, err := s.uploader.Upload(&s3manager.UploadInput{
Bucket: aws.String(s.bucket),
Key: aws.String(key),
Body: body,
Bucket: aws.String(s.bucket),
Key: aws.String(key),
Body: body,
BucketKeyEnabled: s.bucketKeyEnabled,
SSEKMSKeyId: s.sseKmsKeyId,
ServerSideEncryption: s.serverSideEncryption,
})
return err
}

// NewS3Client creates a new S3.client to send Results to S3
func NewS3Client(accessKeyID, secretAccessKey, region, endpoint, bucket string, pathStyle bool) AWSClient {
func NewS3Client(accessKeyID, secretAccessKey, region, endpoint, bucket string, pathStyle bool, opts ...Options) AWSClient {
config := &aws.Config{
Region: aws.String(region),
Endpoint: aws.String(endpoint),
Expand All @@ -48,10 +69,16 @@ func NewS3Client(accessKeyID, secretAccessKey, region, endpoint, bucket string,
return nil
}

return &s3Client{
bucket,
s3manager.NewUploader(sess),
s3Client := &s3Client{
bucket: bucket,
uploader: s3manager.NewUploader(sess),
}

for _, opt := range opts {
opt(s3Client)
}

return s3Client
}

type kinesisClient struct {
Expand Down
2 changes: 1 addition & 1 deletion pkg/helper/gcp.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -34,7 +34,7 @@ func (c *gcsClient) Upload(body *bytes.Buffer, key string) error {
return writer.Close()
}

// NewS3Client creates a new S3.client to send Results to S3
// NewGCSClient creates a new GCS.client to send Results to GCS Bucket
func NewGCSClient(ctx context.Context, credentials, bucket string) GCPClient {
cred, err := google.CredentialsFromJSON(ctx, []byte(credentials), storage.ScopeReadWrite)
if err != nil {
Expand Down
Loading

0 comments on commit 61d883d

Please sign in to comment.