Merge pull request #215 from jumpserver/dev

v3.3.0

main.go

@@ -22,6 +22,7 @@ import (
 	"lion/pkg/config"
 	"lion/pkg/jms-sdk-go/model"
 	"lion/pkg/jms-sdk-go/service"
+	"lion/pkg/jms-sdk-go/service/videoworker"
 	"lion/pkg/logger"
 	"lion/pkg/middleware"
 	"lion/pkg/session"
@@ -58,6 +59,7 @@ func main() {
 	config.Setup(configPath)
 	logger.SetupLogger(config.GlobalConfig)
 	jmsService := MustJMService()
+	videoWorkerClient := NewWorkerClient(*config.GlobalConfig)
 	bootstrap(jmsService)
 	tunnelService := tunnel.GuacamoleTunnelServer{
 		Cache: &tunnel.GuaTunnelCacheManager{
@@ -66,8 +68,9 @@ func main() {
 		SessCache: &tunnel.SessionCache{
 			Sessions: make(map[string]*session.TunnelSession),
 		},
-		JmsService:     jmsService,
-		SessionService: &session.Server{JmsService: jmsService},
+		JmsService: jmsService,
+		SessionService: &session.Server{JmsService: jmsService,
+			VideoWorkerClient: videoWorkerClient},
 	}
 	eng := registerRouter(jmsService, &tunnelService)
 	go runHeartTask(jmsService, tunnelService.Cache)
@@ -453,3 +456,79 @@ func MustValidKey(key model.AccessKey) model.AccessKey {
 	os.Exit(1)
 	return key
 }
+
+func NewWorkerClient(cfg config.Config) *videoworker.Client {
+	if !cfg.EnableVideoWorker {
+		return nil
+	}
+	workerURL := cfg.VideoWorkerHost
+	var key model.AccessKey
+	if err := key.LoadFromFile(cfg.AccessKeyFilePath); err != nil {
+		logger.Errorf("Create video worker client failed: loading access key err %s", err)
+		return nil
+	}
+	workClient := videoworker.NewClient(workerURL, key, cfg.IgnoreVerifyCerts)
+	if workClient == nil {
+		logger.Errorf("Create video worker client failed: worker url %s", workerURL)
+		return nil
+	}
+	go KeepWsConnect(workClient)
+	return workClient
+}
+
+func KeepWsConnect(s *videoworker.Client) {
+	if err := s.Login(); err != nil {
+		logger.Errorf("Worker Ws client login failed: %s, try next 10s", err)
+		time.Sleep(10 * time.Second)
+		go KeepWsConnect(s)
+		return
+	}
+	wsCon, err := s.GetWsClient()
+	if err != nil {
+		time.Sleep(10 * time.Second)
+		go KeepWsConnect(s)
+		return
+	}
+	defer wsCon.Close()
+	logger.Info("Start worker ws client beat success")
+	done := make(chan struct{}, 2)
+	go func() {
+		defer close(done)
+		for {
+			msgType, message, err2 := wsCon.ReadMessage()
+			if err2 != nil {
+				logger.Errorf("Worker Ws client read err: %s", err2)
+				return
+			}
+			switch msgType {
+			case websocket.PingMessage,
+				websocket.PongMessage:
+				logger.Debug("Worker Ws client ping/pong Message")
+				continue
+			case websocket.CloseMessage:
+				logger.Debug("Worker Ws client close Message")
+				return
+			}
+			logger.Debugf("Worker Ws client read message: %s", message)
+		}
+	}()
+	beatTicker := time.NewTicker(time.Second * 30)
+	defer beatTicker.Stop()
+	pingEvent := map[string]string{
+		"event": "ping",
+	}
+	for {
+		select {
+		case <-done:
+			logger.Error("Worker Ws heart beat closed, try reconnect after 10s")
+			time.Sleep(10 * time.Second)
+			go KeepWsConnect(s)
+			return
+		case <-beatTicker.C:
+			if err1 := wsCon.WriteJSON(pingEvent); err1 != nil {
+				logger.Errorf("Ws client write stat data failed: %s", err1)
+				continue
+			}
+		}
+	}
+}

pkg/config/config.go

@@ -44,6 +44,10 @@ type Config struct {
 	RedisSentinelPassword string `mapstructure:"REDIS_SENTINEL_PASSWORD"`
 	RedisSentinelHosts    string `mapstructure:"REDIS_SENTINEL_HOSTS"`
 	RedisUseSSL           bool   `mapstructure:"REDIS_USE_SSL"`
+
+	EnableVideoWorker bool   `mapstructure:"ENABLE_VIDEO_WORKER"`
+	VideoWorkerHost   string `mapstructure:"VIDEO_WORKER_HOST"`
+	IgnoreVerifyCerts bool   `mapstructure:"IGNORE_VERIFY_CERTS"`
 }
 
 func Setup(configPath string) {

pkg/guacd/information.go

@@ -55,4 +55,17 @@ type ClientInformation struct {
 	 * The timezone reported by the client.
 	 */
 	Timezone string
+
+	/**
+	 * qwerty keyboard layout
+	 */
+	KeyboardLayout string
+}
+
+func (info *ClientInformation) ExtraConfig() map[string]string {
+	ret := make(map[string]string)
+	if layout, ok := RDPServerLayouts[info.KeyboardLayout]; ok {
+		ret[RDPServerLayout] = layout
+	}
+	return ret
 }

pkg/guacd/qwertz.go

@@ -0,0 +1,54 @@
+package guacd
+
+var RDPServerLayouts = map[string]string{
+	"de-de-qwertz":    "de-de-qwertz",
+	"de-ch-qwertz":    "de-ch-qwertz",
+	"en-gb-qwerty":    "en-gb-qwerty",
+	"en-us-qwerty":    "en-us-qwerty",
+	"es-es-qwerty":    "es-es-qwerty",
+	"es-latam-qwerty": "es-latam-qwerty",
+	"failsafe":        "failsafe",
+	"fr-be-azerty":    "fr-be-azerty",
+	"fr-fr-azerty":    "fr-fr-azerty",
+	"fr-ca-qwerty":    "fr-ca-qwerty",
+	"fr-ch-qwertz":    "fr-ch-qwertz",
+	"hu-hu-qwertz":    "hu-hu-qwertz",
+	"it-it-qwerty":    "it-it-qwerty",
+	"ja-jp-qwerty":    "ja-jp-qwerty",
+	"no-no-qwerty":    "no-no-qwerty",
+	"pl-pl-qwerty":    "pl-pl-qwerty",
+	"pt-br-qwerty":    "pt-br-qwerty",
+	"pt-pt-qwerty":    "pt-pt-qwerty",
+	"ro-ro-qwerty":    "ro-ro-qwerty",
+	"sv-se-qwerty":    "sv-se-qwerty",
+	"da-dk-qwerty":    "da-dk-qwerty",
+	"tr-tr-qwerty":    "tr-tr-qwerty",
+}
+
+/*
+https://github.com/apache/guacamole-client/blob/fe6677bf4ebaa8662418013ab1af8c7060224ef5/guacamole/src/main/frontend/src/translations/zh.json
+
+   "FIELD_OPTION_SERVER_LAYOUT_DE_CH_QWERTZ" : "Swiss German (Qwertz)",
+   "FIELD_OPTION_SERVER_LAYOUT_DE_DE_QWERTZ" : "German (Qwertz)",
+   "FIELD_OPTION_SERVER_LAYOUT_EMPTY"        : "",
+   "FIELD_OPTION_SERVER_LAYOUT_EN_GB_QWERTY" : "UK English (Qwerty)",
+   "FIELD_OPTION_SERVER_LAYOUT_EN_US_QWERTY" : "US English (Qwerty)",
+   "FIELD_OPTION_SERVER_LAYOUT_ES_ES_QWERTY" : "Spanish (Qwerty)",
+   "FIELD_OPTION_SERVER_LAYOUT_ES_LATAM_QWERTY" : "Latin American (Qwerty)",
+   "FIELD_OPTION_SERVER_LAYOUT_FAILSAFE"     : "Unicode",
+   "FIELD_OPTION_SERVER_LAYOUT_FR_BE_AZERTY" : "Belgian French (Azerty)",
+   "FIELD_OPTION_SERVER_LAYOUT_FR_CA_QWERTY" : "Canadian French (Qwerty)",
+   "FIELD_OPTION_SERVER_LAYOUT_FR_CH_QWERTZ" : "Swiss French (Qwertz)",
+   "FIELD_OPTION_SERVER_LAYOUT_FR_FR_AZERTY" : "French (Azerty)",
+   "FIELD_OPTION_SERVER_LAYOUT_HU_HU_QWERTZ" : "Hungarian (Qwertz)",
+   "FIELD_OPTION_SERVER_LAYOUT_IT_IT_QWERTY" : "Italian (Qwerty)",
+   "FIELD_OPTION_SERVER_LAYOUT_JA_JP_QWERTY" : "Japanese (Qwerty)",
+   "FIELD_OPTION_SERVER_LAYOUT_NO_NO_QWERTY" : "Norwegian (Qwerty)",
+   "FIELD_OPTION_SERVER_LAYOUT_PL_PL_QWERTY" : "Polish (Qwerty)",
+   "FIELD_OPTION_SERVER_LAYOUT_PT_BR_QWERTY" : "Portuguese Brazilian (Qwerty)",
+   "FIELD_OPTION_SERVER_LAYOUT_PT_PT_QWERTY" : "Portuguese (Qwerty)",
+   "FIELD_OPTION_SERVER_LAYOUT_RO_RO_QWERTY" : "Romanian (Qwerty)",
+   "FIELD_OPTION_SERVER_LAYOUT_SV_SE_QWERTY" : "Swedish (Qwerty)",
+   "FIELD_OPTION_SERVER_LAYOUT_DA_DK_QWERTY" : "Danish (Qwerty)",
+   "FIELD_OPTION_SERVER_LAYOUT_TR_TR_QWERTY" : "Turkish-Q (Qwerty)",
+*/

pkg/jms-sdk-go/httplib/client.go

@@ -22,23 +22,40 @@ type AuthSign interface {
 
 const miniTimeout = time.Second * 30
 
-func NewClient(baseUrl string, timeout time.Duration) (*Client, error) {
+type Opt func(*opt)
+
+type opt struct {
+	insecure bool
+}
+
+func WithInsecure() Opt {
+	return func(c *opt) {
+		c.insecure = true
+	}
+}
+
+func NewClient(baseUrl string, timeout time.Duration, settings ...Opt) (*Client, error) {
 	_, err := url.Parse(baseUrl)
 	if err != nil {
 		return nil, err
 	}
 	if timeout < miniTimeout {
 		timeout = miniTimeout
 	}
-
+	var cfg opt
+	for _, setter := range settings {
+		setter(&cfg)
+	}
 	jar := &customCookieJar{
 		data: map[string]string{},
 	}
 	con := http.Client{
-		Timeout: timeout,
-		Jar:     jar,
+		Timeout:   timeout,
+		Jar:       jar,
+		Transport: NewTransport(cfg.insecure),
 	}
 	return &Client{
+		Jar:     jar,
 		Timeout: timeout,
 		baseUrl: baseUrl,
 		cookies: make(map[string]string),
@@ -54,6 +71,7 @@ type Client struct {
 	headers  map[string]string
 	http     *http.Client
 	authSign AuthSign
+	Jar      http.CookieJar
 }
 
 func (c *Client) Clone() Client {
@@ -213,6 +231,11 @@ func (c *Client) Patch(reqUrl string, data interface{}, res interface{}, params
 }
 
 func (c *Client) UploadFile(reqUrl string, gFile string, res interface{}, params ...map[string]string) (err error) {
+	reqUrl = c.parseQueryUrl(reqUrl, params)
+	return c.PostFileWithFields(reqUrl, gFile, nil, res)
+}
+
+func (c *Client) PostFileWithFields(reqUrl string, gFile string, fields map[string]string, res interface{}) error {
 	fd, err := os.Open(gFile)
 	if err != nil {
 		return err
@@ -223,8 +246,19 @@ func (c *Client) UploadFile(reqUrl string, gFile string, res interface{}, params
 	if err != nil {
 		return err
 	}
-	contentType, contentLen, bodyReader := getFileMultipartBodyReader("file", fd.Name(), fi.Size(), bufferFd)
-	reqUrl = c.parseUrl(reqUrl, params)
+	var size = fi.Size()
+	startPartBuf := bytes.NewBufferString("")
+	partWriter := multipart.NewWriter(startPartBuf)
+	for name, value := range fields {
+		_ = partWriter.WriteField(name, value)
+	}
+	_, _ = partWriter.CreateFormFile("file", fi.Name())
+	boundary := partWriter.Boundary()
+	endString := fmt.Sprintf("\r\n--%s--\r\n", boundary)
+	endPartBuf := bytes.NewBufferString(endString)
+	bodyReader := io.MultiReader(startPartBuf, bufferFd, endPartBuf)
+	contentLen := int64(startPartBuf.Len()) + size + int64(endPartBuf.Len())
+	reqUrl = c.parseUrl(reqUrl, nil)
 	req, err := http.NewRequest(http.MethodPost, reqUrl, bodyReader)
 	if err != nil {
 		return err
@@ -233,7 +267,8 @@ func (c *Client) UploadFile(reqUrl string, gFile string, res interface{}, params
 		return err
 	}
 	req.ContentLength = contentLen
-	req.Header.Set("Content-Type", contentType)
+	req.Header.Set("Content-Type", partWriter.FormDataContentType())
+
 	client := http.Client{
 		Jar: c.http.Jar,
 	}
@@ -242,45 +277,33 @@ func (c *Client) UploadFile(reqUrl string, gFile string, res interface{}, params
 		return err
 	}
 	defer resp.Body.Close()
-	body, err := ioutil.ReadAll(resp.Body)
-	if resp.StatusCode >= 400 {
-		msg := fmt.Sprintf("%s %s failed, get code: %d, %s", req.Method, req.URL, resp.StatusCode, string(body))
-		err = errors.New(msg)
-		return
-	}
+	return c.handleResp(resp, res)
+}
 
+func (c *Client) handleResp(resp *http.Response, res interface{}) (err error) {
+	req := resp.Request
 	// If is buffer return the raw response body
 	if buf, ok := res.(*bytes.Buffer); ok {
-		buf.Write(body)
-		return
+		_, err = buf.ReadFrom(resp.Body)
+		return err
 	}
-	// Unmarshal response body to result struct
 	if res != nil {
 		switch {
 		case strings.Contains(resp.Header.Get("Content-Type"), "application/json"):
-			err = json.Unmarshal(body, res)
+			err = json.NewDecoder(resp.Body).Decode(res)
 			if err != nil {
-				msg := fmt.Sprintf("%s %s failed, unmarshal '%s' response failed: %s", req.Method, req.URL, body, err)
-				err = errors.New(msg)
-				return
+				msg := fmt.Sprintf("%s %s failed, json unmarshal failed: %s", req.Method, req.URL, err)
+				return fmt.Errorf("%w: %s", err, msg)
 			}
 		}
 	}
-	return
-}
-
-func getFileMultipartBodyReader(field, filename string, size int64,
-	reader io.Reader) (contentType string, contentLen int64, bodyReader io.Reader) {
-	startPartBuf := bytes.NewBufferString("")
-	bodyWriter := multipart.NewWriter(startPartBuf)
-	// use the body_writer to write the Part headers to the buffer
-	_, _ = bodyWriter.CreateFormFile(field, filename)
-	boundary := bodyWriter.Boundary()
-	endString := fmt.Sprintf("\r\n--%s--\r\n", boundary)
-	endPartBuf := bytes.NewBufferString(endString)
-
-	bodyReader = io.MultiReader(startPartBuf, reader, endPartBuf)
-	contentLen = int64(startPartBuf.Len()) + size + int64(endPartBuf.Len())
-	contentType = bodyWriter.FormDataContentType()
-	return
+	if resp.StatusCode >= 400 {
+		var buf bytes.Buffer
+		_, _ = buf.ReadFrom(resp.Body)
+		msg := fmt.Sprintf("%s %s failed, get code: %d %s",
+			req.Method, req.URL, resp.StatusCode, buf.String())
+		err = errors.New(msg)
+		return
+	}
+	return nil
 }

pkg/jms-sdk-go/httplib/transport.go

@@ -0,0 +1,16 @@
+package httplib
+
+import (
+	"crypto/tls"
+	"net/http"
+)
+
+// 创建 Transport 支持使用不安全的 https
+
+func NewTransport(insecure bool) http.RoundTripper {
+	transport := http.DefaultTransport.(*http.Transport).Clone()
+	if insecure {
+		transport.TLSClientConfig = &tls.Config{InsecureSkipVerify: true}
+	}
+	return transport
+}