proto.go

package gobroke

import (
	"bytes"
	"encoding/binary"
	"errors"
	"fmt"
	"time"
	"unicode/utf8"

	"github.com/RoanBrand/gobroke/internal/model"
	"github.com/RoanBrand/gobroke/internal/queue"
	log "github.com/sirupsen/logrus"
)

// MQTT packet parser states
const (
	// Fixed header
	controlAndFlags = iota
	length

	variableHeaderLen
	variableHeader

	propertiesLen
	properties

	payload
)

const maxVarLenMul = 128 * 128 * 128

var (
	protoVersionToName = map[uint8][]byte{
		3: {'M', 'Q', 'I', 's', 'd', 'p'},
		4: {'M', 'Q', 'T', 'T'},
		5: {'M', 'Q', 'T', 'T'},
	}

	pingRespPacket = []byte{model.PINGRESP, 0}

	isSharedSub = []byte("$share")

	// Close the connection normally. Do not send the Will Message.
	errGotNormalDiscon = errors.New("Normal")
	// The Client wishes to disconnect but requires
	// that the Server also publishes its Will Message.
	errGotDisconWithWill = errors.New("WithWill")
	// Publish received with expiry == 0. Do not forward.
	errPubExpired = errors.New("EOA")
)

func (s *Server) parseStream(ses *session, rx []byte) error {
	p, l := &ses.packet, len(rx)

	for i := 0; i < l; {
		switch p.rxState {
		case controlAndFlags:
			p.controlType, p.flags = rx[i]&0xF0, rx[i]&0x0F

			switch p.controlType { // [MQTT-2.2.2-1, 2-2]
			case model.PUBLISH:
				break
			case model.PUBACK, model.PUBREC, model.PUBCOMP, model.PINGREQ:
				if p.flags != 0 {
					return errors.New("malformed packet: fixed header flags must be 0 (reserved)")
				}
			case model.PUBREL:
				f := p.flags
				if ses.protoVersion == 3 {
					f &= 0x07
				}
				if f != 0x02 {
					return errors.New("malformed PUBREL")
				}
			case model.SUBSCRIBE:
				f := p.flags
				if ses.protoVersion == 3 {
					f &= 0x07
				}
				if f != 0x02 { // [MQTT-3.8.1-1]
					return errors.New("malformed SUBSCRIBE: bad fixed header reserved flags")
				}
			case model.UNSUBSCRIBE:
				f := p.flags
				if ses.protoVersion == 3 {
					f &= 0x07
				}
				if f != 0x02 { // [MQTT-3.10.1-1]
					return errors.New("malformed UNSUBSCRIBE: bad fixed header reserved flags")
				}
			case model.DISCONNECT:
				if p.flags != 0 { // [MQTT-3.14.1-1]
					ses.disconnectReasonCode = model.MalformedPacket
					return errors.New("malformed DISCONNECT: fixed header flags must be 0 (reserved)")
				}
			case model.CONNECT:
				if ses.connectSent { // [MQTT-3.1.0-2]
					return errors.New("second CONNECT packet received")
				}
				if p.flags != 0 {
					return errors.New("malformed packet: fixed header flags must be 0 (reserved)")
				}
			default:
				return errors.New("invalid MQTT Control Packet type")
			}

			p.rxState = length
			p.lenMul = 1
			i++
		case length:
			p.remainingLength += int(rx[i]&127) * p.lenMul
			if p.lenMul > maxVarLenMul {
				return errors.New("malformed packet: bad remaining length")
			}

			if rx[i]&128 == 0 {
				switch p.controlType {
				case model.PUBLISH:
					p.vhToRead = 2
					p.vhBuf = p.vhBuf[:0]
					p.rxState = variableHeaderLen
				case model.PUBACK, model.PUBREC, model.PUBREL, model.PUBCOMP:
					p.vhToRead = 2
					if ses.protoVersion == 5 && p.remainingLength > 2 {
						p.vhToRead++ // Reason Code
					}

					p.vhBuf = p.vhBuf[:0]
					p.rxState = variableHeader
				case model.SUBSCRIBE:
					// [MQTT-3.8.3-3]
					if p.remainingLength < 5 || (ses.protoVersion == 5 && p.remainingLength < 6) {
						return errors.New("malformed SUBSCRIBE: no Topic Filter present")
					}

					p.vhToRead = 2
					p.vhBuf = p.vhBuf[:0]
					p.rxState = variableHeader
				case model.UNSUBSCRIBE:
					if p.remainingLength < 4 || (ses.protoVersion == 5 && p.remainingLength < 5) { // [MQTT-3.10.3-2]
						return errors.New("malformed UNSUBSCRIBE: no Topic Filter present")
					}

					p.vhToRead = 2
					p.vhBuf = p.vhBuf[:0]
					p.rxState = variableHeader
				case model.PINGREQ:
					if err := ses.writePacket(pingRespPacket); err != nil {
						return err
					}

					ses.updateTimeout()
					p.rxState = controlAndFlags
				case model.CONNECT:
					p.vhToRead = 2
					p.vhBuf = p.vhBuf[:0]
					p.rxState = variableHeaderLen
				case model.DISCONNECT:
					if p.remainingLength == 0 {
						log.WithFields(log.Fields{
							"ClientId": ses.clientId,
							"Reason":   "Normal disconnection",
						}).Debug("DISCONNECT received")

						return errGotNormalDiscon
					} else if ses.protoVersion < 5 {
						return errors.New("malformed DISCONNECT: proto < v5")
					}

					p.vhToRead = 1 // Reason Code
					p.vhBuf = p.vhBuf[:0]
					p.rxState = variableHeader
				}
			} else {
				p.lenMul *= 128
			}

			i++
		case variableHeaderLen:
			toRead := p.vhToRead
			if avail := l - i; avail < toRead {
				toRead = avail
			}

			p.vhBuf = append(p.vhBuf, rx[i:i+toRead]...)
			p.vhToRead -= toRead
			p.remainingLength -= toRead

			if p.vhToRead == 0 {
				p.vhToRead = int(binary.BigEndian.Uint16(p.vhBuf))
				if p.controlType == model.PUBLISH {
					rawQoS := p.flags & 0x06
					if rawQoS > 0 { // Qos > 0
						if rawQoS == 6 { // [MQTT-3.3.1-4]
							ses.disconnectReasonCode = model.MalformedPacket
							return errors.New("malformed PUBLISH: no QoS3 allowed")
						}

						p.vhToRead += 2
					} else if p.flags&0x08 > 0 { // [MQTT-3.3.1-2]
						return errors.New("malformed PUBLISH: DUP set for QoS0")
					}

				} else { // CONNECT
					p.vhToRead += 4 // ProtoVersion + ConnectFlags + KeepAlive
				}
				p.rxState = variableHeader
			}

			i += toRead
		case variableHeader:
			toRead := p.vhToRead
			if avail := l - i; avail < toRead {
				toRead = avail
			}

			p.vhBuf = append(p.vhBuf, rx[i:i+toRead]...)
			p.remainingLength -= toRead
			p.vhToRead -= toRead

			if p.vhToRead == 0 {
				c := ses.client
				switch p.controlType {
				case model.PUBLISH:
					tLen := binary.BigEndian.Uint16(p.vhBuf)
					if tLen > 0 {
						if err := checkUTF8(p.vhBuf[2:2+tLen], true); err != nil { // [MQTT-3.3.2-1]
							return errors.New("malformed PUBLISH: bad Topic Name: " + err.Error())
						}
					}
					if p.flags&0x06 > 0 { // Qos > 0
						p.pID = binary.BigEndian.Uint16(p.vhBuf[2+tLen:])
					}
				case model.PUBACK:
					if p.remainingLength == 0 {
						c.qos1Done(binary.BigEndian.Uint16(p.vhBuf))
					}
					c.session.incSendQuota()
					if len(p.vhBuf) > 2 && ses.protoVersion == 5 && p.vhBuf[2] != 0 {
						log.WithFields(log.Fields{
							"ClientId":    ses.clientId,
							"Reason Code": p.vhBuf[2],
						}).Debug("PUBACK received")
					}
				case model.PUBREC:
					if p.remainingLength == 0 {
						if err := ses.handlePubrec(); err != nil {
							return err
						}
					}
					if len(p.vhBuf) > 2 && ses.protoVersion == 5 && p.vhBuf[2] != 0 {
						if p.vhBuf[2] >= model.UnspecifiedError {
							c.session.incSendQuota()
						}
						log.WithFields(log.Fields{
							"ClientId":    ses.clientId,
							"Reason Code": p.vhBuf[2],
						}).Debug("PUBREC received")
					}
				case model.PUBREL: // [MQTT-4.3.3-2]
					if p.remainingLength == 0 {
						if err := ses.handlePubrel(); err != nil {
							return err
						}
					}
					if len(p.vhBuf) > 2 && ses.protoVersion == 5 && p.vhBuf[2] != 0 {
						log.WithFields(log.Fields{
							"ClientId":    ses.clientId,
							"Reason Code": p.vhBuf[2],
						}).Debug("PUBREL received")
					}
				case model.PUBCOMP:
					if p.remainingLength == 0 {
						c.qos2Part2Done(binary.BigEndian.Uint16(p.vhBuf))
					}
					c.session.incSendQuota()
					if len(p.vhBuf) > 2 && ses.protoVersion == 5 && p.vhBuf[2] != 0 {
						log.WithFields(log.Fields{
							"ClientId":    ses.clientId,
							"Reason Code": p.vhBuf[2],
						}).Debug("PUBCOMP received")
					}
				case model.CONNECT:
					pnLen := binary.BigEndian.Uint16(p.vhBuf)
					ses.protoVersion = p.vhBuf[2+pnLen]

					if !bytes.Equal(protoVersionToName[ses.protoVersion], p.vhBuf[2:2+pnLen]) { // [MQTT-3.1.2-1]
						if ses.protoVersion < 5 {
							ses.sendConnackFail(1) // [MQTT-3.1.2-2]
						} else {
							ses.sendConnackFail(model.UnsupportedProtocolVersion)
						}

						return errors.New("unsupported client protocol. Must be MQTT 3 (v3.1), 4 (v3.1.1) or 5 (v5)")
					}

					switch ses.protoVersion {
					case 3:
						if p.remainingLength < 3 {
							return errors.New("invalid CONNECT: ClientId must be between 1-23 characters long for MQTT 3 (v3.1)")
						}
					case 4:
						if p.remainingLength < 2 { // [MQTT-3.1.3-3]
							return errors.New("malformed CONNECT: absent ClientId in payload")
						}
					case 5:
						if p.remainingLength < 3 {
							return errors.New("malformed CONNECT: too short (missing Properties Length or ClientId")
						}
					}

					ses.connectFlags = p.vhBuf[3+pnLen]
					if ses.connectFlags&0x01 > 0 { // [MQTT-3.1.2-3]
						return errors.New("malformed CONNECT: reserved header flag must be 0")
					}

					ses.keepAlive = time.Duration(binary.BigEndian.Uint16(p.vhBuf[4+pnLen:]))
				case model.DISCONNECT:
					if p.remainingLength == 0 {
						return ses.handleDisconnect()
					}
				}

				//p.payload = p.payload[:0]
				if p.remainingLength == 0 {
					if p.controlType == model.PUBLISH {
						p.payload = p.payload[:0]
						if err := s.handlePublish(ses); err != nil {
							return err
						}
					}
					ses.updateTimeout()
					p.rxState = controlAndFlags
				} else if ses.protoVersion > 4 {
					p.lenMul = 1
					p.rxState = propertiesLen
				} else {
					p.payload = p.payload[:0]
					p.rxState = payload
				}
			}

			i += toRead
		case payload:
			toRead := p.remainingLength
			if avail := l - i; avail < toRead {
				toRead = avail
			}

			p.payload = append(p.payload, rx[i:i+toRead]...)
			p.remainingLength -= toRead

			if p.remainingLength == 0 {
				var err error
				switch p.controlType {
				case model.PUBLISH:
					err = s.handlePublish(ses)
				case model.SUBSCRIBE:
					err = s.handleSubscribe(ses)
				case model.UNSUBSCRIBE:
					err = s.handleUnsubscribe(ses)
				case model.CONNECT:
					err = s.handleConnect(ses)
				}
				if err != nil {
					return err
				}

				ses.updateTimeout()
				p.rxState = controlAndFlags
			}

			i += toRead
		case propertiesLen:
			p.vhPropToRead += int(rx[i]&127) * p.lenMul
			if p.lenMul > maxVarLenMul {
				return errors.New("malformed packet: bad Properties Length")
			}

			p.remainingLength--
			if rx[i]&128 == 0 {
				if p.vhPropToRead == 0 {
					switch p.controlType {
					case model.PUBLISH:
						p.payload = p.payload[:0]
						if p.remainingLength == 0 {
							if err := s.handlePublish(ses); err != nil {
								return err
							}
							ses.updateTimeout()
							p.rxState = controlAndFlags
						} else {
							p.rxState = payload
						}
					case model.PUBACK:
						ses.client.qos1Done(binary.BigEndian.Uint16(p.vhBuf))
						p.rxState = controlAndFlags
					case model.PUBREC:
						if err := ses.handlePubrec(); err != nil {
							return err
						}
						p.rxState = controlAndFlags
					case model.PUBREL:
						if err := ses.handlePubrel(); err != nil {
							return err
						}
						p.rxState = controlAndFlags
					case model.PUBCOMP:
						ses.client.qos2Part2Done(binary.BigEndian.Uint16(p.vhBuf))
						p.rxState = controlAndFlags
					case model.DISCONNECT:
						return ses.handleDisconnect()
					default: // subscribe, unsubscribe, connect
						p.payload = p.payload[:0]
						p.rxState = payload
					}
				} else {
					p.rxState = properties
				}
			} else {
				p.lenMul *= 128
			}

			i++
		case properties:
			toRead := p.vhPropToRead
			if avail := l - i; avail < toRead {
				toRead = avail
			}

			p.vhBuf = append(p.vhBuf, rx[i:i+toRead]...)
			p.remainingLength -= toRead
			p.vhPropToRead -= toRead

			if p.vhPropToRead == 0 {
				switch p.controlType {
				case model.PUBLISH:
					p.payload = p.payload[:0]
					if p.remainingLength == 0 {
						if err := s.handlePublish(ses); err != nil {
							return err
						}
						ses.updateTimeout()
						p.rxState = controlAndFlags
					} else {
						p.rxState = payload
					}
				case model.PUBACK:
					// TODO: handle PUBACK props
					ses.client.qos1Done(binary.BigEndian.Uint16(p.vhBuf))
					p.rxState = controlAndFlags
				case model.PUBREC:
					// TODO: handle PUBREC props
					if err := ses.handlePubrec(); err != nil {
						return err
					}
					p.rxState = controlAndFlags
				case model.PUBREL:
					// TODO: handle PUBREL props
					if err := ses.handlePubrel(); err != nil {
						return err
					}
					p.rxState = controlAndFlags
				case model.PUBCOMP:
					// TODO: handle PUBCOMP props
					ses.client.qos2Part2Done(binary.BigEndian.Uint16(p.vhBuf))
					p.rxState = controlAndFlags
				case model.SUBSCRIBE:
					/*if err := ses.handleSubscribeProperties(); err != nil {
						return err
					}*/
					p.payload = p.payload[:0]
					p.rxState = payload
				case model.UNSUBSCRIBE:
					if err := s.handleUnSubscribeProperties(ses); err != nil {
						return err
					}
					p.payload = p.payload[:0]
					p.rxState = payload
				case model.CONNECT:
					if err := s.handleConnectProperties(ses); err != nil {
						return err
					}
					p.payload = p.payload[:0]
					p.rxState = payload
				case model.DISCONNECT:
					return s.handleDisconnectProperties(ses)
				}
			}

			i += toRead
		}
	}

	return nil
}

