move to p2p dir

al.go

+package secio
+
+import (
+	"errors"
+	"fmt"
+	"strings"
+
+	"crypto/aes"
+	"crypto/cipher"
+	"crypto/hmac"
+	"crypto/sha1"
+	"crypto/sha256"
+	"crypto/sha512"
+	"hash"
+
+	bfish "github.com/ipfs/go-ipfs/Godeps/_workspace/src/golang.org/x/crypto/blowfish"
+	ci "github.com/ipfs/go-ipfs/p2p/crypto"
+)
+
+// List of supported ECDH curves
+var SupportedExchanges = "P-256,P-384,P-521"
+
+// List of supported Ciphers
+var SupportedCiphers = "AES-256,AES-128,Blowfish"
+
+// List of supported Hashes
+var SupportedHashes = "SHA256,SHA512"
+
+type HMAC struct {
+	hash.Hash
+	size int
+}
+
+// encParams represent encryption parameters
+type encParams struct {
+	// keys
+	permanentPubKey ci.PubKey
+	ephemeralPubKey []byte
+	keys            ci.StretchedKeys
+
+	// selections
+	curveT  string
+	cipherT string
+	hashT   string
+
+	// cipher + mac
+	cipher cipher.Stream
+	mac    HMAC
+}
+
+func (e *encParams) makeMacAndCipher() error {
+	m, err := newMac(e.hashT, e.keys.MacKey)
+	if err != nil {
+		return err
+	}
+
+	bc, err := newBlockCipher(e.cipherT, e.keys.CipherKey)
+	if err != nil {
+		return err
+	}
+
+	e.cipher = cipher.NewCTR(bc, e.keys.IV)
+	e.mac = m
+	return nil
+}
+
+func newMac(hashType string, key []byte) (HMAC, error) {
+	switch hashType {
+	case "SHA1":
+		return HMAC{hmac.New(sha1.New, key), sha1.Size}, nil
+	case "SHA512":
+		return HMAC{hmac.New(sha512.New, key), sha512.Size}, nil
+	case "SHA256":
+		return HMAC{hmac.New(sha256.New, key), sha256.Size}, nil
+	default:
+		return HMAC{}, fmt.Errorf("Unrecognized hash type: %s", hashType)
+	}
+}
+
+func newBlockCipher(cipherT string, key []byte) (cipher.Block, error) {
+	switch cipherT {
+	case "AES-128", "AES-256":
+		return aes.NewCipher(key)
+	case "Blowfish":
+		return bfish.NewCipher(key)
+	default:
+		return nil, fmt.Errorf("Unrecognized cipher type: %s", cipherT)
+	}
+}
+
+// Determines which algorithm to use.  Note:  f(a, b) = f(b, a)
+func selectBest(order int, p1, p2 string) (string, error) {
+	var f, s []string
+	switch {
+	case order < 0:
+		f = strings.Split(p2, ",")
+		s = strings.Split(p1, ",")
+	case order > 0:
+		f = strings.Split(p1, ",")
+		s = strings.Split(p2, ",")
+	default: // Exact same preferences.
+		p := strings.Split(p1, ",")
+		return p[0], nil
+	}
+
+	for _, fc := range f {
+		for _, sc := range s {
+			if fc == sc {
+				return fc, nil
+			}
+		}
+	}
+
+	return "", errors.New("No algorithms in common!")
+}

interface.go

