th5/record_process.go

// Copyright 2009 The Go Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.

package tls

// A recordProcessor accepts reassembled records, decrypts and verifies them
// and routes them either to the handshake processor, to up to the application.
// It also accepts requests from the application for the current connection
// state, or for a notification when the state changes.

import (
	"bytes";
	"container/list";
	"crypto/subtle";
	"hash";
)

// getConnectionState is a request from the application to get the current
// ConnectionState.
type getConnectionState struct {
	reply chan<- ConnectionState;
}

// waitConnectionState is a request from the application to be notified when
// the connection state changes.
type waitConnectionState struct {
	reply chan<- ConnectionState;
}

// connectionStateChange is a message from the handshake processor that the
// connection state has changed.
type connectionStateChange struct {
	connState ConnectionState;
}

// changeCipherSpec is a message send to the handshake processor to signal that
// the peer is switching ciphers.
type changeCipherSpec struct{}

// newCipherSpec is a message from the handshake processor that future
// records should be processed with a new cipher and MAC function.
type newCipherSpec struct {
	encrypt	encryptor;
	mac	hash.Hash;
}

type recordProcessor struct {
	decrypt		encryptor;
	mac		hash.Hash;
	seqNum		uint64;
	handshakeBuf	[]byte;
	appDataChan	chan<- []byte;
	requestChan	<-chan interface{};
	controlChan	<-chan interface{};
	recordChan	<-chan *record;
	handshakeChan	chan<- interface{};

	// recordRead is nil when we don't wish to read any more.
	recordRead	<-chan *record;
	// appDataSend is nil when len(appData) == 0.
	appDataSend	chan<- []byte;
	// appData contains any application data queued for upstream.
	appData	[]byte;
	// A list of channels waiting for connState to change.
	waitQueue	*list.List;
	connState	ConnectionState;
	shutdown	bool;
	header		[13]byte;
}

// drainRequestChannel processes messages from the request channel until it's closed.
func drainRequestChannel(requestChan <-chan interface{}, c ConnectionState) {
	for v := range requestChan {
		if closed(requestChan) {
			return;
		}
		switch r := v.(type) {
		case getConnectionState:
			r.reply <- c;
		case waitConnectionState:
			r.reply <- c;
		}
	}
}

func (p *recordProcessor) loop(appDataChan chan<- []byte, requestChan <-chan interface{}, controlChan <-chan interface{}, recordChan <-chan *record, handshakeChan chan<- interface{}) {
	noop := nop{};
	p.decrypt = noop;
	p.mac = noop;
	p.waitQueue = list.New();

	p.appDataChan = appDataChan;
	p.requestChan = requestChan;
	p.controlChan = controlChan;
	p.recordChan = recordChan;
	p.handshakeChan = handshakeChan;
	p.recordRead = recordChan;

	for !p.shutdown {
		select {
		case p.appDataSend <- p.appData:
			p.appData = nil;
			p.appDataSend = nil;
			p.recordRead = p.recordChan;
		case c := <-controlChan:
			p.processControlMsg(c);
		case r := <-requestChan:
			p.processRequestMsg(r);
		case r := <-p.recordRead:
			p.processRecord(r);
		}
	}

	p.wakeWaiters();
	go drainRequestChannel(p.requestChan, p.connState);
	go func() {
		for _ = range controlChan {
		}
	}();

	close(handshakeChan);
	if len(p.appData) > 0 {
		appDataChan <- p.appData;
	}
	close(appDataChan);
}

func (p *recordProcessor) processRequestMsg(requestMsg interface{}) {
	if closed(p.requestChan) {
		p.shutdown = true;
		return;
	}

	switch r := requestMsg.(type) {
	case getConnectionState:
		r.reply <- p.connState;
	case waitConnectionState:
		if p.connState.HandshakeComplete {
			r.reply <- p.connState;
		}
		p.waitQueue.PushBack(r.reply);
	}
}

func (p *recordProcessor) processControlMsg(msg interface{}) {
	connState, ok := msg.(ConnectionState);
	if !ok || closed(p.controlChan) {
		p.shutdown = true;
		return;
	}

	p.connState = connState;
	p.wakeWaiters();
}

func (p *recordProcessor) wakeWaiters() {
	for i := p.waitQueue.Front(); i != nil; i = i.Next() {
		i.Value.(chan<- ConnectionState) <- p.connState;
	}
	p.waitQueue.Init();
}

