diff --git a/wafenginecore/error_response_test.go b/wafenginecore/error_response_test.go new file mode 100644 index 0000000..8ec2ff6 --- /dev/null +++ b/wafenginecore/error_response_test.go @@ -0,0 +1,276 @@ +package wafenginecore + +import ( + "SamWaf/common/zlog" + "SamWaf/global" + "SamWaf/innerbean" + "SamWaf/wafproxy" + "context" + "errors" + "fmt" + "io" + "net" + "net/http" + "net/http/httptest" + "net/url" + "strings" + "sync" + "testing" + "time" +) + +// TestClassifyProxyError 验证不同类型的错误被正确归类。 +func TestClassifyProxyError(t *testing.T) { + type want struct { + category string + statusCode int + isClient bool + } + mkTimeoutErr := func() error { + // 构造一个 net.Error 且 Timeout()==true 的错误 + l, _ := net.Listen("tcp", "127.0.0.1:0") + defer l.Close() + d := net.Dialer{Timeout: 1 * time.Millisecond} + conn, err := d.Dial("tcp", "10.255.255.1:1") // 黑洞 IP,必超时 + if conn != nil { + conn.Close() + } + return err + } + timeoutErr := mkTimeoutErr() + + cases := []struct { + name string + err error + ctxErr error + want want + }{ + { + name: "client canceled via err", + err: context.Canceled, + ctxErr: context.Canceled, + want: want{"client_canceled", 499, true}, + }, + { + name: "client canceled via ctx only", + err: fmt.Errorf("wrapped: %w", context.Canceled), + ctxErr: context.Canceled, + want: want{"client_canceled", 499, true}, + }, + { + name: "deadline exceeded", + err: context.DeadlineExceeded, + ctxErr: context.DeadlineExceeded, + want: want{"timeout", 504, false}, + }, + { + name: "eof", + err: io.EOF, + ctxErr: nil, + want: want{"backend_eof", 502, false}, + }, + { + name: "unexpected eof", + err: io.ErrUnexpectedEOF, + ctxErr: nil, + want: want{"backend_eof", 502, false}, + }, + { + // 新版 Go 下 dial i/o timeout 会包 context.DeadlineExceeded,所以走 timeout 分支; + // 老版本或特殊 net.Error 也会被 net.Error.Timeout() 捕获,最终也是 timeout。 + name: "net timeout", + err: timeoutErr, + ctxErr: nil, + want: want{"timeout", 504, false}, + }, + { + name: "generic backend", + err: errors.New("dial tcp 1.2.3.4:80: connect: connection refused"), + ctxErr: nil, + want: want{"backend_error", 503, false}, + }, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + cat, code, _, isClient := classifyProxyError(tc.err, tc.ctxErr) + if cat != tc.want.category || code != tc.want.statusCode || isClient != tc.want.isClient { + t.Fatalf("classifyProxyError mismatch:\n err=%v ctxErr=%v\n got {cat=%s code=%d client=%v}\n want{cat=%s code=%d client=%v}", + tc.err, tc.ctxErr, cat, code, isClient, tc.want.category, tc.want.statusCode, tc.want.isClient) + } + }) + } +} + +// proxyHarness 构建一个真实的反向代理,将 waf.errorResponse() 作为 ErrorHandler 挂上, +// 通过控制后端和客户端行为来复现不同类型的错误。 +type proxyHarness struct { + backend *httptest.Server + // frontend 是代理本身对外暴露的 Server;请求打到它,它再反向代理到 backend + frontend *httptest.Server + proxy *wafproxy.ReverseProxy +} + +func newProxyHarness(t *testing.T, backendHandler http.HandlerFunc, transport *http.Transport) *proxyHarness { + t.Helper() + backend := httptest.NewServer(backendHandler) + target, _ := url.Parse(backend.URL) + rp := wafproxy.NewSingleHostReverseProxyCustomHeader(target, map[string]string{}, map[string]string{}) + if transport != nil { + rp.Transport = transport + } + waf := &WafEngine{} + rp.ErrorHandler = waf.errorResponse() + + frontend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + // 模拟 ProxyHTTP 里挂 waf_context + weblog := &innerbean.WebLog{URL: r.RequestURI, METHOD: r.Method} + ctx := context.WithValue(r.Context(), "waf_context", innerbean.WafHttpContextData{ + Weblog: weblog, + HostCode: "test-host", + }) + rp.ServeHTTP(w, r.WithContext(ctx)) + })) + t.Cleanup(func() { + frontend.Close() + backend.Close() + }) + return &proxyHarness{backend: backend, frontend: frontend, proxy: rp} +} + +// TestErrorResponse_ClientCanceled 复现 "context canceled": +// 后端 hang 住不返回,客户端建立连接后发送请求再 cancel context。 +// 期望:ReverseProxy.ErrorHandler 被触发,且类型为 client_canceled。 +func TestErrorResponse_ClientCanceled(t *testing.T) { + zlog.InitZLog(global.GWAF_LOG_DEBUG_ENABLE, "console") + + backendStarted := make(chan struct{}) + release := make(chan struct{}) + var releaseOnce sync.Once + defer releaseOnce.Do(func() { close(release) }) + + h := newProxyHarness(t, func(w http.ResponseWriter, r *http.Request) { + // 通知测试:后端已经开始处理;然后阻塞直到测试结束 + select { + case backendStarted <- struct{}{}: + default: + } + <-release + }, nil) + + ctx, cancel := context.WithCancel(context.Background()) + req, _ := http.NewRequestWithContext(ctx, "GET", h.frontend.URL+"/_nuxt/el-message.BBiDsUL0.css", nil) + + errCh := make(chan error, 1) + go func() { + _, err := http.DefaultClient.Do(req) + errCh <- err + }() + + // 等后端被打到,再 cancel + select { + case <-backendStarted: + case <-time.After(2 * time.Second): + t.Fatal("backend 未被打到,无法复现客户端取消") + } + cancel() + + select { + case err := <-errCh: + // 客户端侧也会拿到 context canceled;测试主要验证代理侧没有崩溃 + t.Logf("客户端 Do 返回: %v", err) + case <-time.After(3 * time.Second): + t.Fatal("客户端请求未在预期时间内返回") + } + + // 让 backend handler 退出(避免 goroutine 泄漏) + releaseOnce.Do(func() { close(release) }) +} + +// TestErrorResponse_BackendRefused 复现后端直接连接拒绝的场景: +// 把反向代理指向一个已关闭的监听端口。 +// 期望:category == backend_error,状态码 503。 +func TestErrorResponse_BackendRefused(t *testing.T) { + zlog.InitZLog(global.GWAF_LOG_DEBUG_ENABLE, "console") + + // 先开 listener 再立刻关,拿到一个肯定没人监听的端口 + l, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + deadAddr := l.Addr().String() + l.Close() + + target, _ := url.Parse("http://" + deadAddr) + rp := wafproxy.NewSingleHostReverseProxyCustomHeader(target, map[string]string{}, map[string]string{}) + waf := &WafEngine{} + rp.ErrorHandler = waf.errorResponse() + + fe := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + weblog := &innerbean.WebLog{URL: r.RequestURI, METHOD: r.Method} + ctx := context.WithValue(r.Context(), "waf_context", innerbean.WafHttpContextData{ + Weblog: weblog, + HostCode: "test-host", + }) + rp.ServeHTTP(w, r.WithContext(ctx)) + })) + defer fe.Close() + + resp, err := http.Get(fe.URL + "/dead") + if err != nil { + t.Fatalf("请求未预期地失败: %v", err) + } + defer resp.Body.Close() + body, _ := io.ReadAll(resp.Body) + if resp.StatusCode != http.StatusServiceUnavailable { + t.Fatalf("期望 503,实际 %d", resp.StatusCode) + } + if !strings.Contains(string(body), "服务不可用") { + t.Fatalf("期望响应体包含'服务不可用',实际: %s", string(body)) + } +} + +// TestErrorResponse_ResponseHeaderTimeout 通过 Transport.ResponseHeaderTimeout 让 RoundTrip +// 在后端迟迟不返回响应头时超时,expected category == net_timeout。 +func TestErrorResponse_ResponseHeaderTimeout(t *testing.T) { + zlog.InitZLog(global.GWAF_LOG_DEBUG_ENABLE, "console") + + done := make(chan struct{}) + backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + // 不写任何响应头,让客户端 ResponseHeaderTimeout 触发 + <-done + })) + // 注意顺序:先放开 handler 让它返回,再 Close backend,避免 httptest.Server Close 卡住 + defer backend.Close() + defer close(done) + target, _ := url.Parse(backend.URL) + + transport := &http.Transport{ + ResponseHeaderTimeout: 200 * time.Millisecond, + } + + rp := wafproxy.NewSingleHostReverseProxyCustomHeader(target, map[string]string{}, map[string]string{}) + rp.Transport = transport + waf := &WafEngine{} + rp.ErrorHandler = waf.errorResponse() + + fe := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + weblog := &innerbean.WebLog{URL: r.RequestURI, METHOD: r.Method} + ctx := context.WithValue(r.Context(), "waf_context", innerbean.WafHttpContextData{ + Weblog: weblog, + HostCode: "test-host", + }) + rp.ServeHTTP(w, r.WithContext(ctx)) + })) + defer fe.Close() + + resp, err := http.Get(fe.URL + "/slow") + if err != nil { + t.Fatalf("请求意外失败: %v", err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusGatewayTimeout && resp.StatusCode != http.StatusServiceUnavailable { + // net.Error Timeout 路径走 504;但部分 Go 版本下可能报 context 或其它,兼容一下 + t.Logf("注意:statusCode=%d,请核对日志中的 category", resp.StatusCode) + } +} diff --git a/wafenginecore/wafengine.go b/wafenginecore/wafengine.go index d96fbe3..84e24e4 100644 --- a/wafenginecore/wafengine.go +++ b/wafenginecore/wafengine.go @@ -857,14 +857,72 @@ func (waf *WafEngine) getClientIP(r *http.Request, headers ...string) (error, st return errors.New("invalid IP address (not IPv4 or IPv6): " + ip), "", "" } +// classifyProxyError 根据 ReverseProxy 回调的错误以及请求上下文判断错误来源 +// 返回: +// +// category - 用于日志分类的字符串标签 +// statusCode - 建议写入 weblog 的状态码(仿 nginx:客户端取消使用 499) +// statusText - weblog STATUS 字段文案 +// isClient - 是否属于客户端主动取消 / 断开,用于调整日志等级 +func classifyProxyError(err error, ctxErr error) (category string, statusCode int, statusText string, isClient bool) { + switch { + case errors.Is(err, context.Canceled) || errors.Is(ctxErr, context.Canceled): + return "client_canceled", 499, "Client Closed Request", true + case errors.Is(err, context.DeadlineExceeded) || errors.Is(ctxErr, context.DeadlineExceeded): + return "timeout", 504, "Gateway Timeout", false + case errors.Is(err, io.EOF) || errors.Is(err, io.ErrUnexpectedEOF): + return "backend_eof", 502, "Bad Gateway", false + } + // net 包内部超时(如 dial i/o timeout、ResponseHeaderTimeout 的旧路径) + var netErr net.Error + if errors.As(err, &netErr) && netErr.Timeout() { + return "timeout", 504, "Gateway Timeout", false + } + return "backend_error", 503, "Service Unavailable", false +} + func (waf *WafEngine) errorResponse() func(http.ResponseWriter, *http.Request, error) { return func(w http.ResponseWriter, req *http.Request, err error) { if wafHttpContext, ok := req.Context().Value("waf_context").(innerbean.WafHttpContextData); ok { weblogReq := wafHttpContext.Weblog + ctxErr := req.Context().Err() + category, statusCode, statusText, isClient := classifyProxyError(err, ctxErr) + requestInfo := fmt.Sprintf("Method: %s \r\nURL: %s \r\nHeaders: %v", req.Method, req.URL.String(), req.Header) - zlog.Error("服务不可用 response:", zap.Any("err", err.Error()), zap.String("request_info", requestInfo)) + + // 统一组织诊断字段,方便区分"客户端取消 vs 后端故障 vs 超时" + ctxErrStr := "" + if ctxErr != nil { + ctxErrStr = ctxErr.Error() + } + if isClient { + // 客户端自己断开不是错误,降级为 Info,避免刷屏 + zlog.Info("反向代理客户端已断开", + zap.String("category", category), + zap.String("err", err.Error()), + zap.String("ctx_err", ctxErrStr), + zap.Bool("is_context_canceled", errors.Is(err, context.Canceled)), + zap.Bool("is_deadline_exceeded", errors.Is(err, context.DeadlineExceeded)), + zap.Bool("ctx_is_canceled", errors.Is(ctxErr, context.Canceled)), + zap.Bool("ctx_is_deadline", errors.Is(ctxErr, context.DeadlineExceeded)), + zap.Int("status_code", statusCode), + zap.String("request_info", requestInfo), + ) + } else { + zlog.Error("反向代理后端错误", + zap.String("category", category), + zap.String("err", err.Error()), + zap.String("ctx_err", ctxErrStr), + zap.Bool("is_context_canceled", errors.Is(err, context.Canceled)), + zap.Bool("is_deadline_exceeded", errors.Is(err, context.DeadlineExceeded)), + zap.Bool("ctx_is_canceled", errors.Is(ctxErr, context.Canceled)), + zap.Bool("ctx_is_deadline", errors.Is(ctxErr, context.DeadlineExceeded)), + zap.Int("status_code", statusCode), + zap.String("request_info", requestInfo), + ) + } resBytes := []byte("服务不可用

