frp/src/cmd/frps/control.go

// Copyright 2016 fatedier, fatedier@gmail.com
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package main

import (
	"encoding/json"
	"fmt"
	"io"
	"strings"
	"time"

	"github.com/fatedier/frp/src/models/consts"
	"github.com/fatedier/frp/src/models/metric"
	"github.com/fatedier/frp/src/models/msg"
	"github.com/fatedier/frp/src/models/server"
	"github.com/fatedier/frp/src/utils/conn"
	"github.com/fatedier/frp/src/utils/log"
	"github.com/fatedier/frp/src/utils/pcrypto"
)

func ProcessControlConn(l *conn.Listener) {
	for {
		c, err := l.Accept()
		if err != nil {
			return
		}
		log.Debug("Get new connection, %v", c.GetRemoteAddr())
		go controlWorker(c)
	}
}

// connection from every client and server
func controlWorker(c *conn.Conn) {
	// if login message type is NewWorkConn, don't close this connection
	var closeFlag bool = true
	var s *server.ProxyServer
	defer func() {
		if closeFlag {
			c.Close()
			if s != nil {
				s.Close()
			}
		}
	}()

	// get login message
	buf, err := c.ReadLine()
	if err != nil {
		log.Warn("Read error, %v", err)
		return
	}
	log.Debug("Get msg from frpc: %s", buf)

	cliReq := &msg.ControlReq{}
	if err := json.Unmarshal([]byte(buf), &cliReq); err != nil {
		log.Warn("Parse msg from frpc error: %v : %s", err, buf)
		return
	}

	// login when type is NewCtlConn or NewWorkConn
	ret, info := doLogin(cliReq, c)
	// if login type is NewWorkConn, nothing will be send to frpc
	if cliReq.Type == consts.NewCtlConn {
		cliRes := &msg.ControlRes{
			Type: consts.NewCtlConnRes,
			Code: ret,
			Msg:  info,
		}
		byteBuf, _ := json.Marshal(cliRes)
		err = c.WriteString(string(byteBuf) + "\n")
		if err != nil {
			log.Warn("ProxyName [%s], write to client error, proxy exit", cliReq.ProxyName)
			return
		}
	} else {
		closeFlag = false
		return
	}

	// if login failed, just return
	if ret > 0 {
		return
	}

	s, ok := server.GetProxyServer(cliReq.ProxyName)
	if !ok {
		log.Warn("ProxyName [%s] does not exist now", cliReq.ProxyName)
		return
	}

	// create a channel for sending messages
	msgSendChan := make(chan interface{}, 1024)
	go msgSender(s, c, msgSendChan)
	go noticeUserConn(s, msgSendChan)

	// loop for reading control messages from frpc and deal with different types
	msgReader(s, c, msgSendChan)

	close(msgSendChan)
	log.Info("ProxyName [%s], I'm dead!", s.Name)
	return
}

// when frps get one new user connection, send NoticeUserConn message to frpc and accept one new WorkConn later
func noticeUserConn(s *server.ProxyServer, msgSendChan chan interface{}) {
	for {
		closeFlag := s.WaitUserConn()
		if closeFlag {
			log.Debug("ProxyName [%s], goroutine for noticing user conn is closed", s.Name)
			break
		}
		notice := &msg.ControlRes{
			Type: consts.NoticeUserConn,
		}
		msgSendChan <- notice
		log.Debug("ProxyName [%s], notice client to add work conn", s.Name)
	}
}

