Handle errors when reading packets

Author: akokhanovskyi

packets/connack.go

@@ -38,10 +38,14 @@ func (ca *ConnackPacket) Write(w io.Writer) error {
 //Unpack decodes the details of a ControlPacket after the fixed
 //header has been read
 func (ca *ConnackPacket) Unpack(b io.Reader) error {
-	ca.SessionPresent = 1&decodeByte(b) > 0
-	ca.ReturnCode = decodeByte(b)
+	flags, err := decodeByte(b)
+	if err != nil {
+		return err
+	}
+	ca.SessionPresent = 1&flags > 0
+	ca.ReturnCode, err = decodeByte(b)
 
-	return nil
+	return err
 }
 
 //Details returns a Details struct containing the Qos and

packets/connect.go

@@ -65,27 +65,55 @@ func (c *ConnectPacket) Write(w io.Writer) error {
 //Unpack decodes the details of a ControlPacket after the fixed
 //header has been read
 func (c *ConnectPacket) Unpack(b io.Reader) error {
-	c.ProtocolName = decodeString(b)
-	c.ProtocolVersion = decodeByte(b)
-	options := decodeByte(b)
+	var err error
+	c.ProtocolName, err = decodeString(b)
+	if err != nil {
+		return err
+	}
+	c.ProtocolVersion, err = decodeByte(b)
+	if err != nil {
+		return err
+	}
+	options, err := decodeByte(b)
+	if err != nil {
+		return err
+	}
 	c.ReservedBit = 1 & options
 	c.CleanSession = 1&(options>>1) > 0
 	c.WillFlag = 1&(options>>2) > 0
 	c.WillQos = 3 & (options >> 3)
 	c.WillRetain = 1&(options>>5) > 0
 	c.PasswordFlag = 1&(options>>6) > 0
 	c.UsernameFlag = 1&(options>>7) > 0
-	c.Keepalive = decodeUint16(b)
-	c.ClientIdentifier = decodeString(b)
+	c.Keepalive, err = decodeUint16(b)
+	if err != nil {
+		return err
+	}
+	c.ClientIdentifier, err = decodeString(b)
+	if err != nil {
+		return err
+	}
 	if c.WillFlag {
-		c.WillTopic = decodeString(b)
-		c.WillMessage = decodeBytes(b)
+		c.WillTopic, err = decodeString(b)
+		if err != nil {
+			return err
+		}
+		c.WillMessage, err = decodeBytes(b)
+		if err != nil {
+			return err
+		}
 	}
 	if c.UsernameFlag {
-		c.Username = decodeString(b)
+		c.Username, err = decodeString(b)
+		if err != nil {
+			return err
+		}
 	}
 	if c.PasswordFlag {
-		c.Password = decodeBytes(b)
+		c.Password, err = decodeBytes(b)
+		if err != nil {
+			return err
+		}
 	}
 
 	return nil

packets/packets.go

@@ -98,19 +98,25 @@ var ConnErrors = map[byte]error{
 //to read an MQTT packet from the stream. It returns a ControlPacket
 //representing the decoded MQTT packet and an error. One of these returns will
 //always be nil, a nil ControlPacket indicating an error occurred.
-func ReadPacket(r io.Reader) (cp ControlPacket, err error) {
+func ReadPacket(r io.Reader) (ControlPacket, error) {
 	var fh FixedHeader
 	b := make([]byte, 1)
 
-	_, err = io.ReadFull(r, b)
+	_, err := io.ReadFull(r, b)
 	if err != nil {
 		return nil, err
 	}
-	fh.unpack(b[0], r)
-	cp = NewControlPacketWithHeader(fh)
-	if cp == nil {
-		return nil, errors.New("Bad data from client")
+
+	err = fh.unpack(b[0], r)
+	if err != nil {
+		return nil, err
+	}
+
+	cp, err := NewControlPacketWithHeader(fh)
+	if err != nil {
+		return nil, err
 	}
+
 	packetBytes := make([]byte, fh.RemainingLength)
 	n, err := io.ReadFull(r, packetBytes)
 	if err != nil {
@@ -128,79 +134,75 @@ func ReadPacket(r io.Reader) (cp ControlPacket, err error) {
 //by packetType, this is usually done by reference to the packet type constants
 //defined in packets.go. The newly created ControlPacket is empty and a pointer
 //is returned.
-func NewControlPacket(packetType byte) (cp ControlPacket) {
+func NewControlPacket(packetType byte) ControlPacket {
 	switch packetType {
 	case Connect:
-		cp = &ConnectPacket{FixedHeader: FixedHeader{MessageType: Connect}}
+		return &ConnectPacket{FixedHeader: FixedHeader{MessageType: Connect}}
 	case Connack:
-		cp = &ConnackPacket{FixedHeader: FixedHeader{MessageType: Connack}}
+		return &ConnackPacket{FixedHeader: FixedHeader{MessageType: Connack}}
 	case Disconnect:
-		cp = &DisconnectPacket{FixedHeader: FixedHeader{MessageType: Disconnect}}
+		return &DisconnectPacket{FixedHeader: FixedHeader{MessageType: Disconnect}}
 	case Publish:
-		cp = &PublishPacket{FixedHeader: FixedHeader{MessageType: Publish}}
+		return &PublishPacket{FixedHeader: FixedHeader{MessageType: Publish}}
 	case Puback:
-		cp = &PubackPacket{FixedHeader: FixedHeader{MessageType: Puback}}
+		return &PubackPacket{FixedHeader: FixedHeader{MessageType: Puback}}
 	case Pubrec:
-		cp = &PubrecPacket{FixedHeader: FixedHeader{MessageType: Pubrec}}
+		return &PubrecPacket{FixedHeader: FixedHeader{MessageType: Pubrec}}
 	case Pubrel:
-		cp = &PubrelPacket{FixedHeader: FixedHeader{MessageType: Pubrel, Qos: 1}}
+		return &PubrelPacket{FixedHeader: FixedHeader{MessageType: Pubrel, Qos: 1}}
 	case Pubcomp:
-		cp = &PubcompPacket{FixedHeader: FixedHeader{MessageType: Pubcomp}}
+		return &PubcompPacket{FixedHeader: FixedHeader{MessageType: Pubcomp}}
 	case Subscribe:
-		cp = &SubscribePacket{FixedHeader: FixedHeader{MessageType: Subscribe, Qos: 1}}
+		return &SubscribePacket{FixedHeader: FixedHeader{MessageType: Subscribe, Qos: 1}}
 	case Suback:
-		cp = &SubackPacket{FixedHeader: FixedHeader{MessageType: Suback}}
+		return &SubackPacket{FixedHeader: FixedHeader{MessageType: Suback}}
 	case Unsubscribe:
-		cp = &UnsubscribePacket{FixedHeader: FixedHeader{MessageType: Unsubscribe, Qos: 1}}
+		return &UnsubscribePacket{FixedHeader: FixedHeader{MessageType: Unsubscribe, Qos: 1}}
 	case Unsuback:
-		cp = &UnsubackPacket{FixedHeader: FixedHeader{MessageType: Unsuback}}
+		return &UnsubackPacket{FixedHeader: FixedHeader{MessageType: Unsuback}}
 	case Pingreq:
-		cp = &PingreqPacket{FixedHeader: FixedHeader{MessageType: Pingreq}}
+		return &PingreqPacket{FixedHeader: FixedHeader{MessageType: Pingreq}}
 	case Pingresp:
-		cp = &PingrespPacket{FixedHeader: FixedHeader{MessageType: Pingresp}}
-	default:
-		return nil
+		return &PingrespPacket{FixedHeader: FixedHeader{MessageType: Pingresp}}
 	}
-	return cp
+	return nil
 }
 
 //NewControlPacketWithHeader is used to create a new ControlPacket of the type
 //specified within the FixedHeader that is passed to the function.
 //The newly created ControlPacket is empty and a pointer is returned.
-func NewControlPacketWithHeader(fh FixedHeader) (cp ControlPacket) {
+func NewControlPacketWithHeader(fh FixedHeader) (ControlPacket, error) {
 	switch fh.MessageType {
 	case Connect:
-		cp = &ConnectPacket{FixedHeader: fh}
+		return &ConnectPacket{FixedHeader: fh}, nil
 	case Connack:
-		cp = &ConnackPacket{FixedHeader: fh}
+		return &ConnackPacket{FixedHeader: fh}, nil
 	case Disconnect:
-		cp = &DisconnectPacket{FixedHeader: fh}
+		return &DisconnectPacket{FixedHeader: fh}, nil
 	case Publish:
-		cp = &PublishPacket{FixedHeader: fh}
+		return &PublishPacket{FixedHeader: fh}, nil
 	case Puback:
-		cp = &PubackPacket{FixedHeader: fh}
+		return &PubackPacket{FixedHeader: fh}, nil
 	case Pubrec:
-		cp = &PubrecPacket{FixedHeader: fh}
+		return &PubrecPacket{FixedHeader: fh}, nil
 	case Pubrel:
-		cp = &PubrelPacket{FixedHeader: fh}
+		return &PubrelPacket{FixedHeader: fh}, nil
 	case Pubcomp:
-		cp = &PubcompPacket{FixedHeader: fh}
+		return &PubcompPacket{FixedHeader: fh}, nil
 	case Subscribe:
-		cp = &SubscribePacket{FixedHeader: fh}
+		return &SubscribePacket{FixedHeader: fh}, nil
 	case Suback:
-		cp = &SubackPacket{FixedHeader: fh}
+		return &SubackPacket{FixedHeader: fh}, nil
 	case Unsubscribe:
-		cp = &UnsubscribePacket{FixedHeader: fh}
+		return &UnsubscribePacket{FixedHeader: fh}, nil
 	case Unsuback:
-		cp = &UnsubackPacket{FixedHeader: fh}
+		return &UnsubackPacket{FixedHeader: fh}, nil
 	case Pingreq:
-		cp = &PingreqPacket{FixedHeader: fh}
+		return &PingreqPacket{FixedHeader: fh}, nil
 	case Pingresp:
-		cp = &PingrespPacket{FixedHeader: fh}
-	default:
-		return nil
+		return &PingrespPacket{FixedHeader: fh}, nil
 	}
-	return cp
+	return nil, fmt.Errorf("unsupported packet type 0x%x", fh.MessageType)
 }
 
 //Details struct returned by the Details() function called on
@@ -241,24 +243,34 @@ func (fh *FixedHeader) pack() bytes.Buffer {
 	return header
 }
 
-func (fh *FixedHeader) unpack(typeAndFlags byte, r io.Reader) {
+func (fh *FixedHeader) unpack(typeAndFlags byte, r io.Reader) error {
 	fh.MessageType = typeAndFlags >> 4
 	fh.Dup = (typeAndFlags>>3)&0x01 > 0
 	fh.Qos = (typeAndFlags >> 1) & 0x03
 	fh.Retain = typeAndFlags&0x01 > 0
-	fh.RemainingLength = decodeLength(r)
+
+	var err error
+	fh.RemainingLength, err = decodeLength(r)
+	return err
 }
 
-func decodeByte(b io.Reader) byte {
+func decodeByte(b io.Reader) (byte, error) {
 	num := make([]byte, 1)
-	b.Read(num)
-	return num[0]
+	_, err := b.Read(num)
+	if err != nil {
+		return 0, err
+	}
+
+	return num[0], nil
 }
 
-func decodeUint16(b io.Reader) uint16 {
+func decodeUint16(b io.Reader) (uint16, error) {
 	num := make([]byte, 2)
-	b.Read(num)
-	return binary.BigEndian.Uint16(num)
+	_, err := b.Read(num)
+	if err != nil {
+		return 0, err
+	}
+	return binary.BigEndian.Uint16(num), nil
 }
 
 func encodeUint16(num uint16) []byte {
@@ -268,19 +280,27 @@ func encodeUint16(num uint16) []byte {
 }
 
 func encodeString(field string) []byte {
-
 	return encodeBytes([]byte(field))
 }
 
-func decodeString(b io.Reader) string {
-	return string(decodeBytes(b))
+func decodeString(b io.Reader) (string, error) {
+	buf, err := decodeBytes(b)
+	return string(buf), err
 }
 
-func decodeBytes(b io.Reader) []byte {
-	fieldLength := decodeUint16(b)
+func decodeBytes(b io.Reader) ([]byte, error) {
+	fieldLength, err := decodeUint16(b)
+	if err != nil {
+		return nil, err
+	}
+
 	field := make([]byte, fieldLength)
-	b.Read(field)
-	return field
+	_, err = b.Read(field)
+	if err != nil {
+		return nil, err
+	}
+
+	return field, nil
 }
 
 func encodeBytes(field []byte) []byte {
@@ -305,18 +325,22 @@ func encodeLength(length int) []byte {
 	return encLength
 }
 
-func decodeLength(r io.Reader) int {
+func decodeLength(r io.Reader) (int, error) {
 	var rLength uint32
 	var multiplier uint32
 	b := make([]byte, 1)
 	for multiplier < 27 { //fix: Infinite '(digit & 128) == 1' will cause the dead loop
-		io.ReadFull(r, b)
+		_, err := io.ReadFull(r, b)
+		if err != nil {
+			return 0, err
+		}
+
 		digit := b[0]
 		rLength |= uint32(digit&127) << multiplier
 		if (digit & 128) == 0 {
 			break
 		}
 		multiplier += 7
 	}
-	return int(rLength)
+	return int(rLength), nil
 }

packets/packets_test.go

@@ -192,11 +192,11 @@ func TestPackUnpackControlPackets(t *testing.T) {
 }
 
 func TestEncoding(t *testing.T) {
-	if res := decodeByte(bytes.NewBuffer([]byte{0x56})); res != 0x56 {
-		t.Errorf("decodeByte([0x56]) did not return 0x56 but 0x%X", res)
+	if res, err := decodeByte(bytes.NewBuffer([]byte{0x56})); res != 0x56 || err != nil {
+		t.Errorf("decodeByte([0x56]) did not return (0x56, nil) but (0x%X, %v)", res, err)
 	}
-	if res := decodeUint16(bytes.NewBuffer([]byte{0x56, 0x78})); res != 22136 {
-		t.Errorf("decodeUint16([0x5678]) did not return 22136 but %d", res)
+	if res, err := decodeUint16(bytes.NewBuffer([]byte{0x56, 0x78})); res != 22136 || err != nil {
+		t.Errorf("decodeUint16([0x5678]) did not return (22136, nil) but (%d, %v)", res, err)
 	}
 	if res := encodeUint16(22136); !bytes.Equal(res, []byte{0x56, 0x78}) {
 		t.Errorf("encodeUint16(22136) did not return [0x5678] but [0x%X]", res)
@@ -208,11 +208,11 @@ func TestEncoding(t *testing.T) {
 		"A\U0002A6D4": []byte{0x00, 0x05, 'A', 0xF0, 0xAA, 0x9B, 0x94},
 	}
 	for str, encoded := range strings {
-		if res := decodeString(bytes.NewBuffer(encoded)); res != str {
-			t.Errorf(`decodeString(%v) did not return "%s", but "%s"`, encoded, str, res)
+		if res, err := decodeString(bytes.NewBuffer(encoded)); res != str || err != nil {
+			t.Errorf("decodeString(%v) did not return (%q, nil), but (%q, %v)", encoded, str, res, err)
 		}
 		if res := encodeString(str); !bytes.Equal(res, encoded) {
-			t.Errorf(`encodeString("%s") did not return [0x%X], but [0x%X]`, str, encoded, res)
+			t.Errorf("encodeString(%q) did not return [0x%X], but [0x%X]", str, encoded, res)
 		}
 	}
 
@@ -227,8 +227,8 @@ func TestEncoding(t *testing.T) {
 		268435455: []byte{0xFF, 0xFF, 0xFF, 0x7F},
 	}
 	for length, encoded := range lengths {
-		if res := decodeLength(bytes.NewBuffer(encoded)); res != length {
-			t.Errorf("decodeLength([0x%X]) did not return %d, but %d", encoded, length, res)
+		if res, err := decodeLength(bytes.NewBuffer(encoded)); res != length || err != nil {
+			t.Errorf("decodeLength([0x%X]) did not return (%d, nil) but (%d, %v)", encoded, length, res, err)
 		}
 		if res := encodeLength(length); !bytes.Equal(res, encoded) {
 			t.Errorf("encodeLength(%d) did not return [0x%X], but [0x%X]", length, encoded, res)

packets/puback.go

@@ -32,9 +32,10 @@ func (pa *PubackPacket) Write(w io.Writer) error {
 //Unpack decodes the details of a ControlPacket after the fixed
 //header has been read
 func (pa *PubackPacket) Unpack(b io.Reader) error {
-	pa.MessageID = decodeUint16(b)
+	var err error
+	pa.MessageID, err = decodeUint16(b)
 
-	return nil
+	return err
 }
 
 //Details returns a Details struct containing the Qos and

packets/pubcomp.go

@@ -32,9 +32,10 @@ func (pc *PubcompPacket) Write(w io.Writer) error {
 //Unpack decodes the details of a ControlPacket after the fixed
 //header has been read
 func (pc *PubcompPacket) Unpack(b io.Reader) error {
-	pc.MessageID = decodeUint16(b)
+	var err error
+	pc.MessageID, err = decodeUint16(b)
 
-	return nil
+	return err
 }
 
 //Details returns a Details struct containing the Qos and