Merge pull request v2ray#224 from rprx/master

VLESS PREVIEW 2

infra/conf/vless.go

@@ -48,9 +48,12 @@ func (c *VLessInboundConfig) Build() (proto.Message, error) {
 			return nil, newError(`VLESS clients: invalid user`).Base(err)
 		}
 
-		if account.Flow != "" {
-			return nil, newError(`VLESS clients: "flow" is not available in this version`)
+		switch account.Flow {
+		case "", "xtls-rprx-origin":
+		default:
+			return nil, newError(`VLESS clients: "flow" only accepts "", "xtls-rprx-origin" in this version`)
 		}
+
 		if account.Encryption != "" {
 			return nil, newError(`VLESS clients: "encryption" should not in inbound settings`)
 		}
@@ -161,9 +164,12 @@ func (c *VLessOutboundConfig) Build() (proto.Message, error) {
 				return nil, newError(`VLESS users: invalid user`).Base(err)
 			}
 
-			if account.Flow != "" {
-				return nil, newError(`VLESS users: "flow" is not available in this version`)
+			switch account.Flow {
+			case "", "xtls-rprx-origin", "xtls-rprx-origin-udp443":
+			default:
+				return nil, newError(`VLESS users: "flow" only accepts "", "xtls-rprx-origin", "xtls-rprx-origin-udp443" in this version`)
 			}
+
 			if account.Encryption != "none" {
 				return nil, newError(`VLESS users: please add/set "encryption":"none" for every user`)
 			}

infra/conf/vless_test.go

@@ -26,6 +26,7 @@ func TestVLessOutbound(t *testing.T) {
 					"users": [
 						{
 							"id": "27848739-7e62-4138-9fd3-098a63964b6b",
+							"flow": "xtls-rprx-origin-udp443",
 							"encryption": "none",
 							"level": 0
 						}
@@ -46,6 +47,7 @@ func TestVLessOutbound(t *testing.T) {
 							{
 								Account: serial.ToTypedMessage(&vless.Account{
 									Id:         "27848739-7e62-4138-9fd3-098a63964b6b",
+									Flow:       "xtls-rprx-origin-udp443",
 									Encryption: "none",
 								}),
 								Level: 0,
@@ -69,6 +71,7 @@ func TestVLessInbound(t *testing.T) {
 				"clients": [
 					{
 						"id": "27848739-7e62-4138-9fd3-098a63964b6b",
+						"flow": "xtls-rprx-origin",
 						"level": 0,
 						"email": "love@v2fly.org"
 					}
@@ -94,7 +97,8 @@ func TestVLessInbound(t *testing.T) {
 				Clients: []*protocol.User{
 					{
 						Account: serial.ToTypedMessage(&vless.Account{
-							Id: "27848739-7e62-4138-9fd3-098a63964b6b",
+							Id:   "27848739-7e62-4138-9fd3-098a63964b6b",
+							Flow: "xtls-rprx-origin",
 						}),
 						Level: 0,
 						Email: "love@v2fly.org",

proxy/vless/encoding/addons.go

@@ -9,11 +9,24 @@ import (
 
 	"v2ray.com/core/common/buf"
 	"v2ray.com/core/common/protocol"
+	"v2ray.com/core/proxy/vless"
 )
 
 func EncodeHeaderAddons(buffer *buf.Buffer, addons *Addons) error {
 
 	switch addons.Flow {
+	case vless.XRO:
+
+		if bytes, err := proto.Marshal(addons); err != nil {
+			return newError("failed to marshal addons protobuf value").Base(err)
+		}
+		if err := buffer.WriteByte(byte(len(bytes))); err != nil {
+			return newError("failed to write addons protobuf length").Base(err)
+		}
+		if _, err := buffer.Write(bytes); err != nil {
+			return newError("failed to write addons protobuf value").Base(err)
+		}
+
 	default:
 
 		if err := buffer.WriteByte(0); err != nil {
@@ -62,22 +75,136 @@ func DecodeHeaderAddons(buffer *buf.Buffer, reader io.Reader) (*Addons, error) {
 func EncodeBodyAddons(writer io.Writer, request *protocol.RequestHeader, addons *Addons) buf.Writer {
 
 	switch addons.Flow {
-	default:
+	case vless.XRO:
 
-		return buf.NewWriter(writer)
+		if request.Command == protocol.RequestCommandUDP {
+			return NewMultiLengthPacketWriter(writer.(buf.Writer))
+		}
 
 	}
 
+	return buf.NewWriter(writer)
+
 }
 
 // DecodeBodyAddons returns a Reader from which caller can fetch decrypted body.
 func DecodeBodyAddons(reader io.Reader, request *protocol.RequestHeader, addons *Addons) buf.Reader {
 
 	switch addons.Flow {
-	default:
+	case vless.XRO:
+
+		if request.Command == protocol.RequestCommandUDP {
+			return NewLengthPacketReader(reader)
+		}
 
-		return buf.NewReader(reader)
+	}
+
+	return buf.NewReader(reader)
+
+}
 
+func NewMultiLengthPacketWriter(writer buf.Writer) *MultiLengthPacketWriter {
+	return &MultiLengthPacketWriter{
+		Writer: writer,
 	}
+}
+
+type MultiLengthPacketWriter struct {
+	buf.Writer
+}
 
+func (w *MultiLengthPacketWriter) WriteMultiBuffer(mb buf.MultiBuffer) error {
+	defer buf.ReleaseMulti(mb)
+	mb2Write := make(buf.MultiBuffer, 0, len(mb)+1)
+	for _, b := range mb {
+		length := b.Len()
+		if length == 0 || length+2 > buf.Size {
+			continue
+		}
+		eb := buf.New()
+		if err := eb.WriteByte(byte(length >> 8)); err != nil {
+			eb.Release()
+			continue
+		}
+		if err := eb.WriteByte(byte(length)); err != nil {
+			eb.Release()
+			continue
+		}
+		if _, err := eb.Write(b.Bytes()); err != nil {
+			eb.Release()
+			continue
+		}
+		mb2Write = append(mb2Write, eb)
+	}
+	if mb2Write.IsEmpty() {
+		return nil
+	}
+	return w.Writer.WriteMultiBuffer(mb2Write)
+}
+
+func NewLengthPacketWriter(writer io.Writer) *LengthPacketWriter {
+	return &LengthPacketWriter{
+		Writer: writer,
+		cache:  make([]byte, 0, 65536),
+	}
+}
+
+type LengthPacketWriter struct {
+	io.Writer
+	cache []byte
+}
+
+func (w *LengthPacketWriter) WriteMultiBuffer(mb buf.MultiBuffer) error {
+	length := mb.Len() // none of mb is nil
+	//fmt.Println("Write", length)
+	if length == 0 {
+		return nil
+	}
+	defer func() {
+		w.cache = w.cache[:0]
+	}()
+	w.cache = append(w.cache, byte(length>>8), byte(length))
+	for i, b := range mb {
+		w.cache = append(w.cache, b.Bytes()...)
+		b.Release()
+		mb[i] = nil
+	}
+	if _, err := w.Write(w.cache); err != nil {
+		return newError("failed to write a packet").Base(err)
+	}
+	return nil
+}
+
+func NewLengthPacketReader(reader io.Reader) *LengthPacketReader {
+	return &LengthPacketReader{
+		Reader: reader,
+		cache:  make([]byte, 2),
+	}
+}
+
+type LengthPacketReader struct {
+	io.Reader
+	cache []byte
+}
+
+func (r *LengthPacketReader) ReadMultiBuffer() (buf.MultiBuffer, error) {
+	if _, err := io.ReadFull(r.Reader, r.cache); err != nil { // maybe EOF
+		return nil, newError("failed to read packet length").Base(err)
+	}
+	length := int(r.cache[0])<<8 | int(r.cache[1])
+	//fmt.Println("Read", length)
+	mb := make(buf.MultiBuffer, 0, length/buf.Size+1)
+	for length > 0 {
+		size := length
+		if length > buf.Size {
+			size = buf.Size
+		}
+		length -= size
+		b := buf.New()
+		if _, err := b.ReadFullFrom(r.Reader, int32(size)); err != nil {
+			return nil, newError("failed to read packet payload").Base(err)
+		}
+		mb = append(mb, b)
+	}
+	return mb, nil
 }

proxy/vless/encoding/encoding.go

@@ -153,23 +153,23 @@ func EncodeResponseHeader(writer io.Writer, request *protocol.RequestHeader, res
 }
 
 // DecodeResponseHeader decodes and returns (if successful) a ResponseHeader from an input stream.
-func DecodeResponseHeader(reader io.Reader, request *protocol.RequestHeader, responseAddons *Addons) error {
+func DecodeResponseHeader(reader io.Reader, request *protocol.RequestHeader) (*Addons, error) {
 
 	buffer := buf.StackNew()
 	defer buffer.Release()
 
 	if _, err := buffer.ReadFullFrom(reader, 1); err != nil {
-		return newError("failed to read response version").Base(err)
+		return nil, newError("failed to read response version").Base(err)
 	}
 
 	if buffer.Byte(0) != request.Version {
-		return newError("unexpected response version. Expecting ", int(request.Version), " but actually ", int(buffer.Byte(0)))
+		return nil, newError("unexpected response version. Expecting ", int(request.Version), " but actually ", int(buffer.Byte(0)))
 	}
 
 	responseAddons, err := DecodeHeaderAddons(&buffer, reader)
 	if err != nil {
-		return newError("failed to decode response header addons").Base(err)
+		return nil, newError("failed to decode response header addons").Base(err)
 	}
 
-	return nil
+	return responseAddons, nil
 }