diff --git a/go.mod b/go.mod index 8b1dc3aed..cbcf613db 100644 --- a/go.mod +++ b/go.mod @@ -38,7 +38,7 @@ require ( github.com/netfoundry/secretstream v0.1.2 github.com/openziti/fabric v0.14.16 github.com/openziti/foundation v0.14.15 - github.com/openziti/sdk-golang v0.13.61 + github.com/openziti/sdk-golang v0.14.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 4fa0fc03f..f2582ffb9 100644 --- a/go.sum +++ b/go.sum @@ -463,8 +463,8 @@ github.com/openziti/fabric v0.14.16 h1:l+5ZX4FoXHusYDq7bFRaqOFtLpIVuZ/lqYibTKEJw github.com/openziti/fabric v0.14.16/go.mod h1:aNECd87kH0bKntsHwPHtjbYf/9bk3MnlzQj21maO6FQ= github.com/openziti/foundation v0.14.15 h1:9QqOqo82wzws8zf5bfqezrmRj9E7AU1hnZHz8wiI9JY= github.com/openziti/foundation v0.14.15/go.mod h1:z98OIbCyr2B57EtamNgvDVIoQXct+zVVxOif9KsOOaM= -github.com/openziti/sdk-golang v0.13.61 h1:MWoqH04LkDVFHTpUwgo02c/WdwOxafQeRxQsz3h3M8E= -github.com/openziti/sdk-golang v0.13.61/go.mod h1:oIpkQERshTJxFE1k0c3Nw+i4mG+w07t3fhjCvxQXqHc= +github.com/openziti/sdk-golang v0.14.2 h1:YxH7Mq5A+4sgvKq0BoEId67vU7VVV8Z5GE1XPmWTAhM= +github.com/openziti/sdk-golang v0.14.2/go.mod h1:oIpkQERshTJxFE1k0c3Nw+i4mG+w07t3fhjCvxQXqHc= 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/fabric.go b/router/xgress_edge/fabric.go index 6c4f1cf85..215727cae 100644 --- a/router/xgress_edge/fabric.go +++ b/router/xgress_edge/fabric.go @@ -26,6 +26,19 @@ import ( "io" ) +const ( + PayloadFlagsHeader uint8 = 0x10 +) + +// headers to pass through fabric to the other side +var headersTofabric = map[int32]uint8{ + edge.FlagsHeader: PayloadFlagsHeader, +} + +var headersFromFabric = map[uint8]int32{ + PayloadFlagsHeader: edge.FlagsHeader, +} + type localMessageSink struct { edge.MsgChannel seq sequencer.Sequencer @@ -94,7 +107,15 @@ func (conn *localMessageSink) ReadPayload() ([]byte, map[uint8][]byte, error) { func (conn *localMessageSink) WritePayload(p []byte, headers map[uint8][]byte) (n int, err error) { msgUUID := headers[xgress.HeaderKeyUUID] - return conn.WriteTraced(p, msgUUID) + + edgeHdrs := make(map[int32][]byte) + for k, v := range headers { + if edgeHeader, found := headersFromFabric[k]; found { + edgeHdrs[edgeHeader] = v + } + } + + return conn.WriteTraced(p, msgUUID, edgeHdrs) } func (conn *localMessageSink) Close() error { @@ -141,9 +162,17 @@ func (conn *localMessageSink) Accept(event *edge.MsgEvent) { } func (conn *localMessageSink) getHeaderMap(message *channel2.Message) map[uint8][]byte { + headers := make(map[uint8][]byte) msgUUID, found := message.Headers[edge.UUIDHeader] if found { - return map[uint8][]byte{xgress.HeaderKeyUUID: msgUUID} + headers[xgress.HeaderKeyUUID] = msgUUID } - return nil + + for k, v := range message.Headers { + if pHdr, found := headersTofabric[k]; found { + headers[pHdr] = v + } + } + + return headers } diff --git a/router/xgress_edge_transport/connection.go b/router/xgress_edge_transport/connection.go index fdc6530ed..45f4cdcca 100644 --- a/router/xgress_edge_transport/connection.go +++ b/router/xgress_edge_transport/connection.go @@ -17,9 +17,13 @@ package xgress_edge_transport import ( + "errors" "fmt" + "github.com/michaelquigley/pfxlog" "github.com/netfoundry/secretstream" + "github.com/openziti/edge/router/xgress_edge" "github.com/openziti/foundation/transport" + "github.com/openziti/sdk-golang/ziti/edge" "io" "sync" ) @@ -37,6 +41,10 @@ type edgeTransportXgressConn struct { readCh chan *readResult startRead sync.Once startCrypto sync.Once + + writeDone chan bool + finSent bool + finRecv bool } func newEdgeTransportXgressConn(conn transport.Connection) *edgeTransportXgressConn { @@ -44,6 +52,7 @@ func newEdgeTransportXgressConn(conn transport.Connection) *edgeTransportXgressC Connection: conn, cryptoCtx: nil, readCh: make(chan *readResult), + writeDone: make(chan bool), } } @@ -65,7 +74,7 @@ func (c *edgeTransportXgressConn) readFromServer() { c.readCh <- result - if err == io.EOF { + if err != nil { return } } @@ -75,8 +84,26 @@ func (c *edgeTransportXgressConn) read() *readResult { return <-c.readCh } +var finHeaders = map[uint8][]byte{ + xgress_edge.PayloadFlagsHeader: {0x1, 0, 0, 0}, +} + func (c *edgeTransportXgressConn) ReadPayload() ([]byte, map[uint8][]byte, error) { + log := pfxlog.ContextLogger(c.Conn().RemoteAddr().String()) + + // block read to prevent xgress from tearing down connection until write side is also done + if c.finRecv { + log.Debug("waiting for write done") + <-c.writeDone + return nil, nil, io.EOF + } + readResult := c.read() + if readResult.err == io.EOF { + log.Debug("received EOF, returning FIN") + c.finRecv = true + return nil, finHeaders, nil + } if readResult.err != nil { return nil, nil, readResult.err @@ -148,5 +175,24 @@ func (c *edgeTransportXgressConn) Write(p []byte) (n int, err error) { } func (c *edgeTransportXgressConn) WritePayload(p []byte, headers map[uint8][]byte) (n int, err error) { - return c.Write(p) + if c.finSent { + return 0, errors.New("unexpected writePayload") + } + + if len(p) > 0 { + n, err = c.Write(p) + } + + if flags, found := headers[xgress_edge.PayloadFlagsHeader]; found { + if flags[0]|edge.FIN != 0 { + c.finSent = true + conn, ok := c.Conn().(edge.CloseWriter) + // if connection does not support half-close just let xgress tear it down + if ok { + _ = conn.CloseWrite() + } + close(c.writeDone) + } + } + return n, err } diff --git a/tests/context.go b/tests/context.go index fe2ca5abe..5cf63680f 100644 --- a/tests/context.go +++ b/tests/context.go @@ -340,6 +340,7 @@ func (ctx *TestContext) EnrollIdentity(identityId string) *sdkconfig.Config { flags := sdkenroll.EnrollmentFlags{ Token: tkn, + KeyAlg: "EC", } conf, err := sdkenroll.Enroll(flags) ctx.Req.NoError(err) diff --git a/tests/data_flow_close_test.go b/tests/data_flow_close_test.go index f0a9328dd..5081a4f85 100644 --- a/tests/data_flow_close_test.go +++ b/tests/data_flow_close_test.go @@ -316,3 +316,131 @@ func Test_ClientContextClosePropagation(t *testing.T) { ctx.Req.Equal(0, n) ctx.Req.Equal(err, io.EOF) } + +func Test_ServerConnCloseWritePropagation(t *testing.T) { + ctx := NewTestContext(t) + defer ctx.Teardown() + ctx.StartServer() + ctx.RequireAdminLogin() + + ctx.CreateEnrollAndStartEdgeRouter() + + service := ctx.AdminSession.RequireNewServiceAccessibleToAll("smartrouting") + + _, context := ctx.AdminSession.RequireCreateSdkContext() + listener, err := context.Listen(service.Name) + ctx.Req.NoError(err) + + clientIdentity := ctx.AdminSession.RequireNewIdentityWithOtt(false) + clientConfig := ctx.EnrollIdentity(clientIdentity.Id) + clientContext := ziti.NewContextWithConfig(clientConfig) + + errC := make(chan error, 1) + + go func() { + defer func() { + val := recover() + if val != nil { + if err, ok := val.(error); ok { + errC <- err + } else if str, ok := val.(string); ok { + errC <- errors.New(str) + } else { + errC <- errors.New(fmt.Sprintf("%v", val)) + } + } + close(errC) + }() + + conn := ctx.WrapConn(clientContext.Dial(service.Name)) + name := conn.ReadString(512, 2*time.Second) + n, err := conn.Read(make([]byte, 128)) + if err != io.EOF { + errC <- fmt.Errorf("did not receive EOF err(%v) %d", err, n) + } + conn.WriteString("hello, "+name+"\nI got your FIN!", time.Second) + conn.RequireClose() + }() + + conn := ctx.WrapNetConn(listener.Accept()) + name := eid.New() + conn.WriteString(name, time.Second) + _ = conn.CloseWrite() + + select { + case err := <-errC: + ctx.Req.NoError(err) + case <-time.After(2 * time.Second): + t.Fatal("timed out after 2 seconds") + } + + ctx.Req.NoError(conn.SetReadDeadline(time.Now().Add(time.Second))) + conn.ReadExpected("hello, "+name+"\nI got your FIN!", time.Second) + + n, err := conn.Read(make([]byte, 1024)) + ctx.Req.Equal(0, n) + ctx.Req.Equal(err, io.EOF) +} + +func Test_ClientConnCloseWritePropagation(t *testing.T) { + ctx := NewTestContext(t) + defer ctx.Teardown() + ctx.StartServer() + ctx.RequireAdminLogin() + + ctx.CreateEnrollAndStartEdgeRouter() + + service := ctx.AdminSession.RequireNewServiceAccessibleToAll("smartrouting") + + _, context := ctx.AdminSession.RequireCreateSdkContext() + listener, err := context.Listen(service.Name) + ctx.Req.NoError(err) + + defer func() { + ctx.Req.NoError(listener.Close()) + }() + + errC := make(chan error, 1) + + go func() { + defer func() { + val := recover() + if val != nil { + err := val.(error) + errC <- err + } + close(errC) + }() + + conn := ctx.WrapNetConn(listener.Accept()) + name := conn.ReadString(512, time.Second) + n, err := conn.Read(make([]byte, 128)) + if err != io.EOF { + errC <- fmt.Errorf("did not receive EOF err(%v) %d", err, n) + } + conn.WriteString("hello, "+name+"\nI got your FIN!", time.Second) + conn.RequireClose() + }() + + clientIdentity := ctx.AdminSession.RequireNewIdentityWithOtt(false) + clientConfig := ctx.EnrollIdentity(clientIdentity.Id) + clientContext := ziti.NewContextWithConfig(clientConfig) + + conn := ctx.WrapConn(clientContext.Dial(service.Name)) + name := eid.New() + conn.WriteString(name, time.Second) + _ = conn.CloseWrite() + conn.ReadExpected("hello, "+name+"\nI got your FIN!", time.Second) + + select { + case err := <-errC: + ctx.Req.NoError(err) + case <-time.After(2 * time.Second): + t.Fatal("timed out after 2 seconds") + } + + ctx.Req.NoError(conn.SetReadDeadline(time.Now().Add(time.Second))) + n, err := conn.Read(make([]byte, 1024)) + ctx.Req.Equal(0, n) + ctx.Req.Equal(err, io.EOF) +} diff --git a/tunnel/intercept/protocols/tcp/conn.go b/tunnel/intercept/protocols/tcp/conn.go index 082ee13c0..a199f0b94 100644 --- a/tunnel/intercept/protocols/tcp/conn.go +++ b/tunnel/intercept/protocols/tcp/conn.go @@ -17,6 +17,7 @@ package tcp import ( + "fmt" "github.com/openziti/edge/tunnel/intercept/protocols/ip" "github.com/sirupsen/logrus" "io" @@ -54,6 +55,7 @@ type ClientConn struct { mss uint tState tcpState + tStateMtx sync.Locker rcvAdv uint16 // advertised window from peer - written in tcpRecv. read in tcpSend (waitForClientAcks) rcvWndScale uint8 rcvAckNum uint32 @@ -63,6 +65,10 @@ type ClientConn struct { log *logrus.Entry } +func (conn ClientConn) String() string { + return fmt.Sprintf("tcpConn(%v -> %v)", conn.srcAddr, conn.dstAddr) +} + func NewClientConn(clientAddr, interceptAddr string, rxq chan *tcpQItem, dev io.ReadWriter, tunMTU uint) (*ClientConn, error) { srcAddr, err := net.ResolveTCPAddr("tcp", clientAddr) if err != nil { @@ -78,7 +84,7 @@ func NewClientConn(clientAddr, interceptAddr string, rxq chan *tcpQItem, dev io. for i := 0; i < cap(txq); i++ { txq <- make([]byte, tunMTU) } - + return &ClientConn{ clientKey: clientAddr, svcKey: interceptAddr, @@ -89,6 +95,7 @@ func NewClientConn(clientAddr, interceptAddr string, rxq chan *tcpQItem, dev io. dev: dev, mss: mss, tState: tcpsListen, + tStateMtx: new(sync.Mutex), seqCond: maxSegCond, log: log.WithFields(logrus.Fields{"src": srcAddr.String(), "dst": dstAddr.String()}), }, nil @@ -98,6 +105,7 @@ func (conn *ClientConn) tcpRecv(segment TCP, buf []byte) (int, error) { conn.log.Debugf("got packet: flags=%d, state=%d", segment.Flags(), conn.tState) conn.seqCond.L.Lock() + defer conn.seqCond.L.Unlock() conn.rcvAdv = segment.WindowSize() if segment.HasFlags(TCPFlagAck) { conn.rcvAckNum = segment.AckNumber() @@ -110,11 +118,15 @@ func (conn *ClientConn) tcpRecv(segment TCP, buf []byte) (int, error) { if segment.HasFlags(TCPFlagFin) { conn.sndAckNum += 1 } + n := 0 + + conn.tStateMtx.Lock() + defer conn.tStateMtx.Unlock() switch { // LISTEN --> SYN_RCVD case conn.tState == tcpsListen && segment.HasFlags(TCPFlagSyn): - synOpts := ParseSynOptions([]byte(segment[TCPMinimumSize:segment.DataOffset()]), segment.HasFlags(TCPFlagAck)) + synOpts := ParseSynOptions(segment[TCPMinimumSize:segment.DataOffset()], segment.HasFlags(TCPFlagAck)) if synOpts.WS > 0 { conn.rcvWndScale = uint8(synOpts.WS) } @@ -132,12 +144,11 @@ func (conn *ClientConn) tcpRecv(segment TCP, buf []byte) (int, error) { conn.tcpSend(nil, TCPFlagAck) } else { // DATA TRANSFER - n := len(segment.Payload()) + n = len(segment.Payload()) if n > 0 { conn.sndAckNum += uint32(n) conn.log.Debugf("writing %d bytes to ziti", n) conn.tcpSend(nil, TCPFlagAck) - conn.seqCond.L.Unlock() copy(buf, segment.Payload()) return n, nil } @@ -148,20 +159,29 @@ func (conn *ClientConn) tcpRecv(segment TCP, buf []byte) (int, error) { // CONNECTION TERMINATED by ziti service case conn.tState == tcpsFinWait1: + n = len(segment.Payload()) + conn.sndAckNum += uint32(n) + copy(buf, segment.Payload()) + if n > 0 || segment.HasFlags(TCPFlagFin) { + conn.tcpSend(nil, TCPFlagAck) + } switch { case segment.HasFlags(TCPFlagFin | TCPFlagAck): - conn.tcpSend(nil, TCPFlagAck) conn.tState = tcpsTimeWait case segment.HasFlags(TCPFlagFin): - conn.tcpSend(nil, TCPFlagAck) conn.tState = tcpsClosing case segment.HasFlags(TCPFlagAck): conn.tState = tcpsFinWait2 } - case conn.tState == tcpsFinWait2 && segment.HasFlags(TCPFlagFin): + case conn.tState == tcpsFinWait2: + n = len(segment.Payload()) + conn.sndAckNum += uint32(n) + copy(buf, segment.Payload()) + if segment.HasFlags(TCPFlagFin) { + conn.tState = tcpsTimeWait + } conn.tcpSend(nil, TCPFlagAck) - conn.tState = tcpsTimeWait case conn.tState == tcpsClosing && segment.HasFlags(TCPFlagAck): conn.tState = tcpsTimeWait @@ -181,14 +201,11 @@ func (conn *ClientConn) tcpRecv(segment TCP, buf []byte) (int, error) { conn.log.Infof("closing %s", conn.clientKey) // TODO return 0, io.EOF? case tcpsCloseWait: - conn.tState = tcpsLastAck conn.tcpSend(nil, TCPFlagFin) - conn.seqCond.L.Unlock() return 0, io.EOF } - conn.seqCond.L.Unlock() - return 0, nil + return n, nil } // Reads the next packet from the local client @@ -262,6 +279,14 @@ func (conn *ClientConn) tcpSend(payload []byte, flags uint8) { } func (conn *ClientConn) Write(payload []byte) (int, error) { + + switch conn.tState { + case tcpsSynReceived, tcpsEstablished, tcpsCloseWait: // allowed write states + default: + conn.log.Errorf("attempted write in %v", conn.tState) + return 0, fmt.Errorf("attempted write in %v state", conn.tState) + } + n := uint(len(payload)) var chunkLen uint for i := uint(0); i < n; i += conn.mss { @@ -292,23 +317,38 @@ func (conn *ClientConn) waitForClientAcks(i uint32) { } } +func (conn *ClientConn) CloseWrite() error { + conn.log.Debug("CloseWrite()") + conn.tStateMtx.Lock() + defer conn.tStateMtx.Unlock() + + sendFin := true + switch conn.tState { + case tcpsEstablished, tcpsSynReceived: + conn.tState = tcpsFinWait1 + case tcpsCloseWait: + conn.tState = tcpsLastAck + default: + sendFin = false + } + + if sendFin { + conn.tcpSend(nil, TCPFlagFin|TCPFlagAck) + conn.log.Debug("FIN sent") + } + + return nil +} + func (conn *ClientConn) Close() error { + _ = conn.CloseWrite() + queuesMtx.Lock() delete(queues, conn.clientKey) // prevent writes on outQueue which is about to be closed queuesMtx.Unlock() close(conn.rxq) - conn.log.Debugf("closing state = %d", conn.tState) - if conn.tState == tcpsEstablished || conn.tState == tcpsSynReceived { - conn.seqCond.L.Lock() - conn.waitForClientAcks(0) - conn.tcpSend(nil, TCPFlagFin) - conn.seqCond.L.Unlock() - conn.log.Debug("FIN sent") - conn.tState = tcpsFinWait1 - } - return nil } diff --git a/tunnel/intercept/tproxy/tproxy_linux.go b/tunnel/intercept/tproxy/tproxy_linux.go index 05dd0d4a1..c26ae3d6a 100644 --- a/tunnel/intercept/tproxy/tproxy_linux.go +++ b/tunnel/intercept/tproxy/tproxy_linux.go @@ -137,37 +137,35 @@ func New() (intercept.Interceptor, error) { } func (t *tProxyInterceptor) Start(context ziti.Context) { - t.accept(context) - t.acceptUDP(context) + go t.accept(context) + go t.acceptUDP(context) } func (t *tProxyInterceptor) accept(context ziti.Context) { log := pfxlog.Logger() - go func() { - for { - client, err := t.tcpLn.Accept() - if err != nil { - log.Errorf("error while accepting: %v", err) - } - if client == nil { - log.Info("shutting down") - return - } - log.Infof("received connection: %s --> %s", client.LocalAddr().String(), client.RemoteAddr().String()) - service, err := t.interceptLUT.GetByAddress(client.LocalAddr()) - if service == nil { - log.Warnf("received connection for %s, which does not map to an intercepted service", client.LocalAddr().String()) - client.Close() - continue - } - go tunnel.DialAndRun(context, service.Name, client) + for { + client, err := t.tcpLn.Accept() + if err != nil { + log.Errorf("error while accepting: %v", err) } - }() + if client == nil { + log.Info("shutting down") + return + } + log.Infof("received connection: %s --> %s", client.LocalAddr().String(), client.RemoteAddr().String()) + service, err := t.interceptLUT.GetByAddress(client.LocalAddr()) + if service == nil { + log.Warnf("received connection for %s, which does not map to an intercepted service", client.LocalAddr().String()) + client.Close() + continue + } + go tunnel.DialAndRun(context, service.Name, client) + } } func (t *tProxyInterceptor) acceptUDP(context ziti.Context) { vconnMgr := udp_vconn.NewManager(context, udp_vconn.NewUnlimitedConnectionPolicy(), udp_vconn.NewDefaultExpirationPolicy()) - go t.generateReadEvents(vconnMgr) + t.generateReadEvents(vconnMgr) } func (t *tProxyInterceptor) generateReadEvents(manager udp_vconn.Manager) { diff --git a/tunnel/tunnel.go b/tunnel/tunnel.go index a35d5734c..3aa4e205c 100644 --- a/tunnel/tunnel.go +++ b/tunnel/tunnel.go @@ -18,6 +18,7 @@ package tunnel import ( "github.com/openziti/sdk-golang/ziti" + "github.com/openziti/sdk-golang/ziti/edge" "github.com/sirupsen/logrus" "io" "net" @@ -51,6 +52,11 @@ func Run(zitiConn net.Conn, clientConn net.Conn) { go myCopy(zitiConn, clientConn, doneRecv) + defer func() { + _ = clientConn.Close() + _ = zitiConn.Close() + }() + var n1, n2 int64 for count := 2; count > 0; { select { @@ -68,8 +74,17 @@ func myCopy(dst net.Conn, src net.Conn, done chan int64) { "src-remote": src.RemoteAddr(), "src-local": src.LocalAddr(), "dst-local": dst.LocalAddr(), "dst-remote": dst.RemoteAddr()} - defer dst.Close() - defer log.WithFields(loggerFields).Info("stopping pipe") + logger := log.WithFields(loggerFields) + defer func() { + if cw, ok := dst.(edge.CloseWriter); ok { + _ = cw.CloseWrite() + } else { + _ = dst.Close() + } + + }() + + defer logger.Info("stopping pipe") // use smaller copyBuf so UDP payloads aren't chunked when sending to tunnelers with smaller MTU. // 17 bytes covers encryption overhead. copyBuf := make([]byte, 0x4000-17) diff --git a/version b/version index 8eac30c38..50653ad0a 100644 --- a/version +++ b/version @@ -1 +1 @@ -0.16 +0.17