gsadmin/ziti
1
0
Fork 0
mirror of https://github.com/openziti/ziti.git synced 2026-10-13 23:59:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #367 from openziti/support-passing-edge-headers

Browse Source 
implement TCP half-close support
This commit is contained in:
ekoby
2020-10-22 08:25:17 -04:00
committed by GitHub
parent 3406f0dc67 6126f179ad
commit 1b78a10740
10 changed files with 312 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files
go.mod
+1 -1
View File
@@ -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
go.sum
+2 -2
View File
@@ -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=
router/xgress_edge/fabric.go
+32 -3
View File
@@ -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
}
router/xgress_edge_transport/connection.go
+48 -2
View File
@@ -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
}
tests/context.go
+1
View File
@@ -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)
tests/data_flow_close_test.go
+128
View File
@@ -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)
}
tunnel/intercept/protocols/tcp/conn.go
+62 -22
View File
@@ -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
}
tunnel/intercept/tproxy/tproxy_linux.go
+20 -22
View File
@@ -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) {
tunnel/tunnel.go
+17 -2
View File
@@ -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)
version
+1 -1
View File
@@ -1 +1 @@
0.16
0.17