BP: Base64 decode / encode processors

pkg/plugin/processor/builtin/base64/decode.go

@@ -0,0 +1,121 @@
+// Copyright © 2024 Meroxa, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//go:generate paramgen -output=decode_paramgen.go decodeConfig
+
+package base64
+
+import (
+	"context"
+	"encoding/base64"
+
+	"github.com/conduitio/conduit-commons/opencdc"
+	sdk "github.com/conduitio/conduit-processor-sdk"
+	"github.com/conduitio/conduit/pkg/foundation/cerrors"
+	"github.com/conduitio/conduit/pkg/foundation/log"
+)
+
+type decodeProcessor struct {
+	sdk.UnimplementedProcessor
+
+	config            decodeConfig
+	referenceResolver sdk.ReferenceResolver
+}
+
+type decodeConfig struct {
+	// Field is the target field, as it would be addressed in a Go template (e.g. `.Payload.After.foo`).
+	// Note that it is not allowed to base64 decode the `.Position` field.
+	Field string `json:"field" validate:"required,exclusion=.Position"`
+}
+
+func NewDecodeProcessor(log.CtxLogger) sdk.Processor {
+	return &decodeProcessor{}
+}
+
+func (p *decodeProcessor) Specification() (sdk.Specification, error) {
+	return sdk.Specification{
+		Name:    "base64.decode",
+		Summary: "Decode a field to base64.",
+		Description: `The processor will decode the value of the target field from base64 and store the
+result in the target field. It is not allowed to decode the ` + "`.Position`" + ` field.`,
+		Version:    "v0.1.0",
+		Author:     "Meroxa, Inc.",
+		Parameters: decodeConfig{}.Parameters(),
+	}, nil
+}
+
+func (p *decodeProcessor) Configure(ctx context.Context, m map[string]string) error {
+	err := sdk.ParseConfig(ctx, m, &p.config, p.config.Parameters())
+	if err != nil {
+		return cerrors.Errorf("failed to parse configuration: %w", err)
+	}
+
+	resolver, err := sdk.NewReferenceResolver(p.config.Field)
+	if err != nil {
+		return cerrors.Errorf(`failed to parse the "field" parameter: %w`, err)
+	}
+	p.referenceResolver = resolver
+
+	return nil
+}
+
+func (p *decodeProcessor) Open(context.Context) error { return nil }
+
+func (p *decodeProcessor) Process(ctx context.Context, records []opencdc.Record) []sdk.ProcessedRecord {
+	out := make([]sdk.ProcessedRecord, 0, len(records))
+	for _, rec := range records {
+		rec, err := p.base64Decode(rec)
+		if err != nil {
+			return append(out, sdk.ErrorRecord{Error: err})
+		}
+		out = append(out, rec)
+	}
+	return out
+}
+
+func (p *decodeProcessor) base64Decode(rec opencdc.Record) (sdk.ProcessedRecord, error) {
+	ref, err := p.referenceResolver.Resolve(&rec)
+	if err != nil {
+		return nil, cerrors.Errorf("failed to resolve the field: %w", err)
+	}
+
+	var raw []byte
+	switch val := ref.Get().(type) {
+	case []byte:
+		raw = val
+	case opencdc.RawData:
+		raw = val
+	case string:
+		raw = []byte(val)
+	case nil:
+		return sdk.SingleRecord(rec), nil
+	default:
+		return nil, cerrors.Errorf("unexpected data type %T", val)
+	}
+
+	decoded := make([]byte, base64.StdEncoding.DecodedLen(len(raw)))
+	n, err := base64.StdEncoding.Decode(decoded, raw)
+	if err != nil {
+		return nil, cerrors.Errorf("failed to decode the value: %w", err)
+	}
+
+	err = ref.Set(string(decoded[:n]))
+	if err != nil {
+		return nil, cerrors.Errorf("failed to set the decoded value into the record: %w", err)
+	}
+
+	return sdk.SingleRecord(rec), nil
+}
+
+func (p *decodeProcessor) Teardown(context.Context) error { return nil }

