fix(handler/http2): apply traffic limiter + stats to forward proxy path, decouple auth I/O, add idle timeout

- Wrap non-CONNECT upstream with traffic_wrapper and stats_wrapper (was only on CONNECT path)
- Write 503 error response on forwardRequest failure instead of silent connection close
- Decouple resp.Header from w.Header() to prevent metadata header leakage/doubling in 407 responses
- Return pipeTo signal from authenticate instead of dialing inline (matches http handler pattern)
- Add idleTimeout metadata field and pass to xnet.Pipe for CONNECT tunnels
- Add 4 tests for probe resistance host forwarding and knock bypass
This commit is contained in:
ginuerzh
2026-05-30 20:06:56 +08:00
parent 6ccaae0573
commit e543c0b0fd
6 changed files with 144 additions and 30 deletions
+3 -14
View File
@@ -4,7 +4,6 @@ import (
"context"
"encoding/base64"
"fmt"
"net"
"net/http"
"net/http/httputil"
"os"
@@ -36,10 +35,10 @@ func (h *http2Handler) basicProxyAuth(proxyAuth string) (username, password stri
return cs[:s], cs[s+1:], true
}
func (h *http2Handler) authenticate(ctx context.Context, w http.ResponseWriter, r *http.Request, resp *http.Response, log logger.Logger) (id string, ok bool) {
func (h *http2Handler) authenticate(ctx context.Context, w http.ResponseWriter, r *http.Request, resp *http.Response, log logger.Logger) (id string, ok bool, pipeTo string) {
u, p, _ := h.basicProxyAuth(r.Header.Get("Proxy-Authorization"))
if h.options.Auther == nil {
return "", true
return "", true, ""
}
if id, ok = h.options.Auther.Authenticate(ctx, u, p, auth.WithService(h.options.Service)); ok {
return
@@ -70,17 +69,7 @@ func (h *http2Handler) authenticate(ctx context.Context, w http.ResponseWriter,
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()
if err := h.forwardRequest(w, r, cc); err != nil {
log.Error(err)
}
return
return "", false, pr.Value
case "file":
f, _ := os.Open(pr.Value)
if f != nil {
+43 -6
View File
@@ -86,7 +86,7 @@ func TestAuthenticate(t *testing.T) {
w := httptest.NewRecorder()
resp := &http.Response{Header: w.Header(), Body: io.NopCloser(strings.NewReader(""))}
id, ok := h.authenticate(context.Background(), w, req, resp, &testLogger{})
id, ok, _ := h.authenticate(context.Background(), w, req, resp, &testLogger{})
if !ok {
t.Error("expected ok without auther")
}
@@ -102,7 +102,7 @@ func TestAuthenticate(t *testing.T) {
w := httptest.NewRecorder()
resp := &http.Response{Header: w.Header(), Body: io.NopCloser(strings.NewReader(""))}
id, ok := h.authenticate(context.Background(), w, req, resp, &testLogger{})
id, ok, _ := h.authenticate(context.Background(), w, req, resp, &testLogger{})
if !ok {
t.Error("expected ok with valid auth")
}
@@ -117,7 +117,7 @@ func TestAuthenticate(t *testing.T) {
w := httptest.NewRecorder()
resp := &http.Response{Header: w.Header(), Body: io.NopCloser(strings.NewReader(""))}
_, ok := h.authenticate(context.Background(), w, req, resp, &testLogger{})
_, ok, _ := h.authenticate(context.Background(), w, req, resp, &testLogger{})
if ok {
t.Error("expected !ok with failed auth")
}
@@ -138,7 +138,7 @@ func TestAuthenticate_ProbeResistanceCode(t *testing.T) {
w := httptest.NewRecorder()
resp := &http.Response{Header: http.Header{}, Body: io.NopCloser(strings.NewReader(""))}
_, ok := h.authenticate(context.Background(), w, req, resp, &testLogger{})
_, ok, _ := h.authenticate(context.Background(), w, req, resp, &testLogger{})
if ok {
t.Error("expected !ok")
}
@@ -155,7 +155,7 @@ func TestAuthenticate_ProbeResistanceKnockBypass(t *testing.T) {
w := httptest.NewRecorder()
resp := &http.Response{Header: http.Header{}, Body: io.NopCloser(strings.NewReader(""))}
_, ok := h.authenticate(context.Background(), w, req, resp, &testLogger{})
_, ok, _ := h.authenticate(context.Background(), w, req, resp, &testLogger{})
if ok {
t.Error("expected !ok")
}
@@ -164,6 +164,43 @@ func TestAuthenticate_ProbeResistanceKnockBypass(t *testing.T) {
}
}
func TestAuthenticate_ProbeResistanceHost(t *testing.T) {
h := newTestHandler(handler.AutherOption(&testAuther{ok: false}))
h.md.probeResistance = &probeResistance{Type: "host", Value: "127.0.0.1:9999"}
req := httptest.NewRequest(http.MethodGet, "http://example.com/", nil)
w := httptest.NewRecorder()
resp := &http.Response{Header: http.Header{}, Body: io.NopCloser(strings.NewReader(""))}
_, ok, pipeTo := h.authenticate(context.Background(), w, req, resp, &testLogger{})
if ok {
t.Error("expected !ok")
}
if pipeTo != "127.0.0.1:9999" {
t.Errorf("pipeTo = %q, want %q", pipeTo, "127.0.0.1:9999")
}
}
func TestAuthenticate_ProbeResistanceHostKnockMatch(t *testing.T) {
h := newTestHandler(handler.AutherOption(&testAuther{ok: false}))
h.md.probeResistance = &probeResistance{Type: "host", Value: "127.0.0.1:9999", Knock: "example.com"}
req := httptest.NewRequest(http.MethodGet, "http://example.com/", nil)
w := httptest.NewRecorder()
resp := &http.Response{Header: http.Header{}, Body: io.NopCloser(strings.NewReader(""))}
_, ok, pipeTo := h.authenticate(context.Background(), w, req, resp, &testLogger{})
if ok {
t.Error("expected !ok")
}
if pipeTo != "" {
t.Errorf("pipeTo = %q, want empty (knock matched, probe resistance bypassed)", pipeTo)
}
if resp.StatusCode != http.StatusProxyAuthRequired {
t.Errorf("status = %d, want %d", resp.StatusCode, http.StatusProxyAuthRequired)
}
}
func TestAuthenticate_CustomRealm(t *testing.T) {
h := newTestHandler(handler.AutherOption(&testAuther{ok: false}))
h.md.authBasicRealm = "testrealm"
@@ -172,7 +209,7 @@ func TestAuthenticate_CustomRealm(t *testing.T) {
w := httptest.NewRecorder()
resp := &http.Response{Header: w.Header(), Body: io.NopCloser(strings.NewReader(""))}
_, ok := h.authenticate(context.Background(), w, req, resp, &testLogger{})
_, ok, _ := h.authenticate(context.Background(), w, req, resp, &testLogger{})
if ok {
t.Error("expected !ok")
}
+3 -2
View File
@@ -17,9 +17,10 @@
// ├─ decodeServerName (GOST 2.x compatibility headers)
// ├─ authenticate (basic proxy auth + probe resistance)
// ├─ bypass check
// ├─ set metadata response headers
// ├─ Router.Dial (connect to upstream)
// ├─ forwardRequest (non-CONNECT: proxy a single request)
// └─ CONNECT tunnel (bi-directional pipe)
// ├─ forwardRequest (non-CONNECT: wrap with traffic limiter + stats, proxy request + error response on failure)
// └─ CONNECT tunnel (bi-directional pipe with idle timeout)
package http2
import (
+3
View File
@@ -20,6 +20,7 @@ type metadata struct {
authBasicRealm string
observerPeriod time.Duration
observerResetTraffic bool
idleTimeout time.Duration
limiterRefreshInterval time.Duration
limiterCleanupInterval time.Duration
@@ -55,6 +56,8 @@ func (h *http2Handler) parseMetadata(md mdata.Metadata) {
}
h.md.observerResetTraffic = mdutil.GetBool(md, "observer.resetTraffic")
h.md.idleTimeout = mdutil.GetDuration(md, "readTimeout", "read.timeout")
h.md.limiterRefreshInterval = mdutil.GetDuration(md, "limiter.refreshInterval")
h.md.limiterCleanupInterval = mdutil.GetDuration(md, "limiter.cleanupInterval")
}
+45 -8
View File
@@ -81,14 +81,10 @@ func (h *http2Handler) roundTrip(ctx context.Context, w http.ResponseWriter, req
}
log.Debugf("%s >> %s", req.RemoteAddr, host)
for k := range h.md.header {
w.Header().Set(k, h.md.header.Get(k))
}
resp := &http.Response{
ProtoMajor: 2,
ProtoMinor: 0,
Header: w.Header(),
Header: http.Header{},
Body: io.NopCloser(bytes.NewReader([]byte{})),
}
@@ -108,8 +104,20 @@ func (h *http2Handler) roundTrip(ctx context.Context, w http.ResponseWriter, req
ro.HTTP.Response.Header = resp.Header
}()
clientID, ok := h.authenticate(ctx, w, req, resp, log)
clientID, ok, pipeTo := h.authenticate(ctx, w, req, resp, log)
if !ok {
if pipeTo != "" {
cc, err := net.Dial("tcp", pipeTo)
if err != nil {
log.Error(err)
resp.StatusCode = http.StatusServiceUnavailable
h.writeResponse(w, resp)
return ErrAuthFailed
}
defer cc.Close()
h.forwardRequest(w, req, cc)
return ErrAuthFailed
}
return ErrAuthFailed
}
@@ -131,6 +139,10 @@ func (h *http2Handler) roundTrip(ctx context.Context, w http.ResponseWriter, req
req.Header.Del("Gost-Target")
req.Header.Del("X-Gost-Target")
for k := range h.md.header {
w.Header().Set(k, h.md.header.Get(k))
}
switch h.md.hash {
case "host":
ctx = xctx.ContextWithHash(ctx, &xctx.Hash{Source: host})
@@ -152,9 +164,34 @@ func (h *http2Handler) roundTrip(ctx context.Context, w http.ResponseWriter, req
ro.DstAddr = cc.RemoteAddr().String()
if req.Method != http.MethodConnect {
rw := traffic_wrapper.WrapReadWriter(
h.limiter,
cc,
clientID,
limiter.ScopeOption(limiter.ScopeClient),
limiter.ServiceOption(h.options.Service),
limiter.NetworkOption("tcp"),
limiter.AddrOption(host),
limiter.ClientOption(clientID),
limiter.SrcOption(req.RemoteAddr),
)
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)
}
start := time.Now()
log.Infof("%s <-> %s", req.RemoteAddr, host)
err = h.forwardRequest(w, req, cc)
err = h.forwardRequest(w, req, rw)
if err != nil {
resp.StatusCode = http.StatusServiceUnavailable
if werr := h.writeResponse(w, resp); werr != nil {
log.Error("write error response: ", werr)
}
}
log.WithFields(map[string]any{
"duration": time.Since(start),
}).Infof("%s >-< %s", req.RemoteAddr, host)
@@ -207,7 +244,7 @@ func (h *http2Handler) roundTrip(ctx context.Context, w http.ResponseWriter, req
start := time.Now()
log.Infof("%s <-> %s", req.RemoteAddr, host)
// xnet.Transport(rw, cc)
xnet.Pipe(ctx, xio.NewReadWriteCloser(rw, rw, req.Body), cc)
xnet.Pipe(ctx, xio.NewReadWriteCloser(rw, rw, req.Body), cc, xnet.WithReadTimeout(h.md.idleTimeout))
log.WithFields(map[string]any{
"duration": time.Since(start),
}).Infof("%s >-< %s", req.RemoteAddr, host)
+47
View File
@@ -156,3 +156,50 @@ func TestRoundTrip_AuthFailed(t *testing.T) {
t.Errorf("err = %v, want ErrAuthFailed", err)
}
}
func TestRoundTrip_ProbeResistanceHost(t *testing.T) {
decoy := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.WriteHeader(http.StatusOK)
w.Write([]byte("decoy response"))
}))
defer decoy.Close()
h := newTestHandler(handler.AutherOption(&testAuther{ok: false}))
decoyAddr := strings.TrimPrefix(decoy.URL, "http://")
h.md.probeResistance = &probeResistance{Type: "host", Value: decoyAddr}
h.Init(testMD(map[string]any{}))
req := httptest.NewRequest(http.MethodGet, "http://example.com/", nil)
w := httptest.NewRecorder()
ro := &xrecorder.HandlerRecorderObject{}
err := h.roundTrip(context.Background(), w, req, ro, &testLogger{})
if err != ErrAuthFailed {
t.Errorf("err = %v, want ErrAuthFailed", err)
}
if w.Code != http.StatusOK {
t.Errorf("decoy status = %d, want %d", w.Code, http.StatusOK)
}
if w.Body.String() != "decoy response" {
t.Errorf("decoy body = %q, want %q", w.Body.String(), "decoy response")
}
}
func TestRoundTrip_ProbeResistanceHostDialError(t *testing.T) {
h := newTestHandler(handler.AutherOption(&testAuther{ok: false}))
h.md.probeResistance = &probeResistance{Type: "host", Value: "127.0.0.1:1"} // unreachable
h.Init(testMD(map[string]any{}))
req := httptest.NewRequest(http.MethodGet, "http://example.com/", nil)
w := httptest.NewRecorder()
ro := &xrecorder.HandlerRecorderObject{}
err := h.roundTrip(context.Background(), w, req, ro, &testLogger{})
if err != ErrAuthFailed {
t.Errorf("err = %v, want ErrAuthFailed", err)
}
// Should get a 503 on dial failure.
if w.Code != http.StatusServiceUnavailable {
t.Errorf("status = %d, want %d", w.Code, http.StatusServiceUnavailable)
}
}