Dbconnect (#178)

* [Update] update db log info

* [Update] support mysql session

* [update] add db name

* [Update] add mysql-client

Dockerfile

@@ -25,6 +25,7 @@ RUN chmod 755 ./entrypoint.sh \
   && apk update \
   && apk add -U tzdata \
   && apk add curl \
+  && apk add mysql-client \
   && cp /usr/share/zoneinfo/Asia/Shanghai /etc/localtime \
   && echo "Asia/Shanghai" > /etc/timezone \
   && apk del tzdata \

cmd/locale/en_US/LC_MESSAGES/koko.po

@@ -49,143 +49,200 @@ msgstr ""
 
 #. i18n.T
 #: pkg/handler/banner.go:52
-msgid "refresh your assets and nodes"
+msgid "display the databases that you have permission"
 msgstr ""
 
 #. i18n.T
 #: pkg/handler/banner.go:53
-msgid "print help"
+msgid "refresh your assets and nodes"
 msgstr ""
 
 #. i18n.T
 #: pkg/handler/banner.go:54
+msgid "print help"
+msgstr ""
+
+#. i18n.T
+#: pkg/handler/banner.go:55
 msgid "exit"
 msgstr ""
 
 #. i18n.T
-#: pkg/handler/banner.go:89
+#: pkg/handler/banner.go:90
 msgid "ID"
 msgstr ""
 
 #. i18n.T
-#: pkg/handler/banner.go:90
+#: pkg/handler/banner.go:91
 msgid "hostname"
 msgstr ""
 
 #. i18n.T
-#: pkg/handler/banner.go:91
+#: pkg/handler/banner.go:92
 msgid "IP"
 msgstr ""
 
 #. i18n.T
-#: pkg/handler/banner.go:92
+#: pkg/handler/banner.go:93
 msgid "comment"
 msgstr ""
 
 #. i18n.T
-#: pkg/handler/banner.go:93
+#: pkg/handler/banner.go:94
 msgid "Page: %d, Count: %d, Total Page: %d, Total Count: %d"
 msgstr ""
 
 #. i18n.T
-#: pkg/handler/banner.go:94
+#: pkg/handler/banner.go:95
 msgid "No Assets"
 msgstr ""
 
 #. i18n.T
-#: pkg/handler/banner.go:95
+#: pkg/handler/banner.go:96
 msgid ""
 "Enter ID number directly login the asset, multiple search use // + field, "
 "such as: //16"
 msgstr ""
 
 #. i18n.T
-#: pkg/handler/banner.go:96
+#: pkg/handler/banner.go:97
 msgid "Page up: b\tPage down: n"
 msgstr ""
 
 #. i18n.T
-#: pkg/handler/banner.go:97
+#: pkg/handler/banner.go:98
 msgid "Node: [ ID.Name(Asset amount) ]"
 msgstr ""
 
 #. i18n.T
-#: pkg/handler/banner.go:98
+#: pkg/handler/banner.go:99
 msgid "Tips: Enter g+NodeID to display the host under the node, such as g1"
 msgstr ""
 
 #. i18n.T
-#: pkg/handler/banner.go:99
+#: pkg/handler/banner.go:100
 msgid "Refresh done"
 msgstr ""
 
 #. i18n.T
-#: pkg/handler/banner.go:100
+#: pkg/handler/banner.go:101
 msgid "Tips: Enter system user ID and directly login the asset [ %s(%s) ]"
 msgstr ""
 
 #. i18n.T
-#: pkg/handler/banner.go:101
+#: pkg/handler/banner.go:102
 msgid "Back: B/b"
 msgstr ""
 
 #. i18n.T
-#: pkg/handler/banner.go:102
+#: pkg/handler/banner.go:103
 msgid "Name"
 msgstr ""
 
 #. i18n.T
-#: pkg/handler/banner.go:103
+#: pkg/handler/banner.go:104
 msgid "Username"
 msgstr ""
 
 #. i18n.T
-#: pkg/handler/banner.go:104
+#: pkg/handler/banner.go:105
 msgid "all"
 msgstr ""
 
 #. i18n.T
-#: pkg/handler/banner.go:105
+#: pkg/handler/banner.go:106
 msgid "Search: %s"
 msgstr ""
 
 #. i18n.T
-#: pkg/proxy/parser.go:130
+#: pkg/handler/banner.go:107
+msgid "DBType"
+msgstr ""
+
+#. i18n.T
+#: pkg/handler/banner.go:108
+msgid "DB Name"
+msgstr ""
+
+#. i18n.T
+#: pkg/handler/banner.go:109
+msgid "No Databases"
+msgstr ""
+
+#. i18n.T
+#: pkg/handler/banner.go:110
+msgid ""
+"Enter ID number directly login the database, multiple search use // + field, "
+"such as: //16"
+msgstr ""
+
+#. i18n.T
+#: pkg/proxy/dbproxy.go:112
+msgid "Database connecting to %s %.1f"
+msgstr ""
+
+#. i18n.T
+#: pkg/proxy/dbproxy.go:131
+msgid "System user <%s> and database <%s> protocol are inconsistent."
+msgstr ""
+
+#. i18n.T
+#: pkg/proxy/dbproxy.go:137
+msgid "Database %s protocol client not installed."
+msgstr ""
+
+#. i18n.T
+#: pkg/proxy/dbswitch.go:148
+msgid "Database connect idle more than %d minutes, disconnect"
+msgstr ""
+
+#. i18n.T
+#: pkg/proxy/dbswitch.go:155
+msgid "Database connection terminated by administrator"
+msgstr ""
+
+#. i18n.T
+#: pkg/proxy/parser.go:140
 msgid "Command `%s` is forbidden"
 msgstr ""
 
 #. i18n.T
-#: pkg/proxy/proxy.go:143
+#: pkg/proxy/proxy.go:146
 msgid "Reuse SSH connections (%s@%s) [Number of connections: %d]"
 msgstr ""
 
 #. i18n.T
-#: pkg/proxy/proxy.go:161
+#: pkg/proxy/proxy.go:164
 msgid "Connecting to %s@%s  %.1f"
 msgstr ""
 
 #. i18n.T
-#: pkg/proxy/proxy.go:180
+#: pkg/proxy/proxy.go:183
 msgid "System user <%s> and asset <%s> protocol are inconsistent."
 msgstr ""
 
 #. i18n.T
-#: pkg/proxy/proxy.go:186
+#: pkg/proxy/proxy.go:189
 msgid ""
 "Terminal only support protocol ssh/telnet, please use web terminal to access"
 msgstr ""
 
 #. i18n.T
-#: pkg/proxy/sessmanager.go:67
+#: pkg/proxy/sessmanager.go:75
 msgid "Connect with api server failed"
 msgstr ""
 
 #. i18n.T
-#: pkg/proxy/switch.go:168
+#: pkg/proxy/sessmanager.go:117
+msgid "Create database session failed"
+msgstr ""
+
+#. i18n.T
+#: pkg/proxy/switch.go:173
 msgid "Connect idle more than %d minutes, disconnect"
 msgstr ""
 
 #. i18n.T
-#: pkg/proxy/switch.go:175
+#: pkg/proxy/switch.go:180
 msgid "Terminated by administrator"
 msgstr ""

cmd/locale/zh_CN/LC_MESSAGES/koko.po

@@ -51,51 +51,57 @@ msgstr "显示您有权限的节点"
 
 #. i18n.T
 #: pkg/handler/banner.go:52
+#, fuzzy
+msgid "display the databases that you have permission"
+msgstr "显示您有权限的数据库"
+
+#. i18n.T
+#: pkg/handler/banner.go:53
 msgid "refresh your assets and nodes"
 msgstr "刷新最新的机器和节点信息"
 
 #. i18n.T
-#: pkg/handler/banner.go:53
+#: pkg/handler/banner.go:54
 msgid "print help"
 msgstr "显示帮助"
 
 #. i18n.T
-#: pkg/handler/banner.go:54
+#: pkg/handler/banner.go:55
 msgid "exit"
 msgstr "退出"
 
 #. i18n.T
-#: pkg/handler/banner.go:89
+#: pkg/handler/banner.go:90
 msgid "ID"
 msgstr "ID"
 
 #. i18n.T
-#: pkg/handler/banner.go:90
+#: pkg/handler/banner.go:91
 msgid "hostname"
 msgstr "主机名"
 
 #. i18n.T
-#: pkg/handler/banner.go:91
+#: pkg/handler/banner.go:92
 msgid "IP"
 msgstr "IP"
 
 #. i18n.T
-#: pkg/handler/banner.go:92
+#: pkg/handler/banner.go:93
 msgid "comment"
 msgstr "备注"
 
 #. i18n.T
-#: pkg/handler/banner.go:93
+#: pkg/handler/banner.go:94
 msgid "Page: %d, Count: %d, Total Page: %d, Total Count: %d"
 msgstr "页码：%d，每页行数：%d，总页数：%d，总数量：%d"
 
 #. i18n.T
-#: pkg/handler/banner.go:94
+#: pkg/handler/banner.go:95
 msgid "No Assets"
 msgstr "没有资产"
 
 #. i18n.T
-#: pkg/handler/banner.go:95
+#: pkg/handler/banner.go:96
 #, fuzzy
 msgid ""
 "Enter ID number directly login the asset, multiple search use // + field, "
@@ -103,28 +109,28 @@ msgid ""
 msgstr "提示：输入资产ID直接登录，二级搜索使用 // + 字段，如：//192"
 
 #. i18n.T
-#: pkg/handler/banner.go:96
+#: pkg/handler/banner.go:97
 #, fuzzy
 msgid "Page up: b\tPage down: n"
 msgstr "上一页：b 下一页：n"
 
 #. i18n.T
-#: pkg/handler/banner.go:97
+#: pkg/handler/banner.go:98
 msgid "Node: [ ID.Name(Asset amount) ]"
 msgstr "节点：[ ID.名称(资产数量) ]"
 
 #. i18n.T
-#: pkg/handler/banner.go:98
+#: pkg/handler/banner.go:99
 msgid "Tips: Enter g+NodeID to display the host under the node, such as g1"
 msgstr "提示：输入 g+节点ID 显示节点下主机，如: g1"
 
 #. i18n.T
-#: pkg/handler/banner.go:99
+#: pkg/handler/banner.go:100
 msgid "Refresh done"
 msgstr "刷新完成"
 
 #. i18n.T
-#: pkg/handler/banner.go:100
+#: pkg/handler/banner.go:101
 #, fuzzy
 msgid "Tips: Enter system user ID and directly login the asset [ %s(%s) ]"
 msgstr ""
@@ -132,69 +138,127 @@ msgstr ""
 "提示：输入系统用户ID，登录资产[ %s(%s) ]\n"
 
 #. i18n.T
-#: pkg/handler/banner.go:101
+#: pkg/handler/banner.go:102
 msgid "Back: B/b"
 msgstr "返回：B/b"
 
 #. i18n.T
-#: pkg/handler/banner.go:102
+#: pkg/handler/banner.go:103
 msgid "Name"
 msgstr "名称"
 
 #. i18n.T
-#: pkg/handler/banner.go:103
+#: pkg/handler/banner.go:104
 #, fuzzy
 msgid "Username"
 msgstr "用户名"
 
 #. i18n.T
-#: pkg/handler/banner.go:104
+#: pkg/handler/banner.go:105
 msgid "all"
 msgstr "所有"
 
 #. i18n.T
-#: pkg/handler/banner.go:105
+#: pkg/handler/banner.go:106
 #, fuzzy
 msgid "Search: %s"
 msgstr "搜索: %s"
 
 #. i18n.T
-#: pkg/proxy/parser.go:130
+#: pkg/handler/banner.go:107
+msgid "DBType"
+msgstr "数据库类型"
+
+#. i18n.T
+#: pkg/handler/banner.go:108
+#, fuzzy
+msgid "DB Name"
+msgstr "数据库名称"
+
+#. i18n.T
+#: pkg/handler/banner.go:109
+msgid "No Databases"
+msgstr "无数据库"
+
+#. i18n.T
+#: pkg/handler/banner.go:110
+#, fuzzy
+msgid ""
+"Enter ID number directly login the database, multiple search use // + field, "
+"such as: //16"
+msgstr "提示：输入数据库ID直接登录，二级搜索使用 // + 字段，如：//192"
+
+#. i18n.T
+#: pkg/proxy/dbproxy.go:112
+#, fuzzy
+msgid "Database connecting to %s %.1f"
+msgstr "连接数据库 %s  %.1f"
+
+#. i18n.T
+#: pkg/proxy/dbproxy.go:131
+#, fuzzy
+msgid "System user <%s> and database <%s> protocol are inconsistent."
+msgstr "系统用户<%s>和资产<%s>协议不一致"
+
+#. i18n.T
+#: pkg/proxy/dbproxy.go:137
+msgid "Database %s protocol client not installed."
+msgstr "%s 协议的数据库客户端未安装"
+
+#. i18n.T
+#: pkg/proxy/dbswitch.go:148
+#, fuzzy
+msgid "Database connect idle more than %d minutes, disconnect"
+msgstr "数据库连接空闲时间超过 %d 分钟，断开连接"
+
+#. i18n.T
+#: pkg/proxy/dbswitch.go:155
+#, fuzzy
+msgid "Database connection terminated by administrator"
+msgstr "管理员中断数据库连接"
+
+#. i18n.T
+#: pkg/proxy/parser.go:140
 msgid "Command `%s` is forbidden"
 msgstr "命令 `%s` 是被禁止的 ..."
 
 #. i18n.T
-#: pkg/proxy/proxy.go:143
+#: pkg/proxy/proxy.go:146
 msgid "Reuse SSH connections (%s@%s) [Number of connections: %d]"
 msgstr "复用SSH连接（%s@%s）[连接数量: %d]"
 
 #. i18n.T
-#: pkg/proxy/proxy.go:161
+#: pkg/proxy/proxy.go:164
 msgid "Connecting to %s@%s  %.1f"
 msgstr "开始连接到 %s@%s  %.1f"
 
 #. i18n.T
-#: pkg/proxy/proxy.go:180
+#: pkg/proxy/proxy.go:183
 msgid "System user <%s> and asset <%s> protocol are inconsistent."
 msgstr "系统用户<%s>和资产<%s>协议不一致"
 
 #. i18n.T
-#: pkg/proxy/proxy.go:186
+#: pkg/proxy/proxy.go:189
 msgid ""
 "Terminal only support protocol ssh/telnet, please use web terminal to access"
 msgstr "终端仅支持ssh/telnet协议，请使用web终端登录"
 
 #. i18n.T
-#: pkg/proxy/sessmanager.go:67
+#: pkg/proxy/sessmanager.go:75
 msgid "Connect with api server failed"
 msgstr "连接API服务失败"
 
 #. i18n.T
-#: pkg/proxy/switch.go:168
+#: pkg/proxy/sessmanager.go:117
+msgid "Create database session failed"
+msgstr "创建数据库会话失败"
+
+#. i18n.T
+#: pkg/proxy/switch.go:173
 msgid "Connect idle more than %d minutes, disconnect"
 msgstr "空闲时间超过%d分钟，断开连接"
 
 #. i18n.T
-#: pkg/proxy/switch.go:175
+#: pkg/proxy/switch.go:180
 msgid "Terminated by administrator"
 msgstr "管理员中断连接"

go.mod

@@ -11,6 +11,7 @@ require (
 	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/creack/pty v1.1.9
 	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

go.sum

@@ -17,6 +17,8 @@ github.com/aws/aws-sdk-go v1.19.46 h1:lRqljzjkGmEeiawkw4z1QgtCnU/S5Jw8lNeUuvmydU
 github.com/aws/aws-sdk-go v1.19.46/go.mod h1:KmX6BPdI08NWTb3/sm4ZGu5ShLoqVDhKgpiN924inxo=
 github.com/baiyubin/aliyun-sts-go-sdk v0.0.0-20180326062324-cfa1a18b161f h1:ZNv7On9kyUzm7fvRZumSyy/IUiSC7AzL0I1jKKtwooA=
 github.com/baiyubin/aliyun-sts-go-sdk v0.0.0-20180326062324-cfa1a18b161f/go.mod h1:AuiFmCCPBSrqvVMvuqFuk0qogytodnVFVSN5CeJB8Gc=
+github.com/creack/pty v1.1.9 h1:uDmaGzcdjhF4i/plgjmEsriH11Y0o7RKapEf/LDaM3w=
+github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=

pkg/handler/banner.go

@@ -49,9 +49,10 @@ func Initial() {
 		{id: 2, instruct: i18n.T("/ + IP, Hostname, Comment"), helpText: i18n.T("to search, such as: /192.168")},
 		{id: 3, instruct: "p", helpText: i18n.T("display the host you have permission")},
 		{id: 4, instruct: "g", helpText: i18n.T("display the node that you have permission")},
-		{id: 5, instruct: "r", helpText: i18n.T("refresh your assets and nodes")},
-		{id: 6, instruct: "h", helpText: i18n.T("print help")},
-		{id: 7, instruct: "q", helpText: i18n.T("exit")},
+		{id: 5, instruct: "d", helpText: i18n.T("display the databases that you have permission")},
+		{id: 6, instruct: "r", helpText: i18n.T("refresh your assets and nodes")},
+		{id: 7, instruct: "h", helpText: i18n.T("print help")},
+		{id: 8, instruct: "q", helpText: i18n.T("exit")},
 	}
 }
 
@@ -103,6 +104,10 @@ func getI18nFromMap(name string) string {
 			"Username":          i18n.T("Username"),
 			"All":               i18n.T("all"),
 			"SearchTip":         i18n.T("Search: %s"),
+			"DBType":            i18n.T("DBType"),
+			"DBName":            i18n.T("DB Name"),
+			"NoDatabases":       i18n.T("No Databases"),
+			"DBLoginTip":        i18n.T("Enter ID number directly login the database, multiple search use // + field, such as: //16"),
 		}
 	})
 	return i18nMap[name]

