veyron/runtimes/google/ipc/stream: Implement control-channel encryption.
For VIFs, perform versioning, authentication, and encrypt the control
data (VC data is already encrypted).
In this implementation, authentication/encryption is initiated by the
client.
- The VIF starts off using the standard unencrypted protocol.
- The client calculates the expected IPC version by intersecting
its own supported version range with the server's range that
is published via the endpoint. If the version suports
encryption, the client immediately begins negotiating
with the server.
Negotiation is performed using Encrypt control messages to set up a
bidirectional flow between the client and server. This is like a VC,
but it is hop-by-hop, instead of end-to-end.
- Each party sends its acceptable verson ranges together with
public data for initiating encryption. There is only one
supported encryption scheme now, but the options should
eventually include data for each of the encryption protocols
the party is willing to support.
- Currently the only encryption protocol is based on
code.google.com/p/go.crypto/nacl/box,
which uses Curve25519, XSalsa20 and Poly1305 to encrypt and
authenticate messages. The parties exchange public keys
to create a box cipher.
- Perform the standard Veyron handshake using the cipher for
encryption.
- On success, use the cipher to encrypt the control data.
RULES:
- VIF message formats never change, except in compatible ways.
New developments can only change the VIF protocol by adding
more message types.
- The format of the public data header in the Encrypt
negotiation never changes for all time.
A VIF frame has two parts, a (Type + PayloadSize) header word, and
some payload. The Type+PayloadSize is encrypted with the cipher's
XORKeyStream. The rest of the control data is encrypted with the
cipher's Seal method. This means that none of the data is observable
by an adversary, but the type and length are subject to corruption
(the rest of the data is not). This doesn't matter -- if the Type
or PayloadSize is corrupted by an adversary, the payload will be
misread, and will fail to validate.
We could potentially pass the Type and PayloadSize in the clear,
but then the framing would be observable, a (probably minor)
information leak. There is no reason to do so, we encrypt everything.
TIMES:
$ veyron go test veyron.io/veyron/veyron/runtimes/google/ipc/stream/benchmark -bench=Flow -run=X
Before:
BenchmarkFlow_1VIF_1VC_1Flow 50000 45029 ns/op 1137.02 MB/s
BenchmarkFlow_1VIF_1VC_2Flow 50000 39475 ns/op 1296.99 MB/s
BenchmarkFlow_1VIF_1VC_8Flow 50000 42355 ns/op 1208.82 MB/s
BenchmarkFlow_1VIF_2VC_2Flow 50000 40518 ns/op 1263.62 MB/s
BenchmarkFlow_1VIF_2VC_8Flow 50000 42861 ns/op 1194.55 MB/s
BenchmarkFlow_2VIF_4VC_8Flow 50000 43486 ns/op 1177.38 MB/s
BenchmarkFlow_1VIF_1VC_1FlowTLS 5000 521504 ns/op 98.18 MB/s
BenchmarkFlow_1VIF_1VC_2FlowTLS 5000 503743 ns/op 101.64 MB/s
BenchmarkFlow_1VIF_1VC_8FlowTLS 5000 508953 ns/op 100.60 MB/s
BenchmarkFlow_1VIF_2VC_2FlowTLS 5000 503255 ns/op 101.74 MB/s
BenchmarkFlow_1VIF_2VC_8FlowTLS 5000 510379 ns/op 100.32 MB/s
BenchmarkFlow_2VIF_4VC_8FlowTLS 5000 511113 ns/op 100.17 MB/s
ok veyron.io/veyron/veyron/runtimes/google/ipc/stream/benchmark 32.253s
After:
BenchmarkFlow_1VIF_1VC_1Flow 50000 43946 ns/op 1165.05 MB/s
BenchmarkFlow_1VIF_1VC_2Flow 50000 39394 ns/op 1299.67 MB/s
BenchmarkFlow_1VIF_1VC_8Flow 50000 43537 ns/op 1175.98 MB/s
BenchmarkFlow_1VIF_2VC_2Flow 50000 39945 ns/op 1281.74 MB/s
BenchmarkFlow_1VIF_2VC_8Flow 50000 41531 ns/op 1232.80 MB/s
BenchmarkFlow_2VIF_4VC_8Flow 50000 43568 ns/op 1175.17 MB/s
BenchmarkFlow_1VIF_1VC_1FlowTLS 5000 538491 ns/op 95.08 MB/s
BenchmarkFlow_1VIF_1VC_2FlowTLS 5000 522869 ns/op 97.92 MB/s
BenchmarkFlow_1VIF_1VC_8FlowTLS 5000 526982 ns/op 97.16 MB/s
BenchmarkFlow_1VIF_2VC_2FlowTLS 5000 522381 ns/op 98.01 MB/s
BenchmarkFlow_1VIF_2VC_8FlowTLS 5000 528707 ns/op 96.84 MB/s
BenchmarkFlow_2VIF_4VC_8FlowTLS 5000 530732 ns/op 96.47 MB/s
ok veyron.io/veyron/veyron/runtimes/google/ipc/stream/benchmark 33.302s
Change-Id: I3fcb0346dd23f77d549744fe061695a2dd606215
diff --git a/runtimes/google/ipc/stream/crypto/box_cipher.go b/runtimes/google/ipc/stream/crypto/box_cipher.go
new file mode 100644
index 0000000..d1e4987
--- /dev/null
+++ b/runtimes/google/ipc/stream/crypto/box_cipher.go
@@ -0,0 +1,136 @@
+package crypto
+
+import (
+ "encoding/binary"
+ "errors"
+
+ "code.google.com/p/go.crypto/nacl/box"
+ "code.google.com/p/go.crypto/salsa20/salsa"
+)
+
+// cbox implements a ControlCipher using go.crypto/nacl/box.
+type cbox struct {
+ sharedKey [32]byte
+ enc cboxStream
+ dec cboxStream
+}
+
+// cboxStream implements one stream of encryption or decryption.
+type cboxStream struct {
+ counter uint64
+ nonce [24]byte
+ // buffer is a temporary used for in-place crypto.
+ buffer []byte
+}
+
+const (
+ cboxMACSize = box.Overhead
+)
+
+var (
+ errMessageTooShort = errors.New("control cipher: message is too short")
+)
+
+func (s *cboxStream) alloc(n int) []byte {
+ if len(s.buffer) < n {
+ s.buffer = make([]byte, n*2)
+ }
+ return s.buffer[:0]
+}
+
+func (s *cboxStream) currentNonce() *[24]byte {
+ return &s.nonce
+}
+
+func (s *cboxStream) advanceNonce() {
+ s.counter++
+ binary.LittleEndian.PutUint64(s.nonce[:], s.counter)
+}
+
+// setupXSalsa20 produces a sub-key and Salsa20 counter given a nonce and key.
+//
+// See, "Extending the Salsa20 nonce," by Daniel J. Bernsten, Department of
+// Computer Science, University of Illinois at Chicago, 2008.
+func setupXSalsa20(subKey *[32]byte, counter *[16]byte, nonce *[24]byte, key *[32]byte) {
+ // We use XSalsa20 for encryption so first we need to generate a
+ // key and nonce with HSalsa20.
+ var hNonce [16]byte
+ copy(hNonce[:], nonce[:])
+ salsa.HSalsa20(subKey, &hNonce, key, &salsa.Sigma)
+
+ // The final 8 bytes of the original nonce form the new nonce.
+ copy(counter[:], nonce[16:])
+}
+
+// NewControlCipher returns a ControlCipher for IPC V6.
+func NewControlCipherIPC6(peersPublicKey, privateKey *[32]byte, isServer bool) ControlCipher {
+ var c cbox
+ box.Precompute(&c.sharedKey, peersPublicKey, privateKey)
+ // The stream is full-duplex, and we want the directions to use different
+ // nonces, so we set bit (1 << 64) in the server-to-client stream, and leave
+ // it cleared in the client-to-server stream. advanceNone touches only the
+ // first 8 bytes, so this change is permanent for the duration of the
+ // stream.
+ if isServer {
+ c.enc.nonce[8] = 1
+ } else {
+ c.dec.nonce[8] = 1
+ }
+ return &c
+}
+
+// MACSize implements the ControlCipher method.
+func (c *cbox) MACSize() int {
+ return cboxMACSize
+}
+
+// Seal implements the ControlCipher method.
+func (c *cbox) Seal(data []byte) error {
+ n := len(data)
+ if n < cboxMACSize {
+ return errMessageTooShort
+ }
+ tmp := c.enc.alloc(n)
+ nonce := c.enc.currentNonce()
+ out := box.SealAfterPrecomputation(tmp, data[:n-cboxMACSize], nonce, &c.sharedKey)
+ c.enc.advanceNonce()
+ copy(data, out)
+ return nil
+}
+
+// Open implements the ControlCipher method.
+func (c *cbox) Open(data []byte) bool {
+ n := len(data)
+ if n < cboxMACSize {
+ return false
+ }
+ tmp := c.dec.alloc(n - cboxMACSize)
+ nonce := c.dec.currentNonce()
+ out, ok := box.OpenAfterPrecomputation(tmp, data, nonce, &c.sharedKey)
+ if !ok {
+ return false
+ }
+ c.dec.advanceNonce()
+ copy(data, out)
+ return true
+}
+
+// Encrypt implements the ControlCipher method.
+func (c *cbox) Encrypt(data []byte) {
+ var subKey [32]byte
+ var counter [16]byte
+ nonce := c.enc.currentNonce()
+ setupXSalsa20(&subKey, &counter, nonce, &c.sharedKey)
+ c.enc.advanceNonce()
+ salsa.XORKeyStream(data, data, &counter, &subKey)
+}
+
+// Decrypt implements the ControlCipher method.
+func (c *cbox) Decrypt(data []byte) {
+ var subKey [32]byte
+ var counter [16]byte
+ nonce := c.dec.currentNonce()
+ setupXSalsa20(&subKey, &counter, nonce, &c.sharedKey)
+ c.dec.advanceNonce()
+ salsa.XORKeyStream(data, data, &counter, &subKey)
+}
diff --git a/runtimes/google/ipc/stream/crypto/box_cipher_test.go b/runtimes/google/ipc/stream/crypto/box_cipher_test.go
new file mode 100644
index 0000000..795c9c3
--- /dev/null
+++ b/runtimes/google/ipc/stream/crypto/box_cipher_test.go
@@ -0,0 +1,129 @@
+package crypto_test
+
+import (
+ "bytes"
+ "crypto/rand"
+ "testing"
+
+ "code.google.com/p/go.crypto/nacl/box"
+
+ "veyron.io/veyron/veyron/runtimes/google/ipc/stream/crypto"
+)
+
+// Add space for a MAC.
+func newMessage(s string) []byte {
+ b := make([]byte, len(s)+box.Overhead)
+ copy(b, []byte(s))
+ return b
+}
+
+func TestOpenSeal(t *testing.T) {
+ pub1, pvt1, err := box.GenerateKey(rand.Reader)
+ if err != nil {
+ t.Fatalf("can't generate key")
+ }
+ pub2, pvt2, err := box.GenerateKey(rand.Reader)
+ if err != nil {
+ t.Fatalf("can't generate key")
+ }
+ c1 := crypto.NewControlCipherIPC6(pub2, pvt1, true)
+ c2 := crypto.NewControlCipherIPC6(pub1, pvt2, false)
+
+ msg1 := newMessage("hello")
+ if err := c1.Seal(msg1); err != nil {
+ t.Errorf("unexpected error: %s", err)
+ }
+ msg2 := newMessage("world")
+ if err := c1.Seal(msg2); err != nil {
+ t.Errorf("unexpected error: %s", err)
+ }
+ msg3 := newMessage("hello")
+ if err := c1.Seal(msg3); err != nil {
+ t.Errorf("unexpected error: %s", err)
+ }
+ if bytes.Compare(msg1, msg3) == 0 {
+ t.Errorf("message should differ: %q, %q", msg1, msg3)
+ }
+
+ // Check that the client does not encrypt the same.
+ msg4 := newMessage("hello")
+ if err := c2.Seal(msg4); err != nil {
+ t.Errorf("unexpected error: %s", err)
+ }
+ if bytes.Compare(msg4, msg1) == 0 {
+ t.Errorf("messages should differ %q vs. %q", msg4, msg1)
+ }
+
+ // Corrupted message should not decrypt.
+ msg1[0] ^= 1
+ if ok := c2.Open(msg1); ok {
+ t.Errorf("expected error")
+ }
+
+ // Fix the message and try again.
+ msg1[0] ^= 1
+ if ok := c2.Open(msg1); !ok {
+ t.Errorf("Open failed")
+ }
+ if bytes.Compare(msg1[:5], []byte("hello")) != 0 {
+ t.Errorf("got %q, expected %q", msg1[:5], "hello")
+ }
+
+ // msg3 should not decrypt.
+ if ok := c2.Open(msg3); ok {
+ t.Errorf("expected error")
+ }
+
+ // Resume.
+ if ok := c2.Open(msg2); !ok {
+ t.Errorf("Open failed")
+ }
+ if bytes.Compare(msg2[:5], []byte("world")) != 0 {
+ t.Errorf("got %q, expected %q", msg2[:5], "world")
+ }
+ if ok := c2.Open(msg3); !ok {
+ t.Errorf("Open failed")
+ }
+ if bytes.Compare(msg3[:5], []byte("hello")) != 0 {
+ t.Errorf("got %q, expected %q", msg3[:5], "hello")
+ }
+}
+
+func TestXORKeyStream(t *testing.T) {
+ pub1, pvt1, err := box.GenerateKey(rand.Reader)
+ if err != nil {
+ t.Fatalf("can't generate key")
+ }
+ pub2, pvt2, err := box.GenerateKey(rand.Reader)
+ if err != nil {
+ t.Fatalf("can't generate key")
+ }
+ c1 := crypto.NewControlCipherIPC6(pub2, pvt1, true)
+ c2 := crypto.NewControlCipherIPC6(pub1, pvt2, false)
+
+ msg1 := []byte("hello")
+ msg2 := []byte("world")
+ msg3 := []byte("hello")
+ c1.Encrypt(msg1)
+ c1.Encrypt(msg2)
+ c1.Encrypt(msg3)
+ if bytes.Compare(msg1, msg3) == 0 {
+ t.Errorf("messages should differ: %q, %q", msg1, msg3)
+ }
+
+ c2.Decrypt(msg1)
+ c2.Decrypt(msg2)
+ c2.Decrypt(msg3)
+ s1 := string(msg1)
+ s2 := string(msg2)
+ s3 := string(msg3)
+ if s1 != "hello" {
+ t.Errorf("got %q, expected 'hello'", s1)
+ }
+ if s2 != "world" {
+ t.Errorf("got %q, expected 'world'", s2)
+ }
+ if s3 != "hello" {
+ t.Errorf("got %q, expected 'hello'", s3)
+ }
+}
diff --git a/runtimes/google/ipc/stream/crypto/control_cipher.go b/runtimes/google/ipc/stream/crypto/control_cipher.go
new file mode 100644
index 0000000..8258bcf
--- /dev/null
+++ b/runtimes/google/ipc/stream/crypto/control_cipher.go
@@ -0,0 +1,23 @@
+package crypto
+
+// ControlCipher provides the ciphers and MAC for control channel encryption.
+// Encryption and decryption are performed in place.
+type ControlCipher interface {
+ // MACSize returns the number of bytes in the MAC.
+ MACSize() int
+
+ // Seal replaces the message with an authenticated and encrypted version.
+ // The trailing MACSize bytes of the data are used for the MAC; they are
+ // discarded and overwritten.
+ Seal(data []byte) error
+
+ // Open authenticates and decrypts a box produced by Seal. The trailing
+ // MACSize bytes are not changed. Returns true on success.
+ Open(data []byte) bool
+
+ // Encrypt encrypts the data in place. No MAC is added.
+ Encrypt(data []byte)
+
+ // Decrypt decrypts the data in place. No MAC is verified.
+ Decrypt(data []byte)
+}
diff --git a/runtimes/google/ipc/stream/crypto/crypto.go b/runtimes/google/ipc/stream/crypto/crypto.go
index b191bf2..0d46e64 100644
--- a/runtimes/google/ipc/stream/crypto/crypto.go
+++ b/runtimes/google/ipc/stream/crypto/crypto.go
@@ -26,7 +26,7 @@
Encrypter
Decrypter
// ChannelBinding Returns a byte slice that is unique for the the
- // particular crypter (and the parties between which it is operationg).
+ // particular crypter (and the parties between which it is operating).
// Having both parties assert out of the band that they are indeed
// participating in a connection with that channel binding value is
// sufficient to authenticate the data received through the crypter.
diff --git a/runtimes/google/ipc/stream/crypto/null_cipher.go b/runtimes/google/ipc/stream/crypto/null_cipher.go
new file mode 100644
index 0000000..9a39cc7
--- /dev/null
+++ b/runtimes/google/ipc/stream/crypto/null_cipher.go
@@ -0,0 +1,23 @@
+package crypto
+
+// NullControlCipher is a cipher that does nothing.
+type NullControlCipher struct{}
+
+func (NullControlCipher) MACSize() int { return 0 }
+func (NullControlCipher) Seal(data []byte) error { return nil }
+func (NullControlCipher) Open(data []byte) bool { return true }
+func (NullControlCipher) Encrypt(data []byte) {}
+func (NullControlCipher) Decrypt(data []byte) {}
+
+type disabledControlCipher struct {
+ NullControlCipher
+ macSize int
+}
+
+func (c *disabledControlCipher) MACSize() int { return c.macSize }
+
+// NewDisabledControlCipher returns a cipher that has the correct MACSize, but
+// encryption and decryption are disabled.
+func NewDisabledControlCipher(c ControlCipher) ControlCipher {
+ return &disabledControlCipher{macSize: c.MACSize()}
+}
diff --git a/runtimes/google/ipc/stream/manager/listener.go b/runtimes/google/ipc/stream/manager/listener.go
index a0f2789..bd6d272 100644
--- a/runtimes/google/ipc/stream/manager/listener.go
+++ b/runtimes/google/ipc/stream/manager/listener.go
@@ -157,7 +157,7 @@
vlog.VI(1).Infof("Connecting to proxy at %v", ln.proxyEP)
// TODO(cnicolaou, ashankar): probably want to set a timeout here.
var timeout time.Duration
- vf, err := ln.manager.FindOrDialVIF(ln.proxyEP.Addr(), timeout)
+ vf, err := ln.manager.FindOrDialVIF(ln.proxyEP, &DialTimeout{Duration: timeout})
if err != nil {
return nil, nil, err
}
diff --git a/runtimes/google/ipc/stream/manager/manager.go b/runtimes/google/ipc/stream/manager/manager.go
index cfb4ad0..8620a61 100644
--- a/runtimes/google/ipc/stream/manager/manager.go
+++ b/runtimes/google/ipc/stream/manager/manager.go
@@ -16,6 +16,7 @@
inaming "veyron.io/veyron/veyron/runtimes/google/naming"
"veyron.io/veyron/veyron2/ipc/stream"
+ ipcversion "veyron.io/veyron/veyron2/ipc/version"
"veyron.io/veyron/veyron2/naming"
"veyron.io/veyron/veyron2/verror"
"veyron.io/veyron/veyron2/vlog"
@@ -70,7 +71,17 @@
// FindOrDialVIF returns the network connection (VIF) to the provided address
// from the cache in the manager. If not already present in the cache, a new
// connection will be created using net.Dial.
-func (m *manager) FindOrDialVIF(addr net.Addr, timeout time.Duration) (*vif.VIF, error) {
+func (m *manager) FindOrDialVIF(remote naming.Endpoint, opts ...stream.VCOpt) (*vif.VIF, error) {
+ // Extract options.
+ var timeout time.Duration
+ for _, o := range opts {
+ switch v := o.(type) {
+ case *DialTimeout:
+ timeout = v.Duration
+ }
+ }
+
+ addr := remote.Addr()
network, address := addr.Network(), addr.String()
if vf := m.vifs.Find(network, address); vf != nil {
return vf, nil
@@ -92,7 +103,22 @@
conn.Close()
return vf, nil
}
- vf, err := vif.InternalNewDialedVIF(conn, m.rid, nil)
+ vRange := version.SupportedRange
+ if ep, ok := remote.(*inaming.Endpoint); ok {
+ epRange := &version.Range{Min: ep.MinIPCVersion, Max: ep.MaxIPCVersion}
+ if epRange.Max == ipcversion.UnknownIPCVersion {
+ // If the server's version is unknown, we need to back off to
+ // something we are sure it can understand.
+ //
+ // TODO(jyh): Remove this once the min version is IPC6, where we can
+ // support version negotiation.
+ epRange.Max = ipcversion.IPCVersion5
+ }
+ if r, err := vRange.Intersect(epRange); err == nil {
+ vRange = r
+ }
+ }
+ vf, err := vif.InternalNewDialedVIF(conn, m.rid, vRange, opts...)
if err != nil {
conn.Close()
return nil, fmt.Errorf("failed to create VIF: %v", err)
@@ -110,17 +136,10 @@
}
func (m *manager) Dial(remote naming.Endpoint, opts ...stream.VCOpt) (stream.VC, error) {
- var timeout time.Duration
- for _, o := range opts {
- switch v := o.(type) {
- case *DialTimeout:
- timeout = v.Duration
- }
- }
// If vif.Dial fails because the cached network connection was broken, remove from
// the cache and try once more.
for retry := true; true; retry = false {
- vf, err := m.FindOrDialVIF(remote.Addr(), timeout)
+ vf, err := m.FindOrDialVIF(remote, opts...)
if err != nil {
return nil, err
}
diff --git a/runtimes/google/ipc/stream/message/coding.go b/runtimes/google/ipc/stream/message/coding.go
index 378bd03..23eb4a5 100644
--- a/runtimes/google/ipc/stream/message/coding.go
+++ b/runtimes/google/ipc/stream/message/coding.go
@@ -139,3 +139,12 @@
}
return
}
+
+func readAndDiscardToError(r io.Reader) {
+ var data [1024]byte
+ for {
+ if _, err := r.Read(data[:]); err != nil {
+ return
+ }
+ }
+}
diff --git a/runtimes/google/ipc/stream/message/control.go b/runtimes/google/ipc/stream/message/control.go
index ce58df0..86f849d 100644
--- a/runtimes/google/ipc/stream/message/control.go
+++ b/runtimes/google/ipc/stream/message/control.go
@@ -6,13 +6,14 @@
"io"
"veyron.io/veyron/veyron/runtimes/google/ipc/stream/id"
+ "veyron.io/veyron/veyron/runtimes/google/ipc/version"
inaming "veyron.io/veyron/veyron/runtimes/google/naming"
"veyron.io/veyron/veyron2/naming"
)
// Control is the interface implemented by all control messages.
type Control interface {
- readFrom(r io.Reader) error
+ readFrom(r *bytes.Buffer) error
writeTo(w io.Writer) error
}
@@ -50,6 +51,53 @@
InitialCounters uint32
}
+// HopSetup is a control message used to negotiate VIF options on a
+// hop-by-hop basis.
+type HopSetup struct {
+ Versions version.Range
+ Options []HopSetupOption
+}
+
+// HopSetupOption is the base interface for optional HopSetup options.
+type HopSetupOption interface {
+ // code is the identifier for the option.
+ code() hopSetupOptionCode
+
+ // size returns the number of bytes needed to represent the option.
+ size() uint16
+
+ // write the option to the writer.
+ write(w io.Writer) error
+
+ // read the option from the reader.
+ read(r io.Reader) error
+}
+
+// NaclBox is a HopSetupOption that specifies the public key for the NaclBox
+// encryption protocol.
+type NaclBox struct {
+ PublicKey [32]byte
+}
+
+// HopSetupStream is a byte stream used to negotiate VIF setup. During VIF setup,
+// each party sends a HopSetup message to the other party containing their version
+// and options. If the version requires further negotiation (such as for authentication),
+// the HopSetupStream is used for the negotiation.
+//
+// The protocol used on the stream is version-specific, it is not specified here. See
+// vif/auth.go for an example.
+type HopSetupStream struct {
+ Data []byte
+}
+
+// Setup option codes.
+type hopSetupOptionCode uint16
+
+const (
+ naclBoxPublicKey hopSetupOptionCode = 0
+)
+
+// Command enum.
type command uint8
const (
@@ -57,6 +105,8 @@
closeVCCommand command = 1
addReceiveBuffersCommand command = 2
openFlowCommand command = 3
+ hopSetupCommand command = 4
+ hopSetupStreamCommand command = 5
)
func writeControl(w io.Writer, m Control) error {
@@ -70,6 +120,10 @@
command = addReceiveBuffersCommand
case *OpenFlow:
command = openFlowCommand
+ case *HopSetup:
+ command = hopSetupCommand
+ case *HopSetupStream:
+ command = hopSetupStreamCommand
default:
return fmt.Errorf("unrecognized VC control message: %T", m)
}
@@ -101,6 +155,10 @@
m = new(AddReceiveBuffers)
case openFlowCommand:
m = new(OpenFlow)
+ case hopSetupCommand:
+ m = new(HopSetup)
+ case hopSetupStreamCommand:
+ m = new(HopSetupStream)
default:
return nil, fmt.Errorf("unrecognized VC control message command(%d)", command)
}
@@ -126,7 +184,7 @@
return nil
}
-func (m *OpenVC) readFrom(r io.Reader) (err error) {
+func (m *OpenVC) readFrom(r *bytes.Buffer) (err error) {
if err = readInt(r, &m.VCI); err != nil {
return
}
@@ -159,7 +217,7 @@
return
}
-func (m *CloseVC) readFrom(r io.Reader) (err error) {
+func (m *CloseVC) readFrom(r *bytes.Buffer) (err error) {
if err = readInt(r, &m.VCI); err != nil {
return
}
@@ -173,7 +231,7 @@
return writeCounters(w, m.Counters)
}
-func (m *AddReceiveBuffers) readFrom(r io.Reader) (err error) {
+func (m *AddReceiveBuffers) readFrom(r *bytes.Buffer) (err error) {
m.Counters, err = readCounters(r)
return
}
@@ -191,7 +249,7 @@
return
}
-func (m *OpenFlow) readFrom(r io.Reader) (err error) {
+func (m *OpenFlow) readFrom(r *bytes.Buffer) (err error) {
if err = readInt(r, &m.VCI); err != nil {
return
}
@@ -203,3 +261,99 @@
}
return
}
+
+func (m *HopSetup) writeTo(w io.Writer) error {
+ if err := writeInt(w, m.Versions.Min); err != nil {
+ return err
+ }
+ if err := writeInt(w, m.Versions.Max); err != nil {
+ return err
+ }
+ for _, opt := range m.Options {
+ if err := writeInt(w, opt.code()); err != nil {
+ return err
+ }
+ if err := writeInt(w, opt.size()); err != nil {
+ return err
+ }
+ if err := opt.write(w); err != nil {
+ return err
+ }
+ }
+ return nil
+}
+
+func (m *HopSetup) readFrom(r *bytes.Buffer) error {
+ if err := readInt(r, &m.Versions.Min); err != nil {
+ return err
+ }
+ if err := readInt(r, &m.Versions.Max); err != nil {
+ return err
+ }
+ for {
+ var code hopSetupOptionCode
+ err := readInt(r, &code)
+ if err == io.EOF {
+ return nil
+ }
+ if err != nil {
+ return err
+ }
+
+ var size uint16
+ if err := readInt(r, &size); err != nil {
+ return err
+ }
+ l := &io.LimitedReader{R: r, N: int64(size)}
+
+ switch code {
+ case naclBoxPublicKey:
+ var opt NaclBox
+ if err := opt.read(l); err != nil {
+ return err
+ }
+ m.Options = append(m.Options, &opt)
+ }
+
+ // Consume any data remaining.
+ readAndDiscardToError(l)
+ }
+}
+
+// NaclBox returns the first NaclBox option, or nil if there is none.
+func (m *HopSetup) NaclBox() *NaclBox {
+ for _, opt := range m.Options {
+ if b, ok := opt.(*NaclBox); ok {
+ return b
+ }
+ }
+ return nil
+}
+
+func (*NaclBox) code() hopSetupOptionCode {
+ return naclBoxPublicKey
+}
+
+func (m *NaclBox) size() uint16 {
+ return uint16(len(m.PublicKey))
+}
+
+func (m *NaclBox) write(w io.Writer) error {
+ _, err := w.Write(m.PublicKey[:])
+ return err
+}
+
+func (m *NaclBox) read(r io.Reader) error {
+ _, err := io.ReadFull(r, m.PublicKey[:])
+ return err
+}
+
+func (m *HopSetupStream) writeTo(w io.Writer) error {
+ _, err := w.Write(m.Data)
+ return err
+}
+
+func (m *HopSetupStream) readFrom(r *bytes.Buffer) error {
+ m.Data = r.Bytes()
+ return nil
+}
diff --git a/runtimes/google/ipc/stream/message/message.go b/runtimes/google/ipc/stream/message/message.go
index db5d58b..7c5358b 100644
--- a/runtimes/google/ipc/stream/message/message.go
+++ b/runtimes/google/ipc/stream/message/message.go
@@ -20,7 +20,11 @@
// +---------------------------------------------+
// | 0 | PayloadSize (3 bytes) |
// +---------------------------------------------+
-// | Cmd (1 byte) | Data (PayloadSize - 1 bytes)|
+// | Cmd (1 byte) |
+// +---------------------------------------------+
+// | Data (PayloadSize - MACSize - 1 bytes) |
+// +---------------------------------------------+
+// | MAC (MACSize bytes) |
// +---------------------------------------------+
// Where Data is the serialized Control interface object.
//
@@ -28,19 +32,39 @@
// +---------------------------------------------+
// | 1 | PayloadSize (3 bytes) |
// +---------------------------------------------+
-// | id.VCI (4-bytes) |
+// | id.VCI (4 bytes) |
// +---------------------------------------------+
-// | id.Flow (4-bytes) |
+// | id.Flow (4 bytes) |
// +---------------------------------------------+
-// | Flags (1 byte)| Data (PayloadSize - 9 bytes)|
+// | Flags (1 byte) |
// +---------------------------------------------+
-// Where Data is the application data.
+// | MAC (MACSize bytes) |
+// +---------------------------------------------+
+// | Data (PayloadSize - 9 - MACSize bytes) |
+// +---------------------------------------------+
+// Where Data is the application data. The Data is encrypted separately; it is
+// not included in the MAC.
+//
+// A crypto.ControlCipher is used to encrypt the control data. The MACSize
+// comes from the ControlCipher. When used, the first word of the header,
+// containing the Type and PayloadSize, is encrypted with the cipher's Encrypt
+// method. The rest of the control data is encrypted with the cipher's Seal
+// method. This means that none of the data is observable by an adversary, but
+// the type and length are subject to corruption (the rest of the data is not).
+// This doesn't matter -- if the Type or PayloadSize is corrupted by an
+// adversary, the payload will be misread, and will fail to validate.
+//
+// We could potentially pass the Type and PayloadSize in the clear, but then the
+// framing would be observable, a (probably minor) information leak. There is
+// no reason to do so, we encrypt everything.
import (
"bytes"
+ "errors"
"fmt"
"io"
+ "veyron.io/veyron/veyron/runtimes/google/ipc/stream/crypto"
"veyron.io/veyron/veyron/runtimes/google/ipc/stream/id"
"veyron.io/veyron/veyron/runtimes/google/lib/iobuf"
"veyron.io/veyron/veyron2/vlog"
@@ -58,6 +82,11 @@
dataType = 1
)
+var (
+ emptyMessageErr = errors.New("message is empty")
+ corruptedMessageErr = errors.New("corrupted message")
+)
+
// T is the interface implemented by all messages communicated over a VIF.
type T interface {
}
@@ -75,11 +104,12 @@
// case Control:
// handleControlMessage(m)
// }
-func ReadFrom(r *iobuf.Reader) (T, error) {
+func ReadFrom(r *iobuf.Reader, c crypto.ControlCipher) (T, error) {
header, err := r.Read(commonHeaderSizeBytes)
if err != nil {
return nil, fmt.Errorf("failed to read VC header: %v", err)
}
+ c.Decrypt(header.Contents)
msgType := header.Contents[0]
msgPayloadSize := read3ByteUint(header.Contents[1:4])
header.Release()
@@ -87,19 +117,28 @@
if err != nil {
return nil, fmt.Errorf("failed to read payload of %d bytes for type %d: %v", msgPayloadSize, msgType, err)
}
+ macSize := c.MACSize()
switch msgType {
case controlType:
- m, err := readControl(bytes.NewBuffer(payload.Contents))
+ if !c.Open(payload.Contents) {
+ payload.Release()
+ return nil, corruptedMessageErr
+ }
+ m, err := readControl(bytes.NewBuffer(payload.Contents[:msgPayloadSize-macSize]))
payload.Release()
return m, err
case dataType:
+ if !c.Open(payload.Contents[0 : dataHeaderSizeBytes+macSize]) {
+ payload.Release()
+ return nil, corruptedMessageErr
+ }
m := &Data{
VCI: id.VC(read4ByteUint(payload.Contents[0:4])),
Flow: id.Flow(read4ByteUint(payload.Contents[4:8])),
flags: payload.Contents[8],
Payload: payload,
}
- m.Payload.TruncateFront(dataHeaderSizeBytes)
+ m.Payload.TruncateFront(uint(dataHeaderSizeBytes + macSize))
return m, nil
default:
payload.Release()
@@ -107,8 +146,6 @@
}
}
-var headerSpace [commonHeaderSizeBytes]byte
-
// WriteTo serializes message and makes a single call to w.Write.
// It is the inverse of ReadFrom.
//
@@ -117,19 +154,22 @@
//
// If message is a Data message, the Payload contents will be Released
// irrespective of the return value of this method.
-func WriteTo(w io.Writer, message T) error {
+func WriteTo(w io.Writer, message T, c crypto.ControlCipher) error {
+ macSize := c.MACSize()
switch m := message.(type) {
case *Data:
- payloadSize := m.PayloadSize() + dataHeaderSizeBytes
- msg := mkHeaderSpace(m.Payload, HeaderSizeBytes)
- header := msg.Contents[0:HeaderSizeBytes]
+ payloadSize := m.PayloadSize() + dataHeaderSizeBytes + macSize
+ msg := mkHeaderSpace(m.Payload, uint(HeaderSizeBytes+macSize))
+ header := msg.Contents[0 : HeaderSizeBytes+macSize]
header[0] = dataType
if err := write3ByteUint(header[1:4], payloadSize); err != nil {
return err
+
}
write4ByteUint(header[4:8], uint32(m.VCI))
write4ByteUint(header[8:12], uint32(m.Flow))
header[12] = m.flags
+ EncryptMessage(msg.Contents, c)
_, err := w.Write(msg.Contents)
msg.Release()
return err
@@ -139,17 +179,21 @@
// something that is large enough for typical control messages.
buf.Grow(256)
// Reserve space for the header
- if _, err := buf.Write(headerSpace[:]); err != nil {
+ if err := extendBuffer(&buf, commonHeaderSizeBytes); err != nil {
return err
}
if err := writeControl(&buf, m); err != nil {
return err
}
+ if err := extendBuffer(&buf, macSize); err != nil {
+ return err
+ }
msg := buf.Bytes()
msg[0] = controlType
if err := write3ByteUint(msg[1:4], buf.Len()-commonHeaderSizeBytes); err != nil {
return err
}
+ EncryptMessage(msg, c)
_, err := w.Write(msg)
return err
default:
@@ -158,6 +202,25 @@
return nil
}
+// EncryptMessage encrypts the message's control data in place.
+func EncryptMessage(msg []byte, c crypto.ControlCipher) error {
+ if len(msg) == 0 {
+ return emptyMessageErr
+ }
+ n := len(msg)
+ switch msgType := msg[0]; msgType {
+ case controlType:
+ // skip
+ case dataType:
+ n = HeaderSizeBytes + c.MACSize()
+ default:
+ return fmt.Errorf("unrecognized message type: %d", msgType)
+ }
+ c.Encrypt(msg[0:commonHeaderSizeBytes])
+ c.Seal(msg[commonHeaderSizeBytes:n])
+ return nil
+}
+
func mkHeaderSpace(slice *iobuf.Slice, space uint) *iobuf.Slice {
if slice == nil {
return iobuf.NewSlice(make([]byte, space))
@@ -171,3 +234,10 @@
slice.Release()
return iobuf.NewSlice(contents)
}
+
+var emptyBytes [256]byte
+
+func extendBuffer(buf *bytes.Buffer, size int) error {
+ _, err := buf.Write(emptyBytes[:size])
+ return err
+}
diff --git a/runtimes/google/ipc/stream/message/message_test.go b/runtimes/google/ipc/stream/message/message_test.go
index bd1fba8..9e98320 100644
--- a/runtimes/google/ipc/stream/message/message_test.go
+++ b/runtimes/google/ipc/stream/message/message_test.go
@@ -2,6 +2,7 @@
import (
"bytes"
+ "encoding/binary"
"reflect"
"testing"
@@ -10,6 +11,54 @@
"veyron.io/veyron/veyron2/naming"
)
+// testControlCipher is a super-simple cipher that xor's each byte of the
+// payload with 0xaa.
+type testControlCipher struct{}
+
+const testMACSize = 4
+
+func (*testControlCipher) MACSize() int {
+ return testMACSize
+}
+
+func testMAC(data []byte) []byte {
+ var h uint32
+ for _, b := range data {
+ h = (h << 1) ^ uint32(b)
+ }
+ var hash [4]byte
+ binary.BigEndian.PutUint32(hash[:], h)
+ return hash[:]
+}
+
+func (c *testControlCipher) Decrypt(data []byte) {
+ for i, _ := range data {
+ data[i] ^= 0xaa
+ }
+}
+
+func (c *testControlCipher) Encrypt(data []byte) {
+ for i, _ := range data {
+ data[i] ^= 0xaa
+ }
+}
+
+func (c *testControlCipher) Open(data []byte) bool {
+ mac := testMAC(data[:len(data)-testMACSize])
+ if bytes.Compare(mac, data[len(data)-testMACSize:]) != 0 {
+ return false
+ }
+ c.Decrypt(data[:len(data)-testMACSize])
+ return true
+}
+
+func (c *testControlCipher) Seal(data []byte) error {
+ c.Encrypt(data[:len(data)-testMACSize])
+ mac := testMAC(data[:len(data)-testMACSize])
+ copy(data[len(data)-testMACSize:], mac)
+ return nil
+}
+
func TestControl(t *testing.T) {
counters := NewCounters()
counters.Add(12, 13, 10240)
@@ -32,17 +81,28 @@
&AddReceiveBuffers{Counters: counters},
&OpenFlow{VCI: 1, Flow: 10, InitialCounters: 1 << 24},
+
+ &HopSetup{
+ Versions: version.Range{Min: 21, Max: 71},
+ Options: []HopSetupOption{
+ &NaclBox{PublicKey: [32]byte{'h', 'e', 'l', 'l', 'o', 'w', 'o', 'r', 'l', 'd'}},
+ &NaclBox{PublicKey: [32]byte{7, 67, 31}},
+ },
+ },
+
+ &HopSetupStream{Data: []byte("HelloWorld")},
}
+ var c testControlCipher
pool := iobuf.NewPool(0)
for i, msg := range tests {
var buf bytes.Buffer
- if err := WriteTo(&buf, msg); err != nil {
+ if err := WriteTo(&buf, msg, &c); err != nil {
t.Errorf("WriteTo(%T) (test #%d) failed: %v", msg, i, err)
continue
}
reader := iobuf.NewReader(pool, &buf)
- read, err := ReadFrom(reader)
+ read, err := ReadFrom(reader, &c)
reader.Close()
if err != nil {
t.Errorf("ReadFrom failed (test #%d): %v", i, err)
@@ -63,18 +123,19 @@
{Data{VCI: 10, Flow: 3, flags: 1}, "batman"},
}
+ var c testControlCipher
pool := iobuf.NewPool(0)
- allocator := iobuf.NewAllocator(pool, HeaderSizeBytes)
+ allocator := iobuf.NewAllocator(pool, HeaderSizeBytes+testMACSize)
for i, test := range tests {
var buf bytes.Buffer
msgW := test.Header
msgW.Payload = allocator.Copy([]byte(test.Payload))
- if err := WriteTo(&buf, &msgW); err != nil {
+ if err := WriteTo(&buf, &msgW, &c); err != nil {
t.Errorf("WriteTo(%v) failed: %v", i, err)
continue
}
reader := iobuf.NewReader(pool, &buf)
- read, err := ReadFrom(reader)
+ read, err := ReadFrom(reader, &c)
if err != nil {
t.Errorf("ReadFrom(%v) failed: %v", i, err)
continue
@@ -100,14 +161,15 @@
{VCI: 10, Flow: 3},
{VCI: 11, Flow: 4, flags: 10},
}
+ var c testControlCipher
pool := iobuf.NewPool(0)
for _, test := range tests {
var buf bytes.Buffer
- if err := WriteTo(&buf, &test); err != nil {
+ if err := WriteTo(&buf, &test, &c); err != nil {
t.Errorf("WriteTo(%v) failed: %v", test, err)
continue
}
- read, err := ReadFrom(iobuf.NewReader(pool, &buf))
+ read, err := ReadFrom(iobuf.NewReader(pool, &buf), &c)
if err != nil {
t.Errorf("ReadFrom(%v) failed: %v", test, err)
continue
diff --git a/runtimes/google/ipc/stream/proxy/proxy.go b/runtimes/google/ipc/stream/proxy/proxy.go
index 8e9c509..a0d794e 100644
--- a/runtimes/google/ipc/stream/proxy/proxy.go
+++ b/runtimes/google/ipc/stream/proxy/proxy.go
@@ -6,16 +6,17 @@
"net"
"sync"
+ "veyron.io/veyron/veyron/runtimes/google/ipc/stream/crypto"
"veyron.io/veyron/veyron/runtimes/google/ipc/stream/id"
"veyron.io/veyron/veyron/runtimes/google/ipc/stream/message"
"veyron.io/veyron/veyron/runtimes/google/ipc/stream/vc"
+ "veyron.io/veyron/veyron/runtimes/google/ipc/stream/vif"
"veyron.io/veyron/veyron/runtimes/google/ipc/version"
"veyron.io/veyron/veyron/runtimes/google/lib/bqueue"
"veyron.io/veyron/veyron/runtimes/google/lib/bqueue/drrqueue"
"veyron.io/veyron/veyron/runtimes/google/lib/iobuf"
"veyron.io/veyron/veyron/runtimes/google/lib/upcqueue"
- "veyron.io/veyron/veyron2/ipc/stream"
"veyron.io/veyron/veyron2/naming"
"veyron.io/veyron/veyron2/security"
"veyron.io/veyron/veyron2/verror"
@@ -34,7 +35,7 @@
type Proxy struct {
ln net.Listener
rid naming.RoutingID
- principal stream.ListenerOpt
+ principal security.Principal
mu sync.RWMutex
servers *servermap
processes map[*process]struct{}
@@ -45,6 +46,8 @@
// associated with the process at the other end of the network connection.
type process struct {
Conn net.Conn
+ isSetup bool
+ ctrlCipher crypto.ControlCipher
Queue *upcqueue.T
mu sync.RWMutex
routingTable map[id.VC]*destination
@@ -141,7 +144,7 @@
servers: &servermap{m: make(map[naming.RoutingID]*server)},
processes: make(map[*process]struct{}),
pubAddress: pubAddress,
- principal: vc.LocalPrincipal{principal},
+ principal: principal,
}
go proxy.listenLoop()
return proxy, nil
@@ -155,18 +158,24 @@
proxyLog().Infof("Exiting listenLoop of proxy %q: %v", p.Endpoint(), err)
return
}
- process := &process{
- Conn: conn,
- Queue: upcqueue.New(),
- routingTable: make(map[id.VC]*destination),
- servers: make(map[id.VC]*vc.VC),
- BQ: drrqueue.New(vc.MaxPayloadSizeBytes),
- }
- go writeLoop(process)
- go serverVCsLoop(process)
- go p.readLoop(process)
+ go p.acceptProcess(conn)
}
}
+
+func (p *Proxy) acceptProcess(conn net.Conn) {
+ process := &process{
+ Conn: conn,
+ ctrlCipher: &crypto.NullControlCipher{},
+ Queue: upcqueue.New(),
+ routingTable: make(map[id.VC]*destination),
+ servers: make(map[id.VC]*vc.VC),
+ BQ: drrqueue.New(vc.MaxPayloadSizeBytes),
+ }
+ go writeLoop(process)
+ go serverVCsLoop(process)
+ go p.readLoop(process)
+}
+
func writeLoop(process *process) {
defer processLog().Infof("Exited writeLoop for %v", process)
defer process.Close()
@@ -178,7 +187,7 @@
}
return
}
- if err = message.WriteTo(process.Conn, item.(message.T)); err != nil {
+ if err = message.WriteTo(process.Conn, item.(message.T), process.ctrlCipher); err != nil {
processLog().Infof("message.WriteTo on %v failed: %v", process, err)
return
}
@@ -245,7 +254,7 @@
reader := iobuf.NewReader(iobuf.NewPool(0), process.Conn)
defer reader.Close()
for {
- msg, err := message.ReadFrom(reader)
+ msg, err := message.ReadFrom(reader, process.ctrlCipher)
if err != nil {
processLog().Infof("Read on %v failed: %v", process, err)
return
@@ -317,7 +326,7 @@
p.routeCounters(process, m.Counters)
if vcObj != nil {
server := &server{Process: process, VC: vcObj}
- go p.runServer(server, vcObj.HandshakeAcceptedVC(p.principal))
+ go p.runServer(server, vcObj.HandshakeAcceptedVC(vc.LocalPrincipal{p.principal}))
}
break
}
@@ -344,6 +353,24 @@
m.VCI = dstVCI
dstprocess.Queue.Put(m)
p.routeCounters(process, counters)
+ case *message.HopSetup:
+ // Set up the hop. This takes over the process during negotiation.
+ if process.isSetup {
+ // Already performed authentication. We don't do it again.
+ processLog().Infof("Process %v is already setup", process)
+ return
+ }
+ var blessings security.Blessings
+ if p.principal != nil {
+ blessings = p.principal.BlessingStore().Default()
+ }
+ c, err := vif.AuthenticateAsServer(process.Conn, nil, p.principal, blessings, nil, m)
+ if err != nil {
+ processLog().Infof("Process %v failed to authenticate: %s", process, err)
+ return
+ }
+ process.ctrlCipher = c
+ process.isSetup = true
default:
processLog().Infof("Closing %v because of unrecognized message %T", process, m)
return
diff --git a/runtimes/google/ipc/stream/vc/auth.go b/runtimes/google/ipc/stream/vc/auth.go
index 1871dae..cc27b09 100644
--- a/runtimes/google/ipc/stream/vc/auth.go
+++ b/runtimes/google/ipc/stream/vc/auth.go
@@ -22,17 +22,18 @@
var (
errSameChannelPublicKey = errors.New("same public keys for both ends of the channel")
errChannelIDMismatch = errors.New("channel id does not match expectation")
+ errChecksumMismatch = errors.New("checksum mismatch")
errInvalidSignatureInMessage = errors.New("signature does not verify in authentication handshake message")
errNoCertificatesReceived = errors.New("no certificates received")
errSingleCertificateRequired = errors.New("exactly one X.509 certificate chain with exactly one certificate is required")
)
-// authenticateAsServer executes the authentication protocol at the server and
+// AuthenticateAsServer executes the authentication protocol at the server and
// returns the blessings used to authenticate the client.
-func authenticateAsServer(conn io.ReadWriteCloser, principal security.Principal, server security.Blessings, dc DischargeClient, crypter crypto.Crypter, v version.IPCVersion) (client security.Blessings, clientDischarges map[string]security.Discharge, err error) {
+func AuthenticateAsServer(conn io.ReadWriteCloser, principal security.Principal, server security.Blessings, dc DischargeClient, crypter crypto.Crypter, v version.IPCVersion) (client security.Blessings, clientDischarges map[string]security.Discharge, err error) {
defer conn.Close()
if server == nil {
- return nil, nil, fmt.Errorf("BlessingStore does not contain a default set of blessings, cannot act as a server")
+ return nil, nil, errors.New("no blessings to present as a server")
}
var discharges []security.Discharge
if tpcavs := server.ThirdPartyCaveats(); len(tpcavs) > 0 && dc != nil {
@@ -47,7 +48,7 @@
return
}
-// authenticateAsClient executes the authentication protocol at the client and
+// AuthenticateAsClient executes the authentication protocol at the client and
// returns the blessings used to authenticate both ends.
//
// The client will only share its identity if its blessing store has one marked
@@ -55,7 +56,7 @@
//
// TODO(ashankar): Seems like there is no way the blessing store
// can say that it does NOT want to share the default blessing with the server?
-func authenticateAsClient(conn io.ReadWriteCloser, principal security.Principal, dc DischargeClient, crypter crypto.Crypter, v version.IPCVersion) (server, client security.Blessings, serverDischarges map[string]security.Discharge, err error) {
+func AuthenticateAsClient(conn io.ReadWriteCloser, principal security.Principal, dc DischargeClient, crypter crypto.Crypter, v version.IPCVersion) (server, client security.Blessings, serverDischarges map[string]security.Discharge, err error) {
defer conn.Close()
if server, serverDischarges, err = readBlessings(conn, authServerContextTag, crypter, v); err != nil {
return
diff --git a/runtimes/google/ipc/stream/vc/init.go b/runtimes/google/ipc/stream/vc/init.go
index ee13672..ae639c5 100644
--- a/runtimes/google/ipc/stream/vc/init.go
+++ b/runtimes/google/ipc/stream/vc/init.go
@@ -10,7 +10,7 @@
"veyron.io/veyron/veyron2/vlog"
)
-var anonymousPrincipal security.Principal
+var AnonymousPrincipal security.Principal
func init() {
key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
@@ -18,10 +18,10 @@
vlog.Fatalf("could not create private key for anonymous principal: %v", err)
}
store := &anonymousBlessingStore{k: security.NewECDSAPublicKey(&key.PublicKey)}
- if anonymousPrincipal, err = security.CreatePrincipal(security.NewInMemoryECDSASigner(key), store, nil); err != nil {
+ if AnonymousPrincipal, err = security.CreatePrincipal(security.NewInMemoryECDSASigner(key), store, nil); err != nil {
vlog.Fatalf("could not create anonymous principal: %v", err)
}
- if store.b, err = anonymousPrincipal.BlessSelf("anonymous"); err != nil {
+ if store.b, err = AnonymousPrincipal.BlessSelf("anonymous"); err != nil {
vlog.Fatalf("failed to generate the one blessing to be used by the anonymous principal: %v", err)
}
}
diff --git a/runtimes/google/ipc/stream/vc/vc.go b/runtimes/google/ipc/stream/vc/vc.go
index 8d2f84f..f5783cb 100644
--- a/runtimes/google/ipc/stream/vc/vc.go
+++ b/runtimes/google/ipc/stream/vc/vc.go
@@ -377,7 +377,7 @@
switch securityLevel {
case options.VCSecurityConfidential:
if principal == nil {
- principal = anonymousPrincipal
+ principal = AnonymousPrincipal
}
case options.VCSecurityNone:
return nil
@@ -403,14 +403,14 @@
// to readFromUntil in
// https://code.google.com/p/go/source/browse/src/pkg/crypto/tls/conn.go?spec=svn654b2703fcc466a29692068ab56efedd09fb3d05&r=654b2703fcc466a29692068ab56efedd09fb3d05#539).
// This is not a problem when tls.Conn is used as intended (to wrap over a stream), but
- // becomes a problem when shoehoring a block encrypter (Crypter interface) over this
+ // becomes a problem when shoehorning a block encrypter (Crypter interface) over this
// stream API.
authFID := vc.allocFID()
authConn, err := vc.connectFID(authFID)
if err != nil {
return vc.err(fmt.Errorf("failed to create a Flow for authentication: %v", err))
}
- rBlessings, lBlessings, rDischarges, err := authenticateAsClient(authConn, principal, dischargeClient, crypter, vc.version)
+ rBlessings, lBlessings, rDischarges, err := AuthenticateAsClient(authConn, principal, dischargeClient, crypter, vc.version)
if err != nil {
return vc.err(fmt.Errorf("authentication failed: %v", err))
}
@@ -477,7 +477,7 @@
switch securityLevel {
case options.VCSecurityConfidential:
if principal == nil {
- principal = anonymousPrincipal
+ principal = AnonymousPrincipal
}
if lBlessings == nil {
lBlessings = principal.BlessingStore().Default()
@@ -524,7 +524,7 @@
vc.mu.Lock()
vc.authFID = vc.findFlowLocked(authConn)
vc.mu.Unlock()
- rBlessings, rDischarges, err := authenticateAsServer(authConn, principal, lBlessings, dischargeClient, crypter, vc.version)
+ rBlessings, rDischarges, err := AuthenticateAsServer(authConn, principal, lBlessings, dischargeClient, crypter, vc.version)
if err != nil {
sendErr(fmt.Errorf("authentication failed %v", err))
return
diff --git a/runtimes/google/ipc/stream/vif/auth.go b/runtimes/google/ipc/stream/vif/auth.go
new file mode 100644
index 0000000..6a05496
--- /dev/null
+++ b/runtimes/google/ipc/stream/vif/auth.go
@@ -0,0 +1,265 @@
+package vif
+
+import (
+ "crypto/rand"
+ "errors"
+ "fmt"
+ "io"
+ "net"
+
+ "code.google.com/p/go.crypto/nacl/box"
+
+ "veyron.io/veyron/veyron/runtimes/google/ipc/stream/crypto"
+ "veyron.io/veyron/veyron/runtimes/google/ipc/stream/message"
+ "veyron.io/veyron/veyron/runtimes/google/ipc/stream/vc"
+ "veyron.io/veyron/veyron/runtimes/google/ipc/version"
+ "veyron.io/veyron/veyron/runtimes/google/lib/iobuf"
+ "veyron.io/veyron/veyron2/ipc/stream"
+ ipcversion "veyron.io/veyron/veyron2/ipc/version"
+ "veyron.io/veyron/veyron2/options"
+ "veyron.io/veyron/veyron2/security"
+)
+
+var (
+ errUnsupportedEncryptVersion = errors.New("unsupported encryption version")
+ errVersionNegotiationFailed = errors.New("encryption version negotiation failed")
+ nullCipher crypto.NullControlCipher
+)
+
+// privateData includes secret data we need for encryption.
+type privateData struct {
+ naclBoxPrivateKey [32]byte
+}
+
+// AuthenticateAsClient sends a HopSetup message if possible. If so, it chooses
+// encryption based on the max supported version.
+//
+// The sequence is initiated by the client.
+//
+// - If the versions include IPCVersion6 or greater, the client sends a
+// HopSetup message to the server, containing the client's supported
+// versions, and the client's crypto options. The HopSetup message
+// is sent in the clear.
+//
+// - When the server receives the HopSetup message, it calls
+// AuthenticateAsServer, which constructs a response HopSetup containing
+// the server's version range, and any crypto options.
+//
+// - For IPCVersion6, the client and server generate fresh public/private key
+// pairs, sending the public key to the peer as a crypto option. The
+// remainder of the communication is encrypted as HopSetupStream messages
+// using NewControlCipherIPC6, which is based on
+// code.google.com/p/go.crypto/nacl/box.
+//
+// - Once the encrypted HopSetupStream channel is setup, the client and
+// server authenticate using the vc.AuthenticateAs{Client,Server} protocol.
+//
+// Note that the HopSetup messages are sent in the clear, so they are subject to
+// modification by a man-in-the-middle, which can currently force a downgrade by
+// modifying the acceptable version ranges downward. This can be addressed by
+// including a hash of the HopSetup message in the encrypted stream. It is
+// likely that this will be addressed in subsequent protocol versions (or it may
+// not be addressed at all if IPCVersion6 becomes the only supported version).
+func AuthenticateAsClient(conn net.Conn, versions *version.Range, principal security.Principal, dc vc.DischargeClient) (crypto.ControlCipher, error) {
+ if versions == nil {
+ versions = version.SupportedRange
+ }
+ if principal == nil {
+ // If there is no principal, we do not support encryption/authentication.
+ var err error
+ versions, err = versions.Intersect(&version.Range{Min: 0, Max: ipcversion.IPCVersion5})
+ if err != nil {
+ return nil, err
+ }
+ }
+ if versions.Max < ipcversion.IPCVersion6 {
+ return nullCipher, nil
+ }
+
+ // The client has not yet sent its public data. Construct it and send it.
+ pvt, pub, err := makeHopSetup(versions)
+ if err != nil {
+ return nil, err
+ }
+ if err := message.WriteTo(conn, &pub, nullCipher); err != nil {
+ return nil, err
+ }
+
+ // Read the server's public data.
+ pool := iobuf.NewPool(0)
+ reader := iobuf.NewReader(pool, conn)
+ pmsg, err := message.ReadFrom(reader, nullCipher)
+ if err != nil {
+ return nil, err
+ }
+ ppub, ok := pmsg.(*message.HopSetup)
+ if !ok {
+ return nil, errVersionNegotiationFailed
+ }
+
+ // Choose the max version in the intersection.
+ vrange, err := pub.Versions.Intersect(&ppub.Versions)
+ if err != nil {
+ return nil, err
+ }
+ v := vrange.Max
+ if v < ipcversion.IPCVersion6 {
+ return nullCipher, nil
+ }
+
+ // Perform the authentication.
+ switch v {
+ case ipcversion.IPCVersion6:
+ return authenticateAsClientIPC6(conn, reader, principal, dc, &pvt, &pub, ppub)
+ default:
+ return nil, errUnsupportedEncryptVersion
+ }
+}
+
+func authenticateAsClientIPC6(writer io.Writer, reader *iobuf.Reader, principal security.Principal, dc vc.DischargeClient,
+ pvt *privateData, pub, ppub *message.HopSetup) (crypto.ControlCipher, error) {
+ pbox := ppub.NaclBox()
+ if pbox == nil {
+ return nil, errVersionNegotiationFailed
+ }
+ c := crypto.NewControlCipherIPC6(&pbox.PublicKey, &pvt.naclBoxPrivateKey, false)
+ sconn := newSetupConn(writer, reader, c)
+ // TODO(jyh): act upon the authentication results.
+ _, _, _, err := vc.AuthenticateAsClient(sconn, principal, dc, crypto.NewNullCrypter(), ipcversion.IPCVersion6)
+ if err != nil {
+ return nil, err
+ }
+ return c, nil
+}
+
+// AuthenticateAsServer handles a HopSetup message, choosing authentication
+// based on the max common version.
+//
+// See AuthenticateAsClient for a description of the negotiation.
+func AuthenticateAsServer(conn net.Conn, versions *version.Range, principal security.Principal, lBlessings security.Blessings,
+ dc vc.DischargeClient, ppub *message.HopSetup) (crypto.ControlCipher, error) {
+ var err error
+ if versions == nil {
+ versions = version.SupportedRange
+ }
+ if principal == nil {
+ // If there is no principal, we don't support encryption/authentication.
+ versions, err = versions.Intersect(&version.Range{Min: 0, Max: ipcversion.IPCVersion5})
+ if err != nil {
+ return nil, err
+ }
+ }
+
+ // Create our public data and send it to the client.
+ pvt, pub, err := makeHopSetup(versions)
+ if err != nil {
+ return nil, err
+ }
+ if err := message.WriteTo(conn, &pub, nullCipher); err != nil {
+ return nil, err
+ }
+
+ // Choose the max version in common.
+ vrange, err := pub.Versions.Intersect(&ppub.Versions)
+ if err != nil {
+ return nil, err
+ }
+ v := vrange.Max
+ if v < ipcversion.IPCVersion6 {
+ return nullCipher, nil
+ }
+
+ // Perform authentication.
+ switch v {
+ case ipcversion.IPCVersion6:
+ return authenticateAsServerIPC6(conn, principal, lBlessings, dc, &pvt, &pub, ppub)
+ default:
+ return nil, errUnsupportedEncryptVersion
+ }
+}
+
+func authenticateAsServerIPC6(conn net.Conn, principal security.Principal, lBlessings security.Blessings, dc vc.DischargeClient,
+ pvt *privateData, pub, ppub *message.HopSetup) (crypto.ControlCipher, error) {
+ box := ppub.NaclBox()
+ if box == nil {
+ return nil, errVersionNegotiationFailed
+ }
+ c := crypto.NewControlCipherIPC6(&box.PublicKey, &pvt.naclBoxPrivateKey, true)
+ pool := iobuf.NewPool(0)
+ reader := iobuf.NewReader(pool, conn)
+ sconn := newSetupConn(conn, reader, c)
+ // TODO(jyh): act upon authentication results.
+ _, _, err := vc.AuthenticateAsServer(sconn, principal, lBlessings, dc, crypto.NewNullCrypter(), ipcversion.IPCVersion6)
+ if err != nil {
+ return nil, err
+ }
+ return c, nil
+}
+
+// serverAuthOptions extracts the Principal from the options list.
+func serverAuthOptions(lopts []stream.ListenerOpt) (principal security.Principal, lBlessings security.Blessings, dischargeClient vc.DischargeClient, err error) {
+ var securityLevel options.VCSecurityLevel
+ for _, o := range lopts {
+ switch v := o.(type) {
+ case vc.DischargeClient:
+ dischargeClient = v
+ case vc.LocalPrincipal:
+ principal = v.Principal
+ case options.VCSecurityLevel:
+ securityLevel = v
+ case options.ServerBlessings:
+ lBlessings = v.Blessings
+ }
+ }
+ switch securityLevel {
+ case options.VCSecurityConfidential:
+ if principal == nil {
+ principal = vc.AnonymousPrincipal
+ }
+ if lBlessings == nil {
+ lBlessings = principal.BlessingStore().Default()
+ }
+ case options.VCSecurityNone:
+ principal = nil
+ default:
+ err = fmt.Errorf("unrecognized VC security level: %v", securityLevel)
+ }
+ return
+}
+
+// clientAuthOptions extracts the client authentication options from the options
+// list.
+func clientAuthOptions(lopts []stream.VCOpt) (principal security.Principal, dischargeClient vc.DischargeClient, err error) {
+ var securityLevel options.VCSecurityLevel
+ for _, o := range lopts {
+ switch v := o.(type) {
+ case vc.DischargeClient:
+ dischargeClient = v
+ case vc.LocalPrincipal:
+ principal = v.Principal
+ case options.VCSecurityLevel:
+ securityLevel = v
+ }
+ }
+ switch securityLevel {
+ case options.VCSecurityConfidential:
+ if principal == nil {
+ principal = vc.AnonymousPrincipal
+ }
+ case options.VCSecurityNone:
+ principal = nil
+ default:
+ err = fmt.Errorf("unrecognized VC security level: %v", securityLevel)
+ }
+ return
+}
+
+// makeHopSetup constructs the options that this process can support.
+func makeHopSetup(versions *version.Range) (pvt privateData, pub message.HopSetup, err error) {
+ pub.Versions = *versions
+ var pubKey, pvtKey *[32]byte
+ pubKey, pvtKey, err = box.GenerateKey(rand.Reader)
+ pub.Options = append(pub.Options, &message.NaclBox{PublicKey: *pubKey})
+ pvt.naclBoxPrivateKey = *pvtKey
+ return
+}
diff --git a/runtimes/google/ipc/stream/vif/set_test.go b/runtimes/google/ipc/stream/vif/set_test.go
index 85c68b3..09b11e5 100644
--- a/runtimes/google/ipc/stream/vif/set_test.go
+++ b/runtimes/google/ipc/stream/vif/set_test.go
@@ -1,26 +1,34 @@
package vif_test
import (
+ "fmt"
"io/ioutil"
"net"
"path"
"testing"
- "veyron.io/veyron/veyron2/naming"
-
"veyron.io/veyron/veyron/runtimes/google/ipc/stream/vif"
+ "veyron.io/veyron/veyron2/naming"
)
func TestSetWithPipes(t *testing.T) {
var (
- conn1, _ = net.Pipe()
- conn2, _ = net.Pipe()
- addr1 = conn1.RemoteAddr()
- addr2 = conn2.RemoteAddr()
+ conn1, conn1s = net.Pipe()
+ conn2, _ = net.Pipe()
+ addr1 = conn1.RemoteAddr()
+ addr2 = conn2.RemoteAddr()
- vf, err = vif.InternalNewDialedVIF(conn1, naming.FixedRoutingID(1), nil)
- set = vif.NewSet()
+ set = vif.NewSet()
)
+ // The server is needed to negotiate with the dialed VIF. Otherwise, the
+ // InternalNewDialedVIF below will block.
+ go func() {
+ _, err := vif.InternalNewAcceptedVIF(conn1s, naming.FixedRoutingID(2), nil)
+ if err != nil {
+ panic(fmt.Sprintf("failed to create server: %s", err))
+ }
+ }()
+ vf, err := vif.InternalNewDialedVIF(conn1, naming.FixedRoutingID(1), nil, nil, nil)
if err != nil {
t.Fatal(err)
}
@@ -50,10 +58,16 @@
// with the same "address" (empty string).
go func() {
for i := 0; i < 2; i++ {
- if _, err := net.Dial("unix", sockname); err != nil {
+ conn, err := net.Dial("unix", sockname)
+ if err != nil {
ln.Close()
- t.Fatal(err)
+ panic(fmt.Sprintf("dial failure: %s", err))
}
+ go func() {
+ if _, err := vif.InternalNewDialedVIF(conn, naming.FixedRoutingID(1), nil, nil, nil); err != nil {
+ panic(fmt.Sprintf("failed to dial VIF: %s", err))
+ }
+ }()
}
}()
@@ -76,6 +90,9 @@
if err != nil {
t.Fatal(err)
}
+ if _, err := vif.InternalNewAcceptedVIF(conn2, naming.FixedRoutingID(1), nil); err != nil {
+ t.Fatal(err)
+ }
set := vif.NewSet()
set.Insert(vif1)
if found := set.Find(addr2.Network(), addr2.String()); found != nil {
diff --git a/runtimes/google/ipc/stream/vif/setup_conn.go b/runtimes/google/ipc/stream/vif/setup_conn.go
new file mode 100644
index 0000000..40a659b
--- /dev/null
+++ b/runtimes/google/ipc/stream/vif/setup_conn.go
@@ -0,0 +1,64 @@
+package vif
+
+import (
+ "io"
+
+ "veyron.io/veyron/veyron/runtimes/google/ipc/stream/crypto"
+ "veyron.io/veyron/veyron/runtimes/google/ipc/stream/message"
+ "veyron.io/veyron/veyron/runtimes/google/lib/iobuf"
+)
+
+// setupConn writes the data to the net.Conn using HopSetupStream messages.
+type setupConn struct {
+ writer io.Writer
+ reader *iobuf.Reader
+ cipher crypto.ControlCipher
+ rbuffer []byte // read buffer
+}
+
+var _ io.ReadWriteCloser = (*setupConn)(nil)
+
+const maxFrameSize = 8192
+
+func newSetupConn(writer io.Writer, reader *iobuf.Reader, c crypto.ControlCipher) *setupConn {
+ return &setupConn{writer: writer, reader: reader, cipher: c}
+}
+
+// Read implements the method from net.Conn.
+func (s *setupConn) Read(buf []byte) (int, error) {
+ for len(s.rbuffer) == 0 {
+ msg, err := message.ReadFrom(s.reader, s.cipher)
+ if err != nil {
+ return 0, err
+ }
+ emsg, ok := msg.(*message.HopSetupStream)
+ if !ok {
+ return 0, errVersionNegotiationFailed
+ }
+ s.rbuffer = emsg.Data
+ }
+ n := copy(buf, s.rbuffer)
+ s.rbuffer = s.rbuffer[n:]
+ return n, nil
+}
+
+// Write implements the method from net.Conn.
+func (s *setupConn) Write(buf []byte) (int, error) {
+ amount := 0
+ for len(buf) > 0 {
+ n := len(buf)
+ if n > maxFrameSize {
+ n = maxFrameSize
+ }
+ emsg := message.HopSetupStream{Data: buf[:n]}
+ if err := message.WriteTo(s.writer, &emsg, s.cipher); err != nil {
+ return 0, err
+ }
+ buf = buf[n:]
+ amount += n
+ }
+ return amount, nil
+}
+
+// Close does nothing.
+func (s *setupConn) Close() error { return nil }
diff --git a/runtimes/google/ipc/stream/vif/setup_conn_test.go b/runtimes/google/ipc/stream/vif/setup_conn_test.go
new file mode 100644
index 0000000..776f5d3
--- /dev/null
+++ b/runtimes/google/ipc/stream/vif/setup_conn_test.go
@@ -0,0 +1,162 @@
+package vif
+
+import (
+ "bytes"
+ "encoding/binary"
+ "io"
+ "net"
+ "sync"
+ "testing"
+
+ "veyron.io/veyron/veyron/runtimes/google/lib/iobuf"
+)
+
+const (
+ text = `Lorem ipsum dolor sit amet, consectetur adipisicing elit, sed do eiusmod tempor incididunt ut labore et dolore magna aliqua. Ut enim ad minim veniam, quis nostrud exercitation ullamco laboris nisi ut aliquip ex ea commodo consequat. Duis aute irure dolor in reprehenderit in voluptate velit esse cillum dolore eu fugiat nulla pariatur. Excepteur sint occaecat cupidatat non proident, sunt in culpa qui officia deserunt mollit anim id est laborum.`
+)
+
+func min(i, j int) int {
+ if i < j {
+ return i
+ }
+ return j
+}
+
+// testControlCipher is a super-simple cipher that xor's each byte of the
+// payload with 0xaa.
+type testControlCipher struct{}
+
+const testMACSize = 4
+
+func (*testControlCipher) MACSize() int {
+ return testMACSize
+}
+
+func testMAC(data []byte) []byte {
+ var h uint32
+ for _, b := range data {
+ h = (h << 1) ^ uint32(b)
+ }
+ var hash [4]byte
+ binary.BigEndian.PutUint32(hash[:], h)
+ return hash[:]
+}
+
+func (c *testControlCipher) Decrypt(data []byte) {
+ for i, _ := range data {
+ data[i] ^= 0xaa
+ }
+}
+
+func (c *testControlCipher) Encrypt(data []byte) {
+ for i, _ := range data {
+ data[i] ^= 0xaa
+ }
+}
+
+func (c *testControlCipher) Open(data []byte) bool {
+ mac := testMAC(data[:len(data)-testMACSize])
+ if bytes.Compare(mac, data[len(data)-testMACSize:]) != 0 {
+ return false
+ }
+ c.Decrypt(data[:len(data)-testMACSize])
+ return true
+}
+
+func (c *testControlCipher) Seal(data []byte) error {
+ c.Encrypt(data[:len(data)-testMACSize])
+ mac := testMAC(data[:len(data)-testMACSize])
+ copy(data[len(data)-testMACSize:], mac)
+ return nil
+}
+
+// shortConn performs at most 3 bytes of IO at a time.
+type shortConn struct {
+ io.ReadWriteCloser
+}
+
+func (s *shortConn) Read(data []byte) (int, error) {
+ if len(data) > 3 {
+ data = data[:3]
+ }
+ return s.ReadWriteCloser.Read(data)
+}
+
+func (s *shortConn) Write(data []byte) (int, error) {
+ n := len(data)
+ for i := 0; i < n; i += 3 {
+ j := min(n, i+3)
+ m, err := s.ReadWriteCloser.Write(data[i:j])
+ if err != nil {
+ return i + m, err
+ }
+ }
+ return n, nil
+}
+
+func TestConn(t *testing.T) {
+ p1, p2 := net.Pipe()
+ pool := iobuf.NewPool(0)
+ r1 := iobuf.NewReader(pool, p1)
+ r2 := iobuf.NewReader(pool, p2)
+ f1 := newSetupConn(p1, r1, &testControlCipher{})
+ f2 := newSetupConn(p2, r2, &testControlCipher{})
+ testConn(t, f1, f2)
+}
+
+func TestShortInnerConn(t *testing.T) {
+ p1, p2 := net.Pipe()
+ s1 := &shortConn{p1}
+ s2 := &shortConn{p2}
+ pool := iobuf.NewPool(0)
+ r1 := iobuf.NewReader(pool, s1)
+ r2 := iobuf.NewReader(pool, s2)
+ f1 := newSetupConn(s1, r1, &testControlCipher{})
+ f2 := newSetupConn(s2, r2, &testControlCipher{})
+ testConn(t, f1, f2)
+}
+
+func TestShortOuterConn(t *testing.T) {
+ p1, p2 := net.Pipe()
+ pool := iobuf.NewPool(0)
+ r1 := iobuf.NewReader(pool, p1)
+ r2 := iobuf.NewReader(pool, p2)
+ e1 := newSetupConn(p1, r1, &testControlCipher{})
+ e2 := newSetupConn(p2, r2, &testControlCipher{})
+ f1 := &shortConn{e1}
+ f2 := &shortConn{e2}
+ testConn(t, f1, f2)
+}
+
+// Write prefixes of the text onto the framed pipe and verify the frame content.
+func testConn(t *testing.T, f1, f2 io.ReadWriteCloser) {
+ // Reader loop.
+ var pending sync.WaitGroup
+ pending.Add(1)
+ go func() {
+ var buf [1024]byte
+ for i := 1; i != len(text); i++ {
+ n, err := io.ReadFull(f1, buf[:i])
+ if err != nil {
+ t.Errorf("bad read: %s", err)
+ }
+ if n != i {
+ t.Errorf("bad read: got %d bytes, expected %d bytes", n, i)
+ }
+ actual := string(buf[:n])
+ expected := string(text[:n])
+ if actual != expected {
+ t.Errorf("got %q, expected %q", actual, expected)
+ }
+ }
+ pending.Done()
+ }()
+
+ // Writer.
+ for i := 1; i != len(text); i++ {
+ if n, err := f2.Write([]byte(text[:i])); err != nil || n != i {
+ t.Errorf("bad write: i=%d n=%d err=%s", i, n, err)
+ }
+ }
+ pending.Wait()
+}
diff --git a/runtimes/google/ipc/stream/vif/vif.go b/runtimes/google/ipc/stream/vif/vif.go
index 4e444f4..b69cdb4 100644
--- a/runtimes/google/ipc/stream/vif/vif.go
+++ b/runtimes/google/ipc/stream/vif/vif.go
@@ -7,12 +7,14 @@
import (
"bytes"
+ "errors"
"fmt"
"net"
"strings"
"sync"
"time"
+ "veyron.io/veyron/veyron/runtimes/google/ipc/stream/crypto"
"veyron.io/veyron/veyron/runtimes/google/ipc/stream/id"
"veyron.io/veyron/veyron/runtimes/google/ipc/stream/message"
"veyron.io/veyron/veyron/runtimes/google/ipc/stream/vc"
@@ -23,7 +25,6 @@
"veyron.io/veyron/veyron/runtimes/google/lib/pcqueue"
vsync "veyron.io/veyron/veyron/runtimes/google/lib/sync"
"veyron.io/veyron/veyron/runtimes/google/lib/upcqueue"
-
"veyron.io/veyron/veyron2/ipc/stream"
"veyron.io/veyron/veyron2/naming"
"veyron.io/veyron/veyron2/verror"
@@ -39,6 +40,13 @@
pool *iobuf.Pool
localEP naming.Endpoint
+ // control channel encryption.
+ isSetup bool
+ // ctrlCipher is normally guarded by writeMu, however see the exception in
+ // readLoop.
+ ctrlCipher crypto.ControlCipher
+ writeMu sync.Mutex
+
vcMap *vcMap
wpending, rpending vsync.WaitGroup
@@ -87,6 +95,10 @@
sharedFlowID = vc.SharedFlowID
)
+var (
+ errAlreadySetup = errors.New("VIF is already setup")
+)
+
// InternalNewDialedVIF creates a new virtual interface over the provided
// network connection, under the assumption that the conn object was created
// using net.Dial.
@@ -94,8 +106,16 @@
// As the name suggests, this method is intended for use only within packages
// placed inside veyron/runtimes/google. Code outside the
// veyron2/runtimes/google/* packages should never call this method.
-func InternalNewDialedVIF(conn net.Conn, rid naming.RoutingID, versions *version.Range) (*VIF, error) {
- return internalNew(conn, rid, id.VC(vc.NumReservedVCs), versions, nil, nil)
+func InternalNewDialedVIF(conn net.Conn, rid naming.RoutingID, versions *version.Range, opts ...stream.VCOpt) (*VIF, error) {
+ principal, dc, err := clientAuthOptions(opts)
+ if err != nil {
+ return nil, err
+ }
+ c, err := AuthenticateAsClient(conn, versions, principal, dc)
+ if err != nil {
+ return nil, err
+ }
+ return internalNew(conn, rid, id.VC(vc.NumReservedVCs), versions, nil, nil, c)
}
// InternalNewAcceptedVIF creates a new virtual interface over the provided
@@ -109,10 +129,11 @@
// placed inside veyron/runtimes/google. Code outside the
// veyron/runtimes/google/* packages should never call this method.
func InternalNewAcceptedVIF(conn net.Conn, rid naming.RoutingID, versions *version.Range, lopts ...stream.ListenerOpt) (*VIF, error) {
- return internalNew(conn, rid, id.VC(vc.NumReservedVCs)+1, versions, upcqueue.New(), lopts)
+ var nc crypto.NullControlCipher
+ return internalNew(conn, rid, id.VC(vc.NumReservedVCs)+1, versions, upcqueue.New(), lopts, &nc)
}
-func internalNew(conn net.Conn, rid naming.RoutingID, initialVCI id.VC, versions *version.Range, acceptor *upcqueue.T, listenerOpts []stream.ListenerOpt) (*VIF, error) {
+func internalNew(conn net.Conn, rid naming.RoutingID, initialVCI id.VC, versions *version.Range, acceptor *upcqueue.T, listenerOpts []stream.ListenerOpt, c crypto.ControlCipher) (*VIF, error) {
// Some cloud providers (like Google Compute Engine) seem to blackhole
// inactive TCP connections, set a TCP keep alive to prevent that.
// See: https://developers.google.com/compute/docs/troubleshooting#communicatewithinternet
@@ -160,6 +181,7 @@
vif := &VIF{
conn: conn,
pool: iobuf.NewPool(0),
+ ctrlCipher: c,
vcMap: newVCMap(),
acceptor: acceptor,
listenerOpts: listenerOpts,
@@ -298,24 +320,32 @@
defer reader.Close()
defer vif.stopVCDispatchLoops()
for {
- msg, err := message.ReadFrom(reader)
+ // vif.ctrlCipher is guarded by vif.writeMu. However, the only mutation
+ // to it is in handleMessage, which runs in the same goroutine, so a
+ // lock is not required here.
+ msg, err := message.ReadFrom(reader, vif.ctrlCipher)
if err != nil {
vlog.VI(1).Infof("Exiting readLoop of VIF %s because of read error: %v", vif, err)
return
}
vlog.VI(3).Infof("Received %T = [%v] on VIF %s", msg, msg, vif)
- vif.handleMessage(msg)
+ if err := vif.handleMessage(msg); err != nil {
+ vlog.VI(1).Infof("Exiting readLoop of VIF %s because of message error: %v", vif, err)
+ return
+ }
}
}
-func (vif *VIF) handleMessage(msg message.T) {
+// handleMessage handles a single incoming message. Any error returned is
+// fatal, causing the VIF to close.
+func (vif *VIF) handleMessage(msg message.T) error {
switch m := msg.(type) {
case *message.Data:
_, rq, _ := vif.vcMap.Find(m.VCI)
if rq == nil {
vlog.VI(2).Infof("Ignoring message of %d bytes for unrecognized VCI %d on VIF %s", m.Payload.Size(), m.VCI, vif)
m.Release()
- return
+ return nil
}
if err := rq.Put(m, nil); err != nil {
vlog.VI(2).Infof("Failed to put message(%v) on VC queue on VIF %v: %v", m, vif, err)
@@ -332,7 +362,7 @@
VCI: m.VCI,
Error: "VCs not accepted",
})
- return
+ return nil
}
vc, err := vif.newVC(m.VCI, m.DstEndpoint, m.SrcEndpoint, false)
vif.distributeCounters(m.Counters)
@@ -341,7 +371,7 @@
VCI: m.VCI,
Error: err.Error(),
})
- return
+ return nil
}
go vif.acceptFlowsLoop(vc, vc.HandshakeAcceptedVC(lopts...))
case *message.CloseVC:
@@ -349,7 +379,7 @@
vif.vcMap.Delete(vc.VCI())
vlog.VI(2).Infof("CloseVC(%+v) on VIF %s", m, vif)
vc.Close(fmt.Sprintf("remote end closed VC(%v)", m.Error))
- return
+ return nil
}
vlog.VI(2).Infof("Ignoring CloseVC(%+v) for unrecognized VCI on VIF %s", m, vif)
case *message.AddReceiveBuffers:
@@ -361,15 +391,34 @@
cm := &message.Data{VCI: m.VCI, Flow: m.Flow}
cm.SetClose()
vif.sendOnExpressQ(cm)
- return
+ return nil
}
vc.ReleaseCounters(m.Flow, m.InitialCounters)
- return
+ return nil
}
vlog.VI(2).Infof("Ignoring OpenFlow(%+v) for unrecognized VCI on VIF %s", m, m, vif)
+ case *message.HopSetup:
+ // Configure the VIF. This takes over the conn during negotiation.
+ if vif.isSetup {
+ return errAlreadySetup
+ }
+ principal, lBlessings, dischargeClient, err := serverAuthOptions(vif.listenerOpts)
+ if err != nil {
+ return errVersionNegotiationFailed
+ }
+ vif.writeMu.Lock()
+ c, err := AuthenticateAsServer(vif.conn, vif.versions, principal, lBlessings, dischargeClient, m)
+ if err != nil {
+ vif.writeMu.Unlock()
+ return err
+ }
+ vif.ctrlCipher = c
+ vif.writeMu.Unlock()
+ vif.isSetup = true
default:
vlog.Infof("Ignoring unrecognized message %T on VIF %s", m, vif)
}
+ return nil
}
func (vif *VIF) vcDispatchLoop(vc *vc.VC, messages *pcqueue.T) {
@@ -457,8 +506,8 @@
switch writer {
case vif.expressQ:
for _, b := range bufs {
- if n, err := vif.conn.Write(b.Contents); err != nil || n != b.Size() {
- vlog.Errorf("Exiting writeLoop of VIF %s because Control message write failed. Got (%d, %v), want (%d, nil)", vif, n, err, b.Size())
+ if err := vif.writeSerializedMessage(b.Contents); err != nil {
+ vlog.Errorf("Exiting writeLoop of VIF %s because Control message write failed: %s", vif, err)
releaseBufs(bufs)
return
}
@@ -476,7 +525,7 @@
vif.flowMu.Unlock()
if len(msg.Counters) > 0 {
vlog.VI(3).Infof("Sending counters %v on VIF %s", msg.Counters, vif)
- if err := message.WriteTo(vif.conn, msg); err != nil {
+ if err := vif.writeMessage(msg); err != nil {
vlog.VI(1).Infof("Exiting writeLoop of VIF %s because AddReceiveBuffers message write failed: %v", vif, err)
return
}
@@ -511,7 +560,7 @@
vif.shutdownFlow(vc, m.Flow)
}
if err == nil {
- err = message.WriteTo(vif.conn, m)
+ err = vif.writeMessage(m)
}
if err != nil {
// TODO(caprita): Calling closeVCAndSendMsg below causes
@@ -551,12 +600,41 @@
func (vif *VIF) sendOnExpressQ(msg message.T) error {
vlog.VI(1).Infof("sendOnExpressQ(%T = %+v) on VIF %s", msg, msg, vif)
var buf bytes.Buffer
- if err := message.WriteTo(&buf, msg); err != nil {
+ // Don't encrypt yet, because the message ordering isn't yet determined.
+ // Encryption is performed by vif.writeSerializedMessage() when the
+ // message is actually written to vif.conn.
+ vif.writeMu.Lock()
+ c := vif.ctrlCipher
+ vif.writeMu.Unlock()
+ if err := message.WriteTo(&buf, msg, crypto.NewDisabledControlCipher(c)); err != nil {
return err
}
return vif.expressQ.Put(iobuf.NewSlice(buf.Bytes()), nil)
}
+// writeMessage writes the message to the channel. Writes must be serialized so
+// that the control channel can be encrypted, so we acquire the writeMu.
+func (vif *VIF) writeMessage(msg message.T) error {
+ vif.writeMu.Lock()
+ defer vif.writeMu.Unlock()
+ return message.WriteTo(vif.conn, msg, vif.ctrlCipher)
+}
+
+// Write writes the message to the channel, encrypting the control data. Writes
+// must be serialized so that the control channel can be encrypted, so we
+// acquire the writeMu.
+func (vif *VIF) writeSerializedMessage(msg []byte) error {
+ vif.writeMu.Lock()
+ defer vif.writeMu.Unlock()
+ if err := message.EncryptMessage(msg, vif.ctrlCipher); err != nil {
+ return err
+ }
+ if n, err := vif.conn.Write(msg); err != nil {
+ return fmt.Errorf("write failed: got (%d, %v) for %d byte message", n, err, len(msg))
+ }
+ return nil
+}
+
func (vif *VIF) writeDataMessages(writer bqueue.Writer, bufs []*iobuf.Slice) {
vci, fid := unpackIDs(writer.ID())
// iobuf.Coalesce will coalesce buffers only if they are adjacent to
@@ -616,7 +694,7 @@
LocalEP: localEP,
RemoteEP: remoteEP,
Pool: vif.pool,
- ReserveBytes: message.HeaderSizeBytes,
+ ReserveBytes: uint(message.HeaderSizeBytes + vif.ctrlCipher.MACSize()),
Helper: vcHelper{vif},
Version: version,
})
diff --git a/runtimes/google/ipc/stream/vif/vif_test.go b/runtimes/google/ipc/stream/vif/vif_test.go
index 71c16d0..a101f55 100644
--- a/runtimes/google/ipc/stream/vif/vif_test.go
+++ b/runtimes/google/ipc/stream/vif/vif_test.go
@@ -12,7 +12,9 @@
"reflect"
"runtime"
"sort"
+ "sync"
"testing"
+ "time"
"veyron.io/veyron/veyron/lib/testutil"
tsecurity "veyron.io/veyron/veyron/lib/testutil/security"
@@ -318,10 +320,17 @@
type versionTestCase struct {
client, server, ep *iversion.Range
expectError bool
+ expectVIFError bool
}
func (tc *versionTestCase) Run(t *testing.T) {
- client, server := NewVersionedClientServer(tc.client, tc.server)
+ client, server, err := NewVersionedClientServer(tc.client, tc.server)
+ if (err != nil) != tc.expectVIFError {
+ t.Errorf("Error mismatch. Wanted error: %v, got %v; client: %v, server: %v", tc.expectVIFError, err, tc.client, tc.server)
+ }
+ if err != nil {
+ return
+ }
defer client.Close()
ep := tc.ep.Endpoint("test", "addr", naming.FixedRoutingID(0x5))
@@ -344,19 +353,23 @@
}
// TestIncompatibleVersions tests many cases where the client and server
-// have compatbile or incompatible supported version ranges. It ensures
+// have compatible or incompatible supported version ranges. It ensures
// that overlapping ranges work properly, but non-overlapping ranges generate
// errors.
func TestIncompatibleVersions(t *testing.T) {
unknown := &iversion.Range{version.UnknownIPCVersion, version.UnknownIPCVersion}
tests := []versionTestCase{
- {&iversion.Range{1, 1}, &iversion.Range{1, 1}, &iversion.Range{1, 1}, false},
- {&iversion.Range{1, 3}, &iversion.Range{3, 5}, &iversion.Range{3, 5}, false},
- {&iversion.Range{1, 3}, &iversion.Range{3, 5}, unknown, false},
+ {&iversion.Range{1, 1}, &iversion.Range{1, 1}, &iversion.Range{1, 1}, false, false},
+ {&iversion.Range{1, 3}, &iversion.Range{3, 5}, &iversion.Range{3, 5}, false, false},
+ {&iversion.Range{1, 3}, &iversion.Range{3, 5}, unknown, false, false},
- {&iversion.Range{1, 3}, &iversion.Range{4, 5}, &iversion.Range{4, 5}, true},
- {&iversion.Range{1, 3}, &iversion.Range{4, 5}, unknown, true},
- {&iversion.Range{3, 5}, &iversion.Range{1, 3}, unknown, true},
+ // No VIF error because the client does not initiate authentication.
+ {&iversion.Range{1, 3}, &iversion.Range{4, 5}, &iversion.Range{4, 5}, true, false},
+ {&iversion.Range{1, 3}, &iversion.Range{4, 5}, unknown, true, false},
+
+ // VIF error because the client asks for authentication, but the server
+ // doesn't understand it.
+ {&iversion.Range{6, 6}, &iversion.Range{1, 5}, unknown, true, true},
}
for _, tc := range tests {
@@ -366,14 +379,19 @@
func TestNetworkFailure(t *testing.T) {
c1, c2 := pipe()
- client, err := vif.InternalNewDialedVIF(c1, naming.FixedRoutingID(0xc), nil)
- if err != nil {
- t.Fatal(err)
- }
+ result := make(chan *vif.VIF)
+ go func() {
+ client, err := vif.InternalNewDialedVIF(c1, naming.FixedRoutingID(0xc), nil, nil, nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ result <- client
+ }()
server, err := vif.InternalNewAcceptedVIF(c2, naming.FixedRoutingID(0x5), nil)
if err != nil {
t.Fatal(err)
}
+ client := <-result
// If the network connection dies, Dial and Accept should fail.
c1.Close()
if _, err := client.Dial(makeEP(0x5)); err == nil {
@@ -394,35 +412,104 @@
func (a pipeAddr) Network() string { return "pipe" }
func (a pipeAddr) String() string { return a.name }
-// pipeConn provides a LocalAddr and RemoteAddr based on pipeAddr.
+// pipeConn provides a buffered net.Conn, with pipeAddr addressing.
type pipeConn struct {
- net.Conn
+ lock sync.Mutex
+ // w is guarded by lock, to prevent Close from racing with Write. This is a
+ // quick way to prevent the race, but it allows a Write to block the Close.
+ // This isn't a problem in the tests currently.
+ w chan<- []byte
+ r <-chan []byte
+ rdata []byte
laddr, raddr pipeAddr
}
-func (c *pipeConn) LocalAddr() net.Addr { return c.laddr }
-func (c *pipeConn) RemoteAddr() net.Addr { return c.raddr }
+func (c *pipeConn) Read(data []byte) (int, error) {
+ for len(c.rdata) == 0 {
+ d, ok := <-c.r
+ if !ok {
+ return 0, io.EOF
+ }
+ c.rdata = d
+ }
+ n := copy(data, c.rdata)
+ c.rdata = c.rdata[n:]
+ return n, nil
+}
+
+func (c *pipeConn) Write(data []byte) (int, error) {
+ c.lock.Lock()
+ defer c.lock.Unlock()
+ if c.w == nil {
+ return 0, io.EOF
+ }
+ d := make([]byte, len(data))
+ copy(d, data)
+ c.w <- d
+ return len(data), nil
+}
+
+func (c *pipeConn) Close() error {
+ c.lock.Lock()
+ defer c.lock.Unlock()
+ if c.w == nil {
+ return io.EOF
+ }
+ close(c.w)
+ c.w = nil
+ return nil
+}
+
+func (c *pipeConn) LocalAddr() net.Addr { return c.laddr }
+func (c *pipeConn) RemoteAddr() net.Addr { return c.raddr }
+func (c *pipeConn) SetDeadline(t time.Time) error { return nil }
+func (c *pipeConn) SetReadDeadline(t time.Time) error { return nil }
+func (c *pipeConn) SetWriteDeadline(t time.Time) error { return nil }
func pipe() (net.Conn, net.Conn) {
clientAddr := pipeAddr{"client"}
serverAddr := pipeAddr{"server"}
- client, server := net.Pipe()
- return &pipeConn{client, clientAddr, serverAddr}, &pipeConn{server, serverAddr, clientAddr}
+ c1 := make(chan []byte, 10)
+ c2 := make(chan []byte, 10)
+ p1 := &pipeConn{w: c1, r: c2, laddr: clientAddr, raddr: serverAddr}
+ p2 := &pipeConn{w: c2, r: c1, laddr: serverAddr, raddr: clientAddr}
+ return p1, p2
}
func NewClientServer() (client, server *vif.VIF) {
- return NewVersionedClientServer(nil, nil)
+ var err error
+ client, server, err = NewVersionedClientServer(nil, nil)
+ if err != nil {
+ panic(err)
+ }
+ return
}
-func NewVersionedClientServer(clientVersions, serverVersions *iversion.Range) (client, server *vif.VIF) {
+func NewVersionedClientServer(clientVersions, serverVersions *iversion.Range) (client, server *vif.VIF, verr error) {
c1, c2 := pipe()
- var err error
- if client, err = vif.InternalNewDialedVIF(c1, naming.FixedRoutingID(0xc), clientVersions); err != nil {
- panic(err)
+ var cerr error
+ cl := make(chan *vif.VIF)
+ go func() {
+ c, err := vif.InternalNewDialedVIF(c1, naming.FixedRoutingID(0xc), clientVersions, newPrincipal("client"), nil)
+ if err != nil {
+ cerr = err
+ close(cl)
+ } else {
+ cl <- c
+ }
+ }()
+ s, err := vif.InternalNewAcceptedVIF(c2, naming.FixedRoutingID(0x5), serverVersions, newPrincipal("server"))
+ c, ok := <-cl
+ if err != nil {
+ verr = err
+ return
}
- if server, err = vif.InternalNewAcceptedVIF(c2, naming.FixedRoutingID(0x5), serverVersions, newPrincipal("server")); err != nil {
- panic(err)
+ if !ok {
+ verr = cerr
+ return
}
+ server = s
+ client = c
return
}
diff --git a/runtimes/google/ipc/version/version.go b/runtimes/google/ipc/version/version.go
index f037514..9d00f2c 100644
--- a/runtimes/google/ipc/version/version.go
+++ b/runtimes/google/ipc/version/version.go
@@ -15,19 +15,19 @@
}
var (
- // supportedRange represents the range of protocol verions supported by this
+ // SupportedRange represents the range of protocol verions supported by this
// implementation.
// Max should be incremented whenever we make a protocol
// change that's not both forward and backward compatible.
// Min should be incremented whenever we want to remove
// support for old protocol versions.
- supportedRange = &Range{Min: version.IPCVersion4, Max: version.IPCVersion5}
+ SupportedRange = &Range{Min: version.IPCVersion4, Max: version.IPCVersion6}
// Export the methods on supportedRange.
- Endpoint = supportedRange.Endpoint
- ProxiedEndpoint = supportedRange.ProxiedEndpoint
- CommonVersion = supportedRange.CommonVersion
- CheckCompatibility = supportedRange.CheckCompatibility
+ Endpoint = SupportedRange.Endpoint
+ ProxiedEndpoint = SupportedRange.ProxiedEndpoint
+ CommonVersion = SupportedRange.CommonVersion
+ CheckCompatibility = SupportedRange.CheckCompatibility
)
var (
@@ -84,6 +84,15 @@
return intersectRanges(a.MinIPCVersion, a.MaxIPCVersion, b.MinIPCVersion, b.MaxIPCVersion)
}
+func (r1 *Range) Intersect(r2 *Range) (*Range, error) {
+ min, max, err := intersectRanges(r1.Min, r1.Max, r2.Min, r2.Max)
+ if err != nil {
+ return nil, err
+ }
+ r := &Range{Min: min, Max: max}
+ return r, nil
+}
+
// ProxiedEndpoint returns an endpoint with the Min/MaxIPCVersion properly filled in
// to match the intersection of capabilities of this process and the proxy.
func (r *Range) ProxiedEndpoint(rid naming.RoutingID, proxy naming.Endpoint) (naming.Endpoint, error) {
