mirror of
https://github.com/v2fly/v2ray-core.git
synced 2024-12-22 18:17:52 -05:00
281 lines
7.6 KiB
Go
281 lines
7.6 KiB
Go
package inbound
|
|
|
|
//go:generate go run $GOPATH/src/v2ray.com/core/tools/generrorgen/main.go -pkg inbound -path Proxy,VMess,Inbound
|
|
|
|
import (
|
|
"context"
|
|
"io"
|
|
"runtime"
|
|
"sync"
|
|
"time"
|
|
|
|
"v2ray.com/core/app"
|
|
"v2ray.com/core/app/dispatcher"
|
|
"v2ray.com/core/app/log"
|
|
"v2ray.com/core/app/proxyman"
|
|
"v2ray.com/core/common"
|
|
"v2ray.com/core/common/buf"
|
|
"v2ray.com/core/common/errors"
|
|
"v2ray.com/core/common/net"
|
|
"v2ray.com/core/common/protocol"
|
|
"v2ray.com/core/common/serial"
|
|
"v2ray.com/core/common/signal"
|
|
"v2ray.com/core/common/uuid"
|
|
"v2ray.com/core/proxy/vmess"
|
|
"v2ray.com/core/proxy/vmess/encoding"
|
|
"v2ray.com/core/transport/internet"
|
|
"v2ray.com/core/transport/ray"
|
|
)
|
|
|
|
type userByEmail struct {
|
|
sync.RWMutex
|
|
cache map[string]*protocol.User
|
|
defaultLevel uint32
|
|
defaultAlterIDs uint16
|
|
}
|
|
|
|
func NewUserByEmail(users []*protocol.User, config *DefaultConfig) *userByEmail {
|
|
cache := make(map[string]*protocol.User)
|
|
for _, user := range users {
|
|
cache[user.Email] = user
|
|
}
|
|
return &userByEmail{
|
|
cache: cache,
|
|
defaultLevel: config.Level,
|
|
defaultAlterIDs: uint16(config.AlterId),
|
|
}
|
|
}
|
|
|
|
func (v *userByEmail) Get(email string) (*protocol.User, bool) {
|
|
var user *protocol.User
|
|
var found bool
|
|
v.RLock()
|
|
user, found = v.cache[email]
|
|
v.RUnlock()
|
|
if !found {
|
|
v.Lock()
|
|
user, found = v.cache[email]
|
|
if !found {
|
|
account := &vmess.Account{
|
|
Id: uuid.New().String(),
|
|
AlterId: uint32(v.defaultAlterIDs),
|
|
}
|
|
user = &protocol.User{
|
|
Level: v.defaultLevel,
|
|
Email: email,
|
|
Account: serial.ToTypedMessage(account),
|
|
}
|
|
v.cache[email] = user
|
|
}
|
|
v.Unlock()
|
|
}
|
|
return user, found
|
|
}
|
|
|
|
// Handler is an inbound connection handler that handles messages in VMess protocol.
|
|
type Handler struct {
|
|
inboundHandlerManager proxyman.InboundHandlerManager
|
|
clients protocol.UserValidator
|
|
usersByEmail *userByEmail
|
|
detours *DetourConfig
|
|
sessionHistory *encoding.SessionHistory
|
|
}
|
|
|
|
func New(ctx context.Context, config *Config) (*Handler, error) {
|
|
space := app.SpaceFromContext(ctx)
|
|
if space == nil {
|
|
return nil, newError("no space in context")
|
|
}
|
|
|
|
allowedClients := vmess.NewTimedUserValidator(ctx, protocol.DefaultIDHash)
|
|
for _, user := range config.User {
|
|
allowedClients.Add(user)
|
|
}
|
|
|
|
handler := &Handler{
|
|
clients: allowedClients,
|
|
detours: config.Detour,
|
|
usersByEmail: NewUserByEmail(config.User, config.GetDefaultValue()),
|
|
sessionHistory: encoding.NewSessionHistory(ctx),
|
|
}
|
|
|
|
space.OnInitialize(func() error {
|
|
handler.inboundHandlerManager = proxyman.InboundHandlerManagerFromSpace(space)
|
|
if handler.inboundHandlerManager == nil {
|
|
return newError("InboundHandlerManager is not found is space")
|
|
}
|
|
return nil
|
|
})
|
|
|
|
return handler, nil
|
|
}
|
|
|
|
// Network implements proxy.Inbound.Network().
|
|
func (*Handler) Network() net.NetworkList {
|
|
return net.NetworkList{
|
|
Network: []net.Network{net.Network_TCP},
|
|
}
|
|
}
|
|
|
|
func (v *Handler) GetUser(email string) *protocol.User {
|
|
user, existing := v.usersByEmail.Get(email)
|
|
if !existing {
|
|
v.clients.Add(user)
|
|
}
|
|
return user
|
|
}
|
|
|
|
func transferRequest(timer signal.ActivityTimer, session *encoding.ServerSession, request *protocol.RequestHeader, input io.Reader, output ray.OutputStream) error {
|
|
defer output.Close()
|
|
|
|
bodyReader := session.DecodeRequestBody(request, input)
|
|
if err := buf.PipeUntilEOF(timer, bodyReader, output); err != nil {
|
|
return err
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func transferResponse(timer signal.ActivityTimer, session *encoding.ServerSession, request *protocol.RequestHeader, response *protocol.ResponseHeader, input buf.Reader, output io.Writer) error {
|
|
session.EncodeResponseHeader(response, output)
|
|
|
|
bodyWriter := session.EncodeResponseBody(request, output)
|
|
|
|
var reader buf.Reader = input
|
|
if request.Command == protocol.RequestCommandTCP {
|
|
reader = buf.NewMergingReader(input)
|
|
}
|
|
// Optimize for small response packet
|
|
data, err := reader.Read()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
if err := bodyWriter.Write(data); err != nil {
|
|
return err
|
|
}
|
|
data.Release()
|
|
|
|
if bufferedWriter, ok := output.(*buf.BufferedWriter); ok {
|
|
if err := bufferedWriter.SetBuffered(false); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
|
|
if err := buf.PipeUntilEOF(timer, reader, bodyWriter); err != nil {
|
|
return err
|
|
}
|
|
|
|
if request.Option.Has(protocol.RequestOptionChunkStream) {
|
|
if err := bodyWriter.Write(buf.NewLocal(8)); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// Process implements proxy.Inbound.Process().
|
|
func (v *Handler) Process(ctx context.Context, network net.Network, connection internet.Connection, dispatcher dispatcher.Interface) error {
|
|
connection.SetReadDeadline(time.Now().Add(time.Second * 8))
|
|
reader := buf.NewBufferedReader(connection)
|
|
|
|
session := encoding.NewServerSession(v.clients, v.sessionHistory)
|
|
request, err := session.DecodeRequestHeader(reader)
|
|
|
|
if err != nil {
|
|
if errors.Cause(err) != io.EOF {
|
|
log.Access(connection.RemoteAddr(), "", log.AccessRejected, err)
|
|
log.Trace(newError("invalid request from ", connection.RemoteAddr(), ": ", err))
|
|
}
|
|
return err
|
|
}
|
|
log.Access(connection.RemoteAddr(), request.Destination(), log.AccessAccepted, "")
|
|
log.Trace(newError("received request for ", request.Destination()))
|
|
|
|
connection.SetReadDeadline(time.Time{})
|
|
|
|
userSettings := request.User.GetSettings()
|
|
|
|
ctx = protocol.ContextWithUser(ctx, request.User)
|
|
|
|
ctx, timer := signal.CancelAfterInactivity(ctx, userSettings.PayloadTimeout)
|
|
ray, err := dispatcher.Dispatch(ctx, request.Destination())
|
|
if err != nil {
|
|
return newError("failed to dispatch request to ", request.Destination()).Base(err)
|
|
}
|
|
|
|
input := ray.InboundInput()
|
|
output := ray.InboundOutput()
|
|
|
|
reader.SetBuffered(false)
|
|
|
|
requestDone := signal.ExecuteAsync(func() error {
|
|
return transferRequest(timer, session, request, reader, input)
|
|
})
|
|
|
|
writer := buf.NewBufferedWriter(connection)
|
|
response := &protocol.ResponseHeader{
|
|
Command: v.generateCommand(ctx, request),
|
|
}
|
|
|
|
responseDone := signal.ExecuteAsync(func() error {
|
|
return transferResponse(timer, session, request, response, output, writer)
|
|
})
|
|
|
|
if err := signal.ErrorOrFinish2(ctx, requestDone, responseDone); err != nil {
|
|
input.CloseError()
|
|
output.CloseError()
|
|
return newError("connection ends").Base(err)
|
|
}
|
|
|
|
if err := writer.Flush(); err != nil {
|
|
return newError("error during flushing remaining data").Base(err)
|
|
}
|
|
|
|
runtime.KeepAlive(timer)
|
|
|
|
return nil
|
|
}
|
|
|
|
func (v *Handler) generateCommand(ctx context.Context, request *protocol.RequestHeader) protocol.ResponseCommand {
|
|
if v.detours != nil {
|
|
tag := v.detours.To
|
|
if v.inboundHandlerManager != nil {
|
|
handler, err := v.inboundHandlerManager.GetHandler(ctx, tag)
|
|
if err != nil {
|
|
log.Trace(newError("failed to get detour handler: ", tag, err).AtWarning())
|
|
return nil
|
|
}
|
|
proxyHandler, port, availableMin := handler.GetRandomInboundProxy()
|
|
inboundHandler, ok := proxyHandler.(*Handler)
|
|
if ok && inboundHandler != nil {
|
|
if availableMin > 255 {
|
|
availableMin = 255
|
|
}
|
|
|
|
log.Trace(newError("pick detour handler for port ", port, " for ", availableMin, " minutes.").AtDebug())
|
|
user := inboundHandler.GetUser(request.User.Email)
|
|
if user == nil {
|
|
return nil
|
|
}
|
|
account, _ := user.GetTypedAccount()
|
|
return &protocol.CommandSwitchAccount{
|
|
Port: port,
|
|
ID: account.(*vmess.InternalAccount).ID.UUID(),
|
|
AlterIds: uint16(len(account.(*vmess.InternalAccount).AlterIDs)),
|
|
Level: user.Level,
|
|
ValidMin: byte(availableMin),
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func init() {
|
|
common.Must(common.RegisterConfig((*Config)(nil), func(ctx context.Context, config interface{}) (interface{}, error) {
|
|
return New(ctx, config.(*Config))
|
|
}))
|
|
}
|