frp/server/proxy/proxy.go

363 lines
9.6 KiB
Go
Raw Normal View History

2019-01-15 00:11:08 +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.
package proxy
import (
2019-10-12 20:13:12 +08:00
"context"
2019-01-15 00:11:08 +08:00
"fmt"
"io"
2019-03-29 19:01:18 +08:00
"net"
"reflect"
2019-03-29 19:01:18 +08:00
"strconv"
2019-01-15 00:11:08 +08:00
"sync"
"time"
2019-01-15 00:11:08 +08:00
2023-05-29 14:10:34 +08:00
libio "github.com/fatedier/golib/io"
"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"
2020-09-23 13:49:14 +08:00
"github.com/fatedier/frp/pkg/msg"
plugin "github.com/fatedier/frp/pkg/plugin/server"
"github.com/fatedier/frp/pkg/util/limit"
2023-05-29 14:10:34 +08:00
utilnet "github.com/fatedier/frp/pkg/util/net"
2020-09-23 13:49:14 +08:00
"github.com/fatedier/frp/pkg/util/xlog"
2019-01-15 00:11:08 +08:00
"github.com/fatedier/frp/server/controller"
"github.com/fatedier/frp/server/metrics"
2019-01-15 00:11:08 +08:00
)
var proxyFactoryRegistry = map[reflect.Type]func(*BaseProxy) Proxy{}
func RegisterProxyFactory(proxyConfType reflect.Type, factory func(*BaseProxy) Proxy) {
proxyFactoryRegistry[proxyConfType] = factory
}
2019-10-12 20:13:12 +08:00
type GetWorkConnFn func() (net.Conn, error)
2019-01-15 00:11:08 +08:00
type Proxy interface {
2019-10-12 20:13:12 +08:00
Context() context.Context
2019-01-15 00:11:08 +08:00
Run() (remoteAddr string, err error)
GetName() string
GetConfigurer() v1.ProxyConfigurer
2019-10-12 20:13:12 +08:00
GetWorkConnFromPool(src, dst net.Addr) (workConn net.Conn, err error)
2019-01-15 00:11:08 +08:00
GetUsedPortsNum() int
GetResourceController() *controller.ResourceController
GetUserInfo() plugin.UserInfo
GetLimiter() *rate.Limiter
2023-02-22 00:39:56 +08:00
GetLoginMsg() *msg.Login
2019-01-15 00:11:08 +08:00
Close()
}
type BaseProxy struct {
name string
rc *controller.ResourceController
listeners []net.Listener
usedPortsNum int
poolCount int
getWorkConnFn GetWorkConnFn
serverCfg *v1.ServerConfig
limiter *rate.Limiter
userInfo plugin.UserInfo
2023-02-22 00:39:56 +08:00
loginMsg *msg.Login
configurer v1.ProxyConfigurer
2019-01-15 00:11:08 +08:00
2019-10-12 20:13:12 +08:00
mu sync.RWMutex
xl *xlog.Logger
ctx context.Context
2019-01-15 00:11:08 +08:00
}
func (pxy *BaseProxy) GetName() string {
return pxy.name
}
2019-10-12 20:13:12 +08:00
func (pxy *BaseProxy) Context() context.Context {
return pxy.ctx
}
2019-01-15 00:11:08 +08:00
func (pxy *BaseProxy) GetUsedPortsNum() int {
return pxy.usedPortsNum
}
func (pxy *BaseProxy) GetResourceController() *controller.ResourceController {
return pxy.rc
}
func (pxy *BaseProxy) GetUserInfo() plugin.UserInfo {
return pxy.userInfo
}
2023-02-22 00:39:56 +08:00
func (pxy *BaseProxy) GetLoginMsg() *msg.Login {
return pxy.loginMsg
}
func (pxy *BaseProxy) GetLimiter() *rate.Limiter {
return pxy.limiter
}
func (pxy *BaseProxy) GetConfigurer() v1.ProxyConfigurer {
return pxy.configurer
}
2019-01-15 00:11:08 +08:00
func (pxy *BaseProxy) Close() {
2019-10-12 20:13:12 +08:00
xl := xlog.FromContextSafe(pxy.ctx)
xl.Info("proxy closing")
2019-01-15 00:11:08 +08:00
for _, l := range pxy.listeners {
l.Close()
}
}
2019-07-31 00:41:58 +08:00
// GetWorkConnFromPool try to get a new work connections from pool
// for quickly response, we immediately send the StartWorkConn message to frpc after take out one from pool
2019-10-12 20:13:12 +08:00
func (pxy *BaseProxy) GetWorkConnFromPool(src, dst net.Addr) (workConn net.Conn, err error) {
xl := xlog.FromContextSafe(pxy.ctx)
2019-01-15 00:11:08 +08:00
// try all connections from the pool
for i := 0; i < pxy.poolCount+1; i++ {
if workConn, err = pxy.getWorkConnFn(); err != nil {
2019-10-12 20:13:12 +08:00
xl.Warn("failed to get work connection: %v", err)
2019-01-15 00:11:08 +08:00
return
}
xl.Debug("get a new work connection: [%s]", workConn.RemoteAddr().String())
2019-10-12 20:13:12 +08:00
xl.Spawn().AppendPrefix(pxy.GetName())
2023-05-29 14:10:34 +08:00
workConn = utilnet.NewContextConn(pxy.ctx, workConn)
2019-01-15 00:11:08 +08:00
2019-03-29 19:01:18 +08:00
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)
}
2019-01-15 00:11:08 +08:00
err := msg.WriteMsg(workConn, &msg.StartWorkConn{
ProxyName: pxy.GetName(),
2019-03-29 19:01:18 +08:00
SrcAddr: srcAddr,
SrcPort: uint16(srcPort),
DstAddr: dstAddr,
DstPort: uint16(dstPort),
Error: "",
2019-01-15 00:11:08 +08:00
})
if err != nil {
2019-10-12 20:13:12 +08:00
xl.Warn("failed to send message to work connection from pool: %v, times: %d", err, i)
2019-01-15 00:11:08 +08:00
workConn.Close()
} else {
break
}
}
if err != nil {
2019-10-12 20:13:12 +08:00
xl.Error("try to get work connection failed in the end")
2019-01-15 00:11:08 +08:00
return
}
return
}
// startCommonTCPListenersHandler start a goroutine handler for each listener.
func (pxy *BaseProxy) startCommonTCPListenersHandler() {
2019-10-12 20:13:12 +08:00
xl := xlog.FromContextSafe(pxy.ctx)
2019-01-15 00:11:08 +08:00
for _, listener := range pxy.listeners {
2019-10-12 20:13:12 +08:00
go func(l net.Listener) {
var tempDelay time.Duration // how long to sleep on accept failure
2019-01-15 00:11:08 +08:00
for {
// block
// if listener is closed, err returned
c, err := l.Accept()
if err != nil {
if err, ok := err.(interface{ Temporary() bool }); ok && err.Temporary() {
if tempDelay == 0 {
tempDelay = 5 * time.Millisecond
} else {
tempDelay *= 2
}
if max := 1 * time.Second; tempDelay > max {
tempDelay = max
}
xl.Info("met temporary error: %s, sleep for %s ...", err, tempDelay)
time.Sleep(tempDelay)
continue
}
xl.Warn("listener is closed: %s", err)
2019-01-15 00:11:08 +08:00
return
}
xl.Info("get a user connection [%s]", c.RemoteAddr().String())
go pxy.handleUserTCPConnection(c)
2019-01-15 00:11:08 +08:00
}
}(listener)
}
}
2020-05-24 17:48:37 +08:00
// HandleUserTCPConnection is used for incoming user TCP connections.
func (pxy *BaseProxy) handleUserTCPConnection(userConn net.Conn) {
2019-10-12 20:13:12 +08:00
xl := xlog.FromContextSafe(pxy.Context())
2019-01-15 00:11:08 +08:00
defer userConn.Close()
serverCfg := pxy.serverCfg
cfg := pxy.configurer.GetBaseConfig()
// server plugin hook
rc := pxy.GetResourceController()
content := &plugin.NewUserConnContent{
User: pxy.GetUserInfo(),
ProxyName: pxy.GetName(),
ProxyType: cfg.Type,
RemoteAddr: userConn.RemoteAddr().String(),
}
_, err := rc.PluginManager.NewUserConn(content)
if err != nil {
xl.Warn("the user conn [%s] was rejected, err:%v", content.RemoteAddr, err)
return
}
2019-01-15 00:11:08 +08:00
// try all connections from the pool
2019-03-29 19:01:18 +08:00
workConn, err := pxy.GetWorkConnFromPool(userConn.RemoteAddr(), userConn.LocalAddr())
2019-01-15 00:11:08 +08:00
if err != nil {
return
}
defer workConn.Close()
var local io.ReadWriteCloser = workConn
xl.Trace("handler user tcp connection, use_encryption: %t, use_compression: %t",
cfg.Transport.UseEncryption, cfg.Transport.UseCompression)
if cfg.Transport.UseEncryption {
local, err = libio.WithEncryption(local, []byte(serverCfg.Auth.Token))
2019-01-15 00:11:08 +08:00
if err != nil {
2019-10-12 20:13:12 +08:00
xl.Error("create encryption stream error: %v", err)
2019-01-15 00:11:08 +08:00
return
}
}
if cfg.Transport.UseCompression {
var recycleFn func()
local, recycleFn = libio.WithCompressionFromPool(local)
defer recycleFn()
2019-01-15 00:11:08 +08:00
}
if pxy.GetLimiter() != nil {
2023-05-29 14:10:34 +08:00
local = libio.WrapReadWriteCloser(limit.NewReader(local, pxy.GetLimiter()), limit.NewWriter(local, pxy.GetLimiter()), func() error {
return local.Close()
})
}
2019-10-12 20:13:12 +08:00
xl.Debug("join connections, workConn(l[%s] r[%s]) userConn(l[%s] r[%s])", workConn.LocalAddr().String(),
2019-01-15 00:11:08 +08:00
workConn.RemoteAddr().String(), userConn.LocalAddr().String(), userConn.RemoteAddr().String())
name := pxy.GetName()
proxyType := cfg.Type
metrics.Server.OpenConnection(name, proxyType)
2023-05-29 14:10:34 +08:00
inCount, outCount, _ := libio.Join(local, userConn)
metrics.Server.CloseConnection(name, proxyType)
metrics.Server.AddTrafficIn(name, proxyType, inCount)
metrics.Server.AddTrafficOut(name, proxyType, outCount)
2019-10-12 20:13:12 +08:00
xl.Debug("join connections closed")
2019-01-15 00:11:08 +08:00
}
2023-09-20 15:18:50 +08:00
type Options struct {
UserInfo plugin.UserInfo
LoginMsg *msg.Login
PoolCount int
ResourceController *controller.ResourceController
GetWorkConnFn GetWorkConnFn
Configurer v1.ProxyConfigurer
ServerCfg *v1.ServerConfig
}
func NewProxy(ctx context.Context, options *Options) (pxy Proxy, err error) {
configurer := options.Configurer
xl := xlog.FromContextSafe(ctx).Spawn().AppendPrefix(configurer.GetBaseConfig().Name)
var limiter *rate.Limiter
limitBytes := configurer.GetBaseConfig().Transport.BandwidthLimit.Bytes()
if limitBytes > 0 && configurer.GetBaseConfig().Transport.BandwidthLimitMode == types.BandwidthLimitModeServer {
limiter = rate.NewLimiter(rate.Limit(float64(limitBytes)), int(limitBytes))
}
basePxy := BaseProxy{
name: configurer.GetBaseConfig().Name,
2023-09-20 15:18:50 +08:00
rc: options.ResourceController,
listeners: make([]net.Listener, 0),
2023-09-20 15:18:50 +08:00
poolCount: options.PoolCount,
getWorkConnFn: options.GetWorkConnFn,
serverCfg: options.ServerCfg,
limiter: limiter,
xl: xl,
ctx: xlog.NewContext(ctx, xl),
2023-09-20 15:18:50 +08:00
userInfo: options.UserInfo,
loginMsg: options.LoginMsg,
configurer: configurer,
}
factory := proxyFactoryRegistry[reflect.TypeOf(configurer)]
if factory == nil {
return pxy, fmt.Errorf("proxy type not support")
}
pxy = factory(&basePxy)
if pxy == nil {
return nil, fmt.Errorf("proxy not created")
}
return pxy, nil
}
2020-05-24 17:48:37 +08:00
type Manager struct {
2019-01-15 00:11:08 +08:00
// proxies indexed by proxy name
pxys map[string]Proxy
mu sync.RWMutex
}
2020-05-24 17:48:37 +08:00
func NewManager() *Manager {
return &Manager{
2019-01-15 00:11:08 +08:00
pxys: make(map[string]Proxy),
}
}
2020-05-24 17:48:37 +08:00
func (pm *Manager) Add(name string, pxy Proxy) error {
2019-01-15 00:11:08 +08:00
pm.mu.Lock()
defer pm.mu.Unlock()
if _, ok := pm.pxys[name]; ok {
return fmt.Errorf("proxy name [%s] is already in use", name)
}
pm.pxys[name] = pxy
return nil
}
func (pm *Manager) Exist(name string) bool {
pm.mu.RLock()
defer pm.mu.RUnlock()
_, ok := pm.pxys[name]
return ok
}
2020-05-24 17:48:37 +08:00
func (pm *Manager) Del(name string) {
2019-01-15 00:11:08 +08:00
pm.mu.Lock()
defer pm.mu.Unlock()
delete(pm.pxys, name)
}
2020-05-24 17:48:37 +08:00
func (pm *Manager) GetByName(name string) (pxy Proxy, ok bool) {
2019-01-15 00:11:08 +08:00
pm.mu.RLock()
defer pm.mu.RUnlock()
pxy, ok = pm.pxys[name]
return
}