pkg/handler/dbpaginator.go

+package handler
+
+import (
+	"strings"
+	"sync"
+
+	"github.com/jumpserver/koko/pkg/model"
+)
+
+type DatabasePaginator interface {
+	Paginator
+	RetrievePageData(pageIndex int) []model.Database
+	SearchAsset(key string) []model.Database
+	SearchAgain(key string) []model.Database
+	Name() string
+	SearchKeys() []string
+}
+
+func NewLocalDatabasePaginator(data [] model.Database, pageSize int) DatabasePaginator {
+	p := localDatabasePaginator{
+		allData:       data,
+		currentData:   data,
+		pageSize:      pageSize,
+		currentOffset: 0,
+		currentPage:   1,
+		search:        make([]string, 0, 4),
+		lock:          new(sync.RWMutex),
+	}
+	return &p
+}
+
+type localDatabasePaginator struct {
+	allData []model.Database
+
+	currentData []model.Database
+
+	currentPage int
+
+	pageSize  int
+	totalPage int
+
+	currentOffset int
+
+	search []string
+	lock   *sync.RWMutex
+
+	currentResult []model.Database
+}
+
+func (l *localDatabasePaginator) Name() string {
+	return "local"
+}
+
+func (l *localDatabasePaginator) SearchKeys() []string {
+	return l.search
+}
+
+func (l *localDatabasePaginator) HasPrev() bool {
+	l.lock.RLock()
+	defer l.lock.RUnlock()
+	return l.currentPage > 1
+}
+
+func (l *localDatabasePaginator) HasNext() bool {
+	l.lock.RLock()
+	defer l.lock.RUnlock()
+	return l.currentPage < l.totalPage
+}
+
+func (l *localDatabasePaginator) CurrentPage() int {
+	l.lock.RLock()
+	defer l.lock.RUnlock()
+	return l.currentPage
+}
+
+func (l *localDatabasePaginator) TotalCount() int {
+	l.lock.RLock()
+	defer l.lock.RUnlock()
+	return len(l.currentData)
+}
+
+func (l *localDatabasePaginator) TotalPage() int {
+	l.lock.RLock()
+	defer l.lock.RUnlock()
+	return l.totalPage
+}
+
+func (l *localDatabasePaginator) PageSize() int {
+	l.lock.RLock()
+	defer l.lock.RUnlock()
+	return l.pageSize
+}
+
+func (l *localDatabasePaginator) SetPageSize(size int) {
+	if size <= 0 {
+		size = len(l.currentData)
+	}
+	l.lock.Lock()
+	defer l.lock.Unlock()
+
+	if l.pageSize == size {
+		return
+	}
+	l.pageSize = size
+}
+
+func (l *localDatabasePaginator) RetrievePageData(pageIndex int) []model.Database {
+	l.lock.Lock()
+	defer l.lock.Unlock()
+	return l.retrievePageData(pageIndex)
+}
+
+func (l *localDatabasePaginator) SearchAsset(key string) []model.Database {
+	l.lock.Lock()
+	defer l.lock.Unlock()
+	l.search = l.search[:0]
+	l.search = append(l.search, key)
+	l.currentData = searchFromLocalDBs(l.allData, key)
+	l.currentPage = 1
+	l.currentOffset = 0
+	return l.retrievePageData(1)
+}
+
+func (l *localDatabasePaginator) SearchAgain(key string) []model.Database {
+	l.lock.Lock()
+	defer l.lock.Unlock()
+	l.currentData = searchFromLocalDBs(l.currentData, key)
+	l.search = append(l.search, key)
+	l.currentPage = 1
+	l.currentOffset = 0
+	return l.retrievePageData(1)
+}
+
+func (l *localDatabasePaginator) retrievePageData(pageIndex int) []model.Database {
+	offsetPage := pageIndex - l.currentPage
+	totalOffset := offsetPage * l.pageSize
+	l.currentOffset += totalOffset
+
+	switch {
+	case l.currentOffset <= 0:
+		l.currentOffset = 0
+	case l.currentOffset >= len(l.currentData):
+		l.currentOffset = len(l.currentData)
+	case l.pageSize >= len(l.currentData):
+		l.currentOffset = 0
+	}
+
+	end := l.currentOffset + l.pageSize
+	if end >= len(l.currentData) {
+		end = len(l.currentData)
+	}
+	l.currentResult = l.currentData[l.currentOffset:end]
+	l.updatePageInfo()
+	return l.currentResult
+}
+
+func (l *localDatabasePaginator) updatePageInfo() {
+	pageSize := l.pageSize
+	totalCount := len(l.currentData)
+
+	switch totalCount % pageSize {
+	case 0:
+		l.totalPage = totalCount / pageSize
+	default:
+		l.totalPage = (totalCount / pageSize) + 1
+	}
+	offset := l.currentOffset + len(l.currentResult)
+	switch offset % pageSize {
+	case 0:
+		l.currentPage = offset / pageSize
+	default:
+		l.currentPage = (offset / pageSize) + 1
+	}
+}
+
+func searchFromLocalDBs(dbs []model.Database, key string) []model.Database {
+	displayDBs := make([]model.Database, 0, len(dbs))
+	key = strings.ToLower(key)
+	for _, db := range dbs {
+		contents := []string{strings.ToLower(db.Name),strings.ToLower(db.DBName),
+			strings.ToLower(db.Host), strings.ToLower(db.Comment)}
+		if isSubstring(contents, key) {
+			displayDBs = append(displayDBs, db)
+		}
+	}
+	return displayDBs
+}

