mirror of
https://github.com/v2fly/v2ray-core.git
synced 2024-12-22 18:17:52 -05:00
170 lines
4.4 KiB
Go
170 lines
4.4 KiB
Go
package inbound
|
|
|
|
import (
|
|
"context"
|
|
|
|
"v2ray.com/core"
|
|
"v2ray.com/core/app/proxyman"
|
|
"v2ray.com/core/common"
|
|
"v2ray.com/core/common/dice"
|
|
"v2ray.com/core/common/mux"
|
|
"v2ray.com/core/common/net"
|
|
"v2ray.com/core/common/serial"
|
|
"v2ray.com/core/features/policy"
|
|
"v2ray.com/core/features/stats"
|
|
"v2ray.com/core/proxy"
|
|
"v2ray.com/core/transport/internet"
|
|
)
|
|
|
|
func getStatCounter(v *core.Instance, tag string) (stats.Counter, stats.Counter) {
|
|
var uplinkCounter stats.Counter
|
|
var downlinkCounter stats.Counter
|
|
|
|
policy := v.GetFeature(policy.ManagerType()).(policy.Manager)
|
|
if len(tag) > 0 && policy.ForSystem().Stats.InboundUplink {
|
|
statsManager := v.GetFeature(stats.ManagerType()).(stats.Manager)
|
|
name := "inbound>>>" + tag + ">>>traffic>>>uplink"
|
|
c, _ := stats.GetOrRegisterCounter(statsManager, name)
|
|
if c != nil {
|
|
uplinkCounter = c
|
|
}
|
|
}
|
|
if len(tag) > 0 && policy.ForSystem().Stats.InboundDownlink {
|
|
statsManager := v.GetFeature(stats.ManagerType()).(stats.Manager)
|
|
name := "inbound>>>" + tag + ">>>traffic>>>downlink"
|
|
c, _ := stats.GetOrRegisterCounter(statsManager, name)
|
|
if c != nil {
|
|
downlinkCounter = c
|
|
}
|
|
}
|
|
|
|
return uplinkCounter, downlinkCounter
|
|
}
|
|
|
|
type AlwaysOnInboundHandler struct {
|
|
proxy proxy.Inbound
|
|
workers []worker
|
|
mux *mux.Server
|
|
tag string
|
|
}
|
|
|
|
func NewAlwaysOnInboundHandler(ctx context.Context, tag string, receiverConfig *proxyman.ReceiverConfig, proxyConfig interface{}) (*AlwaysOnInboundHandler, error) {
|
|
rawProxy, err := common.CreateObject(ctx, proxyConfig)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
p, ok := rawProxy.(proxy.Inbound)
|
|
if !ok {
|
|
return nil, newError("not an inbound proxy.")
|
|
}
|
|
|
|
h := &AlwaysOnInboundHandler{
|
|
proxy: p,
|
|
mux: mux.NewServer(ctx),
|
|
tag: tag,
|
|
}
|
|
|
|
uplinkCounter, downlinkCounter := getStatCounter(core.MustFromContext(ctx), tag)
|
|
|
|
nl := p.Network()
|
|
pr := receiverConfig.PortRange
|
|
address := receiverConfig.Listen.AsAddress()
|
|
if address == nil {
|
|
address = net.AnyIP
|
|
}
|
|
|
|
mss, err := internet.ToMemoryStreamConfig(receiverConfig.StreamSettings)
|
|
if err != nil {
|
|
return nil, newError("failed to parse stream config").Base(err).AtWarning()
|
|
}
|
|
|
|
if receiverConfig.ReceiveOriginalDestination {
|
|
if mss.SocketSettings == nil {
|
|
mss.SocketSettings = &internet.SocketConfig{}
|
|
}
|
|
if mss.SocketSettings.Tproxy == internet.SocketConfig_Off {
|
|
mss.SocketSettings.Tproxy = internet.SocketConfig_Redirect
|
|
}
|
|
mss.SocketSettings.ReceiveOriginalDestAddress = true
|
|
}
|
|
|
|
for port := pr.From; port <= pr.To; port++ {
|
|
if nl.HasNetwork(net.Network_TCP) {
|
|
newError("creating stream worker on ", address, ":", port).AtDebug().WriteToLog()
|
|
|
|
worker := &tcpWorker{
|
|
address: address,
|
|
port: net.Port(port),
|
|
proxy: p,
|
|
stream: mss,
|
|
recvOrigDest: receiverConfig.ReceiveOriginalDestination,
|
|
tag: tag,
|
|
dispatcher: h.mux,
|
|
sniffingConfig: receiverConfig.GetEffectiveSniffingSettings(),
|
|
uplinkCounter: uplinkCounter,
|
|
downlinkCounter: downlinkCounter,
|
|
}
|
|
h.workers = append(h.workers, worker)
|
|
}
|
|
|
|
if nl.HasNetwork(net.Network_UDP) {
|
|
worker := &udpWorker{
|
|
tag: tag,
|
|
proxy: p,
|
|
address: address,
|
|
port: net.Port(port),
|
|
dispatcher: h.mux,
|
|
uplinkCounter: uplinkCounter,
|
|
downlinkCounter: downlinkCounter,
|
|
stream: mss,
|
|
}
|
|
h.workers = append(h.workers, worker)
|
|
}
|
|
}
|
|
|
|
return h, nil
|
|
}
|
|
|
|
// Start implements common.Runnable.
|
|
func (h *AlwaysOnInboundHandler) Start() error {
|
|
for _, worker := range h.workers {
|
|
if err := worker.Start(); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// Close implements common.Closable.
|
|
func (h *AlwaysOnInboundHandler) Close() error {
|
|
var errors []interface{}
|
|
for _, worker := range h.workers {
|
|
if err := worker.Close(); err != nil {
|
|
errors = append(errors, err)
|
|
}
|
|
}
|
|
if err := h.mux.Close(); err != nil {
|
|
errors = append(errors, err)
|
|
}
|
|
if len(errors) > 0 {
|
|
return newError("failed to close all resources").Base(newError(serial.Concat(errors...)))
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (h *AlwaysOnInboundHandler) GetRandomInboundProxy() (interface{}, net.Port, int) {
|
|
if len(h.workers) == 0 {
|
|
return nil, 0, 0
|
|
}
|
|
w := h.workers[dice.Roll(len(h.workers))]
|
|
return w.Proxy(), w.Port(), 9999
|
|
}
|
|
|
|
func (h *AlwaysOnInboundHandler) Tag() string {
|
|
return h.tag
|
|
}
|
|
|
|
func (h *AlwaysOnInboundHandler) GetInbound() proxy.Inbound {
|
|
return h.proxy
|
|
}
|