ginuerzh
382d47ea12
- handler/http/websocket, forwarder/sniffer_ws, sniffing/sniffer_ws: nil ro2.HTTP in WebSocket copy goroutines to avoid data races on the shared HTTP recorder object - handler/tunnel/bind: remove ingress rule check that incorrectly overrode the endpoint host for non-matching ingress rules - handler/tunnel/entrypoint: add forwarding loop detection by checking Gost-Forwarded-Node header for the entrypoint's own node ID
133 lines
3.0 KiB
Go
133 lines
3.0 KiB
Go
package sniffing
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"io"
|
|
"math"
|
|
"time"
|
|
|
|
"github.com/go-gost/core/logger"
|
|
ws_util "github.com/go-gost/x/internal/util/ws"
|
|
xrecorder "github.com/go-gost/x/recorder"
|
|
"golang.org/x/time/rate"
|
|
)
|
|
|
|
// sniffingWebsocketFrame copies WebSocket frames between rw and cc while
|
|
// recording frame metadata. It runs two goroutines for bidirectional copy
|
|
// and waits for the first error.
|
|
func (h *Sniffer) sniffingWebsocketFrame(ctx context.Context, rw, cc io.ReadWriter, ro *xrecorder.HandlerRecorderObject, log logger.Logger) error {
|
|
errc := make(chan error, 1)
|
|
|
|
sampleRate := h.WebsocketSampleRate
|
|
if sampleRate == 0 {
|
|
sampleRate = DefaultSampleRate
|
|
}
|
|
if sampleRate < 0 {
|
|
sampleRate = math.MaxFloat64
|
|
}
|
|
|
|
go func() {
|
|
ro2 := &xrecorder.HandlerRecorderObject{}
|
|
*ro2 = *ro
|
|
ro2.HTTP = nil
|
|
ro := ro2
|
|
|
|
limiter := rate.NewLimiter(rate.Limit(sampleRate), int(sampleRate))
|
|
|
|
buf := &bytes.Buffer{}
|
|
for {
|
|
start := time.Now()
|
|
|
|
if err := h.copyWebsocketFrame(cc, rw, buf, "client", ro); err != nil {
|
|
errc <- err
|
|
return
|
|
}
|
|
|
|
if limiter.Allow() {
|
|
ro.Duration = time.Since(start)
|
|
ro.Time = time.Now()
|
|
if err := ro.Record(ctx, h.Recorder); err != nil {
|
|
log.Errorf("record: %v", err)
|
|
}
|
|
}
|
|
}
|
|
}()
|
|
|
|
go func() {
|
|
ro2 := &xrecorder.HandlerRecorderObject{}
|
|
*ro2 = *ro
|
|
ro2.HTTP = nil
|
|
ro := ro2
|
|
|
|
limiter := rate.NewLimiter(rate.Limit(sampleRate), int(sampleRate))
|
|
|
|
buf := &bytes.Buffer{}
|
|
for {
|
|
start := time.Now()
|
|
|
|
if err := h.copyWebsocketFrame(rw, cc, buf, "server", ro); err != nil {
|
|
errc <- err
|
|
return
|
|
}
|
|
|
|
if limiter.Allow() {
|
|
ro.Duration = time.Since(start)
|
|
ro.Time = time.Now()
|
|
if err := ro.Record(ctx, h.Recorder); err != nil {
|
|
log.Errorf("record: %v", err)
|
|
}
|
|
}
|
|
}
|
|
}()
|
|
|
|
<-errc
|
|
return nil
|
|
}
|
|
|
|
// copyWebsocketFrame reads one WebSocket frame from r, records its metadata,
|
|
// and writes it to w.
|
|
func (h *Sniffer) copyWebsocketFrame(w io.Writer, r io.Reader, buf *bytes.Buffer, from string, ro *xrecorder.HandlerRecorderObject) (err error) {
|
|
fr := ws_util.Frame{}
|
|
if _, err = fr.ReadFrom(r); err != nil {
|
|
return err
|
|
}
|
|
|
|
ws := &xrecorder.WebsocketRecorderObject{
|
|
From: from,
|
|
Fin: fr.Header.Fin,
|
|
Rsv1: fr.Header.Rsv1,
|
|
Rsv2: fr.Header.Rsv2,
|
|
Rsv3: fr.Header.Rsv3,
|
|
OpCode: int(fr.Header.OpCode),
|
|
Masked: fr.Header.Masked,
|
|
MaskKey: fr.Header.MaskKey,
|
|
Length: fr.Header.PayloadLength,
|
|
}
|
|
|
|
if bodySize := ClampBodySize(h.RecorderOptions); bodySize > 0 {
|
|
buf.Reset()
|
|
if _, err := io.Copy(buf, io.LimitReader(fr.Data, int64(bodySize))); err != nil {
|
|
return err
|
|
}
|
|
ws.Payload = buf.Bytes()
|
|
}
|
|
|
|
ro.Websocket = ws
|
|
length := uint64(fr.Header.Length()) + uint64(fr.Header.PayloadLength)
|
|
if from == "client" {
|
|
ro.InputBytes = length
|
|
ro.OutputBytes = 0
|
|
} else {
|
|
ro.InputBytes = 0
|
|
ro.OutputBytes = length
|
|
}
|
|
|
|
fr.Data = io.MultiReader(bytes.NewReader(buf.Bytes()), fr.Data)
|
|
if _, err := fr.WriteTo(w); err != nil {
|
|
return err
|
|
}
|
|
|
|
return nil
|
|
}
|