Merge pull request #445 from openziti/router-provided-conn-ids

Support router provided connection ids

README.md

@@ -3,5 +3,3 @@
 Ziti is a modern, programmable network overlay with associated edge components, for application-embedded, zero trust network connectivity, written by developers for developers. Ziti allows developers to take control of networking while with secure connectivity and advanced security concepts such as Zero Trust.
 
 The Ziti Edge implements the application-embedded connectivity and zero trust components of the Ziti ecosystem.
-
-Currently working through migrating our internal repositories over to GitHub. Stay tuned.

controller/internal/routes/terminator_api_model.go

@@ -110,7 +110,7 @@ func MapTerminatorToRestEntity(ae *env.AppEnv, _ *response.RequestContext, e mod
 
 func MapTerminatorToRestModel(ae *env.AppEnv, terminator *network.Terminator) (*rest_model.TerminatorDetail, error) {
 
-	service, err := ae.Handlers.EdgeService.Read(terminator.Service)
+	service, err := ae.Handlers.Service.Read(terminator.Service)
 	if err != nil {
 		return nil, err
 	}

go.mod

@@ -1,6 +1,6 @@
 module github.com/openziti/edge
 
-go 1.14
+go 1.15
 
 // replace github.com/openziti/foundation => ../foundation
 

router/xgress_edge/accept.go

@@ -23,6 +23,7 @@ import (
 	"github.com/openziti/edge/router/internal/fabric"
 	"github.com/openziti/foundation/channel2"
 	"github.com/openziti/sdk-golang/ziti/edge"
+	"math"
 )
 
 type Accepter struct {
@@ -45,6 +46,7 @@ func (handler edgeBindHandler) BindChannel(ch channel2.Channel) error {
 		listener:     handler.listener,
 		fingerprints: fpg.FromCerts(ch.Certificates()),
 		ch:           ch,
+		idSeq:        math.MaxUint32 / 2,
 	}
 
 	log.Debug("peer fingerprints ", proxy.fingerprints)

router/xgress_edge/dialer.go

@@ -60,7 +60,7 @@ func newDialer(factory *Factory, options *Options) xgress.Dialer {
 }
 
 func (dialer *dialer) Dial(destination string, sessionId *identity.TokenId, address xgress.Address, bindHandler xgress.BindHandler) (xt.PeerData, error) {
-	log := pfxlog.Logger().WithField("token", sessionId.Token)
+	log := pfxlog.Logger().WithField("session", sessionId.Token)
 	destParts := strings.Split(destination, ":")
 	if len(destParts) != 2 {
 		return nil, fmt.Errorf("destination '%v' format is incorrect", destination)
@@ -77,6 +77,7 @@ func (dialer *dialer) Dial(destination string, sessionId *identity.TokenId, addr
 	if !found {
 		return nil, fmt.Errorf("host for token '%v' not found", token)
 	}
+	log = log.WithField("bindConnId", listenConn.Id())
 
 	callerId := ""
 	if sessionId.Data != nil {
@@ -93,27 +94,67 @@ func (dialer *dialer) Dial(destination string, sessionId *identity.TokenId, addr
 		dialRequest.Headers[edge.AppDataHeader] = appData
 	}
 
-	reply, err := listenConn.SendPrioritizedAndWaitWithTimeout(dialRequest, channel2.Highest, 5*time.Second)
-	if err != nil {
-		return nil, err
+	if listenConn.assignIds {
+		connId := listenConn.nextDialConnId()
+		log = log.WithField("connId", connId)
+		log.Debug("router assigned connId for dial")
+		dialRequest.PutUint32Header(edge.RouterProvidedConnId, connId)
+
+		conn := listenConn.newSink(connId)
+
+		// On the terminator, which this is, this only starts the txer, which pulls data from the link
+		// Since the opposing xgress doesn't start until this call returns, nothing should be coming this way yet
+		x := xgress.NewXgress(sessionId, address, conn, xgress.Terminator, &dialer.options.Options)
+		bindHandler.HandleXgressBind(x)
+		x.Start()
+
+		reply, err := listenConn.SendPrioritizedAndWaitWithTimeout(dialRequest, channel2.Highest, 5*time.Second)
+		if err != nil {
+			conn.close(false, err.Error())
+			x.Close()
+			return nil, err
+		}
+		result, err := edge.UnmarshalDialResult(reply)
+
+		if err != nil {
+			conn.close(false, err.Error())
+			x.Close()
+			return nil, err
+		}
+
+		if !result.Success {
+			msg := fmt.Sprintf("failed to establish connection with token %v. error: (%v)", token, result.Message)
+			conn.close(false, msg)
+			x.Close()
+			return nil, fmt.Errorf(msg)
+		}
+		log.Debug("dial success")
+
+		return nil, nil
+	} else {
+		log.Debug("router not assigning connId for dial")
+		reply, err := listenConn.SendPrioritizedAndWaitWithTimeout(dialRequest, channel2.Highest, 5*time.Second)
+		if err != nil {
+			return nil, err
+		}
+		result, err := edge.UnmarshalDialResult(reply)
+
+		if err != nil {
+			return nil, err
+		}
+
+		if !result.Success {
+			return nil, fmt.Errorf("failed to establish connection with token %v. error: (%v)", token, result.Message)
+		}
+
+		conn := listenConn.newSink(result.NewConnId)
+
+		x := xgress.NewXgress(sessionId, address, conn, xgress.Terminator, &dialer.options.Options)
+		bindHandler.HandleXgressBind(x)
+		x.Start()
+
+		start := edge.NewStateConnectedMsg(result.ConnId)
+		start.ReplyTo(reply)
+		return nil, listenConn.SendState(start)
 	}
-	result, err := edge.UnmarshalDialResult(reply)
-
-	if err != nil {
-		return nil, err
-	}
-
-	if !result.Success {
-		return nil, fmt.Errorf("failed to establish connection with token %v. error: (%v)", token, result.Message)
-	}
-
-	conn := listenConn.newSink(result.NewConnId)
-
-	x := xgress.NewXgress(sessionId, address, conn, xgress.Terminator, &dialer.options.Options)
-	bindHandler.HandleXgressBind(x)
-	x.Start()
-
-	start := edge.NewStateConnectedMsg(result.ConnId)
-	start.ReplyTo(reply)
-	return nil, listenConn.SendState(start)
 }

router/xgress_edge/fabric.go

@@ -23,6 +23,8 @@ import (
 	"github.com/openziti/foundation/util/concurrenz"
 	"github.com/openziti/sdk-golang/ziti/edge"
 	"io"
+	"math"
+	"sync/atomic"
 )
 
 const (
@@ -51,6 +53,16 @@ type localListener struct {
 	terminatorIdRef *concurrenz.AtomicString
 	service         string
 	parent          *ingressProxy
+	assignIds       bool
+}
+
+func (listener *localListener) nextDialConnId() uint32 {
+	nextId := atomic.AddUint32(&listener.parent.idSeq, 1)
+	if nextId < math.MaxUint32/2 {
+		atomic.StoreUint32(&listener.parent.idSeq, math.MaxUint32/2)
+		nextId = atomic.AddUint32(&listener.parent.idSeq, 1)
+	}
+	return nextId
 }
 
 func (conn *localMessageSink) newSink(connId uint32) *localMessageSink {
@@ -93,13 +105,11 @@ func (conn *localMessageSink) ReadPayload() ([]byte, map[uint8][]byte, error) {
 	case edge.ContentTypeStateClosed:
 		log.Debug("received close message, closing connection and returning EOF")
 		conn.close(false, "close message received")
-		conn.closeCB(conn.Id())
 		return nil, nil, io.EOF // io.EOF signals xgress to shutdown
 
 	default:
 		log.Error("unexpected message type, closing connection")
 		conn.close(false, "close message received")
-		conn.closeCB(conn.Id())
 		return nil, nil, io.EOF // io.EOF signals xgress to shutdown
 	}
 }

router/xgress_edge/listener.go

@@ -83,6 +83,7 @@ type ingressProxy struct {
 	listener     *listener
 	fingerprints cert.Fingerprints
 	ch           channel2.Channel
+	idSeq        uint32
 }
 
 func (proxy *ingressProxy) HandleClose(_ channel2.Channel) {
@@ -178,8 +179,9 @@ func (proxy *ingressProxy) processConnect(req *channel2.Message, ch channel2.Cha
 	}
 	sessionInfo, err := xgress.GetSession(proxy.listener.factory, ns.Id, service, peerData)
 	if err != nil {
-		log.Warn("failed to dial fabric ", err)
+		log.WithError(err).Warn("failed to dial fabric")
 		proxy.sendStateClosedReply(err.Error(), req)
+		conn.close(false, "failed to dial fabric")
 		return
 	}
 
@@ -192,9 +194,10 @@ func (proxy *ingressProxy) processConnect(req *channel2.Message, ch channel2.Cha
 
 	x := xgress.NewXgress(sessionInfo.SessionId, sessionInfo.Address, conn, xgress.Initiator, &proxy.listener.options.Options)
 	proxy.listener.bindHandler.HandleXgressBind(x)
-	x.Start()
 
+	// send the state_connected before starting the xgress. That way we can't get a state_closed before we get state_connected
 	proxy.sendStateConnectedReply(req, sessionInfo.SessionId.Data)
+	x.Start()
 }
 
 func (proxy *ingressProxy) processBind(req *channel2.Message, ch channel2.Channel) {
@@ -262,6 +265,9 @@ func (proxy *ingressProxy) processBind(req *channel2.Message, ch channel2.Channe
 		}
 	})
 
+	assignIds, _ := req.GetBoolHeader(edge.RouterProvidedConnId)
+	log.Debugf("client requested router provided connection ids: %v", assignIds)
+
 	log.Debug("establishing listener")
 	messageSink := &localListener{
 		localMessageSink: localMessageSink{
@@ -283,6 +289,7 @@ func (proxy *ingressProxy) processBind(req *channel2.Message, ch channel2.Channe
 		terminatorIdRef: terminatorIdRef,
 		service:         ns.Service.Id,
 		parent:          proxy,
+		assignIds:       assignIds,
 	}
 
 	proxy.listener.factory.hostedServices.Put(token, messageSink)
@@ -400,25 +407,21 @@ func (proxy *ingressProxy) processUpdateBind(req *channel2.Message, ch channel2.
 func (proxy *ingressProxy) sendStateConnectedReply(req *channel2.Message, hostData map[uint32][]byte) {
 	connId, _ := req.GetUint32Header(edge.ConnIdHeader)
 	msg := edge.NewStateConnectedMsg(connId)
+
+	if assignIds, _ := req.GetBoolHeader(edge.RouterProvidedConnId); assignIds {
+		msg.PutBoolHeader(edge.RouterProvidedConnId, true)
+	}
+
 	for k, v := range hostData {
 		msg.Headers[int32(k)] = v
 	}
 	msg.ReplyTo(req)
 
-	syncC, err := proxy.ch.SendAndSyncWithPriority(msg, channel2.High)
+	err := proxy.ch.SendPrioritizedWithTimeout(msg, channel2.High, time.Second*5)
 	if err != nil {
 		pfxlog.Logger().WithFields(edge.GetLoggerFields(msg)).WithError(err).Error("failed to send state response")
 		return
 	}
-
-	select {
-	case err = <-syncC:
-		if err != nil {
-			pfxlog.Logger().WithFields(edge.GetLoggerFields(msg)).WithError(err).Error("failed to send state response")
-		}
-	case <-time.After(time.Second * 5):
-		pfxlog.Logger().WithFields(edge.GetLoggerFields(msg)).WithError(err).Error("timed out sending state response")
-	}
 }
 
 func (proxy *ingressProxy) sendStateClosedReply(message string, req *channel2.Message) {

tests/accept_manual_start_test.go

@@ -0,0 +1,150 @@
+// +build apitests
+
+/*
+	Copyright NetFoundry, Inc.
+
+	Licensed under the Apache License, Version 2.0 (the "License");
+	you may not use this file except in compliance with the License.
+	You may obtain a copy of the License at
+
+	https://www.apache.org/licenses/LICENSE-2.0
+
+	Unless required by applicable law or agreed to in writing, software
+	distributed under the License is distributed on an "AS IS" BASIS,
+	WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+	See the License for the specific language governing permissions and
+	limitations under the License.
+*/
+
+package tests
+
+import (
+	"github.com/michaelquigley/pfxlog"
+	"github.com/openziti/fabric/controller/xt_ha"
+	"github.com/openziti/sdk-golang/ziti"
+	"github.com/openziti/sdk-golang/ziti/edge"
+	"github.com/pkg/errors"
+	"testing"
+	"time"
+)
+
+func Test_ManualStart(t *testing.T) {
+	ctx := NewTestContext(t)
+	defer ctx.Teardown()
+	ctx.StartServer()
+	ctx.RequireAdminLogin()
+
+	service := ctx.AdminSession.RequireNewServiceAccessibleToAll(xt_ha.NewFactory().GetStrategyName())
+
+	ctx.CreateEnrollAndStartEdgeRouter()
+
+	log := pfxlog.Logger()
+
+	type host struct {
+		id       *identity
+		context  ziti.Context
+		listener edge.Listener
+	}
+
+	log.Info("starting listener1")
+	host1 := &host{}
+	host1.id, host1.context = ctx.AdminSession.RequireCreateSdkContext()
+	listener, err := host1.context.ListenWithOptions(service.Name, &ziti.ListenOptions{
+		Precedence:  ziti.PrecedenceRequired,
+		ManualStart: true,
+	})
+	host1.listener = listener
+	ctx.Req.NoError(err)
+
+	log.Info("started listener1")
+
+	defer func() { _ = host1.listener.Close() }()
+
+	log.Info("starting listener2")
+	host2 := &host{}
+	host2.id, host2.context = ctx.AdminSession.RequireCreateSdkContext()
+	listener, err = host2.context.ListenWithOptions(service.Name, &ziti.ListenOptions{
+		Precedence:  ziti.PrecedenceDefault,
+		ManualStart: false,
+	})
+	host2.listener = listener
+	ctx.Req.NoError(err)
+	log.Info("started listener1")
+
+	defer func() { _ = host2.listener.Close() }()
+
+	go func() {
+		count := 0
+		for {
+			conn, err := host1.listener.AcceptEdge()
+			log.Info("accepted connection from listener1")
+			if err != nil {
+				pfxlog.Logger().WithError(err).Error("failure during accept")
+				return
+			}
+			if count < 10 {
+				if err := conn.CompleteAcceptSuccess(); err != nil {
+					pfxlog.Logger().WithError(err).Error("failure during complete accept success")
+					return
+				}
+				if _, err := conn.Write([]byte("one")); err != nil {
+					pfxlog.Logger().WithError(err).Error("failure during write")
+					return
+				}
+			} else {
+				conn.CompleteAcceptFailed(errors.New("test error"))
+			}
+			_ = conn.Close()
+			count++
+		}
+	}()
+
+	go func() {
+		for {
+			conn, err := host2.listener.AcceptEdge()
+			log.Info("accepted connection from listener2")
+			if err != nil {
+				pfxlog.Logger().WithError(err).Error("failure during accept")
+				return
+			}
+			if _, err := conn.Write([]byte("two")); err != nil {
+				pfxlog.Logger().WithError(err).Error("failure during write")
+				return
+			}
+			_ = conn.Close()
+		}
+	}()
+
+	_, context := ctx.AdminSession.RequireCreateSdkContext()
+	for i := 0; i < 10; i++ {
+		log.Infof("dialing: %v", i)
+		conn := ctx.WrapConn(context.Dial(service.Name))
+		log.Infof("connected: %v", i)
+		val := conn.ReadString(3, time.Second)
+		ctx.Req.Equal("one", val)
+		ctx.Req.NoError(conn.Close())
+	}
+
+	// allow up to three failures
+	for i := 0; i < 3; i++ {
+		log.Infof("dialing: %v", i)
+		conn, err := context.Dial(service.Name)
+		if err == nil {
+			conn := ctx.WrapConn(conn, err)
+			val := conn.ReadString(3, time.Second)
+			ctx.Req.Equal("two", val)
+			ctx.Req.NoError(conn.Close())
+		} else {
+			pfxlog.Logger().WithError(err).Info("allowed dial failure")
+		}
+	}
+
+	for i := 0; i < 10; i++ {
+		log.Infof("dialing: %v", i)
+		conn := ctx.WrapConn(context.Dial(service.Name))
+		val := conn.ReadString(3, time.Second)
+		ctx.Req.Equal("two", val)
+		ctx.Req.NoError(conn.Close())
+	}
+
+}

tests/context.go

@@ -76,7 +76,7 @@ const (
 )
 
 func init() {
-	pfxlog.Global(logrus.InfoLevel)
+	pfxlog.Global(logrus.DebugLevel)
 	pfxlog.SetPrefix("github.com/openziti/")
 	logrus.SetFormatter(pfxlog.NewFormatterStartingToday())
 
@@ -672,26 +672,24 @@ func (ctx *TestContext) requireEntityEnrolled(name string, entity *gabs.Containe
 	ctx.Req.Nil(expiresAt, "expected "+name+" with isVerified=true to have an nil enrollment expires at date")
 }
 
-func (ctx *TestContext) WrapNetConn(conn net.Conn, err error) *testConn {
+func (ctx *TestContext) WrapNetConn(conn edge.Conn, err error) *testConn {
 	ctx.Req.NoError(err)
-	serviceConn, ok := conn.(edge.ServiceConn)
-	ctx.Req.True(ok)
 	return &testConn{
-		ServiceConn: serviceConn,
-		ctx:         ctx,
+		Conn: conn,
+		ctx:  ctx,
 	}
 }
 
-func (ctx *TestContext) WrapConn(conn edge.ServiceConn, err error) *testConn {
+func (ctx *TestContext) WrapConn(conn edge.Conn, err error) *testConn {
 	ctx.Req.NoError(err)
 	return &testConn{
-		ServiceConn: conn,
-		ctx:         ctx,
+		Conn: conn,
+		ctx:  ctx,
 	}
 }
 
 type testConn struct {
-	edge.ServiceConn
+	edge.Conn
 	ctx *TestContext
 }
 
@@ -711,13 +709,13 @@ func (conn *testConn) ReadString(maxSize int, timeout time.Duration) string {
 
 	buf := make([]byte, maxSize)
 	n, err := conn.Read(buf)
-	conn.ctx.Req.NoError(err)
+	conn.ctx.Req.NoError(err, "read timeout on connId=%v", conn.Id())
 	return string(buf[:n])
 }
 
 func (conn *testConn) ReadExpected(expected string, timeout time.Duration) {
 	val := conn.ReadString(len(expected)+1, timeout)
-	conn.ctx.Req.Equal(expected, val)
+	conn.ctx.Req.Equal(expected, val, "read failure on connId=%v", conn.Id())
 }
 
 func (conn *testConn) RequireClose() {

tests/data_flow_close_test.go

@@ -58,7 +58,7 @@ func Test_ServerConnClosePropagation(t *testing.T) {
 			close(errC)
 		}()
 
-		conn := ctx.WrapNetConn(listener.Accept())
+		conn := ctx.WrapNetConn(listener.AcceptEdge())
 		name := conn.ReadString(512, time.Second)
 		conn.WriteString("hello, "+name, time.Second)
 		conn.RequireClose()
@@ -117,7 +117,7 @@ func Test_ServerContextClosePropagation(t *testing.T) {
 			close(errC)
 		}()
 
-		conn := ctx.WrapNetConn(listener.Accept())
+		conn := ctx.WrapNetConn(listener.AcceptEdge())
 		name := conn.ReadString(512, time.Second)
 		conn.WriteString("hello, "+name, time.Second)
 		context.Close()
@@ -177,7 +177,7 @@ func Test_ServerCloseListenerPropagation(t *testing.T) {
 			close(errC)
 		}()
 
-		conn := ctx.WrapNetConn(listener.Accept())
+		conn := ctx.WrapNetConn(listener.AcceptEdge())
 		name := conn.ReadString(512, time.Second)
 		conn.WriteString("hello, "+name, time.Second)
 		ctx.Req.NoError(listener.Close())
@@ -239,7 +239,7 @@ func Test_ClientConnClosePropagation(t *testing.T) {
 		conn.RequireClose()
 	}()
 
-	conn := ctx.WrapNetConn(listener.Accept())
+	conn := ctx.WrapNetConn(listener.AcceptEdge())
 	name := eid.New()
 	conn.WriteString(name, time.Second)
 	conn.ReadExpected("hello, "+name, time.Second)
@@ -299,7 +299,7 @@ func Test_ClientContextClosePropagation(t *testing.T) {
 		clientContext.Close()
 	}()
 
-	conn := ctx.WrapNetConn(listener.Accept())
+	conn := ctx.WrapNetConn(listener.AcceptEdge())
 	name := eid.New()
 	conn.WriteString(name, time.Second)
 	conn.ReadExpected("hello, "+name, time.Second)
@@ -362,7 +362,7 @@ func Test_ServerConnCloseWritePropagation(t *testing.T) {
 		conn.RequireClose()
 	}()
 
-	conn := ctx.WrapNetConn(listener.Accept())
+	conn := ctx.WrapNetConn(listener.AcceptEdge())
 	name := eid.New()
 	conn.WriteString(name, time.Second)
 	_ = conn.CloseWrite()
@@ -412,7 +412,7 @@ func Test_ClientConnCloseWritePropagation(t *testing.T) {
 			close(errC)
 		}()
 
-		conn := ctx.WrapNetConn(listener.Accept())
+		conn := ctx.WrapNetConn(listener.AcceptEdge())
 		name := conn.ReadString(512, time.Second)
 		n, err := conn.Read(make([]byte, 128))
 		if err != io.EOF {

tunnel/intercept/svcpoll.go

@@ -175,18 +175,20 @@ func updateServices(context ziti.Context, interceptor Interceptor, resolver dns.
 
 func host(context ziti.Context, svc *entities.Service) {
 	log := pfxlog.Logger()
-	listener, err := context.Listen(svc.Name)
+	options := ziti.DefaultListenOptions()
+	options.ManualStart = true
+	listener, err := context.ListenWithOptions(svc.Name, options)
 	if err != nil {
 		log.WithError(err).WithField("service", svc.Name).Errorf("error listening for service: %v", err)
 		return
 	}
-	defer listener.Close()
+	defer func() { _ = listener.Close() }()
 	config := svc.ServerConfig
 	for {
 		log.WithField("service", svc.Name).
 			WithField("dialAddr", config.String()).
 			Info("hosting service, waiting for connections")
-		conn, err := listener.Accept()
+		conn, err := listener.AcceptEdge()
 		if err != nil {
 			log.WithError(err).WithField("service", svc.Name).Error("closing listener for service")
 			return
@@ -197,9 +199,17 @@ func host(context ziti.Context, svc *entities.Service) {
 				WithField("service", svc.Name).
 				WithField("dialAddr", config.String()).
 				Error("dial failed")
-			conn.Close()
+			conn.CompleteAcceptFailed(err)
+			_ = conn.Close()
 			continue
 		}
+		if err := conn.CompleteAcceptSuccess(); err != nil {
+			log.WithError(err).
+				WithField("service", svc.Name).
+				WithField("dialAddr", config.String()).
+				Error("complete accept success failed")
+			_ = conn.Close()
+		}
 		go tunnel.Run(conn, externalConn)
 	}
 }