+// package secio handles establishing secure communication between two peers.
+package secio
+
+import (
+	"io"
+
+	ci "github.com/ipfs/go-ipfs/p2p/crypto"
+
+	msgio "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-msgio"
+	context "github.com/ipfs/go-ipfs/Godeps/_workspace/src/golang.org/x/net/context"
+	peer "github.com/ipfs/go-ipfs/p2p/peer"
+)
+
+// SessionGenerator constructs secure communication sessions for a peer.
+type SessionGenerator struct {
+	LocalID    peer.ID
+	PrivateKey ci.PrivKey
+}
+
+// NewSession takes an insecure io.ReadWriter, sets up a TLS-like
+// handshake with the other side, and returns a secure session.
+// The handshake isn't run until the connection is read or written to.
+// See the source for the protocol details and security implementation.
+// The provided Context is only needed for the duration of this function.
+func (sg *SessionGenerator) NewSession(ctx context.Context, insecure io.ReadWriteCloser) (Session, error) {
+	return newSecureSession(ctx, sg.LocalID, sg.PrivateKey, insecure)
+}
+
+type Session interface {
+	// ReadWriter returns the encrypted communication channel
+	ReadWriter() msgio.ReadWriteCloser
+
+	// LocalPeer retrieves the local peer.
+	LocalPeer() peer.ID
+
+	// LocalPrivateKey retrieves the local private key
+	LocalPrivateKey() ci.PrivKey
+
+	// RemotePeer retrieves the remote peer.
+	RemotePeer() peer.ID
+
+	// RemotePublicKey retrieves the remote's public key
+	// which was received during the handshake.
+	RemotePublicKey() ci.PubKey
+
+	// Close closes the secure session
+	Close() error
+}
+
+// SecureReadWriter returns the encrypted communication channel
+func (s *secureSession) ReadWriter() msgio.ReadWriteCloser {
+	if err := s.Handshake(); err != nil {
+		return &closedRW{err}
+	}
+	return s.secure
+}
+
+// LocalPeer retrieves the local peer.
+func (s *secureSession) LocalPeer() peer.ID {
+	return s.localPeer
+}
+
+// LocalPrivateKey retrieves the local peer's PrivateKey
+func (s *secureSession) LocalPrivateKey() ci.PrivKey {
+	return s.localKey
+}
+
+// RemotePeer retrieves the remote peer.
+func (s *secureSession) RemotePeer() peer.ID {
+	if err := s.Handshake(); err != nil {
+		return ""
+	}
+	return s.remotePeer
+}
+
+// RemotePeer retrieves the remote peer.
+func (s *secureSession) RemotePublicKey() ci.PubKey {
+	if err := s.Handshake(); err != nil {
+		return nil
+	}
+	return s.remote.permanentPubKey
+}
+
+// Close closes the secure session
+func (s *secureSession) Close() error {
+	s.cancel()
+	s.handshakeMu.Lock()
+	defer s.handshakeMu.Unlock()
+	if s.secure == nil {
+		return s.insecure.Close() // hadn't secured yet.
+	}
+	return s.secure.Close()
+}
+
+// closedRW implements a stub msgio interface that's already
+// closed and errored.
+type closedRW struct {
+	err error
+}
+
+func (c *closedRW) Read(buf []byte) (int, error) {
+	return 0, c.err
+}
+
+func (c *closedRW) Write(buf []byte) (int, error) {
+	return 0, c.err
+}
+
+func (c *closedRW) NextMsgLen() (int, error) {
+	return 0, c.err
+}
+
+func (c *closedRW) ReadMsg() ([]byte, error) {
+	return nil, c.err
+}
+
+func (c *closedRW) WriteMsg(buf []byte) error {
+	return c.err
+}
+
+func (c *closedRW) Close() error {
+	return c.err
+}
+
+func (c *closedRW) ReleaseMsg(m []byte) {
+}

io_test.go

+package secio

pb/Makefile

+PB = $(wildcard *.proto)
+GO = $(PB:.proto=.pb.go)
+
+all: $(GO)
+
+%.pb.go: %.proto
+		protoc --gogo_out=. --proto_path=../../../../../../:/usr/local/opt/protobuf/include:. $<
+
+clean:
+		rm *.pb.go

pb/spipe.pb.go

