Merge pull request #152 from jumpserver/dev

Dev

pkg/handler/sftp.go

@@ -34,7 +34,7 @@ func SftpHandler(sess ssh.Session) {
 	logger.Infof("SFTP request %s: Handler start", reqID)
 	req := sftp.NewRequestServer(sess, handlers)
 	if err := req.Serve(); err == io.EOF {
-		logger.Debug("SFTP request %s: Exited session.", reqID)
+		logger.Debugf("SFTP request %s: Exited session.", reqID)
 	} else if err != nil {
 		logger.Errorf("SFTP request %s: Server completed with error %s", reqID, err)
 	}
@@ -82,7 +82,7 @@ func (fs *sftpHandler) Filecmd(r *sftp.Request) (err error) {
 	case "Setstat":
 		return
 	case "Rename":
-		logger.Debug("%s=>%s", r.Filepath, r.Target)
+		logger.Debugf("%s=>%s", r.Filepath, r.Target)
 		return fs.Rename(r.Filepath, r.Target)
 	case "Rmdir":
 		err = fs.RemoveDirectory(r.Filepath)
@@ -91,7 +91,7 @@ func (fs *sftpHandler) Filecmd(r *sftp.Request) (err error) {
 	case "Mkdir":
 		err = fs.MkdirAll(r.Filepath)
 	case "Symlink":
-		logger.Debug("%s=>%s", r.Filepath, r.Target)
+		logger.Debugf("%s=>%s", r.Filepath, r.Target)
 		err = fs.Symlink(r.Filepath, r.Target)
 	default:
 		return

pkg/handler/wrappersession.go

@@ -61,6 +61,11 @@ func (w *WrapperSession) Read(p []byte) (int, error) {
 }
 
 func (w *WrapperSession) Close() error {
+	select {
+	case <-w.closed:
+		return nil
+	default:
+	}
 	err := w.inWriter.Close()
 	w.initReadPip()
 	return err

pkg/httpd/websshws.go

@@ -56,8 +56,10 @@ func OnNamespaceConnected(c *neffos.NSConn, msg neffos.Message) error {
 	remoteIP = strings.Split(remoteAddr, ",")[0]
 	logger.Infof("Accepted %s connect websocket from %s", user.Username, remoteIP)
 	go func() {
+		tick := time.NewTicker(30 * time.Second)
+		defer tick.Stop()
 		for {
-			<-time.After(30 * time.Second)
+			<-tick.C
 			if c.Conn.IsClosed() {
 				logger.Infof("User %s from %s websocket connect closed", user.Username, remoteIP)
 				return

pkg/proxy/parsercmd.go

@@ -33,14 +33,29 @@ type CmdParser struct {
 }
 
 func (cp *CmdParser) WriteData(p []byte) (int, error) {
+	select {
+	case <-cp.closed:
+		return 0, io.EOF
+	default:
+	}
 	return cp.writer.Write(p)
 }
 
 func (cp *CmdParser) Write(p []byte) (int, error) {
+	select {
+	case <-cp.closed:
+		return 0, io.EOF
+	default:
+	}
 	return len(p), nil
 }
 
 func (cp *CmdParser) Read(p []byte) (int, error) {
+	select {
+	case <-cp.closed:
+		return 0, io.EOF
+	default:
+	}
 	return cp.reader.Read(p)
 }
 
@@ -51,6 +66,7 @@ func (cp *CmdParser) Close() error {
 	default:
 		close(cp.closed)
 	}
+	_ = cp.reader.Close()
 	return cp.writer.Close()
 }
 
@@ -94,7 +110,11 @@ func (cp *CmdParser) parsePS1(s string) string {
 
 // Parse 解析命令或输出
 func (cp *CmdParser) Parse() string {
-	cp.writer.Write([]byte("\r"))
+	select {
+	case <-cp.closed:
+	default:
+		cp.writer.Write([]byte("\r"))
+	}
 	cp.lock.Lock()
 	defer cp.lock.Unlock()
 	output := strings.TrimSpace(strings.Join(cp.currentLines, "\r\n"))
@@ -102,4 +122,4 @@ func (cp *CmdParser) Parse() string {
 	cp.currentLines = make([]string, 0)
 	cp.currentLength = 0
 	return output
-}
\ No newline at end of file
+}

pkg/proxy/recorder.go

@@ -60,6 +60,8 @@ func (c *CommandRecorder) record() {
 	maxRetry := 0
 	logger.Infof("Session %s: Command recorder start", c.sessionID)
 	defer logger.Infof("Session %s: Command recorder close", c.sessionID)
+	tick := time.NewTicker(time.Second * 10)
+	defer tick.Stop()
 	for {
 		select {
 		case <-c.closed:
@@ -74,7 +76,7 @@ func (c *CommandRecorder) record() {
 			if len(cmdList) < 5 {
 				continue
 			}
-		case <-time.After(time.Second * 5):
+		case <-tick.C:
 			if len(cmdList) == 0 {
 				continue
 			}

pkg/proxy/switch.go

@@ -152,10 +152,19 @@ func (s *SwitchSession) Bridge(userConn UserConnection, srvConn srvconn.ServerCo
 	go LoopRead(srvConn, srvInChan)
 
 	winCh := userConn.WinCh()
+	maxIdleTime := s.MaxIdleTime * time.Minute
+	lastActiveTime := time.Now()
+	tick := time.NewTicker(30 * time.Second)
+	defer tick.Stop()
 	for {
 		select {
 		// 检测是否超过最大空闲时间
-		case <-time.After(s.MaxIdleTime * time.Minute):
+		case <-tick.C:
+			now := time.Now()
+			outTime := lastActiveTime.Add(maxIdleTime)
+			if !now.After(outTime) {
+				continue
+			}
 			msg := fmt.Sprintf(i18n.T("Connect idle more than %d minutes, disconnect"), s.MaxIdleTime)
 			logger.Debugf("Session idle more than %d minutes, disconnect: %s", s.MaxIdleTime, s.ID)
 			msg = utils.WrapperWarn(msg)
@@ -190,6 +199,7 @@ func (s *SwitchSession) Bridge(userConn UserConnection, srvConn srvconn.ServerCo
 			}
 			_, err = srvConn.Write(p)
 		}
+		lastActiveTime = time.Now()
 	}
 }
 

pkg/srvconn/sftpconn.go

@@ -548,17 +548,12 @@ func (u *UserSftp) LoopPushFTPLog() {
 	dataChan := make(chan *model.FTPLog)
 	go u.SendFTPLog(dataChan)
 	defer close(dataChan)
-	var timeoutSecond time.Duration
-	for {
-		switch len(ftpLogList) {
-		case 0:
-			timeoutSecond = time.Second * 60
-		default:
-			timeoutSecond = time.Second * 10
-		}
 
+	tick := time.NewTicker(time.Second * 10)
+	defer tick.Stop()
+	for {
 		select {
-		case <-time.After(timeoutSecond):
+		case <-tick.C:
 		case logData, ok := <-u.LogChan:
 			if !ok {
 				return