From 4f5dfd93cc2886dec185a3ca91a1d04976c2e758 Mon Sep 17 00:00:00 2001 From: Paul Lorenz Date: Thu, 10 Dec 2020 14:56:38 -0500 Subject: [PATCH 1/2] Support router provided connection ids --- router/xgress_edge/dialer.go | 78 +++++++++++----- router/xgress_edge/fabric.go | 15 ++- router/xgress_edge/listener.go | 25 ++--- tests/accept_manual_start_test.go | 150 ++++++++++++++++++++++++++++++ 4 files changed, 233 insertions(+), 35 deletions(-) create mode 100644 tests/accept_manual_start_test.go diff --git a/router/xgress_edge/dialer.go b/router/xgress_edge/dialer.go index 2777fd96b..763bb8fa9 100644 --- a/router/xgress_edge/dialer.go +++ b/router/xgress_edge/dialer.go @@ -93,27 +93,61 @@ 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() + dialRequest.PutUint32Header(edge.RouterProvidedConnId, connId) + + conn := listenConn.newSink(connId) + + x := xgress.NewXgress(sessionId, address, conn, xgress.Terminator, &dialer.options.Options) + bindHandler.HandleXgressBind(x) + + 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) + } + + x.Start() + return nil, nil + } else { + 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) } diff --git a/router/xgress_edge/fabric.go b/router/xgress_edge/fabric.go index 8b4423335..0604d53c8 100644 --- a/router/xgress_edge/fabric.go +++ b/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,17 @@ type localListener struct { terminatorIdRef *concurrenz.AtomicString service string parent *ingressProxy + assignIds bool + idSeq uint32 +} + +func (listener *localListener) nextDialConnId() uint32 { + nextId := atomic.AddUint32(&listener.idSeq, 1) + if nextId < math.MaxUint32/2 { + atomic.StoreUint32(&listener.idSeq, math.MaxUint32/2) + nextId = atomic.AddUint32(&listener.idSeq, 1) + } + return nextId } func (conn *localMessageSink) newSink(connId uint32) *localMessageSink { @@ -93,13 +106,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 } } diff --git a/router/xgress_edge/listener.go b/router/xgress_edge/listener.go index 722cac848..03f20df38 100644 --- a/router/xgress_edge/listener.go +++ b/router/xgress_edge/listener.go @@ -30,6 +30,7 @@ import ( "github.com/openziti/foundation/transport" "github.com/openziti/foundation/util/concurrenz" "github.com/openziti/sdk-golang/ziti/edge" + "math" "time" ) @@ -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 } @@ -262,6 +264,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 +288,8 @@ func (proxy *ingressProxy) processBind(req *channel2.Message, ch channel2.Channe terminatorIdRef: terminatorIdRef, service: ns.Service.Id, parent: proxy, + assignIds: assignIds, + idSeq: math.MaxUint32 / 2, } 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) { diff --git a/tests/accept_manual_start_test.go b/tests/accept_manual_start_test.go new file mode 100644 index 000000000..773e9577c --- /dev/null +++ b/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()) + } + +} From 7329106691e3c24665dba7ca236195d6508fea22 Mon Sep 17 00:00:00 2001 From: Paul Lorenz Date: Thu, 10 Dec 2020 15:28:56 -0500 Subject: [PATCH 2/2] Use manual start for tunnel hosting --- README.md | 2 -- .../internal/routes/terminator_api_model.go | 2 +- go.mod | 6 ++--- go.sum | 10 ++++----- router/xgress_edge/accept.go | 2 ++ router/xgress_edge/dialer.go | 11 ++++++++-- router/xgress_edge/fabric.go | 7 +++--- router/xgress_edge/listener.go | 6 ++--- tests/context.go | 22 +++++++++---------- tests/data_flow_close_test.go | 14 ++++++------ tunnel/intercept/svcpoll.go | 18 +++++++++++---- 11 files changed, 56 insertions(+), 44 deletions(-) diff --git a/README.md b/README.md index d37e48304..10497e4a3 100644 --- a/README.md +++ b/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. \ No newline at end of file diff --git a/controller/internal/routes/terminator_api_model.go b/controller/internal/routes/terminator_api_model.go index 8d3308c9f..d775dddf5 100644 --- a/controller/internal/routes/terminator_api_model.go +++ b/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 } diff --git a/go.mod b/go.mod index 1d48a8ba4..03b3754c2 100644 --- a/go.mod +++ b/go.mod @@ -1,6 +1,6 @@ module github.com/openziti/edge -go 1.14 +go 1.15 // replace github.com/openziti/foundation => ../foundation @@ -37,9 +37,9 @@ require ( github.com/miekg/dns v1.1.35 github.com/mitchellh/mapstructure v1.4.0 github.com/netfoundry/secretstream v0.1.2 - github.com/openziti/fabric v0.15.4 + github.com/openziti/fabric v0.15.5 github.com/openziti/foundation v0.14.24 - github.com/openziti/sdk-golang v0.15.0 + github.com/openziti/sdk-golang v0.15.2 github.com/orcaman/concurrent-map v0.0.0-20190826125027-8c72a8bb44f6 github.com/pkg/errors v0.9.1 github.com/sirupsen/logrus v1.7.0 diff --git a/go.sum b/go.sum index 4b1c9bef7..1b054b9ac 100644 --- a/go.sum +++ b/go.sum @@ -456,14 +456,12 @@ github.com/onsi/gomega v1.8.1/go.mod h1:Ho0h+IUsWyvy1OpqCwxlQ/21gkhVunqlU8fDGcoT github.com/onsi/gomega v1.10.1 h1:o0+MgICZLuZ7xjH7Vx6zS/zcu93/BEp1VwkIW1mEXCE= github.com/onsi/gomega v1.10.1/go.mod h1:iN09h71vgCQne3DLsj+A5owkum+a2tYe+TOCB1ybHNo= github.com/openzipkin/zipkin-go v0.1.1/go.mod h1:NtoC/o8u3JlF1lSlyPNswIbeQH9bJTmOf0Erfk+hxe8= -github.com/openziti/fabric v0.15.4 h1:UN1y9cHgeeWWnW9nPjMlQ2Ua/Mn6RKsaShy/0kSS1iU= -github.com/openziti/fabric v0.15.4/go.mod h1:A31vRi5JrbNw1mHOnNMTEDID5ga2kKyTD6TzzUqA5rQ= -github.com/openziti/foundation v0.14.23 h1:SHzPG2v2XCLBaUvh71NFezobctTb+KBEStiWObcziL0= -github.com/openziti/foundation v0.14.23/go.mod h1:jb1fNtW+s4I43Zo7etlCwoDkOglBhR6E0g6D6dp0Bu4= +github.com/openziti/fabric v0.15.5 h1:qaHsjkoGLb+0lHEmvZPSYnR5jKhdh59k3O4xYSzAkEM= +github.com/openziti/fabric v0.15.5/go.mod h1:A31vRi5JrbNw1mHOnNMTEDID5ga2kKyTD6TzzUqA5rQ= github.com/openziti/foundation v0.14.24 h1:bJ9Ni/QecYDx0tNS7tXzdF1789dI/87F9NzUisjpfBY= github.com/openziti/foundation v0.14.24/go.mod h1:jb1fNtW+s4I43Zo7etlCwoDkOglBhR6E0g6D6dp0Bu4= -github.com/openziti/sdk-golang v0.15.0 h1:Vuerj0IkHluKB92Fxmiyw9KUahABcdH/0uBZy3yiAac= -github.com/openziti/sdk-golang v0.15.0/go.mod h1:J2/UCmA3TXBiqTggu4QbceXP+kiqbfgHU7HbhAqnG2E= +github.com/openziti/sdk-golang v0.15.2 h1:A/ZXY7DZvLsuBUxtA7R5zltx1espHFFlsfkoki9hsFY= +github.com/openziti/sdk-golang v0.15.2/go.mod h1:wpMXCScWQ8Sdj/SX8GCp+/n/vQ4ZFJYTdZdOx9rdG/Y= github.com/orcaman/concurrent-map v0.0.0-20190826125027-8c72a8bb44f6 h1:lNCW6THrCKBiJBpz8kbVGjC7MgdCGKwuvBgc7LoD6sw= github.com/orcaman/concurrent-map v0.0.0-20190826125027-8c72a8bb44f6/go.mod h1:Lu3tH6HLW3feq74c2GC+jIMS/K2CFcDWnWD9XkenwhI= github.com/pascaldekloe/goe v0.0.0-20180627143212-57f6aae5913c/go.mod h1:lzWF7FIEvWOWxwDKqyGYQf6ZUaNfKdP144TG7ZOy1lc= diff --git a/router/xgress_edge/accept.go b/router/xgress_edge/accept.go index c875260d9..5ab6f6c42 100644 --- a/router/xgress_edge/accept.go +++ b/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) diff --git a/router/xgress_edge/dialer.go b/router/xgress_edge/dialer.go index 763bb8fa9..b039a9c4d 100644 --- a/router/xgress_edge/dialer.go +++ b/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 { @@ -95,12 +96,17 @@ func (dialer *dialer) Dial(destination string, sessionId *identity.TokenId, addr 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 { @@ -122,10 +128,11 @@ func (dialer *dialer) Dial(destination string, sessionId *identity.TokenId, addr x.Close() return nil, fmt.Errorf(msg) } + log.Debug("dial success") - x.Start() 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 diff --git a/router/xgress_edge/fabric.go b/router/xgress_edge/fabric.go index 0604d53c8..70a061fa6 100644 --- a/router/xgress_edge/fabric.go +++ b/router/xgress_edge/fabric.go @@ -54,14 +54,13 @@ type localListener struct { service string parent *ingressProxy assignIds bool - idSeq uint32 } func (listener *localListener) nextDialConnId() uint32 { - nextId := atomic.AddUint32(&listener.idSeq, 1) + nextId := atomic.AddUint32(&listener.parent.idSeq, 1) if nextId < math.MaxUint32/2 { - atomic.StoreUint32(&listener.idSeq, math.MaxUint32/2) - nextId = atomic.AddUint32(&listener.idSeq, 1) + atomic.StoreUint32(&listener.parent.idSeq, math.MaxUint32/2) + nextId = atomic.AddUint32(&listener.parent.idSeq, 1) } return nextId } diff --git a/router/xgress_edge/listener.go b/router/xgress_edge/listener.go index 03f20df38..221b72ba2 100644 --- a/router/xgress_edge/listener.go +++ b/router/xgress_edge/listener.go @@ -30,7 +30,6 @@ import ( "github.com/openziti/foundation/transport" "github.com/openziti/foundation/util/concurrenz" "github.com/openziti/sdk-golang/ziti/edge" - "math" "time" ) @@ -84,6 +83,7 @@ type ingressProxy struct { listener *listener fingerprints cert.Fingerprints ch channel2.Channel + idSeq uint32 } func (proxy *ingressProxy) HandleClose(_ channel2.Channel) { @@ -194,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) { @@ -289,7 +290,6 @@ func (proxy *ingressProxy) processBind(req *channel2.Message, ch channel2.Channe service: ns.Service.Id, parent: proxy, assignIds: assignIds, - idSeq: math.MaxUint32 / 2, } proxy.listener.factory.hostedServices.Put(token, messageSink) diff --git a/tests/context.go b/tests/context.go index de4ea6fee..4a7a5b966 100644 --- a/tests/context.go +++ b/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() { diff --git a/tests/data_flow_close_test.go b/tests/data_flow_close_test.go index 5081a4f85..b1d4f2377 100644 --- a/tests/data_flow_close_test.go +++ b/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 { diff --git a/tunnel/intercept/svcpoll.go b/tunnel/intercept/svcpoll.go index 51591af41..56b97acdd 100644 --- a/tunnel/intercept/svcpoll.go +++ b/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) } }