+// Code generated by protoc-gen-gogo.
+// source: spipe.proto
+// DO NOT EDIT!
+
+/*
+Package spipe_pb is a generated protocol buffer package.
+
+It is generated from these files:
+	spipe.proto
+
+It has these top-level messages:
+	Propose
+	Exchange
+*/
+package spipe_pb
+
+import proto "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/gogo/protobuf/proto"
+import math "math"
+
+// Reference imports to suppress errors if they are not otherwise used.
+var _ = proto.Marshal
+var _ = math.Inf
+
+type Propose struct {
+	Rand             []byte  `protobuf:"bytes,1,opt,name=rand" json:"rand,omitempty"`
+	Pubkey           []byte  `protobuf:"bytes,2,opt,name=pubkey" json:"pubkey,omitempty"`
+	Exchanges        *string `protobuf:"bytes,3,opt,name=exchanges" json:"exchanges,omitempty"`
+	Ciphers          *string `protobuf:"bytes,4,opt,name=ciphers" json:"ciphers,omitempty"`
+	Hashes           *string `protobuf:"bytes,5,opt,name=hashes" json:"hashes,omitempty"`
+	XXX_unrecognized []byte  `json:"-"`
+}
+
+func (m *Propose) Reset()         { *m = Propose{} }
+func (m *Propose) String() string { return proto.CompactTextString(m) }
+func (*Propose) ProtoMessage()    {}
+
+func (m *Propose) GetRand() []byte {
+	if m != nil {
+		return m.Rand
+	}
+	return nil
+}
+
+func (m *Propose) GetPubkey() []byte {
+	if m != nil {
+		return m.Pubkey
+	}
+	return nil
+}
+
+func (m *Propose) GetExchanges() string {
+	if m != nil && m.Exchanges != nil {
+		return *m.Exchanges
+	}
+	return ""
+}
+
+func (m *Propose) GetCiphers() string {
+	if m != nil && m.Ciphers != nil {
+		return *m.Ciphers
+	}
+	return ""
+}
+
+func (m *Propose) GetHashes() string {
+	if m != nil && m.Hashes != nil {
+		return *m.Hashes
+	}
+	return ""
+}
+
+type Exchange struct {
+	Epubkey          []byte `protobuf:"bytes,1,opt,name=epubkey" json:"epubkey,omitempty"`
+	Signature        []byte `protobuf:"bytes,2,opt,name=signature" json:"signature,omitempty"`
+	XXX_unrecognized []byte `json:"-"`
+}
+
+func (m *Exchange) Reset()         { *m = Exchange{} }
+func (m *Exchange) String() string { return proto.CompactTextString(m) }
+func (*Exchange) ProtoMessage()    {}
+
+func (m *Exchange) GetEpubkey() []byte {
+	if m != nil {
+		return m.Epubkey
+	}
+	return nil
+}
+
+func (m *Exchange) GetSignature() []byte {
+	if m != nil {
+		return m.Signature
+	}
+	return nil
+}
+
+func init() {
+}

pb/spipe.proto

+package spipe.pb;
+
+message Propose {
+	optional bytes rand = 1;
+	optional bytes pubkey = 2;
+	optional string exchanges = 3;
+	optional string ciphers = 4;
+	optional string hashes = 5;
+}
+
+message Exchange {
+	optional bytes epubkey = 1;
+	optional bytes signature = 2;
+}

protocol.go