func (s *Server) handleConnect(ses *session) error {
	p := ses.packet.payload
	pLen := len(p)

	// ClientId
	clientIdLen := int(binary.BigEndian.Uint16(p))
	offs := 2 + clientIdLen
	if pLen < offs {
		return errors.New("malformed CONNECT: payload too short for ClientId")
	}

	if clientIdLen > 0 {
		if ses.protoVersion == 3 && clientIdLen > 23 {
			ses.sendConnackFail(2)
			return errors.New("invalid CONNECT: ClientId must be between 1-23 characters long for MQTT 3 (v3.1)")
		}

		if err := checkUTF8(p[2:offs], false); err != nil { // [MQTT-3.1.3-4]
			return errors.New("malformed CONNECT: bad ClientId: " + err.Error())
		}

		ses.clientId = string(p[2:offs])
	} else {
		if ses.protoVersion < 5 && !ses.cleanStart() { // [MQTT-3.1.3-7]
			ses.sendConnackFail(2) // [MQTT-3.1.3-8]
			return errors.New("malformed CONNECT: must have ClientId with CleanSession set to 0")
		}

		ses.clientId = "auto-" + string(generateRandomID())
		ses.assignedCId = true
	}

	// Will Properties, Will Topic & Will Message/Payload
	if ses.connectFlags&0x04 > 0 {
		// Properties
		if ses.protoVersion == 5 {
			wpLen, vbLen, err := variableLengthDecode(p[offs:])
			if err != nil {
				return errors.New("malformed CONNECT: bad Will Properties Length")
			}
			// TODO: store and use Will Properties
			offs += vbLen + wpLen
		}

		// Topic
		if pLen < 2+offs {
			return errors.New("malformed CONNECT: no Will Topic in payload")
		}

		wTopicUTFStart := offs
		wTopicLen := int(binary.BigEndian.Uint16(p[offs:]))
		offs += 2
		if pLen < offs+wTopicLen {
			return errors.New("malformed CONNECT: payload too short for Will Topic")
		}

		wTopicUTFEnd := offs + wTopicLen
		offs += wTopicLen
		if pLen < 2+offs {
			return errors.New("malformed CONNECT: no Will Message in payload")
		}

		wTopicUTF8 := p[wTopicUTFStart:wTopicUTFEnd]
		if err := checkUTF8(wTopicUTF8[2:], true); err != nil { // [MQTT-3.1.3-10]
			return errors.New("malformed CONNECT: bad Will Topic: " + err.Error())
		}

		wMsgLen := int(binary.BigEndian.Uint16(p[offs:]))
		offs += 2
		if pLen < offs+wMsgLen {
			return errors.New("malformed CONNECT: payload too short for Will Message")
		}

		wQoS := (ses.connectFlags & 0x18) >> 3
		if wQoS > 2 { // [MQTT-3.1.2-14]
			return errors.New("malformed CONNECT: invalid Will QoS level")
		}

		willPubFlags := wQoS << 1
		if ses.connectFlags&0x20 > 0 {
			willPubFlags |= 0x01 // retain
		}

		ses.will = model.NewPubOld(willPubFlags, wTopicUTF8, p[offs:offs+wMsgLen])
		ses.will.Publisher = ses.clientId
		offs += wMsgLen

	} else if ses.connectFlags&0x38 > 0 { // [MQTT-3.1.2-11, 2-13, 2-15]
		return errors.New("malformed CONNECT: bad Will Flags")
	}

	// Username & Password
	var userName, password []byte
	if ses.connectFlags&0x80 > 0 {
		if pLen < 2+offs {
			if ses.protoVersion < 5 {
				ses.sendConnackFail(4)
			}
			return errors.New("malformed CONNECT: no User Name in payload")
		}

		userLen := int(binary.BigEndian.Uint16(p[offs:]))
		offs += 2
		if pLen < offs+userLen {
			if ses.protoVersion < 5 {
				ses.sendConnackFail(4)
			}
			return errors.New("malformed CONNECT: payload too short for User Name")
		}

		userName = p[offs : offs+userLen]
		if err := checkUTF8(userName, false); err != nil { // [MQTT-3.1.3-11]
			if ses.protoVersion < 5 {
				ses.sendConnackFail(4)
			}
			return errors.New("malformed CONNECT: bad User Name: " + err.Error())
		}

		offs += userLen
		if ses.connectFlags&0x40 > 0 {
			if pLen < 2+offs {
				if ses.protoVersion < 5 {
					ses.sendConnackFail(4)
				}
				return errors.New("malformed CONNECT: no Password in payload")
			}

			passLen := int(binary.BigEndian.Uint16(p[offs:]))
			offs += 2
			if pLen < offs+passLen {
				if ses.protoVersion < 5 {
					ses.sendConnackFail(4)
				}
				return errors.New("malformed CONNECT: payload too short for Password")
			}

			password = p[offs : offs+passLen]
			offs += passLen
		}
	} else if ses.connectFlags&0x40 > 0 {
		return errors.New("malformed CONNECT: Password present without User Name")
	}

	if s.Auther != nil {
		if err := s.Auther.AuthUser(ses.clientId, userName, password); err != nil {
			if ses.protoVersion < 5 {
				ses.sendConnackFail(5)
			} else {
				ses.sendConnackFail(model.NotAuthorized)
			}
			return errors.New("failed authentication for client " + ses.clientId + ": " + err.Error())
		}
	}

	if offs != pLen {
		return errors.New("malformed CONNECT: unexpected extra payload fields (Will Properties, Topic, Message, or User Name or Password)")
	}

	if err := ses.conn.SetReadDeadline(time.Time{}); err != nil { // CONNECT packet timeout cancel
		return err
	}

	ses.connectSent = true
	if ses.protoVersion > 4 {
		ses.taFromClient = make(map[uint16][]byte)
	}
	sessionIsPresent := s.addSession(ses)
	ses.ended.Add(1)
	go ses.startWriter()

	// [MQTT-3.2.2-1, 2-2, 2-3]
	ses.sendConnackSuccess(&s.Config, sessionIsPresent)
	ses.run(s.TimeoutQoS12MQTT34)
	return nil
}

