Close listener when service is deleted or permission is lost to the service. Fixes #813

Author: plorenz

ziti/edge/network/hosting_conn.go

@@ -367,7 +367,6 @@ func (conn *edgeHostConn) listen(session *rest_model.SessionDetail, service *res
 		baseListener: baseListener{
 			service: service,
 			acceptC: make(chan edge.Conn, 10),
-			errorC:  make(chan error, 1),
 		},
 		token:       *session.Token,
 		edgeChan:    conn,

ziti/edge/network/listener.go

@@ -18,24 +18,26 @@ package network
 
 import (
 	"fmt"
-	"github.com/michaelquigley/pfxlog"
-	"github.com/openziti/edge-api/rest_model"
-	"github.com/openziti/sdk-golang/xgress"
-	"github.com/openziti/sdk-golang/ziti/edge"
-	"github.com/pkg/errors"
 	"math"
 	"net"
 	"reflect"
 	"strings"
 	"sync"
 	"sync/atomic"
 	"time"
+
+	"github.com/michaelquigley/pfxlog"
+	"github.com/openziti/edge-api/rest_model"
+	"github.com/openziti/foundation/v2/concurrenz"
+	"github.com/openziti/sdk-golang/xgress"
+	"github.com/openziti/sdk-golang/ziti/edge"
+	"github.com/pkg/errors"
 )
 
 type baseListener struct {
 	service *rest_model.ServiceDetail
 	acceptC chan edge.Conn
-	errorC  chan error
+	err     concurrenz.AtomicValue[error]
 	closed  atomic.Bool
 }
 
@@ -79,10 +81,8 @@ func (listener *baseListener) AcceptEdge() (edge.Conn, error) {
 		}
 	}
 