+package secio
+
+import (
+	"bytes"
+	"crypto/rand"
+	"errors"
+	"fmt"
+	"io"
+	"sync"
+	"time"
+
+	msgio "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-msgio"
+	context "github.com/ipfs/go-ipfs/Godeps/_workspace/src/golang.org/x/net/context"
+	ci "github.com/ipfs/go-ipfs/p2p/crypto"
+	pb "github.com/ipfs/go-ipfs/p2p/crypto/secio/pb"
+	peer "github.com/ipfs/go-ipfs/p2p/peer"
+	u "github.com/ipfs/go-ipfs/util"
+	logging "github.com/ipfs/go-ipfs/vendor/go-log-v1.0.0"
+)
+
+var log = logging.Logger("secio")
+
+// ErrUnsupportedKeyType is returned when a private key cast/type switch fails.
+var ErrUnsupportedKeyType = errors.New("unsupported key type")
+
+// ErrClosed signals the closing of a connection.
+var ErrClosed = errors.New("connection closed")
+
+// ErrEcho is returned when we're attempting to handshake with the same keys and nonces.
+var ErrEcho = errors.New("same keys and nonces. one side talking to self.")
+
+// HandshakeTimeout governs how long the handshake will be allowed to take place for.
+// Making this number large means there could be many bogus connections waiting to
+// timeout in flight. Typical handshakes take ~3RTTs, so it should be completed within
+// seconds across a typical planet in the solar system.
+var HandshakeTimeout = time.Second * 30
+
+// nonceSize is the size of our nonces (in bytes)
+const nonceSize = 16
+
+// secureSession encapsulates all the parameters needed for encrypting
+// and decrypting traffic from an insecure channel.
+type secureSession struct {
+	ctx    context.Context
+	cancel context.CancelFunc
+
+	secure    msgio.ReadWriteCloser
+	insecure  io.ReadWriteCloser
+	insecureM msgio.ReadWriter
+
+	localKey   ci.PrivKey
+	localPeer  peer.ID
+	remotePeer peer.ID
+
+	local  encParams
+	remote encParams
+
+	sharedSecret []byte
+
+	handshakeMu   sync.Mutex // guards handshakeDone + handshakeErr
+	handshakeDone bool
+	handshakeErr  error
+}
+
+func (s *secureSession) Loggable() map[string]interface{} {
+	m := make(map[string]interface{})
+	m["localPeer"] = s.localPeer.Pretty()
+	m["remotePeer"] = s.remotePeer.Pretty()
+	m["established"] = (s.secure != nil)
+	return m
+}
+
+func newSecureSession(ctx context.Context, local peer.ID, key ci.PrivKey, insecure io.ReadWriteCloser) (*secureSession, error) {
+	s := &secureSession{localPeer: local, localKey: key}
+	s.ctx, s.cancel = context.WithCancel(ctx)
+
+	switch {
+	case s.localPeer == "":
+		return nil, errors.New("no local id provided")
+	case s.localKey == nil:
+		return nil, errors.New("no local private key provided")
+	case !s.localPeer.MatchesPrivateKey(s.localKey):
+		return nil, fmt.Errorf("peer.ID does not match PrivateKey")
+	case insecure == nil:
+		return nil, fmt.Errorf("insecure ReadWriter is nil")
+	}
+
+	s.ctx = ctx
+	s.insecure = insecure
+	s.insecureM = msgio.NewReadWriter(insecure)
+	return s, nil
+}
+
+func (s *secureSession) Handshake() error {
+	s.handshakeMu.Lock()
+	defer s.handshakeMu.Unlock()
+
+	if s.handshakeErr != nil {
+		return s.handshakeErr
+	}
+
+	if !s.handshakeDone {
+		s.handshakeErr = s.runHandshake()
+		s.handshakeDone = true
+	}
+	return s.handshakeErr
+}
+
+// runHandshake performs initial communication over insecure channel to share
+// keys, IDs, and initiate communication, assigning all necessary params.
+// requires the duplex channel to be a msgio.ReadWriter (for framed messaging)
+func (s *secureSession) runHandshake() error {
+	ctx, cancel := context.WithTimeout(s.ctx, HandshakeTimeout) // remove
+	defer cancel()
+
+	// =============================================================================
+	// step 1. Propose -- propose cipher suite + send pubkeys + nonce
+
+	// Generate and send Hello packet.
+	// Hello = (rand, PublicKey, Supported)
+	nonceOut := make([]byte, nonceSize)
+	_, err := rand.Read(nonceOut)
+	if err != nil {
+		return err
+	}
+
+	defer log.EventBegin(ctx, "secureHandshake", s).Done()
+
+	s.local.permanentPubKey = s.localKey.GetPublic()
+	myPubKeyBytes, err := s.local.permanentPubKey.Bytes()
+	if err != nil {
+		return err
+	}
+
+	proposeOut := new(pb.Propose)
+	proposeOut.Rand = nonceOut
+	proposeOut.Pubkey = myPubKeyBytes
+	proposeOut.Exchanges = &SupportedExchanges
+	proposeOut.Ciphers = &SupportedCiphers
+	proposeOut.Hashes = &SupportedHashes
+
+	// log.Debugf("1.0 Propose: nonce:%s exchanges:%s ciphers:%s hashes:%s",
+	// 	nonceOut, SupportedExchanges, SupportedCiphers, SupportedHashes)
+
+	// Send Propose packet (respects ctx)
+	proposeOutBytes, err := writeMsgCtx(ctx, s.insecureM, proposeOut)
+	if err != nil {
+		return err
+	}
+
+	// Receive + Parse their Propose packet and generate an Exchange packet.
+	proposeIn := new(pb.Propose)
+	proposeInBytes, err := readMsgCtx(ctx, s.insecureM, proposeIn)
+	if err != nil {
+		return err
+	}
+
+	// log.Debugf("1.0.1 Propose recv: nonce:%s exchanges:%s ciphers:%s hashes:%s",
+	// 	proposeIn.GetRand(), proposeIn.GetExchanges(), proposeIn.GetCiphers(), proposeIn.GetHashes())
+
+	// =============================================================================
+	// step 1.1 Identify -- get identity from their key
+
+	// get remote identity
+	s.remote.permanentPubKey, err = ci.UnmarshalPublicKey(proposeIn.GetPubkey())
+	if err != nil {
+		return err
+	}
+
+	// get peer id
+	s.remotePeer, err = peer.IDFromPublicKey(s.remote.permanentPubKey)
+	if err != nil {
+		return err
+	}
+
+	log.Debugf("1.1 Identify: %s Remote Peer Identified as %s", s.localPeer, s.remotePeer)
+
+	// =============================================================================
+	// step 1.2 Selection -- select/agree on best encryption parameters
+
+	// to determine order, use cmp(H(remote_pubkey||local_rand), H(local_pubkey||remote_rand)).
+	oh1 := u.Hash(append(proposeIn.GetPubkey(), nonceOut...))
+	oh2 := u.Hash(append(myPubKeyBytes, proposeIn.GetRand()...))
+	order := bytes.Compare(oh1, oh2)
+	if order == 0 {
+		return ErrEcho // talking to self (same socket. must be reuseport + dialing self)
+	}
+
+	s.local.curveT, err = selectBest(order, SupportedExchanges, proposeIn.GetExchanges())
+	if err != nil {
+		return err
+	}
+
+	s.local.cipherT, err = selectBest(order, SupportedCiphers, proposeIn.GetCiphers())
+	if err != nil {
+		return err
+	}
+
+	s.local.hashT, err = selectBest(order, SupportedHashes, proposeIn.GetHashes())
+	if err != nil {
+		return err
+	}
+
+	// we use the same params for both directions (must choose same curve)
+	// WARNING: if they dont SelectBest the same way, this won't work...
+	s.remote.curveT = s.local.curveT
+	s.remote.cipherT = s.local.cipherT
+	s.remote.hashT = s.local.hashT
+
+	// log.Debugf("1.2 selection: exchange:%s cipher:%s hash:%s",
+	// 	s.local.curveT, s.local.cipherT, s.local.hashT)
+
+	// =============================================================================
+	// step 2. Exchange -- exchange (signed) ephemeral keys. verify signatures.
+
+	// Generate EphemeralPubKey
+	var genSharedKey ci.GenSharedKey
+	s.local.ephemeralPubKey, genSharedKey, err = ci.GenerateEKeyPair(s.local.curveT)
+
+	// Gather corpus to sign.
+	selectionOut := new(bytes.Buffer)
+	selectionOut.Write(proposeOutBytes)
+	selectionOut.Write(proposeInBytes)
+	selectionOut.Write(s.local.ephemeralPubKey)
+	selectionOutBytes := selectionOut.Bytes()
+
+	// log.Debugf("2.0 exchange: %v", selectionOutBytes)
+	exchangeOut := new(pb.Exchange)
+	exchangeOut.Epubkey = s.local.ephemeralPubKey
+	exchangeOut.Signature, err = s.localKey.Sign(selectionOutBytes)
+	if err != nil {
+		return err
+	}
+
+	// Send Propose packet (respects ctx)
+	if _, err := writeMsgCtx(ctx, s.insecureM, exchangeOut); err != nil {
+		return err
+	}
+
+	// Receive + Parse their Exchange packet.
+	exchangeIn := new(pb.Exchange)
+	if _, err := readMsgCtx(ctx, s.insecureM, exchangeIn); err != nil {
+		return err
+	}
+
+	// =============================================================================
+	// step 2.1. Verify -- verify their exchange packet is good.
+
+	// get their ephemeral pub key
+	s.remote.ephemeralPubKey = exchangeIn.GetEpubkey()
+
+	selectionIn := new(bytes.Buffer)
+	selectionIn.Write(proposeInBytes)
+	selectionIn.Write(proposeOutBytes)
+	selectionIn.Write(s.remote.ephemeralPubKey)
+	selectionInBytes := selectionIn.Bytes()
+	// log.Debugf("2.0.1 exchange recv: %v", selectionInBytes)
+
+	// u.POut("Remote Peer Identified as %s\n", s.remote)
+	sigOK, err := s.remote.permanentPubKey.Verify(selectionInBytes, exchangeIn.GetSignature())
+	if err != nil {
+		// log.Error("2.1 Verify: failed: %s", err)
+		return err
+	}
+
+	if !sigOK {
+		err := errors.New("Bad signature!")
+		// log.Error("2.1 Verify: failed: %s", err)
+		return err
+	}
+	// log.Debugf("2.1 Verify: signature verified.")
+
+	// =============================================================================
+	// step 2.2. Keys -- generate keys for mac + encryption
+
+	// OK! seems like we're good to go.
+	s.sharedSecret, err = genSharedKey(exchangeIn.GetEpubkey())
+	if err != nil {
+		return err
+	}
+
+	// generate two sets of keys (stretching)
+	k1, k2 := ci.KeyStretcher(s.local.cipherT, s.local.hashT, s.sharedSecret)
+
+	// use random nonces to decide order.
+	switch {
+	case order > 0:
+		// just break
+	case order < 0:
+		k1, k2 = k2, k1 // swap
+	default:
+		// we should've bailed before this. but if not, bail here.
+		return ErrEcho
+	}
+	s.local.keys = k1
+	s.remote.keys = k2
+
+	// log.Debug("2.2 keys:\n\tshared: %v\n\tk1: %v\n\tk2: %v",
+	// 	s.sharedSecret, s.local.keys, s.remote.keys)
+
+	// =============================================================================
+	// step 2.3. MAC + Cipher -- prepare MAC + cipher
+
+	if err := s.local.makeMacAndCipher(); err != nil {
+		return err
+	}
+
+	if err := s.remote.makeMacAndCipher(); err != nil {
+		return err
+	}
+
+	// log.Debug("2.3 mac + cipher.")
+
+	// =============================================================================
+	// step 3. Finish -- send expected message to verify encryption works (send local nonce)
+
+	// setup ETM ReadWriter
+	w := NewETMWriter(s.insecure, s.local.cipher, s.local.mac)
+	r := NewETMReader(s.insecure, s.remote.cipher, s.remote.mac)
+	s.secure = msgio.Combine(w, r).(msgio.ReadWriteCloser)
+
+	// log.Debug("3.0 finish. sending: %v", proposeIn.GetRand())
+	// send their Nonce.
+	if _, err := s.secure.Write(proposeIn.GetRand()); err != nil {
+		return fmt.Errorf("Failed to write Finish nonce: %s", err)
+	}
+
+	// read our Nonce
+	nonceOut2 := make([]byte, len(nonceOut))
+	if _, err := io.ReadFull(s.secure, nonceOut2); err != nil {
+		return fmt.Errorf("Failed to read Finish nonce: %s", err)
+	}
+
+	// log.Debug("3.0 finish.\n\texpect: %v\n\tactual: %v", nonceOut, nonceOut2)
+	if !bytes.Equal(nonceOut, nonceOut2) {
+		return fmt.Errorf("Failed to read our encrypted nonce: %s != %s", nonceOut2, nonceOut)
+	}
+
+	// Whew! ok, that's all folks.
+	return nil
+}

