Catch errors when unmarshalling delta/doubleDelta encoded chunks

This is (hopefully) a fix for #1653

Specifically, this makes it so that if the length for the stored
delta/doubleDelta is somehow corrupted to be too small, the attempt to
unmarshal will return an error.

The current (broken) behavior is to return a malformed chunk, which can
then lead to a panic when there is an attempt to read header values.

The referenced issue proposed creating chunks with a minimum length -- I
instead opted to just error on the attempt to unmarshal, since I'm not
clear on how it could be safe to proceed when the length is
incorrect/unknown.

The issue also talked about possibly "quarantining series", but I don't
know the surrounding code well enough to understand how to make that
happen.

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,101 @@
+// 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 verifyUnmarshallingError(t *testing.T, err error, typ string, badLen int) {
+
+	if err == nil {
+		t.Errorf("Failed to obtain an error when unmarshalling %s with corrupt length of %d", typ, badLen)
+		return
+	}
+
+	expectedStr := "header size"
+	if !strings.Contains(err.Error(), expectedStr) {
+		t.Errorf(
+			"'%s' not present in error when unmarshalling %s with corrupt length %d: '%s'",
+			expectedStr,
+			typ,
+			badLen,
+			err.Error())
+	}
+}
+
+func TestUnmarshalingCorruptedDeltaReturnsAnError(t *testing.T) {
+	dec := newDeltaEncodedChunk(d1, d4, false, chunkLen)
+
+	cs, err := dec.add(model.SamplePair{
+		Timestamp: model.Now(),
+		Value:     model.SampleValue(100),
+	})
+	if err != nil {
+		t.Fatalf("Couldn't add sample to empty deltaEncodedChunk: %s", err)
+	}
+
+	buf := make([]byte, chunkLen)
+
+	cs[0].marshalToBuf(buf)
+
+	// Corrupt the length to be every possible too-small value
+	for i := 0; i < deltaHeaderBytes; i++ {
+		binary.LittleEndian.PutUint16(buf[deltaHeaderBufLenOffset:], uint16(i))
+
+		err = cs[0].unmarshalFromBuf(buf)
+		verifyUnmarshallingError(t, err, "deltaEncodedChunk (from buf)", i)
+
+		err = cs[0].unmarshal(bytes.NewBuffer(buf))
+		verifyUnmarshallingError(t, err, "deltaEncodedChunk (from Reader)", i)
+	}
+}
+
+func TestUnmarshalingCorruptedDoubleDeltaReturnsAnError(t *testing.T) {
+	ddec := newDoubleDeltaEncodedChunk(d1, d4, false, chunkLen)
+
+	cs, err := ddec.add(model.SamplePair{
+		Timestamp: model.Now(),
+		Value:     model.SampleValue(100),
+	})
+	if err != nil {
+		t.Fatalf("Couldn't add sample to empty doubleDeltaEncodedChunk: %s", err)
+	}
+
+	buf := make([]byte, chunkLen)
+
+	cs[0].marshalToBuf(buf)
+
+	// Corrupt the length to be every possible too-small value
+	for i := 0; i < doubleDeltaHeaderBytes; i++ {
+
+		binary.LittleEndian.PutUint16(buf[doubleDeltaHeaderBufLenOffset:], uint16(i))
+
+		err = cs[0].unmarshalFromBuf(buf)
+		verifyUnmarshallingError(t, err, "doubleDeltaEncodedChunk (from buf)", i)
+
+		err = cs[0].unmarshal(bytes.NewBuffer(buf))
+		verifyUnmarshallingError(t, err, "doubleDeltaEncodedChunk (from Reader)", i)
+	}
+
+}

storage/local/doubledelta.go

@@ -250,6 +250,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) < doubleDeltaHeaderBytes {
+		return fmt.Errorf("chunk length less than header size: %d < %d", l, doubleDeltaHeaderBytes)
+	}
+
 	*c = (*c)[:l]
 	return nil
 }
@@ -262,6 +266,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) < doubleDeltaHeaderBytes {
+		return fmt.Errorf("chunk length less than header size: %d < %d", l, doubleDeltaHeaderBytes)
+	}
 	*c = (*c)[:l]
 	return nil
 }