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

334 lines
7.0 KiB
Go
Raw Normal View History

package inbound
import (
"context"
"io"
"sync"
"sync/atomic"
"time"
"v2ray.com/core"
"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-04-11 18:10:14 -04:00
"v2ray.com/core/common/session"
2018-02-08 09:39:46 -05:00
"v2ray.com/core/common/signal"
"v2ray.com/core/proxy"
"v2ray.com/core/transport/internet"
"v2ray.com/core/transport/internet/tcp"
"v2ray.com/core/transport/internet/udp"
)
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
stream *internet.StreamConfig
recvOrigDest bool
tag string
dispatcher core.Dispatcher
sniffers []proxyman.KnownProtocols
uplinkCounter core.StatCounter
downlinkCounter core.StatCounter
2018-02-08 09:39:46 -05:00
hub internet.Listener
}
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 {
2017-04-18 06:35:54 -04:00
dest, err := tcp.GetOriginalDestination(conn)
if err != nil {
2018-02-22 09:26:00 -05:00
newError("failed to get original destination").WithContext(ctx).Base(err).WriteToLog()
2017-04-18 06:35:54 -04:00
}
if dest.IsValid() {
2017-02-09 16:49:38 -05:00
ctx = proxy.ContextWithOriginalTarget(ctx, dest)
}
}
if len(w.tag) > 0 {
ctx = proxy.ContextWithInboundTag(ctx, w.tag)
}
ctx = proxy.ContextWithInboundEntryPoint(ctx, net.TCPDestination(w.address, w.port))
ctx = proxy.ContextWithSource(ctx, net.DestinationFromAddr(conn.RemoteAddr()))
if len(w.sniffers) > 0 {
ctx = proxyman.ContextWithProtocolSniffers(ctx, w.sniffers)
}
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 {
2018-02-22 09:26:00 -05:00
newError("connection ends").Base(err).WithContext(ctx).WriteToLog()
}
cancel()
2018-02-12 15:35:10 -05:00
if err := conn.Close(); err != nil {
2018-02-22 09:26:00 -05:00
newError("failed to close connection").Base(err).WithContext(ctx).WriteToLog()
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-02-08 09:39:46 -05:00
ctx := internet.ContextWithStreamSettings(context.Background(), w.stream)
hub, err := internet.ListenTCP(ctx, w.address, w.port, func(conn internet.Connection) {
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 {
2017-02-05 02:44:08 -05:00
if w.hub != nil {
2018-02-08 09:39:46 -05:00
common.Close(w.hub)
common.Close(w.proxy)
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
2017-01-28 10:27:46 -05:00
input chan *buf.Buffer
output func([]byte) (int, error)
remote net.Addr
local net.Addr
2018-02-08 09:39:46 -05:00
done *signal.Done
2018-04-11 18:10:14 -04:00
uplink core.StatCounter
downlink core.StatCounter
}
func (c *udpConn) updateActivity() {
atomic.StoreInt64(&c.lastActivityTime, time.Now().Unix())
}
func (c *udpConn) Read(buf []byte) (int, error) {
2018-02-02 16:35:18 -05:00
select {
case in := <-c.input:
defer in.Release()
c.updateActivity()
2018-04-11 18:10:14 -04:00
nBytes := copy(buf, in.Bytes())
if c.uplink != nil {
c.uplink.Add(int64(nBytes))
}
return nBytes, nil
2018-04-15 14:40:47 -04:00
case <-c.done.Wait():
return 0, io.EOF
}
}
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())
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
recvOrigDest bool
tag string
dispatcher core.Dispatcher
uplinkCounter core.StatCounter
downlinkCounter core.StatCounter
2018-02-08 09:39:46 -05:00
done *signal.Done
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()
2017-10-27 11:13:43 -04:00
if conn, found := w.activeConn[id]; found {
return conn, true
}
conn := &udpConn{
2017-01-28 10:27:46 -05:00
input: make(chan *buf.Buffer, 32),
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-04-11 18:10:14 -04:00
done: signal.NewDone(),
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)
2017-01-28 10:27:46 -05:00
select {
case conn.input <- b:
2018-04-15 14:40:47 -04:00
case <-conn.done.Wait():
2018-02-08 09:39:46 -05:00
b.Release()
2017-01-28 10:27:46 -05:00
default:
b.Release()
}
if !existing {
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() {
2017-02-09 16:49:38 -05:00
ctx = proxy.ContextWithOriginalTarget(ctx, originalDest)
}
if len(w.tag) > 0 {
ctx = proxy.ContextWithInboundTag(ctx, w.tag)
}
ctx = proxy.ContextWithSource(ctx, source)
ctx = proxy.ContextWithInboundEntryPoint(ctx, net.UDPDestination(w.address, w.port))
if err := w.proxy.Process(ctx, net.Network_UDP, conn, w.dispatcher); err != nil {
2017-12-19 15:28:12 -05:00
newError("connection ends").Base(err).WriteToLog()
}
2018-02-08 09:39:46 -05:00
conn.Close()
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()
}
func (w *udpWorker) Start() error {
2018-02-08 16:09:55 -05:00
w.activeConn = make(map[connID]*udpConn, 16)
2018-02-08 09:39:46 -05:00
w.done = signal.NewDone()
2018-02-12 09:08:20 -05:00
h, err := udp.ListenUDP(w.address, w.port, w.callback, udp.HubReceiveOriginalDestination(w.recvOrigDest), udp.HubCapacity(256))
if err != nil {
return err
}
2017-01-27 08:27:02 -05:00
go w.monitor()
w.hub = h
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()
2017-02-05 02:38:25 -05:00
if w.hub != nil {
w.hub.Close()
}
2018-02-12 15:35:10 -05:00
2018-02-14 07:41:21 -05:00
if w.done != nil {
common.Must(w.done.Close())
}
2018-02-12 15:35:10 -05:00
common.Close(w.proxy)
2018-02-08 09:39:46 -05:00
return nil
}
func (w *udpWorker) monitor() {
2017-05-08 18:01:15 -04:00
timer := time.NewTicker(time.Second * 16)
defer timer.Stop()
for {
select {
2018-04-15 14:40:47 -04:00
case <-w.done.Wait():
return
2017-05-08 18:01:15 -04:00
case <-timer.C:
nowSec := time.Now().Unix()
w.Lock()
for addr, conn := range w.activeConn {
2017-02-27 15:34:35 -05:00
if nowSec-atomic.LoadInt64(&conn.lastActivityTime) > 8 {
2017-01-27 08:27:02 -05:00
delete(w.activeConn, addr)
2018-02-08 09:39:46 -05:00
conn.Close()
}
}
2017-01-27 08:27:02 -05:00
w.Unlock()
}
}
}
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
}