nats-server/server/parser.go

// Copyright 2012-2019 The NATS 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.

package server

import (
	"fmt"
)

type pubArg struct {
	arg     []byte
	pacache []byte
	account []byte
	subject []byte
	reply   []byte
	szb     []byte
	queues  [][]byte
	size    int
}

type parserState int
type parseState struct {
	state   parserState
	as      int
	drop    int
	pa      pubArg
	argBuf  []byte
	msgBuf  []byte
	scratch [MAX_CONTROL_LINE_SIZE]byte
}

// Parser constants
const (
	OP_START parserState = iota
	OP_PLUS
	OP_PLUS_O
	OP_PLUS_OK
	OP_MINUS
	OP_MINUS_E
	OP_MINUS_ER
	OP_MINUS_ERR
	OP_MINUS_ERR_SPC
	MINUS_ERR_ARG
	OP_C
	OP_CO
	OP_CON
	OP_CONN
	OP_CONNE
	OP_CONNEC
	OP_CONNECT
	CONNECT_ARG
	OP_P
	OP_PU
	OP_PUB
	OP_PUB_SPC
	PUB_ARG
	OP_PI
	OP_PIN
	OP_PING
	OP_PO
	OP_PON
	OP_PONG
	MSG_PAYLOAD
	MSG_END_R
	MSG_END_N
	OP_S
	OP_SU
	OP_SUB
	OP_SUB_SPC
	SUB_ARG
	OP_A
	OP_ASUB
	OP_ASUB_SPC
	ASUB_ARG
	OP_AUSUB
	OP_AUSUB_SPC
	AUSUB_ARG
	OP_L
	OP_LS
	OP_R
	OP_RS
	OP_U
	OP_UN
	OP_UNS
	OP_UNSU
	OP_UNSUB
	OP_UNSUB_SPC
	UNSUB_ARG
	OP_M
	OP_MS
	OP_MSG
	OP_MSG_SPC
	MSG_ARG
	OP_I
	OP_IN
	OP_INF
	OP_INFO
	INFO_ARG
)

