opensource/SamWaf
3
0
Fork 0
mirror of https://gitee.com/samwaf/SamWaf.git synced 2026-10-31 02:01:44 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat: context canceled

Browse Source 
#771
This commit is contained in:
samwaf
2026-04-17 14:58:43 +08:00
parent d828c6d725
commit a7275bdc34
2 changed files with 349 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
wafenginecore/error_response_test.go
+276
View File
@@ -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)
}
}
wafenginecore/wafengine.go
+73 -7
View File
@@ -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 := "<nil>"
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("<html><head><title>服务不可用</title></head><body><center><h1>服务不可用</h1> <br><h3></h3></center></body> </html>")
@@ -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