frp/client/proxy/proxy.go

219 lines
6.2 KiB
Go
Raw Normal View History

2017-03-08 19:03:47 +01: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 15:06:22 +01:00
package proxy
2017-03-08 19:03:47 +01:00
import (
2019-10-12 14:13:12 +02:00
"context"
2017-03-08 19:03:47 +01:00
"io"
2017-03-12 19:44:47 +01:00
"net"
"reflect"
2019-03-11 08:53:58 +01:00
"strconv"
"strings"
2017-04-24 18:34:14 +02:00
"sync"
"time"
2017-03-08 19:03:47 +01:00
2023-05-29 08:10:34 +02:00
libio "github.com/fatedier/golib/io"
2022-01-20 13:03:07 +01:00
libdial "github.com/fatedier/golib/net/dial"
2019-03-29 12:01:18 +01:00
pp "github.com/pires/go-proxyproto"
2019-11-02 18:20:49 +01:00
"golang.org/x/time/rate"
2022-08-28 19:02:53 +02:00
"github.com/fatedier/frp/pkg/config/types"
v1 "github.com/fatedier/frp/pkg/config/v1"
2022-08-28 19:02:53 +02:00
"github.com/fatedier/frp/pkg/msg"
plugin "github.com/fatedier/frp/pkg/plugin/client"
"github.com/fatedier/frp/pkg/transport"
2022-08-28 19:02:53 +02:00
"github.com/fatedier/frp/pkg/util/limit"
"github.com/fatedier/frp/pkg/util/xlog"
2017-03-08 19:03:47 +01: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 11:35:05 +01:00
// Proxy defines how to handle work connections for different proxy type.
2017-03-08 19:03:47 +01:00
type Proxy interface {
2017-03-11 19:03:24 +01:00
Run() error
2017-03-09 18:42:06 +01:00
// InWorkConn accept work connections registered to server.
2019-10-12 14:13:12 +02:00
InWorkConn(net.Conn, *msg.StartWorkConn)
2018-01-16 18:09:33 +01:00
2017-03-08 19:03:47 +01:00
Close()
}
func NewProxy(
ctx context.Context,
pxyConf v1.ProxyConfigurer,
clientCfg *v1.ClientCommonConfig,
msgTransporter transport.MessageTransporter,
) (pxy Proxy) {
2019-11-02 18:20:49 +01:00
var limiter *rate.Limiter
limitBytes := pxyConf.GetBaseConfig().Transport.BandwidthLimit.Bytes()
if limitBytes > 0 && pxyConf.GetBaseConfig().Transport.BandwidthLimitMode == types.BandwidthLimitModeClient {
2019-11-02 18:20:49 +01:00
limiter = rate.NewLimiter(rate.Limit(float64(limitBytes)), int(limitBytes))
}
2017-03-12 19:44:47 +01:00
baseProxy := BaseProxy{
baseCfg: pxyConf.GetBaseConfig(),
clientCfg: clientCfg,
limiter: limiter,
msgTransporter: msgTransporter,
xl: xlog.FromContextSafe(ctx),
ctx: ctx,
2017-03-12 19:44:47 +01:00
}
factory := proxyFactoryRegistry[reflect.TypeOf(pxyConf)]
if factory == nil {
return nil
2017-03-08 19:03:47 +01:00
}
return factory(&baseProxy, pxyConf)
2017-03-08 19:03:47 +01:00
}
2017-03-12 19:44:47 +01: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.
proxyPlugin plugin.Plugin
2019-10-12 14:13:12 +02:00
mu sync.RWMutex
xl *xlog.Logger
ctx context.Context
2017-03-12 19:44:47 +01: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()
}
}
func (pxy *BaseProxy) InWorkConn(conn net.Conn, m *msg.StartWorkConn) {
pxy.HandleTCPWorkConnection(conn, m, []byte(pxy.clientCfg.Auth.Token))
2017-06-25 21:02:33 +02:00
}
2017-03-09 19:01:17 +01: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-09 19:01:17 +01:00
remote = workConn
if pxy.limiter != nil {
remote = libio.WrapReadWriteCloser(limit.NewReader(workConn, pxy.limiter), limit.NewWriter(workConn, pxy.limiter), func() error {
2019-11-02 18:20:49 +01:00
return workConn.Close()
})
}
2018-01-23 09:31:59 +01:00
2019-10-12 14:13:12 +02:00
xl.Trace("handle tcp work connection, use_encryption: %t, use_compression: %t",
baseCfg.Transport.UseEncryption, baseCfg.Transport.UseCompression)
if baseCfg.Transport.UseEncryption {
2023-05-29 08:10:34 +02:00
remote, err = libio.WithEncryption(remote, encKey)
2017-03-09 19:01:17 +01:00
if err != nil {
2018-01-23 09:31:59 +01:00
workConn.Close()
2019-10-12 14:13:12 +02:00
xl.Error("create encryption stream error: %v", err)
2017-03-09 19:01:17 +01:00
return
}
}
var compressionResourceRecycleFn func()
if baseCfg.Transport.UseCompression {
remote, compressionResourceRecycleFn = libio.WithCompressionFromPool(remote)
2017-03-09 19:01:17 +01:00
}
// check if we need to send proxy protocol info
2023-09-20 09:18:50 +02: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 09:18:50 +02:00
extraInfo.ProxyProtocolHeader = h
}
}
if pxy.proxyPlugin != nil {
// if plugin is set, let plugin handle connection first
xl.Debug("handle by plugin: %s", pxy.proxyPlugin.Name())
2023-09-20 09:18:50 +02:00
pxy.proxyPlugin.Handle(remote, workConn, &extraInfo)
2019-10-12 14:13:12 +02:00
xl.Debug("handle by plugin finished")
return
2020-05-24 11:48:37 +02:00
}
localConn, err := libdial.Dial(
net.JoinHostPort(baseCfg.LocalIP, strconv.Itoa(baseCfg.LocalPort)),
libdial.WithTimeout(10*time.Second),
)
2020-05-24 11:48:37 +02:00
if err != nil {
workConn.Close()
xl.Error("connect to local service [%s:%d] error: %v", baseCfg.LocalIP, baseCfg.LocalPort, err)
2020-05-24 11:48:37 +02:00
return
}
2019-03-29 12:01:18 +01:00
2020-05-24 11:48:37 +02:00
xl.Debug("join connections, localConn(l[%s] r[%s]) workConn(l[%s] r[%s])", localConn.LocalAddr().String(),
localConn.RemoteAddr().String(), workConn.LocalAddr().String(), workConn.RemoteAddr().String())
2019-03-29 12:01:18 +01:00
2023-09-20 09:18:50 +02:00
if extraInfo.ProxyProtocolHeader != nil {
if _, err := extraInfo.ProxyProtocolHeader.WriteTo(localConn); err != nil {
2022-08-28 19:02:53 +02:00
workConn.Close()
2023-09-20 09:18:50 +02:00
xl.Error("write proxy protocol header to local conn error: %v", err)
2022-08-28 19:02:53 +02:00
return
}
}
2020-05-24 11:48:37 +02:00
2023-05-29 08:10:34 +02:00
_, _, errs := libio.Join(localConn, remote)
2020-05-24 11:48:37 +02:00
xl.Debug("join connections closed")
2023-03-11 12:34:06 +01:00
if len(errs) > 0 {
xl.Trace("join connections errors: %v", errs)
}
if compressionResourceRecycleFn != nil {
compressionResourceRecycleFn()
}
2017-03-08 19:03:47 +01:00
}