frp 源码阅读与分析(二)：TCP内网穿透的实现
在 上一篇 文章中，我们介绍了frp中的一些 概念和基础知识，这一篇中，我们在此前的基础之上，来看看frp是怎么实现TCP内网穿透的。
我们知道，要使用frp，必须有个服务端，然后要有个客户端。因此，我们从这里开始入手。
可以参考 《如何阅读源代码》：https://jiajunhuang.com/articles/2018_08_04-how_to_read_source_code.md.html
frps
cmd/frps/main.go 是frps的入口处，我们从这里开始，main 函数的主体：
func main() {
crypto.DefaultSalt = "frp"
Execute()
}
因此我们需要看到 Execute() 函数的内容，其实它是使用了 cobra 这个库，所以实际的入口在
var rootCmd = &cobra.Command{
Use: "frps",
Short: "frps is the server of frp (https://github.com/fatedier/frp)",
RunE: func(cmd *cobra.Command, args []string) error {
if showVersion {
fmt.Println(version.Full())
return nil
}
var err error
if cfgFile != "" {
var content string
content, err = config.GetRenderedConfFromFile(cfgFile)
if err != nil {
return err
}
g.GlbServerCfg.CfgFile = cfgFile
err = parseServerCommonCfg(CfgFileTypeIni, content)
} else {
err = parseServerCommonCfg(CfgFileTypeCmd, "")
}
if err != nil {
return err
}
err = runServer()
if err != nil {
fmt.Println(err)
os.Exit(1)
}
return nil
},
}
最终，也就是 runServer() 这个函数：
func runServer() (err error) {
log.InitLog(g.GlbServerCfg.LogWay, g.GlbServerCfg.LogFile, g.GlbServerCfg.LogLevel,
g.GlbServerCfg.LogMaxDays)
svr, err := server.NewService()
if err != nil {
return err
}
log.Info("Start frps success")
server.ServerService = svr
svr.Run()
return
}
svr.Run()，其中的 svr 是来自 server.NewService()，仔细看一下，server.NewService() 其实就是初始化了一大堆东西。
我们直接看 svr.Run() 做了什么：
func (svr *Service) Run() {
if svr.rc.NatHoleController != nil {
go svr.rc.NatHoleController.Run()
}
if g.GlbServerCfg.KcpBindPort > 0 {
go svr.HandleListener(svr.kcpListener)
}
go svr.HandleListener(svr.websocketListener)
go svr.HandleListener(svr.tlsListener)
svr.HandleListener(svr.listener)
}
可以看到，最后frps会执行到 svr.HandleListener(svr.listener)，前面的都是什么 nat hole punching, kcp, websocket, tls等等，我们不看。直接看tcp。
func (svr *Service) HandleListener(l frpNet.Listener) {
// Listen for incoming connections from client.
for {
c, err := l.Accept()
if err != nil {
log.Warn("Listener for incoming connections from client closed")
return
}
c = frpNet.CheckAndEnableTLSServerConn(c, svr.tlsConfig)
// Start a new goroutine for dealing connections.
go func(frpConn frpNet.Conn) {
...
}
}
}
这里就是监听之后，每来一个新的连接，就起一个goroutine去处理，也就是 go func()... 这一段，然后我们看看内容：
switch m := rawMsg.(type) {
case *msg.Login:
err = svr.RegisterControl(conn, m)
// If login failed, send error message there.
// Otherwise send success message in control's work goroutine.
if err != nil {
conn.Warn("%v", err)
msg.WriteMsg(conn, &msg.LoginResp{
Version: version.Full(),
Error: err.Error(),
})
conn.Close()
}
case *msg.NewWorkConn:
svr.RegisterWorkConn(conn, m)
case *msg.NewVisitorConn:
if err = svr.RegisterVisitorConn(conn, m); err != nil {
conn.Warn("%v", err)
msg.WriteMsg(conn, &msg.NewVisitorConnResp{
ProxyName: m.ProxyName,
Error: err.Error(),
})
conn.Close()
} else {
msg.WriteMsg(conn, &msg.NewVisitorConnResp{
ProxyName: m.ProxyName,
Error: "",
})
}
default:
log.Warn("Error message type for the new connection [%s]", conn.RemoteAddr().String())
conn.Close()
}
这就是服务端启动之后，卡住的地方了。客户端建立连接之后，会发送一个消息，它的类型可能是 msg.Login, msg.NewWorkConn, msg.NewVisitorConn。上一篇我们说了，visitor
是用于stcp也就是端对端加密通信的，我们不看。workConn就是用于转发流量的，Login就是新的客户端连上去之后进行启动。
frpc
同样，我们从 cmd/frpc/main.go 看起：
func main() {
crypto.DefaultSalt = "frp"
sub.Execute()
}
跳转到 sub.Execute()：
var rootCmd = &cobra.Command{
Use: "frpc",
Short: "frpc is the client of frp (https://github.com/fatedier/frp)",
RunE: func(cmd *cobra.Command, args []string) error {
if showVersion {
fmt.Println(version.Full())
return nil
}
// Do not show command usage here.
err := runClient(cfgFile)
if err != nil {
fmt.Println(err)
os.Exit(1)
}
return nil
},
}
func Execute() {
if err := rootCmd.Execute(); err != nil {
os.Exit(1)
}
}
然后我们看 runClient 函数：
func runClient(cfgFilePath string) (err error) {
var content string
content, err = config.GetRenderedConfFromFile(cfgFilePath)
if err != nil {
return
}
g.GlbClientCfg.CfgFile = cfgFilePath
err = parseClientCommonCfg(CfgFileTypeIni, content)
if err != nil {
return
}
pxyCfgs, visitorCfgs, err := config.LoadAllConfFromIni(g.GlbClientCfg.User, content, g.GlbClientCfg.Start)
if err != nil {
return err
}
err = startService(pxyCfgs, visitorCfgs)
return
}
基本上就是解析配置文件（因为frpc启动的时候要一个配置文件），然后执行 startService：
func startService(pxyCfgs map[string]config.ProxyConf, visitorCfgs map[string]config.VisitorConf) (err error) {
log.InitLog(g.GlbClientCfg.LogWay, g.GlbClientCfg.LogFile, g.GlbClientCfg.LogLevel, g.GlbClientCfg.LogMaxDays)
if g.GlbClientCfg.DnsServer != "" {
s := g.GlbClientCfg.DnsServer
if !strings.Contains(s, ":") {
s += ":53"
}
// Change default dns server for frpc
net.DefaultResolver = &net.Resolver{
PreferGo: true,
Dial: func(ctx context.Context, network, address string) (net.Conn, error) {
return net.Dial("udp", s)
},
}
}
svr, errRet := client.NewService(pxyCfgs, visitorCfgs)
if errRet != nil {
err = errRet
return
}
// Capture the exit signal if we use kcp.
if g.GlbClientCfg.Protocol == "kcp" {
go handleSignal(svr)
}
err = svr.Run()
if g.GlbClientCfg.Protocol == "kcp" {
<-kcpDoneCh
}
return
}
同样的，执行 client.NewService 之后执行 svr.Run()，我们看看 svr.Run() 是什么：
func (svr *Service) Run() error {
// first login
for {
conn, session, err := svr.login()
if err != nil {
log.Warn("login to server failed: %v", err)
// if login_fail_exit is true, just exit this program
// otherwise sleep a while and try again to connect to server
if g.GlbClientCfg.LoginFailExit {
return err
} else {
time.Sleep(10 * time.Second)
}
} else {
// login success
ctl := NewControl(svr.runId, conn, session, svr.pxyCfgs, svr.visitorCfgs)
ctl.Run()
svr.ctlMu.Lock()
svr.ctl = ctl
svr.ctlMu.Unlock()
break
}
}
go svr.keepControllerWorking()
if g.GlbClientCfg.AdminPort != 0 {
err := svr.RunAdminServer(g.GlbClientCfg.AdminAddr, g.GlbClientCfg.AdminPort)
if err != nil {
log.Warn("run admin server error: %v", err)
}
log.Info("admin server listen on %s:%d", g.GlbClientCfg.AdminAddr, g.GlbClientCfg.AdminPort)
}
<-svr.closedCh
return nil
}
可以看到，客户端启动之后，就是一个 for 循环，写入 login 信息，也就是刚才 frps 里的 msg.loginMsg，
然后起一个 goroutine 执行 keepControllerWorking()，之后主 goroutine 就阻塞在 <-svr.closedCh。
看看 keepControllerWorking() 的内容：
func (svr *Service) keepControllerWorking() {
maxDelayTime := 20 * time.Second
delayTime := time.Second
for {
<-svr.ctl.ClosedDoneCh()
if atomic.LoadUint32(&svr.exit) != 0 {
return
}
for {
log.Info("try to reconnect to server...")
conn, session, err := svr.login()
if err != nil {
log.Warn("reconnect to server error: %v", err)
time.Sleep(delayTime)
delayTime = delayTime * 2
if delayTime > maxDelayTime {
delayTime = maxDelayTime
}
continue
}
// reconnect success, init delayTime
delayTime = time.Second
ctl := NewControl(svr.runId, conn, session, svr.pxyCfgs, svr.visitorCfgs)
ctl.Run()
svr.ctlMu.Lock()
svr.ctl = ctl
svr.ctlMu.Unlock()
break
}
}
}
基本上就是一个循环，里面最终是为了成功连接然后执行 ctl.Run()：
func (ctl *Control) Run() {
go ctl.worker()
// start all proxies
ctl.pm.Reload(ctl.pxyCfgs)
// start all visitors
go ctl.vm.Run()
return
}
// If controler is notified by closedCh, reader and writer and handler will exit
func (ctl *Control) worker() {
go ctl.msgHandler()
go ctl.reader()
go ctl.writer()
select {
case <-ctl.closedCh:
// close related channels and wait until other goroutines done
close(ctl.readCh)
ctl.readerShutdown.WaitDone()
ctl.msgHandlerShutdown.WaitDone()
close(ctl.sendCh)
ctl.writerShutdown.WaitDone()
ctl.pm.Close()
ctl.vm.Close()
close(ctl.closedDoneCh)
if ctl.session != nil {
ctl.session.Close()
}
return
}
}
// msgHandler handles all channel events and do corresponding operations.
func (ctl *Control) msgHandler() {
defer func() {
if err := recover(); err != nil {
ctl.Error("panic error: %v", err)
ctl.Error(string(debug.Stack()))
}
}()
defer ctl.msgHandlerShutdown.Done()
hbSend := time.NewTicker(time.Duration(g.GlbClientCfg.HeartBeatInterval) * time.Second)
defer hbSend.Stop()
hbCheck := time.NewTicker(time.Second)
defer hbCheck.Stop()
ctl.lastPong = time.Now()
for {
select {
case <-hbSend.C:
// send heartbeat to server
ctl.Debug("send heartbeat to server")
ctl.sendCh <- &msg.Ping{}
case <-hbCheck.C:
if time.Since(ctl.lastPong) > time.Duration(g.GlbClientCfg.HeartBeatTimeout)*time.Second {
ctl.Warn("heartbeat timeout")
// let reader() stop
ctl.conn.Close()
return
}
case rawMsg, ok := <-ctl.readCh:
if !ok {
return
}
switch m := rawMsg.(type) {
case *msg.ReqWorkConn:
go ctl.HandleReqWorkConn(m)
case *msg.NewProxyResp:
ctl.HandleNewProxyResp(m)
case *msg.Pong:
ctl.lastPong = time.Now()
ctl.Debug("receive heartbeat from server")
}
}
}
}
// reader read all messages from frps and send to readCh
func (ctl *Control) reader() {
defer func() {
if err := recover(); err != nil {
ctl.Error("panic error: %v", err)
ctl.Error(string(debug.Stack()))
}
}()
defer ctl.readerShutdown.Done()
defer close(ctl.closedCh)
encReader := crypto.NewReader(ctl.conn, []byte(g.GlbClientCfg.Token))
for {
if m, err := msg.ReadMsg(encReader); err != nil {
if err == io.EOF {
ctl.Debug("read from control connection EOF")
return
} else {
ctl.Warn("read error: %v", err)
ctl.conn.Close()
return
}
} else {
ctl.readCh <- m
}
}
}
// writer writes messages got from sendCh to frps
func (ctl *Control) writer() {
defer ctl.writerShutdown.Done()
encWriter, err := crypto.NewWriter(ctl.conn, []byte(g.GlbClientCfg.Token))
if err != nil {
ctl.conn.Error("crypto new writer error: %v", err)
ctl.conn.Close()
return
}
for {
if m, ok := <-ctl.sendCh; !ok {
ctl.Info("control writer is closing")
return
} else {
if err := msg.WriteMsg(encWriter, m); err != nil {
ctl.Warn("write message to control connection error: %v", err)
return
}
}
}
}
reader 从frps收信息，然后写到 ctl.readCh 这个 channel 里，writer 则相反，
从 ctl.sendCh 收信息，写到 frps，而 msgHandler 则从frpc里读取信息，放到 ctl.sendCh，
从 ctl.readCh 读取信息，处理之。
之所以这样设计，是为了能够异步处理所有消息。看看 msgHandler 的关键部分：
switch m := rawMsg.(type) {
case *msg.ReqWorkConn:
go ctl.HandleReqWorkConn(m)
case *msg.NewProxyResp:
ctl.HandleNewProxyResp(m)
case *msg.Pong:
ctl.lastPong = time.Now()
ctl.Debug("receive heartbeat from server")
}
我们说过了，frps 每次收到一个请求之后，然后下发一个指令给frpc，要求frpc建立连接，
然后frps再把新来的连接与请求所在的连接串起来，完成代理，msg.ReqWorkConn 就是这个指令。
那么 frps 是在哪里下发指令的呢？
frps 下发指令
每当公网来一个新的请求的时候，frps就会下发一个指令给frpc，要求建立一个新的连接，代码如下：
func (pxy *BaseProxy) GetWorkConnFromPool(src, dst net.Addr) (workConn frpNet.Conn, err error) {
// try all connections from the pool
for i := 0; i < pxy.poolCount+1; i++ {
if workConn, err = pxy.getWorkConnFn(); err != nil {
pxy.Warn("failed to get work connection: %v", err)
return
}
pxy.Info("get a new work connection: [%s]", workConn.RemoteAddr().String())
workConn.AddLogPrefix(pxy.GetName())
var (
srcAddr string
dstAddr string
srcPortStr string
dstPortStr string
srcPort int
dstPort int
)
if src != nil {
srcAddr, srcPortStr, _ = net.SplitHostPort(src.String())
srcPort, _ = strconv.Atoi(srcPortStr)
}
if dst != nil {
dstAddr, dstPortStr, _ = net.SplitHostPort(dst.String())
dstPort, _ = strconv.Atoi(dstPortStr)
}
message := msg.StartWorkConn{
ProxyName: pxy.GetName(),
SrcAddr: srcAddr,
SrcPort: uint16(srcPort),
DstAddr: dstAddr,
DstPort: uint16(dstPort),
}
err := msg.WriteMsg(workConn, &message)
workConn.Warn("===== HERE! HERE! HERE!, message is: %+v", message)
if err != nil {
workConn.Warn("failed to send message to work connection from pool: %v, times: %d", err, i)
workConn.Close()
} else {
break
}
}
if err != nil {
pxy.Error("try to get work connection failed in the end")
return
}
return
}
那个 ===== HERE! HERE! HERE!, message is: %+v 是我加上去的，为了方便看每次服务端下发什么信息。我们看的是TCP的代理，
它继承于 server/proxy/proxy.go 里的 BaseProxy，实现是在 server/proxy/tcp.go 里的 type TcpProxy struct。
那么什么时候会初始化 TcpProxy 呢？我们注意到，frps 接收的消息里，就有一种是消息类型是 msg.NewProxy，这是客户端和
服务端都启动，并且客户端成功login之后，客户端发送给服务端的消息。也就是 frpc 的配置文件里具体的代理，例如：
[common]
server_addr = xxxxx
server_port = 12345
tls_enable = true
[ssh]
type = tcp
local_ip = 127.0.0.1
local_port = 22
remote_port = 12346
中的 ssh 就是一个TCP代理。收到 msg.NewProxy 之后，服务端会起一个新的监听器监听在对应的端口，然后开始处理请求。
总结
这一篇中，我们在第一篇的基础之上看了frpc和frps的交互流程，了解了frp是如何进行TCP代理的。
更多文章
邮件 订阅
RSS 订阅
Web开发简介系列
数据结构的实际使用
Golang 简明教程
Python 教程
本站热门
- socks5 协议详解
- zerotier简明教程
- 搞定面试中的系统设计题
- frp 源码阅读与分析(一)：流程和概念
- 用peewee代替SQLAlchemy
- Golang(Go语言)中实现典型的fork调用
- DNSCrypt简明教程
- 一个Gunicorn worker数量引发的血案
- Golang validator使用教程
- Docker组件介绍（一）：runc和containerd
- Docker组件介绍（二）：shim, docker-init和docker-proxy
- 使用Go语言实现一个异步任务框架
- 协程(coroutine)简介 - 什么是协程？
- SQLAlchemy简明教程
- Go Module 简明教程