rw.go

+package secio
+
+import (
+	"crypto/cipher"
+	"errors"
+	"fmt"
+	"io"
+	"sync"
+
+	"crypto/hmac"
+
+	proto "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/gogo/protobuf/proto"
+	msgio "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-msgio"
+	mpool "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-msgio/mpool"
+	context "github.com/ipfs/go-ipfs/Godeps/_workspace/src/golang.org/x/net/context"
+)
+
+const MaxMsgSize = 8 * 1024 * 1024
+
+var ErrMaxMessageSize = errors.New("attempted to read message larger than max size")
+
+// ErrMACInvalid signals that a MAC verification failed
+var ErrMACInvalid = errors.New("MAC verification failed")
+
+// bufPool is a ByteSlicePool for messages. we need buffers because (sadly)
+// we cannot encrypt in place-- the user needs their buffer back.
+var bufPool = mpool.ByteSlicePool
+
+type etmWriter struct {
+	// params
+	pool mpool.Pool        // for the buffers with encrypted data
+	msg  msgio.WriteCloser // msgio for knowing where boundaries lie
+	str  cipher.Stream     // the stream cipher to encrypt with
+	mac  HMAC              // the mac to authenticate data with
+
+	sync.Mutex
+}
+
+// NewETMWriter Encrypt-Then-MAC
+func NewETMWriter(w io.Writer, s cipher.Stream, mac HMAC) msgio.WriteCloser {
+	return &etmWriter{msg: msgio.NewWriter(w), str: s, mac: mac, pool: bufPool}
+}
+
+// Write writes passed in buffer as a single message.
+func (w *etmWriter) Write(b []byte) (int, error) {
+	if err := w.WriteMsg(b); err != nil {
+		return 0, err
+	}
+	return len(b), nil
+}
+
+// WriteMsg writes the msg in the passed in buffer.
+func (w *etmWriter) WriteMsg(b []byte) error {
+	w.Lock()
+	defer w.Unlock()
+
+	// encrypt.
+	data := w.pool.Get(uint32(len(b))).([]byte)
+	data = data[:len(b)] // the pool's buffer may be larger
+	w.str.XORKeyStream(data, b)
+
+	// log.Debugf("ENC plaintext (%d): %s %v", len(b), b, b)
+	// log.Debugf("ENC ciphertext (%d): %s %v", len(data), data, data)
+
+	// then, mac.
+	if _, err := w.mac.Write(data); err != nil {
+		return err
+	}
+
+	// Sum appends.
+	data = w.mac.Sum(data)
+	w.mac.Reset()
+	// it's sad to append here. our buffers are -- hopefully -- coming from
+	// a shared buffer pool, so the append may not actually cause allocation
+	// one can only hope. i guess we'll see.
+
+	return w.msg.WriteMsg(data)
+}
+
+func (w *etmWriter) Close() error {
+	return w.msg.Close()
+}
+
+type etmReader struct {
+	msgio.Reader
+	io.Closer
+
+	// buffer
+	buf []byte
+
+	// params
+	msg msgio.ReadCloser // msgio for knowing where boundaries lie
+	str cipher.Stream    // the stream cipher to encrypt with
+	mac HMAC             // the mac to authenticate data with
+
+	sync.Mutex
+}
+
+// NewETMReader Encrypt-Then-MAC
+func NewETMReader(r io.Reader, s cipher.Stream, mac HMAC) msgio.ReadCloser {
+	return &etmReader{msg: msgio.NewReader(r), str: s, mac: mac}
+}
+
+func (r *etmReader) NextMsgLen() (int, error) {
+	return r.msg.NextMsgLen()
+}
+
+func (r *etmReader) drainBuf(buf []byte) int {
+	if r.buf == nil {
+		return 0
+	}
+
+	n := copy(buf, r.buf)
+	r.buf = r.buf[n:]
+	return n
+}
+
+func (r *etmReader) Read(buf []byte) (int, error) {
+	r.Lock()
+	defer r.Unlock()
+
+	// first, check if we have anything in the buffer
+	copied := r.drainBuf(buf)
+	buf = buf[copied:]
+	if copied > 0 {
+		return copied, nil
+		// return here to avoid complicating the rest...
+		// user can call io.ReadFull.
+	}
+
+	// check the buffer has enough space for the next msg
+	fullLen, err := r.msg.NextMsgLen()
+	if err != nil {
+		return 0, err
+	}
+
+	if fullLen > MaxMsgSize {
+		return 0, ErrMaxMessageSize
+	}
+
+	buf2 := buf
+	changed := false
+	// if not enough space, allocate a new buffer.
+	if cap(buf) < fullLen {
+		buf2 = make([]byte, fullLen)
+		changed = true
+	}
+	buf2 = buf2[:fullLen]
+
+	n, err := io.ReadFull(r.msg, buf2)
+	if err != nil {
+		return n, err
+	}
+
+	m, err := r.macCheckThenDecrypt(buf2)
+	if err != nil {
+		return 0, err
+	}
+	buf2 = buf2[:m]
+	if !changed {
+		return m, nil
+	}
+
+	n = copy(buf, buf2)
+	if len(buf2) > len(buf) {
+		r.buf = buf2[len(buf):] // had some left over? save it.
+	}
+	return n, nil
+}
+
+func (r *etmReader) ReadMsg() ([]byte, error) {
+	r.Lock()
+	defer r.Unlock()
+
+	msg, err := r.msg.ReadMsg()
+	if err != nil {
+		return nil, err
+	}
+
+	n, err := r.macCheckThenDecrypt(msg)
+	if err != nil {
+		return nil, err
+	}
+	return msg[:n], nil
+}
+
+func (r *etmReader) macCheckThenDecrypt(m []byte) (int, error) {
+	l := len(m)
+	if l < r.mac.size {
+		return 0, fmt.Errorf("buffer (%d) shorter than MAC size (%d)", l, r.mac.size)
+	}
+
+	mark := l - r.mac.size
+	data := m[:mark]
+	macd := m[mark:]
+
+	r.mac.Write(data)
+	expected := r.mac.Sum(nil)
+	r.mac.Reset()
+
+	// check mac. if failed, return error.
+	if !hmac.Equal(macd, expected) {
+		log.Debug("MAC Invalid:", expected, "!=", macd)
+		return 0, ErrMACInvalid
+	}
+
+	// ok seems good. decrypt. (can decrypt in place, yay!)
+	// log.Debugf("DEC ciphertext (%d): %s %v", len(data), data, data)
+	r.str.XORKeyStream(data, data)
+	// log.Debugf("DEC plaintext (%d): %s %v", len(data), data, data)
+
+	return mark, nil
+}
+
+func (w *etmReader) Close() error {
+	return w.msg.Close()
+}
+
+// ReleaseMsg signals a buffer can be reused.
+func (r *etmReader) ReleaseMsg(b []byte) {
+	r.msg.ReleaseMsg(b)
+}
+
+// writeMsgCtx is used by the
+func writeMsgCtx(ctx context.Context, w msgio.Writer, msg proto.Message) ([]byte, error) {
+	enc, err := proto.Marshal(msg)
+	if err != nil {
+		return nil, err
+	}
+
+	// write in a goroutine so we can exit when our context is cancelled.
+	done := make(chan error)
+	go func(m []byte) {
+		err := w.WriteMsg(m)
+		select {
+		case done <- err:
+		case <-ctx.Done():
+		}
+	}(enc)
+
+	select {
+	case <-ctx.Done():
+		return nil, ctx.Err()
+	case e := <-done:
+		return enc, e
+	}
+}
+
+func readMsgCtx(ctx context.Context, r msgio.Reader, p proto.Message) ([]byte, error) {
+	var msg []byte
+
+	// read in a goroutine so we can exit when our context is cancelled.
+	done := make(chan error)
+	go func() {
+		var err error
+		msg, err = r.ReadMsg()
+		select {
+		case done <- err:
+		case <-ctx.Done():
+		}
+	}()
+
+	select {
+	case <-ctx.Done():
+		return nil, ctx.Err()
+	case e := <-done:
+		if e != nil {
+			return nil, e
+		}
+	}
+
+	return msg, proto.Unmarshal(msg, p)
+}