add blowfish code and refactor pipes and duplex a little

Godeps/Godeps.json

@@ -20,6 +20,11 @@
 			"Rev": "7dda39b2e7d5e265014674c5af696ba4186679e9"
 		},
 		{
+			"ImportPath": "code.google.com/p/go.crypto/blowfish",
+			"Comment": "null-219",
+			"Rev": "00a7d3b31bbab5795b4a51933c04fc2768242970"
+		},
+		{
 			"ImportPath": "code.google.com/p/go.crypto/sha3",
 			"Comment": "null-219",
 			"Rev": "00a7d3b31bbab5795b4a51933c04fc2768242970"

Godeps/_workspace/src/code.google.com/p/go.crypto/blowfish/block.go

+// Copyright 2010 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 blowfish
+
+// getNextWord returns the next big-endian uint32 value from the byte slice
+// at the given position in a circular manner, updating the position.
+func getNextWord(b []byte, pos *int) uint32 {
+	var w uint32
+	j := *pos
+	for i := 0; i < 4; i++ {
+		w = w<<8 | uint32(b[j])
+		j++
+		if j >= len(b) {
+			j = 0
+		}
+	}
+	*pos = j
+	return w
+}
+
+// ExpandKey performs a key expansion on the given *Cipher. Specifically, it
+// performs the Blowfish algorithm's key schedule which sets up the *Cipher's
+// pi and substitution tables for calls to Encrypt. This is used, primarily,
+// by the bcrypt package to reuse the Blowfish key schedule during its
+// set up. It's unlikely that you need to use this directly.
+func ExpandKey(key []byte, c *Cipher) {
+	j := 0
+	for i := 0; i < 18; i++ {
+		// Using inlined getNextWord for performance.
+		var d uint32
+		for k := 0; k < 4; k++ {
+			d = d<<8 | uint32(key[j])
+			j++
+			if j >= len(key) {
+				j = 0
+			}
+		}
+		c.p[i] ^= d
+	}
+
+	var l, r uint32
+	for i := 0; i < 18; i += 2 {
+		l, r = encryptBlock(l, r, c)
+		c.p[i], c.p[i+1] = l, r
+	}
+
+	for i := 0; i < 256; i += 2 {
+		l, r = encryptBlock(l, r, c)
+		c.s0[i], c.s0[i+1] = l, r
+	}
+	for i := 0; i < 256; i += 2 {
+		l, r = encryptBlock(l, r, c)
+		c.s1[i], c.s1[i+1] = l, r
+	}
+	for i := 0; i < 256; i += 2 {
+		l, r = encryptBlock(l, r, c)
+		c.s2[i], c.s2[i+1] = l, r
+	}
+	for i := 0; i < 256; i += 2 {
+		l, r = encryptBlock(l, r, c)
+		c.s3[i], c.s3[i+1] = l, r
+	}
+}
+
+// This is similar to ExpandKey, but folds the salt during the key
+// schedule. While ExpandKey is essentially expandKeyWithSalt with an all-zero
+// salt passed in, reusing ExpandKey turns out to be a place of inefficiency
+// and specializing it here is useful.
+func expandKeyWithSalt(key []byte, salt []byte, c *Cipher) {
+	j := 0
+	for i := 0; i < 18; i++ {
+		c.p[i] ^= getNextWord(key, &j)
+	}
+
+	j = 0
+	var l, r uint32
+	for i := 0; i < 18; i += 2 {
+		l ^= getNextWord(salt, &j)
+		r ^= getNextWord(salt, &j)
+		l, r = encryptBlock(l, r, c)
+		c.p[i], c.p[i+1] = l, r
+	}
+
+	for i := 0; i < 256; i += 2 {
+		l ^= getNextWord(salt, &j)
+		r ^= getNextWord(salt, &j)
+		l, r = encryptBlock(l, r, c)
+		c.s0[i], c.s0[i+1] = l, r
+	}
+
+	for i := 0; i < 256; i += 2 {
+		l ^= getNextWord(salt, &j)
+		r ^= getNextWord(salt, &j)
+		l, r = encryptBlock(l, r, c)
+		c.s1[i], c.s1[i+1] = l, r
+	}
+
+	for i := 0; i < 256; i += 2 {
+		l ^= getNextWord(salt, &j)
+		r ^= getNextWord(salt, &j)
+		l, r = encryptBlock(l, r, c)
+		c.s2[i], c.s2[i+1] = l, r
+	}
+
+	for i := 0; i < 256; i += 2 {
+		l ^= getNextWord(salt, &j)
+		r ^= getNextWord(salt, &j)
+		l, r = encryptBlock(l, r, c)
+		c.s3[i], c.s3[i+1] = l, r
+	}
+}
+
+func encryptBlock(l, r uint32, c *Cipher) (uint32, uint32) {
+	xl, xr := l, r
+	xl ^= c.p[0]
+	xr ^= ((c.s0[byte(xl>>24)] + c.s1[byte(xl>>16)]) ^ c.s2[byte(xl>>8)]) + c.s3[byte(xl)] ^ c.p[1]
+	xl ^= ((c.s0[byte(xr>>24)] + c.s1[byte(xr>>16)]) ^ c.s2[byte(xr>>8)]) + c.s3[byte(xr)] ^ c.p[2]
+	xr ^= ((c.s0[byte(xl>>24)] + c.s1[byte(xl>>16)]) ^ c.s2[byte(xl>>8)]) + c.s3[byte(xl)] ^ c.p[3]
+	xl ^= ((c.s0[byte(xr>>24)] + c.s1[byte(xr>>16)]) ^ c.s2[byte(xr>>8)]) + c.s3[byte(xr)] ^ c.p[4]
+	xr ^= ((c.s0[byte(xl>>24)] + c.s1[byte(xl>>16)]) ^ c.s2[byte(xl>>8)]) + c.s3[byte(xl)] ^ c.p[5]
+	xl ^= ((c.s0[byte(xr>>24)] + c.s1[byte(xr>>16)]) ^ c.s2[byte(xr>>8)]) + c.s3[byte(xr)] ^ c.p[6]
+	xr ^= ((c.s0[byte(xl>>24)] + c.s1[byte(xl>>16)]) ^ c.s2[byte(xl>>8)]) + c.s3[byte(xl)] ^ c.p[7]
+	xl ^= ((c.s0[byte(xr>>24)] + c.s1[byte(xr>>16)]) ^ c.s2[byte(xr>>8)]) + c.s3[byte(xr)] ^ c.p[8]
+	xr ^= ((c.s0[byte(xl>>24)] + c.s1[byte(xl>>16)]) ^ c.s2[byte(xl>>8)]) + c.s3[byte(xl)] ^ c.p[9]
+	xl ^= ((c.s0[byte(xr>>24)] + c.s1[byte(xr>>16)]) ^ c.s2[byte(xr>>8)]) + c.s3[byte(xr)] ^ c.p[10]
+	xr ^= ((c.s0[byte(xl>>24)] + c.s1[byte(xl>>16)]) ^ c.s2[byte(xl>>8)]) + c.s3[byte(xl)] ^ c.p[11]
+	xl ^= ((c.s0[byte(xr>>24)] + c.s1[byte(xr>>16)]) ^ c.s2[byte(xr>>8)]) + c.s3[byte(xr)] ^ c.p[12]
+	xr ^= ((c.s0[byte(xl>>24)] + c.s1[byte(xl>>16)]) ^ c.s2[byte(xl>>8)]) + c.s3[byte(xl)] ^ c.p[13]
+	xl ^= ((c.s0[byte(xr>>24)] + c.s1[byte(xr>>16)]) ^ c.s2[byte(xr>>8)]) + c.s3[byte(xr)] ^ c.p[14]
+	xr ^= ((c.s0[byte(xl>>24)] + c.s1[byte(xl>>16)]) ^ c.s2[byte(xl>>8)]) + c.s3[byte(xl)] ^ c.p[15]
+	xl ^= ((c.s0[byte(xr>>24)] + c.s1[byte(xr>>16)]) ^ c.s2[byte(xr>>8)]) + c.s3[byte(xr)] ^ c.p[16]
+	xr ^= c.p[17]
+	return xr, xl
+}
+
+func decryptBlock(l, r uint32, c *Cipher) (uint32, uint32) {
+	xl, xr := l, r
+	xl ^= c.p[17]
+	xr ^= ((c.s0[byte(xl>>24)] + c.s1[byte(xl>>16)]) ^ c.s2[byte(xl>>8)]) + c.s3[byte(xl)] ^ c.p[16]
+	xl ^= ((c.s0[byte(xr>>24)] + c.s1[byte(xr>>16)]) ^ c.s2[byte(xr>>8)]) + c.s3[byte(xr)] ^ c.p[15]
+	xr ^= ((c.s0[byte(xl>>24)] + c.s1[byte(xl>>16)]) ^ c.s2[byte(xl>>8)]) + c.s3[byte(xl)] ^ c.p[14]
+	xl ^= ((c.s0[byte(xr>>24)] + c.s1[byte(xr>>16)]) ^ c.s2[byte(xr>>8)]) + c.s3[byte(xr)] ^ c.p[13]
+	xr ^= ((c.s0[byte(xl>>24)] + c.s1[byte(xl>>16)]) ^ c.s2[byte(xl>>8)]) + c.s3[byte(xl)] ^ c.p[12]
+	xl ^= ((c.s0[byte(xr>>24)] + c.s1[byte(xr>>16)]) ^ c.s2[byte(xr>>8)]) + c.s3[byte(xr)] ^ c.p[11]
+	xr ^= ((c.s0[byte(xl>>24)] + c.s1[byte(xl>>16)]) ^ c.s2[byte(xl>>8)]) + c.s3[byte(xl)] ^ c.p[10]
+	xl ^= ((c.s0[byte(xr>>24)] + c.s1[byte(xr>>16)]) ^ c.s2[byte(xr>>8)]) + c.s3[byte(xr)] ^ c.p[9]
+	xr ^= ((c.s0[byte(xl>>24)] + c.s1[byte(xl>>16)]) ^ c.s2[byte(xl>>8)]) + c.s3[byte(xl)] ^ c.p[8]
+	xl ^= ((c.s0[byte(xr>>24)] + c.s1[byte(xr>>16)]) ^ c.s2[byte(xr>>8)]) + c.s3[byte(xr)] ^ c.p[7]
+	xr ^= ((c.s0[byte(xl>>24)] + c.s1[byte(xl>>16)]) ^ c.s2[byte(xl>>8)]) + c.s3[byte(xl)] ^ c.p[6]
+	xl ^= ((c.s0[byte(xr>>24)] + c.s1[byte(xr>>16)]) ^ c.s2[byte(xr>>8)]) + c.s3[byte(xr)] ^ c.p[5]
+	xr ^= ((c.s0[byte(xl>>24)] + c.s1[byte(xl>>16)]) ^ c.s2[byte(xl>>8)]) + c.s3[byte(xl)] ^ c.p[4]
+	xl ^= ((c.s0[byte(xr>>24)] + c.s1[byte(xr>>16)]) ^ c.s2[byte(xr>>8)]) + c.s3[byte(xr)] ^ c.p[3]
+	xr ^= ((c.s0[byte(xl>>24)] + c.s1[byte(xl>>16)]) ^ c.s2[byte(xl>>8)]) + c.s3[byte(xl)] ^ c.p[2]
+	xl ^= ((c.s0[byte(xr>>24)] + c.s1[byte(xr>>16)]) ^ c.s2[byte(xr>>8)]) + c.s3[byte(xr)] ^ c.p[1]
+	xr ^= c.p[0]
+	return xr, xl
+}

Godeps/_workspace/src/code.google.com/p/go.crypto/blowfish/cipher.go

+// Copyright 2010 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 blowfish implements Bruce Schneier's Blowfish encryption algorithm.
+package blowfish
+
+// The code is a port of Bruce Schneier's C implementation.
+// See http://www.schneier.com/blowfish.html.
+
+import "strconv"
+
+// The Blowfish block size in bytes.
+const BlockSize = 8
+
+// A Cipher is an instance of Blowfish encryption using a particular key.
+type Cipher struct {
+	p              [18]uint32
+	s0, s1, s2, s3 [256]uint32
+}
+
+type KeySizeError int
+
+func (k KeySizeError) Error() string {
+	return "crypto/blowfish: invalid key size " + strconv.Itoa(int(k))
+}
+
+// NewCipher creates and returns a Cipher.
+// The key argument should be the Blowfish key, from 1 to 56 bytes.
+func NewCipher(key []byte) (*Cipher, error) {
+	var result Cipher
+	if k := len(key); k < 1 || k > 56 {
+		return nil, KeySizeError(k)
+	}
+	initCipher(&result)
+	ExpandKey(key, &result)
+	return &result, nil
+}
+
+// NewSaltedCipher creates a returns a Cipher that folds a salt into its key
+// schedule. For most purposes, NewCipher, instead of NewSaltedCipher, is
+// sufficient and desirable. For bcrypt compatiblity, the key can be over 56
+// bytes.
+func NewSaltedCipher(key, salt []byte) (*Cipher, error) {
+	if len(salt) == 0 {
+		return NewCipher(key)
+	}
+	var result Cipher
+	if k := len(key); k < 1 {
+		return nil, KeySizeError(k)
+	}
+	initCipher(&result)
+	expandKeyWithSalt(key, salt, &result)
+	return &result, nil
+}
+
+// BlockSize returns the Blowfish block size, 8 bytes.
+// It is necessary to satisfy the Block interface in the
+// package "crypto/cipher".
+func (c *Cipher) BlockSize() int { return BlockSize }
+
+// Encrypt encrypts the 8-byte buffer src using the key k
+// and stores the result in dst.
+// Note that for amounts of data larger than a block,
+// it is not safe to just call Encrypt on successive blocks;
+// instead, use an encryption mode like CBC (see crypto/cipher/cbc.go).
+func (c *Cipher) Encrypt(dst, src []byte) {
+	l := uint32(src[0])<<24 | uint32(src[1])<<16 | uint32(src[2])<<8 | uint32(src[3])
+	r := uint32(src[4])<<24 | uint32(src[5])<<16 | uint32(src[6])<<8 | uint32(src[7])
+	l, r = encryptBlock(l, r, c)
+	dst[0], dst[1], dst[2], dst[3] = byte(l>>24), byte(l>>16), byte(l>>8), byte(l)
+	dst[4], dst[5], dst[6], dst[7] = byte(r>>24), byte(r>>16), byte(r>>8), byte(r)
+}
+
+// Decrypt decrypts the 8-byte buffer src using the key k
+// and stores the result in dst.
+func (c *Cipher) Decrypt(dst, src []byte) {
+	l := uint32(src[0])<<24 | uint32(src[1])<<16 | uint32(src[2])<<8 | uint32(src[3])
+	r := uint32(src[4])<<24 | uint32(src[5])<<16 | uint32(src[6])<<8 | uint32(src[7])
+	l, r = decryptBlock(l, r, c)
+	dst[0], dst[1], dst[2], dst[3] = byte(l>>24), byte(l>>16), byte(l>>8), byte(l)
+	dst[4], dst[5], dst[6], dst[7] = byte(r>>24), byte(r>>16), byte(r>>8), byte(r)
+}
+
+func initCipher(c *Cipher) {
+	copy(c.p[0:], p[0:])
+	copy(c.s0[0:], s0[0:])
+	copy(c.s1[0:], s1[0:])
+	copy(c.s2[0:], s2[0:])
+	copy(c.s3[0:], s3[0:])
+}

blockstore/blockstore.go

@@ -5,6 +5,7 @@ import (
 
 	ds "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore"
 
+	mh "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multihash"
 	blocks "github.com/jbenet/go-ipfs/blocks"
 	u "github.com/jbenet/go-ipfs/util"
 )
@@ -35,7 +36,12 @@ func (bs *blockstore) Get(k u.Key) (*blocks.Block, error) {
 	if !ok {
 		return nil, ValueTypeMismatch
 	}
-	return blocks.NewBlock(bdata), nil
+	//TODO: we *could* verify data coming in from the datastore here
+	//		but its probably very unecessary
+	return &blocks.Block{
+		Data:      bdata,
+		Multihash: mh.Multihash(k),
+	}, nil
 }
 
 func (bs *blockstore) Put(block *blocks.Block) error {

crypto/spipe/handshake.go

@@ -8,7 +8,6 @@ import (
 	"fmt"
 	"strings"
 
-	bfish "code.google.com/p/go.crypto/blowfish"
 	"crypto/aes"
 	"crypto/cipher"
 	"crypto/hmac"
@@ -16,6 +15,7 @@ import (
 	"crypto/sha1"
 	"crypto/sha256"
 	"crypto/sha512"
+	bfish "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.crypto/blowfish"
 	"hash"
 
 	proto "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/goprotobuf/proto"
@@ -108,17 +108,17 @@ func (s *SecurePipe) handshake() error {
 	}
 	log.Debugf("%s Remote Peer Identified as %s", s.local, s.remote)
 
-	exchange, err := selectBest(SupportedExchanges, proposeResp.GetExchanges())
+	exchange, err := SelectBest(SupportedExchanges, proposeResp.GetExchanges())
 	if err != nil {
 		return err
 	}
 
-	cipherType, err := selectBest(SupportedCiphers, proposeResp.GetCiphers())
+	cipherType, err := SelectBest(SupportedCiphers, proposeResp.GetCiphers())
 	if err != nil {
 		return err
 	}
 
-	hashType, err := selectBest(SupportedHashes, proposeResp.GetHashes())
+	hashType, err := SelectBest(SupportedHashes, proposeResp.GetHashes())
 	if err != nil {
 		return err
 	}
@@ -330,7 +330,7 @@ func (s *SecurePipe) handleSecureOut(hashType, cipherType string, mIV, mCKey, mM
 }
 
 // Determines which algorithm to use.  Note:  f(a, b) = f(b, a)
-func selectBest(myPrefs, theirPrefs string) (string, error) {
+func SelectBest(myPrefs, theirPrefs string) (string, error) {
 	// Person with greatest hash gets first choice.
 	myHash := u.Hash([]byte(myPrefs))
 	theirHash := u.Hash([]byte(theirPrefs))

crypto/spipe/pipe.go

@@ -5,18 +5,14 @@ import (
 
 	context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
 	peer "github.com/jbenet/go-ipfs/peer"
-)
 
-// Duplex is a simple duplex channel
-type Duplex struct {
-	In  chan []byte
-	Out chan []byte
-}
+	pipes "github.com/jbenet/go-ipfs/pipes"
+)
 
 // SecurePipe objects represent a bi-directional message channel.
 type SecurePipe struct {
-	Duplex
-	insecure Duplex
+	pipes.Duplex
+	insecure pipes.Duplex
 
 	local  peer.Peer
 	remote peer.Peer
@@ -34,12 +30,12 @@ type params struct {
 
 // NewSecurePipe constructs a pipe with channels of a given buffer size.
 func NewSecurePipe(ctx context.Context, bufsize int, local peer.Peer,
-	peers peer.Peerstore, insecure Duplex) (*SecurePipe, error) {
+	peers peer.Peerstore, insecure pipes.Duplex) (*SecurePipe, error) {
 
 	ctx, cancel := context.WithCancel(ctx)
 
 	sp := &SecurePipe{
-		Duplex: Duplex{
+		Duplex: pipes.Duplex{
 			In:  make(chan []byte, bufsize),
 			Out: make(chan []byte, bufsize),
 		},

crypto/spipe/signedpipe.go

@@ -5,17 +5,18 @@ import (
 	"crypto/rand"
 	"errors"
 
-	"code.google.com/p/goprotobuf/proto"
+	"github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/goprotobuf/proto"
 
 	"github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
 	ci "github.com/jbenet/go-ipfs/crypto"
 	pb "github.com/jbenet/go-ipfs/crypto/spipe/internal/pb"
 	"github.com/jbenet/go-ipfs/peer"
+	"github.com/jbenet/go-ipfs/pipes"
 )
 
 type SignedPipe struct {
-	Duplex
-	insecure Duplex
+	pipes.Duplex
+	insecure pipes.Duplex
 
 	local  peer.Peer
 	remote peer.Peer
@@ -26,12 +27,12 @@ type SignedPipe struct {
 }
 
 func NewSignedPipe(parctx context.Context, bufsize int, local peer.Peer,
-	peers peer.Peerstore, insecure Duplex) (*SignedPipe, error) {
+	peers peer.Peerstore, insecure pipes.Duplex) (*SignedPipe, error) {
 
 	ctx, cancel := context.WithCancel(parctx)
 
 	sp := &SignedPipe{
-		Duplex: Duplex{
+		Duplex: pipes.Duplex{
 			In:  make(chan []byte, bufsize),
 			Out: make(chan []byte, bufsize),
 		},

crypto/spipe/spipe_test.go

@@ -7,6 +7,7 @@ import (
 
 	ci "github.com/jbenet/go-ipfs/crypto"
 	"github.com/jbenet/go-ipfs/peer"
+	"github.com/jbenet/go-ipfs/pipes"
 	"github.com/jbenet/go-ipfs/util"
 )
 
@@ -24,7 +25,7 @@ func getPeer(tb testing.TB) peer.Peer {
 	return p
 }
 
-func bindDuplexNoCopy(a, b Duplex) {
+func bindDuplexNoCopy(a, b pipes.Duplex) {
 	go func() {
 		for m := range b.Out {
 			a.In <- m
@@ -35,7 +36,7 @@ func bindDuplexNoCopy(a, b Duplex) {
 	}
 }
 
-func bindDuplexWithCopy(a, b Duplex) {
+func bindDuplexWithCopy(a, b pipes.Duplex) {
 	dup := func(byt []byte) []byte {
 		n := make([]byte, len(byt))
 		copy(n, byt)
@@ -82,14 +83,8 @@ func runEncryptBenchmark(b *testing.B) {
 
 	pa := getPeer(b)
 	pb := getPeer(b)
-	duplexa := Duplex{
-		In:  make(chan []byte),
-		Out: make(chan []byte),
-	}
-	duplexb := Duplex{
-		In:  make(chan []byte),
-		Out: make(chan []byte),
-	}
+	duplexa := pipes.NewDuplex(16)
+	duplexb := pipes.NewDuplex(16)
 
 	go bindDuplexNoCopy(duplexa, duplexb)
 
@@ -140,14 +135,8 @@ func BenchmarkSignedChannel(b *testing.B) {
 
 	pa := getPeer(b)
 	pb := getPeer(b)
-	duplexa := Duplex{
-		In:  make(chan []byte),
-		Out: make(chan []byte),
-	}
-	duplexb := Duplex{
-		In:  make(chan []byte),
-		Out: make(chan []byte),
-	}
+	duplexa := pipes.NewDuplex(16)
+	duplexb := pipes.NewDuplex(16)
 
 	go bindDuplexNoCopy(duplexa, duplexb)
 
@@ -192,14 +181,8 @@ func BenchmarkSignedChannel(b *testing.B) {
 }
 
 func BenchmarkDataTransfer(b *testing.B) {
-	duplexa := Duplex{
-		In:  make(chan []byte),
-		Out: make(chan []byte),
-	}
-	duplexb := Duplex{
-		In:  make(chan []byte),
-		Out: make(chan []byte),
-	}
+	duplexa := pipes.NewDuplex(16)
+	duplexb := pipes.NewDuplex(16)
 
 	go bindDuplexWithCopy(duplexa, duplexb)
 

net/conn/secure_conn.go

@@ -8,6 +8,7 @@ import (
 
 	spipe "github.com/jbenet/go-ipfs/crypto/spipe"
 	peer "github.com/jbenet/go-ipfs/peer"
+	"github.com/jbenet/go-ipfs/pipes"
 	ctxc "github.com/jbenet/go-ipfs/util/ctxcloser"
 )
 
@@ -54,7 +55,7 @@ func (c *secureConn) secureHandshake(peers peer.Peerstore) error {
 	insecureSC := c.insecure.(*singleConn)
 
 	// setup a Duplex pipe for spipe
-	insecureD := spipe.Duplex{
+	insecureD := pipes.Duplex{
 		In:  insecureSC.msgio.incoming.MsgChan,
 		Out: insecureSC.msgio.outgoing.MsgChan,
 	}

pipes/duplex.go

+package pipes
+
+// Duplex is a simple duplex channel
+type Duplex struct {
+	In  chan []byte
+	Out chan []byte
+}
+
+func NewDuplex(bufsize int) Duplex {
+	return Duplex{
+		In:  make(chan []byte, bufsize),
+		Out: make(chan []byte, bufsize),
+	}
+}

util/testutil/gen.go

@@ -2,6 +2,9 @@ package testutil
 
 import (
 	"testing"
+	crand "crypto/rand"
+
+	"github.com/jbenet/go-ipfs/peer"
 
 	ds "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore"
 	bsrv "github.com/jbenet/go-ipfs/blockservice"
@@ -16,3 +19,9 @@ func GetDAGServ(t testing.TB) dag.DAGService {
 	}
 	return dag.NewDAGService(bserv)
 }
+
+func RandPeer() peer.Peer {
+	id := make(peer.ID, 16)
+	crand.Read(id)
+	return peer.WithID(id)
+}
\ No newline at end of file