Merge pull request #267 from secrethub/feature/mask-rewrite

Masking rewrite

Author: jpcoenen

internals/cli/masker/masker.go

@@ -0,0 +1,139 @@
+package masker
+
+import (
+	"io"
+	"time"
+)
+
+// Masker handles the creation and synchronization of streams that have all their writes scanned for secrets and
+// have them redacted if any matches are found. Masking of secrets is a best effort attempt. Output on all streams is
+// buffered to increase the chance of finding secrets if they are spread across multiple writes, but it cannot be
+// guaranteed that these secrets are masked. The duration bytes spend in the buffer is constant.
+//
+// Usage:
+// 1. Create a new Masker using New()
+// 2. Add one more streams using AddStream()
+// 3. Run the Start() method in a separate goroutine
+// 4. After everything has been written to the io.Writers, flush all buffers using Stop()
+type Masker struct {
+	bufferDelay time.Duration
+	sequences   [][]byte
+	frames      chan frame
+	stopChan    chan struct{}
+	err         error
+}
+
+// Options for configuring masking behavior.
+type Options struct {
+	// DisableBuffer completely disables the buffering of the masker. This increases output responsiveness
+	// but also increases the chance of a secret not being masked.
+	DisableBuffer bool
+
+	// BufferDelay is the constant duration for which input to a stream is buffered. A higher value increases
+	// the chance of secrets being detected for masking. Especially when writes have a variable delay between them,
+	// for example in the case data arrives over an unstable network connection.
+	// Defaults to 50ms if not set.
+	BufferDelay time.Duration
+
+	// FrameBufferLength is the number of frames that can be in the buffer simultaneously.
+	// If the frame buffer is full, writing to a stream blocks until there is space.
+	FrameBufferLength int
+}
+
+// New creates a new Masker that scans all streams for the given sequences and masks them.
+func New(sequences [][]byte, opts *Options) *Masker {
+	masker := &Masker{
+		bufferDelay: time.Millisecond * 50,
+		sequences:   sequences,
+		stopChan:    make(chan struct{}),
+	}
+	frameChanlength := 1024
+	if opts != nil {
+		if opts.DisableBuffer {
+			masker.bufferDelay = 0
+			frameChanlength = 0
+		} else {
+			if opts.BufferDelay > 0 {
+				masker.bufferDelay = opts.BufferDelay
+			}
+			if opts.FrameBufferLength > 0 {
+				frameChanlength = opts.FrameBufferLength
+			}
+		}
+
+	}
+	masker.frames = make(chan frame, frameChanlength)
+
+	return masker
+}
+
+// AddStream takes in an io.Writer to mask secrets on and returns an io.Writer that has secrets on its output masked.
+func (m *Masker) AddStream(w io.Writer) io.Writer {
+	s := stream{
+		dest:          w,
+		registerFrame: m.registerFrame,
+		matches:       matches{},
+		matcher:       newMatcher(m.sequences),
+	}
+	return &s
+}
+
+// Start continuously flushes the input buffer for each frame for which the buffer delay has passed.
+// This method blocks until Stop() is called.
+func (m *Masker) Start() {
+	for {
+		select {
+		case <-m.stopChan:
+			for t := range m.frames {
+				err := t.stream.flush(t.length)
+				if err != nil {
+					m.handleErr(err)
+				}
+			}
+			m.stopChan <- struct{}{}
+			return
+		case trigger := <-m.frames:
+			<-trigger.timer.C
+
+			err := trigger.stream.flush(trigger.length)
+			if err != nil {
+				m.handleErr(err)
+			}
+		}
+	}
+}
+
+// Stop all pending frames and wait for this to complete.
+// This should be run after all input has been written to the io.Writers of the streams.
+// Calling Write() on a stream after calling Stop() will lead to a panic.
+func (m *Masker) Stop() error {
+	m.stopChan <- struct{}{}
+	close(m.frames)
+	<-m.stopChan
+
+	return m.err
+}
+
+// registerFrame adds a new frame to the frames channel with a timeout of bufferDelay plus the given offset.
+// After this timer has passed, the frame will be flushed to the output.
+func (m *Masker) registerFrame(s *stream, offset time.Duration, l int) {
+	m.frames <- frame{
+		length: l,
+		stream: s,
+		timer:  time.NewTimer(offset + m.bufferDelay),
+	}
+}
+
+func (m *Masker) handleErr(err error) {
+	if err != nil && m.err == nil {
+		m.err = err
+	}
+}
+
+// frame represent a set of bytes in the buffer of a stream that were written in a single call of Write().
+// The bytes are written to the destination after the timer has expired.
+type frame struct {
+	length int
+	stream *stream
+	timer  *time.Timer
+}

