Merge pull request #39 from jumpserver/dev

Dev

Merge pull request #39 from jumpserver/dev
Dev
58db16cb · 老广 · GitHub · 1d0d0d6a · 46744e8f · 58db16cb
Unverified Commit 58db16cb authored Jul 18, 2019 by 老广 Committed by GitHub Jul 18, 2019
30 changed files
--- a/.gitignore
+++ b/.gitignore
@@ -16,4 +16,7 @@ log/
 vendor/
 config.yml
 host_key
-cmd/coco
+cmd/koko
+cmd/logs
+cmd/kokodir
+build
--- a/cmd/static/js/neffos.min.js
+++ b/cmd/static/js/neffos.min.js
--- a/cmd/templates/elfinder/file_manager.html
+++ b/cmd/templates/elfinder/file_manager.html
@@ -2,21 +2,39 @@
 <body style="margin: 0">
    <script type="text/javascript" src="/static/js/jquery-2.1.1.js"></script>
    <script type="text/javascript" src="/static/js/jquery-ui-1.10.4.min.js"></script>
-    <script type="text/javascript" src="/static/js/socket.io.js"></script>
+    <script type="text/javascript" src="/static/js/neffos.min.js"></script>
    <script type="text/javascript" src="/static/plugins/elfinder/elfinder.full.js"></script>
    <script type="text/javascript" src="/static/plugins/elfinder/i18n/elfinder.pl.js"></script>
    <link rel="stylesheet" type="text/css" media="screen" href="/static/plugins/elfinder/css/elfinder.min.css">
    <link rel="stylesheet" type="text/css" media="screen" href="/static/plugins/elfinder/css/theme-gray.css">
    <script type="text/javascript" charset="utf-8">