func (c *client) parse(buf []byte) error {
	var i int
	var b byte

	// Snapshots
	c.mu.Lock()
	// Snapshot and then reset when we receive a
	// proper CONNECT if needed.
	authSet := c.awaitingAuth()
	// Snapshot max control line as well.
	mcl := c.mcl
	c.mu.Unlock()

	// Move to loop instead of range syntax to allow jumping of i
	for i = 0; i < len(buf); i++ {
		b = buf[i]

		switch c.state {
		case OP_START:
			if b != 'C' && b != 'c' {
				if authSet {
					goto authErr
				}
				// If the connection is a gateway connection, make sure that
				// if this is an inbound, it starts with a CONNECT.
				if c.kind == GATEWAY && !c.gw.outbound && !c.gw.connected {
					// Use auth violation since no CONNECT was sent.
					// It could be a parseErr too.
					goto authErr
				}
			}
			switch b {
			case 'P', 'p':
				c.state = OP_P
			case 'S', 's':
				c.state = OP_S
			case 'U', 'u':
				c.state = OP_U
			case 'R', 'r':
				if c.kind == CLIENT {
					goto parseErr
				} else {
					c.state = OP_R
				}
			case 'L', 'l':
				if c.kind != LEAF {
					goto parseErr
				} else {
					c.state = OP_L
				}
			case 'A', 'a':
				if c.kind == CLIENT {
					goto parseErr
				} else {
					c.state = OP_A
				}
			case 'C', 'c':
				c.state = OP_C
			case 'I', 'i':
				c.state = OP_I
			case '+':
				c.state = OP_PLUS
			case '-':
				c.state = OP_MINUS
			default:
				goto parseErr
			}
		case OP_P:
			switch b {
			case 'U', 'u':
				c.state = OP_PU
			case 'I', 'i':
				c.state = OP_PI
			case 'O', 'o':
				c.state = OP_PO
			default:
				goto parseErr
			}
		case OP_PU:
			switch b {
			case 'B', 'b':
				c.state = OP_PUB
			default:
				goto parseErr
			}
		case OP_PUB:
			switch b {
			case ' ', '\t':
				c.state = OP_PUB_SPC
			default:
				goto parseErr
			}
		case OP_PUB_SPC:
			switch b {
			case ' ', '\t':
				continue
			default:
				c.state = PUB_ARG
				c.as = i
			}
		case PUB_ARG:
			switch b {
			case '\r':
				c.drop = 1
			case '\n':
				var arg []byte
				if c.argBuf != nil {
					arg = c.argBuf
					c.argBuf = nil
				} else {
					arg = buf[c.as : i-c.drop]
				}
				if err := c.processPub(c.trace, arg); err != nil {
					return err
				}
				c.drop, c.as, c.state = 0, i+1, MSG_PAYLOAD
				// If we don't have a saved buffer then jump ahead with
				// the index. If this overruns what is left we fall out
				// and process split buffer.
				if c.msgBuf == nil {
					i = c.as + c.pa.size - LEN_CR_LF
				}
			default:
				if c.argBuf != nil {
					c.argBuf = append(c.argBuf, b)
				}
			}
		case MSG_PAYLOAD:
			if c.msgBuf != nil {
				// copy as much as we can to the buffer and skip ahead.
				toCopy := c.pa.size - len(c.msgBuf)
				avail := len(buf) - i
				if avail < toCopy {
					toCopy = avail
				}
				if toCopy > 0 {
					start := len(c.msgBuf)
					// This is needed for copy to work.
					c.msgBuf = c.msgBuf[:start+toCopy]
					copy(c.msgBuf[start:], buf[i:i+toCopy])
					// Update our index
					i = (i + toCopy) - 1
				} else {
					// Fall back to append if needed.
					c.msgBuf = append(c.msgBuf, b)
				}
				if len(c.msgBuf) >= c.pa.size {
					c.state = MSG_END_R
				}
			} else if i-c.as+1 >= c.pa.size {
				c.state = MSG_END_R
			}
		case MSG_END_R:
			if b != '\r' {
				goto parseErr
			}
			if c.msgBuf != nil {
				c.msgBuf = append(c.msgBuf, b)
			}
			c.state = MSG_END_N
		case MSG_END_N:
			if b != '\n' {
				goto parseErr
			}
			if c.msgBuf != nil {
				c.msgBuf = append(c.msgBuf, b)
			} else {
				c.msgBuf = buf[c.as : i+1]
			}
			c.processInboundMsg(c.msgBuf)
			c.argBuf, c.msgBuf = nil, nil
			c.drop, c.as, c.state = 0, i+1, OP_START
			// Drop all pub args
			c.pa.arg, c.pa.pacache, c.pa.account, c.pa.subject = nil, nil, nil, nil
			c.pa.reply, c.pa.size, c.pa.szb, c.pa.queues = nil, 0, nil, nil
		case OP_A:
			switch b {
			case '+':
				c.state = OP_ASUB
			case '-', 'u':
				c.state = OP_AUSUB
			default:
				goto parseErr
			}
		case OP_ASUB:
			switch b {
			case ' ', '\t':
				c.state = OP_ASUB_SPC
			default:
				goto parseErr
			}
		case OP_ASUB_SPC:
			switch b {
			case ' ', '\t':
				continue
			default:
				c.state = ASUB_ARG
				c.as = i
			}
		case ASUB_ARG:
			switch b {
			case '\r':
				c.drop = 1
			case '\n':
				var arg []byte
				if c.argBuf != nil {
					arg = c.argBuf
					c.argBuf = nil
				} else {
					arg = buf[c.as : i-c.drop]
				}
				if err := c.processAccountSub(arg); err != nil {
					return err
				}
				c.drop, c.as, c.state = 0, i+1, OP_START
			default:
				if c.argBuf != nil {
					c.argBuf = append(c.argBuf, b)
				}
			}
		case OP_AUSUB:
			switch b {
			case ' ', '\t':
				c.state = OP_AUSUB_SPC
			default:
				goto parseErr
			}
		case OP_AUSUB_SPC:
			switch b {
			case ' ', '\t':
				continue
			default:
				c.state = AUSUB_ARG
				c.as = i
			}
		case AUSUB_ARG:
			switch b {
			case '\r':
				c.drop = 1
			case '\n':
				var arg []byte
				if c.argBuf != nil {
					arg = c.argBuf
					c.argBuf = nil
				} else {
					arg = buf[c.as : i-c.drop]
				}
				c.processAccountUnsub(arg)
				c.drop, c.as, c.state = 0, i+1, OP_START
			default:
				if c.argBuf != nil {
					c.argBuf = append(c.argBuf, b)
				}
			}
		case OP_S:
			switch b {
			case 'U', 'u':
				c.state = OP_SU
			default:
				goto parseErr
			}
		case OP_SU:
			switch b {
			case 'B', 'b':
				c.state = OP_SUB
			default:
				goto parseErr
			}
		case OP_SUB:
			switch b {
			case ' ', '\t':
				c.state = OP_SUB_SPC
			default:
				goto parseErr
			}
		case OP_SUB_SPC:
			switch b {
			case ' ', '\t':
				continue
			default:
				c.state = SUB_ARG
				c.as = i
			}
		case SUB_ARG:
			switch b {
			case '\r':
				c.drop = 1
			case '\n':
				var arg []byte
				if c.argBuf != nil {
					arg = c.argBuf
					c.argBuf = nil
				} else {
					arg = buf[c.as : i-c.drop]
				}
				var err error

				switch c.kind {
				case CLIENT:
					_, err = c.processSub(arg, false)
				case ROUTER:
					err = c.processRemoteSub(arg)
				case GATEWAY:
					err = c.processGatewayRSub(arg)
				case LEAF:
					err = c.processLeafSub(arg)
				}
				if err != nil {
					return err
				}
				c.drop, c.as, c.state = 0, i+1, OP_START
			default:
				if c.argBuf != nil {
					c.argBuf = append(c.argBuf, b)
				}
			}
		case OP_L:
			switch b {
			case 'S', 's':
				c.state = OP_LS
			case 'M', 'm':
				c.state = OP_M
			default:
				goto parseErr
			}
		case OP_LS:
			switch b {
			case '+':
				c.state = OP_SUB
			case '-':
				c.state = OP_UNSUB
			default:
				goto parseErr
			}
		case OP_R:
			switch b {
			case 'S', 's':
				c.state = OP_RS
			case 'M', 'm':
				c.state = OP_M
			default:
				goto parseErr
			}
		case OP_RS:
			switch b {
			case '+':
				c.state = OP_SUB
			case '-':
				c.state = OP_UNSUB
			default:
				goto parseErr
			}
		case OP_U:
			switch b {
			case 'N', 'n':
				c.state = OP_UN
			default:
				goto parseErr
			}
		case OP_UN:
			switch b {
			case 'S', 's':
				c.state = OP_UNS
			default:
				goto parseErr
			}
		case OP_UNS:
			switch b {
			case 'U', 'u':
				c.state = OP_UNSU
			default:
				goto parseErr
			}
		case OP_UNSU:
			switch b {
			case 'B', 'b':
				c.state = OP_UNSUB
			default:
				goto parseErr
			}
		case OP_UNSUB:
			switch b {
			case ' ', '\t':
				c.state = OP_UNSUB_SPC
			default:
				goto parseErr
			}
		case OP_UNSUB_SPC:
			switch b {
			case ' ', '\t':
				continue
			default:
				c.state = UNSUB_ARG
				c.as = i
			}
		case UNSUB_ARG:
			switch b {
			case '\r':
				c.drop = 1
			case '\n':
				var arg []byte
				if c.argBuf != nil {
					arg = c.argBuf
					c.argBuf = nil
				} else {
					arg = buf[c.as : i-c.drop]
				}
				var err error

				switch c.kind {
				case CLIENT:
					err = c.processUnsub(arg)
				case ROUTER:
					err = c.processRemoteUnsub(arg)
				case GATEWAY:
					err = c.processGatewayRUnsub(arg)
				case LEAF:
					err = c.processLeafUnsub(arg)
				}
				if err != nil {
					return err
				}
				c.drop, c.as, c.state = 0, i+1, OP_START
			default:
				if c.argBuf != nil {
					c.argBuf = append(c.argBuf, b)
				}
			}
		case OP_PI:
			switch b {
			case 'N', 'n':
				c.state = OP_PIN
			default:
				goto parseErr
			}
		case OP_PIN:
			switch b {
			case 'G', 'g':
				c.state = OP_PING
			default:
				goto parseErr
			}
		case OP_PING:
			switch b {
			case '\n':
				c.processPing()
				c.drop, c.state = 0, OP_START
			}
		case OP_PO:
			switch b {
			case 'N', 'n':
				c.state = OP_PON
			default:
				goto parseErr
			}
		case OP_PON:
			switch b {
			case 'G', 'g':
				c.state = OP_PONG
			default:
				goto parseErr
			}
		case OP_PONG:
			switch b {
			case '\n':
				c.processPong()
				c.drop, c.state = 0, OP_START
			}
		case OP_C:
			switch b {
			case 'O', 'o':
				c.state = OP_CO
			default:
				goto parseErr
			}
		case OP_CO:
			switch b {
			case 'N', 'n':
				c.state = OP_CON
			default:
				goto parseErr
			}
		case OP_CON:
			switch b {
			case 'N', 'n':
				c.state = OP_CONN
			default:
				goto parseErr
			}
		case OP_CONN:
			switch b {
			case 'E', 'e':
				c.state = OP_CONNE
			default:
				goto parseErr
			}
		case OP_CONNE:
			switch b {
			case 'C', 'c':
				c.state = OP_CONNEC
			default:
				goto parseErr
			}
		case OP_CONNEC:
			switch b {
			case 'T', 't':
				c.state = OP_CONNECT
			default:
				goto parseErr
			}
		case OP_CONNECT:
			switch b {
			case ' ', '\t':
				continue
			default:
				c.state = CONNECT_ARG
				c.as = i
			}
		case CONNECT_ARG:
			switch b {
			case '\r':
				c.drop = 1
			case '\n':
				var arg []byte
				if c.argBuf != nil {
					arg = c.argBuf
					c.argBuf = nil
				} else {
					arg = buf[c.as : i-c.drop]
				}
				if err := c.processConnect(arg); err != nil {
					return err
				}
				c.drop, c.state = 0, OP_START
				// Reset notion on authSet
				c.mu.Lock()
				authSet = c.awaitingAuth()
				c.mu.Unlock()
			default:
				if c.argBuf != nil {
					c.argBuf = append(c.argBuf, b)
				}
			}
		case OP_M:
			switch b {
			case 'S', 's':
				c.state = OP_MS
			default:
				goto parseErr
			}
		case OP_MS:
			switch b {
			case 'G', 'g':
				c.state = OP_MSG
			default:
				goto parseErr
			}
		case OP_MSG:
			switch b {
			case ' ', '\t':
				c.state = OP_MSG_SPC
			default:
				goto parseErr
			}
		case OP_MSG_SPC:
			switch b {
			case ' ', '\t':
				continue
			default:
				c.state = MSG_ARG
				c.as = i
			}
		case MSG_ARG:
			switch b {
			case '\r':
				c.drop = 1
			case '\n':
				var arg []byte
				if c.argBuf != nil {
					arg = c.argBuf
					c.argBuf = nil
				} else {
					arg = buf[c.as : i-c.drop]
				}
				var err error
				if c.kind == ROUTER || c.kind == GATEWAY {
					err = c.processRoutedMsgArgs(c.trace, arg)
				} else if c.kind == LEAF {
					err = c.processLeafMsgArgs(c.trace, arg)
				}
				if err != nil {
					return err
				}
				c.drop, c.as, c.state = 0, i+1, MSG_PAYLOAD

				// jump ahead with the index. If this overruns
				// what is left we fall out and process split
				// buffer.
				i = c.as + c.pa.size - LEN_CR_LF
			default:
				if c.argBuf != nil {
					c.argBuf = append(c.argBuf, b)
				}
			}
		case OP_I:
			switch b {
			case 'N', 'n':
				c.state = OP_IN
			default:
				goto parseErr
			}
		case OP_IN:
			switch b {
			case 'F', 'f':
				c.state = OP_INF
			default:
				goto parseErr
			}
		case OP_INF:
			switch b {
			case 'O', 'o':
				c.state = OP_INFO
			default:
				goto parseErr
			}
		case OP_INFO:
			switch b {
			case ' ', '\t':
				continue
			default:
				c.state = INFO_ARG
				c.as = i
			}
		case INFO_ARG:
			switch b {
			case '\r':
				c.drop = 1
			case '\n':
				var arg []byte
				if c.argBuf != nil {
					arg = c.argBuf
					c.argBuf = nil
				} else {
					arg = buf[c.as : i-c.drop]
				}
				if err := c.processInfo(arg); err != nil {
					return err
				}
				c.drop, c.as, c.state = 0, i+1, OP_START
			default:
				if c.argBuf != nil {
					c.argBuf = append(c.argBuf, b)
				}
			}
		case OP_PLUS:
			switch b {
			case 'O', 'o':
				c.state = OP_PLUS_O
			default:
				goto parseErr
			}
		case OP_PLUS_O:
			switch b {
			case 'K', 'k':
				c.state = OP_PLUS_OK
			default:
				goto parseErr
			}
		case OP_PLUS_OK:
			switch b {
			case '\n':
				c.drop, c.state = 0, OP_START
			}
		case OP_MINUS:
			switch b {
			case 'E', 'e':
				c.state = OP_MINUS_E
			default:
				goto parseErr
			}
		case OP_MINUS_E:
			switch b {
			case 'R', 'r':
				c.state = OP_MINUS_ER
			default:
				goto parseErr
			}
		case OP_MINUS_ER:
			switch b {
			case 'R', 'r':
				c.state = OP_MINUS_ERR
			default:
				goto parseErr
			}
		case OP_MINUS_ERR:
			switch b {
			case ' ', '\t':
				c.state = OP_MINUS_ERR_SPC
			default:
				goto parseErr
			}
		case OP_MINUS_ERR_SPC:
			switch b {
			case ' ', '\t':
				continue
			default:
				c.state = MINUS_ERR_ARG
				c.as = i
			}
		case MINUS_ERR_ARG:
			switch b {
			case '\r':
				c.drop = 1
			case '\n':
				var arg []byte
				if c.argBuf != nil {
					arg = c.argBuf
					c.argBuf = nil
				} else {
					arg = buf[c.as : i-c.drop]
				}
				c.processErr(string(arg))
				c.drop, c.as, c.state = 0, i+1, OP_START
			default:
				if c.argBuf != nil {
					c.argBuf = append(c.argBuf, b)
				}
			}
		default:
			goto parseErr
		}
	}

	// Check for split buffer scenarios for any ARG state.
	if c.state == SUB_ARG || c.state == UNSUB_ARG || c.state == PUB_ARG ||
		c.state == ASUB_ARG || c.state == AUSUB_ARG ||
		c.state == MSG_ARG || c.state == MINUS_ERR_ARG ||
		c.state == CONNECT_ARG || c.state == INFO_ARG {
		// Setup a holder buffer to deal with split buffer scenario.
		if c.argBuf == nil {
			c.argBuf = c.scratch[:0]
			c.argBuf = append(c.argBuf, buf[c.as:i-c.drop]...)
		}
		// Check for violations of control line length here. Note that this is not
		// exact at all but the performance hit is too great to be precise, and
		// catching here should prevent memory exhaustion attacks.
		if len(c.argBuf) > int(mcl) {
			c.sendErr("Maximum Control Line Exceeded")
			c.closeConnection(MaxControlLineExceeded)
			return ErrMaxControlLine
		}
	}

	// Check for split msg
	if (c.state == MSG_PAYLOAD || c.state == MSG_END_R || c.state == MSG_END_N) && c.msgBuf == nil {
		// We need to clone the pubArg if it is still referencing the
		// read buffer and we are not able to process the msg.
		if c.argBuf == nil {
			// Works also for MSG_ARG, when message comes from ROUTE.
			if err := c.clonePubArg(); err != nil {
				goto parseErr
			}
		}

		// If we will overflow the scratch buffer, just create a
		// new buffer to hold the split message.
		if c.pa.size > cap(c.scratch)-len(c.argBuf) {
			lrem := len(buf[c.as:])

			// Consider it a protocol error when the remaining payload
			// is larger than the reported size for PUB. It can happen
			// when processing incomplete messages from rogue clients.
			if lrem > c.pa.size+LEN_CR_LF {
				goto parseErr
			}
			c.msgBuf = make([]byte, lrem, c.pa.size+LEN_CR_LF)
			copy(c.msgBuf, buf[c.as:])
		} else {
			c.msgBuf = c.scratch[len(c.argBuf):len(c.argBuf)]
			c.msgBuf = append(c.msgBuf, (buf[c.as:])...)
		}
	}

	return nil

authErr:
	c.authViolation()
	return ErrAuthentication

parseErr:
	c.sendErr("Unknown Protocol Operation")
	snip := protoSnippet(i, buf)
	err := fmt.Errorf("%s parser ERROR, state=%d, i=%d: proto='%s...'",
		c.typeString(), c.state, i, snip)
	return err
}

func protoSnippet(start int, buf []byte) string {
	stop := start + PROTO_SNIPPET_SIZE
	bufSize := len(buf)
	if start >= bufSize {
		return `""`
	}
	if stop > bufSize {
		stop = bufSize - 1
	}
	return fmt.Sprintf("%q", buf[start:stop])
}

// clonePubArg is used when the split buffer scenario has the pubArg in the existing read buffer, but
// we need to hold onto it into the next read.
func (c *client) clonePubArg() error {
	// Just copy and re-process original arg buffer.
	c.argBuf = c.scratch[:0]
	c.argBuf = append(c.argBuf, c.pa.arg...)

	switch c.kind {
	case ROUTER, GATEWAY:
		return c.processRoutedMsgArgs(false, c.argBuf)
	case LEAF:
		return c.processLeafMsgArgs(false, c.argBuf)
	default:
		return c.processPub(false, c.argBuf)
	}
}