func (s *Server) handleConnectProperties(ses *session) error {
	vh := ses.packet.vhBuf
	pnLen := binary.BigEndian.Uint16(vh)
	props := vh[pnLen+6:]
	var gotSesExp, gotRxMax, gotMaxPSize, gotTopAlias, gotRRI, gotRPI bool

	for i := 0; i < len(props); {
		remain := len(props) - i
		switch props[i] {
		case model.SessionExpiryInterval:
			if gotSesExp {
				return errors.New("malformed CONNECT: Session Expiry Interval included more than once")
			}
			if remain < 5 {
				return errors.New("malformed CONNECT: bad Session Expiry Interval")
			}

			ses.expiryInterval = binary.BigEndian.Uint32(props[i+1:])
			gotSesExp = true
			i += 5
		case model.ReceiveMaximum:
			if gotRxMax {
				return errors.New("malformed CONNECT: Receive Maximum included more than once")
			}
			if remain < 3 {
				return errors.New("malformed CONNECT: bad Receive Maximum")
			}

			ses.receiveMax = binary.BigEndian.Uint16(props[i+1:])
			if ses.receiveMax == 0 {
				return errors.New("malformed CONNECT: Receive Maximum 0 not allowed")
			}
			ses.sendQuota = make(chan struct{}, ses.receiveMax)
			for i := 0; i < int(ses.receiveMax); i++ {
				ses.sendQuota <- struct{}{}
			}

			gotRxMax = true
			i += 3
		case model.MaximumPacketSize:
			if gotMaxPSize {
				return errors.New("malformed CONNECT: Maximum Packet Size included more than once")
			}
			if remain < 5 {
				return errors.New("malformed CONNECT: bad Maximum Packet Size")
			}

			ses.maxPacketSize = binary.BigEndian.Uint32(props[i+1:])
			if ses.maxPacketSize == 0 {
				return errors.New("malformed CONNECT: Maximum Packet Size 0 not allowed")
			}

			gotMaxPSize = true
			i += 5
		case model.TopicAliasMaximum:
			if gotTopAlias {
				return errors.New("malformed CONNECT: Topic Alias Maximum included more than once")
			}
			if remain < 3 {
				return errors.New("malformed CONNECT: bad Topic Alias Maximum")
			}

			ses.topicAliasMax = binary.BigEndian.Uint16(props[i+1:])
			if ses.topicAliasMax > 0 {
				ses.taToClient.aliases = make(map[string]uint16, ses.topicAliasMax)
				ses.taToClient.left = int32(ses.topicAliasMax)
			}

			gotTopAlias = true
			i += 3
		case model.RequestResponseInformation:
			if gotRRI {
				return errors.New("malformed CONNECT: Request Response Information included more than once")
			}
			if remain < 2 {
				return errors.New("malformed CONNECT: bad Request Response Information")
			}
			if props[i+1] != 0 && props[i+1] != 1 {
				return errors.New("malformed CONNECT: bad Request Response Information")
			}

			ses.reqRespInfo = props[i+1] == 1
			gotRRI = true
			i += 2
		case model.RequestProblemInformation:
			if gotRPI {
				return errors.New("malformed CONNECT: Request Problem Information included more than once")
			}
			if remain < 2 {
				return errors.New("malformed CONNECT: bad Request Problem Information")
			}
			if props[i+1] != 0 && props[i+1] != 1 {
				return errors.New("malformed CONNECT: bad Request Problem Information")
			}

			ses.reqProblemInfo = props[i+1] == 1
			gotRPI = true
			i += 2
		case model.UserProperty:
			if remain < 5 {
				return errors.New("malformed CONNECT: bad User Property")
			}

			kLen := int(binary.BigEndian.Uint16(props[i+1:]))
			if remain < 5+kLen {
				return errors.New("malformed CONNECT: bad User Property")
			}

			vLen := int(binary.BigEndian.Uint16(props[i+3+kLen:]))
			expect := 5 + kLen + vLen
			if remain < expect {
				return errors.New("malformed CONNECT: bad User Property")
			}

			i += expect
		case model.AuthenticationMethod:
			if remain < 3 {
				return errors.New("malformed CONNECT: bad Authentication Method Property")
			}

			l := int(binary.BigEndian.Uint16(props[i+1:]))
			if remain < 3+l {
				return errors.New("malformed CONNECT: bad Authentication Method Property")
			}

			i += 3 + l
		case model.AuthenticationData:
			if remain < 3 {
				return errors.New("malformed CONNECT: bad Authentication Data Property")
			}

			l := int(binary.BigEndian.Uint16(props[i+1:]))
			if remain < 3+l {
				return errors.New("malformed CONNECT: bad Authentication Data Property")
			}

			i += 3 + l
		default:
			return fmt.Errorf("malformed CONNECT: unknown property %d (0x%x)", props[i], props[i])
		}
	}
	return nil
}