-        var socket = io.connect('/elfinder');
+        var scheme = document.location.protocol == "https:" ? "wss" : "ws";
-        socket.on('connect', function () {
+        var port = document.location.port ? ":" + document.location.port : "";
-            console.log("Connect websocket done")
+        var wsURL = scheme + "://" + document.location.hostname + port + "/socket.io/";
-        });
+        dial(wsURL, {
-        socket.on('data', function (msg) {
+            "elfinder": {
-            var sid = msg.sid;
+                _OnNamespaceConnected: function (nsConn, msg) {
-            init_elfinder(sid);
+                    console.log("Connect websocket done")
+                },
+                data: function (nsConn, msg) {
+                    var data = msg.unmarshal();
+                    var sid = data.sid;
+                    initElfinder(sid);
+                }
+            }
+        }).then(conn => {
+            conn.connect("elfinder");
        });
-        function init_elfinder(sid) {
+        // var socket = io.connect('/elfinder');
+        // socket.on('connect', function () {
+        //     console.log("Connect websocket done")
+        // });
+        // socket.on('data', function (msg) {
+        //     var sid = msg.sid;
+        //     initElfinder(sid);
+        // });
+        function initElfinder(sid) {
            var elf;
            var opts = {
 	    		uiOptions : {

--- a/go.mod
+++ b/go.mod
@@ -3,40 +3,41 @@ module github.com/jumpserver/koko
 go 1.12
 require (
-	github.com/Azure/azure-pipeline-go v0.1.9
+	github.com/Azure/azure-pipeline-go v0.1.9 // indirect
 	github.com/Azure/azure-storage-blob-go v0.6.0
-	github.com/LeeEirc/elfinder v0.0.0-20190604073433-f4f8357e9220
+	github.com/BurntSushi/toml v0.3.1 // indirect
+	github.com/LeeEirc/elfinder v0.0.0-20190718023636-5679c8bdb7bf
 	github.com/aliyun/aliyun-oss-go-sdk v1.9.8
-	github.com/anmitsu/go-shlex v0.0.0-20161002113705-648efa622239
+	github.com/anmitsu/go-shlex v0.0.0-20161002113705-648efa622239 // indirect
 	github.com/aws/aws-sdk-go v1.19.46
+	github.com/baiyubin/aliyun-sts-go-sdk v0.0.0-20180326062324-cfa1a18b161f // indirect
 	github.com/elastic/go-elasticsearch v0.0.0
+	github.com/flynn/go-shlex v0.0.0-20150515145356-3f9db97f8568 // indirect
 	github.com/gliderlabs/ssh v0.2.3-0.20190711180243-866d0ddf7991
-	github.com/go-playground/form v3.1.4+incompatible
+	github.com/go-playground/form v3.1.4+incompatible // indirect
-	github.com/googollee/go-socket.io v1.4.2-0.20190317095603-ed07a7212e28
 	github.com/gorilla/mux v1.7.2
-	github.com/gorilla/websocket v1.4.0
 	github.com/jarcoal/httpmock v1.0.4
-	github.com/jmespath/go-jmespath v0.0.0-20180206201540-c2b33e8439af
+	github.com/kataras/neffos v0.0.7
-	github.com/konsorten/go-windows-terminal-sequences v1.0.2
+	github.com/konsorten/go-windows-terminal-sequences v1.0.2 // indirect
-	github.com/kr/fs v0.1.0
+	github.com/kr/fs v0.1.0 // indirect
-	github.com/kr/pty v1.1.4
 	github.com/leonelquinteros/gotext v1.4.0
-	github.com/mattn/go-runewidth v0.0.4
+	github.com/mattn/go-runewidth v0.0.4 // indirect
 	github.com/olekukonko/tablewriter v0.0.1
 	github.com/pkg/errors v0.8.1
 	github.com/pkg/sftp v1.10.0
 	github.com/satori/go.uuid v1.2.0
 	github.com/sirupsen/logrus v1.4.2
+	github.com/stretchr/testify v1.3.0 // indirect
 	github.com/xlab/treeprint v0.0.0-20181112141820-a009c3971eca
 	golang.org/x/crypto v0.0.0-20190701094942-4def268fd1a4
+	golang.org/x/text v0.3.1-0.20180807135948-17ff2d5776d2 // indirect
 	golang.org/x/time v0.0.0-20190308202827-9d24e82272b4 // indirect
+	gopkg.in/go-playground/assert.v1 v1.2.1 // indirect
 	gopkg.in/natefinch/lumberjack.v2 v2.0.0-20170531160350-a96e63847dc3
 	gopkg.in/yaml.v2 v2.2.2
 )
 replace (
 	github.com/gliderlabs/ssh v0.2.3-0.20190711180243-866d0ddf7991 => github.com/ibuler/ssh v0.1.6-0.20190509065047-1c00c8e8b607
-	github.com/googollee/go-engine.io v1.4.1 => github.com/ibuler/go-engine.io v1.4.2-0.20190529094538-7786d3a289b9
-	github.com/googollee/go-socket.io v1.4.2-0.20190317095603-ed07a7212e28 => github.com/LeeEirc/go-socket.io v1.4.2-0.20190610105739-e344e8b5a55a
 	golang.org/x/crypto v0.0.0-20190701094942-4def268fd1a4 => github.com/ibuler/crypto v0.0.0-20190509101200-a7099eef26a7
 )
--- a/go.sum
+++ b/go.sum
--- a/pkg/handler/session.go
+++ b/pkg/handler/session.go
@@ -167,7 +167,7 @@ func (h *interactiveHandler) Dispatch(ctx cctx.Context) {
 		if err != nil {
 			if err != io.EOF {
-				logger.Debug("user disconnected")
+				logger.Debug("User disconnected")
 			} else {
 				logger.Error("Read from user err: ", err)
 			}
@@ -323,7 +323,7 @@ func (h *interactiveHandler) refreshAssetsAndNodesData() {
 }
 func (h *interactiveHandler) loadUserAssets(cachePolicy string) {
-	assets := service.GetUserAssets(h.user.ID, cachePolicy)
+	assets := service.GetUserAssets(h.user.ID, cachePolicy, "")
 	userAssetsCached.SetValue(h.user.ID, assets)
 	h.mu.Lock()
 	h.assets = assets

--- a/pkg/handler/sftp.go
+++ b/pkg/handler/sftp.go
--- a/pkg/httpd/client.go
+++ b/pkg/httpd/client.go
 package httpd
 import (
+	"encoding/json"
 	"io"
 	"sync"
 	"github.com/gliderlabs/ssh"
-	socketio "github.com/googollee/go-socket.io"
+	"github.com/kataras/neffos"
+	"github.com/jumpserver/koko/pkg/logger"
 	"github.com/jumpserver/koko/pkg/model"
 )
 type Client struct {
 	Uuid      string
 	Cid       string
@@ -18,10 +21,10 @@ type Client struct {
 	WinChan   chan ssh.Window
 	UserRead  io.Reader
 	UserWrite io.WriteCloser
-	Conn      socketio.Conn
+	Conn *neffos.NSConn
 	Closed    bool
 	pty       ssh.Pty
-	lock      *sync.RWMutex
+	mu        *sync.RWMutex
 }
 func (c *Client) WinCh() <-chan ssh.Window {
@@ -41,14 +44,23 @@ func (c *Client) Read(p []byte) (n int, err error) {
 }
 func (c *Client) Write(p []byte) (n int, err error) {
-	c.lock.RLock()
+	c.mu.RLock()
-	defer c.lock.RUnlock()
+	defer c.mu.RUnlock()
 	if c.Closed {
 		return
 	}
-	data := DataMsg{Data: string(p), Room: c.Uuid}
+	data := DataMsg{Data: string(p)}
+	msg, err := json.Marshal(data)
+	if err != nil {
+		return
+	}
 	n = len(p)
-	c.Conn.Emit("data", data)
+	room := c.Conn.Room(c.Uuid)
+	if room == nil {
+		logger.Error("room not found: ", c.Uuid)
+		return
+	}
+	room.Emit("data", msg)
 	return
 }
@@ -57,11 +69,17 @@ func (c *Client) Pty() ssh.Pty {
 }
 func (c *Client) Close() (err error) {
-	c.lock.Lock()
+	c.mu.Lock()
-	defer c.lock.Unlock()
+	defer c.mu.Unlock()
 	if c.Closed {
 		return
 	}
 	c.Closed = true
 	return c.UserWrite.Close()
 }
+func (c *Client) SetWinSize(size ssh.Window) {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+	c.WinChan <- size
+}
--- a/pkg/httpd/clients.go
+++ b/pkg/httpd/clients.go
+package httpd
+import (
+	"fmt"
+	"sync"
+	"github.com/jumpserver/koko/pkg/logger"
+)
+var conns = &Connections{container: make(map[string][]string), mu: new(sync.RWMutex)}
+var clients = &Clients{container: make(map[string]*Client), mu: new(sync.RWMutex)}
+type Clients struct {
+	container map[string]*Client
+	mu        *sync.RWMutex
+}
+func (c *Clients) GetClient(cID string) (client *Client) {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+	client = c.container[cID]
+	return
+}
+func (c *Clients) DeleteClient(cID string) {
+	c.mu.RLock()
+	client, ok := c.container[cID]
+	c.mu.RUnlock()
+	if !ok {
+		return
+	}
+	_ = client.Close()
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	delete(c.container, cID)
+	logger.Debug("Remain clients count: ", len(c.container))
+}
+func (c *Clients) AddClient(cID string, conn *Client) {
+	fmt.Println("Add Client id: ", cID)
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	c.container[cID] = conn
+	logger.Debug("Now clients count: ", len(c.container))
+}
+type Connections struct {
+	container map[string][]string
+	mu        *sync.RWMutex
+}
+func (c *Connections) AddClient(cID, clientID string) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	clients, ok := c.container[cID]
+	if ok {
+		clients = append(clients, clientID)
+	} else {
+		clients = []string{clientID}
+	}
+	c.container[cID] = clients
+}
+func (c *Connections) GetClients(cID string) (clients []string) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	return c.container[cID]
+}
+func (c *Connections) DeleteClients(cID string) {
+	if clientIDs := c.GetClients(cID); clientIDs != nil{
+		for _, clientID := range clientIDs {
+			clients.DeleteClient(clientID)
+		}
+	}
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	delete(c.container, cID)
+}
--- a/pkg/httpd/connection.go
+++ b/pkg/httpd/connection.go
-package httpd
-import (
-	"sync"
-	"github.com/gliderlabs/ssh"
-	socketio "github.com/googollee/go-socket.io"
-	"github.com/jumpserver/koko/pkg/model"
-)
-var conns = &connections{container: make(map[string]*WebConn), mu: new(sync.RWMutex)}
-type connections struct {
-	container map[string]*WebConn
-	mu        *sync.RWMutex
-}
-func (c *connections) GetWebConn(conID string) (conn *WebConn) {
-	c.mu.RLock()
-	defer c.mu.RUnlock()
-	conn = c.container[conID]
-	return
-}
-func (c *connections) DeleteWebConn(conID string) {
-	c.mu.RLock()
-	webC, ok := c.container[conID]
-	c.mu.RUnlock()
-	if !ok {
-		return
-	}
-	webC.Close()
-	c.mu.Lock()
-	defer c.mu.Unlock()
-	delete(c.container, conID)
-}
-func (c *connections) AddWebConn(conID string, conn *WebConn) {
-	c.mu.Lock()
-	defer c.mu.Unlock()
-	c.container[conID] = conn
-}
-func newWebConn(id string, sock socketio.Conn, addr string, user *model.User) *WebConn {
-	conn := &WebConn{Cid: id, Sock: sock, Addr: addr, User: user, mu: new(sync.RWMutex), Clients: make(map[string]*Client)}
-	return conn
-}
-type WebConn struct {
-	Cid     string
-	Sock    socketio.Conn
-	Addr    string
-	User    *model.User
-	Clients map[string]*Client
-	mu      *sync.RWMutex
-}
-func (w *WebConn) GetClient(clientID string) (conn *Client) {
-	w.mu.RLock()
-	defer w.mu.RUnlock()
-	return w.Clients[clientID]
-}
-func (w *WebConn) DeleteClient(clientID string) {
-	w.mu.Lock()
-	defer w.mu.Unlock()
-	delete(w.Clients, clientID)
-}
-func (w *WebConn) AddClient(clientID string, conn *Client) {
-	w.mu.Lock()
-	defer w.mu.Unlock()
-	w.Clients[clientID] = conn
-}
-func (w *WebConn) GetAllClients() (clients []string) {
-	clients = make([]string, 0)
-	w.mu.RLock()
-	defer w.mu.RUnlock()
-	for k := range w.Clients {
-		clients = append(clients, k)
-	}
-	return clients
-}
-func (w *WebConn) SetWinSize(winSize ssh.Window) {
-	w.mu.RLock()
-	defer w.mu.RUnlock()
-	for _, client := range w.Clients {
-		client.WinChan <- winSize
-	}
-}
-func (w *WebConn) Close() {
-	w.mu.Lock()
-	defer w.mu.Unlock()
-	clientsCopy := make(map[string]*Client)
-	for k, v := range w.Clients {
-		clientsCopy[k] = v
-	}
-	for k, client := range clientsCopy {
-		_ = client.Close()
-		delete(w.Clients, k)
-	}
-}
--- a/pkg/httpd/context.go
+++ b/pkg/httpd/context.go
@@ -4,6 +4,6 @@ import "github.com/jumpserver/koko/pkg/model"
 type WebContext struct {
 	User       *model.User
-	Connection *WebConn
+	Connection *Client
 	Client     *Client
 }
--- a/pkg/httpd/data.go
+++ b/pkg/httpd/data.go
@@ -20,7 +20,6 @@ type TokenMsg struct {
 type DataMsg struct {
 	Data string `json:"data"`
-	Room string `json:"room"`
 }
 type RoomMsg struct {

--- a/pkg/httpd/elfhandler.go
+++ b/pkg/httpd/elfhandler.go
@@ -8,7 +8,6 @@ import (
 	"strings"
 	"github.com/LeeEirc/elfinder"
-	socketio "github.com/googollee/go-socket.io"
 	"github.com/gorilla/mux"
 	"github.com/jumpserver/koko/pkg/cctx"
@@ -49,17 +48,7 @@ func AuthDecorator(handler http.HandlerFunc) http.HandlerFunc {
 	}
 }
-func OnELFinderConnect(s socketio.Conn) error {
-	data := EmitSidMsg{Sid: s.ID()}
-	s.Emit("data", data)
-	return nil
-}
-func OnELFinderDisconnect(s socketio.Conn, msg string) {
-	logger.Debug("disconnect: ", s.ID())
-	logger.Debug("disconnect msg ", msg)
-	removeUserVolume(s.ID())
-}
 func sftpHostFinder(wr http.ResponseWriter, req *http.Request) {
 	vars := mux.Vars(req)
@@ -74,6 +63,8 @@ func sftpFinder(wr http.ResponseWriter, req *http.Request) {
 }
 func sftpHostConnectorView(wr http.ResponseWriter, req *http.Request) {
+	vars := mux.Vars(req)
+	hostID := vars["host"]
 	user := req.Context().Value(cctx.ContextKeyUser).(*model.User)
 	remoteIP := req.Context().Value(cctx.ContextKeyRemoteAddr).(string)
 	switch req.Method {
@@ -92,10 +83,15 @@ func sftpHostConnectorView(wr http.ResponseWriter, req *http.Request) {
 	sid := req.Form.Get("sid")
 	userV, ok := GetUserVolume(sid)
 	if !ok {
-		userV = NewUserVolume(user, remoteIP)
+		switch strings.TrimSpace(hostID) {
+		case "_":
+			userV = NewUserVolume(user, remoteIP,"")
+		default:
+			userV = NewUserVolume(user, remoteIP, hostID)
+		}
 		addUserVolume(sid, userV)
 	}
-	logger.Debugf("sid: %s", sid)
+	logger.Debugf("Elfinder connector sid: %s", sid)
-	con := elfinder.NewElFinderConnector([]elfinder.Volume{userV})
+	conn := elfinder.NewElFinderConnector([]elfinder.Volume{userV})
-	con.ServeHTTP(wr, req)
+	conn.ServeHTTP(wr, req)
 }
--- a/pkg/httpd/elfws.go
+++ b/pkg/httpd/elfws.go
+package httpd
+import (
+	"github.com/kataras/neffos"
+	"github.com/jumpserver/koko/pkg/logger"
+)
+func OnELFinderConnect(c *neffos.NSConn, msg neffos.Message) error {
+	data := EmitSidMsg{Sid: c.Conn.ID()}
+	c.Emit("data", neffos.Marshal(data))
+	return nil
+}
+func OnELFinderDisconnect(c *neffos.NSConn, msg neffos.Message) (error) {
+	logger.Debug("disconnect: ", c.Conn.ID())
+	removeUserVolume(c.Conn.ID())
+	return nil
+}
--- a/pkg/httpd/server.go
+++ b/pkg/httpd/server.go
@@ -5,8 +5,9 @@ import (
 	"net/http"
 	"path/filepath"
-	"github.com/googollee/go-socket.io"
 	"github.com/gorilla/mux"
+	"github.com/kataras/neffos"
+	"github.com/kataras/neffos/gorilla"
 	"github.com/jumpserver/koko/pkg/config"
 	"github.com/jumpserver/koko/pkg/logger"
@@ -16,35 +17,43 @@ var (
 	httpServer *http.Server
 )
+var wsEvents = neffos.Namespaces{
+	"ssh": neffos.Events{
+		neffos.OnNamespaceConnected: OnNamespaceConnected,
+		neffos.OnNamespaceDisconnect: OnNamespaceDisconnect,
+		neffos.OnRoomJoined: func(c *neffos.NSConn, msg neffos.Message) error {
+			return nil
+		},
+		neffos.OnRoomLeft: func(c *neffos.NSConn, msg neffos.Message) error {
+			return nil
+		},
+		"data": OnDataHandler,
+		"resize": OnResizeHandler,
+		"host": OnHostHandler,
+		"logout": OnLogoutHandler,
+		"token": OnTokenHandler,
+	},
+	"elfinder": neffos.Events{
+		neffos.OnNamespaceConnected: OnELFinderConnect,
+		neffos.OnNamespaceDisconnect: OnELFinderDisconnect,
+	},
+}
 func StartHTTPServer() {
 	conf := config.GetConf()
-	server, err := socketio.NewServer(nil)
+	sshWs := neffos.New(gorilla.DefaultUpgrader, wsEvents)
-	if err != nil {
+	sshWs.IDGenerator = func(w http.ResponseWriter, r *http.Request) string {
-		logger.Fatal(err)
+		return neffos.DefaultIDGenerator(w, r)
+	}
+	sshWs.OnUpgradeError = func(err error) {
 	}
-	server.OnConnect("/ssh", OnConnectHandler)
-	server.OnDisconnect("/ssh", OnDisconnect)
-	server.OnError("/ssh", OnErrorHandler)
-	server.OnEvent("/ssh", "host", OnHostHandler)
-	server.OnEvent("/ssh", "token", OnTokenHandler)
-	server.OnEvent("/ssh", "data", OnDataHandler)
-	server.OnEvent("/ssh", "resize", OnResizeHandler)
-	server.OnEvent("/ssh", "logout", OnLogoutHandler)
-	server.OnConnect("/elfinder", OnELFinderConnect)
-	server.OnDisconnect("/elfinder", OnELFinderDisconnect)
-	server.OnError("/elfiner", OnErrorHandler)
-	server.OnDisconnect("", SocketDisconnect)
-	server.OnError("", OnErrorHandler)
-	go server.Serve()
-	defer server.Close()
 	router := mux.NewRouter()
 	fs := http.FileServer(http.Dir(filepath.Join(conf.RootPath, "static")))
 	router.PathPrefix("/static/").Handler(http.StripPrefix("/static/", fs))
-	router.Handle("/socket.io/", server)
+	router.Handle("/socket.io/", sshWs)
 	router.HandleFunc("/coco/elfinder/sftp/{host}/", AuthDecorator(sftpHostFinder))
 	router.HandleFunc("/coco/elfinder/sftp/", AuthDecorator(sftpFinder))
 	router.HandleFunc("/coco/elfinder/sftp/connector/{host}/",
@@ -60,8 +69,4 @@ func StopHTTPServer() {
 	_ = httpServer.Close()
 }
-func SocketDisconnect(s socketio.Conn, msg string) {
-	removeUserVolume(s.ID())
-	conns.DeleteWebConn(s.ID())
-	logger.Debug("clean disconnect")
-}
--- a/pkg/httpd/sftpvolume.go
+++ b/pkg/httpd/sftpvolume.go
--- a/pkg/httpd/volumemanager.go
+++ b/pkg/httpd/volumemanager.go
@@ -2,12 +2,19 @@ package httpd
 import (
 	"sync"
+	"github.com/LeeEirc/elfinder"
 )
-var userVolumes = make(map[string]*UserVolume)
+type VolumeCloser interface {
+	elfinder.Volume
+	Close()
+}
+var userVolumes = make(map[string]VolumeCloser)
 var volumeLock = new(sync.RWMutex)
-func addUserVolume(sid string, v *UserVolume) {
+func addUserVolume(sid string, v VolumeCloser) {
 	volumeLock.Lock()
 	defer volumeLock.Unlock()
 	userVolumes[sid] = v
@@ -27,7 +34,7 @@ func removeUserVolume(sid string) {
 }
-func GetUserVolume(sid string) (*UserVolume, bool) {
+func GetUserVolume(sid string) (VolumeCloser, bool) {
 	volumeLock.RLock()
 	defer volumeLock.RUnlock()
 	v, ok := userVolumes[sid]

--- a/pkg/httpd/handler.go
+++ b/pkg/httpd/handler.go
 package httpd
 import (
+	"encoding/json"
 	"errors"
 	"fmt"
+	"github.com/jumpserver/koko/pkg/model"
 	"io"
 	"net"
 	"strings"
 	"sync"
 	"github.com/gliderlabs/ssh"
-	socketio "github.com/googollee/go-socket.io"
+	"github.com/kataras/neffos"
-	uuid "github.com/satori/go.uuid"
+	"github.com/satori/go.uuid"
 	"github.com/jumpserver/koko/pkg/logger"
 	"github.com/jumpserver/koko/pkg/proxy"
 	"github.com/jumpserver/koko/pkg/service"
 )
 // OnConnectHandler 当websocket连接后触发
-func OnConnectHandler(s socketio.Conn) error {
+func OnNamespaceConnected(c *neffos.NSConn, msg neffos.Message) error {
 	// 首次连接 1.获取当前用户的信息
-	logger.Debug("Web terminal on connect event trigger")
+	cc := c.Conn
-	cookies := strings.Split(s.RemoteHeader().Get("Cookie"), ";")
+	if cc.WasReconnected() {
+		logger.Debugf("Web terminal redirected, with tries: %d", cc.ID(), cc.ReconnectTries)
+	} else {
+		logger.Debug("Web terminal on connect event trigger")
+	}
+	request := cc.Socket().Request()
+	header := request.Header
+	cookies := strings.Split(header.Get("Cookie"), ";")
 	var csrfToken, sessionID, remoteIP string
 	for _, line := range cookies {
 		if strings.Contains(line, "csrftoken") {
@@ -37,17 +51,21 @@ func OnConnectHandler(s socketio.Conn) error {
 		logger.Error(msg)
 		return errors.New(strings.ToLower(msg))
 	}
-	remoteAddr := s.RemoteHeader().Get("X-Forwarded-For")
+	cc.Set("currentUser", user)
+	remoteAddr := header.Get("X-Forwarded-For")
 	if remoteAddr == "" {
-		remoteIP = s.RemoteAddr().String()
+		remoteAddr = request.RemoteAddr
-	} else {
-		remoteIP = strings.Split(remoteAddr, ",")[0]
 	}
+	remoteIP = strings.Split(remoteAddr, ",")[0]
 	logger.Infof("Accepted %s connect websocket from %s", user.Username, remoteIP)
-	conn := newWebConn(s.ID(), s, remoteIP, user)
+	return nil
-	ctx := WebContext{User: user, Connection: conn}
+}
-	s.SetContext(ctx)
-	conns.AddWebConn(s.ID(), conn)
+// OnDisconnect websocket断开后触发
+func OnNamespaceDisconnect(c *neffos.NSConn, msg neffos.Message) (err error){
+	logger.Debug("On disconnect event trigger")
+	conns.DeleteClients(c.Conn.ID())
 	return nil
 }
@@ -57,8 +75,15 @@ func OnErrorHandler(e error) {
 }
 // OnHostHandler 当用户连接Host时触发
-func OnHostHandler(s socketio.Conn, message HostMsg) {
+func OnHostHandler(c *neffos.NSConn, msg neffos.Message) (err error) {
 	logger.Debug("Web terminal on host event trigger")
+	cc := c.Conn
+	var message HostMsg
+	err = msg.Unmarshal(&message)
+	if err != nil {
+		return
+	}
+	fmt.Println("Host msg: ", message)
 	win := ssh.Window{Height: 24, Width: 80}
 	assetID := message.Uuid
 	systemUserID := message.UserID
@@ -70,141 +95,151 @@ func OnHostHandler(s socketio.Conn, message HostMsg) {
 	if height != 0 {
 		win.Height = height
 	}
-	clientID := uuid.NewV4().String()
+	roomID := uuid.NewV4().String()
-	emitMsg := RoomMsg{clientID, secret}
+	emitMsg := RoomMsg{roomID, secret}
-	s.Emit("room", emitMsg)
+	joinRoomMsg, _ := json.Marshal(emitMsg)
+	c.Emit("room", joinRoomMsg)
+	if err != nil {
+		logger.Debug("Join room error occur: ", err)
+		return
+	}
 	asset := service.GetAsset(assetID)
 	systemUser := service.GetSystemUser(systemUserID)
 	if asset.ID == "" || systemUser.ID == "" {
+		logger.Debug("No asset id or system user id found, exit")
 		return
 	}
 	logger.Debug("Web terminal want to connect host: ", asset.Hostname)
+	currentUser, ok := cc.Get("currentUser").(*model.User)
+	if !ok {
+		return errors.New("not found current user")
+	}
-	ctx := s.Context().(WebContext)
 	userR, userW := io.Pipe()
-	conn := conns.GetWebConn(s.ID())
+	addr, _, _ := net.SplitHostPort(cc.Socket().Request().RemoteAddr)
-	addr, _, _ := net.SplitHostPort(s.RemoteAddr().String())
 	client := &Client{
-		Uuid: clientID, Cid: conn.Cid, user: conn.User, addr: addr,
+		Uuid: roomID, user: currentUser, addr: addr,
-		WinChan: make(chan ssh.Window, 100), Conn: s,
+		WinChan: make(chan ssh.Window, 100), Conn: c,
-		UserRead: userR, UserWrite: userW, lock: new(sync.RWMutex),
+		UserRead: userR, UserWrite: userW, mu: new(sync.RWMutex),
 		pty: ssh.Pty{Term: "xterm", Window: win},
 	}
+	user := cc.Get("currentUser").(*model.User)
 	client.WinChan <- win
-	conn.AddClient(clientID, client)
+	clients.AddClient(roomID, client)
+	conns.AddClient(cc.ID(), roomID)
 	proxySrv := proxy.ProxyServer{
-		UserConn: client, User: ctx.User,
+		UserConn: client, User: user,
 		Asset: &asset, SystemUser: &systemUser,
 	}
 	go func() {
 		defer logger.Debug("web proxy end")
+		logger.Debug("Start proxy")
 		proxySrv.Proxy()
-		s.Emit("logout", RoomMsg{Room: clientID})
+		logoutMsg, _ := json.Marshal(RoomMsg{Room: roomID})
+		c.Emit("logout", logoutMsg)
+		clients.DeleteClient(roomID)
 	}()
+	return nil
 }
 // OnTokenHandler 当使用token连接时触发
-func OnTokenHandler(s socketio.Conn, message TokenMsg) {
+func OnTokenHandler(c *neffos.NSConn, msg neffos.Message) (err error) {
 	logger.Debug("Web terminal on token event trigger")
-	win := ssh.Window{Height: 24, Width: 80}
+	cc := c.Conn
+	var message TokenMsg
+	err = msg.Unmarshal(&message)
+	if err != nil {
+		return
+	}
 	token := message.Token
 	secret := message.Secret
-	width, height := message.Size[0], message.Size[1]
-	if width != 0 {
-		win.Width = width
-	}
-	if height != 0 {
-		win.Height = height
-	}
 	clientID := uuid.NewV4().String()
-	emitMs := RoomMsg{clientID, secret}
+	roomMsg := RoomMsg{clientID, secret}
-	s.Emit("room", emitMs)
+	c.Emit("room", neffos.Marshal(roomMsg))
 	// check token
 	if token == "" || secret == "" {
 		msg := fmt.Sprintf("Token or secret is None: %s %s", token, secret)
 		dataMsg := EmitDataMsg{Data: msg, Room: clientID}
-		s.Emit("data", dataMsg)
+		c.Emit("data", neffos.Marshal(dataMsg))
-		s.Emit("disconnect")
+		c.Emit("disconnect", nil)
 	}
 	tokenUser := service.GetTokenAsset(token)
 	if tokenUser.UserID == "" {
 		msg := "Token info is none, maybe token expired"
 		dataMsg := EmitDataMsg{Data: msg, Room: clientID}
-		s.Emit("data", dataMsg)
+		c.Emit("data", neffos.Marshal(dataMsg))
-		s.Emit("disconnect")
+		c.Emit("disconnect", nil)
 	}
 	currentUser := service.GetUserDetail(tokenUser.UserID)
-	asset := service.GetAsset(tokenUser.AssetID)
-	systemUser := service.GetSystemUser(tokenUser.SystemUserID)
-	if asset.ID == "" || systemUser.ID == "" {
+	if currentUser == nil {
-		return
+		msg := "User id error"
+		dataMsg := EmitDataMsg{Data: msg, Room: clientID}
+		c.Emit("data", neffos.Marshal(dataMsg))
+		c.Emit("disconnect", nil)
 	}
-	userR, userW := io.Pipe()
+	cc.Set("currentUser", currentUser)
-	conn := conns.GetWebConn(s.ID())
+	hostMsg := HostMsg{
-	conn.User = currentUser
+		Uuid: tokenUser.AssetID, UserID: tokenUser.SystemUserID,
-	client := Client{
+		Size: message.Size, Secret:secret,
-		Uuid: clientID, Cid: conn.Cid, user: conn.User,
-		WinChan: make(chan ssh.Window, 100), Conn: s,
-		UserRead: userR, UserWrite: userW, lock: new(sync.RWMutex),
-		pty: ssh.Pty{Term: "xterm", Window: win},
 	}
-	client.WinChan <- win
+	fmt.Println("Host msg: ", hostMsg)
-	conn.AddClient(clientID, &client)
+	hostWsMsg := neffos.Message{
+		Body:neffos.Marshal(hostMsg),
-	proxySrv := proxy.ProxyServer{
-		UserConn: &client, User: currentUser,
-		Asset: &asset, SystemUser: &systemUser,
 	}
-	go func() {
+	return OnHostHandler(c, hostWsMsg)
-		defer logger.Debug("web proxy end")
-		proxySrv.Proxy()
-		s.Emit("logout", RoomMsg{Room: clientID})
-	}()
 }
 // OnDataHandler 收发数据时触发
-func OnDataHandler(s socketio.Conn, message DataMsg) {
+func OnDataHandler(c *neffos.NSConn, msg neffos.Message) (err error) {
-	cid := message.Room
+	roomID := msg.Room
-	conn := conns.GetWebConn(s.ID())
+	client := clients.GetClient(roomID)
-	client := conn.GetClient(cid)
 	if client == nil {
 		return
 	}
-	_, _ = client.UserWrite.Write([]byte(message.Data))
+	var message DataMsg
+	err = msg.Unmarshal(&message)
+	if err != nil {
+		return
+	}
+	_, err = client.UserWrite.Write([]byte(message.Data))
+	return err
 }
 // OnResizeHandler 用户窗口改变时触发
-func OnResizeHandler(s socketio.Conn, message ResizeMsg) {
+func OnResizeHandler(c *neffos.NSConn, msg neffos.Message) (err error) {
+	var message ResizeMsg
+	err = msg.Unmarshal(&message)
+	if err != nil {
+		return
+	}
 	logger.Debugf("Web terminal on resize event trigger: %d*%d", message.Width, message.Height)
 	winSize := ssh.Window{Height: message.Height, Width: message.Width}
-	conn := conns.GetWebConn(s.ID())
+	for _, room := range c.Rooms() {
-	conn.SetWinSize(winSize)
+		roomID := room.Name
+		client := clients.GetClient(roomID)
+		if client != nil {
+			client.SetWinSize(winSize)
+		}
+	}
+	return nil
 }
 // OnLogoutHandler 用户登出一个会话时触发
-func OnLogoutHandler(s socketio.Conn, message string) {
+func OnLogoutHandler(c *neffos.NSConn, msg neffos.Message) (err error){
-	logger.Debug("Web terminal on logout event trigger")
+	logger.Debug("Web terminal on logout event trigger: ", msg.Room)
-	conn := conns.GetWebConn(s.ID())
+	var message RoomMsg
-	if conn == nil {
+	err = msg.Unmarshal(&message)
-		logger.Error("No conn found")
+	if err != nil {
-		return
-	}
-	client := conn.GetClient(message)
-	if client == nil {
-		logger.Error("No client found")
 		return
 	}
-	_ = client.Close()
+	roomID := message.Room
+	clients.DeleteClient(roomID)
+	return
 }
-// OnDisconnect websocket断开后触发
-func OnDisconnect(s socketio.Conn, msg string) {
-	logger.Debug("On disconnect event trigger")
-	conn := conns.GetWebConn(s.ID())
-	conn.Close()
-}
--- a/pkg/koko/task.go
+++ b/pkg/koko/task.go
@@ -70,7 +70,7 @@ func uploadRemainReplay(rootPath string) {
 			}
 			_ = os.Remove(path)
 		}
-		relayRecord := &proxy.ReplyRecorder{}
+		relayRecord := &proxy.ReplyRecorder{SessionID:sid}
 		relayRecord.AbsGzFilePath = absGzPath
 		relayRecord.Target, _ = filepath.Rel(path, rootPath)
 		relayRecord.UploadGzipFile(3)

--- a/pkg/model/assets.go
+++ b/pkg/model/assets.go
@@ -220,6 +220,7 @@ func SortAssetNodesByKey(assetNodes []Node) {
 const LoginModeManual = "manual"
 const (
+	AllAction      = "all"
 	ConnectAction  = "connect"
 	UploadAction   = "upload_file"
 	DownloadAction = "download_file"

--- a/pkg/model/connection.go
+++ b/pkg/model/connection.go
 package model
+const (
+	OperateRemoveDir = "Rmdir"
+	OperateDownaload = "Download"
+	OperateUpload    = "Upload"
+	OperateRename    = "Rename"
+	OperateMkdir     = "Mkdir"
+	OperateDelete    = "Delete"
+	OperateSymlink   = "Symlink"
+)
--- a/pkg/proxy/recorder.go
+++ b/pkg/proxy/recorder.go
@@ -22,7 +22,7 @@ func NewCommandRecorder(sid string) (recorder *CommandRecorder) {
 }
 func NewReplyRecord(sid string) (recorder *ReplyRecorder) {
-	recorder = &ReplyRecorder{sessionID: sid}
+	recorder = &ReplyRecorder{SessionID: sid}
 	recorder.initial()
 	return recorder
 }
@@ -93,7 +93,7 @@ func (c *CommandRecorder) record() {
 }
 type ReplyRecorder struct {
-	sessionID string
+	SessionID string
 	absFilePath   string
 	AbsGzFilePath string
@@ -119,7 +119,7 @@ func (r *ReplyRecorder) Record(b []byte) {
 }
 func (r *ReplyRecorder) prepare() {
-	sessionID := r.sessionID
+	sessionID := r.SessionID
 	rootPath := config.GetConf().RootPath
 	today := time.Now().UTC().Format("2006-01-02")
 	gzFileName := sessionID + ".replay.gz"
@@ -179,7 +179,7 @@ func (r *ReplyRecorder) UploadGzipFile(maxRetry int) {
 		err := r.storage.Upload(r.AbsGzFilePath, r.Target)
 		if err == nil {
 			_ = os.Remove(r.AbsGzFilePath)
-			service.FinishReply(r.sessionID)
+			service.FinishReply(r.SessionID)
 			break
 		}
 		// 如果还是失败，使用备用storage再传一次

--- a/pkg/service/assets.go
+++ b/pkg/service/assets.go
@@ -57,7 +57,7 @@ func GetSystemUserFilterRules(systemUserID string) (rules []model.SystemUserFilt
 	        "filter": "de7693ca-75d5-4639-986b-44ed390260a0"
 	    }
 	]`*/
-	Url := fmt.Sprintf(SystemUserCmdFilterRules, systemUserID)
+	Url := fmt.Sprintf(SystemUserCmdFilterRulesListURL, systemUserID)
 	err = authClient.Get(Url, &rules)
 	if err != nil {
@@ -94,7 +94,7 @@ func GetDomainWithGateway(gID string) (domain model.Domain) {
 }
 func GetTokenAsset(token string) (tokenUser model.TokenUser) {
-	Url := fmt.Sprintf(TokenAssetUrl, token)
+	Url := fmt.Sprintf(TokenAssetURL, token)
 	err := authClient.Get(Url, &tokenUser)
 	if err != nil {
 		logger.Error("Get Token Asset info failed: ", err)

--- a/pkg/service/perms.go
+++ b/pkg/service/perms.go
@@ -7,11 +7,14 @@ import (
 	"github.com/jumpserver/koko/pkg/model"
 )
-func GetUserAssets(userID, cachePolicy string) (assets model.AssetList) {
+func GetUserAssets(userID, cachePolicy, assetId string) (assets model.AssetList) {
 	if cachePolicy == "" {
 		cachePolicy = "1"
 	}
 	payload := map[string]string{"cache_policy": cachePolicy}
+	if assetId != "" {
+		payload["id"] = assetId
+	}
 	Url := fmt.Sprintf(UserAssetsURL, userID)
 	err := authClient.Get(Url, &assets, payload)
 	if err != nil {

--- a/pkg/service/terminal.go
+++ b/pkg/service/terminal.go
@@ -46,12 +46,12 @@ func FinishSession(data map[string]interface{}) {
 	var res map[string]interface{}
 	if sid, ok := data["id"]; ok {
-		playborad := map[string]interface{}{
+		payload := map[string]interface{}{
 			"is_finished": true,
 			"date_end":    data["date_end"],
 		}
 		Url := fmt.Sprintf(SessionDetailURL, sid)
-		err := authClient.Patch(Url, playborad, &res)
+		err := authClient.Patch(Url, payload, &res)
 		if err != nil {
 			logger.Error(err)
 		}
@@ -102,7 +102,7 @@ func PushSessionCommand(commands []*model.Command) (err error) {
 }
 func PushFTPLog(data *model.FTPLog) (err error) {
-	err = authClient.Post(FTPLogList, data, nil)
+	err = authClient.Post(FTPLogListURL, data, nil)
 	if err != nil {
 		logger.Error(err)
 	}

--- a/pkg/service/urls.go
+++ b/pkg/service/urls.go
@@ -3,16 +3,16 @@ package service
 const (
 	UserAuthURL    = "/api/authentication/v1/auth/"                      // post 验证用户登陆
 	UserProfileURL = "/api/users/v1/profile/"                            // 获取当前用户的基本信息
-	UserListUrl    = "/api/users/v1/users/"                              // 用户列表地址
+	UserListURL    = "/api/users/v1/users/"                              // 用户列表地址
 	UserDetailURL  = "/api/users/v1/users/%s/"                           // 获取用户信息
 	UserAuthOTPURL = "/api/authentication/v1/otp/auth/"                  // 验证OTP
-	TokenAssetUrl  = "/api/authentication/v1/connection-token/?token=%s" // Token name
+	TokenAssetURL  = "/api/authentication/v1/connection-token/?token=%s" // Token name
-	SystemUserAssetAuthURL   = "/api/assets/v1/system-user/%s/asset/%s/auth-info/" // 该系统用户对某资产的授权
+	SystemUserAssetAuthURL          = "/api/assets/v1/system-user/%s/asset/%s/auth-info/" // 该系统用户对某资产的授权
-	SystemUserCmdFilterRules = "/api/assets/v1/system-user/%s/cmd-filter-rules/"   // 过滤规则url
+	SystemUserCmdFilterRulesListURL = "/api/assets/v1/system-user/%s/cmd-filter-rules/"   // 过滤规则url
-	SystemUserDetailURL      = "/api/assets/v1/system-user/%s/"                    // 某个系统用户的信息
+	SystemUserDetailURL             = "/api/assets/v1/system-user/%s/"                    // 某个系统用户的信息
-	AssetDetailURL           = "/api/assets/v1/assets/%s/"                         // 某一个资产信息
+	AssetDetailURL                  = "/api/assets/v1/assets/%s/"                         // 某一个资产信息
-	DomainDetailURL          = "/api/assets/v1/domain/%s/?gateway=1"
+	DomainDetailURL                 = "/api/assets/v1/domain/%s/?gateway=1"
 	TerminalRegisterURL  = "/api/terminal/v2/terminal-registrations/" // 注册当前coco
 	TerminalConfigURL    = "/api/terminal/v1/terminal/config/"        // 从jumpserver获取coco的配置
@@ -22,11 +22,11 @@ const (
 	SessionDetailURL  = "/api/terminal/v1/sessions/%s/"        // finish session的时候发送
 	SessionReplayURL  = "/api/terminal/v1/sessions/%s/replay/" //上传录像
 	SessionCommandURL = "/api/terminal/v1/command/"            //上传批量命令
-	FTPLogList        = "/api/audits/v1/ftp-log/"              // 上传 ftp日志
+	FinishTaskURL     = "/api/terminal/v1/tasks/%s/"
-	FinishTaskURL = "/api/terminal/v1/tasks/%s/"
+	FTPLogListURL = "/api/audits/v1/ftp-log/" // 上传 ftp日志
-	UserAssetsURL                  = "/api/perms/v1/user/%s/assets/"                 //获取用户授权的所有资产
+	UserAssetsURL                  = "/api/perms/v1/users/%s/assets/"                 //获取用户授权的所有资产
-	UserNodesAssetsURL             = "/api/perms/v1/user/%s/nodes-assets/"           // 获取用户授权的所有节点信息 节点分组
+	UserNodesAssetsURL             = "/api/perms/v1/users/%s/nodes-assets/"           // 获取用户授权的所有节点信息 节点分组
-	ValidateUserAssetPermissionURL = "/api/perms/v1/asset-permission/user/validate/" //0不使用缓存 1 使用缓存 2 刷新缓存
+	ValidateUserAssetPermissionURL = "/api/perms/v1/asset-permissions/user/validate/" //0不使用缓存 1 使用缓存 2 刷新缓存
 )
--- a/pkg/service/users.go
+++ b/pkg/service/users.go
@@ -44,7 +44,7 @@ func GetProfile() (user *model.User, err error) {
 func GetUserByUsername(username string) (user *model.User, err error) {
 	var users []*model.User
 	payload := map[string]string{"username": username}
-	err = authClient.Get(UserListUrl, &users, payload)
+	err = authClient.Get(UserListURL, &users, payload)
 	if err != nil {
 		return
 	}

--- a/pkg/srvconn/sftpconn.go
+++ b/pkg/srvconn/sftpconn.go
--- a/pkg/srvconn/utils.go
+++ b/pkg/srvconn/utils.go
 package srvconn
 import (
-	"golang.org/x/crypto/ssh"
 	"io/ioutil"
+	"golang.org/x/crypto/ssh"
 )
 func GetPubKeyFromFile(keypath string) (ssh.Signer, error) {

--- a/pkg/sshd/server.go
+++ b/pkg/sshd/server.go
@@ -23,7 +23,7 @@ func StartServer() {
 	}
 	addr := net.JoinHostPort(conf.BindHost, conf.SSHPort)
-	logger.Infof("Start ssh server at %s", addr)
+	logger.Infof("Start SSH server at %s", addr)
 	sshServer = &ssh.Server{
 		Addr:                       addr,
 		KeyboardInteractiveHandler: auth.CheckMFA,