internals/cli/masker/masker_diagram.drawio

@@ -0,0 +1 @@
+<mxfile host="Electron" modified="2020-04-01T14:17:44.267Z" agent="Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) draw.io/12.6.5 Chrome/80.0.3987.141 Electron/8.1.1 Safari/537.36" etag="-4ZPzU0b57b_415PivDi" version="12.6.5" type="device"><diagram id="oNAHip3EKjUoHj94W4kH" name="Page-1">7Vttc6M2EP41nrYf7EESAvtjnNebadrOpZ27fFSMjLlg5Ao5tu/XV4BkXkRyxDaY3NRfghZJwO6zj3ZXygBdLre3nKwW98yj4QBa3naArgYQQsuB8k8i2WUSAPA4k/g88JQsFzwE36kSWkq6DjwalzoKxkIRrMrCGYsiOhMlGeGcbcrd5iwsP3VFfGoIHmYkNKVfAk8sMukYurn8jgb+Qj8ZOJPszpLozupL4gXx2KYgQtcDdMkZE9nVcntJw0R7Wi/ZuJtX7u5fjNNINBnwffP4zbqLbvGfjN3dfA44vVgPkbLGCwnX6ovvSfxMuXpnsdOK4GwdeTSZyxqg6WYRCPqwIrPk7kbaXsoWYhnKFpCXL5SLQCrxIgz8SMoESzqYb6w+IulOtwWR+oJbypZU8J3sou7aSGlT4Wls4ay9yY0DHNVnUTTMWAmJAoS/nzvXmbxQanuHCgE2VPiFS+38+puhQ+pJbKkm42LBfBaR8DqXTstazvv8zhINprr9RoXYKUcha8HKmqfbQHwtXD8mU42wal1t1cxpY6cbkdTD12KjMCpp5sPSlh73qj1jtuYz+pbOFC0Iwn0q3upoZx0Tzb2JD80vysqchkQEL2UnrjN9OtMF52RX6LBiQSTiwoP+SgQ5CgEuoxCDiuu9r7+8yN6gMlq/DpvPY6mlKlD3X38EdqGB3RVnMxrHv8QpdD22FsdRwSmcvqw927VMn7dqfL49l3f76PLWCOKC14OGPq9G7d2+Pae3mzq909DpFT7kJ5QAMkT4EBbo0vGg6Xg9QFDPF42m+NEzNscPQHrMUetIpwjC5+WaItVYHxUpGDVEyqE4UAhD+kEKXqi6NGVvqka1ABbHoBtO/SAWlN9wsqyjHbk2i7LtiQrmZ9IeMkEwo/xl4HkZzGgcfCdP6VSJJVU8JefF0wG+kpKQPNFwSmbPfgrJSxYynj4XzdNfrfnf9AMjnNjnhOpFBsW0q44GhtbIcly7vJTouY4EAHRGuLJGVeZoMcizDePHglOyPC6way3Hg1bv4r2JocF7ImYLGp+Xg91DSFiN0jwM2mNhp2m81zTJyx21EvDB3gd8wGTgZQqhIystLfibY53b33TJzFCW6W+dK6uaykNTWbhGV6g1XQFDV4HUyJZ60/V83gN4VWOfWjqHncLLLHp6NBZBJLmDRedT2D73wM6ZHRCZDvjP6m/2KQFWn1JUeFiOCttLPRBquOgh932LngxOoV0OTsH4uFVPL6egTGmoPEF7a6LdzzLaIUUQMOgotW2ML01y78AX1mnirgKM/oZVWh01uY0TJvnrk1wAHV+kOq9KOPXILH3PKFHESUKM1rIiMCl7KbBsYK4SdaEHAK0tE68FtT2I06rqAtBUV6eBmj3uD6u5vaM1t+myefKK3XFGNSsDT2nYLW0GraedqHGFj1JZ04A9SWUNlctfp6mqDcelSSddBS7IDFz6lXPZtrFY2OZiUZd07fOM07vKeTcsDtraGnW0NaqX8g9Hf2YAMA/X8SKJ4K15srPwcdnvZPsKSeZmuSV/HMKT8B+ulEH1UaQOUjez6uQz2X6QCBaJ9X8cfW9IkHzKXBpojxVLBMv+nj2rkOrYrjl71mkhS5P6R9oDHnW0+4AmDSlVQ7knlIrMiDJ1jvOnVNWzlxNcU8gFYOQUf67pDgi3pDpsVi17fZi1WhmHTo1Cuz3Mil/h9c/r6CdldeRUjmbU7X51yup6Vf+f1V8/sPPjkz39YnVs1il7wurG9lzdifqOWP1fMb3849lxV5g88oftEk/j1dBkpJskxQgiXzNM1yWWxHnizI+SqecsEsqtIGqn2Ou6NUYZt1S9rLUCNKzwKfLTTdOf2QzOZOS4Vv4r51uwZu9/b7kjrSKb+b8GZVlb/h9W6Po/</diagram></mxfile>