// loop for reading messages from frpc after control connection is established
func msgReader(s *server.ProxyServer, c *conn.Conn, msgSendChan chan interface{}) error {
	// for heartbeat
	var heartbeatTimeout bool = false
	timer := time.AfterFunc(time.Duration(server.HeartBeatTimeout)*time.Second, func() {
		heartbeatTimeout = true
		s.Close()
		log.Error("ProxyName [%s], client heartbeat timeout", s.Name)
	})
	defer timer.Stop()

	for {
		buf, err := c.ReadLine()
		if err != nil {
			if err == io.EOF {
				log.Warn("ProxyName [%s], client is dead!", s.Name)
				s.Close()
				return err
			} else if c == nil || c.IsClosed() {
				log.Warn("ProxyName [%s], client connection is closed", s.Name)
				s.Close()
				return err
			}
			log.Warn("ProxyName [%s], read error: %v", s.Name, err)
			continue
		}

		cliReq := &msg.ControlReq{}
		if err := json.Unmarshal([]byte(buf), &cliReq); err != nil {
			log.Warn("ProxyName [%s], parse msg from frpc error: %v : %s", s.Name, err, buf)
			continue
		}

		switch cliReq.Type {
		case consts.HeartbeatReq:
			log.Debug("ProxyName [%s], get heartbeat", s.Name)
			timer.Reset(time.Duration(server.HeartBeatTimeout) * time.Second)
			heartbeatRes := &msg.ControlRes{
				Type: consts.HeartbeatRes,
			}
			msgSendChan <- heartbeatRes
		default:
			log.Warn("ProxyName [%s}, unsupport msgType [%d]", s.Name, cliReq.Type)
		}
	}
	return nil
}

// loop for sending messages from channel to frpc
func msgSender(s *server.ProxyServer, c *conn.Conn, msgSendChan chan interface{}) {
	for {
		msg, ok := <-msgSendChan
		if !ok {
			break
		}

		buf, _ := json.Marshal(msg)
		err := c.WriteString(string(buf) + "\n")
		if err != nil {
			log.Warn("ProxyName [%s], write to client error, proxy exit", s.Name)
			s.Close()
			break
		}
	}
}

