ref: Move to NaclBox for channel encryption and move versions out of endpoints.
This change does several things:
1) Create a new endpoint version (version 5) this will eventually
replace all previous endpoint version and the only valid format.
The difference from v4 is the removal of versioning information.
2) Switch to NaclBox for channel encryption.
3) Change the low level protocol to use SetupVC instead of OpenVC for
new messages. This does version and encryption negotiation all
at once as the first thing on a new VC..
There are several reasons for this change:
1) We save round-trips vs TLS.
2) Versioning is reliable since it's negotiated. This is compared
to now where the root mounttable must be updated before new
clients are deployed. This has proven an unrealistic requirement.
Old benchmarks:
Benchmark_dial_VC_TLS 100 36758152 ns/op
--- Histogram (unit: ms)
Count: 100 Min: 22 Max: 49 Avg: 36.21
------------------------------------------------------------
[ 22, 23) 1 1.0% 1.0%
[ 23, 24) 0 0.0% 1.0%
[ 24, 25) 0 0.0% 1.0%
[ 25, 26) 0 0.0% 1.0%
[ 26, 28) 0 0.0% 1.0%
[ 28, 31) 5 5.0% 6.0% #
[ 31, 34) 38 38.0% 44.0% ####
[ 34, 38) 14 14.0% 58.0% #
[ 38, 43) 28 28.0% 86.0% ###
[ 43, 50) 14 14.0% 100.0% #
[ 50, 59) 0 0.0% 100.0%
[ 59, 70) 0 0.0% 100.0%
[ 70, 83) 0 0.0% 100.0%
[ 83, 100) 0 0.0% 100.0%
[100, 121) 0 0.0% 100.0%
[121, 148) 0 0.0% 100.0%
[148, inf) 0 0.0% 100.0%
Benchmark_throughput_Flow_1VIF_1VC_1Flow 50000 36354 ns/op 1408.36 MB/s
Benchmark_throughput_Flow_1VIF_1VC_2Flow 50000 36849 ns/op 1389.43 MB/s
Benchmark_throughput_Flow_1VIF_1VC_8Flow 30000 44898 ns/op 1140.35 MB/s
Benchmark_throughput_Flow_1VIF_2VC_2Flow 50000 37180 ns/op 1377.06 MB/s
Benchmark_throughput_Flow_1VIF_2VC_8Flow 30000 44599 ns/op 1147.99 MB/s
Benchmark_throughput_Flow_2VIF_4VC_8Flow 30000 44945 ns/op 1139.15 MB/s
Benchmark_throughput_Flow_1VIF_1VC_1FlowTLS 3000 543051 ns/op 94.28 MB/s
Benchmark_throughput_Flow_1VIF_1VC_2FlowTLS 3000 534945 ns/op 95.71 MB/s
Benchmark_throughput_Flow_1VIF_1VC_8FlowTLS 3000 552158 ns/op 92.73 MB/s
Benchmark_throughput_Flow_1VIF_2VC_2FlowTLS 3000 760363 ns/op 67.34 MB/s
Benchmark_throughput_Flow_1VIF_2VC_8FlowTLS 3000 462552 ns/op 110.69 MB/s
Benchmark_throughput_Flow_2VIF_4VC_8FlowTLS 3000 463113 ns/op 110.56 MB/s
ok v.io/x/ref/profiles/internal/rpc/stream/benchmark 37.809s
New benchmarks:
--- Histogram (unit: ms)
Count: 100 Min: 14 Max: 39 Avg: 30.63
------------------------------------------------------------
[ 14, 15) 1 1.0% 1.0%
[ 15, 16) 0 0.0% 1.0%
[ 16, 17) 0 0.0% 1.0%
[ 17, 18) 0 0.0% 1.0%
[ 18, 20) 0 0.0% 1.0%
[ 20, 22) 0 0.0% 1.0%
[ 22, 25) 0 0.0% 1.0%
[ 25, 29) 0 0.0% 1.0%
[ 29, 34) 97 97.0% 98.0% ##########
[ 34, 40) 2 2.0% 100.0%
[ 40, 48) 0 0.0% 100.0%
[ 48, 58) 0 0.0% 100.0%
[ 58, 71) 0 0.0% 100.0%
[ 71, 87) 0 0.0% 100.0%
[ 87, 107) 0 0.0% 100.0%
[107, 131) 0 0.0% 100.0%
[131, inf) 0 0.0% 100.0%
Benchmark_throughput_Flow_1VIF_1VC_1Flow 50000 32806 ns/op 1560.65 MB/s
Benchmark_throughput_Flow_1VIF_1VC_2Flow 50000 31752 ns/op 1612.47 MB/s
Benchmark_throughput_Flow_1VIF_1VC_8Flow 50000 39765 ns/op 1287.54 MB/s
Benchmark_throughput_Flow_1VIF_2VC_2Flow 50000 31967 ns/op 1601.62 MB/s
Benchmark_throughput_Flow_1VIF_2VC_8Flow 50000 39513 ns/op 1295.74 MB/s
Benchmark_throughput_Flow_2VIF_4VC_8Flow 30000 40676 ns/op 1258.72 MB/s
Benchmark_throughput_Flow_1VIF_1VC_1FlowTLS 10000 237259 ns/op 215.80 MB/s
Benchmark_throughput_Flow_1VIF_1VC_2FlowTLS 10000 233769 ns/op 219.02 MB/s
Benchmark_throughput_Flow_1VIF_1VC_8FlowTLS 10000 244584 ns/op 209.33 MB/s
Benchmark_throughput_Flow_1VIF_2VC_2FlowTLS 10000 235281 ns/op 217.61 MB/s
Benchmark_throughput_Flow_1VIF_2VC_8FlowTLS 10000 238344 ns/op 214.81 MB/s
Benchmark_throughput_Flow_2VIF_4VC_8FlowTLS 10000 239573 ns/op 213.71 MB/s
ok v.io/x/ref/profiles/internal/rpc/stream/benchmark 39.893s
MultiPart: 2/2
Change-Id: Ia2397c445116d12d0b037ad65f686ddd9846f33b
diff --git a/profiles/internal/naming/endpoint.go b/profiles/internal/naming/endpoint.go
index 88b333c..e757878 100644
--- a/profiles/internal/naming/endpoint.go
+++ b/profiles/internal/naming/endpoint.go
@@ -77,6 +77,8 @@
err = ep.parseV3(parts)
case 4:
err = ep.parseV4(parts)
+ case 5:
+ err = ep.parseV5(parts)
default:
err = errInvalidEndpointString
}
@@ -125,7 +127,7 @@
}
v, err := strconv.ParseUint(input, 10, 32)
if err != nil {
- err = fmt.Errorf("invalid RPC version: %s, %v", err)
+ err = fmt.Errorf("invalid RPC version: %s, %v", input, err)
}
return version.RPCVersion(v), err
}
@@ -201,6 +203,26 @@
return nil
}
+func (ep *Endpoint) parseV5(parts []string) error {
+ if len(parts) < 5 {
+ return errInvalidEndpointString
+ }
+ var err error
+ if err = ep.parseV1(parts[:4]); err != nil {
+ return err
+ }
+ if ep.IsMountTable, ep.IsLeaf, err = parseMountTableFlag(parts[4]); err != nil {
+ return fmt.Errorf("invalid mount table flag: %v", err)
+ }
+ // Join the remaining and re-split.
+ if str := strings.Join(parts[5:], separator); len(str) > 0 {
+ ep.Blessings = strings.Split(str, blessingsSeparator)
+ }
+ ep.MinRPCVersion = version.DeprecatedRPCVersion
+ ep.MaxRPCVersion = version.DeprecatedRPCVersion
+ return nil
+}
+
func (ep *Endpoint) RoutingID() naming.RoutingID {
//nologcall
return ep.RID
@@ -244,6 +266,17 @@
ep.Protocol, ep.Address, ep.RID,
printRPCVersion(ep.MinRPCVersion), printRPCVersion(ep.MaxRPCVersion),
mt, blessings)
+ case 5:
+ mt := "s"
+ switch {
+ case ep.IsLeaf:
+ mt = "l"
+ case ep.IsMountTable:
+ mt = "m"
+ }
+ blessings := strings.Join(ep.Blessings, blessingsSeparator)
+ return fmt.Sprintf("@5@%s@%s@%s@%s@%s@@",
+ ep.Protocol, ep.Address, ep.RID, mt, blessings)
}
}
diff --git a/profiles/internal/naming/endpoint_test.go b/profiles/internal/naming/endpoint_test.go
index a011512..2b9d204 100644
--- a/profiles/internal/naming/endpoint_test.go
+++ b/profiles/internal/naming/endpoint_test.go
@@ -91,6 +91,34 @@
// Blessings that look similar to other parts of the endpoint.
Blessings: []string{"@@", "@s", "@m"},
}
+ v5 := &Endpoint{
+ Protocol: "tcp",
+ Address: "batman.com:2345",
+ RID: naming.FixedRoutingID(0xba77),
+ IsMountTable: true,
+ Blessings: []string{"dev.v.io/foo@bar.com", "dev.v.io/bar@bar.com/delegate"},
+ MinRPCVersion: version.DeprecatedRPCVersion,
+ MaxRPCVersion: version.DeprecatedRPCVersion,
+ }
+ v5b := &Endpoint{
+ Protocol: "tcp",
+ Address: "batman.com:2345",
+ RID: naming.FixedRoutingID(0xba77),
+ IsMountTable: true,
+ // Blessings that look similar to other parts of the endpoint.
+ Blessings: []string{"@@", "@s", "@m"},
+ MinRPCVersion: version.DeprecatedRPCVersion,
+ MaxRPCVersion: version.DeprecatedRPCVersion,
+ }
+ v5c := &Endpoint{
+ Protocol: "tcp",
+ Address: "batman.com:2345",
+ RID: naming.FixedRoutingID(0xba77),
+ IsMountTable: false,
+ Blessings: []string{"dev.v.io/foo@bar.com", "dev.v.io/bar@bar.com/delegate"},
+ MinRPCVersion: version.DeprecatedRPCVersion,
+ MaxRPCVersion: version.DeprecatedRPCVersion,
+ }
testcasesA := []struct {
endpoint naming.Endpoint
@@ -107,7 +135,7 @@
}
}
- // Test v3 & v4 endpoints.
+ // Test v3, v4, v5 endpoints.
testcasesC := []struct {
Endpoint naming.Endpoint
String string
@@ -120,6 +148,9 @@
{v3s, "@4@@batman.com:2345@00000000000000000000000000000000@2@3@s@@@", 4},
{v4, "@4@tcp@batman.com:2345@0000000000000000000000000000ba77@4@5@m@dev.v.io/foo@bar.com,dev.v.io/bar@bar.com/delegate@@", 4},
{v4b, "@4@tcp@batman.com:2345@0000000000000000000000000000ba77@4@5@m@@@,@s,@m@@", 4},
+ {v5, "@5@tcp@batman.com:2345@0000000000000000000000000000ba77@m@dev.v.io/foo@bar.com,dev.v.io/bar@bar.com/delegate@@", 5},
+ {v5b, "@5@tcp@batman.com:2345@0000000000000000000000000000ba77@m@@@,@s,@m@@", 5},
+ {v5c, "@5@tcp@batman.com:2345@0000000000000000000000000000ba77@s@dev.v.io/foo@bar.com,dev.v.io/bar@bar.com/delegate@@", 5},
}
for _, test := range testcasesC {
diff --git a/profiles/internal/naming/namespace/all_test.go b/profiles/internal/naming/namespace/all_test.go
index 721ea34..386d169 100644
--- a/profiles/internal/naming/namespace/all_test.go
+++ b/profiles/internal/naming/namespace/all_test.go
@@ -674,10 +674,10 @@
// The namespace root from the context should be authorized as well.
ctx, ns, _ := v23.SetNewNamespace(clientCtx, naming.JoinAddressName(root, ""))
if e, err := ns.Resolve(ctx, "mt/server"); verror.ErrorID(err) != verror.ErrNotTrusted.ID {
- t.Errorf("resolve with root=%q returned (%v, errorid=%v %v), wanted errorid=%v", root, e, verror.ErrorID(err), err, verror.ErrNotTrusted.ID)
+ t.Errorf("resolve with root=%q returned (%v, errorid=%v %v), wanted errorid=%v: %s", root, e, verror.ErrorID(err), err, verror.ErrNotTrusted.ID, verror.DebugString(err))
}
if _, err := ns.Resolve(ctx, "mt/server", options.SkipServerEndpointAuthorization{}); err != nil {
- t.Errorf("resolve with root=%q should have succeeded when authorization checks are skipped. Got %v", err)
+ t.Errorf("resolve with root=%q should have succeeded when authorization checks are skipped. Got %v: %s", root, err, verror.DebugString(err))
}
}
}
@@ -692,7 +692,7 @@
}
if e, err := resolve("mt/server", options.SkipServerEndpointAuthorization{}); err != nil {
- t.Errorf("Resolve should succeed when skipping server authorization. Got (%v, %v)", e, err)
+ t.Errorf("Resolve should succeed when skipping server authorization. Got (%v, %v) %s", e, err, verror.DebugString(err))
} else if e, err := resolve("mt/server"); verror.ErrorID(err) != verror.ErrNotTrusted.ID {
t.Errorf("Resolve should have failed with %q because an attacker has taken over the intermediate mounttable. Got (%+v, errorid=%q:%v)", verror.ErrNotTrusted.ID, e, verror.ErrorID(err), err)
}
diff --git a/profiles/internal/rpc/stream/crypto/box.go b/profiles/internal/rpc/stream/crypto/box.go
index 0dfba87..e16c5ec 100644
--- a/profiles/internal/rpc/stream/crypto/box.go
+++ b/profiles/internal/rpc/stream/crypto/box.go
@@ -9,13 +9,10 @@
"crypto/rand"
"encoding/binary"
"fmt"
- "io"
- "net"
"golang.org/x/crypto/nacl/box"
"v.io/v23/verror"
-
"v.io/x/ref/profiles/internal/lib/iobuf"
"v.io/x/ref/profiles/internal/rpc/stream"
)
@@ -32,42 +29,47 @@
// strings to avoid repeating these n-times in the final error
// message visible to the user.
errCipherTextTooShort = reg(".errCipherTextTooShort", "ciphertext too short")
+ errRemotePublicKey = reg(".errRemotePublicKey", "failed to get remote public key")
errMessageAuthFailed = reg(".errMessageAuthFailed", "message authentication failed")
errUnrecognizedCipherText = reg(".errUnrecognizedCipherText", "CipherSuite {3} is not recognized. Must use one that uses Diffie-Hellman as the key exchange algorithm")
)
type boxcrypter struct {
- conn net.Conn
alloc *iobuf.Allocator
sharedKey [32]byte
sortedPubkeys []byte
writeNonce, readNonce uint64
}
+type BoxKey [32]byte
+
+// BoxKeyExchanger is used to communicate public keys between the two ends of
+// communication.
+type BoxKeyExchanger func(myPublicKey *BoxKey) (theirPublicKey *BoxKey, err error)
+
// NewBoxCrypter uses Curve25519, XSalsa20 and Poly1305 to encrypt and
// authenticate messages (as defined in http://nacl.cr.yp.to/box.html).
// An ephemeral Diffie-Hellman key exchange is performed per invocation
// of NewBoxCrypter; the data sent has forward security with connection
// granularity. One round-trip is required before any data can be sent.
// BoxCrypter does NOT do anything to verify the identity of the peer.
-func NewBoxCrypter(conn net.Conn, pool *iobuf.Pool) (Crypter, error) {
+func NewBoxCrypter(exchange BoxKeyExchanger, pool *iobuf.Pool) (Crypter, error) {
pk, sk, err := box.GenerateKey(rand.Reader)
if err != nil {
return nil, err
}
- var theirPK [32]byte
- errChan := make(chan error)
- defer close(errChan)
- go func() { _, err := conn.Write(pk[:]); errChan <- err }()
- go func() { _, err := io.ReadFull(conn, theirPK[:]); errChan <- err }()
- if err := <-errChan; err != nil {
+
+ theirPK, err := exchange((*BoxKey)(pk))
+ if err != nil {
return nil, err
}
- if err := <-errChan; err != nil {
- return nil, err
+ if theirPK == nil {
+ return nil, verror.New(errRemotePublicKey, nil)
}
- ret := &boxcrypter{conn: conn, alloc: iobuf.NewAllocator(pool, 0)}
- box.Precompute(&ret.sharedKey, &theirPK, sk)
+
+ ret := &boxcrypter{alloc: iobuf.NewAllocator(pool, 0)}
+
+ box.Precompute(&ret.sharedKey, (*[32]byte)(theirPK), sk)
// Distinct messages between the same {sender, receiver} set are required
// to have distinct nonces. The server with the lexicographically smaller
// public key will be sending messages with 0, 2, 4... and the other will
diff --git a/profiles/internal/rpc/stream/crypto/box_cipher.go b/profiles/internal/rpc/stream/crypto/box_cipher.go
index 5dc0cb8..a5a5f6b 100644
--- a/profiles/internal/rpc/stream/crypto/box_cipher.go
+++ b/profiles/internal/rpc/stream/crypto/box_cipher.go
@@ -73,10 +73,10 @@
copy(counter[:], nonce[16:])
}
-// NewControlCipher returns a ControlCipher for RPC V6.
-func NewControlCipherRPC6(peersPublicKey, privateKey *[32]byte, isServer bool) ControlCipher {
+// NewControlCipher returns a ControlCipher for RPC versions greater than 6.
+func NewControlCipherRPC6(peersPublicKey, privateKey *BoxKey, isServer bool) ControlCipher {
var c cbox
- box.Precompute(&c.sharedKey, peersPublicKey, privateKey)
+ box.Precompute(&c.sharedKey, (*[32]byte)(peersPublicKey), (*[32]byte)(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
diff --git a/profiles/internal/rpc/stream/crypto/box_cipher_test.go b/profiles/internal/rpc/stream/crypto/box_cipher_test.go
index e26e6ce..dc29318 100644
--- a/profiles/internal/rpc/stream/crypto/box_cipher_test.go
+++ b/profiles/internal/rpc/stream/crypto/box_cipher_test.go
@@ -30,8 +30,8 @@
if err != nil {
t.Fatalf("can't generate key")
}
- c1 := crypto.NewControlCipherRPC6(pub2, pvt1, true)
- c2 := crypto.NewControlCipherRPC6(pub1, pvt2, false)
+ c1 := crypto.NewControlCipherRPC6((*crypto.BoxKey)(pub2), (*crypto.BoxKey)(pvt1), true)
+ c2 := crypto.NewControlCipherRPC6((*crypto.BoxKey)(pub1), (*crypto.BoxKey)(pvt2), false)
msg1 := newMessage("hello")
if err := c1.Seal(msg1); err != nil {
@@ -102,8 +102,8 @@
if err != nil {
t.Fatalf("can't generate key")
}
- c1 := crypto.NewControlCipherRPC6(pub2, pvt1, true)
- c2 := crypto.NewControlCipherRPC6(pub1, pvt2, false)
+ c1 := crypto.NewControlCipherRPC6((*crypto.BoxKey)(pub2), (*crypto.BoxKey)(pvt1), true)
+ c2 := crypto.NewControlCipherRPC6((*crypto.BoxKey)(pub1), (*crypto.BoxKey)(pvt2), false)
msg1 := []byte("hello")
msg2 := []byte("world")
diff --git a/profiles/internal/rpc/stream/crypto/crypto_test.go b/profiles/internal/rpc/stream/crypto/crypto_test.go
index 5fa3c82..f5ef433 100644
--- a/profiles/internal/rpc/stream/crypto/crypto_test.go
+++ b/profiles/internal/rpc/stream/crypto/crypto_test.go
@@ -73,8 +73,7 @@
}
func TestBox(t *testing.T) {
- server, client := net.Pipe()
- c1, c2 := boxCrypters(t, server, client)
+ c1, c2 := boxCrypters(t, nil, nil)
testSimple(t, c1, c2)
}
@@ -168,16 +167,25 @@
return c1, c2
}
-func boxCrypters(t testing.TB, serverConn, clientConn net.Conn) (Crypter, Crypter) {
+func boxCrypters(t testing.TB, _, _ net.Conn) (Crypter, Crypter) {
+ server, client := make(chan *BoxKey, 1), make(chan *BoxKey, 1)
+ clientExchanger := func(pubKey *BoxKey) (*BoxKey, error) {
+ client <- pubKey
+ return <-server, nil
+ }
+ serverExchanger := func(pubKey *BoxKey) (*BoxKey, error) {
+ server <- pubKey
+ return <-client, nil
+ }
crypters := make(chan Crypter)
- for _, conn := range []net.Conn{serverConn, clientConn} {
- go func(conn net.Conn) {
- crypter, err := NewBoxCrypter(conn, iobuf.NewPool(0))
+ for _, ex := range []BoxKeyExchanger{clientExchanger, serverExchanger} {
+ go func(exchanger BoxKeyExchanger) {
+ crypter, err := NewBoxCrypter(exchanger, iobuf.NewPool(0))
if err != nil {
t.Fatal(err)
}
crypters <- crypter
- }(conn)
+ }(ex)
}
return <-crypters, <-crypters
}
diff --git a/profiles/internal/rpc/stream/manager/error_test.go b/profiles/internal/rpc/stream/manager/error_test.go
index 9c33348..7d6e27e 100644
--- a/profiles/internal/rpc/stream/manager/error_test.go
+++ b/profiles/internal/rpc/stream/manager/error_test.go
@@ -77,7 +77,7 @@
func dropDataDialer(network, address string, timeout time.Duration) (net.Conn, error) {
matcher := func(read bool, msg message.T) bool {
switch msg.(type) {
- case *message.HopSetup:
+ case *message.Setup:
return true
}
return false
diff --git a/profiles/internal/rpc/stream/message/control.go b/profiles/internal/rpc/stream/message/control.go
index 93bfb31..0e94e48 100644
--- a/profiles/internal/rpc/stream/message/control.go
+++ b/profiles/internal/rpc/stream/message/control.go
@@ -13,6 +13,7 @@
"v.io/v23/verror"
inaming "v.io/x/ref/profiles/internal/naming"
+ "v.io/x/ref/profiles/internal/rpc/stream/crypto"
"v.io/x/ref/profiles/internal/rpc/stream/id"
"v.io/x/ref/profiles/internal/rpc/version"
)
@@ -60,11 +61,10 @@
// the two ends of a VC to establish a protocol version.
type SetupVC struct {
VCI id.VC
- Versions version.Range // Version range supported by the sender.
LocalEndpoint naming.Endpoint // Endpoint of the sender (as seen by the sender), can be nil.
RemoteEndpoint naming.Endpoint // Endpoint of the receiver (as seen by the sender), can be nil.
Counters Counters
- Options []SetupOption
+ Setup Setup // Negotiate versioning and encryption.
}
// AddReceiveBuffers is a Control implementation used by the sender of the
@@ -83,14 +83,13 @@
InitialCounters uint32
}
-// HopSetup is a control message used to negotiate VIF options on a
-// hop-by-hop basis.
-type HopSetup struct {
+// Setup is a control message used to negotiate VIF/VC options.
+type Setup struct {
Versions version.Range
Options []SetupOption
}
-// SetupOption is the base interface for optional HopSetup and SetupVC options.
+// SetupOption is the base interface for optional Setup options.
type SetupOption interface {
// code is the identifier for the option.
code() setupOptionCode
@@ -108,17 +107,17 @@
// NaclBox is a SetupOption that specifies the public key for the NaclBox
// encryption protocol.
type NaclBox struct {
- PublicKey [32]byte
+ PublicKey crypto.BoxKey
}
-// 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
+// SetupStream is a byte stream used to negotiate VIF setup. During VIF setup,
+// each party sends a Setup 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 SetupStream 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 {
+type SetupStream struct {
Data []byte
}
@@ -153,9 +152,9 @@
command = addReceiveBuffersCommand
case *OpenFlow:
command = openFlowCommand
- case *HopSetup:
+ case *Setup:
command = hopSetupCommand
- case *HopSetupStream:
+ case *SetupStream:
command = hopSetupStreamCommand
case *SetupVC:
command = setupVCCommand
@@ -191,9 +190,9 @@
case openFlowCommand:
m = new(OpenFlow)
case hopSetupCommand:
- m = new(HopSetup)
+ m = new(Setup)
case hopSetupStreamCommand:
- m = new(HopSetupStream)
+ m = new(SetupStream)
case setupVCCommand:
m = new(SetupVC)
default:
@@ -268,12 +267,6 @@
if err = writeInt(w, m.VCI); err != nil {
return
}
- if err = writeInt(w, m.Versions.Min); err != nil {
- return
- }
- if err = writeInt(w, m.Versions.Max); err != nil {
- return
- }
var localep string
if m.LocalEndpoint != nil {
localep = m.LocalEndpoint.String()
@@ -291,7 +284,7 @@
if err = writeCounters(w, m.Counters); err != nil {
return
}
- if err = writeSetupOptions(w, m.Options); err != nil {
+ if err = m.Setup.writeTo(w); err != nil {
return
}
return
@@ -301,12 +294,6 @@
if err = readInt(r, &m.VCI); err != nil {
return
}
- if err = readInt(r, &m.Versions.Min); err != nil {
- return
- }
- if err = readInt(r, &m.Versions.Max); err != nil {
- return
- }
var ep string
if err = readString(r, &ep); err != nil {
return
@@ -327,7 +314,7 @@
if m.Counters, err = readCounters(r); err != nil {
return
}
- if m.Options, err = readSetupOptions(r); err != nil {
+ if err = m.Setup.readFrom(r); err != nil {
return
}
return
@@ -368,7 +355,7 @@
return
}
-func (m *HopSetup) writeTo(w io.Writer) (err error) {
+func (m *Setup) writeTo(w io.Writer) (err error) {
if err = writeInt(w, m.Versions.Min); err != nil {
return
}
@@ -381,7 +368,7 @@
return
}
-func (m *HopSetup) readFrom(r *bytes.Buffer) (err error) {
+func (m *Setup) readFrom(r *bytes.Buffer) (err error) {
if err = readInt(r, &m.Versions.Min); err != nil {
return
}
@@ -395,7 +382,7 @@
}
// NaclBox returns the first NaclBox option, or nil if there is none.
-func (m *HopSetup) NaclBox() *NaclBox {
+func (m *Setup) NaclBox() *NaclBox {
for _, opt := range m.Options {
if b, ok := opt.(*NaclBox); ok {
return b
@@ -422,12 +409,12 @@
return err
}
-func (m *HopSetupStream) writeTo(w io.Writer) error {
+func (m *SetupStream) writeTo(w io.Writer) error {
_, err := w.Write(m.Data)
return err
}
-func (m *HopSetupStream) readFrom(r *bytes.Buffer) error {
+func (m *SetupStream) readFrom(r *bytes.Buffer) error {
m.Data = r.Bytes()
return nil
}
diff --git a/profiles/internal/rpc/stream/message/message_test.go b/profiles/internal/rpc/stream/message/message_test.go
index fc37033..913d390 100644
--- a/profiles/internal/rpc/stream/message/message_test.go
+++ b/profiles/internal/rpc/stream/message/message_test.go
@@ -12,6 +12,7 @@
"v.io/v23/naming"
"v.io/x/ref/profiles/internal/lib/iobuf"
+ "v.io/x/ref/profiles/internal/rpc/stream/crypto"
"v.io/x/ref/profiles/internal/rpc/version"
)
@@ -83,23 +84,27 @@
&SetupVC{
VCI: 1,
- Versions: version.Range{Min: 34, Max: 56},
LocalEndpoint: version.Endpoint("tcp", "batman.com:1990", naming.FixedRoutingID(0xba7)),
RemoteEndpoint: version.Endpoint("tcp", "bugsbunny.com:1940", naming.FixedRoutingID(0xbb)),
Counters: counters,
- Options: []SetupOption{
- &NaclBox{PublicKey: [32]byte{'h', 'e', 'l', 'l', 'o', 'w', 'o', 'r', 'l', 'd'}},
- &NaclBox{PublicKey: [32]byte{7, 67, 31}},
+ Setup: Setup{
+ Versions: version.Range{Min: 34, Max: 56},
+ Options: []SetupOption{
+ &NaclBox{PublicKey: crypto.BoxKey{'h', 'e', 'l', 'l', 'o', 'w', 'o', 'r', 'l', 'd'}},
+ &NaclBox{PublicKey: crypto.BoxKey{7, 67, 31}},
+ },
},
},
// SetupVC without endpoints
&SetupVC{
VCI: 1,
- Versions: version.Range{Min: 34, Max: 56},
Counters: counters,
- Options: []SetupOption{
- &NaclBox{PublicKey: [32]byte{'h', 'e', 'l', 'l', 'o', 'w', 'o', 'r', 'l', 'd'}},
- &NaclBox{PublicKey: [32]byte{7, 67, 31}},
+ Setup: Setup{
+ Versions: version.Range{Min: 34, Max: 56},
+ Options: []SetupOption{
+ &NaclBox{PublicKey: crypto.BoxKey{'h', 'e', 'l', 'l', 'o', 'w', 'o', 'r', 'l', 'd'}},
+ &NaclBox{PublicKey: crypto.BoxKey{7, 67, 31}},
+ },
},
},
@@ -108,15 +113,15 @@
&OpenFlow{VCI: 1, Flow: 10, InitialCounters: 1 << 24},
- &HopSetup{
+ &Setup{
Versions: version.Range{Min: 21, Max: 71},
Options: []SetupOption{
- &NaclBox{PublicKey: [32]byte{'h', 'e', 'l', 'l', 'o', 'w', 'o', 'r', 'l', 'd'}},
- &NaclBox{PublicKey: [32]byte{7, 67, 31}},
+ &NaclBox{PublicKey: crypto.BoxKey{'h', 'e', 'l', 'l', 'o', 'w', 'o', 'r', 'l', 'd'}},
+ &NaclBox{PublicKey: crypto.BoxKey{7, 67, 31}},
},
},
- &HopSetupStream{Data: []byte("HelloWorld")},
+ &SetupStream{Data: []byte("HelloWorld")},
}
var c testControlCipher
diff --git a/profiles/internal/rpc/stream/proxy/proxy.go b/profiles/internal/rpc/stream/proxy/proxy.go
index 184b209..09d3586 100644
--- a/profiles/internal/rpc/stream/proxy/proxy.go
+++ b/profiles/internal/rpc/stream/proxy/proxy.go
@@ -14,6 +14,7 @@
"v.io/v23/context"
"v.io/v23/naming"
"v.io/v23/rpc"
+ "v.io/v23/rpc/version"
"v.io/v23/security"
"v.io/v23/verror"
"v.io/v23/vom"
@@ -30,7 +31,7 @@
"v.io/x/ref/profiles/internal/rpc/stream/message"
"v.io/x/ref/profiles/internal/rpc/stream/vc"
"v.io/x/ref/profiles/internal/rpc/stream/vif"
- "v.io/x/ref/profiles/internal/rpc/version"
+ iversion "v.io/x/ref/profiles/internal/rpc/version"
"v.io/x/ref/lib/stats"
)
@@ -64,7 +65,6 @@
errFailedToForwardRxBufs = reg(".errFailedToForwardRxBufs", "failed to forward receive buffers{:3}")
errFailedToFowardDataMsg = reg(".errFailedToFowardDataMsg", "failed to forward data message{:3}")
errFailedToFowardOpenFlow = reg(".errFailedToFowardOpenFlow", "failed to forward open flow{:3}")
- errUnsupportedSetupVC = reg(".errUnsupportedSetupVC", "proxy support for SetupVC not implemented yet")
errServerNotBeingProxied = reg(".errServerNotBeingProxied", "no server with routing id {3} is being proxied")
errServerVanished = reg(".errServerVanished", "server with routing id {3} vanished")
)
@@ -314,7 +314,7 @@
response.Error = verror.Convert(verror.ErrUnknown, nil, err)
} else {
defer p.servers.Remove(server)
- ep, err := version.ProxiedEndpoint(server.VC.RemoteEndpoint().RoutingID(), p.endpoint())
+ ep, err := iversion.ProxiedEndpoint(server.VC.RemoteEndpoint().RoutingID(), p.endpoint())
if err != nil {
response.Error = verror.Convert(verror.ErrInternal, nil, err)
}
@@ -379,7 +379,7 @@
// Endpoint returns the endpoint of the proxy service. By Dialing a VC to this
// endpoint, processes can have their services exported through the proxy.
func (p *Proxy) endpoint() naming.Endpoint {
- ep := version.Endpoint(p.ln.Addr().Network(), p.pubAddress, p.rid)
+ ep := iversion.Endpoint(p.ln.Addr().Network(), p.pubAddress, p.rid)
if prncpl := p.principal; prncpl != nil {
for b, _ := range prncpl.BlessingsInfo(prncpl.BlessingStore().Default()) {
ep.Blessings = append(ep.Blessings, b)
@@ -534,30 +534,66 @@
case *message.AddReceiveBuffers:
p.proxy.routeCounters(p, m.Counters)
case *message.SetupVC:
+ // First let's ensure that we can speak a common protocol verison.
+ intersection, err := iversion.SupportedRange.Intersect(&m.Setup.Versions)
+ if err != nil {
+ p.SendCloseVC(m.VCI, verror.New(stream.ErrProxy, nil,
+ verror.New(errIncompatibleVersions, nil, err)))
+ break
+ }
+
dstrid := m.RemoteEndpoint.RoutingID()
if naming.Compare(dstrid, p.proxy.rid) || naming.Compare(dstrid, naming.NullRoutingID) {
// VC that terminates at the proxy.
- // TODO(ashankar,mattr): Implement this!
- p.SendCloseVC(m.VCI, verror.New(stream.ErrProxy, nil, verror.New(errUnsupportedSetupVC, nil)))
+ // See protocol.vdl for details on the protocol between the server and the proxy.
+ vcObj := p.NewServerSetupVC(m)
+ // route counters after creating the VC so counters to vc are not lost.
p.proxy.routeCounters(p, m.Counters)
+ if vcObj != nil {
+ server := &server{Process: p, VC: vcObj}
+ keyExchanger := func(pubKey *crypto.BoxKey) (*crypto.BoxKey, error) {
+ p.queue.Put(&message.SetupVC{
+ VCI: m.VCI,
+ Setup: message.Setup{
+ // Note that servers send clients not their actual supported versions,
+ // but the intersected range of the server and client ranges. This
+ // is important because proxies may have adjusted the version ranges
+ // along the way, and we should negotiate a version that is compatible
+ // with all intermediate hops.
+ Versions: *intersection,
+ Options: []message.SetupOption{&message.NaclBox{PublicKey: *pubKey}},
+ },
+ RemoteEndpoint: m.LocalEndpoint,
+ LocalEndpoint: p.proxy.endpoint(),
+ // TODO(mattr): Consider adding counters. See associated comment
+ // in vc.go:VC.HandshakeAcceptedVC for more details.
+ })
+ var theirPK *crypto.BoxKey
+ box := m.Setup.NaclBox()
+ if box != nil {
+ theirPK = &box.PublicKey
+ }
+ return theirPK, nil
+ }
+ go p.proxy.runServer(server, vcObj.HandshakeAcceptedVC(intersection.Max, p.proxy.principal, p.proxy.blessings, keyExchanger))
+ }
break
}
- dstprocess := p.proxy.servers.Process(dstrid)
- if dstprocess == nil {
- p.SendCloseVC(m.VCI, verror.New(stream.ErrProxy, nil, verror.New(errServerNotBeingProxied, nil, dstrid)))
- p.proxy.routeCounters(p, m.Counters)
- break
- }
+
srcVCI := m.VCI
+
d := p.Route(srcVCI)
if d == nil {
- // SetupVC involves two messages: One sent by
- // the initiator and one by the acceptor. The
- // routing table gets setup on the first
- // message, so if there is no destination
- // process - setup a routing table entry.
- // If d != nil, then this SetupVC message is
- // likely the one sent by the acceptor.
+ // SetupVC involves two messages: One sent by the initiator
+ // and one by the acceptor. The routing table gets setup on
+ // the first message, so if there is no route -
+ // setup a routing table entry.
+ dstprocess := p.proxy.servers.Process(dstrid)
+ if dstprocess == nil {
+ p.SendCloseVC(m.VCI, verror.New(stream.ErrProxy, nil, verror.New(errServerNotBeingProxied, nil, dstrid)))
+ p.proxy.routeCounters(p, m.Counters)
+ break
+ }
dstVCI := dstprocess.AllocVCI()
startRoutingVC(srcVCI, dstVCI, p, dstprocess)
if d = p.Route(srcVCI); d == nil {
@@ -566,6 +602,7 @@
break
}
}
+
// Forward the SetupVC message.
// Typically, a SetupVC message is accompanied with
// Counters for the new VC. Keep that in the forwarded
@@ -580,7 +617,10 @@
}
}
m.VCI = dstVCI
- dstprocess.queue.Put(m)
+ // Note that proxies rewrite the version range so that the final negotiated
+ // version will be compatible with all intermediate hops.
+ m.Setup.Versions = *intersection
+ d.Process.queue.Put(m)
p.proxy.routeCounters(p, counters)
case *message.OpenVC:
@@ -588,12 +628,12 @@
if naming.Compare(dstrid, p.proxy.rid) || naming.Compare(dstrid, naming.NullRoutingID) {
// VC that terminates at the proxy.
// See protocol.vdl for details on the protocol between the server and the proxy.
- vcObj := p.NewServerVC(m)
+ vcObj, vers := p.NewServerVC(m)
// route counters after creating the VC so counters to vc are not lost.
p.proxy.routeCounters(p, m.Counters)
if vcObj != nil {
server := &server{Process: p, VC: vcObj}
- go p.proxy.runServer(server, vcObj.HandshakeAcceptedVC(p.proxy.principal, p.proxy.blessings))
+ go p.proxy.runServer(server, vcObj.HandshakeAcceptedVC(vers, p.proxy.principal, p.proxy.blessings, nil))
}
break
}
@@ -620,7 +660,8 @@
m.VCI = dstVCI
dstprocess.queue.Put(m)
p.proxy.routeCounters(p, counters)
- case *message.HopSetup:
+
+ case *message.Setup:
// Set up the hop. This takes over the process during negotiation.
if p.isSetup {
// Already performed authentication. We don't do it again.
@@ -714,17 +755,17 @@
return p.servers[vci]
}
-func (p *process) NewServerVC(m *message.OpenVC) *vc.VC {
+func (p *process) NewServerVC(m *message.OpenVC) (*vc.VC, version.RPCVersion) {
p.mu.Lock()
defer p.mu.Unlock()
if vc := p.servers[m.VCI]; vc != nil {
vc.Close(verror.New(stream.ErrProxy, nil, verror.New(errDuplicateOpenVC, nil)))
- return nil
+ return nil, version.UnknownRPCVersion
}
- version, err := version.CommonVersion(m.DstEndpoint, m.SrcEndpoint)
+ vers, err := iversion.CommonVersion(m.DstEndpoint, m.SrcEndpoint)
if err != nil {
p.SendCloseVC(m.VCI, verror.New(stream.ErrProxy, nil, verror.New(errIncompatibleVersions, nil, err)))
- return nil
+ return nil, vers
}
vc := vc.InternalNew(vc.Params{
VCI: m.VCI,
@@ -733,7 +774,26 @@
Pool: p.pool,
ReserveBytes: message.HeaderSizeBytes,
Helper: p,
- Version: version,
+ })
+ p.servers[m.VCI] = vc
+ proxyLog().Infof("Registered VC %v from server on process %v", vc, p)
+ return vc, vers
+}
+
+func (p *process) NewServerSetupVC(m *message.SetupVC) *vc.VC {
+ p.mu.Lock()
+ defer p.mu.Unlock()
+ if vc := p.servers[m.VCI]; vc != nil {
+ vc.Close(verror.New(stream.ErrProxy, nil, verror.New(errDuplicateOpenVC, nil)))
+ return nil
+ }
+ vc := vc.InternalNew(vc.Params{
+ VCI: m.VCI,
+ LocalEP: m.RemoteEndpoint,
+ RemoteEP: m.LocalEndpoint,
+ Pool: p.pool,
+ ReserveBytes: message.HeaderSizeBytes,
+ Helper: p,
})
p.servers[m.VCI] = vc
proxyLog().Infof("Registered VC %v from server on process %v", vc, p)
diff --git a/profiles/internal/rpc/stream/vc/vc.go b/profiles/internal/rpc/stream/vc/vc.go
index 5e9113b..0b4edc4 100644
--- a/profiles/internal/rpc/stream/vc/vc.go
+++ b/profiles/internal/rpc/stream/vc/vc.go
@@ -48,6 +48,8 @@
errUnrecognizedFlow = reg(".errUnrecognizedFlow", "unrecognized flow")
errFailedToCreateWriterForFlow = reg(".errFailedToCreateWriterForFlow", "failed to create writer for Flow{:3}")
errConnectOnClosedVC = reg(".errConnectOnClosedVC", "connect on closed VC{:3}")
+ errHandshakeNotInProgress = reg(".errHandshakeNotInProgress", "Attempted to finish a VC handshake, but no handshake was in progress{:3}")
+ errClosedDuringHandshake = reg(".errCloseDuringHandshake", "VC closed during handshake{:3}")
errFailedToDecryptPayload = reg(".errFailedToDecryptPayload", "failed to decrypt payload{:3}")
errIgnoringMessageOnClosedVC = reg(".errIgnoringMessageOnClosedVC", "ignoring message for Flow {3} on closed VC {4}")
errVomTypeDecoder = reg(".errVomDecoder", "failed to create vom type decoder{:3}")
@@ -55,7 +57,7 @@
errFailedToCreateFlowForWireType = reg(".errFailedToCreateFlowForWireType", "fail to create a Flow for wire type{:3}")
errFlowForWireTypeNotAccepted = reg(".errFlowForWireTypeNotAccepted", "Flow for wire type not accepted{:3}")
errFailedToCreateTLSFlow = reg(".errFailedToCreateTLSFlow", "failed to create a Flow for setting up TLS{3:}")
- errFailedToSetupTLS = reg(".errFailedToSetupTLS", "failed to setup TLS{:3}")
+ errFailedToSetupEncryption = reg(".errFailedToSetupEncryption", "failed to setup channel encryption{:3}")
errFailedToCreateFlowForAuth = reg(".errFailedToCreateFlowForAuth", "failed to create a Flow for authentication{:3}")
errAuthFailed = reg(".errAuthFailed", "authentication failed{:3}")
errNoActiveListener = reg(".errNoActiveListener", "no active listener on VCI {3}")
@@ -109,9 +111,10 @@
closeReason error // reason why the VC was closed, possibly nil
closeCh chan struct{}
closed bool
+ version version.RPCVersion
+ remotePubKeyChan chan *crypto.BoxKey // channel which will receive the remote public key during setup.
helper Helper
- version version.RPCVersion
dataCache *dataCache // dataCache contains information that can shared between Flows from this VC.
}
@@ -210,7 +213,6 @@
Pool *iobuf.Pool // Byte pool used for read and write buffer allocations.
ReserveBytes uint // Number of padding bytes to reserve for headers.
Helper Helper
- Version version.RPCVersion
}
// InternalNew creates a new VC, which implements the stream.VC interface.
@@ -241,7 +243,6 @@
crypter: crypto.NewNullCrypter(),
closeCh: make(chan struct{}),
helper: p.Helper,
- version: p.Version,
dataCache: newDataCache(),
}
}
@@ -251,6 +252,12 @@
return vc.connectFID(vc.allocFID(), normalFlowPriority, opts...)
}
+func (vc *VC) Version() version.RPCVersion {
+ vc.mu.Lock()
+ defer vc.mu.Unlock()
+ return vc.version
+}
+
func (vc *VC) connectFID(fid id.Flow, priority bqueue.Priority, opts ...stream.FlowOpt) (stream.Flow, error) {
writer, err := vc.newWriter(fid, priority)
if err != nil {
@@ -450,15 +457,57 @@
return err
}
+// FinishHandshakeDialedVC should be called from another goroutine
+// after HandshakeDialedVC is called, when version/encryption
+// negotiation is complete.
+func (vc *VC) FinishHandshakeDialedVC(vers version.RPCVersion, remotePubKey *crypto.BoxKey) error {
+ vc.mu.Lock()
+ defer vc.mu.Unlock()
+ if vc.remotePubKeyChan == nil {
+ return verror.New(errHandshakeNotInProgress, nil, vc.VCI)
+ }
+ vc.remotePubKeyChan <- remotePubKey
+ vc.remotePubKeyChan = nil
+ vc.version = vers
+ return nil
+}
+
// HandshakeDialedVC completes initialization of the VC (setting up encryption,
// authentication etc.) under the assumption that the VC was initiated by the
// local process (i.e., the local process "Dial"ed to create the VC).
-func (vc *VC) HandshakeDialedVC(principal security.Principal, opts ...stream.VCOpt) error {
+// HandshakeDialedVC will not return until FinishHandshakeDialedVC is called
+// from another goroutine.
+// TODO(mattr): vers can be removed as a parameter when we get rid of TLS and
+// the version is always obtained via the Setup call.
+func (vc *VC) HandshakeDialedVC(principal security.Principal, vers version.RPCVersion, sendKey func(*crypto.BoxKey) error, opts ...stream.VCOpt) error {
+ var remotePubKeyChan chan *crypto.BoxKey
+ if sendKey != nil {
+ remotePubKeyChan = make(chan *crypto.BoxKey, 1)
+ vc.mu.Lock()
+ vc.remotePubKeyChan = remotePubKeyChan
+ vc.mu.Unlock()
+ }
+
// principal = nil means that we are running in SecurityNone and we don't need
- // to authenticate the VC.
+ // to authenticate the VC. We still need to negotiate versioning information,
+ // so we still set remotePubKeyChan and call sendKey, though we don't care about
+ // the resulting key.
if principal == nil {
+ if sendKey != nil {
+ if err := sendKey(nil); err != nil {
+ return err
+ }
+ // TODO(mattr): Error on some timeout so a non-responding server doesn't
+ // cause this to hang forever.
+ select {
+ case <-remotePubKeyChan:
+ case <-vc.closeCh:
+ return verror.New(errClosedDuringHandshake, nil, vc.VCI)
+ }
+ }
return nil
}
+
var (
tlsSessionCache crypto.TLSClientSessionCache
auth *ServerAuthorizer
@@ -472,19 +521,49 @@
}
}
- // Establish TLS
- handshakeConn, err := vc.connectFID(HandshakeFlowID, systemFlowPriority)
- if err != nil {
- return vc.appendCloseReason(verror.New(stream.ErrSecurity, nil, verror.New(errFailedToCreateTLSFlow, nil, err)))
- }
- crypter, err := crypto.NewTLSClient(handshakeConn, handshakeConn.LocalEndpoint(), handshakeConn.RemoteEndpoint(), tlsSessionCache, vc.pool)
- if err != nil {
- // Assume that we don't trust the server if the TLS handshake fails for any
- // reason other than EOF.
- if err == io.EOF {
- return vc.appendCloseReason(verror.New(stream.ErrNetwork, nil, verror.New(errFailedToSetupTLS, nil, err)))
+ var crypter crypto.Crypter
+
+ if sendKey != nil {
+ exchange := func(pubKey *crypto.BoxKey) (*crypto.BoxKey, error) {
+ if err := sendKey(pubKey); err != nil {
+ return nil, err
+ }
+ // TODO(mattr): Error on some timeout so a non-responding server doesn't
+ // cause this to hang forever.
+ select {
+ case theirKey := <-remotePubKeyChan:
+ return theirKey, nil
+ case <-vc.closeCh:
+ return nil, verror.New(errClosedDuringHandshake, nil, vc.VCI)
+ }
}
- return vc.appendCloseReason(verror.New(stream.ErrNotTrusted, nil, verror.New(errFailedToSetupTLS, nil, err)))
+ var err error
+ crypter, err = crypto.NewBoxCrypter(exchange, vc.pool)
+ if err != nil {
+ return vc.appendCloseReason(verror.New(stream.ErrNetwork, nil,
+ verror.New(errFailedToSetupEncryption, nil, err)))
+ }
+ // The version is set by FinishHandshakeDialedVC and exchange (called by
+ // NewBoxCrypter) will block until FinishHandshakeDialedVC is called, so we
+ // should look up the version now.
+ vers = vc.Version()
+ } else {
+ // Establish TLS
+ handshakeConn, err := vc.connectFID(HandshakeFlowID, systemFlowPriority)
+ if err != nil {
+ return vc.appendCloseReason(verror.New(stream.ErrSecurity, nil, verror.New(errFailedToCreateTLSFlow, nil, err)))
+ }
+ crypter, err = crypto.NewTLSClient(handshakeConn,
+ handshakeConn.LocalEndpoint(), handshakeConn.RemoteEndpoint(),
+ tlsSessionCache, vc.pool)
+ if err != nil {
+ // Assume that we don't trust the server if the TLS handshake fails for any
+ // reason other than EOF.
+ if err == io.EOF {
+ return vc.appendCloseReason(verror.New(stream.ErrNetwork, nil, verror.New(errFailedToSetupEncryption, nil, err)))
+ }
+ return vc.appendCloseReason(verror.New(stream.ErrNotTrusted, nil, verror.New(errFailedToSetupEncryption, nil, err)))
+ }
}
// Authenticate (exchange identities)
@@ -505,7 +584,13 @@
LocalEndpoint: vc.localEP,
RemoteEndpoint: vc.remoteEP,
}
- rBlessings, lBlessings, rDischarges, err := AuthenticateAsClient(authConn, crypter, params, auth, vc.version)
+ vc.mu.Lock()
+ vc.version = vers
+ vc.authFID = AuthFlowID
+ vc.handshakeFID = HandshakeFlowID
+ vc.mu.Unlock()
+
+ rBlessings, lBlessings, rDischarges, err := AuthenticateAsClient(authConn, crypter, params, auth, vers)
if err != nil || len(rBlessings.ThirdPartyCaveats()) == 0 {
authConn.Close()
if err != nil {
@@ -516,8 +601,6 @@
}
vc.mu.Lock()
- vc.handshakeFID = HandshakeFlowID
- vc.authFID = AuthFlowID
vc.crypter = crypter
vc.localPrincipal = principal
vc.localBlessings = lBlessings
@@ -530,7 +613,8 @@
return vc.appendCloseReason(err)
}
- vlog.VI(1).Infof("Client VC %v authenticated. RemoteBlessings:%v, LocalBlessings:%v", vc, rBlessings, lBlessings)
+ vlog.VI(1).Infof("Client VC %v authenticated. RemoteBlessings:%v, LocalBlessings:%v",
+ vc, rBlessings, lBlessings)
return nil
}
@@ -551,7 +635,12 @@
// 'principal' is the principal used by the server used during authentication.
// If principal is nil, then the VC expects to be used for unauthenticated, unencrypted communication.
// 'lBlessings' is presented to the client during authentication.
-func (vc *VC) HandshakeAcceptedVC(principal security.Principal, lBlessings security.Blessings, opts ...stream.ListenerOpt) <-chan HandshakeResult {
+func (vc *VC) HandshakeAcceptedVC(
+ vers version.RPCVersion,
+ principal security.Principal,
+ lBlessings security.Blessings,
+ exchange crypto.BoxKeyExchanger,
+ opts ...stream.ListenerOpt) <-chan HandshakeResult {
result := make(chan HandshakeResult, 1)
finish := func(ln stream.Listener, err error) chan HandshakeResult {
result <- HandshakeResult{ln, err}
@@ -576,39 +665,63 @@
if err != nil {
return finish(nil, err)
}
+ // TODO(mattr): We could instead send counters in the return SetupVC message
+ // and avoid this extra message. It probably doesn't make much difference
+ // so for now I'll leave it. May be a nice cleanup after we are always
+ // using SetupVC.
vc.helper.AddReceiveBuffers(vc.VCI(), SharedFlowID, DefaultBytesBufferedPerFlow)
// principal = nil means that we are running in SecurityNone and we don't need
// to authenticate the VC.
if principal == nil {
- return finish(ln, nil)
+ if exchange == nil {
+ return finish(ln, nil)
+ }
+ go func() {
+ _, err = exchange(nil)
+ result <- HandshakeResult{ln, err}
+ }()
+ return result
}
+ var crypter crypto.Crypter
+
go func() {
sendErr := func(err error) {
ln.Close()
result <- HandshakeResult{nil, vc.appendCloseReason(err)}
}
- // TODO(ashankar): There should be a timeout on this Accept
- // call. Otherwise, a malicious (or incompetent) client can
- // consume server resources by sending many OpenVC messages but
- // not following up with the handshake protocol. Same holds for
- // the identity exchange protocol.
- handshakeConn, err := ln.Accept()
- if err != nil {
- sendErr(verror.New(stream.ErrNetwork, nil, verror.New(errTLSFlowNotAccepted, nil, err)))
- return
- }
+
vc.mu.Lock()
vc.acceptHandshakeDone = make(chan struct{})
- vc.handshakeFID = vc.findFlowLocked(handshakeConn)
vc.mu.Unlock()
- // Establish TLS
- crypter, err := crypto.NewTLSServer(handshakeConn, handshakeConn.LocalEndpoint(), handshakeConn.RemoteEndpoint(), vc.pool)
- if err != nil {
- sendErr(verror.New(stream.ErrSecurity, nil, verror.New(errFailedToSetupTLS, nil, err)))
- return
+ if exchange != nil {
+ crypter, err = crypto.NewBoxCrypter(exchange, vc.pool)
+ if err != nil {
+ sendErr(verror.New(stream.ErrSecurity, nil, verror.New(errFailedToSetupEncryption, nil, err)))
+ return
+ }
+ } else {
+ // TODO(ashankar): There should be a timeout on this Accept
+ // call. Otherwise, a malicious (or incompetent) client can
+ // consume server resources by sending many OpenVC messages but
+ // not following up with the handshake protocol. Same holds for
+ // the identity exchange protocol.
+ handshakeConn, err := ln.Accept()
+ if err != nil {
+ sendErr(verror.New(stream.ErrNetwork, nil, verror.New(errTLSFlowNotAccepted, nil, err)))
+ return
+ }
+ vc.mu.Lock()
+ vc.handshakeFID = vc.findFlowLocked(handshakeConn)
+ vc.mu.Unlock()
+ // Establish TLS
+ crypter, err = crypto.NewTLSServer(handshakeConn, handshakeConn.LocalEndpoint(), handshakeConn.RemoteEndpoint(), vc.pool)
+ if err != nil {
+ sendErr(verror.New(stream.ErrSecurity, nil, verror.New(errFailedToSetupEncryption, nil, err)))
+ return
+ }
}
// Authenticate (exchange identities)
@@ -617,11 +730,12 @@
sendErr(verror.New(stream.ErrNetwork, nil, verror.New(errAuthFlowNotAccepted, nil, err)))
return
}
+
vc.mu.Lock()
vc.authFID = vc.findFlowLocked(authConn)
vc.mu.Unlock()
- rBlessings, lDischarges, err := AuthenticateAsServer(authConn, principal, lBlessings, dischargeClient, crypter, vc.version)
+ rBlessings, lDischarges, err := AuthenticateAsServer(authConn, principal, lBlessings, dischargeClient, crypter, vers)
if err != nil {
authConn.Close()
sendErr(verror.New(stream.ErrSecurity, nil, verror.New(errAuthFailed, nil, err)))
@@ -629,6 +743,7 @@
}
vc.mu.Lock()
+ vc.version = vers
vc.crypter = crypter
vc.localPrincipal = principal
vc.localBlessings = lBlessings
@@ -741,7 +856,7 @@
}
func (vc *VC) connectSystemFlows() error {
- if vc.version < version.RPCVersion8 {
+ if vc.Version() < version.RPCVersion8 {
return nil
}
conn, err := vc.connectFID(TypeFlowID, systemFlowPriority)
@@ -764,7 +879,7 @@
}
func (vc *VC) acceptSystemFlows(ln stream.Listener) error {
- if vc.version < version.RPCVersion8 {
+ if vc.Version() < version.RPCVersion8 {
return nil
}
conn, err := ln.Accept()
diff --git a/profiles/internal/rpc/stream/vc/vc_test.go b/profiles/internal/rpc/stream/vc/vc_test.go
index 649d671..a8a8e09 100644
--- a/profiles/internal/rpc/stream/vc/vc_test.go
+++ b/profiles/internal/rpc/stream/vc/vc_test.go
@@ -30,8 +30,10 @@
"v.io/x/ref/profiles/internal/lib/bqueue/drrqueue"
"v.io/x/ref/profiles/internal/lib/iobuf"
"v.io/x/ref/profiles/internal/rpc/stream"
+ "v.io/x/ref/profiles/internal/rpc/stream/crypto"
"v.io/x/ref/profiles/internal/rpc/stream/id"
"v.io/x/ref/profiles/internal/rpc/stream/vc"
+ iversion "v.io/x/ref/profiles/internal/rpc/version"
"v.io/x/ref/test/testutil"
)
@@ -46,12 +48,12 @@
// Convenience alias to avoid conflicts between the package name "vc" and variables called "vc".
DefaultBytesBufferedPerFlow = vc.DefaultBytesBufferedPerFlow
// Shorthands
- SecurityNone = options.SecurityNone
- SecurityTLS = options.SecurityConfidential
-
- LatestVersion = version.RPCVersion7
+ SecurityNone = options.SecurityNone
+ SecurityDefault = options.SecurityConfidential
)
+var LatestVersion = iversion.SupportedRange.Max
+
// testFlowEcho writes a random string of 'size' bytes on the flow and then
// ensures that the same string is read back.
func testFlowEcho(t *testing.T, flow stream.Flow, size int) {
@@ -86,7 +88,7 @@
}
}
-func TestHandshake(t *testing.T) {
+func TestHandshakeNoSecurity(t *testing.T) {
// When the principals are nil, no blessings should be sent over the wire.
h, vc, err := New(LatestVersion, nil, nil, nil, nil)
if err != nil {
@@ -170,7 +172,7 @@
// Test that mockDischargeClient implements vc.DischargeClient.
var _ vc.DischargeClient = (mockDischargeClient)(nil)
-func TestHandshakeTLS(t *testing.T) {
+func TestHandshake(t *testing.T) {
matchesError := func(got error, want string) error {
if (got == nil) && len(want) == 0 {
return nil
@@ -281,12 +283,16 @@
}
testFlowEcho(t, flow, 10)
}
-func TestConnect_Small(t *testing.T) { testConnect_Small(t, LatestVersion, SecurityNone) }
-func TestConnect_SmallTLS(t *testing.T) { testConnect_Small(t, LatestVersion, SecurityTLS) }
-func TestConnect_Small7(t *testing.T) { testConnect_Small(t, version.RPCVersion7, SecurityNone) }
-func TestConnect_Small7TLS(t *testing.T) { testConnect_Small(t, version.RPCVersion7, SecurityTLS) }
-func TestConnect_Small8(t *testing.T) { testConnect_Small(t, version.RPCVersion8, SecurityNone) }
-func TestConnect_Small8TLS(t *testing.T) { testConnect_Small(t, version.RPCVersion8, SecurityTLS) }
+func TestConnect_SmallNoSecurity(t *testing.T) { testConnect_Small(t, LatestVersion, SecurityNone) }
+func TestConnect_Small(t *testing.T) { testConnect_Small(t, LatestVersion, SecurityDefault) }
+func TestConnect_Small7NoSecurity(t *testing.T) {
+ testConnect_Small(t, version.RPCVersion7, SecurityNone)
+}
+func TestConnect_Small7(t *testing.T) { testConnect_Small(t, version.RPCVersion7, SecurityDefault) }
+func TestConnect_Small8NoSecurity(t *testing.T) {
+ testConnect_Small(t, version.RPCVersion8, SecurityNone)
+}
+func TestConnect_Small8(t *testing.T) { testConnect_Small(t, version.RPCVersion8, SecurityDefault) }
func testConnect(t *testing.T, securityLevel options.SecurityLevel) {
h, vc, err := NewSimple(LatestVersion, securityLevel)
@@ -300,8 +306,8 @@
}
testFlowEcho(t, flow, 10*DefaultBytesBufferedPerFlow)
}
-func TestConnect(t *testing.T) { testConnect(t, SecurityNone) }
-func TestConnectTLS(t *testing.T) { testConnect(t, SecurityTLS) }
+func TestConnectNoSecurity(t *testing.T) { testConnect(t, SecurityNone) }
+func TestConnect(t *testing.T) { testConnect(t, SecurityDefault) }
func testConnect_Version7(t *testing.T, securityLevel options.SecurityLevel) {
h, vc, err := NewSimple(LatestVersion, securityLevel)
@@ -315,8 +321,8 @@
}
testFlowEcho(t, flow, 10)
}
-func TestConnect_Version7(t *testing.T) { testConnect_Version7(t, SecurityNone) }
-func TestConnect_Version7TLS(t *testing.T) { testConnect_Version7(t, SecurityTLS) }
+func TestConnect_Version7NoSecurity(t *testing.T) { testConnect_Version7(t, SecurityNone) }
+func TestConnect_Version7(t *testing.T) { testConnect_Version7(t, SecurityDefault) }
// helper function for testing concurrent operations on multiple flows over the
// same VC. Such tests are most useful when running the race detector.
@@ -346,11 +352,11 @@
wg.Wait()
}
-func TestConcurrentFlows_1(t *testing.T) { testConcurrentFlows(t, SecurityNone, 10, 1) }
-func TestConcurrentFlows_1TLS(t *testing.T) { testConcurrentFlows(t, SecurityTLS, 10, 1) }
+func TestConcurrentFlows_1NOSecurity(t *testing.T) { testConcurrentFlows(t, SecurityNone, 10, 1) }
+func TestConcurrentFlows_1(t *testing.T) { testConcurrentFlows(t, SecurityDefault, 10, 1) }
-func TestConcurrentFlows_10(t *testing.T) { testConcurrentFlows(t, SecurityNone, 10, 10) }
-func TestConcurrentFlows_10TLS(t *testing.T) { testConcurrentFlows(t, SecurityTLS, 10, 10) }
+func TestConcurrentFlows_10NoSecurity(t *testing.T) { testConcurrentFlows(t, SecurityNone, 10, 10) }
+func TestConcurrentFlows_10(t *testing.T) { testConcurrentFlows(t, SecurityDefault, 10, 10) }
func testListen(t *testing.T, securityLevel options.SecurityLevel) {
h, vc, err := NewSimple(LatestVersion, securityLevel)
@@ -400,8 +406,8 @@
t.Errorf("Got (%d, %v) want (0, %v)", n, err, io.EOF)
}
}
-func TestListen(t *testing.T) { testListen(t, SecurityNone) }
-func TestListenTLS(t *testing.T) { testListen(t, SecurityTLS) }
+func TestListenNoSecurity(t *testing.T) { testListen(t, SecurityNone) }
+func TestListen(t *testing.T) { testListen(t, SecurityDefault) }
func testNewFlowAfterClose(t *testing.T, securityLevel options.SecurityLevel) {
h, _, err := NewSimple(LatestVersion, securityLevel)
@@ -414,8 +420,8 @@
t.Fatalf("New flows should not be accepted once the VC is closed")
}
}
-func TestNewFlowAfterClose(t *testing.T) { testNewFlowAfterClose(t, SecurityNone) }
-func TestNewFlowAfterCloseTLS(t *testing.T) { testNewFlowAfterClose(t, SecurityTLS) }
+func TestNewFlowAfterCloseNoSecurity(t *testing.T) { testNewFlowAfterClose(t, SecurityNone) }
+func TestNewFlowAfterClose(t *testing.T) { testNewFlowAfterClose(t, SecurityDefault) }
func testConnectAfterClose(t *testing.T, securityLevel options.SecurityLevel) {
h, vc, err := NewSimple(LatestVersion, securityLevel)
@@ -428,8 +434,8 @@
t.Fatalf("Got (%v, %v), want (nil, %q)", f, err, "myerr")
}
}
-func TestConnectAfterClose(t *testing.T) { testConnectAfterClose(t, SecurityNone) }
-func TestConnectAfterCloseTLS(t *testing.T) { testConnectAfterClose(t, SecurityTLS) }
+func TestConnectAfterCloseNoSecurity(t *testing.T) { testConnectAfterClose(t, SecurityNone) }
+func TestConnectAfterClose(t *testing.T) { testConnectAfterClose(t, SecurityDefault) }
// helper implements vc.Helper and also sets up a single VC.
type helper struct {
@@ -441,7 +447,7 @@
}
func createPrincipals(securityLevel options.SecurityLevel) (client, server security.Principal) {
- if securityLevel == SecurityTLS {
+ if securityLevel == SecurityDefault {
client = testutil.NewPrincipal("client")
server = testutil.NewPrincipal("server")
}
@@ -472,7 +478,6 @@
RemoteEP: serverEP,
Pool: iobuf.NewPool(0),
Helper: clientH,
- Version: v,
}
serverParams := vc.Params{
VCI: vci,
@@ -480,7 +485,6 @@
RemoteEP: clientEP,
Pool: iobuf.NewPool(0),
Helper: serverH,
- Version: v,
}
clientH.VC = vc.InternalNew(clientParams)
@@ -506,8 +510,38 @@
if server != nil {
bserver = server.BlessingStore().Default()
}
- c := serverH.VC.HandshakeAcceptedVC(server, bserver, lopts...)
- if err := clientH.VC.HandshakeDialedVC(client, vcopts...); err != nil {
+
+ var clientExchanger func(*crypto.BoxKey) error
+ var serverExchanger func(*crypto.BoxKey) (*crypto.BoxKey, error)
+ if v >= version.RPCVersion9 {
+ server, client := make(chan *crypto.BoxKey, 1), make(chan *crypto.BoxKey, 1)
+ clientExchanger = func(pubKey *crypto.BoxKey) error {
+ client <- pubKey
+ return clientH.VC.FinishHandshakeDialedVC(v, <-server)
+ }
+ serverExchanger = func(pubKey *crypto.BoxKey) (*crypto.BoxKey, error) {
+ server <- pubKey
+ return <-client, nil
+ }
+ }
+
+ var wg sync.WaitGroup
+ wg.Add(2)
+
+ var c <-chan vc.HandshakeResult
+ go func() {
+ defer wg.Done()
+ c = serverH.VC.HandshakeAcceptedVC(v, server, bserver, serverExchanger, lopts...)
+ }()
+ var err error
+ go func() {
+ defer wg.Done()
+ err = clientH.VC.HandshakeDialedVC(client, v, clientExchanger, vcopts...)
+ }()
+
+ wg.Wait()
+
+ if err != nil {
go func() { <-c }()
return nil, nil, err
}
diff --git a/profiles/internal/rpc/stream/vif/auth.go b/profiles/internal/rpc/stream/vif/auth.go
index d1cb403..19b262b 100644
--- a/profiles/internal/rpc/stream/vif/auth.go
+++ b/profiles/internal/rpc/stream/vif/auth.go
@@ -36,36 +36,36 @@
// privateData includes secret data we need for encryption.
type privateData struct {
- naclBoxPrivateKey [32]byte
+ naclBoxPrivateKey crypto.BoxKey
}
-// AuthenticateAsClient sends a HopSetup message if possible. If so, it chooses
+// AuthenticateAsClient sends a Setup 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 RPCVersion6 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
+// Setup message to the server, containing the client's supported
+// versions, and the client's crypto options. The Setup message
// is sent in the clear.
//
-// - When the server receives the HopSetup message, it calls
-// AuthenticateAsServer, which constructs a response HopSetup containing
+// - When the server receives the Setup message, it calls
+// AuthenticateAsServer, which constructs a response Setup containing
// the server's version range, and any crypto options.
//
// - For RPCVersion6 and RPCVersion7, 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 NewControlCipherRPC6, which is based on
+// SetupStream messages using NewControlCipherRPC6, which is based on
// code.google.com/p/go.crypto/nacl/box.
//
-// - Once the encrypted HopSetupStream channel is setup, the client and
+// - Once the encrypted SetupStream 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
+// Note that the Setup 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
+// including a hash of the Setup 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 RPCVersion6 becomes the only supported version).
func AuthenticateAsClient(writer io.Writer, reader *iobuf.Reader, versions *version.Range, params security.CallParams, auth *vc.ServerAuthorizer) (crypto.ControlCipher, error) {
@@ -74,6 +74,8 @@
}
if params.LocalPrincipal == nil {
// If there is no principal, we do not support encryption/authentication.
+ // TODO(ashankar, mattr): We should still exchange version information even
+ // if we don't do encryption.
var err error
versions, err = versions.Intersect(&version.Range{Min: 0, Max: rpcversion.RPCVersion5})
if err != nil {
@@ -85,7 +87,7 @@
}
// The client has not yet sent its public data. Construct it and send it.
- pvt, pub, err := makeHopSetup(versions)
+ pvt, pub, err := makeSetup(versions)
if err != nil {
return nil, verror.New(stream.ErrSecurity, nil, err)
}
@@ -98,7 +100,7 @@
if err != nil {
return nil, verror.New(stream.ErrNetwork, nil, err)
}
- ppub, ok := pmsg.(*message.HopSetup)
+ ppub, ok := pmsg.(*message.Setup)
if !ok {
return nil, verror.New(stream.ErrSecurity, nil, verror.New(errVersionNegotiationFailed, nil))
}
@@ -118,7 +120,7 @@
}
func authenticateAsClient(writer io.Writer, reader *iobuf.Reader, params security.CallParams, auth *vc.ServerAuthorizer,
- pvt *privateData, pub, ppub *message.HopSetup, version rpcversion.RPCVersion) (crypto.ControlCipher, error) {
+ pvt *privateData, pub, ppub *message.Setup, version rpcversion.RPCVersion) (crypto.ControlCipher, error) {
if version < rpcversion.RPCVersion6 {
return nil, verror.New(errUnsupportedEncryptVersion, nil, version, rpcversion.RPCVersion6)
}
@@ -136,12 +138,12 @@
return c, nil
}
-// AuthenticateAsServer handles a HopSetup message, choosing authentication
+// AuthenticateAsServer handles a Setup message, choosing authentication
// based on the max common version.
//
// See AuthenticateAsClient for a description of the negotiation.
func AuthenticateAsServer(writer io.Writer, reader *iobuf.Reader, versions *version.Range, principal security.Principal, lBlessings security.Blessings,
- dc vc.DischargeClient, ppub *message.HopSetup) (crypto.ControlCipher, error) {
+ dc vc.DischargeClient, ppub *message.Setup) (crypto.ControlCipher, error) {
var err error
if versions == nil {
versions = version.SupportedRange
@@ -155,7 +157,7 @@
}
// Create our public data and send it to the client.
- pvt, pub, err := makeHopSetup(versions)
+ pvt, pub, err := makeSetup(versions)
if err != nil {
return nil, err
}
@@ -178,7 +180,7 @@
}
func authenticateAsServerRPC6(writer io.Writer, reader *iobuf.Reader, principal security.Principal, lBlessings security.Blessings, dc vc.DischargeClient,
- pvt *privateData, pub, ppub *message.HopSetup, version rpcversion.RPCVersion) (crypto.ControlCipher, error) {
+ pvt *privateData, pub, ppub *message.Setup, version rpcversion.RPCVersion) (crypto.ControlCipher, error) {
if version < rpcversion.RPCVersion6 {
return nil, verror.New(errUnsupportedEncryptVersion, nil, version, rpcversion.RPCVersion6)
}
@@ -208,8 +210,8 @@
return nil
}
-// makeHopSetup constructs the options that this process can support.
-func makeHopSetup(versions *version.Range) (pvt privateData, pub message.HopSetup, err error) {
+// makeSetup constructs the options that this process can support.
+func makeSetup(versions *version.Range) (pvt privateData, pub message.Setup, err error) {
pub.Versions = *versions
var pubKey, pvtKey *[32]byte
pubKey, pvtKey, err = box.GenerateKey(rand.Reader)
diff --git a/profiles/internal/rpc/stream/vif/setup_conn.go b/profiles/internal/rpc/stream/vif/setup_conn.go
index 06d71a6..2359cbf 100644
--- a/profiles/internal/rpc/stream/vif/setup_conn.go
+++ b/profiles/internal/rpc/stream/vif/setup_conn.go
@@ -15,7 +15,7 @@
"v.io/x/ref/profiles/internal/rpc/stream/message"
)
-// setupConn writes the data to the net.Conn using HopSetupStream messages.
+// setupConn writes the data to the net.Conn using SetupStream messages.
type setupConn struct {
writer io.Writer
reader *iobuf.Reader
@@ -38,7 +38,7 @@
if err != nil {
return 0, err
}
- emsg, ok := msg.(*message.HopSetupStream)
+ emsg, ok := msg.(*message.SetupStream)
if !ok {
return 0, verror.New(stream.ErrSecurity, nil, verror.New(errVersionNegotiationFailed, nil))
}
@@ -57,7 +57,7 @@
if n > maxFrameSize {
n = maxFrameSize
}
- emsg := message.HopSetupStream{Data: buf[:n]}
+ emsg := message.SetupStream{Data: buf[:n]}
if err := message.WriteTo(s.writer, &emsg, s.cipher); err != nil {
return 0, err
}
diff --git a/profiles/internal/rpc/stream/vif/vif.go b/profiles/internal/rpc/stream/vif/vif.go
index dcd37b4..e59a153 100644
--- a/profiles/internal/rpc/stream/vif/vif.go
+++ b/profiles/internal/rpc/stream/vif/vif.go
@@ -20,6 +20,7 @@
"v.io/v23/context"
"v.io/v23/naming"
+ "v.io/v23/rpc/version"
"v.io/v23/security"
"v.io/v23/verror"
"v.io/v23/vtrace"
@@ -36,7 +37,7 @@
"v.io/x/ref/profiles/internal/rpc/stream/id"
"v.io/x/ref/profiles/internal/rpc/stream/message"
"v.io/x/ref/profiles/internal/rpc/stream/vc"
- "v.io/x/ref/profiles/internal/rpc/version"
+ iversion "v.io/x/ref/profiles/internal/rpc/version"
)
const pkgPath = "v.io/x/ref/profiles/internal/rpc/stream/vif"
@@ -115,7 +116,7 @@
// non-nil only in testing. nil is equivalent to using the versions
// actually supported by this RPC implementation (which is always
// what you want outside of tests).
- versions *version.Range
+ versions *iversion.Range
isClosedMu sync.Mutex
isClosed bool // GUARDED_BY(isClosedMu)
@@ -164,7 +165,7 @@
// As the name suggests, this method is intended for use only within packages
// placed inside v.io/x/ref/profiles/internal. Code outside the
// v.io/x/ref/profiles/internal/* packages should never call this method.
-func InternalNewDialedVIF(conn net.Conn, rid naming.RoutingID, principal security.Principal, versions *version.Range, onClose func(*VIF), opts ...stream.VCOpt) (*VIF, error) {
+func InternalNewDialedVIF(conn net.Conn, rid naming.RoutingID, principal security.Principal, versions *iversion.Range, onClose func(*VIF), opts ...stream.VCOpt) (*VIF, error) {
ctx := getDialContext(opts)
if ctx != nil {
var span vtrace.Span
@@ -210,7 +211,7 @@
// As the name suggests, this method is intended for use only within packages
// placed inside v.io/x/ref/profiles/internal. Code outside the
// v.io/x/ref/profiles/internal/* packages should never call this method.
-func InternalNewAcceptedVIF(conn net.Conn, rid naming.RoutingID, principal security.Principal, blessings security.Blessings, versions *version.Range, onClose func(*VIF), lopts ...stream.ListenerOpt) (*VIF, error) {
+func InternalNewAcceptedVIF(conn net.Conn, rid naming.RoutingID, principal security.Principal, blessings security.Blessings, versions *iversion.Range, onClose func(*VIF), lopts ...stream.ListenerOpt) (*VIF, error) {
pool := iobuf.NewPool(0)
reader := iobuf.NewReader(pool, conn)
var startTimeout time.Duration
@@ -223,7 +224,7 @@
return internalNew(conn, pool, reader, rid, id.VC(vc.NumReservedVCs)+1, versions, principal, blessings, startTimeout, onClose, upcqueue.New(), lopts, &crypto.NullControlCipher{})
}
-func internalNew(conn net.Conn, pool *iobuf.Pool, reader *iobuf.Reader, rid naming.RoutingID, initialVCI id.VC, versions *version.Range, principal security.Principal, blessings security.Blessings, startTimeout time.Duration, onClose func(*VIF), acceptor *upcqueue.T, listenerOpts []stream.ListenerOpt, c crypto.ControlCipher) (*VIF, error) {
+func internalNew(conn net.Conn, pool *iobuf.Pool, reader *iobuf.Reader, rid naming.RoutingID, initialVCI id.VC, versions *iversion.Range, principal security.Principal, blessings security.Blessings, startTimeout time.Duration, onClose func(*VIF), acceptor *upcqueue.T, listenerOpts []stream.ListenerOpt, c crypto.ControlCipher) (*VIF, error) {
var (
// Choose IDs that will not conflict with any other (VC, Flow)
// pairs. VCI 0 is never used by the application (it is
@@ -253,6 +254,10 @@
}
stopQ.Release(-1) // Disable flow control
+ if versions == nil {
+ versions = iversion.SupportedRange
+ }
+
vif := &VIF{
conn: conn,
pool: pool,
@@ -304,27 +309,52 @@
}
counters := message.NewCounters()
counters.Add(vc.VCI(), sharedFlowID, defaultBytesBufferedPerFlow)
- // TODO(ashankar,mattr): If remoteEP/localEP version ranges allow, then
- // use message.SetupVC instead of message.OpenVC.
- // Rough outline:
- // (1) Switch to NaclBox for VC encryption (thus the VC handshake will
- // no longer require the TLS flow and roundtrips for that).
- // (2) Send an appropriate SetupVC message in response to a received
- // SetupVC message.
- // (3) Use the SetupVC received from the remote end to establish the
- // exact protocol version to use.
- err = vif.sendOnExpressQ(&message.OpenVC{
- VCI: vc.VCI(),
- DstEndpoint: remoteEP,
- SrcEndpoint: vif.localEP,
- Counters: counters})
- if err != nil {
- vif.deleteVC(vc.VCI())
- err = verror.New(stream.ErrNetwork, nil, verror.New(errSendOnExpressQFailed, nil, err))
- vc.Close(err)
- return nil, err
+
+ ver, err := vif.versions.CommonVersion(vif.localEP, remoteEP)
+
+ switch {
+ case verror.ErrorID(err) == iversion.ErrDeprecatedVersion.ID || ver >= version.RPCVersion9:
+ sendPublicKey := func(pubKey *crypto.BoxKey) error {
+ var options []message.SetupOption
+ if pubKey != nil {
+ options = []message.SetupOption{&message.NaclBox{PublicKey: *pubKey}}
+ }
+ err := vif.sendOnExpressQ(&message.SetupVC{
+ VCI: vc.VCI(),
+ RemoteEndpoint: remoteEP,
+ LocalEndpoint: vif.localEP,
+ Counters: counters,
+ Setup: message.Setup{
+ Versions: *vif.versions,
+ Options: options,
+ },
+ })
+ if err != nil {
+ err = verror.New(stream.ErrNetwork, nil,
+ verror.New(errSendOnExpressQFailed, nil, err))
+ }
+ return err
+ }
+ if err = vc.HandshakeDialedVC(principal, ver, sendPublicKey, opts...); err != nil {
+ break
+ }
+ case err != nil:
+ break
+ default:
+ err = vif.sendOnExpressQ(&message.OpenVC{
+ VCI: vc.VCI(),
+ DstEndpoint: remoteEP,
+ SrcEndpoint: vif.localEP,
+ Counters: counters})
+ if err != nil {
+ err = verror.New(stream.ErrNetwork, nil, verror.New(errSendOnExpressQFailed, nil, err))
+ break
+ }
+ if err = vc.HandshakeDialedVC(principal, ver, nil, opts...); err != nil {
+ break
+ }
}
- if err := vc.HandshakeDialedVC(principal, opts...); err != nil {
+ if err != nil {
vif.deleteVC(vc.VCI())
vc.Close(err)
return nil, err
@@ -530,16 +560,103 @@
})
return nil
}
- go vif.acceptFlowsLoop(vc, vc.HandshakeAcceptedVC(vif.principal, vif.blessings, lopts...))
+ vers, err := vif.versions.CommonVersion(m.DstEndpoint, m.SrcEndpoint)
+ if err != nil {
+ vif.sendOnExpressQ(&message.CloseVC{
+ VCI: m.VCI,
+ Error: err.Error(),
+ })
+ return nil
+ }
+ go vif.acceptFlowsLoop(vc, vc.HandshakeAcceptedVC(vers, vif.principal, vif.blessings, nil, lopts...))
case *message.SetupVC:
- // TODO(ashankar,mattr): Handle this! See comment about SetupVC
- // in vif.Dial
+ // First, find the public key we need out of the message.
+ var theirPK *crypto.BoxKey
+ box := m.Setup.NaclBox()
+ if box != nil {
+ theirPK = &box.PublicKey
+ }
+
+ // If we dialed this VC, then this is a response and we should finish
+ // the vc handshake. Otherwise, this message is opening a new VC.
+ if vif.dialedVCI(m.VCI) {
+ vif.distributeCounters(m.Counters)
+ if vc, _, _ := vif.vcMap.Find(m.VCI); vc != nil {
+ intersection, err := vif.versions.Intersect(&m.Setup.Versions)
+ if err != nil {
+ vif.closeVCAndSendMsg(vc, false, err)
+ } else if err := vc.FinishHandshakeDialedVC(intersection.Max, theirPK); err != nil {
+ vif.closeVCAndSendMsg(vc, false, err)
+ }
+ return nil
+ }
+ vlog.VI(2).Infof("Ignoring SetupVC message %+v for unknown dialed VC", m)
+ return nil
+ }
+
+ // This is an accepted VC.
+ intersection, err := vif.versions.Intersect(&m.Setup.Versions)
+ if err != nil {
+ vlog.VI(2).Infof("SetupVC message %+v to VIF %s did not present compatible versions: %v", m, vif, err)
+ vif.sendOnExpressQ(&message.CloseVC{
+ VCI: m.VCI,
+ Error: err.Error(),
+ })
+ return nil
+ }
+ vif.muListen.Lock()
+ closed := vif.acceptor == nil || vif.acceptor.IsClosed()
+ lopts := vif.listenerOpts
+ vif.muListen.Unlock()
+ if closed {
+ vlog.VI(2).Infof("Ignoring SetupVC message %+v as VIF %s does not accept VCs", m, vif)
+ vif.sendOnExpressQ(&message.CloseVC{
+ VCI: m.VCI,
+ Error: "VCs not accepted",
+ })
+ return nil
+ }
+ var idleTimeout time.Duration
+ for _, o := range lopts {
+ switch v := o.(type) {
+ case vc.IdleTimeout:
+ idleTimeout = v.Duration
+ }
+ }
+ vc, err := vif.newVC(m.VCI, m.RemoteEndpoint, m.LocalEndpoint, idleTimeout, false)
+ if err != nil {
+ vif.sendOnExpressQ(&message.CloseVC{
+ VCI: m.VCI,
+ Error: err.Error(),
+ })
+ return nil
+ }
vif.distributeCounters(m.Counters)
- vif.sendOnExpressQ(&message.CloseVC{
- VCI: m.VCI,
- Error: "SetupVC handling not implemented yet",
- })
- vlog.VI(2).Infof("Received SetupVC message, but handling not yet implemented")
+ keyExchanger := func(pubKey *crypto.BoxKey) (*crypto.BoxKey, error) {
+ var options []message.SetupOption
+ if pubKey != nil {
+ options = []message.SetupOption{&message.NaclBox{PublicKey: *pubKey}}
+ }
+ err = vif.sendOnExpressQ(&message.SetupVC{
+ VCI: m.VCI,
+ Setup: message.Setup{
+ // Note that servers send clients not their actual supported versions,
+ // but the intersected range of the server and client ranges. This
+ // is important because proxies may have adjusted the version ranges
+ // along the way, and we should negotiate a version that is compatible
+ // with all intermediate hops.
+ Versions: *intersection,
+ Options: options,
+ },
+ RemoteEndpoint: m.LocalEndpoint,
+ LocalEndpoint: vif.localEP,
+ // TODO(mattr): Consider adding counters. See associated comment
+ // in vc.go:VC.HandshakeAcceptedVC for more details.
+ })
+ return theirPK, err
+ }
+ go vif.acceptFlowsLoop(vc, vc.HandshakeAcceptedVC(intersection.Max, vif.principal, vif.blessings, keyExchanger, lopts...))
+
case *message.CloseVC:
if vc, _, _ := vif.vcMap.Find(m.VCI); vc != nil {
vif.deleteVC(vc.VCI())
@@ -567,7 +684,7 @@
return nil
}
vlog.VI(2).Infof("Ignoring OpenFlow(%+v) for unrecognized VCI on VIF %s", m, m, vif)
- case *message.HopSetup:
+ case *message.Setup:
// Configure the VIF. This takes over the conn during negotiation.
if vif.isSetup {
return verror.New(stream.ErrNetwork, nil, verror.New(errVIFAlreadySetup, nil))
@@ -601,7 +718,7 @@
}
m := qm.(*message.Data)
if err := vc.DispatchPayload(m.Flow, m.Payload); err != nil {
- vlog.VI(2).Infof("Ignoring data message %v for on VIF %v: %v", m, vif, err)
+ vlog.VI(2).Infof("Ignoring data message %v for on VIF %s: %v", m, vif, err)
}
if m.Close() {
vif.shutdownFlow(vc, m.Flow)
@@ -852,6 +969,10 @@
}
}
+func (vif *VIF) dialedVCI(VCI id.VC) bool {
+ return vif.nextVCI%2 == VCI%2
+}
+
func (vif *VIF) allocVCI() id.VC {
vif.muNextVCI.Lock()
ret := vif.nextVCI
@@ -861,13 +982,6 @@
}
func (vif *VIF) newVC(vci id.VC, localEP, remoteEP naming.Endpoint, idleTimeout time.Duration, dialed bool) (*vc.VC, error) {
- version, err := version.CommonVersion(localEP, remoteEP)
- if vif.versions != nil {
- version, err = vif.versions.CommonVersion(localEP, remoteEP)
- }
- if err != nil {
- return nil, err
- }
vif.muStartTimer.Lock()
if vif.startTimer != nil {
vif.startTimer.Stop()
@@ -889,7 +1003,6 @@
Pool: vif.pool,
ReserveBytes: uint(message.HeaderSizeBytes + macSize),
Helper: vcHelper{vif},
- Version: version,
})
added, rq, wq := vif.vcMap.Insert(vc)
if added {
@@ -1061,9 +1174,9 @@
// The addition of the endpoints is left as an excercise to higher layers of
// the stack, where the desire to share or hide blessings from the endpoint is
// clearer.
-func localEP(conn net.Conn, rid naming.RoutingID, versions *version.Range) naming.Endpoint {
+func localEP(conn net.Conn, rid naming.RoutingID, versions *iversion.Range) naming.Endpoint {
localAddr := conn.LocalAddr()
- ep := version.Endpoint(localAddr.Network(), localAddr.String(), rid)
+ ep := iversion.Endpoint(localAddr.Network(), localAddr.String(), rid)
if versions != nil {
ep = versions.Endpoint(localAddr.Network(), localAddr.String(), rid)
}
diff --git a/profiles/internal/rpc/stream/vif/vif_test.go b/profiles/internal/rpc/stream/vif/vif_test.go
index 6e20202..d598953 100644
--- a/profiles/internal/rpc/stream/vif/vif_test.go
+++ b/profiles/internal/rpc/stream/vif/vif_test.go
@@ -212,7 +212,7 @@
}
}
if nDiffs > 0 {
- t.Errorf("#Mismatches:%d #ReadSamples:%d #WriteSamples:", nDiffs, len(dataRead), len(dataWritten))
+ t.Errorf("#Mismatches:%d #ReadSamples:%d #WriteSamples:%d", nDiffs, len(dataRead), len(dataWritten))
}
}
}
@@ -246,7 +246,7 @@
var message = []byte("bugs bunny")
go func() {
if n, err := clientFlow.Write(message); n != len(message) || err != nil {
- t.Fatal("Wrote (%d, %v), want (%d, nil)", n, err, len(message))
+ t.Fatalf("Wrote (%d, %v), want (%d, nil)", n, err, len(message))
}
client.Close()
}()
@@ -258,7 +258,7 @@
}
// subsequent reads should fail, since the VIF should be closed.
if n, err := serverFlow.Read(buf); n != 0 || err == nil {
- t.Fatal("Got (%d, %v) = %q, want (0, nil)", n, err, buf[:n])
+ t.Fatalf("Got (%d, %v) = %q, want (0, nil)", n, err, buf[:n])
}
server.Close()
}
@@ -646,17 +646,17 @@
func TestIncompatibleVersions(t *testing.T) {
unknown := &iversion.Range{version.UnknownRPCVersion, version.UnknownRPCVersion}
tests := []versionTestCase{
- {&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{2, 2}, &iversion.Range{2, 2}, &iversion.Range{2, 2}, false, false},
+ {&iversion.Range{2, 3}, &iversion.Range{3, 5}, &iversion.Range{3, 5}, false, false},
+ {&iversion.Range{2, 3}, &iversion.Range{3, 5}, unknown, false, false},
// 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},
+ {&iversion.Range{2, 3}, &iversion.Range{4, 5}, &iversion.Range{4, 5}, true, false},
+ {&iversion.Range{2, 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},
+ {&iversion.Range{6, 6}, &iversion.Range{2, 5}, unknown, true, true},
}
for _, tc := range tests {
diff --git a/profiles/internal/rpc/version/version.go b/profiles/internal/rpc/version/version.go
index 4bcb1be..f536b12 100644
--- a/profiles/internal/rpc/version/version.go
+++ b/profiles/internal/rpc/version/version.go
@@ -26,13 +26,21 @@
// 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.RPCVersion5, Max: version.RPCVersion8}
+ SupportedRange = &Range{Min: version.RPCVersion5, Max: version.RPCVersion9}
// Export the methods on supportedRange.
Endpoint = SupportedRange.Endpoint
ProxiedEndpoint = SupportedRange.ProxiedEndpoint
CommonVersion = SupportedRange.CommonVersion
CheckCompatibility = SupportedRange.CheckCompatibility
+
+ // Which version to guess servers support if it's unknown.
+ // TODO(mattr): This is a hack. Once RPCVersion9 is released and versions
+ // are negotiated, we wont have to guess anymore and this code should
+ // be removed. This is required until version 9 is live.
+ // In fact, once version9 is the minimum supported version, much of the
+ // code in this file can be eliminated.
+ maxVersionGuess = version.RPCVersion8
)
const pkgPath = "v.io/x/ref/profiles/internal/rpc/version"
@@ -46,8 +54,9 @@
// level errors and hence {1}{2} is omitted from their format
// strings to avoid repeating these n-times in the final error
// message visible to the user.
- ErrNoCompatibleVersion = reg(".errNoCompatibleVersionErr", "No compatible RPC version available{:3} not in range {4}..{5}")
- ErrUnknownVersion = reg(".errUnknownVersionErr", "There was not enough information to determine a version")
+ ErrNoCompatibleVersion = reg(".errNoCompatibleVersionErr", "no compatible RPC version available{:3} not in range {4}..{5}")
+ ErrUnknownVersion = reg(".errUnknownVersionErr", "there was not enough information to determine a version")
+ ErrDeprecatedVersion = reg(".errDeprecatedVersionError", "some of the provided version information is deprecated")
errInternalTypeConversionError = reg(".errInternalTypeConversionError", "failed to convert {3} to v.io/ref/profiles/internal/naming.Endpoint {3}")
)
@@ -78,6 +87,14 @@
// a == (2, 4) and b == (Unknown, Unknown), intersect(a,b) == (2, 4)
// a == (2, Unknown) and b == (3, 4), intersect(a,b) == (3, 4)
func intersectRanges(amin, amax, bmin, bmax version.RPCVersion) (min, max version.RPCVersion, err error) {
+ // TODO(mattr): this may be incorrect. Ensure that when we talk to a server who
+ // advertises (5,8) and we support (5, 9) but v5 EPs (so we may get d, d here) that
+ // we use v8 and don't send setupVC.
+ d := version.DeprecatedRPCVersion
+ if amin == d || amax == d || bmin == d || bmax == d {
+ return d, d, verror.New(ErrDeprecatedVersion, nil)
+ }
+
u := version.UnknownRPCVersion
min = amin
@@ -88,6 +105,12 @@
if max == u || (bmax != u && bmax < max) {
max = bmax
}
+ // TODO(mattr): This is a hack. Once RPCVersion9 is released and versions
+ // are negotiated, we wont have to guess anymore and this code should
+ // be removed. This is required until version 9 is live.
+ if max > maxVersionGuess && (amax == u || bmax == u) {
+ max = maxVersionGuess
+ }
if min == u || max == u {
err = verror.New(ErrUnknownVersion, nil)
diff --git a/profiles/internal/rpc/version/version_test.go b/profiles/internal/rpc/version/version_test.go
index bc54048..a22e6cc 100644
--- a/profiles/internal/rpc/version/version_test.go
+++ b/profiles/internal/rpc/version/version_test.go
@@ -1,7 +1,6 @@
// Copyright 2015 The Vanadium 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 version
import (
@@ -15,7 +14,7 @@
)
func TestCommonVersion(t *testing.T) {
- r := &Range{Min: 1, Max: 3}
+ r := &Range{Min: 2, Max: 4}
type testCase struct {
localMin, localMax version.RPCVersion
@@ -25,13 +24,13 @@
}
tests := []testCase{
{0, 0, 0, 0, 0, ErrUnknownVersion},
- {0, 1, 2, 3, 0, ErrNoCompatibleVersion},
- {2, 3, 0, 1, 0, ErrNoCompatibleVersion},
- {0, 5, 5, 6, 0, ErrNoCompatibleVersion},
- {0, 2, 2, 4, 2, verror.ErrUnknown},
- {0, 2, 1, 3, 2, verror.ErrUnknown},
- {1, 3, 1, 3, 3, verror.ErrUnknown},
- {3, 3, 3, 3, 3, verror.ErrUnknown},
+ {0, 2, 3, 4, 0, ErrNoCompatibleVersion},
+ {3, 4, 0, 2, 0, ErrNoCompatibleVersion},
+ {0, 6, 6, 7, 0, ErrNoCompatibleVersion},
+ {0, 3, 3, 5, 3, verror.ErrUnknown},
+ {0, 3, 2, 4, 3, verror.ErrUnknown},
+ {2, 4, 2, 4, 4, verror.ErrUnknown},
+ {4, 4, 4, 4, 4, verror.ErrUnknown},
}
for _, tc := range tests {
local := &inaming.Endpoint{
@@ -61,13 +60,13 @@
expectError bool
}
tests := []testCase{
- {1, 3, 1, 2, 1, 2, false},
- {1, 3, 3, 5, 3, 3, false},
- {1, 3, 0, 1, 1, 1, false},
- {1, 3, 0, 1, 1, 1, false},
+ {2, 4, 2, 3, 2, 3, false},
+ {2, 4, 4, 6, 4, 4, false},
+ {2, 4, 0, 2, 2, 2, false},
+ {2, 4, 0, 2, 2, 2, false},
{0, 0, 0, 0, 0, 0, true},
- {2, 5, 0, 1, 0, 0, true},
- {2, 5, 6, 7, 0, 0, true},
+ {3, 4, 0, 2, 0, 0, true},
+ {3, 6, 7, 8, 0, 0, true},
}
rid := naming.FixedRoutingID(1)
@@ -102,11 +101,11 @@
}
tests := []testCase{
{0, 0, 0, 0, ErrUnknownVersion},
- {5, 10, 1, 4, ErrNoCompatibleVersion},
- {1, 4, 5, 10, ErrNoCompatibleVersion},
- {1, 10, 2, 9, verror.ErrUnknown},
- {3, 8, 1, 4, verror.ErrUnknown},
- {3, 8, 7, 9, verror.ErrUnknown},
+ {6, 11, 2, 5, ErrNoCompatibleVersion},
+ {2, 5, 6, 11, ErrNoCompatibleVersion},
+ {2, 11, 3, 10, verror.ErrUnknown},
+ {4, 9, 2, 5, verror.ErrUnknown},
+ {4, 9, 8, 10, verror.ErrUnknown},
}
for _, tc := range tests {