-	select {
-	case err := <-listener.errorC:
+	if err := listener.err.Load(); err != nil {
 		return nil, fmt.Errorf("listener is closed (%w)", err)
-	default:
 	}
 
 	return nil, errors.New("listener is closed")
@@ -188,7 +188,6 @@ func NewMultiListener(service *rest_model.ServiceDetail, getSessionF func() *res
 		baseListener: baseListener{
 			service: service,
 			acceptC: make(chan edge.Conn),
-			errorC:  make(chan error),
 		},
 		listeners:      map[*edgeListener]struct{}{},
 		getSessionF:    getSessionF,
@@ -405,37 +404,37 @@ func (self *multiListener) accept(conn edge.Conn, ticker *time.Ticker) {
 }
 
 func (self *multiListener) Close() error {
-	self.closed.Store(true)
-
-	self.listenerLock.Lock()
-	defer self.listenerLock.Unlock()
+	if self.closed.CompareAndSwap(false, true) {
+		self.listenerLock.Lock()
+		defer self.listenerLock.Unlock()
 
-	var resultErrors []error
-	for child := range self.listeners {
-		if err := child.Close(); err != nil {
-			resultErrors = append(resultErrors, err)
+		var resultErrors []error
+		for child := range self.listeners {
+			if err := child.Close(); err != nil {
+				resultErrors = append(resultErrors, err)
+			}
 		}
-	}
 
-	self.listeners = nil
+		self.listeners = nil
 
-	select {
-	case self.acceptC <- nil:
-	default:
-		// If the queue is full, bail out, we're just popping a nil on the
-		// accept queue to let it return from accept more quickly
+		select {
+		case self.acceptC <- nil:
+		default:
+			// If the queue is full, bail out, we're just popping a nil on the
+			// accept queue to let it return from accept more quickly
+		}
+
+		return self.condenseErrors(resultErrors)
 	}
 
-	return self.condenseErrors(resultErrors)
+	return nil
 }
 
 func (self *multiListener) CloseWithError(err error) {
-	select {
-	case self.errorC <- err:
-	default:
+	self.err.Store(err)
+	if closeErr := self.Close(); closeErr != nil {
+		pfxlog.Logger().WithError(err).Error("error closing edge listener")
 	}
-
-	self.closed.Store(true)
 }
 
 type MultipleErrors []error

ziti/ziti.go

@@ -25,6 +25,7 @@ import (
 	"net"
 	"net/url"
 	"reflect"
+	"slices"
 	"strconv"
 	"strings"
 	"sync"
@@ -1757,7 +1758,12 @@ func (context *ContextImpl) createSessionWithBackoff(service *rest_model.Service
 	var session *rest_model.SessionDetail
 	operation := func() error {
 		latestSvc, _ := context.services.Get(*service.Name)
-		if latestSvc != nil && *latestSvc.ID != *service.ID {
+
+		if latestSvc == nil {
+			return backoff.Permanent(fmt.Errorf("service %v not found", *service.Name))
+		}
+
+		if *latestSvc.ID != *service.ID {
 			pfxlog.Logger().
 				WithField("serviceName", *service.Name).
 				WithField("oldServiceId", *service.ID).
@@ -1782,6 +1788,14 @@ func (context *ContextImpl) createSessionWithBackoff(service *rest_model.Service
 	return session, backoff.Retry(operation, expBackoff)
 }
 
+func (context *ContextImpl) isNotFoundApiError(err error) bool {
+	apiFormattedErr := &rest_util.APIFormattedError{}
+	if errors.As(err, &apiFormattedErr) {
+		return apiFormattedErr.APIError != nil && apiFormattedErr.Code == errorz.NotFoundCode
+	}
+	return false
+}
+
 func (context *ContextImpl) isUnauthorizedApiError(err error) bool {
 	apiFormattedErr := &rest_util.APIFormattedError{}
 	if errors.As(err, &apiFormattedErr) {
@@ -1815,7 +1829,7 @@ func (context *ContextImpl) createSession(service *rest_model.ServiceDetail, ses
 		}
 
 		var createSessionNotFound = &rest_session.CreateSessionNotFound{}
-		if errors.As(err, &createSessionNotFound) {
+		if context.isNotFoundApiError(err) || errors.As(err, &createSessionNotFound) {
 			if refreshErr := context.refreshServices(false, false); refreshErr != nil {
 				logger.WithError(refreshErr).Info("failed to refresh services after create session returned 404 (likely for service)")
 			}
@@ -2022,9 +2036,16 @@ func (mgr *listenerManager) notify(eventType ListenEventType) {
 
 func (mgr *listenerManager) run() {
 	log := pfxlog.Logger().WithField("service", stringz.OrEmpty(mgr.service.Name))
+
+	start := time.Now()
 	// need to either establish a session, or fail if we can't create one
 	for mgr.session == nil {
-		mgr.createSessionWithBackoff()
+		if err := mgr.createSessionWithBackoff(); err != nil {
+			if time.Since(start) > mgr.options.ConnectTimeout {
+				log.WithError(err).Error("timed out trying to create session to bind service")
+				return
+			}
+		}
 	}
 
 	mgr.makeMoreListeners()
@@ -2234,6 +2255,17 @@ func (mgr *listenerManager) makeMoreListeners() {
 	}
 }
 
+func (mgr *listenerManager) createSessionWithPermissionCheck() {
+	log := pfxlog.Logger().WithField("service", stringz.OrEmpty(mgr.service.Name))
+
+	if err := mgr.createSessionWithBackoff(); err != nil {
+		if IsServiceAccessDeniedError(err) {
+			log.WithError(err).Error("service access lost, closing service listener")
+			mgr.listener.CloseWithError(err)
+		}
+	}
+}
+
 func (mgr *listenerManager) refreshSession() {
 	if time.Since(mgr.lastSessionRefresh) < 30*time.Second {
 		return
@@ -2242,7 +2274,7 @@ func (mgr *listenerManager) refreshSession() {
 	log := pfxlog.Logger().WithField("service", stringz.OrEmpty(mgr.service.Name))
 	if mgr.session == nil {
 		log.Debug("establishing initial session")
-		mgr.createSessionWithBackoff()
+		mgr.createSessionWithPermissionCheck()
 		return
 	}
 
@@ -2254,7 +2286,7 @@ func (mgr *listenerManager) refreshSession() {
 		var detailSessionNotFound = &rest_session.DetailSessionNotFound{}
 		if errors.As(err, &detailSessionNotFound) {
 			// try to create new session
-			mgr.createSessionWithBackoff()
+			mgr.createSessionWithPermissionCheck()
 			return
 		}
 
@@ -2286,7 +2318,7 @@ func (mgr *listenerManager) refreshSession() {
 			log.WithError(err).Errorf("failed to to refresh session %v", *mgr.session.ID)
 
 			// try to create new session
-			mgr.createSessionWithBackoff()
+			mgr.createSessionWithPermissionCheck()
 		}
 	}
 
@@ -2297,10 +2329,15 @@ func (mgr *listenerManager) refreshSession() {
 	}
 }
 
-func (mgr *listenerManager) createSessionWithBackoff() {
+func (mgr *listenerManager) createSessionWithBackoff() error {
 	log := pfxlog.Logger().WithField("serviceName", *mgr.service.Name)
 	latestSvc, _ := mgr.context.services.Get(*mgr.service.Name)
-	if latestSvc != nil && *latestSvc.ID != *mgr.service.ID {
+
+	if latestSvc == nil || !slices.Contains(latestSvc.Permissions, rest_model.DialBindBind) {
+		return ServiceAccessDeniedError{serviceName: *mgr.service.Name}
+	}
+
+	if *latestSvc.ID != *mgr.service.ID {
 		log.WithField("oldServiceId", *mgr.service.ID).
 			WithField("newServiceId", *latestSvc.ID).
 			Info("service id changed, service was recreated")
@@ -2311,9 +2348,11 @@ func (mgr *listenerManager) createSessionWithBackoff() {
 	if session != nil {
 		mgr.sessionRefreshed(session)
 		log.Debug("new service session created")
-	} else {
-		log.WithError(err).Errorf("failed to create bind session for service %v", mgr.service.Name)
+		return nil
 	}
+
+	log.WithError(err).Errorf("failed to create bind session for service %v", mgr.service.Name)
+	return err
 }
 
 func (mgr *listenerManager) GetCurrentSession() *rest_model.SessionDetail {
@@ -2401,3 +2440,15 @@ const (
 type ListenEventObserver interface {
 	Notify(eventType ListenEventType)
 }
+
+func IsServiceAccessDeniedError(err error) bool {
+	return errors.As(err, &ServiceAccessDeniedError{})
+}
+
+type ServiceAccessDeniedError struct {
+	serviceName string
+}
+
+func (s ServiceAccessDeniedError) Error() string {
+	return fmt.Sprintf("identity does not have permission to bind service '%v', or service does not exist", s.serviceName)
+}