func (s *Server) handlePublish(ses *session) error {
	p := &ses.packet
	topicLen := int(binary.BigEndian.Uint16(p.vhBuf))
	var topicUTF8 []byte

	// Topic Alias
	if topicLen == 0 {
		if p.topicAlias == 0 { // [MQTT-4.7.3-1]
			return errors.New("malformed PUBLISH: empty Topic Name")
		} else {
			var ok bool
			topicUTF8, ok = ses.taFromClient[p.topicAlias]
			if !ok {
				ses.disconnectReasonCode = model.ProtocolError
				return errors.New("malformed PUBLISH: unknown Topic Alias")
			}
			p.topicAlias = 0
		}
	} else {
		topicUTF8 = p.vhBuf[:topicLen+2]
		if p.topicAlias != 0 {
			t := make([]byte, len(topicUTF8))
			copy(t, topicUTF8)
			ses.taFromClient[p.topicAlias] = t
			p.topicAlias = 0
		}
	}

	if s.Auther != nil {
		if err := s.Auther.AuthPublish(ses.clientId, topicUTF8[2:]); err != nil {
			return errors.New("failed publish authorization for client " + ses.clientId + ": " + err.Error())
		}
	}

	requiredLen := 1 + len(topicUTF8) + len(p.payload)
	pub := model.NewPub(requiredLen)

	pub.B[0] = p.flags
	pub.B = append(pub.B, topicUTF8...)
	pub.B = append(pub.B, p.payload...)

	propsIdx := 2 + topicLen
	if p.flags&0x06 > 0 { // Qos > 0
		propsIdx += 2
	}

	props := p.vhBuf[propsIdx:]
	if len(props) > 0 {
		if cap(pub.Props) < len(props) {
			pub.Props = make([]byte, 0, len(props))
		} else {
			pub.Props = pub.Props[:0]
		}

		var err error
		pub.Props, err = ses.handlePublishProperties(pub.Props, props)
		if err != nil {
			pub.FreeIfLastUser()
			if err == errPubExpired {
				if qos := pub.RxQoS(); qos == 1 {
					return ses.sendPuback(p.pID)
				} else if qos == 2 {
					return ses.sendPubrec(p.pID)
				}
				return nil
			}
			return err
		}

		if p.expiry != 0 {
			pub.Expiry = time.Now().Add(time.Duration(p.expiry) * time.Second).Unix()
			p.expiry = 0
		}
	}

	pub.Publisher = ses.clientId

	/*if log.IsLevelEnabled(log.DebugLevel) {
		lf := log.Fields{
			"ClientId":   ses.clientId,
			"Topic Name": string(p.vhBuf[2 : topicLen+2]),
			"QoS":        pub.RxQoS(),
		}
		if pub.Duplicate() {
			lf["duplicate"] = true
		}
		if pub.ToRetain() {
			lf["retain"] = true
		}
		log.WithFields(lf).Debug("PUBLISH received")
	}*/

	switch pub.RxQoS() {
	case 0:
		s.pubs.Add(queue.GetItem(pub))
	case 1:
		s.pubs.Add(queue.GetItem(pub))
		return ses.sendPuback(p.pID)
	case 2: // [MQTT-4.3.3-2]
		if _, ok := ses.client.q2RxLookup[p.pID]; !ok {
			ses.client.q2RxLookup[p.pID] = struct{}{}
			s.pubs.Add(queue.GetItem(pub))
		}
		return ses.sendPubrec(p.pID)
	}

	return nil
}

