func (p *Parser) OnData(reply, endStream bool, dataBuffers [][]byte) (proxylib.OpType, int) {
	if reply {
		if p.injectFromQueue() {
			return proxylib.INJECT, len(DeniedMsgBase)
		}
		if len(dataBuffers) == 0 {
			return proxylib.NOP, 0
		}
	}

	//TODO don't copy data from buffers
	data := bytes.Join(dataBuffers, []byte{})
	log.Debugf("Data length: %d", len(data))

	if headerSize > len(data) {
		headerMissing := headerSize - len(data)
		log.Debugf("Did not receive needed header data, need %d more bytes", headerMissing)
		return proxylib.MORE, headerMissing
	}

	bodyLength := binary.BigEndian.Uint32(data[8:12])

	keyLength := binary.BigEndian.Uint16(data[2:4])
	extrasLength := data[4]

	if keyLength > 0 {
		neededData := headerSize + int(keyLength) + int(extrasLength)
		if neededData > len(data) {
			keyMissing := neededData - len(data)
			log.Debugf("Did not receive enough bytes for key, need %d more bytes", keyMissing)
			return proxylib.MORE, keyMissing
		}
	}

	opcode, key, err := p.getOpcodeAndKey(data, extrasLength, keyLength)
	if err != 0 {
		return proxylib.ERROR, int(err)
	}

	logEntry := &cilium.LogEntry_GenericL7{
		GenericL7: &cilium.L7LogEntry{
			Proto: "binarymemcached",
			Fields: map[string]string{
				"opcode": strconv.Itoa(int(opcode)),
				"key":    string(key),
			},
		},
	}

	// we don't filter reply traffic
	if reply {
		log.Debugf("reply, passing %d bytes", len(data))
		p.connection.Log(cilium.EntryType_Response, logEntry)
		p.replyCount++
		return proxylib.PASS, int(bodyLength + headerSize)
	}

	p.requestCount++

	matches := p.connection.Matches(meta.MemcacheMeta{
		Opcode: opcode,
		Keys:   [][]byte{key},
	})
	if matches {
		p.connection.Log(cilium.EntryType_Request, logEntry)
		return proxylib.PASS, int(bodyLength + headerSize)
	}

	magic := ResponseMagic | data[0]

	// This is done to ensure in-order replies
	if p.requestCount == p.replyCount+1 {
		p.injectDeniedMessage(magic)
	} else {
		p.injectQueue = append(p.injectQueue, queuedInject{magic, p.requestCount})
	}

	p.injectQueue = append(p.injectQueue, queuedInject{magic, p.requestCount})

	p.connection.Log(cilium.EntryType_Denied, logEntry)
	return proxylib.DROP, int(bodyLength + headerSize)
}
// OnData parses binary memcached data
OnData parses binary memcached data
// OnData parses binary memcached data
func (p *Parser) OnData(reply, endStream bool, dataBuffers [][]byte) (proxylib.OpType, int) {
	if reply {
		if p.injectFromQueue() {
			return proxylib.INJECT, len(DeniedMsgBase)
		}
		if len(dataBuffers) == 0 {
			return proxylib.NOP, 0
		}
	}
	data := bytes.Join(dataBuffers, []byte{})
	log.Debugf("Data length: %d", len(data))
	if headerSize > len(data) {
		headerMissing := headerSize - len(data)
		log.Debugf("Did not receive needed header data, need %d more bytes", headerMissing)
		return proxylib.MORE, headerMissing
	}
	bodyLength := binary.BigEndian.Uint32(data[8:12])
	keyLength := binary.BigEndian.Uint16(data[2:4])
	extrasLength := data[4]
	if keyLength > 0 {
		neededData := headerSize + int(keyLength) + int(extrasLength)
		if neededData > len(data) {
			keyMissing := neededData - len(data)
			log.Debugf("Did not receive enough bytes for key, need %d more bytes", keyMissing)
			return proxylib.MORE, keyMissing
		}
	}
	opcode, key, err := p.getOpcodeAndKey(data, extrasLength, keyLength)
	if err != 0 {
		return proxylib.ERROR, int(err)
	}
	logEntry := &cilium.LogEntry_GenericL7{
		GenericL7: &cilium.L7LogEntry{
			Proto: "binarymemcached",
			Fields: map[string]string{
				"opcode": strconv.Itoa(int(opcode)),
				"key":    string(key),
			},
		},
	}
	if reply {
		log.Debugf("reply, passing %d bytes", len(data))
		p.connection.Log(cilium.EntryType_Response, logEntry)
		p.replyCount++
		return proxylib.PASS, int(bodyLength + headerSize)
	}
	p.requestCount++
	matches := p.connection.Matches(meta.MemcacheMeta{
		Opcode: opcode,
		Keys:   [][]byte{key},
	})
	if matches {
		p.connection.Log(cilium.EntryType_Request, logEntry)
		return proxylib.PASS, int(bodyLength + headerSize)
	}
	magic := ResponseMagic | data[0]
	if p.requestCount == p.replyCount+1 {
		p.injectDeniedMessage(magic)
	} else {
		p.injectQueue = append(p.injectQueue, queuedInject{magic, p.requestCount})
	}
	p.injectQueue = append(p.injectQueue, queuedInject{magic, p.requestCount})
	p.connection.Log(cilium.EntryType_Denied, logEntry)
	return proxylib.DROP, int(bodyLength + headerSize)
}
func ( p * Parser ) OnData ( reply , endStream bool , dataBuffers [ ] [ ] byte ) ( proxylib . OpType , int ) { if reply { if p . injectFromQueue ( ) { return proxylib . INJECT , len ( DeniedMsgBase ) } if len ( dataBuffers ) == 0 { return proxylib . NOP , 0 } } data := bytes . Join ( dataBuffers , [ ] byte { } ) log . Debugf ( " " , len ( data ) ) if headerSize > len ( data ) { headerMissing := headerSize - len ( data ) log . Debugf ( " " , headerMissing ) return proxylib . MORE , headerMissing } bodyLength := binary . BigEndian . Uint32 ( data [ 8 : 12 ] ) keyLength := binary . BigEndian . Uint16 ( data [ 2 : 4 ] ) extrasLength := data [ 4 ] if keyLength > 0 { neededData := headerSize + int ( keyLength ) + int ( extrasLength ) if neededData > len ( data ) { keyMissing := neededData - len ( data ) log . Debugf ( " " , keyMissing ) return proxylib . MORE , keyMissing } } opcode , key , err := p . getOpcodeAndKey ( data , extrasLength , keyLength ) if err != 0 { return proxylib . ERROR , int ( err ) } logEntry := & cilium . LogEntry_GenericL7 { GenericL7 : & cilium . L7LogEntry { Proto : " " , Fields : map [ string ] string { " " : strconv . Itoa ( int ( opcode ) ) , " " : string ( key ) , } , } , } if reply { log . Debugf ( " " , len ( data ) ) p . connection . Log ( cilium . EntryType_Response , logEntry ) p . replyCount ++ return proxylib . PASS , int ( bodyLength + headerSize ) } p . requestCount ++ matches := p . connection . Matches ( meta . MemcacheMeta { Opcode : opcode , Keys : [ ] [ ] byte { key } , } ) if matches { p . connection . Log ( cilium . EntryType_Request , logEntry ) return proxylib . PASS , int ( bodyLength + headerSize ) } magic := ResponseMagic | data [ 0 ] if p . requestCount == p . replyCount + 1 { p . injectDeniedMessage ( magic ) } else { p . injectQueue = append ( p . injectQueue , queuedInject { magic , p . requestCount } ) } p . injectQueue = append ( p . injectQueue , queuedInject { magic , p . requestCount } ) p . connection . Log ( cilium . EntryType_Denied , logEntry ) return proxylib . DROP , int ( bodyLength + headerSize ) }
OnData parses binary memcached data
OnData parses binary memcached data
//TODO don't copy data from buffers // we don't filter reply traffic // This is done to ensure in-order replies