internals/cli/masker/masker_diagram.png

@@ -0,0 +1,244 @@
+package masker
+
+import (
+	"bytes"
+	"fmt"
+	"io"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/secrethub/secrethub-go/internals/assert"
+	"github.com/secrethub/secrethub-go/pkg/randchar"
+)
+
+var maskString = "<redacted by SecretHub>"
+
+func TestMasker(t *testing.T) {
+	delay10s := time.Second * 10
+	delay1us := time.Microsecond * 1
+
+	randomIn, err := randchar.NewGenerator(true).Generate(10000)
+	assert.OK(t, err)
+
+	tests := map[string]struct {
+		maskStrings []string
+		inputFunc   func(io.Writer)
+		options     *Options
+		expected    string
+	}{
+		"no_masking": {
+			maskStrings: []string{"foo", "bar"},
+			inputFunc: func(w io.Writer) {
+				_, err := w.Write([]byte("test"))
+				assert.OK(t, err)
+			},
+			expected: "test",
+		},
+		"single mask": {
+			maskStrings: []string{"foo", "bar"},
+			inputFunc: func(w io.Writer) {
+				_, err := w.Write([]byte("test foo test"))
+				assert.OK(t, err)
+			},
+			expected: "test " + maskString + " test",
+		},
+		"multiple masks": {
+			maskStrings: []string{"foo", "bar"},
+			inputFunc: func(w io.Writer) {
+				_, err := w.Write([]byte("test foo bar"))
+				assert.OK(t, err)
+			},
+			expected: "test " + maskString + " " + maskString,
+		},
+		"incomplete mask": {
+			maskStrings: []string{"foobar"},
+			inputFunc: func(w io.Writer) {
+				_, err := w.Write([]byte("test foo"))
+				assert.OK(t, err)
+			},
+			expected: "test foo",
+		},
+		"mask within a mask": {
+			maskStrings: []string{"foo", "bar", "testfoobartestfoo"},
+			inputFunc: func(w io.Writer) {
+				_, err := w.Write([]byte("testfoobartestfoo bar foo"))
+				assert.OK(t, err)
+			},
+			expected: maskString + " " + maskString + " " + maskString,
+		},
+		"across multiple writes": {
+			maskStrings: []string{"foo", "bar"},
+			inputFunc: func(w io.Writer) {
+				_, err := w.Write([]byte("fo"))
+				assert.OK(t, err)
+				_, err = w.Write([]byte("o bar f"))
+				assert.OK(t, err)
+				_, err = w.Write([]byte("o"))
+				assert.OK(t, err)
+			},
+			expected: maskString + " " + maskString + " fo",
+		},
+		"within buffer delay": {
+			maskStrings: []string{"foo", "bar"},
+			inputFunc: func(w io.Writer) {
+				_, err := w.Write([]byte("fo"))
+				assert.OK(t, err)
+				time.Sleep(time.Nanosecond * 1)
+				_, err = w.Write([]byte("o test"))
+				assert.OK(t, err)
+			},
+			options:  &Options{BufferDelay: delay10s},
+			expected: maskString + " test",
+		},
+		"outside buffer delay": {
+			maskStrings: []string{"foo", "bar"},
+			inputFunc: func(w io.Writer) {
+				_, err := w.Write([]byte("fo"))
+				assert.OK(t, err)
+				time.Sleep(time.Millisecond * 10)
+				_, err = w.Write([]byte("o bar test"))
+				assert.OK(t, err)
+			},
+			options:  &Options{BufferDelay: delay1us},
+			expected: "foo " + maskString + " test",
+		},
+		"no buffering": {
+			maskStrings: []string{"foo", "bar"},
+			inputFunc: func(w io.Writer) {
+				_, err := w.Write([]byte("test foo test"))
+				assert.OK(t, err)
+			},
+			options:  &Options{DisableBuffer: true},
+			expected: "test " + maskString + " test",
+		},
+		"long input": {
+			maskStrings: []string{},
+			inputFunc: func(w io.Writer) {
+				for _, c := range randomIn {
+					_, err := w.Write([]byte{c})
+					assert.OK(t, err)
+				}
+			},
+			expected: string(randomIn),
+		},
+		"reuse input buffer": {
+			maskStrings: []string{},
+			inputFunc: func(w io.Writer) {
+				tmp := make([]byte, 1)
+				for _, c := range randomIn {
+					copy(tmp, []byte{c})
+					_, err := w.Write(tmp)
+					assert.OK(t, err)
+				}
+			},
+			expected: string(randomIn),
+		},
+		"masking unicode": {
+			maskStrings: []string{
+				"ⓗⓔⓛⓛⓞ",
+			},
+			inputFunc: func(w io.Writer) {
+				_, err := w.Write([]byte("ⓗⓔⓛⓛⓞ world"))
+				assert.OK(t, err)
+			},
+			expected: maskString + " world",
+		},
+	}
+
+	for name, tc := range tests {
+		t.Run(name, func(t *testing.T) {
+			var buf bytes.Buffer
+
+			var maskStrings [][]byte
+			for _, s := range tc.maskStrings {
+				maskStrings = append(maskStrings, []byte(s))
+			}
+
+			m := New(maskStrings, tc.options)
+
+			writer := m.AddStream(&buf)
+			go m.Start()
+			tc.inputFunc(writer)
+
+			err = m.Stop()
+
+			assert.OK(t, err)
+			assert.Equal(t, buf.String(), tc.expected)
+		})
+	}
+}
+
+type errWriter struct {
+	err error
+}
+
+func (w errWriter) Write(p []byte) (n int, err error) {
+	return 0, w.err
+}
+
+func TestMasker_WriteError(t *testing.T) {
+	expectedErr := fmt.Errorf("test")
+
+	m := New([][]byte{[]byte("test")}, nil)
+	writer := m.AddStream(&errWriter{err: expectedErr})
+
+	go m.Start()
+	_, err := writer.Write([]byte{0x01})
+	assert.OK(t, err)
+
+	err = m.Stop()
+	assert.Equal(t, err, expectedErr)
+}
+
+func TestMasker_MultipleStreams(t *testing.T) {
+	sequences := [][]byte{
+		[]byte("Gandalf"),
+		[]byte("uruk-hai army"),
+		[]byte("Aragorn, son of Arathorn"),
+		[]byte("hobbit"),
+	}
+
+	input := [][]byte{
+		[]byte("line 1 "),
+		[]byte("line 2 "),
+		[]byte("line 3 "),
+		[]byte("message from Gandalf the Grey "),
+		[]byte("line 5 "),
+		[]byte("an uruk-hai army appears "),
+		[]byte("say hobbit hobbit hobbit "),
+	}
+
+	bufferDelay := 10 * time.Millisecond
+
+	m := New(sequences, &Options{
+		BufferDelay: bufferDelay,
+	})
+
+	var outputBuffer bytes.Buffer
+	var streams [3]io.Writer
+
+	for i := range streams {
+		streams[i] = m.AddStream(&outputBuffer)
+	}
+
+	go m.Start()
+
+	expected := ""
+
+	for i, b := range input {
+		n, err := streams[i%3].Write(b)
+		assert.OK(t, err)
+		assert.Equal(t, n, len(b))
+
+		expected += string(b)
+	}
+
+	err := m.Stop()
+	assert.OK(t, err)
+
+	for _, sequence := range sequences {
+		expected = strings.ReplaceAll(expected, string(sequence), maskString)
+	}
+	assert.Equal(t, outputBuffer.String(), expected)
+}