func (s *session) handlePublishProperties(tx, rx []byte) ([]byte, error) {
	var gotMEI, gotTA, gotRT, gotCD, gotCT bool

	for i := 0; i < len(rx); {
		remain := len(rx) - i
		switch rx[i] {
		case model.PayloadFormatIndicator:
			if remain < 2 {
				s.disconnectReasonCode = model.MalformedPacket
				return nil, errors.New("malformed PUBLISH: bad Payload Format Indicator")
			}
			if rx[i+1] == 1 {
				if err := checkUTF8(s.packet.payload, false); err != nil {
					s.disconnectReasonCode = model.PayloadFormatInvalid
					return nil, errors.New("malformed PUBLISH: payload not valid UTF8")
				}
			}
			tx = append(tx, rx[i:i+2]...)
			i += 2
		case model.MessageExpiryInterval: // just use the last value
			if remain < 5 {
				s.disconnectReasonCode = model.MalformedPacket
				return nil, errors.New("malformed PUBLISH: bad Message Expiry Interval")
			}
			s.packet.expiry = binary.BigEndian.Uint32(rx[i+1:])
			gotMEI = true
			i += 5
		case model.TopicAlias:
			if gotTA {
				s.disconnectReasonCode = model.ProtocolError
				return nil, errors.New("malformed PUBLISH: Topic Alias included more than once")
			}
			if remain < 3 {
				s.disconnectReasonCode = model.MalformedPacket
				return nil, errors.New("malformed PUBLISH: bad Topic Alias")
			}
			s.packet.topicAlias = binary.BigEndian.Uint16(rx[i+1:])
			if s.packet.topicAlias == 0 {
				s.disconnectReasonCode = model.ProtocolError
				return nil, errors.New("malformed PUBLISH: bad Topic Alias")
			}
			gotTA = true
			i += 3
		case model.ResponseTopic:
			if gotRT {
				s.disconnectReasonCode = model.ProtocolError
				return nil, errors.New("malformed PUBLISH: Response Topic included more than once")
			}
			if remain < 3 {
				s.disconnectReasonCode = model.MalformedPacket
				return nil, errors.New("malformed PUBLISH: bad Response Topic")
			}
			l := int(binary.BigEndian.Uint16(rx[i+1:]))
			if remain < 3+l {
				s.disconnectReasonCode = model.MalformedPacket
				return nil, errors.New("malformed PUBLISH: bad Response Topic")
			}
			tx = append(tx, rx[i:i+3+l]...)
			gotRT = true
			i += 3 + l
		case model.CorrelationData:
			if gotCD {
				s.disconnectReasonCode = model.ProtocolError
				return nil, errors.New("malformed PUBLISH: Response Topic included more than once")
			}
			if remain < 3 {
				s.disconnectReasonCode = model.MalformedPacket
				return nil, errors.New("malformed PUBLISH: bad Response Topic")
			}
			l := int(binary.BigEndian.Uint16(rx[i+1:]))
			if remain < 3+l {
				s.disconnectReasonCode = model.MalformedPacket
				return nil, errors.New("malformed PUBLISH: bad Response Topic")
			}
			tx = append(tx, rx[i:i+3+l]...)
			gotCD = true
			i += 3 + l
		case model.UserProperty:
			if remain < 5 {
				s.disconnectReasonCode = model.MalformedPacket
				return nil, errors.New("malformed PUBLISH: bad User Property")
			}
			kLen := int(binary.BigEndian.Uint16(rx[i+1:]))
			if remain < 5+kLen {
				s.disconnectReasonCode = model.MalformedPacket
				return nil, errors.New("malformed PUBLISH: bad User Property")
			}
			vLen := int(binary.BigEndian.Uint16(rx[i+3+kLen:]))
			expect := 5 + kLen + vLen
			if remain < expect {
				s.disconnectReasonCode = model.MalformedPacket
				return nil, errors.New("malformed PUBLISH: bad User Property")
			}
			tx = append(tx, rx[i:i+expect]...)
			i += expect
		case model.SubscriptionIdentifier:
			s.disconnectReasonCode = model.ProtocolError // v5[MQTT-3.3.4-6]
			return nil, errors.New("malformed PUBLISH: Subscription Identifier not allowed from Client")
		case model.ContentType:
			if gotCT {
				s.disconnectReasonCode = model.ProtocolError
				return nil, errors.New("malformed PUBLISH: Response Topic included more than once")
			}
			if remain < 3 {
				s.disconnectReasonCode = model.MalformedPacket
				return nil, errors.New("malformed PUBLISH: bad Response Topic")
			}
			l := int(binary.BigEndian.Uint16(rx[i+1:]))
			if remain < 3+l {
				s.disconnectReasonCode = model.MalformedPacket
				return nil, errors.New("malformed PUBLISH: bad Response Topic")
			}
			tx = append(tx, rx[i:i+3+l]...)
			gotCT = true
			i += 3 + l
		default:
			s.disconnectReasonCode = model.MalformedPacket
			return nil, fmt.Errorf("malformed PUBLISH: unknown property %d (0x%x)", rx[i], rx[i])
		}
	}

	if gotMEI && s.packet.expiry == 0 {
		return nil, errPubExpired
	}

	return tx, nil
}

