mirrors/Xray-core
1
0
Fork 0
mirror of https://github.com/XTLS/Xray-core.git synced 2025-05-18 01:58:41 +00:00
Code Issues Projects Releases Packages Wiki Activity

Experiment: pingpong seed

Browse source
This commit is contained in:
yuhan6665 2024-09-21 22:02:08 -04:00
parent 6a11542925
commit f0e6bdf4ce
11 changed files with 158 additions and 297 deletions
Download patch file Download diff file Expand all files Collapse all files

16
proxy/vless/encoding/addons.go
View file

@ -4,6 +4,7 @@ import (
"bytes"
"context"
"io"
"strings"
"github.com/xtls/xray-core/common/buf"
"github.com/xtls/xray-core/common/errors"
@ -54,15 +55,6 @@ func DecodeHeaderAddons(buffer *buf.Buffer, reader io.Reader) (*proxy.Addons, er
return addons, nil
}
// EncodeBodyAddons returns a Writer that auto-encrypt content written by caller.
func EncodeBodyAddons(writer buf.Writer, request *protocol.RequestHeader, addons *proxy.Addons, state *proxy.TrafficState, isUplink bool, context context.Context) buf.Writer {
w := proxy.NewVisionWriter(writer, addons, state, isUplink, context)
if request.Command == protocol.RequestCommandUDP {
return NewMultiLengthPacketWriter(w)
}
return w
}
// DecodeBodyAddons returns a Reader from which caller can fetch decrypted body.
func DecodeBodyAddons(reader io.Reader, request *protocol.RequestHeader, addons *proxy.Addons, state *proxy.TrafficState, isUplink bool, context context.Context) buf.Reader {
r := proxy.NewVisionReader(buf.NewReader(reader), addons, state, isUplink, context)
@ -181,7 +173,7 @@ func (r *LengthPacketReader) ReadMultiBuffer() (buf.MultiBuffer, error) {
func PopulateSeed(seed string, addons *proxy.Addons) {
if len(seed) > 0 {
addons.Seed = []byte {1} // only turn on, more TBD
addons.Mode = proxy.SeedMode_PaddingPlusDelay
addons.Mode = proxy.SeedMode_IndependentScheduler
addons.Duration = "0-8"
addons.Padding = &proxy.PaddingConfig{
RegularMin: 0,
@ -196,6 +188,7 @@ func PopulateSeed(seed string, addons *proxy.Addons) {
// }
addons.Scheduler = &proxy.SchedulerConfig{
TimeoutMillis: 600,
PingPong: strings.Contains(seed, "pingpong"),
}
} else if addons.Flow == vless.XRV {
addons.Seed = []byte {1} // only turn on, more TBD
@ -244,7 +237,8 @@ func CheckSeed(requestAddons *proxy.Addons, responseAddons *proxy.Addons) error
return errors.New("Delay of one is nil but the other is not nil")
}
if requestAddons.Scheduler != nil && responseAddons.Scheduler != nil {
if requestAddons.Scheduler.TimeoutMillis != responseAddons.Scheduler.TimeoutMillis {
if requestAddons.Scheduler.TimeoutMillis != responseAddons.Scheduler.TimeoutMillis ||
requestAddons.Scheduler.PingPong != responseAddons.Scheduler.PingPong {
return errors.New("Scheduler not match")
}
} else if requestAddons.Scheduler != nil || responseAddons.Scheduler != nil {

7
proxy/vless/encoding/encoding.go
View file

@ -172,7 +172,7 @@ func DecodeResponseHeader(reader io.Reader, request *protocol.RequestHeader) (*p
}
// XtlsRead filter and read xtls protocol
func XtlsRead(reader buf.Reader, writer buf.Writer, timer *signal.ActivityTimer, conn net.Conn, input *bytes.Reader, rawInput *bytes.Buffer, trafficState *proxy.TrafficState, ob *session.Outbound, isUplink bool, ctx context.Context) error {
func XtlsRead(reader buf.Reader, writer buf.Writer, timer *signal.ActivityTimer, scheduler *proxy.Scheduler, conn net.Conn, input *bytes.Reader, rawInput *bytes.Buffer, trafficState *proxy.TrafficState, ob *session.Outbound, isUplink bool, ctx context.Context) error {
err := func() error {
for {
if isUplink && trafficState.Inbound.UplinkReaderDirectCopy || !isUplink && trafficState.Outbound.DownlinkReaderDirectCopy {
@ -188,11 +188,10 @@ func XtlsRead(reader buf.Reader, writer buf.Writer, timer *signal.ActivityTimer,
ob.CanSpliceCopy = 1
}
}
return proxy.CopyRawConnIfExist(ctx, conn, writerConn, writer, timer, inTimer)
return proxy.CopyRawConnIfExist(ctx, conn, writerConn, writer, timer, inTimer, scheduler)
}
buffer, err := reader.ReadMultiBuffer()
if !buffer.IsEmpty() {
timer.Update()
if isUplink && trafficState.Inbound.UplinkReaderDirectCopy || !isUplink && trafficState.Outbound.DownlinkReaderDirectCopy {
// XTLS Vision processes struct TLS Conn's input and rawInput
if inputBuffer, err := buf.ReadFrom(input); err == nil {
@ -209,7 +208,9 @@ func XtlsRead(reader buf.Reader, writer buf.Writer, timer *signal.ActivityTimer,
if werr := writer.WriteMultiBuffer(buffer); werr != nil {
return werr
}
timer.Update()
}
scheduler.Trigger <- 2
if err != nil {
return err
}

17
proxy/vless/inbound/inbound.go
View file

@ -532,6 +532,12 @@ func (h *Handler) Process(ctx context.Context, network net.Network, connection s
serverReader := link.Reader // .(*pipe.Reader)
serverWriter := link.Writer // .(*pipe.Writer)
trafficState := proxy.NewTrafficState(account.ID.Bytes(), account.Flow)
bufferWriter := buf.NewBufferedWriter(buf.NewWriter(connection))
var clientWriter buf.Writer
v := proxy.NewVisionWriter(bufferWriter, requestAddons, trafficState, false, ctx)
scheduler := v.Scheduler
clientWriter = v
postRequest := func() error {
defer timer.SetTimeout(sessionPolicy.Timeouts.DownlinkOnly)
@ -542,10 +548,10 @@ func (h *Handler) Process(ctx context.Context, network net.Network, connection s
if requestAddons.Flow == vless.XRV {
ctx1 := session.ContextWithInbound(ctx, nil) // TODO enable splice
err = encoding.XtlsRead(clientReader, serverWriter, timer, connection, input, rawInput, trafficState, nil, true, ctx1)
err = encoding.XtlsRead(clientReader, serverWriter, timer, scheduler, connection, input, rawInput, trafficState, nil, true, ctx1)
} else {
// from clientReader.ReadMultiBuffer to serverWriter.WriteMultiBuffer
err = buf.Copy(clientReader, serverWriter, buf.UpdateActivity(timer))
err = buf.Copy(clientReader, serverWriter, buf.UpdateActivity(timer), proxy.TriggerScheduler(scheduler))
}
if err != nil {
@ -558,13 +564,14 @@ func (h *Handler) Process(ctx context.Context, network net.Network, connection s
getResponse := func() error {
defer timer.SetTimeout(sessionPolicy.Timeouts.UplinkOnly)
bufferWriter := buf.NewBufferedWriter(buf.NewWriter(connection))
if err := encoding.EncodeResponseHeader(bufferWriter, request, responseAddons); err != nil {
return errors.New("failed to encode response header").Base(err).AtWarning()
}
// default: clientWriter := bufferWriter
clientWriter := encoding.EncodeBodyAddons(bufferWriter, request, responseAddons, trafficState, false, ctx)
scheduler.Start()
if request.Command == protocol.RequestCommandUDP {
clientWriter = encoding.NewMultiLengthPacketWriter(clientWriter)
}
multiBuffer, err1 := serverReader.ReadMultiBuffer()
if err1 != nil {
return err1 // ...

24
proxy/vless/outbound/outbound.go
View file

@ -185,18 +185,23 @@ func (h *Handler) Process(ctx context.Context, link *transport.Link, dialer inte
request.Address = net.DomainAddress("v1.mux.cool")
request.Port = net.Port(666)
}
bufferWriter := buf.NewBufferedWriter(buf.NewWriter(conn))
var serverWriter buf.Writer
v := proxy.NewVisionWriter(bufferWriter, requestAddons, trafficState, true, ctx)
scheduler := v.Scheduler
serverWriter = v
postRequest := func() error {
defer timer.SetTimeout(sessionPolicy.Timeouts.DownlinkOnly)
bufferWriter := buf.NewBufferedWriter(buf.NewWriter(conn))
if err := encoding.EncodeRequestHeader(bufferWriter, request, requestAddons); err != nil {
return errors.New("failed to encode request header").Base(err).AtWarning()
}
// default: serverWriter := bufferWriter
serverWriter := encoding.EncodeBodyAddons(bufferWriter, request, requestAddons, trafficState, true, ctx)
if request.Command == protocol.RequestCommandMux && request.Port == 666 {
scheduler.Start()
if request.Command == protocol.RequestCommandUDP {
serverWriter = encoding.NewMultiLengthPacketWriter(serverWriter)
} else if request.Command == protocol.RequestCommandMux && request.Port == 666 {
serverWriter = xudp.NewPacketWriter(serverWriter, target, xudp.GetGlobalID(ctx))
}
timeoutReader, ok := clientReader.(buf.TimeoutReader)
@ -222,6 +227,11 @@ func (h *Handler) Process(ctx context.Context, link *transport.Link, dialer inte
if err := bufferWriter.SetBuffered(false); err != nil {
return errors.New("failed to write A request payload").Base(err).AtWarning()
}
if requestAddons.Scheduler != nil && requestAddons.Scheduler.PingPong {
go func() {
scheduler.Trigger <- 2 // client kickstart the pingpong!
}()
}
var err error
if requestAddons.Flow == vless.XRV {
@ -266,10 +276,10 @@ func (h *Handler) Process(ctx context.Context, link *transport.Link, dialer inte
}
if requestAddons.Flow == vless.XRV {
err = encoding.XtlsRead(serverReader, clientWriter, timer, conn, input, rawInput, trafficState, ob, false, ctx)
err = encoding.XtlsRead(serverReader, clientWriter, timer, scheduler, conn, input, rawInput, trafficState, ob, false, ctx)
} else {
// from serverReader.ReadMultiBuffer to clientWriter.WriteMultiBuffer
err = buf.Copy(serverReader, clientWriter, buf.UpdateActivity(timer))
err = buf.Copy(serverReader, clientWriter, buf.UpdateActivity(timer), proxy.TriggerScheduler(scheduler))
}
if err != nil {