frp/client/proxy/proxy.go

228 lines
6.7 KiB
Go
Raw Normal View History

2017-03-09 02:03:47 +08:00
// Copyright 2017 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.
2018-12-09 22:06:22 +08:00
package proxy
2017-03-09 02:03:47 +08:00
import (
2019-10-12 20:13:12 +08:00
"context"
2017-03-09 02:03:47 +08:00
"io"
2017-03-13 02:44:47 +08:00
"net"
"reflect"
2019-03-11 15:53:58 +08:00
"strconv"
"strings"
2017-04-25 00:34:14 +08:00
"sync"
"time"
2017-03-09 02:03:47 +08:00
2023-05-29 14:10:34 +08:00
libio "github.com/fatedier/golib/io"
2022-01-20 20:03:07 +08:00
libdial "github.com/fatedier/golib/net/dial"
2019-03-29 19:01:18 +08:00
pp "github.com/pires/go-proxyproto"
2019-11-03 01:20:49 +08:00
"golang.org/x/time/rate"
2022-08-29 01:02:53 +08:00
"github.com/fatedier/frp/pkg/config/types"
v1 "github.com/fatedier/frp/pkg/config/v1"
2022-08-29 01:02:53 +08:00
"github.com/fatedier/frp/pkg/msg"
plugin "github.com/fatedier/frp/pkg/plugin/client"
"github.com/fatedier/frp/pkg/transport"
2022-08-29 01:02:53 +08:00
"github.com/fatedier/frp/pkg/util/limit"
"github.com/fatedier/frp/pkg/util/xlog"
2017-03-09 02:03:47 +08:00
)
var proxyFactoryRegistry = map[reflect.Type]func(*BaseProxy, v1.ProxyConfigurer) Proxy{}
func RegisterProxyFactory(proxyConfType reflect.Type, factory func(*BaseProxy, v1.ProxyConfigurer) Proxy) {
proxyFactoryRegistry[proxyConfType] = factory
}
2018-11-06 18:35:05 +08:00
// Proxy defines how to handle work connections for different proxy type.
2017-03-09 02:03:47 +08:00
type Proxy interface {
2017-03-12 02:03:24 +08:00
Run() error
2017-03-10 01:42:06 +08:00
// InWorkConn accept work connections registered to server.
2019-10-12 20:13:12 +08:00
InWorkConn(net.Conn, *msg.StartWorkConn)
2023-11-21 11:19:35 +08:00
SetInWorkConnCallback(func(*v1.ProxyBaseConfig, net.Conn, *msg.StartWorkConn) /* continue */ bool)
2017-03-09 02:03:47 +08:00
Close()
}
func NewProxy(
ctx context.Context,
pxyConf v1.ProxyConfigurer,
clientCfg *v1.ClientCommonConfig,
msgTransporter transport.MessageTransporter,
) (pxy Proxy) {
2019-11-03 01:20:49 +08:00
var limiter *rate.Limiter
limitBytes := pxyConf.GetBaseConfig().Transport.BandwidthLimit.Bytes()
if limitBytes > 0 && pxyConf.GetBaseConfig().Transport.BandwidthLimitMode == types.BandwidthLimitModeClient {
2019-11-03 01:20:49 +08:00
limiter = rate.NewLimiter(rate.Limit(float64(limitBytes)), int(limitBytes))
}
2017-03-13 02:44:47 +08:00
baseProxy := BaseProxy{
baseCfg: pxyConf.GetBaseConfig(),
clientCfg: clientCfg,
limiter: limiter,
msgTransporter: msgTransporter,
xl: xlog.FromContextSafe(ctx),
ctx: ctx,
2017-03-13 02:44:47 +08:00
}
factory := proxyFactoryRegistry[reflect.TypeOf(pxyConf)]
if factory == nil {
return nil
2017-03-09 02:03:47 +08:00
}
return factory(&baseProxy, pxyConf)
2017-03-09 02:03:47 +08:00
}
2017-03-13 02:44:47 +08:00
type BaseProxy struct {
baseCfg *v1.ProxyBaseConfig
clientCfg *v1.ClientCommonConfig
msgTransporter transport.MessageTransporter
limiter *rate.Limiter
// proxyPlugin is used to handle connections instead of dialing to local service.
// It's only validate for TCP protocol now.
2023-11-21 11:19:35 +08:00
proxyPlugin plugin.Plugin
inWorkConnCallback func(*v1.ProxyBaseConfig, net.Conn, *msg.StartWorkConn) /* continue */ bool
2019-10-12 20:13:12 +08:00
mu sync.RWMutex
xl *xlog.Logger
ctx context.Context
2017-03-13 02:44:47 +08:00
}
func (pxy *BaseProxy) Run() error {
if pxy.baseCfg.Plugin.Type != "" {
p, err := plugin.Create(pxy.baseCfg.Plugin.Type, pxy.baseCfg.Plugin.ClientPluginOptions)
if err != nil {
return err
}
pxy.proxyPlugin = p
}
return nil
}
func (pxy *BaseProxy) Close() {
if pxy.proxyPlugin != nil {
pxy.proxyPlugin.Close()
}
}
2023-11-21 11:19:35 +08:00
func (pxy *BaseProxy) SetInWorkConnCallback(cb func(*v1.ProxyBaseConfig, net.Conn, *msg.StartWorkConn) bool) {
pxy.inWorkConnCallback = cb
}
func (pxy *BaseProxy) InWorkConn(conn net.Conn, m *msg.StartWorkConn) {
2023-11-21 11:19:35 +08:00
if pxy.inWorkConnCallback != nil {
if !pxy.inWorkConnCallback(pxy.baseCfg, conn, m) {
return
}
}
pxy.HandleTCPWorkConnection(conn, m, []byte(pxy.clientCfg.Auth.Token))
2017-06-26 03:02:33 +08:00
}
2017-03-10 02:01:17 +08:00
// Common handler for tcp work connections.
func (pxy *BaseProxy) HandleTCPWorkConnection(workConn net.Conn, m *msg.StartWorkConn, encKey []byte) {
xl := pxy.xl
baseCfg := pxy.baseCfg
var (
remote io.ReadWriteCloser
err error
)
2017-03-10 02:01:17 +08:00
remote = workConn
if pxy.limiter != nil {
remote = libio.WrapReadWriteCloser(limit.NewReader(workConn, pxy.limiter), limit.NewWriter(workConn, pxy.limiter), func() error {
2019-11-03 01:20:49 +08:00
return workConn.Close()
})
}
2018-01-23 16:31:59 +08:00
2024-03-12 13:58:53 +08:00
xl.Tracef("handle tcp work connection, useEncryption: %t, useCompression: %t",
baseCfg.Transport.UseEncryption, baseCfg.Transport.UseCompression)
if baseCfg.Transport.UseEncryption {
2023-05-29 14:10:34 +08:00
remote, err = libio.WithEncryption(remote, encKey)
2017-03-10 02:01:17 +08:00
if err != nil {
2018-01-23 16:31:59 +08:00
workConn.Close()
2024-03-12 13:58:53 +08:00
xl.Errorf("create encryption stream error: %v", err)
2017-03-10 02:01:17 +08:00
return
}
}
var compressionResourceRecycleFn func()
if baseCfg.Transport.UseCompression {
remote, compressionResourceRecycleFn = libio.WithCompressionFromPool(remote)
2017-03-10 02:01:17 +08:00
}
// check if we need to send proxy protocol info
2023-09-20 15:18:50 +08:00
var extraInfo plugin.ExtraInfo
if baseCfg.Transport.ProxyProtocolVersion != "" {
if m.SrcAddr != "" && m.SrcPort != 0 {
if m.DstAddr == "" {
m.DstAddr = "127.0.0.1"
}
srcAddr, _ := net.ResolveTCPAddr("tcp", net.JoinHostPort(m.SrcAddr, strconv.Itoa(int(m.SrcPort))))
dstAddr, _ := net.ResolveTCPAddr("tcp", net.JoinHostPort(m.DstAddr, strconv.Itoa(int(m.DstPort))))
h := &pp.Header{
Command: pp.PROXY,
SourceAddr: srcAddr,
DestinationAddr: dstAddr,
}
if strings.Contains(m.SrcAddr, ".") {
h.TransportProtocol = pp.TCPv4
} else {
h.TransportProtocol = pp.TCPv6
}
if baseCfg.Transport.ProxyProtocolVersion == "v1" {
h.Version = 1
} else if baseCfg.Transport.ProxyProtocolVersion == "v2" {
h.Version = 2
}
2023-09-20 15:18:50 +08:00
extraInfo.ProxyProtocolHeader = h
}
}
if pxy.proxyPlugin != nil {
// if plugin is set, let plugin handle connection first
2024-03-12 13:58:53 +08:00
xl.Debugf("handle by plugin: %s", pxy.proxyPlugin.Name())
2023-09-20 15:18:50 +08:00
pxy.proxyPlugin.Handle(remote, workConn, &extraInfo)
2024-03-12 13:58:53 +08:00
xl.Debugf("handle by plugin finished")
return
2020-05-24 17:48:37 +08:00
}
localConn, err := libdial.Dial(
net.JoinHostPort(baseCfg.LocalIP, strconv.Itoa(baseCfg.LocalPort)),
libdial.WithTimeout(10*time.Second),
)
2020-05-24 17:48:37 +08:00
if err != nil {
workConn.Close()
2024-03-12 13:58:53 +08:00
xl.Errorf("connect to local service [%s:%d] error: %v", baseCfg.LocalIP, baseCfg.LocalPort, err)
2020-05-24 17:48:37 +08:00
return
}
2019-03-29 19:01:18 +08:00
2024-03-12 13:58:53 +08:00
xl.Debugf("join connections, localConn(l[%s] r[%s]) workConn(l[%s] r[%s])", localConn.LocalAddr().String(),
2020-05-24 17:48:37 +08:00
localConn.RemoteAddr().String(), workConn.LocalAddr().String(), workConn.RemoteAddr().String())
2019-03-29 19:01:18 +08:00
2023-09-20 15:18:50 +08:00
if extraInfo.ProxyProtocolHeader != nil {
if _, err := extraInfo.ProxyProtocolHeader.WriteTo(localConn); err != nil {
2022-08-29 01:02:53 +08:00
workConn.Close()
2024-03-12 13:58:53 +08:00
xl.Errorf("write proxy protocol header to local conn error: %v", err)
2022-08-29 01:02:53 +08:00
return
}
}
2020-05-24 17:48:37 +08:00
2023-05-29 14:10:34 +08:00
_, _, errs := libio.Join(localConn, remote)
2024-03-12 13:58:53 +08:00
xl.Debugf("join connections closed")
2023-03-11 19:34:06 +08:00
if len(errs) > 0 {
2024-03-12 13:58:53 +08:00
xl.Tracef("join connections errors: %v", errs)
2023-03-11 19:34:06 +08:00
}
if compressionResourceRecycleFn != nil {
compressionResourceRecycleFn()
}
2017-03-09 02:03:47 +08:00
}