func (s *Server) handleSubscribe(ses *session) error {
	var subId int

	if ses.protoVersion > 4 {
		var err error
		if subId, err = ses.handleSubscribeProperties(); err != nil {
			return err
		}
	}

	p := ses.packet.payload
	topics, subOps, subRCs := make([][][]byte, 0, 2), make([]uint8, 0, 2), make([]uint8, 0, 2)

	for i := 0; i < len(p); {
		topicL := int(binary.BigEndian.Uint16(p[i:]))
		i += 2
		topicEnd := i + topicL
		subOp := p[topicEnd]

		if ses.protoVersion == 5 {
			if subOp&0xC0 != 0 { // v5[MQTT-3.8.3-5]
				return errors.New("malformed SUBSCRIBE: reserved bits in payload must be 0")
			}

			if retainHandling(subOp) > 2 {
				ses.disconnectReasonCode = model.ProtocolError
				return errors.New("malformed SUBSCRIBE: Retain Handling 3 not allowed")
			}
		} else if subOp&0xFC != 0 { // [MQTT-3-8.3-4]
			return errors.New("malformed SUBSCRIBE: reserved bits in payload must be 0")
		}

		if maxQoS(subOp) == 3 {
			ses.disconnectReasonCode = model.ProtocolError
			return errors.New("malformed SUBSCRIBE: QoS 3 not allowed")
		}

		topic := p[i:topicEnd]
		if err := checkUTF8(topic, false); err != nil { // [MQTT-3.8.3-1]
			return errors.New("malformed SUBSCRIBE: bad Topic Filter: " + err.Error())
		}

		if s.Auther != nil {
			if err := s.Auther.AuthSubscription(ses.clientId, topic); err != nil {
				if ses.protoVersion > 4 {
					subRCs = append(subRCs, model.NotAuthorized)
				} else {
					subRCs = append(subRCs, model.UnspecifiedError)
				}
				i += 1 + topicL
				continue
			}
		}

		topics = append(topics, bytes.Split(topic, []byte{'/'}))
		subOps, subRCs = append(subOps, subOp), append(subRCs, maxQoS(subOp))
		i += 1 + topicL

		if log.IsLevelEnabled(log.DebugLevel) {
			log.WithFields(log.Fields{
				"ClientId":     ses.clientId,
				"Topic Filter": string(topic),
				"QoS":          maxQoS(subOp),
			}).Debug("SUBSCRIBE received")
		}
	}

	if err := s.addSubscriptions(ses.client, topics, subOps, uint32(subId)); err != nil {
		return err
	}

	// [MQTT-3.8.4-1, 4-4, 4-5, 4-6]
	return ses.sendSuback(subRCs)
}