pkg/handler/dispatch.go

@@ -10,6 +10,8 @@ import (
 	"github.com/jumpserver/koko/pkg/config"
 	"github.com/jumpserver/koko/pkg/logger"
 	"github.com/jumpserver/koko/pkg/model"
+	"github.com/jumpserver/koko/pkg/proxy"
+	"github.com/jumpserver/koko/pkg/service"
 	"github.com/jumpserver/koko/pkg/utils"
 )
 
@@ -32,20 +34,34 @@ func (h *interactiveHandler) Dispatch() {
 					h.movePrePage()
 					break
 				}
+				if h.dbPaginator != nil {
+					h.moveDBPrePage()
+					break
+				}
 				if ok := h.searchOrProxy(line); ok {
 					continue
 				}
+			case "d":
+				h.assetPaginator = nil
+				h.dbPaginator = h.getDatabasePaginator()
+				h.currentDBData = h.dbPaginator.RetrievePageData(1)
 			case "n":
 				if h.assetPaginator != nil {
 					h.moveNextPage()
 					break
 				}
+				if h.dbPaginator != nil {
+					h.moveDBNextPage()
+					break
+				}
 				if ok := h.searchOrProxy(line); ok {
 					continue
 				}
 			case "":
 				if h.assetPaginator != nil {
 					h.moveNextPage()
+				} else if h.dbPaginator != nil {
+					h.moveDBNextPage()
 				} else {
 					h.resetPaginator()
 				}
@@ -96,11 +112,19 @@ func (h *interactiveHandler) Dispatch() {
 				}
 			}
 		}
-		h.displayPageAssets()
+		if h.dbPaginator != nil {
+			h.displayPageDatabase()
+		}
+		if h.assetPaginator != nil {
+			h.displayPageAssets()
+		}
+
 	}
 }
 
 func (h *interactiveHandler) resetPaginator() {
+	h.dbPaginator = nil
+	h.currentDBData = nil
 	h.assetPaginator = h.getAssetPaginator()
 	h.currentData = h.assetPaginator.RetrievePageData(1)
 }
