Merge pull request #1923 from dmilstein/fix-delta-unmarshaling

Catch errors when unmarshalling delta/doubleDelta encoded chunks

storage/local/delta.go

@@ -242,6 +242,9 @@ func (c *deltaEncodedChunk) unmarshal(r io.Reader) error {
 	if int(l) > cap(*c) {
 		return fmt.Errorf("chunk length exceeded during unmarshaling: %d", l)
 	}
+	if int(l) < deltaHeaderBytes {
+		return fmt.Errorf("chunk length less than header size: %d < %d", l, deltaHeaderBytes)
+	}
 	*c = (*c)[:l]
 	return nil
 }
@@ -254,6 +257,9 @@ func (c *deltaEncodedChunk) unmarshalFromBuf(buf []byte) error {
 	if int(l) > cap(*c) {
 		return fmt.Errorf("chunk length exceeded during unmarshaling: %d", l)
 	}
+	if int(l) < deltaHeaderBytes {
+		return fmt.Errorf("chunk length less than header size: %d < %d", l, deltaHeaderBytes)
+	}
 	*c = (*c)[:l]
 	return nil
 }

storage/local/delta_test.go

@@ -0,0 +1,103 @@
+// Copyright 2016 The Prometheus Authors
+// 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.
+
+// Note: this file has tests for code in both delta.go and doubledelta.go --
+// it may make sense to split those out later, but given that the tests are
+// near-identical and share a helper, this feels simpler for now.
+
+package local
+
+import (
+	"bytes"
+	"encoding/binary"
+	"strings"
+	"testing"
+
+	"github.com/prometheus/common/model"
+)
+
+func TestUnmarshalingCorruptedDeltaReturnsAnError(t *testing.T) {
+
+	var verifyUnmarshallingError = func(
+		err error,
+		chunkTypeName string,
+		unmarshalMethod string,
+		badLen int) {
+
+		if err == nil {
+			t.Errorf("Failed to obtain an error when unmarshalling %s (from %s) with corrupt length of %d", chunkTypeName, unmarshalMethod, badLen)
+			return
+		}
+
+		expectedStr := "header size"
+		if !strings.Contains(err.Error(), expectedStr) {
+			t.Errorf(
+				"'%s' not present in error when unmarshalling %s (from %s) with corrupt length %d: '%s'",
+				expectedStr,
+				chunkTypeName,
+				unmarshalMethod,
+				badLen,
+				err.Error())
+		}
+	}
+
+	cases := []struct {
+		chunkTypeName    string
+		chunkConstructor func(deltaBytes, deltaBytes, bool, int) chunk
+		minHeaderLen     int
+		chunkLenPos      int
+	}{
+		{
+			chunkTypeName: "deltaEncodedChunk",
+			chunkConstructor: func(a, b deltaBytes, c bool, d int) chunk {
+				return newDeltaEncodedChunk(a, b, c, d)
+			},
+			minHeaderLen: deltaHeaderBytes,
+			chunkLenPos:  deltaHeaderBufLenOffset,
+		},
+		{
+			chunkTypeName: "doubleDeltaEncodedChunk",
+			chunkConstructor: func(a, b deltaBytes, c bool, d int) chunk {
+				return newDoubleDeltaEncodedChunk(a, b, c, d)
+			},
+			minHeaderLen: doubleDeltaHeaderMinBytes,
+			chunkLenPos:  doubleDeltaHeaderBufLenOffset,
+		},
+	}
+	for _, c := range cases {
+		chunk := c.chunkConstructor(d1, d4, false, chunkLen)
+
+		cs, err := chunk.add(model.SamplePair{
+			Timestamp: model.Now(),
+			Value:     model.SampleValue(100),
+		})
+		if err != nil {
+			t.Fatalf("Couldn't add sample to empty %s: %s", c.chunkTypeName, err)
+		}
+
+		buf := make([]byte, chunkLen)
+
+		cs[0].marshalToBuf(buf)
+
+		// Corrupt the length to be every possible too-small value
+		for i := 0; i < c.minHeaderLen; i++ {
+			binary.LittleEndian.PutUint16(buf[c.chunkLenPos:], uint16(i))
+
+			err = cs[0].unmarshalFromBuf(buf)
+			verifyUnmarshallingError(err, c.chunkTypeName, "buf", i)
+
+			err = cs[0].unmarshal(bytes.NewBuffer(buf))
+			verifyUnmarshallingError(err, c.chunkTypeName, "Reader", i)
+		}
+	}
+}

storage/local/doubledelta.go

@@ -33,7 +33,8 @@ import (
 // - base time delta:          8 bytes
 // - base value delta:         8 bytes
 const (
-	doubleDeltaHeaderBytes = 37
+	doubleDeltaHeaderBytes    = 37
+	doubleDeltaHeaderMinBytes = 21 // header isn't full for chunk w/ one sample
 
 	doubleDeltaHeaderBufLenOffset         = 0
 	doubleDeltaHeaderTimeBytesOffset      = 2
@@ -250,6 +251,10 @@ func (c *doubleDeltaEncodedChunk) unmarshal(r io.Reader) error {
 	if int(l) > cap(*c) {
 		return fmt.Errorf("chunk length exceeded during unmarshaling: %d", l)
 	}
+	if int(l) < doubleDeltaHeaderMinBytes {
+		return fmt.Errorf("chunk length less than header size: %d < %d", l, doubleDeltaHeaderMinBytes)
+	}
+
 	*c = (*c)[:l]
 	return nil
 }
@@ -262,6 +267,9 @@ func (c *doubleDeltaEncodedChunk) unmarshalFromBuf(buf []byte) error {
 	if int(l) > cap(*c) {
 		return fmt.Errorf("chunk length exceeded during unmarshaling: %d", l)
 	}
+	if int(l) < doubleDeltaHeaderMinBytes {
+		return fmt.Errorf("chunk length less than header size: %d < %d", l, doubleDeltaHeaderMinBytes)
+	}
 	*c = (*c)[:l]
 	return nil
 }