func (s *session) handleSubscribeProperties() (int, error) {
	props := s.packet.vhBuf[2:]
	var subId int
	var gotSubId bool

	for i := 0; i < len(props); {
		remain := len(props) - i
		switch props[i] {
		case model.SubscriptionIdentifier:
			if gotSubId {
				s.disconnectReasonCode = model.ProtocolError
				return 0, errors.New("malformed SUBSCRIBE: Subscription Identifier included more than once")
			}
			if remain < 2 {
				return 0, errors.New("malformed SUBSCRIBE: bad Subscription Identifier")
			}
			sId, l, err := variableLengthDecode(props[i+1:])
			if err != nil {
				return 0, errors.New("malformed SUBSCRIBE: bad Subscription Identifier")
			}
			if sId == 0 {
				s.disconnectReasonCode = model.ProtocolError
				return 0, errors.New("malformed SUBSCRIBE: Subscription Identifier 0 not allowed")
			}
			subId = sId
			gotSubId = true
			i += 1 + l
		case model.UserProperty:
			if remain < 5 {
				s.disconnectReasonCode = model.MalformedPacket
				return 0, errors.New("malformed SUBSCRIBE: bad User Property")
			}

			kLen := int(binary.BigEndian.Uint16(props[i+1:]))
			if remain < 5+kLen {
				s.disconnectReasonCode = model.MalformedPacket
				return 0, errors.New("malformed SUBSCRIBE: bad User Property")
			}

			vLen := int(binary.BigEndian.Uint16(props[i+3+kLen:]))
			expect := 5 + kLen + vLen
			if remain < expect {
				s.disconnectReasonCode = model.MalformedPacket
				return 0, errors.New("malformed SUBSCRIBE: bad User Property")
			}

			i += expect
		default:
			s.disconnectReasonCode = model.MalformedPacket
			return 0, fmt.Errorf("malformed SUBSCRIBE: unknown property %d (0x%x)", props[i], props[i])
		}
	}
	return subId, nil
}