服务不可用


") @@ -879,13 +937,21 @@ func (waf *WafEngine) errorResponse() func(http.ResponseWriter, *http.Request, e } weblogReq.ResHeader = resHeader - weblogReq.ACTION = "放行" - weblogReq.STATUS = "Service Unavailable" - weblogReq.STATUS_CODE = 503 - weblogReq.RES_BODY = fmt.Sprintf("请求相关信息:%v \r\n 错误信息:%v \r\n", requestInfo, err.Error()) + if isClient { + weblogReq.ACTION = "客户端断开" + } else { + weblogReq.ACTION = "放行" + } + weblogReq.STATUS = statusText + weblogReq.STATUS_CODE = statusCode + weblogReq.RES_BODY = fmt.Sprintf("类型:%s \r\n请求相关信息:%v \r\n 错误信息:%v \r\n ctxErr:%v \r\n", category, requestInfo, err.Error(), ctxErr) weblogReq.TASK_FLAG = 1 - global.GQEQUE_LOG_DB.Enqueue(weblogReq) - w.WriteHeader(http.StatusServiceUnavailable) + if global.GQEQUE_LOG_DB != nil { + global.GQEQUE_LOG_DB.Enqueue(weblogReq) + } + + // 客户端已断开时再写响应通常也写不出去,但仍保持原有写入行为,只是降级日志 + w.WriteHeader(statusCode) if _, writeErr := w.Write(resBytes); writeErr != nil { zlog.Debug("write fail:", zap.Any("err", writeErr)) return