XHTTP: packet-down in stream-down with downlink keep-alive

infra/conf/transport_method.go

@@ -292,6 +292,7 @@ type SplitHTTPConfig struct {
 	ScStreamUpServerSecs Int32Range        `json:"scStreamUpServerSecs"`
 	ServerMaxHeaderBytes int32             `json:"serverMaxHeaderBytes"`
 	Xmux                 XmuxConfig        `json:"xmux"`
+	PacketDown           *PacketDownConfig `json:"packetDown"`
 	DownloadSettings     *StreamConfig     `json:"downloadSettings"`
 	Extra                json.RawMessage   `json:"extra"`
 }
@@ -305,6 +306,12 @@ type XmuxConfig struct {
 	HKeepAlivePeriod int64      `json:"hKeepAlivePeriod"`
 }
 
+type PacketDownConfig struct {
+	KeepAlive       bool       `json:"keepAlive"`
+	KeepAlivePeriod Int32Range `json:"keepAlivePeriod"`
+	KeepAliveLength Int32Range `json:"keepAliveLength"`
+}
+
 func newRangeConfig(input Int32Range) *splithttp.RangeConfig {
 	return &splithttp.RangeConfig{
 		From: input.From,
@@ -505,6 +512,17 @@ func (c *SplitHTTPConfig) Build() (proto.Message, error) {
 		},
 	}
 
+	if c.PacketDown != nil {
+		if c.PacketDown.KeepAlivePeriod.From < 0 || c.PacketDown.KeepAliveLength.From < 0 {
+			return nil, errors.New("invalid negative value in packetDown")
+		}
+		config.PacketDown = &splithttp.PacketDownConfig{
+			KeepAlive:       c.PacketDown.KeepAlive,
+			KeepAlivePeriod: newRangeConfig(c.PacketDown.KeepAlivePeriod),
+			KeepAliveLength: newRangeConfig(c.PacketDown.KeepAliveLength),
+		}
+	}
+
 	if c.DownloadSettings != nil {
 		if c.Mode == "stream-one" {
 			return nil, errors.New(`Can not use "downloadSettings" in "stream-one" mode.`)

infra/conf/transport_method_test.go

@@ -0,0 +1,64 @@
+package conf_test
+
+import (
+	"encoding/json"
+	"testing"
+
+	"github.com/google/go-cmp/cmp"
+	. "github.com/xtls/xray-core/infra/conf"
+	"github.com/xtls/xray-core/transport/internet/splithttp"
+)
+
+func Test_SplitHTTP_PacketDown(t *testing.T) {
+	rawJSON := `{
+	  "path": "/sh",
+	  "packetDown": { "keepAlive": true, "keepAlivePeriod": "5-10", "keepAliveLength": 32 }
+	}`
+	cfg := new(SplitHTTPConfig)
+	if err := json.Unmarshal([]byte(rawJSON), cfg); err != nil {
+		t.Fatal(err)
+	}
+	built, err := cfg.Build()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	pd := built.(*splithttp.Config).GetPacketDown()
+	if pd == nil {
+		t.Fatal("packetDown was not built")
+	}
+	if !pd.GetKeepAlive() {
+		t.Error("packetDown.keepAlive should be true")
+	}
+	if r := cmp.Diff(int32(5), pd.GetKeepAlivePeriod().GetFrom()); r != "" {
+		t.Error(r)
+	}
+	if r := cmp.Diff(int32(32), pd.GetKeepAliveLength().GetTo()); r != "" {
+		t.Error(r)
+	}
+}
+
+func Test_SplitHTTP_PacketDownNegative(t *testing.T) {
+	rawJSON := `{ "path": "/sh", "packetDown": { "keepAlive": true, "keepAlivePeriod": "-1-10" } }`
+	cfg := new(SplitHTTPConfig)
+	parseErr := json.Unmarshal([]byte(rawJSON), cfg)
+	_, buildErr := cfg.Build()
+	if parseErr == nil && buildErr == nil {
+		t.Error("expected negative period to fail at least one stage")
+	}
+}
+
+func Test_SplitHTTP_PacketDownAbsent(t *testing.T) {
+	rawJSON := `{ "path": "/sh" }`
+	cfg := new(SplitHTTPConfig)
+	if err := json.Unmarshal([]byte(rawJSON), cfg); err != nil {
+		t.Fatal(err)
+	}
+	built, err := cfg.Build()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if built.(*splithttp.Config).GetPacketDown() != nil {
+		t.Error("packetDown should stay unset when absent")
+	}
+}

transport/internet/splithttp/client.go

@@ -68,6 +68,15 @@ func (c *DefaultDialerClient) OpenStream(ctx context.Context, url string, sessio
 	}
 	c.transportConfig.FillStreamRequest(req, sessionId, "")
 
+	// stream-down and stream-one responses carry the downlink; ask the
+	// server to run packet-down in it. stream-up responses are discarded
+	// by this client, so they don't need packet-down.
+	downlinkMaybeFramed := false
+	if !uploadOnly && c.transportConfig.PacketDown != nil {
+		req.Header.Set(pdRequestHeader, "1")
+		downlinkMaybeFramed = true
+	}
+
 	wrc = &WaitReadCloser{wait: done.New()}
 	go func() {
 		resp, err := c.client.Do(req)
@@ -91,7 +100,13 @@ func (c *DefaultDialerClient) OpenStream(ctx context.Context, url string, sessio
 			wrc.Close()
 			return
 		}
-		wrc.(*WaitReadCloser).Set(resp.Body)
+		if downlinkMaybeFramed && resp.Header.Get(pdResponseHeader) == "1" {
+			// the body is packet-down frames; strip them before the
+			// payload reaches the dispatcher
+			wrc.(*WaitReadCloser).Set(newPacketDownReader(resp.Body))
+		} else {
+			wrc.(*WaitReadCloser).Set(resp.Body)
+		}
 	}()
 
 	<-gotConn.Wait()

transport/internet/splithttp/config.go

@@ -427,6 +427,40 @@ func (c *Config) ExtractMetaFromRequest(req *http.Request, path string) (session
 	return sessionId, seqStr
 }
 
+func (c *PacketDownConfig) GetNormalizedKeepAlivePeriod() *RangeConfig {
+	if c.KeepAlivePeriod == nil || c.KeepAlivePeriod.To == 0 {
+		return &RangeConfig{
+			From: 10,
+			To:   30,
+		}
+	}
+
+	return c.KeepAlivePeriod
+}
+
+func (c *PacketDownConfig) GetNormalizedKeepAliveLength() *RangeConfig {
+	if c.KeepAliveLength == nil || c.KeepAliveLength.To == 0 {
+		return &RangeConfig{
+			From: 24,
+			To:   64,
+		}
+	}
+
+	if c.KeepAliveLength.From > pdMaxFramePayload {
+		return &RangeConfig{
+			From: pdMaxFramePayload,
+			To:   pdMaxFramePayload,
+		}
+	} else if c.KeepAliveLength.To > pdMaxFramePayload {
+		return &RangeConfig{
+			From: c.KeepAliveLength.From,
+			To:   pdMaxFramePayload,
+		}
+	}
+
+	return c.KeepAliveLength
+}
+
 func (m *XmuxConfig) GetNormalizedMaxConcurrency() *RangeConfig {
 	if m.MaxConcurrency == nil {
 		return &RangeConfig{