masking clean commit

config/test/bloblang/masking.yaml

@@ -0,0 +1,55 @@
+pipeline:
+  processors:
+    - mapping: |
+        root = this
+        root = match env("MASK") {
+          "FIXED_5"      => root.value.mask(5)
+          "LEFT_5"       => root.value.mask(5, "left")
+          "RIGHT_4"      => root.value.mask(4,"right")
+          "RIGHT_4_HASH" => root.value.mask(4,"right", "#")
+          _              => root.value.mask()
+        }
+
+tests:
+    - name: Fixed mask 5
+      target_processors: /pipeline/processors
+      environment: 
+        MASK: FIXED_5
+      input_batch:
+          - content: '{"value": "this is a happy cat meow"}'
+      output_batches:
+          - - content_equals: '*****'
+
+    - name: Left mask 5
+      target_processors: /pipeline/processors
+      environment: 
+        MASK: LEFT_5
+      input_batch:
+          - content: '{"value": "this is a happy cat meow"}'
+      output_batches:
+          - - content_equals: 'this *******************'
+
+    - name: Right mask 4
+      target_processors: /pipeline/processors
+      environment: 
+        MASK: RIGHT_4
+      input_batch:
+          - content: '{"value": "this is a happy cat meow"}'
+      output_batches:
+          - - content_equals: '********************meow'
+
+    - name: Right mask 4 with hash
+      target_processors: /pipeline/processors
+      environment: 
+        MASK: RIGHT_4_HASH
+      input_batch:
+          - content: '{"value": "this is a happy cat meow"}'
+      output_batches:
+          - - content_equals: '####################meow'
+
+    - name: default mask
+      target_processors: /pipeline/processors
+      input_batch:
+          - content: '{"value": "this is a happy cat meow"}'
+      output_batches:
+          - - content_equals: '************************'

internal/impl/pure/bloblang_mask.go

@@ -0,0 +1,79 @@
+package pure
+
+import (
+	"errors"
+	"strings"
+
+	"github.com/redpanda-data/benthos/v4/internal/bloblang/query"
+	"github.com/redpanda-data/benthos/v4/public/bloblang"
+)
+
+func init() {
+	if err := bloblang.RegisterMethodV2("mask",
+		bloblang.NewPluginSpec().
+			Category(query.MethodCategoryParsing).
+			Description(`Masks a string using the given character, leaving X number of characters unmasked and returns a string.`).
+			Param(bloblang.NewInt64Param("count").Description("the number of characters that will not be masked on the left or right hand side, in the case of a all mask, it is the number of mask characters to return giving a fixed length string, default is 0 which will return all characters masked.").Optional().Default(0)).
+			Param(bloblang.NewStringParam("direction").Description("the direction to mask, left, right or all, default is all").Optional().Default("all")).
+			Param(bloblang.NewStringParam("char").Description("the character used for masking, default is *").Optional().Default("*")).
+			Example("Mask the first 13 characters", `root.body_mask = this.body.mask(4, "right")`,
+				[2]string{
+					`{"body":"the cat goes meow"}`,
+					`{"body_mask":"*************meow"}`,
+				},
+			),
+		func(args *bloblang.ParsedParams) (bloblang.Method, error) {
+			countPtr, err := args.GetOptionalInt64("count")
+			if err != nil {
+				return nil, errors.New("failed to get count as int: " + err.Error())
+			}
+			count := int(*countPtr)
+
+			char, err := args.GetString("char")
+			if err != nil {
+				return nil, errors.New("failed to get masking char as string: " + err.Error())
+			}
+
+			direction, err := args.GetString("direction")
+			if err != nil {
+				return nil, errors.New("failed to get direction as string: " + err.Error())
+			}
+
+			direction = strings.ToLower(direction)
+			if direction != "left" && direction != "right" && direction != "all" {
+				return nil, errors.New("direction must be one of left, right or all")
+			}
+
+			return bloblang.StringMethod(func(s string) (any, error) {
+				return maskString(s, char, direction, count), nil
+			}), nil
+		}); err != nil {
+		panic(err)
+	}
+}
+
+// maskString masks the string based on the given parameters.
+func maskString(s string, char string, direction string, count int) string {
+	sLength := len(s)
+	if count == 0 {
+		count = sLength
+	}
+	fixedMask := count
+	if count > sLength {
+		count = sLength
+	}
+
+	switch direction {
+	case "left":
+		unmasked := s[:count]
+		masked := strings.Repeat(char, sLength-count)
+		return unmasked + masked
+	case "right":
+		unmasked := s[sLength-count:]
+		masked := strings.Repeat(char, sLength-count)
+		return masked + unmasked
+	default:
+		masked := strings.Repeat(char, fixedMask)
+		return masked
+	}
+}

internal/impl/pure/bloblang_mask_test.go

@@ -0,0 +1,100 @@
+package pure
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/redpanda-data/benthos/v4/internal/bloblang/query"
+	"github.com/redpanda-data/benthos/v4/internal/value"
+)
+
+func TestMask(t *testing.T) {
+	testCases := []struct {
+		name   string
+		method string
+		target any
+		args   []any
+		exp    any
+		err    string
+	}{
+		{
+			name:   "default fixed string",
+			method: "mask",
+			target: "this is a test",
+			args:   []any{},
+			exp:    "**************",
+			err:    "",
+		},
+		{
+			name:   "default fixed string length 5",
+			method: "mask",
+			target: "this is a test",
+			args:   []any{int64(5)},
+			exp:    "*****",
+			err:    "",
+		},
+		{
+			name:   "Mask left leave left hand four chars unmasked",
+			method: "mask",
+			target: "this is a test",
+			args:   []any{int64(4), "left"},
+			exp:    "this**********",
+			err:    "",
+		},
+		{
+			name:   "Mask right leave right hand four chars unmasked",
+			method: "mask",
+			target: "this is a test",
+			args:   []any{int64(6), "right"},
+			exp:    "********a test",
+			err:    "",
+		},
+		{
+			name:   "Mask right leave right hand four chars unmasked, mask with '%' char",
+			method: "mask",
+			target: "this is a test",
+			args:   []any{int64(6), "right", "%"},
+			exp:    "%%%%%%%%a test",
+			err:    "",
+		},
+		{
+			name:   "invalid direction",
+			method: "mask",
+			target: "this is a test",
+			args:   []any{int64(5), "Fred", "*"},
+			exp:    nil,
+			err:    "direction must be one of left, right or all",
+		},
+	}
+
+	for _, test := range testCases {
+		test := test
+		t.Run(test.name, func(t *testing.T) {
+			targetClone := value.IClone(test.target)
+			argsClone := value.IClone(test.args).([]any)
+
+			fn, err := query.InitMethodHelper(test.method, query.NewLiteralFunction("", targetClone), argsClone...)
+
+			if test.err != "" {
+				require.Error(t, err)
+				assert.EqualError(t, err, test.err)
+				return
+			}
+
+			require.NoError(t, err)
+
+			res, err := fn.Exec(query.FunctionContext{
+				Maps:     map[string]query.Function{},
+				Index:    0,
+				MsgBatch: nil,
+			})
+			require.NoError(t, err)
+
+			assert.Equal(t, test.exp, res)
+			assert.Equal(t, test.target, targetClone)
+			assert.Equal(t, test.args, argsClone)
+		})
+	}
+}