func (s *Server) handleUnsubscribe(ses *session) error {
	p := ses.packet.payload
	topics := make([][][]byte, 0, 2)

	for i := 0; i < len(p); {
		topicL := int(binary.BigEndian.Uint16(p[i:]))
		i += 2
		topicEnd := i + topicL
		topic := p[i:topicEnd]
		if err := checkUTF8(topic, false); err != nil { // [MQTT-3.10.3-1]
			return errors.New("malformed UNSUBSCRIBE: bad Topic Filter: " + err.Error())
		}

		if log.IsLevelEnabled(log.DebugLevel) {
			log.WithFields(log.Fields{
				"ClientId":     ses.clientId,
				"Topic Filter": string(topic),
			}).Debug("UNSUBSCRIBE received")
		}

		topics = append(topics, bytes.Split(topic, []byte{'/'}))
		i += topicEnd
	}

	c := ses.client
	if err := s.removeSubscriptions(c, topics); err != nil {
		return err
	}

	// [MQTT-3.10.4-4, 4-5, 4-6]
	return ses.sendUnsuback(len(topics))
}

func (s *Server) handleUnSubscribeProperties(ses *session) error {
	props := ses.packet.vhBuf[2:]

	for i := 0; i < len(props); {
		remain := len(props) - i
		switch props[i] {
		case model.UserProperty:
			if remain < 5 {
				ses.disconnectReasonCode = model.MalformedPacket
				return errors.New("malformed UNSUBSCRIBE: bad User Property")
			}

			kLen := int(binary.BigEndian.Uint16(props[i+1:]))
			if remain < 5+kLen {
				ses.disconnectReasonCode = model.MalformedPacket
				return errors.New("malformed UNSUBSCRIBE: bad User Property")
			}

			vLen := int(binary.BigEndian.Uint16(props[i+3+kLen:]))
			expect := 5 + kLen + vLen
			if remain < expect {
				ses.disconnectReasonCode = model.MalformedPacket
				return errors.New("malformed UNSUBSCRIBE: bad User Property")
			}

			i += expect
		default:
			ses.disconnectReasonCode = model.MalformedPacket
			return fmt.Errorf("malformed UNSUBSCRIBE: unknown property %d (0x%x)", props[i], props[i])
		}
	}
	return nil
}

func (s *Server) handleDisconnectProperties(ses *session) error {
	props := ses.packet.vhBuf[1:]
	var gotSesExp, gotReasonStr bool

	for i := 0; i < len(props); {
		remain := len(props) - i

		switch props[i] {
		case model.SessionExpiryInterval:
			if gotSesExp {
				ses.disconnectReasonCode = model.ProtocolError
				return errors.New("malformed DISCONNECT: Session Expiry Interval included more than once")
			}
			if remain < 5 {
				ses.disconnectReasonCode = model.MalformedPacket
				return errors.New("malformed DISCONNECT: bad Session Expiry Interval")
			}

			newExp := binary.BigEndian.Uint32(props[i+1:])
			if ses.expiryInterval == 0 && newExp != 0 {
				ses.disconnectReasonCode = model.ProtocolError
				return errors.New("malformed DISCONNECT: bad Session Expiry Interval")
			}

			ses.expiryInterval = newExp
			gotSesExp = true
			i += 5
		case model.ReasonString:
			if gotReasonStr {
				ses.disconnectReasonCode = model.ProtocolError
				return errors.New("malformed DISCONNECT: Reason String included more than once")
			}
			if remain < 3 {
				ses.disconnectReasonCode = model.MalformedPacket
				return errors.New("malformed DISCONNECT: bad Reason String")
			}

			l := int(binary.BigEndian.Uint16(props[i+1:]))
			if remain < 3+l {
				ses.disconnectReasonCode = model.MalformedPacket
				return errors.New("malformed DISCONNECT: bad Reason String")
			}

			gotReasonStr = true
			i += 3 + l
		case model.UserProperty:
			if remain < 5 {
				ses.disconnectReasonCode = model.MalformedPacket
				return errors.New("malformed DISCONNECT: bad User Property")
			}

			kLen := int(binary.BigEndian.Uint16(props[i+1:]))
			if remain < 5+kLen {
				ses.disconnectReasonCode = model.MalformedPacket
				return errors.New("malformed DISCONNECT: bad User Property")
			}

			vLen := int(binary.BigEndian.Uint16(props[i+3+kLen:]))
			expect := 5 + kLen + vLen
			if remain < expect {
				ses.disconnectReasonCode = model.MalformedPacket
				return errors.New("malformed DISCONNECT: bad User Property")
			}

			i += expect
		case model.ServerReference:
			ses.disconnectReasonCode = model.ProtocolError
			return errors.New("malformed DISCONNECT: Server Reference should only be sent by Server")
		default:
			ses.disconnectReasonCode = model.MalformedPacket
			return fmt.Errorf("malformed DISCONNECT: unknown property %d (0x%x)", props[i], props[i])
		}
	}

	return ses.handleDisconnect()
}

var errInvalidUTF = errors.New("invalid UTF8")
var errContainsWildCards = errors.New("contains wildcard characters")

// [MQTT-1.5.3-1] [MQTT-1.5.3-3]
func checkUTF8(str []byte, checkWildCards bool) error {
	for i := 0; i < len(str); {
		if str[i] == 0 { // [MQTT-1.5.3-2]
			return errInvalidUTF
		}

		if checkWildCards && (str[i] == '+' || str[i] == '#') { // [MQTT-3.3.2-2]
			return errContainsWildCards
		} else if str[i]&0x80 == 0 {
			i++
		} else {
			r, size := utf8.DecodeRune(str[i:])
			if r == utf8.RuneError {
				if size != 1 {
					return nil
				} else {
					return errInvalidUTF
				}
			}
			i += size
		}
	}
	return nil
}

// returns decoded length, as well as the number of Variable Length bytes.
func variableLengthDecode(b []byte) (int, int, error) {
	var l, i int
	var lMul int = 1
	var err error

	max := len(b) - 1
	for done := false; !done; i++ {
		if i > max {
			err = errors.New("bad Variable Length")
			break
		}

		l += int(b[i]&127) * lMul
		if lMul > maxVarLenMul {
			err = errors.New("bad Variable Length")
			break
		}

		lMul *= 128
		done = b[i]&128 == 0
	}
	return l, i, err
}