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 03e1993b8..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 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 2777fd96b..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 { @@ -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) } diff --git a/router/xgress_edge/fabric.go b/router/xgress_edge/fabric.go index 8b4423335..70a061fa6 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,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 } } diff --git a/router/xgress_edge/listener.go b/router/xgress_edge/listener.go index 722cac848..221b72ba2 100644 --- a/router/xgress_edge/listener.go +++ b/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) { 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()) + } + +} 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) } }