func (p *recordProcessor) processRecord(r *record) {
	if closed(p.recordChan) {
		p.shutdown = true;
		return;
	}

	p.decrypt.XORKeyStream(r.payload);
	if len(r.payload) < p.mac.Size() {
		p.error(alertBadRecordMAC);
		return;
	}

	fillMACHeader(&p.header, p.seqNum, len(r.payload) - p.mac.Size(), r);
	p.seqNum++;

	p.mac.Reset();
	p.mac.Write(p.header[0:13]);
	p.mac.Write(r.payload[0 : len(r.payload) - p.mac.Size()]);
	macBytes := p.mac.Sum();

	if subtle.ConstantTimeCompare(macBytes, r.payload[len(r.payload) - p.mac.Size() : len(r.payload)]) != 1 {
		p.error(alertBadRecordMAC);
		return;
	}

	switch r.contentType {
	case recordTypeHandshake:
		p.processHandshakeRecord(r.payload[0 : len(r.payload) - p.mac.Size()]);
	case recordTypeChangeCipherSpec:
		if len(r.payload) != 1 || r.payload[0] != 1 {
			p.error(alertUnexpectedMessage);
			return;
		}

		p.handshakeChan <- changeCipherSpec{};
		newSpec, ok := (<-p.controlChan).(*newCipherSpec);
		if !ok {
			p.connState.Error = alertUnexpectedMessage;
			p.shutdown = true;
			return;
		}
		p.decrypt = newSpec.encrypt;
		p.mac = newSpec.mac;
		p.seqNum = 0;
	case recordTypeApplicationData:
		if p.connState.HandshakeComplete == false {
			p.error(alertUnexpectedMessage);
			return;
		}
		p.recordRead = nil;
		p.appData = r.payload;
		p.appDataSend = p.appDataChan;
	default:
		p.error(alertUnexpectedMessage);
		return;
	}
}

func (p *recordProcessor) processHandshakeRecord(data []byte) {
	if p.handshakeBuf == nil {
		p.handshakeBuf = data;
	} else {
		if len(p.handshakeBuf) > maxHandshakeMsg {
			p.error(alertInternalError);
			return;
		}
		newBuf := make([]byte, len(p.handshakeBuf)+len(data));
		bytes.Copy(newBuf, p.handshakeBuf);
		bytes.Copy(newBuf[len(p.handshakeBuf):len(newBuf)], data);
		p.handshakeBuf = newBuf;
	}

	for len(p.handshakeBuf) >= 4 {
		handshakeLen := int(p.handshakeBuf[1])<<16 |
			int(p.handshakeBuf[2])<<8 |
			int(p.handshakeBuf[3]);
		if handshakeLen + 4 > len(p.handshakeBuf) {
			break;
		}

		bytes := p.handshakeBuf[0 : handshakeLen + 4];
		p.handshakeBuf = p.handshakeBuf[handshakeLen + 4 : len(p.handshakeBuf)];
		if bytes[0] == typeFinished {
			// Special case because Finished is synchronous: the
			// handshake handler has to tell us if it's ok to start
			// forwarding application data.
			m := new(finishedMsg);
			if !m.unmarshal(bytes) {
				p.error(alertUnexpectedMessage);
			}
			p.handshakeChan <- m;
			var ok bool;
			p.connState, ok = (<-p.controlChan).(ConnectionState);
			if !ok || p.connState.Error != 0 {
				p.shutdown = true;
				return;
			}
		} else {
			msg, ok := parseHandshakeMsg(bytes);
			if !ok {
				p.error(alertUnexpectedMessage);
				return;
			}
			p.handshakeChan <- msg;
		}
	}
}

func (p *recordProcessor) error(err alertType) {
	close(p.handshakeChan);
	p.connState.Error = err;
	p.wakeWaiters();
	p.shutdown = true;
}

func parseHandshakeMsg(data []byte) (interface{}, bool) {
	var m interface {
		unmarshal([]byte) bool;
	}

	switch data[0] {
	case typeClientHello:
		m = new(clientHelloMsg);
	case typeClientKeyExchange:
		m = new(clientKeyExchangeMsg);
	default:
		return nil, false;
	}

	ok := m.unmarshal(data);
	return m, ok;
}