pkg/plugin/processor/builtin/base64/decode_test.go

@@ -0,0 +1,142 @@
+// Copyright © 2024 Meroxa, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package base64
+
+import (
+	"context"
+	"testing"
+
+	"github.com/conduitio/conduit-commons/opencdc"
+	sdk "github.com/conduitio/conduit-processor-sdk"
+	"github.com/conduitio/conduit/pkg/foundation/cerrors"
+	"github.com/conduitio/conduit/pkg/foundation/log"
+	"github.com/google/go-cmp/cmp"
+	"github.com/matryer/is"
+)
+
+func TestDecodeProcessor_Success(t *testing.T) {
+	ctx := context.Background()
+
+	testCases := []struct {
+		name   string
+		field  string
+		record opencdc.Record
+		want   sdk.SingleRecord
+	}{{
+		name:  "decode raw data",
+		field: ".Key",
+		record: opencdc.Record{
+			Key: opencdc.RawData("Zm9v"),
+		},
+		want: sdk.SingleRecord{
+			Key: opencdc.RawData("foo"),
+		},
+	}, {
+		name:  "decode string",
+		field: ".Key.foo",
+		record: opencdc.Record{
+			Key: opencdc.StructuredData{
+				"foo": "YmFy",
+			},
+		},
+		want: sdk.SingleRecord{
+			Key: opencdc.StructuredData{
+				"foo": "bar",
+			},
+		},
+	}}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			is := is.New(t)
+			proc := NewDecodeProcessor(log.Nop())
+			err := proc.Configure(ctx, map[string]string{"field": tc.field})
+			is.NoErr(err)
+			got := proc.Process(ctx, []opencdc.Record{tc.record})
+			is.Equal(1, len(got))
+			is.Equal("", cmp.Diff(tc.want, got[0], cmpProcessedRecordOpts...))
+		})
+	}
+}
+
+func TestDecodeProcessor_Fail(t *testing.T) {
+	ctx := context.Background()
+
+	testCases := []struct {
+		name    string
+		field   string
+		record  opencdc.Record
+		wantErr error
+	}{{
+		name:  "decode structured data",
+		field: ".Key",
+		record: opencdc.Record{
+			Key: opencdc.StructuredData{
+				"foo": "bar",
+			},
+		},
+		wantErr: cerrors.New("unexpected data type opencdc.StructuredData"),
+	}, {
+		name:  "decode map",
+		field: ".Key.foo",
+		record: opencdc.Record{
+			Key: opencdc.StructuredData{
+				"foo": map[string]any{
+					"bar": "baz",
+				},
+			},
+		},
+		wantErr: cerrors.New("unexpected data type map[string]interface {}"),
+	}, {
+		name:  "decode int",
+		field: ".Key.foo",
+		record: opencdc.Record{
+			Key: opencdc.StructuredData{
+				"foo": 1,
+			},
+		},
+		wantErr: cerrors.New("unexpected data type int"),
+	}, {
+		name:  "decode float",
+		field: ".Key.foo",
+		record: opencdc.Record{
+			Key: opencdc.StructuredData{
+				"foo": 1.1,
+			},
+		},
+		wantErr: cerrors.New("unexpected data type float64"),
+	}, {
+		name:  "invalid base64 string",
+		field: ".Key.foo",
+		record: opencdc.Record{
+			Key: opencdc.StructuredData{
+				"foo": "bar",
+			},
+		},
+		wantErr: cerrors.New("failed to decode the value: illegal base64 data at input byte 0"),
+	}}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			is := is.New(t)
+			proc := NewDecodeProcessor(log.Nop())
+			err := proc.Configure(ctx, map[string]string{"field": tc.field})
+			is.NoErr(err)
+			got := proc.Process(ctx, []opencdc.Record{tc.record})
+			is.Equal(1, len(got))
+			is.Equal("", cmp.Diff(sdk.ErrorRecord{Error: tc.wantErr}, got[0], cmpProcessedRecordOpts...))
+		})
+	}
+}