// if success, ret equals 0, otherwise greater than 0
// NewCtlConn
// NewWorkConn
// NewWorkConnUdp
func doLogin(req *msg.ControlReq, c *conn.Conn) (ret int64, info string) {
	ret = 1
	// check if PrivilegeMode is enabled
	if req.PrivilegeMode && !server.PrivilegeMode {
		info = fmt.Sprintf("ProxyName [%s], PrivilegeMode is disabled in frps", req.ProxyName)
		log.Warn("info")
		return
	}

	var (
		s  *server.ProxyServer
		ok bool
	)
	s, ok = server.GetProxyServer(req.ProxyName)
	if req.PrivilegeMode && req.Type == consts.NewCtlConn {
		log.Debug("ProxyName [%s], doLogin and privilege mode is enabled", req.ProxyName)
	} else {
		if !ok {
			info = fmt.Sprintf("ProxyName [%s] is not exist", req.ProxyName)
			log.Warn(info)
			return
		}
	}

	// check authKey or privilegeKey
	nowTime := time.Now().Unix()
	if req.PrivilegeMode {
		privilegeKey := pcrypto.GetAuthKey(req.ProxyName + server.PrivilegeToken + fmt.Sprintf("%d", req.Timestamp))
		// privilegeKey unavaiable after server.AuthTimeout minutes
		if server.AuthTimeout != 0 && nowTime-req.Timestamp > server.AuthTimeout {
			info = fmt.Sprintf("ProxyName [%s], privilege mode authorization timeout", req.ProxyName)
			log.Warn(info)
			return
		} else if req.PrivilegeKey != privilegeKey {
			info = fmt.Sprintf("ProxyName [%s], privilege mode authorization failed", req.ProxyName)
			log.Warn(info)
			log.Debug("PrivilegeKey [%s] and get [%s]", privilegeKey, req.PrivilegeKey)
			return
		}
	} else {
		authKey := pcrypto.GetAuthKey(req.ProxyName + s.AuthToken + fmt.Sprintf("%d", req.Timestamp))
		if server.AuthTimeout != 0 && nowTime-req.Timestamp > server.AuthTimeout {
			info = fmt.Sprintf("ProxyName [%s], authorization timeout", req.ProxyName)
			log.Warn(info)
			return
		} else if req.AuthKey != authKey {
			info = fmt.Sprintf("ProxyName [%s], authorization failed", req.ProxyName)
			log.Warn(info)
			log.Debug("AuthKey [%s] and get [%s]", authKey, req.AuthKey)
			return
		}
	}

	// control conn
	if req.Type == consts.NewCtlConn {
		if req.PrivilegeMode {
			s = server.NewProxyServerFromCtlMsg(req)
			// we check listen_port if privilege_allow_ports are set
			// and PrivilegeMode is enabled
			if s.Type == "tcp" {
				if len(server.PrivilegeAllowPorts) != 0 {
					_, ok := server.PrivilegeAllowPorts[s.ListenPort]
					if !ok {
						info = fmt.Sprintf("ProxyName [%s], remote_port [%d] isn't allowed", req.ProxyName, s.ListenPort)
						log.Warn(info)
						return
					}
				}
			} else if s.Type == "http" || s.Type == "https" {
				for _, domain := range s.CustomDomains {
					if server.SubDomainHost != "" && strings.Contains(domain, server.SubDomainHost) {
						info = fmt.Sprintf("ProxyName [%s], custom domain [%s] should not belong to subdomain_host [%s]", req.ProxyName, domain, server.SubDomainHost)
						log.Warn(info)
						return
					}
				}
			}
			err := server.CreateProxy(s)
			if err != nil {
				info = fmt.Sprintf("ProxyName [%s], %v", req.ProxyName, err)
				log.Warn(info)
				return
			}
		}

		// check if vhost_port is set
		if s.Type == "http" && server.VhostHttpMuxer == nil {
			info = fmt.Sprintf("ProxyName [%s], type [http] not support when vhost_http_port is not set", req.ProxyName)
			log.Warn(info)
			return
		}
		if s.Type == "https" && server.VhostHttpsMuxer == nil {
			info = fmt.Sprintf("ProxyName [%s], type [https] not support when vhost_https_port is not set", req.ProxyName)
			log.Warn(info)
			return
		}

		// set infomations from frpc
		s.BindAddr = server.BindAddr
		s.UseEncryption = req.UseEncryption
		s.UseGzip = req.UseGzip
		s.HostHeaderRewrite = req.HostHeaderRewrite
		s.HttpUserName = req.HttpUserName
		s.HttpPassWord = req.HttpPassWord

		// package URL
		if req.SubDomain != "" {
			if strings.Contains(req.SubDomain, ".") || strings.Contains(req.SubDomain, "*") {
				info = fmt.Sprintf("ProxyName [%s], '.' or '*' is not supported in subdomain", req.ProxyName)
				log.Warn(info)
				return
			}
			if server.SubDomainHost == "" {
				info = fmt.Sprintf("ProxyName [%s], subdomain in not supported because this feature is not enabled by remote server", req.ProxyName)
				log.Warn(info)
				return
			}
			s.SubDomain = req.SubDomain + "." + server.SubDomainHost
		}
		if req.PoolCount > server.MaxPoolCount {
			s.PoolCount = server.MaxPoolCount
		} else if req.PoolCount < 0 {
			s.PoolCount = 0
		} else {
			s.PoolCount = req.PoolCount
		}

		if s.Status == consts.Working {
			info = fmt.Sprintf("ProxyName [%s], already in use", req.ProxyName)
			log.Warn(info)
			return
		}

		// update metric's proxy status
		metric.SetProxyInfo(s.Name, s.Type, s.BindAddr, s.UseEncryption, s.UseGzip, s.PrivilegeMode, s.CustomDomains, s.Locations, s.ListenPort)

		// start proxy and listen for user connections, no block
		err := s.Start(c)
		if err != nil {
			info = fmt.Sprintf("ProxyName [%s], start proxy error: %v", req.ProxyName, err)
			log.Warn(info)
			return
		}
		log.Info("ProxyName [%s], start proxy success", req.ProxyName)
		if req.PrivilegeMode {
			log.Info("ProxyName [%s], created by PrivilegeMode", req.ProxyName)
		}
	} else if req.Type == consts.NewWorkConn {
		// work conn
		if s.Status != consts.Working {
			log.Warn("ProxyName [%s], is not working when it gets one new work connnection", req.ProxyName)
			return
		}
		// the connection will close after join over
		s.RegisterNewWorkConn(c)
	} else if req.Type == consts.NewWorkConnUdp {
		// work conn for udp
		if s.Status != consts.Working {
			log.Warn("ProxyName [%s], is not working when it gets one new work connnection for udp", req.ProxyName)
			return
		}
		s.RegisterNewWorkConnUdp(c)
	} else {
		info = fmt.Sprintf("Unsupport login message type [%d]", req.Type)
		log.Warn("Unsupport login message type [%d]", req.Type)
		return
	}

	ret = 0
	return
}