@@ -210,6 +234,26 @@ func (h *interactiveHandler) searchAssets(key string) []model.Asset {
 }
 
 func (h *interactiveHandler) searchOrProxy(key string) bool {
+	if h.dbPaginator != nil {
+		if indexNum, err := strconv.Atoi(key); err == nil && len(h.currentDBData) > 0 {
+			if indexNum > 0 && indexNum <= len(h.currentDBData) {
+				dbSelected := h.currentDBData[indexNum-1]
+				h.ProxyDB(dbSelected)
+				h.dbPaginator = nil
+				h.currentDBData = nil
+				return true
+			}
+		}
+		if data := h.dbPaginator.SearchAsset(key); len(data) == 1 {
+			h.ProxyDB(data[0])
+			h.dbPaginator = nil
+			h.currentDBData = nil
+			return true
+		} else {
+			h.currentDBData = data
+		}
+		return false
+	}
 	if indexNum, err := strconv.Atoi(key); err == nil && len(h.currentSortedData) > 0 {
 		if indexNum > 0 && indexNum <= len(h.currentSortedData) {
 			assetSelect := h.currentSortedData[indexNum-1]
@@ -231,10 +275,16 @@ func (h *interactiveHandler) searchOrProxy(key string) bool {
 }
 
 func (h *interactiveHandler) searchAssetAndDisplay(key string) {
+	h.currentDBData = nil
+	h.dbPaginator = nil
 	h.currentData = h.searchAssets(key)
 }
 
 func (h *interactiveHandler) searchAssetsAgain(key string) {
+	if h.dbPaginator != nil {
+		h.currentDBData = h.dbPaginator.SearchAgain(key)
+		return
+	}
 	if h.assetPaginator == nil {
 		h.assetPaginator = h.getAssetPaginator()
 		h.currentData = h.assetPaginator.SearchAsset(key)
@@ -279,3 +329,187 @@ func (h *interactiveHandler) getAssetPaginator() AssetPaginator {
 func (h *interactiveHandler) getNodeAssetPaginator(node model.Node) AssetPaginator {
 	return NewNodeAssetPaginator(*h.user, node, getPageSize(h.term))
 }
+
+func (h *interactiveHandler) getDatabasePaginator() DatabasePaginator {
+	dbs := service.GetUserDatabases(h.user.ID)
+	return NewLocalDatabasePaginator(dbs, getPageSize(h.term))
+}
+
+func (h *interactiveHandler) displayPageDatabase() {
+	if len(h.currentDBData) == 0 {
+		_, _ = h.term.Write([]byte(getI18nFromMap("NoDatabases") + "\n\r"))
+		h.dbPaginator = nil
+		return
+	}
+	Labels := []string{getI18nFromMap("ID"), getI18nFromMap("Name"),
+		getI18nFromMap("IP"), getI18nFromMap("DBType"),
+		getI18nFromMap("DBName"),getI18nFromMap("Comment")}
+	fields := []string{"ID", "name", "IP", "DBType","DBName", "comment"}
+	data := make([]map[string]string, len(h.currentDBData))
+	for i, j := range h.currentDBData {
+		row := make(map[string]string)
+		row["ID"] = strconv.Itoa(i + 1)
+		row["name"] = j.Name
+		row["IP"] = j.Host
+		row["DBType"] = j.DBType
+		row["DBName"] = j.DBName
+
+		comments := make([]string, 0)
+		for _, item := range strings.Split(strings.TrimSpace(j.Comment), "\r\n") {
+			if strings.TrimSpace(item) == "" {
+				continue
+			}
+			comments = append(comments, strings.ReplaceAll(strings.TrimSpace(item), " ", ","))
+		}
+		row["comment"] = strings.Join(comments, "|")
+		data[i] = row
+	}
+	w, _ := h.term.GetSize()
+
+	currentPage := h.dbPaginator.CurrentPage()
+	pageSize := h.dbPaginator.PageSize()
+	totalPage := h.dbPaginator.TotalPage()
+	totalCount := h.dbPaginator.TotalCount()
+
+	caption := fmt.Sprintf(getI18nFromMap("AssetTableCaption"),
+		currentPage, pageSize, totalPage, totalCount)
+
+	caption = utils.WrapperString(caption, utils.Green)
+	table := common.WrapperTable{
+		Fields: fields,
+		Labels: Labels,
+		FieldsSize: map[string][3]int{
+			"ID":      {0, 0, 5},
+			"name":    {0, 8, 0},
+			"IP":      {0, 15, 40},
+			"DBType":  {0, 8, 0},
+			"DBName":  {0, 8, 0},
+			"comment": {0, 0, 0},
+		},
+		Data:        data,
+		TotalSize:   w,
+		Caption:     caption,
+		TruncPolicy: common.TruncMiddle,
+	}
+	table.Initial()
+	header := getI18nFromMap("All")
+	keys := h.dbPaginator.SearchKeys()
+	switch h.dbPaginator.Name() {
+	case "local", "remote":
+		if len(keys) != 0 {
+			header = strings.Join(keys, " ")
+		}
+	default:
+		header = fmt.Sprintf("%s %s", h.dbPaginator.Name(), strings.Join(keys, " "))
+	}
+	searchHeader := fmt.Sprintf(getI18nFromMap("SearchTip"), header)
+	actionTip := fmt.Sprintf("%s %s", getI18nFromMap("DBLoginTip"), getI18nFromMap("PageActionTip"))
+
+	_, _ = h.term.Write([]byte(utils.CharClear))
+	_, _ = h.term.Write([]byte(table.Display()))
+	utils.IgnoreErrWriteString(h.term, utils.WrapperString(actionTip, utils.Green))
+	utils.IgnoreErrWriteString(h.term, utils.CharNewLine)
+	utils.IgnoreErrWriteString(h.term, utils.WrapperString(searchHeader, utils.Green))
+	utils.IgnoreErrWriteString(h.term, utils.CharNewLine)
+}
+
+func (h *interactiveHandler) moveDBPrePage() {
+	if h.dbPaginator == nil || !h.dbPaginator.HasPrev() {
+		return
+	}
+	h.dbPaginator.SetPageSize(getPageSize(h.term))
+	prePage := h.dbPaginator.CurrentPage() - 1
+	h.currentDBData = h.dbPaginator.RetrievePageData(prePage)
+}
+
+func (h *interactiveHandler) moveDBNextPage() {
+	if h.dbPaginator == nil || !h.dbPaginator.HasNext() {
+		return
+	}
+	h.dbPaginator.SetPageSize(getPageSize(h.term))
+	prePage := h.dbPaginator.CurrentPage() + 1
+	h.currentDBData = h.dbPaginator.RetrievePageData(prePage)
+}
+
+func (h *interactiveHandler) ProxyDB(dbSelect model.Database) {
+	systemUsers := service.GetUserDatabaseSystemUsers(h.user.ID, dbSelect.ID)
+	systemUserSelect, ok := h.chooseDBSystemUser(dbSelect, systemUsers)
+	if !ok {
+		return
+	}
+	p := proxy.DBProxyServer{
+		UserConn:   h.sess,
+		User:       h.user,
+		Database:   &dbSelect,
+		SystemUser: &systemUserSelect,
+	}
+	p.Proxy()
+	logger.Infof("Request %s: database %s proxy end", h.sess.Uuid, dbSelect.Name)
+}
+
+func (h *interactiveHandler) chooseDBSystemUser(dbAsset model.Database,
+	systemUsers []model.SystemUser) (systemUser model.SystemUser, ok bool) {
+
+	length := len(systemUsers)
+	switch length {
+	case 0:
+		return model.SystemUser{}, false
+	case 1:
+		return systemUsers[0], true
+	default:
+	}
+	displaySystemUsers := selectHighestPrioritySystemUsers(systemUsers)
+	if len(displaySystemUsers) == 1 {
+		return displaySystemUsers[0], true
+	}
+
+	Labels := []string{getI18nFromMap("ID"), getI18nFromMap("Name"), getI18nFromMap("Username")}
+	fields := []string{"ID", "Name", "Username"}
+
+	data := make([]map[string]string, len(displaySystemUsers))
+	for i, j := range displaySystemUsers {
+		row := make(map[string]string)
+		row["ID"] = strconv.Itoa(i + 1)
+		row["Name"] = j.Name
+		row["Username"] = j.Username
+		data[i] = row
+	}
+	w, _ := h.term.GetSize()
+	table := common.WrapperTable{
+		Fields: fields,
+		Labels: Labels,
+		FieldsSize: map[string][3]int{
+			"ID":       {0, 0, 5},
+			"Name":     {0, 8, 0},
+			"Username": {0, 10, 0},
+		},
+		Data:        data,
+		TotalSize:   w,
+		TruncPolicy: common.TruncMiddle,
+	}
+	table.Initial()
+
+	h.term.SetPrompt("ID> ")
+	defer h.term.SetPrompt("Opt> ")
+	selectUserTip := fmt.Sprintf(getI18nFromMap("SelectUserTip"), dbAsset.Name, dbAsset.Host)
+	for {
+		utils.IgnoreErrWriteString(h.term, table.Display())
+		utils.IgnoreErrWriteString(h.term, selectUserTip)
+		utils.IgnoreErrWriteString(h.term, getI18nFromMap("BackTip"))
+		utils.IgnoreErrWriteString(h.term, "\r\n")
+		line, err := h.term.ReadLine()
+		if err != nil {
+			return
+		}
+		line = strings.TrimSpace(line)
+		switch strings.ToLower(line) {
+		case "q", "b", "quit", "exit", "back":
+			return
+		}
+		if num, err := strconv.Atoi(line); err == nil {
+			if num > 0 && num <= len(displaySystemUsers) {
+				return displaySystemUsers[num-1], true
+			}
+		}
+	}
+}

pkg/handler/session.go

@@ -68,6 +68,9 @@ type interactiveHandler struct {
 	currentData       []model.Asset
 
 	assetPaginator AssetPaginator
+
+	dbPaginator   DatabasePaginator
+	currentDBData []model.Database
 }
 
 func (h *interactiveHandler) Initial() {
@@ -207,6 +210,7 @@ func (h *interactiveHandler) refreshAssetsAndNodesData() {
 		logger.Error("refresh Assets  Nodes err:", err)
 	}
 	h.assetPaginator = nil
+	h.dbPaginator = nil
 }
 
 func (h *interactiveHandler) loadUserNodes(cachePolicy string) {
@@ -238,6 +242,7 @@ func (h *interactiveHandler) Proxy(ctx context.Context) {
 	h.pauseWatchWinSize()
 	p.Proxy()
 	h.resumeWatchWinSize()
+	logger.Infof("Request %s: asset %s proxy end", h.sess.Uuid, h.assetSelect.Hostname)
 }
 
 func ConstructAssetNodeTree(assetNodes []model.Node) treeprint.Tree {
@@ -335,4 +340,4 @@ func getPageSize(term *utils.Terminal) int {
 		pageSize = 1
 	}
 	return pageSize
-}
\ No newline at end of file
+}

pkg/model/assets.go

@@ -90,7 +90,6 @@ type Asset struct {
 	IP              string       `json:"ip"`
 	Os              string       `json:"os"`
 	Domain          string       `json:"domain"`
-	Platform        string       `json:"platform"`
 	Comment         string       `json:"comment"`
 	Protocols       []string     `json:"protocols,omitempty"`
 	OrgID           string       `json:"org_id"`

pkg/model/database.go

+package model
+
+import "fmt"
+
+type Database struct {
+	ID      string `json:"id"`
+	Name    string `json:"name"`
+	DBType  string `json:"type"`
+	Host    string `json:"host"`
+	Port    int    `json:"port"`
+	DBName  string `json:"database"`
+	OrgID   string `json:"org_id"`
+	Comment string `json:"comment"`
+}
+
+func (db Database) String() string {
+	return fmt.Sprintf("%s://%s:%d/%s", db.DBType, db.Host, db.Port, db.DBName)
+}

pkg/proxy/dbparser.go

+package proxy
+
+import (
+	"bytes"
+	"sync"
+
+	"github.com/jumpserver/koko/pkg/logger"
+)
+
+const (
+	DBInputParserName  = "DB Input parser"
+	DBOutputParserName = "DB Output parser"
+)
+
+func newDBParser(id string) DBParser {
+	dbParser := DBParser{
+		id: id,
+	}
+	dbParser.initial()
+	return dbParser
+}
+
+type DBParser struct {
+	id string
+
+	userOutputChan chan []byte
+	srvOutputChan  chan []byte
+	cmdRecordChan  chan [2]string
+
+	inputInitial  bool
+	inputPreState bool
+	inputState    bool
+	once          *sync.Once
+	lock          *sync.RWMutex
+
+	command         string
+	output          string
+	cmdInputParser  *CmdParser
+	cmdOutputParser *CmdParser
+
+	closed chan struct{}
+}
+
+func (p *DBParser) initial() {
+	p.once = new(sync.Once)
+	p.lock = new(sync.RWMutex)
+
+	p.cmdInputParser = NewCmdParser(p.id, DBInputParserName)
+	p.cmdOutputParser = NewCmdParser(p.id, DBOutputParserName)
+
+	p.closed = make(chan struct{})
+	p.cmdRecordChan = make(chan [2]string, 1024)
+}
+
+// ParseStream 解析数据流
+func (p *DBParser) ParseStream(userInChan, srvInChan <-chan []byte) (userOut, srvOut <-chan []byte) {
+
+	p.userOutputChan = make(chan []byte, 1)
+	p.srvOutputChan = make(chan []byte, 1)
+	logger.Infof("DB Session %s: Parser start", p.id)
+	go func() {
+		defer func() {
+			// 会话结束，结算命令结果
+			p.sendCommandRecord()
+			close(p.cmdRecordChan)
+			close(p.userOutputChan)
+			close(p.srvOutputChan)
+			_ = p.cmdOutputParser.Close()
+			_ = p.cmdInputParser.Close()
+			logger.Infof("DB Session %s: Parser routine done", p.id)
+		}()
+		for {
+			select {
+			case <-p.closed:
+				return
+			case b, ok := <-userInChan:
+				if !ok {
+					return
+				}
+				b = p.ParseUserInput(b)
+				select {
+				case <-p.closed:
+					return
+				case p.userOutputChan <- b:
+				}
+
+			case b, ok := <-srvInChan:
+				if !ok {
+					return
+				}
+				b = p.ParseServerOutput(b)
+				select {
+				case <-p.closed:
+					return
+				case p.srvOutputChan <- b:
+				}
+
+			}
+		}
+	}()
+	return p.userOutputChan, p.srvOutputChan
+}
+
+// parseInputState 切换用户输入状态, 并结算命令和结果
+func (p *DBParser) parseInputState(b []byte) []byte {
+	p.inputPreState = p.inputState
+	if bytes.Contains(b, charEnter) {
+		// 连续输入enter key, 结算上一条可能存在的命令结果
+		p.sendCommandRecord()
+		p.inputState = false
+		// 用户输入了Enter，开始结算命令
+		p.parseCmdInput()
+	} else {
+		p.inputState = true
+		// 用户又开始输入，并上次不处于输入状态，开始结算上次命令的结果
+		if !p.inputPreState {
+			p.sendCommandRecord()
+		}
+	}
+	return b
+}
+
+// parseCmdInput 解析命令的输入
+func (p *DBParser) parseCmdInput() {
+	p.command = p.cmdInputParser.Parse()
+}
+
+// parseCmdOutput 解析命令输出
+func (p *DBParser) parseCmdOutput() {
+	p.output = p.cmdOutputParser.Parse()
+}
+
+// ParseUserInput 解析用户的输入
+func (p *DBParser) ParseUserInput(b []byte) []byte {
+	p.lock.Lock()
+	defer p.lock.Unlock()
+	p.once.Do(func() {
+		p.inputInitial = true
+	})
+	nb := p.parseInputState(b)
+	return nb
+}
+
+// splitCmdStream 将服务器输出流分离到命令buffer和命令输出buffer
+func (p *DBParser) splitCmdStream(b []byte) {
+	if !p.inputInitial {
+		return
+	}
+	if p.inputState {
+		_, _ = p.cmdInputParser.WriteData(b)
+		return
+	}
+	_, _ = p.cmdOutputParser.WriteData(b)
+}
+
+// ParseServerOutput 解析服务器输出
+func (p *DBParser) ParseServerOutput(b []byte) []byte {
+	p.lock.Lock()
+	defer p.lock.Unlock()
+	p.splitCmdStream(b)
+	return b
+}
+
+// Close 关闭parser
+func (p *DBParser) Close() {
+	select {
+	case <-p.closed:
+		return
+	default:
+		close(p.closed)
+	}
+}
+
+func (p *DBParser) sendCommandRecord() {
+	if p.command != "" {
+		p.parseCmdOutput()
+		p.cmdRecordChan <- [2]string{p.command, p.output}
+		p.command = ""
+		p.output = ""
+	}
+}

pkg/proxy/dbproxy.go

+package proxy
+
+import (
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/jumpserver/koko/pkg/config"
+	"github.com/jumpserver/koko/pkg/i18n"
+	"github.com/jumpserver/koko/pkg/logger"
+	"github.com/jumpserver/koko/pkg/model"
+	"github.com/jumpserver/koko/pkg/service"
+	"github.com/jumpserver/koko/pkg/srvconn"
+	"github.com/jumpserver/koko/pkg/utils"
+)
+
+type DBProxyServer struct {
+	UserConn   UserConnection
+	User       *model.User
+	Database   *model.Database
+	SystemUser *model.SystemUser
+}
+
+func (p *DBProxyServer) getAuthOrManualSet() error {
+	needManualSet := false
+	if p.SystemUser.LoginMode == model.LoginModeManual {
+		needManualSet = true
+		logger.Debugf("Database %s login mode is: %s", p.Database.Name, model.LoginModeManual)
+	}
+	if p.SystemUser.Password == "" {
+		needManualSet = true
+		logger.Debugf("Database  %s neither has password", p.Database.Name)
+	}
+	if needManualSet {
+		term := utils.NewTerminal(p.UserConn, "password: ")
+		line, err := term.ReadPassword(fmt.Sprintf("%s's password: ", p.SystemUser.Username))
+		if err != nil {
+			logger.Errorf("Get password from user err %s", err.Error())
+			return err
+		}
+		p.SystemUser.Password = line
+		logger.Debug("Get password from user input: ", line)
+	}
+	return nil
+}
+
+func (p *DBProxyServer) getUsernameIfNeed() (err error) {
+	if p.SystemUser.Username == "" {
+		var username string
+		term := utils.NewTerminal(p.UserConn, "username: ")
+		for {
+			username, err = term.ReadLine()
+			if err != nil {
+				return err
+			}
+			username = strings.TrimSpace(username)
+			if username != "" {
+				break
+			}
+		}
+		p.SystemUser.Username = username
+		logger.Debug("Get username from user input: ", username)
+	}
+	return
+}
+
+func (p *DBProxyServer) checkProtocolMatch() bool {
+	return strings.ToLower(p.Database.DBType) == strings.ToLower(p.SystemUser.Protocol)
+}
+
+func (p *DBProxyServer) checkProtocolClientInstalled() bool {
+	switch strings.ToLower(p.Database.DBType) {
+	case "mysql":
+		return utils.IsInstalledMysqlClient()
+	}
+
+	return false
+}
+
+// validatePermission 检查是否有权限连接
+func (p *DBProxyServer) validatePermission() bool {
+	return true
+}
+
+// getSSHConn 获取ssh连接
+func (p *DBProxyServer) getMysqlConn() (srvConn *srvconn.ServerMysqlConnection, err error) {
+	srvConn = srvconn.NewMysqlServer(
+		srvconn.SqlHost(p.Database.Host),
+		srvconn.SqlPort(p.Database.Port),
+		srvconn.SqlUsername(p.SystemUser.Username),
+		srvconn.SqlPassword(p.SystemUser.Password),
+		srvconn.SqlDBName(p.Database.DBName),
+	)
+	err = srvConn.Connect()
+	return
+}
+
+// getServerConn 获取获取server连接
+func (p *DBProxyServer) getServerConn() (srvConn srvconn.ServerConnection, err error) {
+	done := make(chan struct{})
+	defer func() {
+		utils.IgnoreErrWriteString(p.UserConn, "\r\n")
+		close(done)
+	}()
+	go p.sendConnectingMsg(done, config.GetConf().SSHTimeout*time.Second)
+	return p.getMysqlConn()
+}
+
+// sendConnectingMsg 发送连接信息
+func (p *DBProxyServer) sendConnectingMsg(done chan struct{}, delayDuration time.Duration) {
+	delay := 0.0
+	msg := fmt.Sprintf(i18n.T("Database connecting to %s %.1f"), p.Database, delay)
+	utils.IgnoreErrWriteString(p.UserConn, msg)
+	for int(delay) < int(delayDuration/time.Second) {
+		select {
+		case <-done:
+			return
+		default:
+			delayS := fmt.Sprintf("%.1f", delay)
+			data := strings.Repeat("\x08", len(delayS)) + delayS
+			utils.IgnoreErrWriteString(p.UserConn, data)
+			time.Sleep(100 * time.Millisecond)
+			delay += 0.1
+		}
+	}
+}
+
+// preCheckRequisite 检查是否满足条件
+func (p *DBProxyServer) preCheckRequisite() (ok bool) {
+	if !p.checkProtocolMatch() {
+		msg := utils.WrapperWarn(i18n.T("System user <%s> and database <%s> protocol are inconsistent."))
+		msg = fmt.Sprintf(msg, p.SystemUser.Username, p.Database.DBType)
+		utils.IgnoreErrWriteString(p.UserConn, msg)
+		return
+	}
+	if !p.checkProtocolClientInstalled() {
+		msg := utils.WrapperWarn(i18n.T("Database %s protocol client not installed."))
+		msg = fmt.Sprintf(msg, p.Database.DBType)
+		utils.IgnoreErrWriteString(p.UserConn, msg)
+		return
+	}
+	if !p.validatePermission() {
+		msg := fmt.Sprintf("You don't have permission login %s", p.Database.Name)
+		utils.IgnoreErrWriteString(p.UserConn, msg)
+		return
+	}
+	if err := p.checkRequiredAuth(); err != nil {
+		msg := fmt.Sprintf("You get database %s auth info err: %s", p.Database.Name, err)
+		utils.IgnoreErrWriteString(p.UserConn, msg)
+		return
+	}
+	return true
+}
+
+func (p *DBProxyServer) checkRequiredAuth() error {
+	info := service.GetSystemUserDatabaseAuthInfo(p.SystemUser.ID)
+	p.SystemUser.Password = info.Password
+	if err := p.getUsernameIfNeed(); err != nil {
+		logger.Errorf("Get database %s auth username err: %s", p.Database.Name, err)
+		return err
+	}
+
+	if err := p.getAuthOrManualSet(); err != nil {
+		logger.Errorf("Get database %s auth password err: %s", p.Database.Name, err)
+		return err
+	}
+	return nil
+}
+
+// sendConnectErrorMsg 发送连接错误消息
+func (p *DBProxyServer) sendConnectErrorMsg(err error) {
+	msg := fmt.Sprintf("Connect database %s error: %s\r\n", p.Database.Host, err)
+	utils.IgnoreErrWriteString(p.UserConn, msg)
+	logger.Error(msg)
+}
+
+// Proxy 代理
+func (p *DBProxyServer) Proxy() {
+	if !p.preCheckRequisite() {
+		logger.Error("Check requisite failed")
+		return
+	}
+	// 创建Session
+	sw, err := CreateDBSession(p)
+	if err != nil {
+		logger.Error("Create database Session failed")
+		return
+	}
+	defer RemoveDBSession(sw)
+	srvConn, err := p.getServerConn()
+	// 连接后端服务器失败
+	if err != nil {
+		logger.Errorf("Create database server conn failed: %s", err)
+		p.sendConnectErrorMsg(err)
+		return
+	}
+	if err = sw.Bridge(p.UserConn, srvConn); err != nil {
+		logger.Errorf("DB Session %s bridge end: %s", sw.ID, err)
+	}
+
+}

pkg/proxy/dbswitch.go

+package proxy
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"strings"
+	"time"
+
+	uuid "github.com/satori/go.uuid"
+
+	"github.com/jumpserver/koko/pkg/common"
+	"github.com/jumpserver/koko/pkg/config"
+	"github.com/jumpserver/koko/pkg/i18n"
+	"github.com/jumpserver/koko/pkg/logger"
+	"github.com/jumpserver/koko/pkg/model"
+	"github.com/jumpserver/koko/pkg/srvconn"
+	"github.com/jumpserver/koko/pkg/utils"
+)
+
+type DBSwitchSession struct {
+	ID string
+	p  *DBProxyServer
+
+	DateStart string
+	DateEnd   string
+	finished  bool
+
+	MaxIdleTime time.Duration
+
+	ctx    context.Context
+	cancel context.CancelFunc
+}
+
+func (s *DBSwitchSession) Initial() {
+	s.ID = uuid.NewV4().String()
+	s.DateStart = common.CurrentUTCTime()
+	s.MaxIdleTime = config.GetConf().MaxIdleTime
+	s.ctx, s.cancel = context.WithCancel(context.Background())
+}
+
+func (s *DBSwitchSession) Terminate() {
+	select {
+	case <-s.ctx.Done():
+		return
+	default:
+	}
+	s.cancel()
+}
+
+func (s *DBSwitchSession) SessionID() string {
+	return s.ID
+}
+
+func (s *DBSwitchSession) recordCommand(cmdRecordChan chan [2]string) {
+	// 命令记录
+	cmdRecorder := NewCommandRecorder(s.ID)
+	for command := range cmdRecordChan {
+		if command[0] == "" {
+			continue
+		}
+		cmd := s.generateCommandResult(command)
+		cmdRecorder.Record(cmd)
+	}
+	// 关闭命令记录
+	cmdRecorder.End()
+}
+func (s *DBSwitchSession) generateCommandResult(command [2]string) *model.Command {
+	var input string
+	var output string
+	if len(command[0]) > 128 {
+		input = command[0][:128]
+	} else {
+		input = command[0]
+	}
+	i := strings.LastIndexByte(command[1], '\r')
+	if i <= 0 {
+		output = command[1]
+	} else if i > 0 && i < 1024 {
+		output = command[1][:i]
+	} else {
+		output = command[1][:1024]
+	}
+
+	return &model.Command{
+		SessionID:  s.ID,
+		OrgID:      s.p.Database.OrgID,
+		Input:      input,
+		Output:     output,
+		User:       fmt.Sprintf("%s (%s)", s.p.User.Name, s.p.User.Username),
+		Server:     s.p.Database.Name,
+		SystemUser: s.p.SystemUser.Username,
+		Timestamp:  time.Now().Unix(),
+	}
+}
+
+// postBridge 桥接结束以后执行操作
+func (s *DBSwitchSession) postBridge() {
+	s.DateEnd = common.CurrentUTCTime()
+	s.finished = true
+}
+
+// Bridge 桥接两个链接
+func (s *DBSwitchSession) Bridge(userConn UserConnection, srvConn srvconn.ServerConnection) (err error) {
+	var (
+		parser         DBParser
+		replayRecorder ReplyRecorder
+
+		userInChan chan []byte
+		srvInChan  chan []byte
+		done       chan struct{}
+	)
+	parser = newDBParser(s.ID)
+	replayRecorder = NewReplyRecord(s.ID)
+
+	userInChan = make(chan []byte, 1)
+	srvInChan = make(chan []byte, 1)
+	done = make(chan struct{})
+	userOutChan, srvOutChan := parser.ParseStream(userInChan, srvInChan)
+	defer func() {
+		close(done)
+		_ = userConn.Close()
+		_ = srvConn.Close()
+		// 关闭parser
+		parser.Close()
+		// 关闭录像
+		replayRecorder.End()
+		s.postBridge()
+	}()
+
+	go s.recordCommand(parser.cmdRecordChan)
+	go s.LoopReadFromSrv(done, srvConn, srvInChan)
+	go s.LoopReadFromUser(done, userConn, userInChan)
+	winCh := userConn.WinCh()
+	maxIdleTime := s.MaxIdleTime * time.Minute
+	lastActiveTime := time.Now()
+	tick := time.NewTicker(30 * time.Second)
+	defer tick.Stop()
+	for {
+		select {
+		// 检测是否超过最大空闲时间
+		case <-tick.C:
+			now := time.Now()
+			outTime := lastActiveTime.Add(maxIdleTime)
+			if !now.After(outTime) {
+				continue
+			}
+			msg := fmt.Sprintf(i18n.T("Database connect idle more than %d minutes, disconnect"), s.MaxIdleTime)
+			logger.Infof("DB Session idle more than %d minutes, disconnect: %s", s.MaxIdleTime, s.ID)
+			msg = utils.WrapperWarn(msg)
+			utils.IgnoreErrWriteString(userConn, "\n\r"+msg)
+			return
+		// 手动结束
+		case <-s.ctx.Done():
+			msg := i18n.T("Database connection terminated by administrator")
+			msg = utils.WrapperWarn(msg)
+			utils.IgnoreErrWriteString(userConn, "\n\r"+msg)
+			return
+		// 监控窗口大小变化
+		case win, ok := <-winCh:
+			if !ok {
+				return
+			}
+			_ = srvConn.SetWinSize(win.Height, win.Width)
+			logger.Debugf("Window server change: %d*%d", win.Height, win.Width)
+		// 经过parse处理的server数据，发给user
+		case p, ok := <-srvOutChan:
+			if !ok {
+				return
+			}
+			nw, _ := userConn.Write(p)
+			replayRecorder.Record(p[:nw])
+		// 经过parse处理的user数据，发给server
+		case p, ok := <-userOutChan:
+			if !ok {
+				return
+			}
+			_, err = srvConn.Write(p)
+		}
+		lastActiveTime = time.Now()
+	}
+}
+
+func (s *DBSwitchSession) MapData() map[string]interface{} {
+	var dataEnd interface{}
+	if s.DateEnd != "" {
+		dataEnd = s.DateEnd
+	}
+	return map[string]interface{}{
+		"id":             s.ID,
+		"user":           fmt.Sprintf("%s (%s)", s.p.User.Name, s.p.User.Username),
+		"asset":          s.p.Database.Name,
+		"org_id":         s.p.Database.OrgID,
+		"login_from":     s.p.UserConn.LoginFrom(),
+		"system_user":    s.p.SystemUser.Username,
+		"protocol":       s.p.SystemUser.Protocol,
+		"remote_addr":    s.p.UserConn.RemoteAddr(),
+		"is_finished":    s.finished,
+		"date_start":     s.DateStart,
+		"date_end":       dataEnd,
+		"user_id":        s.p.User.ID,
+		"asset_id":       s.p.Database.ID,
+		"system_user_id": s.p.SystemUser.ID,
+	}
+}
+
+func (s *DBSwitchSession) LoopReadFromUser(done chan struct{}, userConn UserConnection, inChan chan<- []byte) {
+	defer logger.Infof("DB Session %s: read from user done", s.ID)
+	s.LoopRead(done, userConn, inChan)
+}
+
+func (s *DBSwitchSession) LoopReadFromSrv(done chan struct{}, srvConn srvconn.ServerConnection, inChan chan<- []byte) {
+	defer logger.Infof("DB Session %s: read from srv done", s.ID)
+	s.LoopRead(done, srvConn, inChan)
+}
+
+func (s *DBSwitchSession) LoopRead(done chan struct{}, read io.Reader, inChan chan<- []byte) {
+loop:
+	for {
+		buf := make([]byte, 1024)
+		nr, err := read.Read(buf)
+		if nr > 0 {
+			select {
+			case <-done:
+				logger.Debug("DB session reader loop break done.")
+				break loop
+			case inChan <- buf[:nr]:
+			}
+		}
+		if err != nil {
+			break
+		}
+	}
+	close(inChan)
+}

pkg/proxy/sessmanager.go

@@ -12,9 +12,14 @@ import (
 	"github.com/jumpserver/koko/pkg/utils"
 )
 
-var sessionMap = make(map[string]*SwitchSession)
+var sessionMap = make(map[string]Session)
 var lock = new(sync.RWMutex)
 
+type Session interface {
+	SessionID() string
+	Terminate()
+}
+
 func HandleSessionTask(task model.TerminalTask) {
 	switch task.Name {
 	case "kill_session":
@@ -50,20 +55,23 @@ func RemoveSession(sw *SwitchSession) {
 	lock.Lock()
 	defer lock.Unlock()
 	delete(sessionMap, sw.ID)
-	finishSession(sw)
+	data := sw.MapData()
+	finishSession(data)
+	logger.Infof("Session %s has finished", sw.ID)
 }
 
-func AddSession(sw *SwitchSession) {
+func AddSession(sw Session) {
 	lock.Lock()
 	defer lock.Unlock()
-	sessionMap[sw.ID] = sw
+	sessionMap[sw.SessionID()] = sw
 }
 
 func CreateSession(p *ProxyServer) (sw *SwitchSession, err error) {
 	// 创建Session
 	sw = NewSwitchSession(p)
 	// Post到Api端
-	ok := postSession(sw)
+	data := sw.MapData()
+	ok := postSession(data)
 	msg := i18n.T("Connect with api server failed")
 	if !ok {
 		msg = utils.WrapperWarn(msg)
@@ -84,8 +92,7 @@ func CreateSession(p *ProxyServer) (sw *SwitchSession, err error) {
 	return
 }
 
-func postSession(s *SwitchSession) bool {
-	data := s.MapData()
+func postSession(data map[string]interface{}) bool {
 	for i := 0; i < 5; i++ {
 		if service.CreateSession(data) {
 			return true
@@ -95,8 +102,33 @@ func postSession(s *SwitchSession) bool {
 	return false
 }
 
-func finishSession(s *SwitchSession) {
-	data := s.MapData()
+func finishSession(data map[string]interface{}) {
 	service.FinishSession(data)
-	logger.Debugf("Session %s has finished", s.ID)
+}
+
+func CreateDBSession(p *DBProxyServer) (sw *DBSwitchSession, err error) {
+	// 创建Session
+	sw = &DBSwitchSession{
+		p: p,
+	}
+	sw.Initial()
+	data := sw.MapData()
+	ok := postSession(data)
+	msg := i18n.T("Create database session failed")
+	if !ok {
+		msg = utils.WrapperWarn(msg)
+		utils.IgnoreErrWriteString(p.UserConn, msg)
+		logger.Error(msg)
+		return sw, errors.New("create database session failed")
+	}
+	AddSession(sw)
+	return
+}
+
+func RemoveDBSession(sw *DBSwitchSession) {
+	lock.Lock()
+	defer lock.Unlock()
+	delete(sessionMap, sw.ID)
+	finishSession(sw.MapData())
+	logger.Infof("DB Session %s has finished", sw.ID)
 }

pkg/proxy/switch.go

@@ -57,6 +57,10 @@ func (s *SwitchSession) Terminate() {
 	s.cancel()
 }
 
+func (s *SwitchSession) SessionID() string {
+	return s.ID
+}
+
 func (s *SwitchSession) recordCommand(cmdRecordChan chan [2]string) {
 	// 命令记录
 	cmdRecorder := NewCommandRecorder(s.ID)

pkg/service/database.go

+package service
+
+import (
+	"fmt"
+
+	"github.com/jumpserver/koko/pkg/logger"
+	"github.com/jumpserver/koko/pkg/model"
+)
+
+func GetUserDatabases(uid string) (res []model.Database) {
+	Url := fmt.Sprintf(DatabaseAPPURL, uid)
+	_, err := authClient.Get(Url, &res)
+	if err != nil {
+		logger.Errorf("Get User databases err: %s", err)
+	}
+	return
+}
+
+func GetUserDatabaseSystemUsers(userID, assetID string) (sysUsers []model.SystemUser) {
+	Url := fmt.Sprintf(UserDatabaseSystemUsersURL, userID, assetID)
+	_, err := authClient.Get(Url, &sysUsers)
+	if err != nil {
+		logger.Error("Get user asset system users error: ", err)
+	}
+	return
+}
+
+func GetSystemUserDatabaseAuthInfo(systemUserID string) (info model.SystemUserAuthInfo) {
+	Url := fmt.Sprintf(SystemUserAuthURL, systemUserID)
+	_, err := authClient.Get(Url, &info)
+	if err != nil {
+		logger.Errorf("Get system user %s auth info failed", systemUserID)
+	}
+	return
+}

pkg/service/urls.go

 package service
 
 const (
-	UserAuthURL    = "/api/v1/authentication/auth/"                      // post 验证用户登陆
 	UserProfileURL = "/api/v1/users/profile/"                            // 获取当前用户的基本信息
 	UserListURL    = "/api/v1/users/users/"                              // 用户列表地址
 	UserDetailURL  = "/api/v1/users/users/%s/"                           // 获取用户信息
-	UserAuthOTPURL = "/api/v1/authentication/otp/auth/"                  // 验证OTP
 	TokenAssetURL  = "/api/v1/authentication/connection-token/?token=%s" // Token name
 
 	SystemUserAssetAuthURL          = "/api/v1/assets/system-users/%s/assets/%s/auth-info/" // 该系统用户对某资产的授权
@@ -41,8 +39,14 @@ const (
 
 // 1.5.5
 const (
-	UserTokenAuthURL = "/api/v1/authentication/tokens/" // 用户登录验证
+	UserTokenAuthURL   = "/api/v1/authentication/tokens/" // 用户登录验证
 	UserConfirmAuthURL = "/api/v1/authentication/login-confirm-ticket/status/"
 
 	NodeTreeWithAssetURL = "/api/v1/perms/users/%s/nodes/children-with-assets/tree/" // 资产树
+
+	DatabaseAPPURL = "/api/v1/perms/users/%s/database-apps/" //数据库app
+
+	UserDatabaseSystemUsersURL = "/api/v1/perms/users/%s/database-apps/%s/system-users/"
+
+	SystemUserAuthURL = "/api/v1/assets/system-users/%s/auth-info/"
 )

pkg/srvconn/mysqlconn.go

+package srvconn
+
+import (
+	"bytes"
+	"errors"
+	"fmt"
+	"os"
+	"os/exec"
+	"os/user"
+	"strconv"
+	"sync"
+	"syscall"
+	"time"
+
+	"github.com/creack/pty"
+
+	"github.com/jumpserver/koko/pkg/logger"
+)
+
+const (
+	mysqlPrompt = "Enter password: "
+)
+
+func NewMysqlServer(ops ...SqlOption) *ServerMysqlConnection {
+	args := &SqlOptions{
+		Username: os.Getenv("USER"),
+		Password: os.Getenv("PASSWORD"),
+		Host:     "127.0.0.1",
+		Port:     3306,
+		DBName:   "",
+	}
+	for _, setter := range ops {
+		setter(args)
+	}
+	return &ServerMysqlConnection{options: args, onceClose: new(sync.Once)}
+}
+
+type ServerMysqlConnection struct {
+	options   *SqlOptions
+	ptyFD     *os.File
+	onceClose *sync.Once
+	cmd       *exec.Cmd
+}
+
+func (dbconn *ServerMysqlConnection) Connect() (err error) {
+	dbconn.cmd = exec.Command("mysql", dbconn.options.CommandArgs()...)
+	nobody, err := user.Lookup("nobody")
+	if err != nil {
+		logger.Errorf("lookup nobody user err: %s", err)
+		return errors.New("nobody user does not exist")
+	}
+	dbconn.cmd.SysProcAttr = &syscall.SysProcAttr{}
+	uid, _ := strconv.Atoi(nobody.Uid)
+	gid, _ := strconv.Atoi(nobody.Gid)
+	dbconn.cmd.SysProcAttr.Credential = &syscall.Credential{Uid: uint32(uid), Gid: uint32(gid)}
+	dbconn.ptyFD, err = pty.Start(dbconn.cmd)
+	if err != nil {
+		logger.Errorf("pty start err: %s", err)
+		return fmt.Errorf("start local pty err: %s", err)
+	}
+	prompt := [len(mysqlPrompt)]byte{}
+	nr, err := dbconn.ptyFD.Read(prompt[:])
+	if err != nil {
+		_ = dbconn.ptyFD.Close()
+		_ = dbconn.cmd.Process.Kill()
+		logger.Errorf("read mysql pty local fd err: %s", err)
+		return fmt.Errorf("mysql conn err: %s", err)
+	}
+	if !bytes.Equal(prompt[:nr], []byte(mysqlPrompt)) {
+		_ = dbconn.cmd.Process.Kill()
+		_ = dbconn.ptyFD.Close()
+		logger.Errorf("mysql login prompt characters did not match: %s", prompt[:nr])
+		return errors.New("failed login mysql")
+	}
+	// 输入密码, 登录mysql
+	_, err = dbconn.ptyFD.Write([]byte(dbconn.options.Password + "\r\n"))
+	if err != nil {
+		_ = dbconn.ptyFD.Close()
+		_ = dbconn.cmd.Process.Kill()
+		logger.Errorf("mysql local pty write err: %s", err)
+		return fmt.Errorf("mysql conn err: %s", err)
+	}
+	logger.Infof("Connect mysql database %s success ", dbconn.options.Host)
+	return
+}
+
+func (dbconn *ServerMysqlConnection) Read(p []byte) (int, error) {
+	if dbconn.ptyFD == nil {
+		return 0, fmt.Errorf("not connect init")
+	}
+	return dbconn.ptyFD.Read(p)
+}
+
+func (dbconn *ServerMysqlConnection) Write(p []byte) (int, error) {
+	if dbconn.ptyFD == nil {
+		return 0, fmt.Errorf("not connect init")
+	}
+	return dbconn.ptyFD.Write(p)
+}
+
+func (dbconn *ServerMysqlConnection) SetWinSize(w, h int) error {
+	if dbconn.ptyFD == nil {
+		return fmt.Errorf("not connect init")
+	}
+	win := pty.Winsize{
+		Rows: uint16(h),
+		Cols: uint16(w),
+	}
+	return pty.Setsize(dbconn.ptyFD, &win)
+}
+
+func (dbconn *ServerMysqlConnection) Close() (err error) {
+	dbconn.onceClose.Do(func() {
+		if dbconn.ptyFD == nil {
+			return
+		}
+		_ = dbconn.ptyFD.Close()
+		err = dbconn.cmd.Process.Kill()
+	})
+	return
+}
+
+func (dbconn *ServerMysqlConnection) Timeout() time.Duration {
+	return time.Duration(10) * time.Second
+}
+
+func (dbconn *ServerMysqlConnection) Protocol() string {
+	return "mysql"
+}
+
+type SqlOptions struct {
+	Username string
+	Password string
+	DBName   string
+	Host     string
+	Port     int
+}
+
+func (opts *SqlOptions) CommandArgs() []string {
+	return []string{
+		fmt.Sprintf("--user=%s", opts.Username),
+		fmt.Sprintf("--host=%s", opts.Host),
+		fmt.Sprintf("--port=%d", opts.Port),
+		"--password",
+		opts.DBName,
+	}
+}
+
+type SqlOption func(*SqlOptions)
+
+func SqlUsername(username string) SqlOption {
+	return func(args *SqlOptions) {
+		args.Username = username
+	}
+}
+
+func SqlPassword(password string) SqlOption {
+	return func(args *SqlOptions) {
+		args.Password = password
+	}
+}
+
+func SqlDBName(dbName string) SqlOption {
+	return func(args *SqlOptions) {
+		args.DBName = dbName
+	}
+}
+
+func SqlHost(host string) SqlOption {
+	return func(args *SqlOptions) {
+		args.Host = host
+	}
+}
+
+func SqlPort(port int) SqlOption {
+	return func(args *SqlOptions) {
+		args.Port = port
+	}
+}

pkg/utils/database.go

+package utils
+
+import (
+	"os/exec"
+	"os/user"
+)
+
+func IsInstalledMysqlClient() bool {
+	if mysqlPath, err := exec.LookPath("mysql"); err == nil {
+		cmd := exec.Command(mysqlPath, "-V")
+		if err = cmd.Start(); err == nil {
+			_ = cmd.Process.Kill()
+			return true
+		}
+	}
+	return false
+}
+
+func IsUserExist(username string) bool {
+	if _, err := user.Lookup(username); err == nil {
+		return true
+	}
+	return false
+}