1
0
mirror of https://github.com/v2fly/v2ray-core.git synced 2024-12-23 02:26:49 -05:00
v2fly/app/proxyman/inbound/worker.go

396 lines
8.7 KiB
Go
Raw Normal View History

package inbound
import (
"context"
"sync"
"sync/atomic"
"time"
"v2ray.com/core/app/proxyman"
2018-02-08 09:39:46 -05:00
"v2ray.com/core/common"
"v2ray.com/core/common/buf"
"v2ray.com/core/common/net"
2018-05-31 05:55:11 -04:00
"v2ray.com/core/common/serial"
2018-04-11 18:10:14 -04:00
"v2ray.com/core/common/session"
2018-05-27 08:42:53 -04:00
"v2ray.com/core/common/signal/done"
2018-05-31 05:55:11 -04:00
"v2ray.com/core/common/task"
"v2ray.com/core/features/routing"
"v2ray.com/core/features/stats"
"v2ray.com/core/proxy"
"v2ray.com/core/transport/internet"
"v2ray.com/core/transport/internet/tcp"
"v2ray.com/core/transport/internet/udp"
2018-09-02 18:56:43 -04:00
"v2ray.com/core/transport/pipe"
)
type worker interface {
Start() error
2018-02-08 09:39:46 -05:00
Close() error
Port() net.Port
2017-01-26 14:57:18 -05:00
Proxy() proxy.Inbound
}
type tcpWorker struct {
2018-04-11 18:10:14 -04:00
address net.Address
port net.Port
proxy proxy.Inbound
2018-09-07 08:50:25 -04:00
stream *internet.MemoryStreamConfig
2018-04-11 18:10:14 -04:00
recvOrigDest bool
tag string
dispatcher routing.Dispatcher
2018-07-16 07:47:00 -04:00
sniffingConfig *proxyman.SniffingConfig
uplinkCounter stats.Counter
downlinkCounter stats.Counter
2018-02-08 09:39:46 -05:00
hub internet.Listener
}
2018-09-23 14:14:07 -04:00
func getTProxyType(s *internet.MemoryStreamConfig) internet.SocketConfig_TProxyMode {
if s == nil || s.SocketSettings == nil {
return internet.SocketConfig_Off
}
return s.SocketSettings.Tproxy
}
func (w *tcpWorker) callback(conn internet.Connection) {
2018-02-08 09:39:46 -05:00
ctx, cancel := context.WithCancel(context.Background())
2018-02-22 09:26:00 -05:00
sid := session.NewID()
ctx = session.ContextWithID(ctx, sid)
if w.recvOrigDest {
2018-09-23 14:14:07 -04:00
var dest net.Destination
switch getTProxyType(w.stream) {
case internet.SocketConfig_Redirect:
d, err := tcp.GetOriginalDestination(conn)
if err != nil {
newError("failed to get original destination").Base(err).WriteToLog(session.ExportIDToError(ctx))
} else {
dest = d
}
case internet.SocketConfig_TProxy:
dest = net.DestinationFromAddr(conn.LocalAddr())
2017-04-18 06:35:54 -04:00
}
if dest.IsValid() {
ctx = session.ContextWithOutbound(ctx, &session.Outbound{
Target: dest,
})
}
}
ctx = session.ContextWithInbound(ctx, &session.Inbound{
Source: net.DestinationFromAddr(conn.RemoteAddr()),
Gateway: net.TCPDestination(w.address, w.port),
Tag: w.tag,
})
2018-07-16 07:47:00 -04:00
if w.sniffingConfig != nil {
ctx = proxyman.ContextWithSniffingConfig(ctx, w.sniffingConfig)
}
2018-04-11 18:10:14 -04:00
if w.uplinkCounter != nil || w.downlinkCounter != nil {
conn = &internet.StatCouterConnection{
Connection: conn,
Uplink: w.uplinkCounter,
Downlink: w.downlinkCounter,
}
}
if err := w.proxy.Process(ctx, net.Network_TCP, conn, w.dispatcher); err != nil {
newError("connection ends").Base(err).WriteToLog(session.ExportIDToError(ctx))
}
cancel()
2018-02-12 15:35:10 -05:00
if err := conn.Close(); err != nil {
newError("failed to close connection").Base(err).WriteToLog(session.ExportIDToError(ctx))
2018-02-12 15:35:10 -05:00
}
}
2017-01-26 14:57:18 -05:00
func (w *tcpWorker) Proxy() proxy.Inbound {
return w.proxy
}
func (w *tcpWorker) Start() error {
2018-11-21 08:55:05 -05:00
ctx := context.Background()
hub, err := internet.ListenTCP(ctx, w.address, w.port, w.stream, func(conn internet.Connection) {
2018-02-08 09:39:46 -05:00
go w.callback(conn)
})
if err != nil {
2017-12-27 16:25:12 -05:00
return newError("failed to listen TCP on ", w.port).AtWarning().Base(err)
}
w.hub = hub
return nil
}
2018-02-08 09:39:46 -05:00
func (w *tcpWorker) Close() error {
2018-05-31 05:55:11 -04:00
var errors []interface{}
2017-02-05 02:44:08 -05:00
if w.hub != nil {
2018-05-31 05:55:11 -04:00
if err := common.Close(w.hub); err != nil {
errors = append(errors, err)
}
if err := common.Close(w.proxy); err != nil {
errors = append(errors, err)
}
}
if len(errors) > 0 {
return newError("failed to close all resources").Base(newError(serial.Concat(errors...)))
2017-02-05 02:44:08 -05:00
}
2018-02-08 09:39:46 -05:00
return nil
}
func (w *tcpWorker) Port() net.Port {
return w.port
}
type udpConn struct {
lastActivityTime int64 // in seconds
2018-09-02 18:56:43 -04:00
reader buf.Reader
writer buf.Writer
output func([]byte) (int, error)
remote net.Addr
local net.Addr
2018-05-27 08:42:53 -04:00
done *done.Instance
uplink stats.Counter
downlink stats.Counter
}
func (c *udpConn) updateActivity() {
atomic.StoreInt64(&c.lastActivityTime, time.Now().Unix())
}
2018-05-26 19:19:05 -04:00
// ReadMultiBuffer implements buf.Reader
func (c *udpConn) ReadMultiBuffer() (buf.MultiBuffer, error) {
2018-09-02 18:56:43 -04:00
mb, err := c.reader.ReadMultiBuffer()
if err != nil {
return nil, err
2018-05-26 19:19:05 -04:00
}
2018-09-01 15:19:01 -04:00
c.updateActivity()
if c.uplink != nil {
2018-09-02 18:56:43 -04:00
c.uplink.Add(int64(mb.Len()))
}
2018-09-02 18:56:43 -04:00
return mb, nil
2018-05-26 19:19:05 -04:00
}
func (c *udpConn) Read(buf []byte) (int, error) {
2018-09-02 18:56:43 -04:00
panic("not implemented")
}
2017-04-21 09:36:05 -04:00
// Write implements io.Writer.
func (c *udpConn) Write(buf []byte) (int, error) {
n, err := c.output(buf)
2018-04-11 18:10:14 -04:00
if c.downlink != nil {
c.downlink.Add(int64(n))
}
if err == nil {
c.updateActivity()
}
return n, err
}
func (c *udpConn) Close() error {
2018-02-12 15:35:10 -05:00
common.Must(c.done.Close())
2018-09-02 18:56:43 -04:00
common.Must(common.Close(c.writer))
return nil
}
func (c *udpConn) RemoteAddr() net.Addr {
return c.remote
}
func (c *udpConn) LocalAddr() net.Addr {
return c.remote
}
func (*udpConn) SetDeadline(time.Time) error {
return nil
}
func (*udpConn) SetReadDeadline(time.Time) error {
return nil
}
func (*udpConn) SetWriteDeadline(time.Time) error {
return nil
}
2018-02-08 16:09:55 -05:00
type connID struct {
2017-10-27 11:13:43 -04:00
src net.Destination
dest net.Destination
}
type udpWorker struct {
sync.RWMutex
2018-04-11 18:10:14 -04:00
proxy proxy.Inbound
hub *udp.Hub
address net.Address
port net.Port
tag string
2018-09-17 09:12:58 -04:00
stream *internet.MemoryStreamConfig
dispatcher routing.Dispatcher
uplinkCounter stats.Counter
downlinkCounter stats.Counter
2018-05-31 05:55:11 -04:00
checker *task.Periodic
2018-02-08 16:09:55 -05:00
activeConn map[connID]*udpConn
}
2018-02-08 16:09:55 -05:00
func (w *udpWorker) getConnection(id connID) (*udpConn, bool) {
w.Lock()
defer w.Unlock()
2018-05-26 19:19:05 -04:00
if conn, found := w.activeConn[id]; found && !conn.done.Done() {
return conn, true
}
2018-09-02 18:56:43 -04:00
pReader, pWriter := pipe.New(pipe.DiscardOverflow(), pipe.WithSizeLimit(16*1024))
conn := &udpConn{
2018-09-02 18:56:43 -04:00
reader: pReader,
writer: pWriter,
output: func(b []byte) (int, error) {
2017-10-27 11:13:43 -04:00
return w.hub.WriteTo(b, id.src)
},
remote: &net.UDPAddr{
2017-10-27 11:13:43 -04:00
IP: id.src.Address.IP(),
Port: int(id.src.Port),
},
local: &net.UDPAddr{
IP: w.address.IP(),
Port: int(w.port),
},
2018-05-27 08:42:53 -04:00
done: done.New(),
2018-04-11 18:10:14 -04:00
uplink: w.uplinkCounter,
downlink: w.downlinkCounter,
}
2017-10-27 11:13:43 -04:00
w.activeConn[id] = conn
conn.updateActivity()
return conn, false
}
func (w *udpWorker) callback(b *buf.Buffer, source net.Destination, originalDest net.Destination) {
2018-02-08 16:09:55 -05:00
id := connID{
2018-03-01 08:58:13 -05:00
src: source,
}
if originalDest.IsValid() {
id.dest = originalDest
2017-10-27 11:13:43 -04:00
}
conn, existing := w.getConnection(id)
2018-09-02 18:56:43 -04:00
// payload will be discarded in pipe is full.
2018-11-16 05:08:12 -05:00
conn.writer.WriteMultiBuffer(buf.MultiBuffer{b}) // nolint: errcheck
if !existing {
common.Must(w.checker.Start())
go func() {
2018-02-08 09:39:46 -05:00
ctx := context.Background()
2018-02-22 09:26:00 -05:00
sid := session.NewID()
ctx = session.ContextWithID(ctx, sid)
if originalDest.IsValid() {
ctx = session.ContextWithOutbound(ctx, &session.Outbound{
Target: originalDest,
})
}
ctx = session.ContextWithInbound(ctx, &session.Inbound{
Source: source,
Gateway: net.UDPDestination(w.address, w.port),
Tag: w.tag,
})
if err := w.proxy.Process(ctx, net.Network_UDP, conn, w.dispatcher); err != nil {
newError("connection ends").Base(err).WriteToLog(session.ExportIDToError(ctx))
}
2018-05-31 05:55:11 -04:00
conn.Close() // nolint: errcheck
2017-10-27 11:13:43 -04:00
w.removeConn(id)
}()
}
}
2018-02-08 16:09:55 -05:00
func (w *udpWorker) removeConn(id connID) {
w.Lock()
2017-10-27 11:13:43 -04:00
delete(w.activeConn, id)
w.Unlock()
}
2018-08-11 16:35:01 -04:00
func (w *udpWorker) handlePackets() {
receive := w.hub.Receive()
for payload := range receive {
w.callback(payload.Content, payload.Source, payload.OriginalDestination)
}
}
func (w *udpWorker) clean() error {
nowSec := time.Now().Unix()
w.Lock()
defer w.Unlock()
if len(w.activeConn) == 0 {
return newError("no more connections. stopping...")
}
for addr, conn := range w.activeConn {
if nowSec-atomic.LoadInt64(&conn.lastActivityTime) > 8 {
delete(w.activeConn, addr)
conn.Close() // nolint: errcheck
}
}
if len(w.activeConn) == 0 {
w.activeConn = make(map[connID]*udpConn, 16)
}
return nil
}
func (w *udpWorker) Start() error {
2018-02-08 16:09:55 -05:00
w.activeConn = make(map[connID]*udpConn, 16)
2018-11-21 08:55:05 -05:00
ctx := context.Background()
h, err := udp.ListenUDP(ctx, w.address, w.port, w.stream, udp.HubCapacity(256))
if err != nil {
return err
}
2018-05-31 05:55:11 -04:00
w.checker = &task.Periodic{
Interval: time.Second * 16,
Execute: w.clean,
2018-05-31 05:55:11 -04:00
}
w.hub = h
2018-08-11 16:35:01 -04:00
go w.handlePackets()
return nil
}
2018-02-08 09:39:46 -05:00
func (w *udpWorker) Close() error {
2018-02-12 15:35:10 -05:00
w.Lock()
defer w.Unlock()
2018-05-31 05:55:11 -04:00
var errors []interface{}
2017-02-05 02:38:25 -05:00
if w.hub != nil {
2018-05-31 05:55:11 -04:00
if err := w.hub.Close(); err != nil {
errors = append(errors, err)
}
2017-02-05 02:38:25 -05:00
}
2018-02-12 15:35:10 -05:00
2018-05-31 05:55:11 -04:00
if w.checker != nil {
if err := w.checker.Close(); err != nil {
errors = append(errors, err)
}
2018-02-14 07:41:21 -05:00
}
2018-05-31 05:55:11 -04:00
if err := common.Close(w.proxy); err != nil {
errors = append(errors, err)
}
2017-05-08 18:01:15 -04:00
2018-05-31 05:55:11 -04:00
if len(errors) > 0 {
return newError("failed to close all resources").Base(newError(serial.Concat(errors...)))
}
2018-05-31 05:55:11 -04:00
return nil
}
func (w *udpWorker) Port() net.Port {
return w.port
}
2017-01-26 14:57:18 -05:00
func (w *udpWorker) Proxy() proxy.Inbound {
return w.proxy
}