Merge tag 'v1.14.3'

protocol/ssh/inbound.go

@@ -106,15 +106,17 @@ func NewInbound(ctx context.Context, router adapter.Router, logger log.ContextLo
 	return inbound, nil
 }
 
-func (h *Inbound) Start(stage adapter.StartStage) error {
+func (h *Inbound) Start(stage adapter.StartStage, scope *adapter.Scope) error {
 	if stage != adapter.StartStateStart {
 		return nil
 	}
-	return h.listener.Start()
-}
-
-func (h *Inbound) Close() error {
-	return E.Errors(h.service.Close(), h.listener.Close())
+	scope.Add(h.service.Close)
+	err := h.listener.Start()
+	if err != nil {
+		return err
+	}
+	scope.Add(h.listener.Close)
+	return nil
 }
 
 func (h *Inbound) UpdateUsers(users []option.SSHUser) {

protocol/ssh/outbound.go

@@ -32,7 +32,11 @@ func RegisterOutbound(registry *outbound.Registry) {
 	outbound.Register[option.SSHOutboundOptions](registry, C.TypeSSH, NewOutbound)
 }
 
-var _ adapter.InterfaceUpdateListener = (*Outbound)(nil)
+var (
+	_ adapter.InterfaceUpdateListener = (*Outbound)(nil)
+	_ adapter.IdleConnectionKeeper    = (*Outbound)(nil)
+	_ adapter.OutboundWithMultiplex   = (*Outbound)(nil)
+)
 
 type Outbound struct {
 	outbound.Adapter
@@ -50,6 +54,8 @@ type Outbound struct {
 	clientAccess      sync.Mutex
 	clientConn        net.Conn
 	client            *ssh.Client
+	streams           int
+	closeIdle         bool
 }
 
 func NewOutbound(ctx context.Context, router adapter.Router, logger log.ContextLogger, tag string, options option.SSHOutboundOptions) (adapter.Outbound, error) {
@@ -127,14 +133,11 @@ func randomVersion() string {
 }
 
 func (s *Outbound) connect(ctx context.Context) (client *ssh.Client, err error) {
-	if s.client != nil {
-		return s.client, nil
-	}
-
 	s.clientAccess.Lock()
 	defer s.clientAccess.Unlock()
 
 	if s.client != nil {
+		s.streams++
 		return s.client, nil
 	}
 
@@ -193,13 +196,16 @@ func (s *Outbound) connect(ctx context.Context) (client *ssh.Client, err error)
 
 	s.clientConn = conn
 	s.client = client
+	s.streams = 1
 
 	go func() {
 		client.Wait()
 		conn.Close()
 		s.clientAccess.Lock()
-		s.client = nil
-		s.clientConn = nil
+		if s.client == client {
+			s.client = nil
+			s.clientConn = nil
+		}
 		s.clientAccess.Unlock()
 	}()
 
@@ -210,8 +216,58 @@ func (s *Outbound) InterfaceUpdated(ctx context.Context) {
 	common.Close(s.clientConn)
 }
 
-func (s *Outbound) Close() error {
-	return common.Close(s.clientConn)
+func (s *Outbound) MultiplexEnabled() bool {
+	return true
+}
+
+func (s *Outbound) SetKeepIdleConnections(keep bool) {
+	s.clientAccess.Lock()
+	s.closeIdle = !keep
+	s.clientAccess.Unlock()
+	if !keep {
+		s.CloseIdleConnections()
+	}
+}
+
+func (s *Outbound) CloseIdleConnections() {
+	s.clientAccess.Lock()
+	if s.client == nil || s.streams > 0 {
+		s.clientAccess.Unlock()
+		return
+	}
+	clientConn := s.clientConn
+	s.client = nil
+	s.clientConn = nil
+	s.clientAccess.Unlock()
+	common.Close(clientConn)
+}
+
+func (s *Outbound) releaseStream(client *ssh.Client, keepSession bool) {
+	s.clientAccess.Lock()
+	if s.client != client {
+		s.clientAccess.Unlock()
+		return
+	}
+	s.streams--
+	if !s.closeIdle || keepSession || s.streams > 0 {
+		s.clientAccess.Unlock()
+		return
+	}
+	clientConn := s.clientConn
+	s.client = nil
+	s.clientConn = nil
+	s.clientAccess.Unlock()
+	common.Close(clientConn)
+}
+
+func (s *Outbound) Start(stage adapter.StartStage, scope *adapter.Scope) error {
+	if stage != adapter.StartStateInitialize {
+		return nil
+	}
+	scope.Add(func() error {
+		return common.Close(s.clientConn)
+	})
+	return nil
 }
 
 func (s *Outbound) DialContext(ctx context.Context, network string, destination M.Socksaddr) (net.Conn, error) {
@@ -221,9 +277,11 @@ func (s *Outbound) DialContext(ctx context.Context, network string, destination
 	}
 	conn, err := client.Dial(network, destination.String())
 	if err != nil {
+		s.releaseStream(client, false)
 		return nil, err
 	}
-	return &chanConnWrapper{Conn: conn}, nil
+	keepSession := adapter.KeepSessionFromContext(ctx)
+	return &chanConnWrapper{Conn: conn, onClose: func() { s.releaseStream(client, keepSession) }}, nil
 }
 
 func (s *Outbound) ListenPacket(ctx context.Context, destination M.Socksaddr) (net.PacketConn, error) {
@@ -232,6 +290,14 @@ func (s *Outbound) ListenPacket(ctx context.Context, destination M.Socksaddr) (n
 
 type chanConnWrapper struct {
 	net.Conn
+	closeOnce sync.Once
+	onClose   func()
+}
+
+func (c *chanConnWrapper) Close() error {
+	err := c.Conn.Close()
+	c.closeOnce.Do(c.onClose)
+	return err
 }
 
 func (c *chanConnWrapper) SetDeadline(t time.Time) error {