Collect stats and possible padding (previously Vision Reader Writer) to all traffic

proxy/proxy.go

@@ -126,8 +126,8 @@ type TrafficState struct {
 	WriterSwitchToDirectCopy bool
 }
 
-func NewTrafficState(userUUID []byte) *TrafficState {
+func NewTrafficState(userUUID []byte, flow string) *TrafficState {
-	return &TrafficState{
+	var state = TrafficState{
 		UserUUID:                 userUUID,
 		StartTime:                time.Time{},
 		ByteSent:                 0,
@@ -140,15 +140,18 @@ func NewTrafficState(userUUID []byte) *TrafficState {
 		IsTLS:                    false,
 		Cipher:                   0,
 		RemainingServerHello:     -1,
-		WithinPaddingBuffers:     true,
 		ReaderSwitchToDirectCopy: false,
 		RemainingCommand:         -1,
 		RemainingContent:         -1,
 		RemainingPadding:         -1,
 		CurrentCommand:           0,
-		IsPadding:                true,
 		WriterSwitchToDirectCopy: false,
+	};
+	if len(flow) > 0 {
+		state.IsPadding = true;
+		state.WithinPaddingBuffers = true;
 	}
+	return &state
 }
 
 // VisionReader is used to read xtls vision protocol

proxy/vless/encoding/addons.go

@@ -55,23 +55,21 @@ func DecodeHeaderAddons(buffer *buf.Buffer, reader io.Reader) (*Addons, error) {
 }
 
 // EncodeBodyAddons returns a Writer that auto-encrypt content written by caller.
-func EncodeBodyAddons(writer io.Writer, request *protocol.RequestHeader, requestAddons *Addons, state *proxy.TrafficState, context context.Context) buf.Writer {
+func EncodeBodyAddons(writer buf.Writer, request *protocol.RequestHeader, requestAddons *Addons, state *proxy.TrafficState, context context.Context) buf.Writer {
+	w := proxy.NewVisionWriter(writer, state, context)
 	if request.Command == protocol.RequestCommandUDP {
-		return NewMultiLengthPacketWriter(writer.(buf.Writer))
+		return NewMultiLengthPacketWriter(w)
-	}
-	w := buf.NewWriter(writer)
-	if requestAddons.Flow == vless.XRV {
-		w = proxy.NewVisionWriter(w, state, context)
 	}
 	return w
 }
 
 // DecodeBodyAddons returns a Reader from which caller can fetch decrypted body.
-func DecodeBodyAddons(reader io.Reader, request *protocol.RequestHeader, addons *Addons) buf.Reader {
+func DecodeBodyAddons(reader io.Reader, request *protocol.RequestHeader, addons *Addons, state *proxy.TrafficState, context context.Context) buf.Reader {
+	r := proxy.NewVisionReader(buf.NewReader(reader), state, context)
 	if request.Command == protocol.RequestCommandUDP {
-		return NewLengthPacketReader(reader)
+		return NewLengthPacketReader(&buf.BufferedReader{Reader: r})
 	}
-	return buf.NewReader(reader)
+	return r
 }
 
 func NewMultiLengthPacketWriter(writer buf.Writer) *MultiLengthPacketWriter {
@@ -204,39 +202,39 @@ func PopulateSeed(seed string, addons *Addons) {
 
 func CheckSeed(requestAddons *Addons, responseAddons *Addons) error {
 	if !bytes.Equal(requestAddons.Seed, responseAddons.Seed) {
-		return newError("Seed bytes not match", requestAddons.Seed, responseAddons.Seed)
+		return errors.New("Seed bytes not match", requestAddons.Seed, responseAddons.Seed)
 	}
 	if requestAddons.Mode != responseAddons.Mode {
-		return newError("Mode not match", requestAddons.Mode, responseAddons.Mode)
+		return errors.New("Mode not match", requestAddons.Mode, responseAddons.Mode)
 	}
 	if requestAddons.Duration != responseAddons.Duration {
-		return newError("Duration not match", requestAddons.Duration, responseAddons.Duration)
+		return errors.New("Duration not match", requestAddons.Duration, responseAddons.Duration)
 	}
 	if requestAddons.Padding != nil && responseAddons.Padding != nil {
 		if requestAddons.Padding.RegularMin != responseAddons.Padding.RegularMin || 
 		requestAddons.Padding.RegularMax != responseAddons.Padding.RegularMax || 
 		requestAddons.Padding.LongMin != responseAddons.Padding.LongMin || 
 		requestAddons.Padding.LongMax != responseAddons.Padding.LongMax {
-			return newError("Padding not match")
+			return errors.New("Padding not match")
 		}
 	} else if requestAddons.Padding != nil || responseAddons.Padding != nil {
-		return newError("Padding of one is nil but the other is not nil")
+		return errors.New("Padding of one is nil but the other is not nil")
 	}
 	if requestAddons.Delay != nil && responseAddons.Delay != nil {
 		if requestAddons.Delay.IsRandom != responseAddons.Delay.IsRandom || 
 		requestAddons.Delay.MinMillis != responseAddons.Delay.MinMillis || 
 		requestAddons.Delay.MaxMillis != responseAddons.Delay.MaxMillis {
-			return newError("Delay not match")
+			return errors.New("Delay not match")
 		}
 	} else if requestAddons.Delay != nil || responseAddons.Delay != nil {
-		return newError("Delay of one is nil but the other is not nil")
+		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 {
-			return newError("Scheduler not match")
+			return errors.New("Scheduler not match")
 		}
 	} else if requestAddons.Scheduler != nil || responseAddons.Scheduler != nil {
-		return newError("Scheduler of one is nil but the other is not nil")
+		return errors.New("Scheduler of one is nil but the other is not nil")
 	}
 	return nil
 }

proxy/vless/inbound/inbound.go

@@ -460,7 +460,7 @@ func (h *Handler) Process(ctx context.Context, network net.Network, connection s
 	}
 	encoding.PopulateSeed(account.Seed, responseAddons)
 	if check := encoding.CheckSeed(requestAddons, responseAddons); check != nil {
-		return newError("Seed configuration mis-match").Base(check).AtWarning()
+		return errors.New("Seed configuration mis-match").Base(check).AtWarning()
 	}
 
 	var input *bytes.Reader
@@ -531,18 +531,17 @@ 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())
+	trafficState := proxy.NewTrafficState(account.ID.Bytes(), account.Flow)
 	postRequest := func() error {
 		defer timer.SetTimeout(sessionPolicy.Timeouts.DownlinkOnly)
 
 		// default: clientReader := reader
-		clientReader := encoding.DecodeBodyAddons(reader, request, requestAddons)
+		clientReader := encoding.DecodeBodyAddons(reader, request, requestAddons, trafficState, ctx)
 
 		var err error
 
 		if requestAddons.Flow == vless.XRV {
 			ctx1 := session.ContextWithInbound(ctx, nil) // TODO enable splice
-			clientReader = proxy.NewVisionReader(clientReader, trafficState, ctx1)
 			err = encoding.XtlsRead(clientReader, serverWriter, timer, connection, input, rawInput, trafficState, nil, ctx1)
 		} else {
 			// from clientReader.ReadMultiBuffer to serverWriter.WriteMultiBuffer

proxy/vless/outbound/outbound.go

@@ -179,7 +179,7 @@ func (h *Handler) Process(ctx context.Context, link *transport.Link, dialer inte
 
 	clientReader := link.Reader // .(*pipe.Reader)
 	clientWriter := link.Writer // .(*pipe.Writer)
-	trafficState := proxy.NewTrafficState(account.ID.Bytes())
+	trafficState := proxy.NewTrafficState(account.ID.Bytes(), account.Flow)
 	if request.Command == protocol.RequestCommandUDP && (requestAddons.Flow == vless.XRV || (h.cone && request.Port != 53 && request.Port != 443)) {
 		request.Command = protocol.RequestCommandMux
 		request.Address = net.DomainAddress("v1.mux.cool")
@@ -260,16 +260,9 @@ func (h *Handler) Process(ctx context.Context, link *transport.Link, dialer inte
 		}
 
 		// default: serverReader := buf.NewReader(conn)
-		serverReader := encoding.DecodeBodyAddons(conn, request, responseAddons)
+		serverReader := encoding.DecodeBodyAddons(conn, request, responseAddons, trafficState, ctx)
-		if requestAddons.Flow == vless.XRV {
-			serverReader = proxy.NewVisionReader(serverReader, trafficState, ctx)
-		}
 		if request.Command == protocol.RequestCommandMux && request.Port == 666 {
-			if requestAddons.Flow == vless.XRV {
 			serverReader = xudp.NewPacketReader(&buf.BufferedReader{Reader: serverReader})
-			} else {
-				serverReader = xudp.NewPacketReader(conn)
-			}
 		}
 
 		if requestAddons.Flow == vless.XRV {