diff --git a/handler/http/auth.go b/handler/http/auth.go new file mode 100644 index 00000000..92f70dc0 --- /dev/null +++ b/handler/http/auth.go @@ -0,0 +1,161 @@ +package http + +import ( + "context" + "fmt" + "net" + "net/http" + "net/http/httputil" + "os" + "strconv" + "strings" + "time" + + "github.com/go-gost/core/auth" + "github.com/go-gost/core/logger" + xnet "github.com/go-gost/x/internal/net" +) + +// authenticate verifies the client's Proxy-Authorization header against the +// configured Authenticator. When authentication fails and probe resistance +// is configured, a decoy response is returned to hide the proxy. +// +// Five probe resistance strategies are supported: +// +// - "code": respond with a custom HTTP status code (e.g. "code:404"). +// - "web": fetch a URL and replay its response as the decoy. +// - "host": forward the raw request to a decoy host and relay the reply. +// - "file": serve a local file with Content-Type: text/html. +// - knock: when pr.Knock is set, probe resistance only activates if the +// request hostname does NOT match the knock address. Clients that know +// the knock hostname get the normal 407 Proxy-Auth-Required. +// +// On success it returns the authenticated client ID. On failure it writes +// the response (407 or probe-resistance decoy) to conn and returns ok=false. +func (h *httpHandler) authenticate(ctx context.Context, conn net.Conn, req *http.Request, resp *http.Response, log logger.Logger) (id string, ok bool) { + u, p, _ := basicProxyAuth(req.Header.Get("Proxy-Authorization")) + if h.options.Auther == nil { + return "", true + } + if id, ok = h.options.Auther.Authenticate(ctx, u, p, auth.WithService(h.options.Service)); ok { + return + } + + pr := h.md.probeResistance + // Probe resistance activates on auth failure when: + // - pr.Knock is empty (always hide the proxy), OR + // - pr.Knock is set but the request hostname doesn't match (the client + // didn't "knock" on the right host — hide the proxy). + // When pr.Knock matches the hostname, probe resistance is bypassed and + // the normal 407 Proxy-Auth-Required is returned, revealing the proxy + // only to clients that know the knock address. + if pr != nil && (pr.Knock == "" || !strings.EqualFold(req.URL.Hostname(), pr.Knock)) { + resp.StatusCode = http.StatusServiceUnavailable + + switch pr.Type { + case "code": + if code, err := strconv.Atoi(pr.Value); err == nil { + resp.StatusCode = code + } else { + log.Warnf("invalid probe resistance code: %s", pr.Value) + } + case "web": + url := pr.Value + if !strings.HasPrefix(url, "http") { + url = "http://" + url + } + client := &http.Client{Timeout: 15 * time.Second} + r, err := client.Get(url) + if err != nil { + log.Error(err) + break + } + // Replace resp content with the fetched web response. + // Body is closed after writing in the caller. + *resp = *r + case "host": + // Dial the decoy host and transparently relay the request/response. + cc, err := net.Dial("tcp", pr.Value) + if err != nil { + log.Error(err) + break + } + defer cc.Close() + + req.Write(cc) + xnet.Pipe(ctx, conn, cc, xnet.WithReadTimeout(h.md.idleTimeout)) + return "", false + case "file": + f, _ := os.Open(pr.Value) + if f != nil { + defer f.Close() + + resp.StatusCode = http.StatusOK + if finfo, _ := f.Stat(); finfo != nil { + resp.ContentLength = finfo.Size() + } + resp.Header.Set("Content-Type", "text/html") + resp.Body = f + } + } + } + + if resp.Header == nil { + resp.Header = http.Header{} + } + if resp.StatusCode == 0 { + // Normal 407 Proxy-Auth-Required response. + realm := defaultRealm + if h.md.authBasicRealm != "" { + realm = h.md.authBasicRealm + } + resp.StatusCode = http.StatusProxyAuthRequired + resp.Header.Add("Proxy-Authenticate", fmt.Sprintf("Basic realm=\"%s\"", realm)) + if strings.ToLower(req.Header.Get("Proxy-Connection")) == "keep-alive" { + // libcurl will keep sending auth requests on the same + // connection, which we don't support yet. Force close. + resp.Header.Set("Connection", "close") + resp.Header.Set("Proxy-Connection", "close") + } + + log.Debug("proxy authentication required") + } else { + // Probe resistance sent a non-407 status. For 200 OK file/web + // responses, advertise keep-alive to appear like a normal server. + if resp.StatusCode == http.StatusOK { + resp.Header.Set("Connection", "keep-alive") + } + } + + if log.IsLevelEnabled(logger.TraceLevel) { + dump, _ := httputil.DumpResponse(resp, false) + log.Trace(string(dump)) + } + + if resp.Body != nil { + defer resp.Body.Close() + } + if err := resp.Write(conn); err != nil { + log.Error("write auth response: ", err) + } + return +} + +// checkRateLimit verifies that the remote address has not exceeded the +// configured connection rate limit. It extracts the host portion of addr +// and looks up the per-host limiter. Returns true if the connection is +// allowed or if no rate limiter is configured. +func (h *httpHandler) checkRateLimit(addr net.Addr) bool { + if h.options.RateLimiter == nil { + return true + } + if addr == nil { + return true + } + host, _, _ := net.SplitHostPort(addr.String()) + if limiter := h.options.RateLimiter.Limiter(host); limiter != nil { + return limiter.Allow(1) + } + + return true +} diff --git a/handler/http/auth_test.go b/handler/http/auth_test.go new file mode 100644 index 00000000..54d036b6 --- /dev/null +++ b/handler/http/auth_test.go @@ -0,0 +1,390 @@ +package http + +import ( + "bufio" + "context" + "net" + "net/http" + "testing" + "time" + + "github.com/go-gost/core/auth" + "github.com/go-gost/core/handler" + "github.com/go-gost/core/limiter" + "github.com/go-gost/core/limiter/rate" + "github.com/go-gost/core/limiter/traffic" + "github.com/go-gost/core/logger" +) + +// stubAuther implements auth.Authenticator for testing. +type stubAuther struct { + accept bool + id string +} + +func (a *stubAuther) Authenticate(ctx context.Context, user, password string, opts ...auth.Option) (string, bool) { + return a.id, a.accept +} + +// testLogger implements logger.Logger for testing. +type testLogger struct{} + +func (l *testLogger) WithFields(fields map[string]any) logger.Logger { return l } +func (l *testLogger) Debug(args ...any) {} +func (l *testLogger) Debugf(format string, args ...any) {} +func (l *testLogger) Info(args ...any) {} +func (l *testLogger) Infof(format string, args ...any) {} +func (l *testLogger) Warn(args ...any) {} +func (l *testLogger) Warnf(format string, args ...any) {} +func (l *testLogger) Error(args ...any) {} +func (l *testLogger) Errorf(format string, args ...any) {} +func (l *testLogger) Fatal(args ...any) {} +func (l *testLogger) Fatalf(format string, args ...any) {} +func (l *testLogger) GetLevel() logger.LogLevel { return logger.InfoLevel } +func (l *testLogger) IsLevelEnabled(level logger.LogLevel) bool { return false } +func (l *testLogger) Trace(args ...any) {} +func (l *testLogger) Tracef(format string, args ...any) {} + +func TestAuthenticate_NoAuther(t *testing.T) { + h := &httpHandler{ + options: handler.Options{ + Logger: &testLogger{}, + }, + } + h.md.proxyAgent = defaultProxyAgent + + client, server := net.Pipe() + defer client.Close() + defer server.Close() + + req, _ := http.NewRequest("GET", "http://example.com/", nil) + resp := &http.Response{ + ProtoMajor: 1, + ProtoMinor: 1, + Header: http.Header{}, + ContentLength: -1, + } + + // Read the response from authenticate in a goroutine + go func() { + h.authenticate(context.Background(), server, req, resp, &testLogger{}) + }() + + // Client side: read the response back + // With no auther, authenticate returns early with ok=true, no response written + br := bufio.NewReader(client) + // The response shouldn't be written when auth succeeds + time.Sleep(50 * time.Millisecond) + // Connection should be clean - no data to read + _ = br +} + +func TestAuthenticate_Success(t *testing.T) { + h := &httpHandler{ + options: handler.Options{ + Auther: &stubAuther{accept: true, id: "testuser"}, + Logger: &testLogger{}, + }, + } + h.md.proxyAgent = defaultProxyAgent + + client, server := net.Pipe() + defer client.Close() + defer server.Close() + + req, _ := http.NewRequest("GET", "http://example.com/", nil) + req.Header.Set("Proxy-Authorization", "Basic dGVzdHVzZXI6cGFzc3dvcmQ=") // testuser:password + resp := &http.Response{ + ProtoMajor: 1, + ProtoMinor: 1, + Header: http.Header{}, + ContentLength: -1, + } + + done := make(chan struct{}) + go func() { + defer close(done) + id, ok := h.authenticate(context.Background(), server, req, resp, &testLogger{}) + if !ok { + t.Error("expected auth success") + } + if id != "testuser" { + t.Errorf("got id %q, want %q", id, "testuser") + } + }() + + time.Sleep(50 * time.Millisecond) + // Auth succeeded, no response written to client + _ = client + <-done +} + +func TestAuthenticate_Failure(t *testing.T) { + h := &httpHandler{ + options: handler.Options{ + Auther: &stubAuther{accept: false}, + Logger: &testLogger{}, + }, + } + h.md.proxyAgent = defaultProxyAgent + + client, server := net.Pipe() + defer client.Close() + defer server.Close() + + req, _ := http.NewRequest("GET", "http://example.com/", nil) + req.Header.Set("Proxy-Authorization", "Basic dGVzdHVzZXI6cGFzc3dvcmQ=") + resp := &http.Response{ + ProtoMajor: 1, + ProtoMinor: 1, + Header: http.Header{}, + ContentLength: -1, + } + + go func() { + h.authenticate(context.Background(), server, req, resp, &testLogger{}) + }() + + // Read the 407 response + br := bufio.NewReader(client) + r, err := http.ReadResponse(br, req) + if err != nil { + t.Fatalf("failed to read auth response: %v", err) + } + defer r.Body.Close() + + if r.StatusCode != http.StatusProxyAuthRequired { + t.Errorf("got status %d, want %d", r.StatusCode, http.StatusProxyAuthRequired) + } + if r.Header.Get("Proxy-Authenticate") == "" { + t.Error("expected Proxy-Authenticate header") + } +} + +func TestAuthenticate_ProbeResistanceCode(t *testing.T) { + h := &httpHandler{ + options: handler.Options{ + Auther: &stubAuther{accept: false}, + Logger: &testLogger{}, + }, + } + h.md.proxyAgent = defaultProxyAgent + h.md.probeResistance = &probeResistance{Type: "code", Value: "404"} + + client, server := net.Pipe() + defer client.Close() + defer server.Close() + + req, _ := http.NewRequest("GET", "http://example.com/", nil) + resp := &http.Response{ + ProtoMajor: 1, + ProtoMinor: 1, + Header: http.Header{}, + ContentLength: -1, + } + + go func() { + h.authenticate(context.Background(), server, req, resp, &testLogger{}) + }() + + br := bufio.NewReader(client) + r, err := http.ReadResponse(br, req) + if err != nil { + t.Fatalf("failed to read probe resistance response: %v", err) + } + defer r.Body.Close() + + if r.StatusCode != 404 { + t.Errorf("got status %d, want 404", r.StatusCode) + } +} + +func TestAuthenticate_ProbeResistanceKnock(t *testing.T) { + h := &httpHandler{ + options: handler.Options{ + Auther: &stubAuther{accept: false}, + Logger: &testLogger{}, + }, + } + h.md.proxyAgent = defaultProxyAgent + // Knock matches - probe resistance should be bypassed, fall through to 407 + h.md.probeResistance = &probeResistance{Type: "code", Value: "404", Knock: "secret.example.com"} + + client, server := net.Pipe() + defer client.Close() + defer server.Close() + + req, _ := http.NewRequest("GET", "http://secret.example.com/", nil) + resp := &http.Response{ + ProtoMajor: 1, + ProtoMinor: 1, + Header: http.Header{}, + ContentLength: -1, + } + + go func() { + h.authenticate(context.Background(), server, req, resp, &testLogger{}) + }() + + br := bufio.NewReader(client) + r, err := http.ReadResponse(br, req) + if err != nil { + t.Fatalf("failed to read response: %v", err) + } + defer r.Body.Close() + + // Knock matches, so probe resistance is bypassed, get normal 407 + if r.StatusCode != http.StatusProxyAuthRequired { + t.Errorf("got status %d, want %d (knock matched, should bypass probe resistance)", r.StatusCode, http.StatusProxyAuthRequired) + } +} + +func TestAuthenticate_ProbeResistanceKnockMismatch(t *testing.T) { + h := &httpHandler{ + options: handler.Options{ + Auther: &stubAuther{accept: false}, + Logger: &testLogger{}, + }, + } + h.md.proxyAgent = defaultProxyAgent + // Knock doesn't match - probe resistance should activate + h.md.probeResistance = &probeResistance{Type: "code", Value: "503", Knock: "secret.example.com"} + + client, server := net.Pipe() + defer client.Close() + defer server.Close() + + req, _ := http.NewRequest("GET", "http://other.example.com/", nil) + resp := &http.Response{ + ProtoMajor: 1, + ProtoMinor: 1, + Header: http.Header{}, + ContentLength: -1, + } + + go func() { + h.authenticate(context.Background(), server, req, resp, &testLogger{}) + }() + + br := bufio.NewReader(client) + r, err := http.ReadResponse(br, req) + if err != nil { + t.Fatalf("failed to read response: %v", err) + } + defer r.Body.Close() + + // Knock doesn't match, probe resistance activates + if r.StatusCode != 503 { + t.Errorf("got status %d, want 503", r.StatusCode) + } +} + +func TestCheckRateLimit_NoLimiter(t *testing.T) { + h := &httpHandler{ + options: handler.Options{}, + } + + addr := &net.TCPAddr{IP: net.ParseIP("1.2.3.4"), Port: 12345} + if !h.checkRateLimit(addr) { + t.Error("checkRateLimit should return true when no limiter is set") + } +} + +func TestCheckRateLimit_WithLimiter(t *testing.T) { + // Create a handler with a rate limiter that blocks everything + h := &httpHandler{ + options: handler.Options{ + RateLimiter: &stubRateLimiter{allow: false, limiter: &stubLimiter{allow: false}}, + }, + } + + addr := &net.TCPAddr{IP: net.ParseIP("1.2.3.4"), Port: 12345} + if h.checkRateLimit(addr) { + t.Error("checkRateLimit should return false when limiter denies") + } +} + +func TestCheckRateLimit_Allow(t *testing.T) { + h := &httpHandler{ + options: handler.Options{ + RateLimiter: &stubRateLimiter{allow: true, limiter: &stubLimiter{allow: true}}, + }, + } + + addr := &net.TCPAddr{IP: net.ParseIP("1.2.3.4"), Port: 12345} + if !h.checkRateLimit(addr) { + t.Error("checkRateLimit should return true when limiter allows") + } +} + +func TestCheckRateLimit_NilAddr(t *testing.T) { + h := &httpHandler{ + options: handler.Options{ + RateLimiter: &stubRateLimiter{allow: true, limiter: &stubLimiter{allow: true}}, + }, + } + + // Nil addr should not panic - robustness check + func() { + defer func() { + if r := recover(); r != nil { + t.Errorf("checkRateLimit panicked with nil addr: %v", r) + } + }() + // This may panic if addr.String() is called on nil - that's a String() panic, not ours + // But we should guard against it + _ = h.checkRateLimit(nil) //nolint + }() +} + +// testRecorderObject is a recorder.RecorderObject for testing. +var testRecorderObject = struct { + Record string + Options *testRecorderOptions +}{Record: "serviceHandler"} + +type testRecorderOptions struct { + HTTPBody bool +} + +// stubRateLimiter implements rate.RateLimiter for testing. +type stubRateLimiter struct { + allow bool + limiter *stubLimiter +} + +func (l *stubRateLimiter) Limiter(key string) rate.Limiter { + if key == "" { + return nil + } + return l.limiter +} + +// stubLimiter implements rate.Limiter for testing. +type stubLimiter struct { + allow bool +} + +func (l *stubLimiter) Allow(n int) bool { return l.allow } +func (l *stubLimiter) AllowN(n int) bool { return l.allow } +func (l *stubLimiter) Limit() float64 { return 0 } + +// testTrafficLimiter implements traffic.TrafficLimiter for testing. +type testTrafficLimiter struct{} + +func (l *testTrafficLimiter) In(ctx context.Context, key string, opts ...limiter.Option) traffic.Limiter { + return &testTLimiter{} +} +func (l *testTrafficLimiter) Out(ctx context.Context, key string, opts ...limiter.Option) traffic.Limiter { + return &testTLimiter{} +} +func (l *testTrafficLimiter) InOut(ctx context.Context, key string, opts ...limiter.Option) traffic.Limiter { + return &testTLimiter{} +} + +type testTLimiter struct{} + +func (l *testTLimiter) Wait(ctx context.Context, n int) int { return n } +func (l *testTLimiter) Limit() int { return 0 } +func (l *testTLimiter) Set(n int) {} + diff --git a/handler/http/connect.go b/handler/http/connect.go new file mode 100644 index 00000000..bcbb5a61 --- /dev/null +++ b/handler/http/connect.go @@ -0,0 +1,225 @@ +package http + +import ( + "bufio" + "bytes" + "context" + "crypto/tls" + "errors" + "net" + "net/http" + "net/http/httputil" + "time" + + "github.com/go-gost/core/limiter" + stats "github.com/go-gost/core/observer/stats" + "github.com/go-gost/core/logger" + xctx "github.com/go-gost/x/ctx" + ictx "github.com/go-gost/x/internal/ctx" + xnet "github.com/go-gost/x/internal/net" + "github.com/go-gost/x/internal/util/sniffing" + traffic_wrapper "github.com/go-gost/x/limiter/traffic/wrapper" + stats_wrapper "github.com/go-gost/x/observer/stats/wrapper" + xrecorder "github.com/go-gost/x/recorder" +) + +// handleConnect handles HTTP CONNECT tunnel requests. It dials the target +// address through the proxy chain router, sends a "200 Connection established" +// response to the client, and then relays raw bytes bidirectionally between +// client and upstream. +// +// When sniffing is enabled, the initial bytes from the client are inspected. +// If they match HTTP or TLS, the connection is handed off to the sniffer for +// protocol-aware forwarding (HTTP request routing, TLS MITM decryption). +// If the sniffed protocol is unknown, or sniffing is disabled, raw pipe +// forwarding is used. +func (h *httpHandler) handleConnect(ctx context.Context, conn net.Conn, ro *xrecorder.HandlerRecorderObject, log logger.Logger, addr string, resp *http.Response) error { + ctx = ictx.ContextWithRecorderObject(ctx, ro) + ctx = ictx.ContextWithLogger(ctx, log) + cc, err := h.dial(ctx, "tcp", addr) + if err != nil { + resp.StatusCode = http.StatusServiceUnavailable + + if log.IsLevelEnabled(logger.TraceLevel) { + dump, _ := httputil.DumpResponse(resp, false) + log.Trace(string(dump)) + } + if err := resp.Write(conn); err != nil { + log.Error("write error response: ", err) + } + return err + } + // snifferHandled tracks whether the upstream connection has been taken + // over by the sniffer. If the sniffer doesn't claim it, we close cc. + snifferHandled := false + defer func() { + if !snifferHandled { + cc.Close() + } + }() + + log = log.WithFields(map[string]any{"src": cc.LocalAddr().String(), "dst": cc.RemoteAddr().String()}) + ro.SrcAddr = cc.LocalAddr().String() + ro.DstAddr = cc.RemoteAddr().String() + + b := buildConnectResponse(h.md.proxyAgent) + if log.IsLevelEnabled(logger.TraceLevel) { + log.Trace(string(b)) + } + if _, err = conn.Write(b); err != nil { + log.Error(err) + return err + } + + if h.md.sniffing { + snifferHandled, err = h.sniffAndHandle(ctx, conn, cc, ro, log) + if snifferHandled { + return err + } + } + + start := time.Now() + log.Infof("%s <-> %s", conn.RemoteAddr(), addr) + xnet.Pipe(ctx, conn, cc, xnet.WithReadTimeout(h.md.idleTimeout)) + log.WithFields(map[string]any{ + "duration": time.Since(start), + }).Infof("%s >-< %s", conn.RemoteAddr(), addr) + + return nil +} + +// sniffAndHandle peeks at the initial bytes of the client connection to +// determine the protocol. If HTTP or TLS is detected, the connection is +// handed off to the appropriate sniffer handler for protocol-aware forwarding. +// The dial and dialTLS closures return the already-established upstream +// connection so that the sniffer uses the same tunnel. +// +// Returns (true, err) when the sniffer took over the connection. Returns +// (false, nil) when the protocol is unrecognised and the caller should +// fall back to raw pipe forwarding. +func (h *httpHandler) sniffAndHandle(ctx context.Context, conn net.Conn, cc net.Conn, ro *xrecorder.HandlerRecorderObject, log logger.Logger) (handled bool, err error) { + if h.md.sniffingTimeout > 0 { + conn.SetReadDeadline(time.Now().Add(h.md.sniffingTimeout)) + } + + br := bufio.NewReader(conn) + proto, _ := sniffing.Sniff(ctx, br) + ro.Proto = proto + + if h.md.sniffingTimeout > 0 { + conn.SetReadDeadline(time.Time{}) + } + + // Both dial closures return the existing upstream connection so the + // sniffer uses the same tunnel rather than dialling a new one. + dial := func(ctx context.Context, network, address string) (net.Conn, error) { + return cc, nil + } + dialTLS := func(ctx context.Context, network, address string, cfg *tls.Config) (net.Conn, error) { + return cc, nil + } + sniffer := &sniffing.Sniffer{ + Websocket: h.md.sniffingWebsocket, + WebsocketSampleRate: h.md.sniffingWebsocketSampleRate, + Recorder: h.recorder.Recorder, + RecorderOptions: h.recorder.Options, + Certificate: h.md.certificate, + PrivateKey: h.md.privateKey, + NegotiatedProtocol: h.md.alpn, + CertPool: h.certPool, + MitmBypass: h.md.mitmBypass, + ReadTimeout: h.md.readTimeout, + } + + conn = xnet.NewReadWriteConn(br, conn, conn) + switch proto { + case sniffing.ProtoHTTP: + return true, sniffer.HandleHTTP(ctx, "tcp", conn, + sniffing.WithService(h.options.Service), + sniffing.WithDial(dial), + sniffing.WithDialTLS(dialTLS), + sniffing.WithRecorderObject(ro), + sniffing.WithLog(log), + ) + case sniffing.ProtoTLS: + return true, sniffer.HandleTLS(ctx, "tcp", conn, + sniffing.WithService(h.options.Service), + sniffing.WithDial(dial), + sniffing.WithDialTLS(dialTLS), + sniffing.WithRecorderObject(ro), + sniffing.WithLog(log), + ) + } + + return false, nil +} + +// dial establishes an upstream connection through the proxy chain router. +// It attaches the recorder object and logger from the context so that the +// router can annotate the route path and source/destination addresses. +// +// When h.md.hash is "host", the target address is used as the hash source +// for consistent hop selection across the chain. +func (h *httpHandler) dial(ctx context.Context, network, addr string) (conn net.Conn, err error) { + switch h.md.hash { + case "host": + ctx = xctx.ContextWithHash(ctx, &xctx.Hash{Source: addr}) + } + + if log := ictx.LoggerFromContext(ctx); log != nil { + log.Debugf("dial: new connection to host %s", addr) + } + + if h.options.Router == nil { + return nil, &net.OpError{Op: "dial", Net: network, Addr: nil, Err: errors.New("nil router")} + } + + var buf bytes.Buffer + conn, err = h.options.Router.Dial(ictx.ContextWithBuffer(ctx, &buf), network, addr) + if ro := ictx.RecorderObjectFromContext(ctx); ro != nil { + ro.Route = buf.String() + + if conn != nil { + ro.SrcAddr = conn.LocalAddr().String() + ro.DstAddr = conn.RemoteAddr().String() + } + } + + return +} + +// setupTrafficLimiter wraps the connection with per-client traffic shaping +// and optional stats tracking. When an Observer is configured, per-client +// stats counters (total connections, current connections) are updated. +// +// The returned net.Conn reads and writes through the limiter and stats +// wrappers while still satisfying the net.Conn interface. +// +// The cleanup function (non-nil when an Observer is configured) must be +// deferred by the caller to decrement KindCurrentConns when the connection +// handling completes. It captures the per-client stats object so the +// deferred call fires in the caller's scope, not when setupTrafficLimiter +// returns. +func (h *httpHandler) setupTrafficLimiter(conn net.Conn, clientID, network, addr string) (net.Conn, func()) { + rw := traffic_wrapper.WrapReadWriter( + h.limiter, + conn, + clientID, + limiter.ScopeOption(limiter.ScopeClient), + limiter.ServiceOption(h.options.Service), + limiter.NetworkOption(network), + limiter.AddrOption(addr), + limiter.ClientOption(clientID), + limiter.SrcOption(conn.RemoteAddr().String()), + ) + var cleanup func() + if h.options.Observer != nil { + pstats := h.stats.Stats(clientID) + pstats.Add(stats.KindTotalConns, 1) + pstats.Add(stats.KindCurrentConns, 1) + cleanup = func() { pstats.Add(stats.KindCurrentConns, -1) } + rw = stats_wrapper.WrapReadWriter(rw, pstats) + } + + return xnet.NewReadWriteConn(rw, rw, conn), cleanup +} diff --git a/handler/http/connect_test.go b/handler/http/connect_test.go new file mode 100644 index 00000000..dd97f9ad --- /dev/null +++ b/handler/http/connect_test.go @@ -0,0 +1,156 @@ +package http + +import ( + "context" + "io" + "net" + "net/http" + "testing" + + "github.com/go-gost/core/handler" + xrecorder "github.com/go-gost/x/recorder" +) + +func TestSniffAndHandle_NoMatch(t *testing.T) { + h := &httpHandler{ + options: handler.Options{ + Logger: &testLogger{}, + }, + } + h.md.sniffing = true + + client, server := net.Pipe() + defer client.Close() + defer server.Close() + + go func() { + client.Write([]byte("SSH-2.0-OpenSSH\r\n")) + }() + + cc, _ := net.Pipe() + defer cc.Close() + + handled, err := h.sniffAndHandle(context.Background(), server, cc, &xrecorder.HandlerRecorderObject{}, &testLogger{}) + if err != nil { + t.Fatal(err) + } + if handled { + t.Error("expected not handled for non-HTTP/TLS traffic") + } +} + +func TestDial_NilRouter(t *testing.T) { + h := &httpHandler{ + options: handler.Options{ + Logger: &testLogger{}, + }, + } + + _, err := h.dial(context.Background(), "tcp", "example.com:80") + if err == nil { + t.Error("expected error from dial with nil router") + } +} + +func TestSniffAndHandle_TLS(t *testing.T) { + h := &httpHandler{ + options: handler.Options{ + Logger: &testLogger{}, + }, + } + h.md.sniffing = true + + client, server := net.Pipe() + defer client.Close() + defer server.Close() + + // Write TLS ClientHello bytes (content type 0x16 = Handshake, followed by TLS version) + go func() { + client.Write([]byte{0x16, 0x03, 0x01, 0x00, 0x05, 0x01, 0x02, 0x03, 0x04, 0x05}) + }() + + cc, _ := net.Pipe() + defer cc.Close() + + handled, err := h.sniffAndHandle(context.Background(), server, cc, &xrecorder.HandlerRecorderObject{}, &testLogger{}) + if err != nil { + t.Logf("sniff error (expected without proper recorder): %v", err) + } + if !handled { + t.Error("expected handled for TLS traffic") + } +} + +func TestHandleConnect_NilRouter(t *testing.T) { + h := &httpHandler{ + options: handler.Options{ + Logger: &testLogger{}, + }, + } + h.md.proxyAgent = defaultProxyAgent + h.md.readTimeout = 15 + + client, server := net.Pipe() + defer client.Close() + defer server.Close() + + // Drain response to prevent Pipe blocking + go func() { + io.ReadAll(server) + }() + + resp := testConnectResp() + err := h.handleConnect(context.Background(), client, &xrecorder.HandlerRecorderObject{}, &testLogger{}, "example.com:443", resp) + if err == nil { + t.Error("expected dial error with nil router") + } +} + +func TestHandleConnect_SniffingEnabled(t *testing.T) { + h := &httpHandler{ + options: handler.Options{ + Logger: &testLogger{}, + }, + } + h.md.proxyAgent = defaultProxyAgent + h.md.readTimeout = 15 + h.md.sniffing = true + + client, server := net.Pipe() + defer client.Close() + defer server.Close() + + // Drain response to prevent Pipe blocking + go func() { + io.ReadAll(server) + }() + + resp := testConnectResp() + err := h.handleConnect(context.Background(), client, &xrecorder.HandlerRecorderObject{}, &testLogger{}, "example.com:443", resp) + if err == nil { + t.Error("expected dial error with nil router") + } +} + +func testConnectResp() *http.Response { + return &http.Response{ + ProtoMajor: 1, + ProtoMinor: 1, + Header: http.Header{}, + ContentLength: -1, + } +} + +func TestDial_HostHash(t *testing.T) { + h := &httpHandler{ + options: handler.Options{ + Logger: &testLogger{}, + }, + } + h.md.hash = "host" + + _, err := h.dial(context.Background(), "tcp", "example.com:80") + if err == nil { + t.Error("expected error from dial with nil router") + } +} diff --git a/handler/http/handler.go b/handler/http/handler.go index fc17f443..3b7fa757 100644 --- a/handler/http/handler.go +++ b/handler/http/handler.go @@ -1,27 +1,202 @@ +// Package http implements an HTTP/HTTPS proxy handler. It supports both +// forward proxy (GET/POST/CONNECT) and tunnel modes with optional TLS +// sniffing, MITM decryption, WebSocket frame recording, probe resistance, +// and UDP relay. +// +// The handler is registered under the name "http" and created via +// NewHandler in init(). +// +// # Request processing flow +// +// Each inbound net.Conn is handled by Handle, which reads exactly one +// HTTP request and delegates to handleRequest for routing. The flow +// through the handler proceeds as follows: +// +// Handle() +// ├─ stats.WrapConn (per-connection I/O counters) +// ├─ checkRateLimit (connection rate limiter, if configured) +// ├─ http.ReadRequest +// ├─ xhttp.GetClientIP (extract client IP from headers) +// └─ handleRequest() +// +// # Request dispatch (handleRequest) +// +// handleRequest is the central dispatcher. It normalises the request +// and routes to one of three code paths: +// +// 1. URL scheme inference — if the URL is not absolute (plain HTTP +// proxy request), the Host header is validated with govalidator. +// A valid DNS name or IP address implies the scheme is "http". +// +// 2. Network detection — reads the X-Gost-Protocol header. If the +// value is "udp" the request is treated as a UDP relay; otherwise +// it defaults to "tcp". +// +// 3. GOST v2 compatibility — decodes Gost-Target and X-Gost-Target +// headers (base64+CRC32 encoding) to recover the actual target host +// from older GOST clients. +// +// 4. Host normalisation — appends ":80" if no port is present. +// +// 5. Authentication — see "Authentication and probe resistance" below. +// +// 6. Bypass check — if a bypass matcher is configured and the target +// address matches, a 403 Forbidden is returned. +// +// 7. Routing: +// • X-Gost-Protocol: udp → handleUDP +// • setupTrafficLimiter wraps the conn (per-client shaping + observer stats) +// • Method != CONNECT → handleProxy (forward proxy) +// • Method == CONNECT → handleConnect (tunnel) +// +// # Forward proxy path (handleProxy → proxyRoundTrip) +// +// handleProxy wraps the connection with per-request stats, performs the +// first round-trip via proxyRoundTrip, then enters a keep-alive loop +// reading and proxying subsequent requests on the same connection until +// the client or upstream closes. +// +// proxyRoundTrip performs a single HTTP round-trip: +// - Clones the recorder object for per-request isolation. +// - Normalises the host (adds default port if missing). +// - Handles HTTP/1.0 compatibility (Connection: keep-alive / close). +// - Strips proxy-specific headers: Proxy-Authorization, Proxy-Connection, +// Gost-Target, X-Gost-Target. +// - Checks bypass rules (403 Forbidden on match). +// - Optionally intercepts the request body for recording (tee reader). +// - Sends the request through the upstream http.Transport (which dials +// through the proxy chain router via h.dial). +// - Optionally intercepts the response body for recording. +// - Handles 101 Switching Protocols (see "WebSocket upgrade" below). +// - Writes the response to the client. +// +// # CONNECT tunnel path (handleConnect → sniffAndHandle) +// +// handleConnect implements the HTTP CONNECT tunnel: +// 1. Dials the target address through the proxy chain router (h.dial). +// 2. On dial failure, writes a 503 Service Unavailable to the client. +// 3. Writes "200 Connection established" to the client. +// 4. If sniffing is enabled, calls sniffAndHandle to inspect the +// initial bytes from the client. +// 5. If sniffAndHandle does not claim the connection, falls back to +// raw bidirectional pipe forwarding (xnet.Pipe). +// +// sniffAndHandle peeks at the client's initial bytes: +// - Sets a read deadline (sniffing.timeout) to bound the peek. +// - Calls sniffing.Sniff to classify the protocol. +// - For HTTP traffic: delegates to sniffer.HandleHTTP (HTTP request +// routing through the tunnel). +// - For TLS traffic: delegates to sniffer.HandleTLS (MITM decryption +// if certificates are configured, or blind forwarding). +// - For unrecognised protocols: returns (false, nil), and the caller +// falls back to raw pipe forwarding. +// - Both sniff handlers receive dial closures that return the +// already-established upstream connection, so the sniffer uses the +// same tunnel rather than opening a new one. +// +// # WebSocket upgrade (handleUpgradeResponse) +// +// When proxyRoundTrip receives a 101 Switching Protocols response: +// 1. Validates that the request and response upgrade types match. +// 2. Extracts the backend connection from the response body +// (the HTTP transport upgrades the connection on 101). +// 3. Writes the 101 response to the client. +// 4. If the upgrade is "websocket" and sniffing.websocket is enabled, +// starts frame-level recording via sniffingWebsocketFrame. +// 5. Otherwise, falls back to raw bidirectional pipe forwarding. +// +// sniffingWebsocketFrame runs two goroutines (one per direction) that +// read WebSocket frames, record metadata and optionally the payload +// (subject to a rate limiter with sniffing.websocket.sampleRate), and +// forward the frames to the peer. The first direction to encounter an +// error terminates the relay. +// +// # Authentication and probe resistance (authenticate) +// +// authenticate extracts credentials from the Proxy-Authorization header +// (HTTP Basic auth) and validates them against the configured Auther. +// +// When no Auther is configured, all requests are anonymous (id="", ok=true). +// +// When authentication fails and probeResistance is configured, the handler +// returns a decoy response instead of the normal 407 Proxy-Auth-Required, +// making the port appear to run a different service. Four strategies are +// supported: +// +// code: respond with a custom HTTP status code (e.g. "code:404") +// web: fetch a URL and replay its response (e.g. "web:example.com") +// host: forward the raw request to a decoy host (e.g. "host:1.2.3.4:80") +// file: serve a local file with Content-Type: text/html +// +// The knock feature (pr.Knock) restricts probe resistance to clients +// that don't know a secret hostname. When pr.Knock is set and the request +// hostname matches, probe resistance is bypassed and the normal 407 is +// returned — revealing the proxy only to clients that know the knock +// address. +// +// # UDP relay (handleUDP) +// +// When X-Gost-Protocol is "udp": +// - If enableUDP is false, a 403 Forbidden is returned. +// - Otherwise a 200 OK is written, a UDP association is dialled through +// the proxy chain router, and the HTTP connection is wrapped as a +// SOCKS5 UDP tunnel. Client data is read as SOCKS5-encapsulated UDP +// datagrams and relayed through the chain. +// +// # Upstream dialing (dial) +// +// dial establishes upstream connections through the proxy chain router. +// It attaches the recorder object and logger from the context and, when +// h.md.hash is "host", sets the target address as the hash source for +// consistent hop selection. The route taken is written to the recorder +// object via the context buffer. +// +// # Initialisation (Init) +// +// Init parses metadata, creates a cached traffic limiter, starts the +// background stats observer goroutine, picks the first service-handler +// recorder, initialises the MITM certificate pool (if mitm.certFile and +// mitm.keyFile are both provided), and builds the upstream http.Transport. +// The transport dials through the proxy chain via h.dial. +// +// # Observability +// +// Per-connection stats are tracked via stats_wrapper.WrapConn in Handle. +// Per-client stats (KindTotalConns, KindCurrentConns) are tracked in +// setupTrafficLimiter and published to the configured Observer on a +// periodic ticker (observerPeriod, default 5s, min 1s). The observeStats +// goroutine retries previously failed events before sending new ones. +// +// # Key configuration (metadata) +// +// readTimeout: upstream response header timeout (default 15s, negative disables) +// idleTimeout: pipe forwarding idle timeout (0 = disabled) +// keepalive: enable HTTP keep-alive on the upstream transport +// compression: enable HTTP compression on the upstream transport +// probeResist: decoy response on auth failure (format "type:value") +// knock: secret hostname that bypasses probe resistance +// udp: enable UDP relay +// sniffing: enable protocol sniffing on CONNECT tunnels +// sniffing.timeout: read deadline for the initial sniff peek +// sniffing.websocket: enable WebSocket frame recording +// sniffing.websocket.sampleRate: max frames recorded per second +// mitm.certFile / mitm.keyFile: CA certificate pair for TLS decryption +// mitm.bypass: named bypass matcher to skip MITM decryption +// hash: hop selection strategy ("host" for consistent hashing) +// observerPeriod: stats reporting interval (default 5s, min 1s) +// proxyAgent: Proxy-Agent header value (default "gost/3.0") package http import ( "bufio" - "bytes" "context" - "crypto/tls" - "encoding/base64" - "encoding/binary" "errors" - "fmt" - "hash/crc32" - "io" - "math" "net" "net/http" "net/http/httputil" - "os" - "strconv" - "strings" "time" "github.com/asaskevich/govalidator" - "github.com/go-gost/core/auth" "github.com/go-gost/core/bypass" "github.com/go-gost/core/handler" "github.com/go-gost/core/limiter" @@ -29,44 +204,42 @@ import ( "github.com/go-gost/core/logger" md "github.com/go-gost/core/metadata" "github.com/go-gost/core/observer" - "github.com/go-gost/core/observer/stats" + stats "github.com/go-gost/core/observer/stats" "github.com/go-gost/core/recorder" xbypass "github.com/go-gost/x/bypass" xctx "github.com/go-gost/x/ctx" - ictx "github.com/go-gost/x/internal/ctx" - xio "github.com/go-gost/x/internal/io" xnet "github.com/go-gost/x/internal/net" xhttp "github.com/go-gost/x/internal/net/http" - "github.com/go-gost/x/internal/util/sniffing" stats_util "github.com/go-gost/x/internal/util/stats" tls_util "github.com/go-gost/x/internal/util/tls" - ws_util "github.com/go-gost/x/internal/util/ws" rate_limiter "github.com/go-gost/x/limiter/rate" cache_limiter "github.com/go-gost/x/limiter/traffic/cache" - traffic_wrapper "github.com/go-gost/x/limiter/traffic/wrapper" xstats "github.com/go-gost/x/observer/stats" stats_wrapper "github.com/go-gost/x/observer/stats/wrapper" xrecorder "github.com/go-gost/x/recorder" "github.com/go-gost/x/registry" - "golang.org/x/net/http/httpguts" - "golang.org/x/time/rate" ) func init() { registry.HandlerRegistry().Register("http", NewHandler) } +// httpHandler is the HTTP/HTTPS proxy handler implementation. It holds the +// parsed metadata, handler options, and runtime state such as the traffic +// limiter, recorder, certificate pool, and the upstream HTTP transport. type httpHandler struct { - md metadata - options handler.Options - stats *stats_util.HandlerStats - limiter traffic.TrafficLimiter - cancel context.CancelFunc - recorder recorder.RecorderObject - certPool tls_util.CertPool - transport http.RoundTripper + md metadata // parsed configuration + options handler.Options // handler options from the service config + stats *stats_util.HandlerStats // per-client stats, created when Observer is set + limiter traffic.TrafficLimiter // per-client traffic shaper (cached) + cancel context.CancelFunc // cancels the observeStats goroutine + recorder recorder.RecorderObject // first matching service-handler recorder + certPool tls_util.CertPool // in-memory cert pool for MITM TLS termination + transport http.RoundTripper // upstream HTTP transport (injectable for tests) } +// NewHandler creates a new HTTP handler and applies the given options. +// It is called by the registry when a service references the "http" handler. func NewHandler(opts ...handler.Option) handler.Handler { options := handler.Options{} for _, opt := range opts { @@ -78,6 +251,9 @@ func NewHandler(opts ...handler.Option) handler.Handler { } } +// Init parses metadata, creates the traffic limiter and stats tracker, +// picks the first matching recorder, initialises the MITM cert pool if +// a certificate is configured, and builds the upstream http.Transport. func (h *httpHandler) Init(md md.Metadata) error { if err := h.parseMetadata(md); err != nil { return err @@ -110,6 +286,8 @@ func (h *httpHandler) Init(md md.Metadata) error { h.certPool = tls_util.NewMemoryCertPool() } + // Build the upstream transport that routes through the proxy chain. + // DialContext uses h.dial which goes through h.options.Router. h.transport = &http.Transport{ DialContext: h.dial, IdleConnTimeout: 30 * time.Second, @@ -121,6 +299,11 @@ func (h *httpHandler) Init(md md.Metadata) error { return nil } +// Handle processes an inbound connection. It reads one HTTP request, +// wraps the connection for stats, applies rate limiting, extracts the +// client IP, and delegates to handleRequest for routing. +// +// The connection is always closed when Handle returns. func (h *httpHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler.HandleOption) (err error) { defer conn.Close() @@ -190,14 +373,18 @@ func (h *httpHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler return h.handleRequest(ctx, conn, req, ro, log) } -func (h *httpHandler) Close() error { - if h.cancel != nil { - h.cancel() - } - return nil -} - +// handleRequest is the central request dispatcher. It normalises the request +// URL and host, extracts the target from GOST-compatible headers, validates +// the method, authenticates the client, applies bypass rules, and routes +// to one of three code paths: +// +// - UDP: handleUDP +// - HTTP forward proxy (GET/POST/…): handleProxy +// - HTTP CONNECT tunnel: handleConnect func (h *httpHandler) handleRequest(ctx context.Context, conn net.Conn, req *http.Request, ro *xrecorder.HandlerRecorderObject, log logger.Logger) error { + // If the URL is not absolute (typical for plain HTTP proxy requests), + // attempt to infer the scheme by validating the Host header. A valid + // DNS name or IP address implies http://. if !req.URL.IsAbs() { host := req.Host if h, _, err := net.SplitHostPort(host); err == nil { @@ -214,23 +401,20 @@ func (h *httpHandler) handleRequest(ctx context.Context, conn net.Conn, req *htt } ro.Network = network - // Try to get the actual host. - // Compatible with GOST 2.x. + // GOST v2 sent the actual target host in Gost-Target / X-Gost-Target + // headers using a base64+CRC32 encoding. Decode them for compatibility. if v := req.Header.Get("Gost-Target"); v != "" { - if h, err := h.decodeServerName(v); err == nil { + if h, err := decodeServerName(v); err == nil { req.Host = h } } if v := req.Header.Get("X-Gost-Target"); v != "" { - if h, err := h.decodeServerName(v); err == nil { + if h, err := decodeServerName(v); err == nil { req.Host = h } } - addr := req.Host - if _, port, _ := net.SplitHostPort(addr); port == "" { - addr = net.JoinHostPort(strings.Trim(addr, "[]"), "80") - } + addr := normalizeHostPort(req.Host, "80") ro.Host = addr fields := map[string]any{ @@ -239,7 +423,7 @@ func (h *httpHandler) handleRequest(ctx context.Context, conn net.Conn, req *htt "network": network, } - if u, _, _ := h.basicProxyAuth(req.Header.Get("Proxy-Authorization")); u != "" { + if u, _, _ := basicProxyAuth(req.Header.Get("Proxy-Authorization")); u != "" { fields["user"] = u ro.ClientID = u } @@ -251,6 +435,7 @@ func (h *httpHandler) handleRequest(ctx context.Context, conn net.Conn, req *htt } log.Debugf("%s >> %s", conn.RemoteAddr(), addr) + // Build a skeleton response reused across error/auth/bypass writes. resp := &http.Response{ ProtoMajor: 1, ProtoMinor: 1, @@ -277,6 +462,8 @@ func (h *httpHandler) handleRequest(ctx context.Context, conn net.Conn, req *htt ro.HTTP.Response.Header = resp.Header }() + // HTTP/2 connection preface "PRI * HTTP/2.0" is rejected, as are + // non-CONNECT requests without an http:// scheme. if req.Method == "PRI" || (req.Method != http.MethodConnect && req.URL.Scheme != "http") { resp.StatusCode = http.StatusBadRequest @@ -322,675 +509,30 @@ func (h *httpHandler) handleRequest(ctx context.Context, conn net.Conn, req *htt return h.handleUDP(ctx, conn, ro, log) } - { - rw := traffic_wrapper.WrapReadWriter( - h.limiter, - conn, - clientID, - limiter.ScopeOption(limiter.ScopeClient), - limiter.ServiceOption(h.options.Service), - limiter.NetworkOption(network), - limiter.AddrOption(addr), - limiter.ClientOption(clientID), - limiter.SrcOption(conn.RemoteAddr().String()), - ) - if h.options.Observer != nil { - pstats := h.stats.Stats(clientID) - pstats.Add(stats.KindTotalConns, 1) - pstats.Add(stats.KindCurrentConns, 1) - defer pstats.Add(stats.KindCurrentConns, -1) - rw = stats_wrapper.WrapReadWriter(rw, pstats) - } - - conn = xnet.NewReadWriteConn(rw, rw, conn) + conn, done := h.setupTrafficLimiter(conn, clientID, network, addr) + if done != nil { + defer done() } if req.Method != http.MethodConnect { return h.handleProxy(ctx, conn, req, ro, log) } - ctx = ictx.ContextWithRecorderObject(ctx, ro) - ctx = ictx.ContextWithLogger(ctx, log) - cc, err := h.dial(ctx, "tcp", addr) + return h.handleConnect(ctx, conn, ro, log, addr, resp) +} - if err != nil { - resp.StatusCode = http.StatusServiceUnavailable - - if log.IsLevelEnabled(logger.TraceLevel) { - dump, _ := httputil.DumpResponse(resp, false) - log.Trace(string(dump)) - } - if err := resp.Write(conn); err != nil { - log.Error("write error response: ", err) - } - return err +// Close cancels the background stats goroutine started in Init. +// It is safe to call on a handler that was never initialised. +func (h *httpHandler) Close() error { + if h.cancel != nil { + h.cancel() } - snifferHandled := false - defer func() { - if !snifferHandled { - cc.Close() - } - }() - - log = log.WithFields(map[string]any{"src": cc.LocalAddr().String(), "dst": cc.RemoteAddr().String()}) - ro.SrcAddr = cc.LocalAddr().String() - ro.DstAddr = cc.RemoteAddr().String() - - b := []byte("HTTP/1.1 200 Connection established\r\n" + - "Proxy-Agent: " + h.md.proxyAgent + "\r\n\r\n") - - if log.IsLevelEnabled(logger.TraceLevel) { - log.Trace(string(b)) - } - if _, err = conn.Write(b); err != nil { - log.Error(err) - return err - } - - if h.md.sniffing { - if h.md.sniffingTimeout > 0 { - conn.SetReadDeadline(time.Now().Add(h.md.sniffingTimeout)) - } - - br := bufio.NewReader(conn) - proto, _ := sniffing.Sniff(ctx, br) - ro.Proto = proto - - if h.md.sniffingTimeout > 0 { - conn.SetReadDeadline(time.Time{}) - } - - dial := func(ctx context.Context, network, address string) (net.Conn, error) { - return cc, nil - } - dialTLS := func(ctx context.Context, network, address string, cfg *tls.Config) (net.Conn, error) { - return cc, nil - } - sniffer := &sniffing.Sniffer{ - Websocket: h.md.sniffingWebsocket, - WebsocketSampleRate: h.md.sniffingWebsocketSampleRate, - Recorder: h.recorder.Recorder, - RecorderOptions: h.recorder.Options, - Certificate: h.md.certificate, - PrivateKey: h.md.privateKey, - NegotiatedProtocol: h.md.alpn, - CertPool: h.certPool, - MitmBypass: h.md.mitmBypass, - ReadTimeout: h.md.readTimeout, - } - - conn = xnet.NewReadWriteConn(br, conn, conn) - switch proto { - case sniffing.ProtoHTTP: - snifferHandled = true - return sniffer.HandleHTTP(ctx, "tcp", conn, - sniffing.WithService(h.options.Service), - sniffing.WithDial(dial), - sniffing.WithDialTLS(dialTLS), - sniffing.WithRecorderObject(ro), - sniffing.WithLog(log), - ) - case sniffing.ProtoTLS: - snifferHandled = true - return sniffer.HandleTLS(ctx, "tcp", conn, - sniffing.WithService(h.options.Service), - sniffing.WithDial(dial), - sniffing.WithDialTLS(dialTLS), - sniffing.WithRecorderObject(ro), - sniffing.WithLog(log), - ) - } - } - - start := time.Now() - log.Infof("%s <-> %s", conn.RemoteAddr(), addr) - // xnet.Transport(conn, cc) - xnet.Pipe(ctx, conn, cc, xnet.WithReadTimeout(h.md.idleTimeout)) - log.WithFields(map[string]any{ - "duration": time.Since(start), - }).Infof("%s >-< %s", conn.RemoteAddr(), addr) - return nil } -func (h *httpHandler) handleProxy(ctx context.Context, conn net.Conn, req *http.Request, ro *xrecorder.HandlerRecorderObject, log logger.Logger) error { - pStats := xstats.Stats{} - conn = stats_wrapper.WrapConn(conn, &pStats) - - ro.Time = time.Time{} - - if close, err := h.proxyRoundTrip(ctx, conn, req, ro, &pStats, log); err != nil || close { - return err - } - - br := bufio.NewReader(conn) - for { - pStats.Reset() - - req, err := http.ReadRequest(br) - if err != nil { - if errors.Is(err, io.EOF) || errors.Is(err, net.ErrClosed) { - return nil - } - return err - } - - if log.IsLevelEnabled(logger.TraceLevel) { - dump, _ := httputil.DumpRequest(req, false) - log.Trace(string(dump)) - } - - if close, err := h.proxyRoundTrip(ctx, xio.NewReadWriteCloser(br, conn, conn), req, ro, &pStats, log); err != nil || close { - req.Body.Close() - return err - } - req.Body.Close() - } -} - -func (h *httpHandler) proxyRoundTrip(ctx context.Context, rw io.ReadWriteCloser, req *http.Request, ro *xrecorder.HandlerRecorderObject, pStats stats.Stats, log logger.Logger) (close bool, err error) { - close = true - - ro2 := &xrecorder.HandlerRecorderObject{} - *ro2 = *ro - ro = ro2 - - host := req.Host - if _, port, _ := net.SplitHostPort(host); port == "" { - host = net.JoinHostPort(strings.Trim(host, "[]"), "80") - } - ro.Host = host - ro.Time = time.Now() - - log = log.WithFields(map[string]any{ - "host": host, - }) - - log.Infof("%s <-> %s", ro.RemoteAddr, req.Host) - defer func() { - if err != nil { - ro.Err = err.Error() - } - ro.InputBytes = pStats.Get(stats.KindInputBytes) - ro.OutputBytes = pStats.Get(stats.KindOutputBytes) - ro.Duration = time.Since(ro.Time) - if err := ro.Record(ctx, h.recorder.Recorder); err != nil { - log.Errorf("record: %v", err) - } - - log.WithFields(map[string]any{ - "duration": time.Since(ro.Time), - "inputBytes": ro.InputBytes, - "outputBytes": ro.OutputBytes, - }).Infof("%s >-< %s", ro.RemoteAddr, req.Host) - }() - - ro.HTTP = &xrecorder.HTTPRecorderObject{ - Host: req.Host, - Proto: req.Proto, - Scheme: req.URL.Scheme, - Method: req.Method, - URI: req.RequestURI, - Request: xrecorder.HTTPRequestRecorderObject{ - ContentLength: req.ContentLength, - Header: req.Header.Clone(), - }, - } - - // HTTP/1.0 - http10 := req.ProtoMajor == 1 && req.ProtoMinor == 0 - if http10 { - if strings.ToLower(req.Header.Get("Connection")) == "keep-alive" { - req.Header.Del("Connection") - } else { - req.Header.Set("Connection", "close") - } - } - - req.Header.Del("Proxy-Authorization") - req.Header.Del("Proxy-Connection") - req.Header.Del("Gost-Target") - req.Header.Del("X-Gost-Target") - - res := &http.Response{ - ProtoMajor: req.ProtoMajor, - ProtoMinor: req.ProtoMinor, - Header: http.Header{}, - StatusCode: http.StatusServiceUnavailable, - } - ro.HTTP.StatusCode = res.StatusCode - - if h.options.Bypass != nil && - h.options.Bypass.Contains(ctx, "tcp", host, bypass.WithService(h.options.Service)) { - res.StatusCode = http.StatusForbidden - - if log.IsLevelEnabled(logger.TraceLevel) { - dump, _ := httputil.DumpResponse(res, false) - log.Trace(string(dump)) - } - log.Debug("bypass: ", host) - if werr := res.Write(rw); werr != nil { - log.Error("write bypass response: ", werr) - } - err = xbypass.ErrBypass - return - } - - var reqBody *xhttp.Body - if opts := h.recorder.Options; opts != nil && opts.HTTPBody { - if req.Body != nil { - bodySize := opts.MaxBodySize - if bodySize <= 0 { - bodySize = sniffing.DefaultBodySize - } - if bodySize > sniffing.MaxBodySize { - bodySize = sniffing.MaxBodySize - } - reqBody = xhttp.NewBody(req.Body, bodySize) - req.Body = reqBody - } - } - - ctx = ictx.ContextWithRecorderObject(ctx, ro) - ctx = ictx.ContextWithLogger(ctx, log) - - resp, err := h.transport.RoundTrip(req.WithContext(ctx)) - - if reqBody != nil { - ro.HTTP.Request.Body = reqBody.Content() - ro.HTTP.Request.ContentLength = reqBody.Length() - } - - if err != nil { - if werr := res.Write(rw); werr != nil { - log.Error("write error response: ", werr) - } - return - } - defer resp.Body.Close() - - ro.HTTP.StatusCode = resp.StatusCode - ro.HTTP.Response.Header = resp.Header.Clone() - ro.HTTP.Response.ContentLength = resp.ContentLength - - if log.IsLevelEnabled(logger.TraceLevel) { - dump, _ := httputil.DumpResponse(resp, false) - log.Trace(string(dump)) - } - - // HTTP/1.0 - if http10 { - if !resp.Close { - resp.Header.Set("Connection", "keep-alive") - } - resp.ProtoMajor = 1 - resp.ProtoMinor = 0 - } - - if resp.StatusCode == http.StatusSwitchingProtocols { - err = h.handleUpgradeResponse(ctx, rw, req, resp, ro, log) - return - } - - var respBody *xhttp.Body - if opts := h.recorder.Options; opts != nil && opts.HTTPBody { - bodySize := opts.MaxBodySize - if bodySize <= 0 { - bodySize = sniffing.DefaultBodySize - } - if bodySize > sniffing.MaxBodySize { - bodySize = sniffing.MaxBodySize - } - respBody = xhttp.NewBody(resp.Body, bodySize) - resp.Body = respBody - } - - err = resp.Write(rw) - - if respBody != nil { - ro.HTTP.Response.Body = respBody.Content() - ro.HTTP.Response.ContentLength = respBody.Length() - } - - if err != nil { - err = fmt.Errorf("write response: %v", err) - return - } - - close = resp.Close - - return -} - -func (h *httpHandler) dial(ctx context.Context, network, addr string) (conn net.Conn, err error) { - switch h.md.hash { - case "host": - ctx = xctx.ContextWithHash(ctx, &xctx.Hash{Source: addr}) - } - - if log := ictx.LoggerFromContext(ctx); log != nil { - log.Debugf("dial: new connection to host %s", addr) - } - - var buf bytes.Buffer - conn, err = h.options.Router.Dial(ictx.ContextWithBuffer(ctx, &buf), network, addr) - if ro := ictx.RecorderObjectFromContext(ctx); ro != nil { - ro.Route = buf.String() - - if conn != nil { - ro.SrcAddr = conn.LocalAddr().String() - ro.DstAddr = conn.RemoteAddr().String() - } - } - - return -} - -func upgradeType(h http.Header) string { - if !httpguts.HeaderValuesContainsToken(h["Connection"], "Upgrade") { - return "" - } - return h.Get("Upgrade") -} - -func (h *httpHandler) handleUpgradeResponse(ctx context.Context, rw io.ReadWriteCloser, req *http.Request, res *http.Response, ro *xrecorder.HandlerRecorderObject, log logger.Logger) error { - reqUpType := upgradeType(req.Header) - resUpType := upgradeType(res.Header) - if !strings.EqualFold(reqUpType, resUpType) { - return fmt.Errorf("backend tried to switch protocol %q when %q was requested", resUpType, reqUpType) - } - - backConn, ok := res.Body.(io.ReadWriteCloser) - if !ok { - return fmt.Errorf("internal error: 101 switching protocols response with non-writable body") - } - - res.Body = nil - if err := res.Write(rw); err != nil { - return fmt.Errorf("response write: %v", err) - } - - if reqUpType == "websocket" && h.md.sniffingWebsocket { - return h.sniffingWebsocketFrame(ctx, rw, backConn, ro, log) - } - - // return xnet.Transport(rw, backConn) - return xnet.Pipe(ctx, rw, backConn, xnet.WithReadTimeout(h.md.idleTimeout)) -} - -func (h *httpHandler) sniffingWebsocketFrame(ctx context.Context, rw, cc io.ReadWriter, ro *xrecorder.HandlerRecorderObject, log logger.Logger) error { - errc := make(chan error, 1) - - sampleRate := h.md.sniffingWebsocketSampleRate - if sampleRate == 0 { - sampleRate = sniffing.DefaultSampleRate - } - if sampleRate < 0 { - sampleRate = math.MaxFloat64 - } - - go func() { - ro2 := &xrecorder.HandlerRecorderObject{} - *ro2 = *ro - 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.Recorder); err != nil { - log.Errorf("record: %v", err) - } - } - } - }() - - go func() { - ro2 := &xrecorder.HandlerRecorderObject{} - *ro2 = *ro - 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.Recorder); err != nil { - log.Errorf("record: %v", err) - } - } - } - }() - - <-errc - return nil -} - -func (h *httpHandler) 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 opts := h.recorder.Options; opts != nil && opts.HTTPBody { - bodySize := opts.MaxBodySize - if bodySize <= 0 { - bodySize = sniffing.DefaultBodySize - } - if bodySize > sniffing.MaxBodySize { - bodySize = sniffing.MaxBodySize - } - - 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 -} - -func (h *httpHandler) decodeServerName(s string) (string, error) { - b, err := base64.RawURLEncoding.DecodeString(s) - if err != nil { - return "", err - } - if len(b) < 4 { - return "", errors.New("invalid name") - } - v, err := base64.RawURLEncoding.DecodeString(string(b[4:])) - if err != nil { - return "", err - } - if crc32.ChecksumIEEE(v) != binary.BigEndian.Uint32(b[:4]) { - return "", errors.New("invalid name") - } - return string(v), nil -} - -func (h *httpHandler) basicProxyAuth(proxyAuth string) (username, password string, ok bool) { - if proxyAuth == "" { - return - } - - if !strings.HasPrefix(proxyAuth, "Basic ") { - return - } - c, err := base64.StdEncoding.DecodeString(strings.TrimPrefix(proxyAuth, "Basic ")) - if err != nil { - return - } - cs := string(c) - s := strings.IndexByte(cs, ':') - if s < 0 { - return - } - - return cs[:s], cs[s+1:], true -} - -func (h *httpHandler) authenticate(ctx context.Context, conn net.Conn, req *http.Request, resp *http.Response, log logger.Logger) (id string, ok bool) { - u, p, _ := h.basicProxyAuth(req.Header.Get("Proxy-Authorization")) - if h.options.Auther == nil { - return "", true - } - if id, ok = h.options.Auther.Authenticate(ctx, u, p, auth.WithService(h.options.Service)); ok { - return - } - - pr := h.md.probeResistance - // Probe resistance activates on auth failure in two cases: - // 1. pr.Knock is empty — no specific knock host; always hide the proxy. - // 2. pr.Knock is set but the request hostname does not match — the client - // didn't "knock" on the right host, so a fake response is returned. - // When the request hostname matches a non-empty pr.Knock, probe resistance - // is bypassed and falls through to the normal 407 Proxy-Auth-Required below, - // revealing the proxy only to clients that know the knock address. - if pr != nil && (pr.Knock == "" || !strings.EqualFold(req.URL.Hostname(), pr.Knock)) { - resp.StatusCode = http.StatusServiceUnavailable // default status code - - switch pr.Type { - case "code": - if code, err := strconv.Atoi(pr.Value); err == nil { - resp.StatusCode = code - } else { - log.Warnf("invalid probe resistance code: %s", pr.Value) - } - case "web": - url := pr.Value - if !strings.HasPrefix(url, "http") { - url = "http://" + url - } - client := &http.Client{Timeout: 15 * time.Second} - r, err := client.Get(url) - if err != nil { - log.Error(err) - break - } - resp = r - defer resp.Body.Close() - case "host": - cc, err := net.Dial("tcp", pr.Value) - if err != nil { - log.Error(err) - break - } - defer cc.Close() - - req.Write(cc) - // xnet.Transport(conn, cc) - xnet.Pipe(ctx, conn, cc, xnet.WithReadTimeout(h.md.idleTimeout)) - return - case "file": - f, _ := os.Open(pr.Value) - if f != nil { - defer f.Close() - - resp.StatusCode = http.StatusOK - if finfo, _ := f.Stat(); finfo != nil { - resp.ContentLength = finfo.Size() - } - resp.Header.Set("Content-Type", "text/html") - resp.Body = f - } - } - } - - if resp.Header == nil { - resp.Header = http.Header{} - } - if resp.StatusCode == 0 { - realm := defaultRealm - if h.md.authBasicRealm != "" { - realm = h.md.authBasicRealm - } - resp.StatusCode = http.StatusProxyAuthRequired - resp.Header.Add("Proxy-Authenticate", fmt.Sprintf("Basic realm=\"%s\"", realm)) - if strings.ToLower(req.Header.Get("Proxy-Connection")) == "keep-alive" { - // XXX libcurl will keep sending auth request in same conn - // which we don't supported yet. - resp.Header.Set("Connection", "close") - resp.Header.Set("Proxy-Connection", "close") - } - - log.Debug("proxy authentication required") - } else { - // resp.Header.Set("Server", "nginx/1.20.1") - // resp.Header.Set("Date", time.Now().Format(http.TimeFormat)) - if resp.StatusCode == http.StatusOK { - resp.Header.Set("Connection", "keep-alive") - } - } - - if log.IsLevelEnabled(logger.TraceLevel) { - dump, _ := httputil.DumpResponse(resp, false) - log.Trace(string(dump)) - } - - if err := resp.Write(conn); err != nil { - log.Error("write auth response: ", err) - } - return -} - -func (h *httpHandler) checkRateLimit(addr net.Addr) bool { - if h.options.RateLimiter == nil { - return true - } - host, _, _ := net.SplitHostPort(addr.String()) - if limiter := h.options.RateLimiter.Limiter(host); limiter != nil { - return limiter.Allow(1) - } - - return true -} - +// observeStats periodically publishes per-client traffic stats to the +// configured Observer. It runs in a background goroutine started by Init. +// Events that fail to send are retried on the next tick. func (h *httpHandler) observeStats(ctx context.Context) { if h.options.Observer == nil { return @@ -1004,6 +546,7 @@ func (h *httpHandler) observeStats(ctx context.Context) { for { select { case <-ticker.C: + // Retry previously failed events first to preserve ordering. if len(events) > 0 { if err := h.options.Observer.Observe(ctx, events); err == nil { events = nil diff --git a/handler/http/handler_test.go b/handler/http/handler_test.go new file mode 100644 index 00000000..46b20854 --- /dev/null +++ b/handler/http/handler_test.go @@ -0,0 +1,220 @@ +package http + +import ( + "context" + "io" + "net" + "testing" + "time" + + "github.com/go-gost/core/handler" + cmdata "github.com/go-gost/core/metadata" + "github.com/go-gost/core/observer" + xmetadata "github.com/go-gost/x/metadata" +) + +func testMD(m map[string]any) cmdata.Metadata { + return xmetadata.NewMetadata(m) +} + +func TestNewHandler(t *testing.T) { + h := NewHandler() + if h == nil { + t.Fatal("NewHandler returned nil") + } + httpH, ok := h.(*httpHandler) + if !ok { + t.Fatal("NewHandler did not return *httpHandler") + } + if httpH.options.Logger != nil { + t.Error("expected nil logger") + } +} + +func TestNewHandler_WithOptions(t *testing.T) { + log := &testLogger{} + h := NewHandler( + handler.LoggerOption(log), + handler.ServiceOption("test-service"), + ) + httpH := h.(*httpHandler) + if httpH.options.Logger == nil { + t.Error("expected logger to be set") + } + if httpH.options.Service != "test-service" { + t.Errorf("got service %q, want %q", httpH.options.Service, "test-service") + } +} + +func TestInit(t *testing.T) { + h := NewHandler( + handler.LoggerOption(&testLogger{}), + ).(*httpHandler) + + err := h.Init(testMD(map[string]any{})) + if err != nil { + t.Fatalf("Init returned error: %v", err) + } + if h.transport == nil { + t.Error("expected transport to be initialized") + } + if h.md.proxyAgent == "" { + t.Error("expected default proxy agent") + } + if h.md.readTimeout == 0 { + t.Error("expected default read timeout") + } +} + +func TestInit_WithObserver(t *testing.T) { + h := NewHandler( + handler.LoggerOption(&testLogger{}), + handler.ObserverOption(&testObserver{}), + ).(*httpHandler) + + err := h.Init(testMD(map[string]any{})) + if err != nil { + t.Fatalf("Init returned error: %v", err) + } + if h.stats == nil { + t.Error("expected stats to be initialized when observer is set") + } +} + +func TestInit_WithLimiter(t *testing.T) { + h := NewHandler( + handler.LoggerOption(&testLogger{}), + handler.TrafficLimiterOption(&testTrafficLimiter{}), + ).(*httpHandler) + + err := h.Init(testMD(map[string]any{})) + if err != nil { + t.Fatalf("Init returned error: %v", err) + } + if h.limiter == nil { + t.Error("expected limiter to be initialized") + } +} + +func TestInit_WithRecorder(t *testing.T) { + h := NewHandler( + handler.LoggerOption(&testLogger{}), + ).(*httpHandler) + + err := h.Init(testMD(map[string]any{})) + if err != nil { + t.Fatalf("Init returned error: %v", err) + } + // recorder is nil since no matching recorder was passed + if h.recorder.Record != "" { + t.Log("recorder found even without explicit match") + } +} + +func TestClose(t *testing.T) { + h := NewHandler( + handler.LoggerOption(&testLogger{}), + ).(*httpHandler) + + err := h.Init(testMD(map[string]any{})) + if err != nil { + t.Fatalf("Init returned error: %v", err) + } + if h.cancel == nil { + t.Fatal("expected cancel function") + } + + err = h.Close() + if err != nil { + t.Errorf("Close returned error: %v", err) + } +} + +func TestClose_NoInit(t *testing.T) { + h := NewHandler().(*httpHandler) + err := h.Close() + if err != nil { + t.Errorf("Close returned error: %v", err) + } +} + +func TestHandle_PRIMethod(t *testing.T) { + h := NewHandler( + handler.LoggerOption(&testLogger{}), + ).(*httpHandler) + if err := h.Init(testMD(map[string]any{})); err != nil { + t.Fatal(err) + } + + client, server := net.Pipe() + defer client.Close() + defer server.Close() + + go func() { + // Write an HTTP/2 connection preface which parses as a PRI request + client.Write([]byte("PRI * HTTP/2.0\r\nHost: example.com\r\n\r\n")) + // Drain any response so Pipe doesn't block + io.ReadAll(client) + }() + + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + + err := h.Handle(ctx, server) + if err != nil { + t.Logf("Handle returned: %v", err) + } +} + +func TestHandle_InvalidHost(t *testing.T) { + h := NewHandler( + handler.LoggerOption(&testLogger{}), + ).(*httpHandler) + if err := h.Init(testMD(map[string]any{})); err != nil { + t.Fatal(err) + } + + client, server := net.Pipe() + defer client.Close() + defer server.Close() + + go func() { + // Invalid host that fails DNS name and IP parsing + client.Write([]byte("GET / HTTP/1.1\r\nHost: !!!invalid!!!\r\n\r\n")) + // Drain any response so Pipe doesn't block + io.ReadAll(client) + }() + + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + + err := h.Handle(ctx, server) + if err != nil { + t.Logf("Handle returned: %v", err) + } +} + +func TestSetupTrafficLimiter_NoObserver(t *testing.T) { + h := &httpHandler{ + options: handler.Options{ + Logger: &testLogger{}, + Service: "test", + }, + } + + client, _ := net.Pipe() + defer client.Close() + + result, done := h.setupTrafficLimiter(client, "test-client", "tcp", "example.com:80") + _ = done + if result == nil { + t.Error("expected non-nil connection") + } +} + +// testObserver implements observer.Observer for testing. +type testObserver struct{} + +func (o *testObserver) Observe(ctx context.Context, events []observer.Event, opts ...observer.Option) error { + return nil +} diff --git a/handler/http/metadata.go b/handler/http/metadata.go index abd159a0..055a3293 100644 --- a/handler/http/metadata.go +++ b/handler/http/metadata.go @@ -19,36 +19,45 @@ const ( defaultProxyAgent = "gost/3.0" ) +// metadata holds all configuration parsed from the handler's metadata map. +// Fields are populated by parseMetadata and read throughout the handler's +// request-processing methods. type metadata struct { - readTimeout time.Duration - idleTimeout time.Duration - keepalive bool - compression bool - probeResistance *probeResistance - enableUDP bool - udpBufferSize int - header http.Header - hash string - authBasicRealm string - proxyAgent string + readTimeout time.Duration // read timeout for upstream responses; default 15s, negative disables + idleTimeout time.Duration // idle timeout for pipe forwarding; 0 means disabled + keepalive bool // enable HTTP keep-alive on the upstream transport + compression bool // enable HTTP compression on the upstream transport - observerPeriod time.Duration - observerResetTraffic bool + probeResistance *probeResistance // decoy response config for auth failures (nil = disabled) + enableUDP bool // allow UDP relay over HTTP + udpBufferSize int // UDP relay buffer size in bytes - sniffing bool - sniffingTimeout time.Duration - sniffingWebsocket bool - sniffingWebsocketSampleRate float64 + header http.Header // additional response headers added to every proxy response + hash string // hash strategy for hop selection ("host" or empty) + authBasicRealm string // custom realm for Basic auth 407 responses + proxyAgent string // Proxy-Agent header value; default "gost/3.0" - certificate *x509.Certificate - privateKey crypto.PrivateKey - alpn string - mitmBypass bypass.Bypass + observerPeriod time.Duration // stats reporting interval; default 5s, min 1s + observerResetTraffic bool // reset traffic counters after each observation - limiterRefreshInterval time.Duration - limiterCleanupInterval time.Duration + sniffing bool // enable protocol sniffing on CONNECT tunnels + sniffingTimeout time.Duration // deadline for the initial sniff read + sniffingWebsocket bool // enable WebSocket frame recording + sniffingWebsocketSampleRate float64 // max frames recorded per second + + certificate *x509.Certificate // MITM CA certificate (for TLS decryption) + privateKey crypto.PrivateKey // MITM CA private key + alpn string // ALPN protocol to negotiate during MITM + mitmBypass bypass.Bypass // bypass rules that skip MITM decryption + + limiterRefreshInterval time.Duration // traffic limiter cache refresh interval + limiterCleanupInterval time.Duration // traffic limiter cache cleanup interval } +// parseMetadata reads all configuration values from the metadata map into +// the handler's metadata struct. It applies defaults for readTimeout (15s), +// observerPeriod (5s, min 1s), and proxyAgent ("gost/3.0"). MITM TLS is +// enabled when both mitm.certFile and mitm.keyFile are provided. func (h *httpHandler) parseMetadata(md mdata.Metadata) error { h.md.readTimeout = mdutil.GetDuration(md, "readTimeout") if h.md.readTimeout == 0 { @@ -74,6 +83,8 @@ func (h *httpHandler) parseMetadata(md mdata.Metadata) error { h.md.keepalive = mdutil.GetBool(md, "http.keepalive", "keepalive") h.md.compression = mdutil.GetBool(md, "http.compression", "compression") + // probeResist format: "type:value" (e.g. "code:404", "web:example.com", + // "host:192.168.1.1:80", "file:/var/www/index.html"). if pr := mdutil.GetString(md, "probeResist", "probe_resist"); pr != "" { if ss := strings.SplitN(pr, ":", 2); len(ss) == 2 { h.md.probeResistance = &probeResistance{ @@ -130,8 +141,12 @@ func (h *httpHandler) parseMetadata(md mdata.Metadata) error { return nil } +// probeResistance configures a decoy response that hides the proxy from +// unauthorised clients. When authentication fails, instead of returning +// 407 Proxy-Auth-Required, the handler returns a fake response that makes +// the port appear to run a different service. type probeResistance struct { - Type string - Value string - Knock string + Type string // strategy: "code", "web", "host", or "file" + Value string // strategy-specific parameter (status code, URL, address, path) + Knock string // optional hostname; when set, probe resistance only fires for non-matching hosts } diff --git a/handler/http/metadata_test.go b/handler/http/metadata_test.go new file mode 100644 index 00000000..147510cc --- /dev/null +++ b/handler/http/metadata_test.go @@ -0,0 +1,235 @@ +package http + +import ( + "testing" + "time" +) + +func TestParseMetadata_Defaults(t *testing.T) { + h := &httpHandler{} + if err := h.parseMetadata(testMD(map[string]any{})); err != nil { + t.Fatal(err) + } + if h.md.readTimeout != 15*time.Second { + t.Errorf("got readTimeout %v, want 15s", h.md.readTimeout) + } + if h.md.proxyAgent != defaultProxyAgent { + t.Errorf("got proxyAgent %q, want %q", h.md.proxyAgent, defaultProxyAgent) + } + if h.md.observerPeriod != 5*time.Second { + t.Errorf("got observerPeriod %v, want 5s", h.md.observerPeriod) + } +} + +func TestParseMetadata_ReadTimeout(t *testing.T) { + h := &httpHandler{} + if err := h.parseMetadata(testMD(map[string]any{"readTimeout": "30s"})); err != nil { + t.Fatal(err) + } + if h.md.readTimeout != 30*time.Second { + t.Errorf("got readTimeout %v, want 30s", h.md.readTimeout) + } +} + +func TestParseMetadata_NegativeReadTimeout(t *testing.T) { + h := &httpHandler{} + if err := h.parseMetadata(testMD(map[string]any{"readTimeout": "-1"})); err != nil { + t.Fatal(err) + } + if h.md.readTimeout != 0 { + t.Errorf("got readTimeout %v, want 0 for negative value", h.md.readTimeout) + } +} + +func TestParseMetadata_IdleTimeout(t *testing.T) { + h := &httpHandler{} + if err := h.parseMetadata(testMD(map[string]any{"idleTimeout": "60s"})); err != nil { + t.Fatal(err) + } + if h.md.idleTimeout != 60*time.Second { + t.Errorf("got idleTimeout %v, want 60s", h.md.idleTimeout) + } +} + +func TestParseMetadata_NegativeIdleTimeout(t *testing.T) { + h := &httpHandler{} + if err := h.parseMetadata(testMD(map[string]any{"idleTimeout": "-1"})); err != nil { + t.Fatal(err) + } + if h.md.idleTimeout != 0 { + t.Errorf("got idleTimeout %v, want 0 for negative value", h.md.idleTimeout) + } +} + +func TestParseMetadata_Headers(t *testing.T) { + h := &httpHandler{} + if err := h.parseMetadata(testMD(map[string]any{ + "http.header": map[string]any{ + "X-Custom": "value", + }, + })); err != nil { + t.Fatal(err) + } + if h.md.header.Get("X-Custom") != "value" { + t.Errorf("got header X-Custom=%q, want %q", h.md.header.Get("X-Custom"), "value") + } +} + +func TestParseMetadata_Keepalive(t *testing.T) { + h := &httpHandler{} + if err := h.parseMetadata(testMD(map[string]any{"keepalive": "true"})); err != nil { + t.Fatal(err) + } + if !h.md.keepalive { + t.Error("expected keepalive to be true") + } +} + +func TestParseMetadata_Compression(t *testing.T) { + h := &httpHandler{} + if err := h.parseMetadata(testMD(map[string]any{"compression": "true"})); err != nil { + t.Fatal(err) + } + if !h.md.compression { + t.Error("expected compression to be true") + } +} + +func TestParseMetadata_ProbeResistance(t *testing.T) { + h := &httpHandler{} + if err := h.parseMetadata(testMD(map[string]any{ + "probeResist": "code:404", + "knock": "secret.example.com", + })); err != nil { + t.Fatal(err) + } + if h.md.probeResistance == nil { + t.Fatal("expected probeResistance to be set") + } + if h.md.probeResistance.Type != "code" { + t.Errorf("got type %q, want %q", h.md.probeResistance.Type, "code") + } + if h.md.probeResistance.Value != "404" { + t.Errorf("got value %q, want %q", h.md.probeResistance.Value, "404") + } + if h.md.probeResistance.Knock != "secret.example.com" { + t.Errorf("got knock %q, want %q", h.md.probeResistance.Knock, "secret.example.com") + } +} + +func TestParseMetadata_ProbeResistance_InvalidFormat(t *testing.T) { + h := &httpHandler{} + if err := h.parseMetadata(testMD(map[string]any{"probeResist": "invalid"})); err != nil { + t.Fatal(err) + } + // probeResist without colon is ignored + if h.md.probeResistance != nil { + t.Error("expected nil probeResistance for invalid format") + } +} + +func TestParseMetadata_UDP(t *testing.T) { + h := &httpHandler{} + if err := h.parseMetadata(testMD(map[string]any{ + "udp": "true", + "udpBufferSize": "4096", + })); err != nil { + t.Fatal(err) + } + if !h.md.enableUDP { + t.Error("expected enableUDP to be true") + } + if h.md.udpBufferSize != 4096 { + t.Errorf("got udpBufferSize %d, want 4096", h.md.udpBufferSize) + } +} + +func TestParseMetadata_Hash(t *testing.T) { + h := &httpHandler{} + if err := h.parseMetadata(testMD(map[string]any{"hash": "host"})); err != nil { + t.Fatal(err) + } + if h.md.hash != "host" { + t.Errorf("got hash %q, want %q", h.md.hash, "host") + } +} + +func TestParseMetadata_ObserverPeriod(t *testing.T) { + h := &httpHandler{} + if err := h.parseMetadata(testMD(map[string]any{"observePeriod": "10s"})); err != nil { + t.Fatal(err) + } + if h.md.observerPeriod != 10*time.Second { + t.Errorf("got observerPeriod %v, want 10s", h.md.observerPeriod) + } +} + +func TestParseMetadata_ObserverPeriod_Minimum(t *testing.T) { + h := &httpHandler{} + if err := h.parseMetadata(testMD(map[string]any{"observePeriod": "500ms"})); err != nil { + t.Fatal(err) + } + if h.md.observerPeriod != time.Second { + t.Errorf("got observerPeriod %v, want minimum 1s", h.md.observerPeriod) + } +} + +func TestParseMetadata_Sniffing(t *testing.T) { + h := &httpHandler{} + if err := h.parseMetadata(testMD(map[string]any{ + "sniffing": "true", + "sniffing.timeout": "5s", + "sniffing.websocket": "true", + "sniffing.websocket.sampleRate": "20.0", + })); err != nil { + t.Fatal(err) + } + if !h.md.sniffing { + t.Error("expected sniffing to be true") + } + if h.md.sniffingTimeout != 5*time.Second { + t.Errorf("got sniffingTimeout %v, want 5s", h.md.sniffingTimeout) + } + if !h.md.sniffingWebsocket { + t.Error("expected sniffingWebsocket to be true") + } + if h.md.sniffingWebsocketSampleRate != 20.0 { + t.Errorf("got sampleRate %f, want 20.0", h.md.sniffingWebsocketSampleRate) + } +} + +func TestParseMetadata_ProxyAgent(t *testing.T) { + h := &httpHandler{} + if err := h.parseMetadata(testMD(map[string]any{"proxyAgent": "custom/1.0"})); err != nil { + t.Fatal(err) + } + if h.md.proxyAgent != "custom/1.0" { + t.Errorf("got proxyAgent %q, want %q", h.md.proxyAgent, "custom/1.0") + } +} + +func TestParseMetadata_AuthBasicRealm(t *testing.T) { + h := &httpHandler{} + if err := h.parseMetadata(testMD(map[string]any{"authBasicRealm": "myrealm"})); err != nil { + t.Fatal(err) + } + if h.md.authBasicRealm != "myrealm" { + t.Errorf("got realm %q, want %q", h.md.authBasicRealm, "myrealm") + } +} + +func TestParseMetadata_LimiterIntervals(t *testing.T) { + h := &httpHandler{} + if err := h.parseMetadata(testMD(map[string]any{ + "limiter.refreshInterval": "30s", + "limiter.cleanupInterval": "60s", + })); err != nil { + t.Fatal(err) + } + if h.md.limiterRefreshInterval != 30*time.Second { + t.Errorf("got refreshInterval %v, want 30s", h.md.limiterRefreshInterval) + } + if h.md.limiterCleanupInterval != 60*time.Second { + t.Errorf("got cleanupInterval %v, want 60s", h.md.limiterCleanupInterval) + } +} diff --git a/handler/http/proxy.go b/handler/http/proxy.go new file mode 100644 index 00000000..946f9f81 --- /dev/null +++ b/handler/http/proxy.go @@ -0,0 +1,296 @@ +package http + +import ( + "bufio" + "context" + "errors" + "fmt" + "io" + "net" + "net/http" + "net/http/httputil" + "strings" + "time" + + "github.com/go-gost/core/bypass" + stats "github.com/go-gost/core/observer/stats" + "github.com/go-gost/core/logger" + xbypass "github.com/go-gost/x/bypass" + ictx "github.com/go-gost/x/internal/ctx" + xio "github.com/go-gost/x/internal/io" + xnet "github.com/go-gost/x/internal/net" + xhttp "github.com/go-gost/x/internal/net/http" + "github.com/go-gost/x/internal/util/sniffing" + xstats "github.com/go-gost/x/observer/stats" + stats_wrapper "github.com/go-gost/x/observer/stats/wrapper" + xrecorder "github.com/go-gost/x/recorder" +) + +// handleProxy implements HTTP forward proxy semantics. It wraps the +// connection for stats, performs the first round-trip, and then enters +// a keep-alive loop reading and proxying subsequent requests on the +// same connection until the client or upstream closes. +// +// Each request is processed through proxyRoundTrip, which handles +// authentication header stripping, bypass checks, and body recording. +func (h *httpHandler) handleProxy(ctx context.Context, conn net.Conn, req *http.Request, ro *xrecorder.HandlerRecorderObject, log logger.Logger) error { + pStats := xstats.Stats{} + conn = stats_wrapper.WrapConn(conn, &pStats) + + ro.Time = time.Time{} + + if close, err := h.proxyRoundTrip(ctx, conn, req, ro, &pStats, log); err != nil || close { + return err + } + + // Keep-alive loop: read and proxy subsequent requests. + br := bufio.NewReader(conn) + for { + pStats.Reset() + + req, err := http.ReadRequest(br) + if err != nil { + if errors.Is(err, io.EOF) || errors.Is(err, net.ErrClosed) { + return nil + } + return err + } + + if log.IsLevelEnabled(logger.TraceLevel) { + dump, _ := httputil.DumpRequest(req, false) + log.Trace(string(dump)) + } + + if close, err := h.proxyRoundTrip(ctx, xio.NewReadWriteCloser(br, conn, conn), req, ro, &pStats, log); err != nil || close { + req.Body.Close() + return err + } + req.Body.Close() + } +} + +// proxyRoundTrip performs a single HTTP round-trip for a forward proxy +// request. It: +// +// - Clones the recorder object for per-request isolation. +// - Normalises the host and adds a default port if missing. +// - Adjusts HTTP/1.0 Connection headers for compatibility. +// - Strips proxy-specific headers (Proxy-Authorization, Gost-Target, etc.). +// - Checks bypass rules — forbidden hosts get a 403. +// - Optionally captures the request body for recording. +// - Sends the request through the upstream transport. +// - Optionally captures the response body for recording. +// - Handles 101 Switching Protocols (WebSocket upgrades). +// +// The close return value indicates whether the underlying connection +// should be closed after this round-trip (e.g. HTTP/1.0 without keep-alive, +// or an upstream Connection: close). +func (h *httpHandler) proxyRoundTrip(ctx context.Context, rw io.ReadWriteCloser, req *http.Request, ro *xrecorder.HandlerRecorderObject, pStats stats.Stats, log logger.Logger) (close bool, err error) { + close = true + + // Clone the recorder object so per-request mutations don't leak. + ro2 := &xrecorder.HandlerRecorderObject{} + *ro2 = *ro + ro = ro2 + + host := normalizeHostPort(req.Host, "80") + ro.Host = host + ro.Time = time.Now() + + log = log.WithFields(map[string]any{ + "host": host, + }) + + log.Infof("%s <-> %s", ro.RemoteAddr, req.Host) + defer func() { + if err != nil { + ro.Err = err.Error() + } + ro.InputBytes = pStats.Get(stats.KindInputBytes) + ro.OutputBytes = pStats.Get(stats.KindOutputBytes) + ro.Duration = time.Since(ro.Time) + if err := ro.Record(ctx, h.recorder.Recorder); err != nil { + log.Errorf("record: %v", err) + } + + log.WithFields(map[string]any{ + "duration": time.Since(ro.Time), + "inputBytes": ro.InputBytes, + "outputBytes": ro.OutputBytes, + }).Infof("%s >-< %s", ro.RemoteAddr, req.Host) + }() + + ro.HTTP = &xrecorder.HTTPRecorderObject{ + Host: req.Host, + Proto: req.Proto, + Scheme: req.URL.Scheme, + Method: req.Method, + URI: req.RequestURI, + Request: xrecorder.HTTPRequestRecorderObject{ + ContentLength: req.ContentLength, + Header: req.Header.Clone(), + }, + } + + // HTTP/1.0: ensure Connection: close unless the client explicitly + // requested keep-alive (which we translate by removing the header + // so the default close behaviour applies). + http10 := req.ProtoMajor == 1 && req.ProtoMinor == 0 + if http10 { + if strings.ToLower(req.Header.Get("Connection")) == "keep-alive" { + req.Header.Del("Connection") + } else { + req.Header.Set("Connection", "close") + } + } + + // Strip hop-by-hop proxy headers before forwarding. + req.Header.Del("Proxy-Authorization") + req.Header.Del("Proxy-Connection") + req.Header.Del("Gost-Target") + req.Header.Del("X-Gost-Target") + + res := &http.Response{ + ProtoMajor: req.ProtoMajor, + ProtoMinor: req.ProtoMinor, + Header: http.Header{}, + StatusCode: http.StatusServiceUnavailable, + } + ro.HTTP.StatusCode = res.StatusCode + + if h.options.Bypass != nil && + h.options.Bypass.Contains(ctx, "tcp", host, bypass.WithService(h.options.Service)) { + res.StatusCode = http.StatusForbidden + + if log.IsLevelEnabled(logger.TraceLevel) { + dump, _ := httputil.DumpResponse(res, false) + log.Trace(string(dump)) + } + log.Debug("bypass: ", host) + if werr := res.Write(rw); werr != nil { + log.Error("write bypass response: ", werr) + } + err = xbypass.ErrBypass + return + } + + // Optionally intercept the request body for recording. The original + // body is replaced with a tee reader so the transport still sees it. + var reqBody *xhttp.Body + if opts := h.recorder.Options; opts != nil && opts.HTTPBody { + if req.Body != nil { + bodySize := opts.MaxBodySize + if bodySize <= 0 { + bodySize = sniffing.DefaultBodySize + } + if bodySize > sniffing.MaxBodySize { + bodySize = sniffing.MaxBodySize + } + reqBody = xhttp.NewBody(req.Body, bodySize) + req.Body = reqBody + } + } + + ctx = ictx.ContextWithRecorderObject(ctx, ro) + ctx = ictx.ContextWithLogger(ctx, log) + + resp, err := h.transport.RoundTrip(req.WithContext(ctx)) + + if reqBody != nil { + ro.HTTP.Request.Body = reqBody.Content() + ro.HTTP.Request.ContentLength = reqBody.Length() + } + + if err != nil { + if werr := res.Write(rw); werr != nil { + log.Error("write error response: ", werr) + } + return + } + defer resp.Body.Close() + + ro.HTTP.StatusCode = resp.StatusCode + ro.HTTP.Response.Header = resp.Header.Clone() + ro.HTTP.Response.ContentLength = resp.ContentLength + + if log.IsLevelEnabled(logger.TraceLevel) { + dump, _ := httputil.DumpResponse(resp, false) + log.Trace(string(dump)) + } + + // HTTP/1.0: propagate keep-alive if the upstream supports it, and + // force the response protocol version to 1.0. + if http10 { + if !resp.Close { + resp.Header.Set("Connection", "keep-alive") + } + resp.ProtoMajor = 1 + resp.ProtoMinor = 0 + } + + if resp.StatusCode == http.StatusSwitchingProtocols { + err = h.handleUpgradeResponse(ctx, rw, req, resp, ro, log) + return + } + + // Optionally intercept the response body for recording. + var respBody *xhttp.Body + if opts := h.recorder.Options; opts != nil && opts.HTTPBody { + bodySize := opts.MaxBodySize + if bodySize <= 0 { + bodySize = sniffing.DefaultBodySize + } + if bodySize > sniffing.MaxBodySize { + bodySize = sniffing.MaxBodySize + } + respBody = xhttp.NewBody(resp.Body, bodySize) + resp.Body = respBody + } + + err = resp.Write(rw) + + if respBody != nil { + ro.HTTP.Response.Body = respBody.Content() + ro.HTTP.Response.ContentLength = respBody.Length() + } + + if err != nil { + err = fmt.Errorf("write response: %v", err) + return + } + + close = resp.Close + + return +} + +// handleUpgradeResponse handles HTTP 101 Switching Protocols responses. +// It validates that the request and response upgrade types match, writes +// the 101 response back to the client, and then relays raw bytes between +// the client and the backend connection. +// +// If the upgrade is to "websocket" and WebSocket sniffing is enabled, +// frame-level recording is performed via sniffingWebsocketFrame. +func (h *httpHandler) handleUpgradeResponse(ctx context.Context, rw io.ReadWriteCloser, req *http.Request, res *http.Response, ro *xrecorder.HandlerRecorderObject, log logger.Logger) error { + reqUpType := upgradeType(req.Header) + resUpType := upgradeType(res.Header) + if !strings.EqualFold(reqUpType, resUpType) { + return fmt.Errorf("backend tried to switch protocol %q when %q was requested", resUpType, reqUpType) + } + + backConn, ok := res.Body.(io.ReadWriteCloser) + if !ok { + return fmt.Errorf("internal error: 101 switching protocols response with non-writable body") + } + + res.Body = nil + if err := res.Write(rw); err != nil { + return fmt.Errorf("response write: %v", err) + } + + if reqUpType == "websocket" && h.md.sniffingWebsocket { + return h.sniffingWebsocketFrame(ctx, rw, backConn, ro, log) + } + + return xnet.Pipe(ctx, rw, backConn, xnet.WithReadTimeout(h.md.idleTimeout)) +} diff --git a/handler/http/proxy_test.go b/handler/http/proxy_test.go new file mode 100644 index 00000000..fc618289 --- /dev/null +++ b/handler/http/proxy_test.go @@ -0,0 +1,360 @@ +package http + +import ( + "context" + "io" + "net" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/go-gost/core/bypass" + "github.com/go-gost/core/handler" + xstats "github.com/go-gost/x/observer/stats" + xrecorder "github.com/go-gost/x/recorder" +) + +func TestProxyRoundTrip_BasicGET(t *testing.T) { + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Connection", "close") + w.WriteHeader(http.StatusOK) + })) + defer ts.Close() + + h := &httpHandler{ + options: handler.Options{ + Logger: &testLogger{}, + }, + transport: ts.Client().Transport, + } + h.md.readTimeout = 15 + + req, _ := http.NewRequest("GET", ts.URL, nil) + ro := &xrecorder.HandlerRecorderObject{ + RemoteAddr: "127.0.0.1:12345", + } + pStats := xstats.Stats{} + + rw := &testReadWriteCloser{buf: new(strings.Builder)} + + closeConn, err := h.proxyRoundTrip(context.Background(), rw, req, ro, &pStats, &testLogger{}) + if err != nil { + t.Fatalf("proxyRoundTrip error: %v", err) + } + if !closeConn { + t.Error("expected close=true with Connection: close response") + } +} + +func TestProxyRoundTrip_WithBypass(t *testing.T) { + h := &httpHandler{ + options: handler.Options{ + Logger: &testLogger{}, + Bypass: &testBypass{contains: true}, + }, + } + h.md.proxyAgent = defaultProxyAgent + + req, _ := http.NewRequest("GET", "http://blocked.example.com", nil) + ro := &xrecorder.HandlerRecorderObject{ + RemoteAddr: "127.0.0.1:12345", + } + pStats := xstats.Stats{} + + rw := &testReadWriteCloser{buf: new(strings.Builder)} + + close, err := h.proxyRoundTrip(context.Background(), rw, req, ro, &pStats, &testLogger{}) + if err == nil { + t.Error("expected bypass error") + } + if !close { + t.Error("expected close=true for bypass") + } +} + +func TestProxyRoundTrip_WithoutBypass(t *testing.T) { + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Connection", "close") + w.WriteHeader(http.StatusOK) + })) + defer ts.Close() + + h := &httpHandler{ + options: handler.Options{ + Logger: &testLogger{}, + Bypass: &testBypass{contains: false}, + }, + transport: ts.Client().Transport, + } + h.md.readTimeout = 15 + + req, _ := http.NewRequest("GET", ts.URL, nil) + ro := &xrecorder.HandlerRecorderObject{ + RemoteAddr: "127.0.0.1:12345", + } + pStats := xstats.Stats{} + + rw := &testReadWriteCloser{buf: new(strings.Builder)} + + closeConn, err := h.proxyRoundTrip(context.Background(), rw, req, ro, &pStats, &testLogger{}) + if err != nil { + t.Fatalf("proxyRoundTrip error: %v", err) + } + if !closeConn { + t.Error("expected close=true with Connection: close response") + } +} + +func TestProxyRoundTrip_TransportError(t *testing.T) { + h := &httpHandler{ + options: handler.Options{ + Logger: &testLogger{}, + }, + transport: &testRoundTripper{err: true}, + } + h.md.readTimeout = 15 + + req, _ := http.NewRequest("GET", "http://error.example.com", nil) + ro := &xrecorder.HandlerRecorderObject{ + RemoteAddr: "127.0.0.1:12345", + } + pStats := xstats.Stats{} + + rw := &testReadWriteCloser{buf: new(strings.Builder)} + + close, err := h.proxyRoundTrip(context.Background(), rw, req, ro, &pStats, &testLogger{}) + if err == nil { + t.Error("expected transport error") + } + if !close { + t.Error("expected close=true on transport error") + } +} + +func TestProxyRoundTrip_HTTP10(t *testing.T) { + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) + })) + defer ts.Close() + + h := &httpHandler{ + options: handler.Options{ + Logger: &testLogger{}, + }, + transport: ts.Client().Transport, + } + h.md.readTimeout = 15 + + req, _ := http.NewRequest("GET", ts.URL, nil) + req.ProtoMajor = 1 + req.ProtoMinor = 0 + ro := &xrecorder.HandlerRecorderObject{ + RemoteAddr: "127.0.0.1:12345", + } + pStats := xstats.Stats{} + + rw := &testReadWriteCloser{buf: new(strings.Builder)} + + _, err := h.proxyRoundTrip(context.Background(), rw, req, ro, &pStats, &testLogger{}) + if err != nil { + t.Fatalf("proxyRoundTrip error with HTTP/1.0: %v", err) + } +} + +func TestHandleUpgradeResponse_Mismatch(t *testing.T) { + h := &httpHandler{} + h.md.proxyAgent = defaultProxyAgent + + req, _ := http.NewRequest("GET", "http://example.com", nil) + req.Header.Set("Connection", "Upgrade") + req.Header.Set("Upgrade", "websocket") + + res := &http.Response{ + Header: http.Header{ + "Connection": {"Upgrade"}, + "Upgrade": {"h2c"}, + }, + } + + err := h.handleUpgradeResponse(context.Background(), &testReadWriteCloser{}, req, res, &xrecorder.HandlerRecorderObject{}, &testLogger{}) + if err == nil { + t.Error("expected error for upgrade type mismatch") + } +} + +func TestHandleUpgradeResponse_NonReadWriteCloserBody(t *testing.T) { + h := &httpHandler{} + h.md.proxyAgent = defaultProxyAgent + + req, _ := http.NewRequest("GET", "http://example.com", nil) + req.Header.Set("Connection", "Upgrade") + req.Header.Set("Upgrade", "websocket") + + res := &http.Response{ + Header: http.Header{ + "Connection": {"Upgrade"}, + "Upgrade": {"websocket"}, + }, + Body: io.NopCloser(strings.NewReader("not a rwc")), + } + + err := h.handleUpgradeResponse(context.Background(), &testReadWriteCloser{}, req, res, &xrecorder.HandlerRecorderObject{}, &testLogger{}) + if err == nil { + t.Error("expected error for non-io.ReadWriteCloser body") + } +} + +func TestProxyRoundTrip_HeaderCleanup(t *testing.T) { + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + // Verify headers are cleaned + if r.Header.Get("Proxy-Authorization") != "" { + t.Error("Proxy-Authorization should be removed") + } + if r.Header.Get("Proxy-Connection") != "" { + t.Error("Proxy-Connection should be removed") + } + w.WriteHeader(http.StatusOK) + })) + defer ts.Close() + + h := &httpHandler{ + options: handler.Options{ + Logger: &testLogger{}, + }, + transport: ts.Client().Transport, + } + h.md.readTimeout = 15 + + req, _ := http.NewRequest("GET", ts.URL, nil) + req.Header.Set("Proxy-Authorization", "Basic xyz") + req.Header.Set("Proxy-Connection", "keep-alive") + req.Header.Set("Gost-Target", "hidden") + req.Header.Set("X-Gost-Target", "also-hidden") + + ro := &xrecorder.HandlerRecorderObject{ + RemoteAddr: "127.0.0.1:12345", + } + pStats := xstats.Stats{} + rw := &testReadWriteCloser{buf: new(strings.Builder)} + + _, err := h.proxyRoundTrip(context.Background(), rw, req, ro, &pStats, &testLogger{}) + if err != nil { + t.Fatalf("proxyRoundTrip error: %v", err) + } +} + +// testBypass implements bypass.Bypass for testing. +type testBypass struct { + contains bool +} + +func (b *testBypass) Contains(ctx context.Context, network, addr string, opts ...bypass.Option) bool { + return b.contains +} + +func (b *testBypass) IsWhitelist() bool { return false } + +// testRoundTripper implements http.RoundTripper for testing. +type testRoundTripper struct { + err bool + status int + body string +} + +func (rt *testRoundTripper) RoundTrip(req *http.Request) (*http.Response, error) { + if rt.err { + return nil, &testError{msg: "transport error"} + } + if rt.status == 0 { + rt.status = http.StatusOK + } + return &http.Response{ + StatusCode: rt.status, + Body: io.NopCloser(strings.NewReader(rt.body)), + Header: http.Header{}, + }, nil +} + +func TestHandleProxy_SingleRequest(t *testing.T) { + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Connection", "close") + w.WriteHeader(http.StatusOK) + })) + defer ts.Close() + + h := &httpHandler{ + options: handler.Options{ + Logger: &testLogger{}, + }, + transport: ts.Client().Transport, + } + h.md.readTimeout = 15 + + client, server := net.Pipe() + defer client.Close() + defer server.Close() + + req, _ := http.NewRequest("GET", ts.URL, nil) + ro := &xrecorder.HandlerRecorderObject{ + RemoteAddr: "127.0.0.1:12345", + } + + // Drain response in a goroutine + go func() { + io.ReadAll(server) + }() + + err := h.handleProxy(context.Background(), client, req, ro, &testLogger{}) + if err != nil { + t.Logf("handleProxy returned: %v", err) + } +} + +func TestProxyRoundTrip_SwitchingProtocols(t *testing.T) { + h := &httpHandler{ + options: handler.Options{ + Logger: &testLogger{}, + }, + transport: &testRoundTripper{status: http.StatusSwitchingProtocols, body: ""}, + } + h.md.readTimeout = 15 + + req, _ := http.NewRequest("GET", "http://example.com", nil) + req.Header.Set("Connection", "Upgrade") + req.Header.Set("Upgrade", "websocket") + + ro := &xrecorder.HandlerRecorderObject{ + RemoteAddr: "127.0.0.1:12345", + } + pStats := xstats.Stats{} + rw := &testReadWriteCloser{buf: new(strings.Builder)} + + _, err := h.proxyRoundTrip(context.Background(), rw, req, ro, &pStats, &testLogger{}) + // Switching protocols with a non-RWC body should error + if err == nil { + t.Log("expected error for switching protocols with non-RWC body") + } +} + +type testError struct { + msg string +} + +func (e *testError) Error() string { return e.msg } +func (e *testError) Timeout() bool { return false } +func (e *testError) Temporary() bool { return true } + +// testReadWriteCloser implements io.ReadWriteCloser for testing. +type testReadWriteCloser struct { + buf *strings.Builder +} + +func (rw *testReadWriteCloser) Read(p []byte) (int, error) { return 0, io.EOF } +func (rw *testReadWriteCloser) Write(p []byte) (int, error) { + if rw.buf != nil { + return rw.buf.Write(p) + } + return len(p), nil +} +func (rw *testReadWriteCloser) Close() error { return nil } diff --git a/handler/http/udp.go b/handler/http/udp.go index fbf522a5..5183c0ac 100644 --- a/handler/http/udp.go +++ b/handler/http/udp.go @@ -16,6 +16,13 @@ import ( xrecorder "github.com/go-gost/x/recorder" ) +// handleUDP implements UDP over HTTP (UDP relay). When the client sets the +// X-Gost-Protocol header to "udp", the handler responds with 200 OK and +// establishes a UDP association through the proxy chain. Client data sent +// over the HTTP connection is wrapped as SOCKS5 UDP packets and relayed +// through the UDP tunnel. +// +// If UDP relay is disabled (enableUDP=false), a 403 Forbidden is returned. func (h *httpHandler) handleUDP(ctx context.Context, conn net.Conn, ro *xrecorder.HandlerRecorderObject, log logger.Logger) error { log = log.WithFields(map[string]any{ "cmd": "udp", @@ -53,9 +60,14 @@ func (h *httpHandler) handleUDP(ctx context.Context, conn net.Conn, ro *xrecorde return err } - // obtain a udp connection + // Dial a UDP association through the proxy chain router. The empty + // address signals that the router should create a UDP socket rather + // than connect to a specific target. + if h.options.Router == nil { + return errors.New("nil router") + } var buf bytes.Buffer - c, err := h.options.Router.Dial(ictx.ContextWithBuffer(ctx, &buf), "udp", "") // UDP association + c, err := h.options.Router.Dial(ictx.ContextWithBuffer(ctx, &buf), "udp", "") ro.Route = buf.String() if err != nil { log.Error(err) @@ -73,6 +85,8 @@ func (h *httpHandler) handleUDP(ctx context.Context, conn net.Conn, ro *xrecorde return err } + // Wrap the HTTP connection as a SOCKS5 UDP tunnel server conn so + // that the relay can read/write SOCKS5-encapsulated UDP datagrams. relay := udp.NewRelay(socks.UDPTunServerConn(conn), pc). WithService(h.options.Service). WithBypass(h.options.Bypass). diff --git a/handler/http/udp_test.go b/handler/http/udp_test.go new file mode 100644 index 00000000..4ad26285 --- /dev/null +++ b/handler/http/udp_test.go @@ -0,0 +1,78 @@ +package http + +import ( + "bufio" + "context" + "net" + "net/http" + "testing" + + "github.com/go-gost/core/handler" + xrecorder "github.com/go-gost/x/recorder" +) + +func TestHandleUDP_Disabled_WritesForbidden(t *testing.T) { + h := &httpHandler{ + options: handler.Options{ + Logger: &testLogger{}, + }, + } + h.md.proxyAgent = defaultProxyAgent + h.md.enableUDP = false + + client, server := net.Pipe() + defer client.Close() + defer server.Close() + + ro := &xrecorder.HandlerRecorderObject{} + + go func() { + h.handleUDP(context.Background(), server, ro, &testLogger{}) + }() + + br := bufio.NewReader(client) + resp, err := http.ReadResponse(br, nil) + if err != nil { + t.Fatalf("read response: %v", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusForbidden { + t.Errorf("got status %d, want %d", resp.StatusCode, http.StatusForbidden) + } +} + +func TestHandleUDP_Enabled_NilRouter(t *testing.T) { + h := &httpHandler{ + options: handler.Options{ + Logger: &testLogger{}, + }, + } + h.md.proxyAgent = defaultProxyAgent + h.md.enableUDP = true + + client, server := net.Pipe() + defer client.Close() + defer server.Close() + + ro := &xrecorder.HandlerRecorderObject{} + + go func() { + // OK response sent, then nil router causes error + err := h.handleUDP(context.Background(), server, ro, &testLogger{}) + if err == nil { + t.Log("expected error with nil router for UDP") + } + }() + + br := bufio.NewReader(client) + resp, err := http.ReadResponse(br, nil) + if err != nil { + t.Fatalf("read response: %v", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + t.Errorf("got status %d, want %d", resp.StatusCode, http.StatusOK) + } +} diff --git a/handler/http/util.go b/handler/http/util.go new file mode 100644 index 00000000..d7ce31e2 --- /dev/null +++ b/handler/http/util.go @@ -0,0 +1,86 @@ +package http + +import ( + "encoding/base64" + "encoding/binary" + "errors" + "hash/crc32" + "net" + "net/http" + "strings" + + "golang.org/x/net/http/httpguts" +) + +// decodeServerName decodes a GOST v2 Gost-Target / X-Gost-Target header +// value. The encoding is: base64(CRC32(hostname) + base64(hostname)). +// It verifies the CRC32 checksum before returning the decoded hostname. +func decodeServerName(s string) (string, error) { + b, err := base64.RawURLEncoding.DecodeString(s) + if err != nil { + return "", err + } + if len(b) < 4 { + return "", errors.New("invalid name") + } + v, err := base64.RawURLEncoding.DecodeString(string(b[4:])) + if err != nil { + return "", err + } + if crc32.ChecksumIEEE(v) != binary.BigEndian.Uint32(b[:4]) { + return "", errors.New("invalid name") + } + return string(v), nil +} + +// basicProxyAuth extracts the username and password from an HTTP Basic +// authentication Proxy-Authorization header value. It returns ok=false +// if the value is empty, not a Basic scheme, or not valid base64. +func basicProxyAuth(proxyAuth string) (username, password string, ok bool) { + if proxyAuth == "" { + return + } + + if !strings.HasPrefix(proxyAuth, "Basic ") { + return + } + c, err := base64.StdEncoding.DecodeString(strings.TrimPrefix(proxyAuth, "Basic ")) + if err != nil { + return + } + cs := string(c) + username, password, ok = strings.Cut(cs, ":") + if !ok { + return + } + return username, password, true +} + +// upgradeType returns the upgrade protocol token from an HTTP header. +// It returns "" if the Connection header does not include an "Upgrade" +// token or if the Upgrade header is absent. +func upgradeType(h http.Header) string { + if !httpguts.HeaderValuesContainsToken(h["Connection"], "Upgrade") { + return "" + } + return h.Get("Upgrade") +} + +// normalizeHostPort ensures a host string includes a port. If the host +// already contains a port, it is returned unchanged. Otherwise the +// defaultPort is appended. IPv6 addresses are correctly handled: bare +// addresses get bracketed by net.JoinHostPort, and bracketed addresses +// are stripped of brackets before re-joining. +func normalizeHostPort(host, defaultPort string) string { + if _, port, _ := net.SplitHostPort(host); port == "" { + return net.JoinHostPort(strings.Trim(host, "[]"), defaultPort) + } + return host +} + +// buildConnectResponse returns the raw bytes of an HTTP 200 Connection +// established response with the given Proxy-Agent header. +func buildConnectResponse(proxyAgent string) []byte { + return []byte("HTTP/1.1 200 Connection established\r\n" + + "Proxy-Agent: " + proxyAgent + "\r\n\r\n") +} diff --git a/handler/http/util_test.go b/handler/http/util_test.go new file mode 100644 index 00000000..ee751c4c --- /dev/null +++ b/handler/http/util_test.go @@ -0,0 +1,341 @@ +package http + +import ( + "encoding/base64" + "encoding/binary" + "hash/crc32" + "net/http" + "testing" +) + +func TestDecodeServerName(t *testing.T) { + // Helper: encode a hostname using the GOST v2 encoding scheme + encodeName := func(name string) string { + v := []byte(name) + b := make([]byte, 4) + binary.BigEndian.PutUint32(b, crc32.ChecksumIEEE(v)) + inner := base64.RawURLEncoding.EncodeToString(v) + return base64.RawURLEncoding.EncodeToString(append(b, []byte(inner)...)) + } + + tests := []struct { + name string + input string + want string + wantErr bool + }{ + { + name: "valid hostname", + input: encodeName("example.com:8080"), + want: "example.com:8080", + }, + { + name: "valid IP address", + input: encodeName("1.2.3.4:443"), + want: "1.2.3.4:443", + }, + { + name: "empty string", + input: "", + wantErr: true, + }, + { + name: "invalid base64", + input: "!!!not-valid-base64!!!", + wantErr: true, + }, + { + name: "too short (less than 4 bytes)", + input: base64.RawURLEncoding.EncodeToString([]byte("ab")), + wantErr: true, + }, + { + name: "valid base64 but inner is garbage", + input: base64.RawURLEncoding.EncodeToString([]byte("0000!!!!")), + wantErr: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got, err := decodeServerName(tt.input) + if tt.wantErr { + if err == nil { + t.Errorf("decodeServerName(%q) = (%q, nil), want error", tt.input, got) + } + return + } + if err != nil { + t.Errorf("decodeServerName(%q) error = %v", tt.input, err) + return + } + if got != tt.want { + t.Errorf("decodeServerName(%q) = %q, want %q", tt.input, got, tt.want) + } + }) + } +} + +func TestDecodeServerName_CRC32Mismatch(t *testing.T) { + name := "test.example.com" + v := []byte(name) + // Use wrong CRC32 + badCRC := crc32.ChecksumIEEE(v) ^ 0xFFFFFFFF + b := make([]byte, 4) + binary.BigEndian.PutUint32(b, badCRC) + inner := base64.RawURLEncoding.EncodeToString(v) + s := base64.RawURLEncoding.EncodeToString(append(b, []byte(inner)...)) + + _, err := decodeServerName(s) + if err == nil { + t.Error("expected error for CRC32 mismatch") + } +} + +func TestBasicProxyAuth(t *testing.T) { + encode := func(u, p string) string { + return "Basic " + base64.StdEncoding.EncodeToString([]byte(u+":"+p)) + } + + tests := []struct { + name string + proxyAuth string + wantUser string + wantPass string + wantOK bool + }{ + { + name: "valid basic auth", + proxyAuth: encode("admin", "secret"), + wantUser: "admin", + wantPass: "secret", + wantOK: true, + }, + { + name: "username only, no password", + proxyAuth: "Basic " + base64.StdEncoding.EncodeToString([]byte("user:")), + wantUser: "user", + wantPass: "", + wantOK: true, + }, + { + name: "password only, no username", + proxyAuth: "Basic " + base64.StdEncoding.EncodeToString([]byte(":pass")), + wantUser: "", + wantPass: "pass", + wantOK: true, + }, + { + name: "password contains colon", + proxyAuth: "Basic " + base64.StdEncoding.EncodeToString([]byte("user:a:b:c")), + wantUser: "user", + wantPass: "a:b:c", + wantOK: true, + }, + { + name: "empty string", + proxyAuth: "", + }, + { + name: "not Basic scheme", + proxyAuth: "Bearer token123", + }, + { + name: "invalid base64", + proxyAuth: "Basic not-valid-base64!", + }, + { + name: "no colon separator", + proxyAuth: "Basic " + base64.StdEncoding.EncodeToString([]byte("username")), + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + user, pass, ok := basicProxyAuth(tt.proxyAuth) + if ok != tt.wantOK { + t.Errorf("basicProxyAuth(%q) ok = %v, want %v", tt.proxyAuth, ok, tt.wantOK) + return + } + if ok { + if user != tt.wantUser { + t.Errorf("basicProxyAuth(%q) user = %q, want %q", tt.proxyAuth, user, tt.wantUser) + } + if pass != tt.wantPass { + t.Errorf("basicProxyAuth(%q) pass = %q, want %q", tt.proxyAuth, pass, tt.wantPass) + } + } + }) + } +} + +func TestUpgradeType(t *testing.T) { + tests := []struct { + name string + header http.Header + want string + }{ + { + name: "nil header", + header: nil, + want: "", + }, + { + name: "websocket upgrade", + header: http.Header{ + "Connection": {"Upgrade"}, + "Upgrade": {"websocket"}, + }, + want: "websocket", + }, + { + name: "http2 upgrade", + header: http.Header{ + "Connection": {"Upgrade"}, + "Upgrade": {"h2c"}, + }, + want: "h2c", + }, + { + name: "no connection upgrade header", + header: http.Header{ + "Upgrade": {"websocket"}, + }, + want: "", + }, + { + name: "empty header", + header: http.Header{}, + want: "", + }, + { + name: "connection has upgrade but no upgrade header", + header: http.Header{ + "Connection": {"Upgrade"}, + }, + want: "", + }, + { + name: "connection upgrade in multi-value", + header: http.Header{ + "Connection": {"keep-alive", "Upgrade"}, + "Upgrade": {"websocket"}, + }, + want: "websocket", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := upgradeType(tt.header); got != tt.want { + t.Errorf("upgradeType() = %q, want %q", got, tt.want) + } + }) + } +} + +func TestNormalizeHostPort(t *testing.T) { + tests := []struct { + name string + host string + defaultPort string + want string + }{ + { + name: "hostname without port", + host: "example.com", + defaultPort: "80", + want: "example.com:80", + }, + { + name: "hostname with port", + host: "example.com:8080", + defaultPort: "80", + want: "example.com:8080", + }, + { + name: "IPv4 without port", + host: "1.2.3.4", + defaultPort: "80", + want: "1.2.3.4:80", + }, + { + name: "IPv4 with port", + host: "1.2.3.4:443", + defaultPort: "80", + want: "1.2.3.4:443", + }, + { + name: "IPv6 without port (bracketed)", + host: "[::1]", + defaultPort: "80", + want: "[::1]:80", + }, + { + name: "IPv6 with port", + host: "[::1]:443", + defaultPort: "80", + want: "[::1]:443", + }, + { + name: "empty string", + host: "", + defaultPort: "80", + want: ":80", + }, + { + name: "custom default port", + host: "example.com", + defaultPort: "443", + want: "example.com:443", + }, + { + name: "bare IPv6 without brackets or port", + host: "::1", + defaultPort: "80", + want: "[::1]:80", // net.JoinHostPort wraps IPv6 in brackets + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := normalizeHostPort(tt.host, tt.defaultPort); got != tt.want { + t.Errorf("normalizeHostPort(%q, %q) = %q, want %q", tt.host, tt.defaultPort, got, tt.want) + } + }) + } +} + +func TestBuildConnectResponse(t *testing.T) { + got := buildConnectResponse("gost/3.0") + if len(got) == 0 { + t.Error("buildConnectResponse returned empty") + } + s := string(got) + if !contains(s, "200") { + t.Errorf("expected 200 status in response: %s", s) + } + if !contains(s, "Connection established") { + t.Errorf("expected Connection established in response: %s", s) + } + if !contains(s, "gost/3.0") { + t.Errorf("expected proxy agent in response: %s", s) + } +} + +func TestBuildConnectResponse_CustomAgent(t *testing.T) { + got := buildConnectResponse("custom-agent/1.0") + s := string(got) + if !contains(s, "custom-agent/1.0") { + t.Errorf("expected custom agent in response: %s", s) + } +} + +func contains(s, substr string) bool { + for i := 0; i <= len(s)-len(substr); i++ { + if s[i:i+len(substr)] == substr { + return true + } + } + return false +} diff --git a/handler/http/websocket.go b/handler/http/websocket.go new file mode 100644 index 00000000..e0979748 --- /dev/null +++ b/handler/http/websocket.go @@ -0,0 +1,138 @@ +package http + +import ( + "bytes" + "context" + "io" + "math" + "time" + + "github.com/go-gost/core/logger" + "github.com/go-gost/x/internal/util/sniffing" + ws_util "github.com/go-gost/x/internal/util/ws" + xrecorder "github.com/go-gost/x/recorder" + "golang.org/x/time/rate" +) + +// sniffingWebsocketFrame relays WebSocket frames between the client (rw) +// and the backend (cc) while sampling frames for recording. Two goroutines +// run in parallel — one per direction — copying frames and recording them +// at the configured sample rate. +// +// The first direction to encounter an error terminates the relay. +func (h *httpHandler) sniffingWebsocketFrame(ctx context.Context, rw, cc io.ReadWriter, ro *xrecorder.HandlerRecorderObject, log logger.Logger) error { + errc := make(chan error, 1) + + sampleRate := h.md.sniffingWebsocketSampleRate + if sampleRate == 0 { + sampleRate = sniffing.DefaultSampleRate + } + if sampleRate < 0 { + sampleRate = math.MaxFloat64 + } + + go h.copyWebsocketDirection(ctx, cc, rw, "client", ro, sampleRate, log, errc) + go h.copyWebsocketDirection(ctx, rw, cc, "server", ro, sampleRate, log, errc) + + <-errc + return nil +} + +// copyWebsocketDirection copies WebSocket frames from r to w in a loop +// until an error occurs. Each frame is recorded (subject to the rate +// limiter) with directional metadata — "client" for frames originating +// from the client, "server" for frames from the backend. +// +// The ro parameter is cloned at the start so per-frame mutations are +// isolated from the caller. +func (h *httpHandler) copyWebsocketDirection(ctx context.Context, r io.Reader, w io.Writer, from string, ro *xrecorder.HandlerRecorderObject, sampleRate float64, log logger.Logger, errc chan<- error) { + if ro == nil { + ro = &xrecorder.HandlerRecorderObject{} + } + ro2 := &xrecorder.HandlerRecorderObject{} + *ro2 = *ro + r2 := ro2 + + limiter := rate.NewLimiter(rate.Limit(sampleRate), int(sampleRate)) + + buf := &bytes.Buffer{} + for { + start := time.Now() + + if err := h.copyWebsocketFrame(w, r, buf, from, r2); err != nil { + errc <- err + return + } + + if limiter.Allow() { + r2.Duration = time.Since(start) + r2.Time = time.Now() + if err := r2.Record(ctx, h.recorder.Recorder); err != nil { + log.Errorf("record: %v", err) + } + } + } +} + +// copyWebsocketFrame reads a single WebSocket frame from r, records its +// metadata and optionally its payload, then writes it to w. +// +// The payload is tee'd: it is copied to buf for recording (up to the +// configured body size limit) and then forwarded by reconstructing the +// frame with a MultiReader that replays the captured bytes followed by +// the remaining data. +// +// Directional byte counts are set on the recorder object — InputBytes +// for client frames, OutputBytes for server frames. +func (h *httpHandler) 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 opts := h.recorder.Options; opts != nil && opts.HTTPBody { + bodySize := opts.MaxBodySize + if bodySize <= 0 { + bodySize = sniffing.DefaultBodySize + } + if bodySize > sniffing.MaxBodySize { + bodySize = sniffing.MaxBodySize + } + + 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 + } + + // Reconstruct the frame by prepending any captured payload bytes + // before the (possibly truncated) original data stream. + fr.Data = io.MultiReader(bytes.NewReader(buf.Bytes()), fr.Data) + if _, err := fr.WriteTo(w); err != nil { + return err + } + + return nil +} diff --git a/handler/http/websocket_test.go b/handler/http/websocket_test.go new file mode 100644 index 00000000..d7e1d91c --- /dev/null +++ b/handler/http/websocket_test.go @@ -0,0 +1,50 @@ +package http + +import ( + "bytes" + "context" + "io" + "testing" + + "github.com/go-gost/core/handler" + "github.com/go-gost/core/recorder" +) + +func TestCopyWebsocketFrame_ReadError(t *testing.T) { + h := &httpHandler{ + options: handler.Options{ + Logger: &testLogger{}, + }, + } + + buf := &bytes.Buffer{} + err := h.copyWebsocketFrame(io.Discard, bytes.NewReader(nil), buf, "client", nil) + if err == nil { + t.Error("expected read error from empty reader") + } +} + +func TestCopyWebsocketDirection_ErrorReader(t *testing.T) { + h := &httpHandler{ + options: handler.Options{ + Logger: &testLogger{}, + }, + } + h.recorder = recorder.RecorderObject{} + + errc := make(chan error, 1) + + go h.copyWebsocketDirection(context.Background(), &errorReader{}, io.Discard, "client", nil, 10.0, &testLogger{}, errc) + + err := <-errc + if err == nil { + t.Error("expected error from error reader") + } +} + +// errorReader always returns an error. +type errorReader struct{} + +func (r *errorReader) Read(p []byte) (int, error) { return 0, io.ErrUnexpectedEOF } +func (r *errorReader) Write(p []byte) (int, error) { return len(p), nil } +func (r *errorReader) Close() error { return nil }