Merge pull request #4515 from BenjaminAaron196/feat/filter-noise-rejected-requests

(fix) 过滤入口拒绝日志并强化鉴权安全边界

backend/internal/handler/admin/ops_auth_cache_health_handler.go

@@ -0,0 +1,21 @@
+package admin
+
+import (
+	"net/http"
+
+	"github.com/Wei-Shaw/sub2api/internal/pkg/response"
+	"github.com/gin-gonic/gin"
+)
+
+// GetAuthCacheInvalidationHealth exposes durable outbox lag and subscriber health.
+func (h *OpsHandler) GetAuthCacheInvalidationHealth(c *gin.Context) {
+	if h.opsService == nil {
+		response.Error(c, http.StatusServiceUnavailable, "Ops service not available")
+		return
+	}
+	if err := h.opsService.RequireMonitoringEnabled(c.Request.Context()); err != nil {
+		response.ErrorFrom(c, err)
+		return
+	}
+	response.Success(c, h.opsService.GetAuthCacheInvalidationHealth(c.Request.Context()))
+}

backend/internal/handler/admin/ops_ingress_reject_handler.go

@@ -0,0 +1,131 @@
+package admin
+
+import (
+	"net/http"
+	"net/netip"
+	"strconv"
+	"strings"
+
+	"github.com/Wei-Shaw/sub2api/internal/pkg/response"
+	"github.com/Wei-Shaw/sub2api/internal/service"
+	"github.com/gin-gonic/gin"
+)
+
+var ingressRejectReasons = map[string]struct{}{
+	"query_api_key_deprecated": {}, "api_key_required": {}, "invalid_api_key": {},
+	"invalid_auth_rate_limited": {},
+	"api_key_auth_overloaded":   {},
+	"api_key_disabled":          {}, "ip_restricted": {}, "user_inactive": {}, "group_deleted": {},
+	"group_disabled": {}, "group_not_allowed": {}, "group_unassigned": {}, "other": {},
+}
+
+var ingressRejectRouteFamilies = map[string]struct{}{
+	"antigravity": {}, "gemini": {}, "codex": {}, "messages": {}, "responses": {},
+	"chat_completions": {}, "images": {}, "videos": {}, "embeddings": {}, "models": {}, "other": {},
+}
+
+var ingressRejectProtocols = map[string]struct{}{
+	"google": {}, "anthropic": {}, "openai": {}, "gateway": {}, "other": {},
+}
+
+// ListIngressRejects returns bounded security aggregates, never raw credentials or request bodies.
+func (h *OpsHandler) ListIngressRejects(c *gin.Context) {
+	if h.opsService == nil {
+		response.Error(c, http.StatusServiceUnavailable, "Ops service not available")
+		return
+	}
+	page, pageSize := response.ParsePagination(c)
+	if pageSize > 200 {
+		pageSize = 200
+	}
+	startTime, endTime, err := parseOpsTimeRange(c, "1h")
+	if err != nil {
+		response.BadRequest(c, err.Error())
+		return
+	}
+	filter := &service.OpsIngressRejectFilter{Page: page, PageSize: pageSize}
+	if !startTime.IsZero() {
+		filter.StartTime = &startTime
+	}
+	if !endTime.IsZero() {
+		filter.EndTime = &endTime
+	}
+	if filter.RejectReason, err = parseIngressRejectEnum(c, "reason", ingressRejectReasons); err != nil {
+		response.BadRequest(c, err.Error())
+		return
+	}
+	if filter.RouteFamily, err = parseIngressRejectEnum(c, "route_family", ingressRejectRouteFamilies); err != nil {
+		response.BadRequest(c, err.Error())
+		return
+	}
+	if filter.Protocol, err = parseIngressRejectEnum(c, "protocol", ingressRejectProtocols); err != nil {
+		response.BadRequest(c, err.Error())
+		return
+	}
+	if raw := strings.TrimSpace(c.Query("client_ip")); raw != "" {
+		addr, parseErr := netip.ParseAddr(raw)
+		if parseErr != nil {
+			response.BadRequest(c, "Invalid client_ip")
+			return
+		}
+		addr = addr.Unmap()
+		if addr.Is6() {
+			addr = netip.PrefixFrom(addr, 64).Masked().Addr()
+		}
+		filter.ClientIP = addr.String()
+	}
+	if filter.UserID, err = parseOptionalPositiveID(c, "user_id"); err != nil {
+		response.BadRequest(c, err.Error())
+		return
+	}
+	if filter.APIKeyID, err = parseOptionalPositiveID(c, "api_key_id"); err != nil {
+		response.BadRequest(c, err.Error())
+		return
+	}
+
+	result, err := h.opsService.ListIngressRejects(c.Request.Context(), filter)
+	if err != nil {
+		response.ErrorFrom(c, err)
+		return
+	}
+	response.Success(c, result)
+}
+
+func (h *OpsHandler) GetIngressRejectHealth(c *gin.Context) {
+	if h.opsService == nil {
+		response.Error(c, http.StatusServiceUnavailable, "Ops service not available")
+		return
+	}
+	if err := h.opsService.RequireMonitoringEnabled(c.Request.Context()); err != nil {
+		response.ErrorFrom(c, err)
+		return
+	}
+	response.Success(c, h.opsService.GetIngressRejectHealth())
+}
+
+func parseIngressRejectEnum(c *gin.Context, name string, allowed map[string]struct{}) (string, error) {
+	value := strings.TrimSpace(c.Query(name))
+	if value == "" {
+		return "", nil
+	}
+	if _, ok := allowed[value]; !ok {
+		return "", &ingressRejectQueryError{message: "Invalid " + name}
+	}
+	return value, nil
+}
+
+func parseOptionalPositiveID(c *gin.Context, name string) (*int64, error) {
+	raw := strings.TrimSpace(c.Query(name))
+	if raw == "" {
+		return nil, nil
+	}
+	value, err := strconv.ParseInt(raw, 10, 64)
+	if err != nil || value <= 0 {
+		return nil, &ingressRejectQueryError{message: "Invalid " + name}
+	}
+	return &value, nil
+}
+
+type ingressRejectQueryError struct{ message string }
+
+func (e *ingressRejectQueryError) Error() string { return e.message }

backend/internal/handler/admin/ops_ingress_reject_handler_test.go

@@ -0,0 +1,41 @@
+package admin
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/Wei-Shaw/sub2api/internal/service"
+	"github.com/gin-gonic/gin"
+	"github.com/stretchr/testify/require"
+)
+
+func newIngressRejectHandlerForTest() *OpsHandler {
+	return NewOpsHandler(service.NewOpsService(nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil))
+}
+
+func TestListIngressRejectsValidatesFilters(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	for _, path := range []string{
+		"/api/v1/admin/ops/ingress-rejections?reason=not-valid",
+		"/api/v1/admin/ops/ingress-rejections?client_ip=not-an-ip",
+		"/api/v1/admin/ops/ingress-rejections?user_id=0",
+		"/api/v1/admin/ops/ingress-rejections?api_key_id=-1",
+	} {
+		t.Run(path, func(t *testing.T) {
+			context, _ := gin.CreateTestContext(httptest.NewRecorder())
+			context.Request = httptest.NewRequest(http.MethodGet, path, nil)
+			newIngressRejectHandlerForTest().ListIngressRejects(context)
+			require.Equal(t, http.StatusBadRequest, context.Writer.Status())
+		})
+	}
+}
+
+func TestGetIngressRejectHealthAvailable(t *testing.T) {
+	recorder := httptest.NewRecorder()
+	context, _ := gin.CreateTestContext(recorder)
+	context.Request = httptest.NewRequest(http.MethodGet, "/api/v1/admin/ops/ingress-rejections/health", nil)
+	newIngressRejectHandlerForTest().GetIngressRejectHealth(context)
+	require.Equal(t, http.StatusOK, recorder.Code)
+	require.Contains(t, recorder.Body.String(), `"capacity":8192`)
+}

backend/internal/handler/admin/setting_handler_update.go

@@ -1688,6 +1688,9 @@ func (h *SettingHandler) UpdateSettings(c *gin.Context) {
 		response.ErrorFrom(c, err)
 		return
 	}
+	if h.opsService != nil {
+		h.opsService.SetMonitoringEnabled(settings.OpsMonitoringEnabled)
+	}
 
 	// Update OpenAI fast policy (stored under dedicated key, only when provided).
 	if req.OpenAIFastPolicySettings != nil {

backend/internal/handler/openai_gateway_credential_failover_test.go

@@ -206,6 +206,10 @@ func TestOpsRecoveredCredentialFailoverUsesAccountAuthAttribution(t *testing.T)
 	require.NotContains(t, job.entry.ErrorMessage, "earlier inference failure")
 	require.NotNil(t, job.entry.UpstreamStatusCode)
 	require.Zero(t, *job.entry.UpstreamStatusCode)
-	require.Len(t, job.entry.UpstreamErrors, 2)
-	require.Equal(t, http.StatusForbidden, job.entry.UpstreamErrors[0].UpstreamStatusCode)
+	require.Nil(t, job.entry.UpstreamErrors)
+	require.NotNil(t, job.entry.UpstreamErrorsJSON)
+	events, err := service.ParseOpsUpstreamErrors(*job.entry.UpstreamErrorsJSON)
+	require.NoError(t, err)
+	require.Len(t, events, 2)
+	require.Equal(t, http.StatusForbidden, events[0].UpstreamStatusCode)
 }

backend/internal/handler/ops_error_logger.go

@@ -72,24 +72,10 @@ const (
 	opsErrorLogMinQueueSize       = 256
 	opsErrorLogMaxQueueSize       = 8192
 	opsErrorLogBatchSize          = 32
+	opsErrorLogMaxQueueBytes      = 32 * 1024 * 1024
+	opsErrorLogMaxUserAgentBytes  = 512
 )
 
-// looksLikeSystemKey 粗筛"形似本系统 key"的输入:长度 16-128 且仅含 [a-zA-Z0-9_-]。
-// 不用前缀匹配(APIKeyPrefix 可配置)。用于反查审计表前挡掉随机扫描的乱码输入。
-func looksLikeSystemKey(key string) bool {
-	if len(key) < 16 || len(key) > 128 {
-		return false
-	}
-	for _, c := range key {
-		allowed := (c >= 'a' && c <= 'z') || (c >= 'A' && c <= 'Z') ||
-			(c >= '0' && c <= '9') || c == '_' || c == '-'
-		if !allowed {
-			return false
-		}
-	}
-	return true
-}
-
 // keyPrefix 返回脱敏前缀(前 n 个字符);不足 n 则原样返回。
 func keyPrefix(key string, n int) string {
 	if len(key) <= n {
@@ -98,43 +84,26 @@ func keyPrefix(key string, n int) string {
 	return key[:n]
 }
 
-// extractAttemptedKey 按认证中间件同样的顺序从请求头提取提交的 key 明文。
-// 与 api_key_auth.go:43-59 一致:Authorization 仅取 Bearer scheme,非 Bearer 则忽略并继续 x-api-key → x-goog-api-key。
-func extractAttemptedKey(c *gin.Context) string {
-	if h := c.GetHeader("Authorization"); h != "" {
-		parts := strings.SplitN(h, " ", 2)
-		if len(parts) == 2 && strings.EqualFold(parts[0], "Bearer") {
-			return strings.TrimSpace(parts[1])
-		}
-		// 非 Bearer:与中间件一致,忽略 Authorization,继续尝试其它 header(不在此 return)。
-	}
-	if k := c.GetHeader("x-api-key"); k != "" {
-		return strings.TrimSpace(k)
-	}
-	if k := c.GetHeader("x-goog-api-key"); k != "" {
-		return strings.TrimSpace(k)
-	}
-	return ""
-}
-
 type opsErrorLogJob struct {
-	ops   *service.OpsService
-	entry *service.OpsInsertErrorLogInput
+	ops         *service.OpsService
+	entry       *service.OpsInsertErrorLogInput
+	queuedBytes int64
 }
 
 var (
 	opsErrorLogOnce  sync.Once
 	opsErrorLogQueue chan opsErrorLogJob
 
-	opsErrorLogStopOnce  sync.Once
-	opsErrorLogWorkersWg sync.WaitGroup
-	opsErrorLogMu        sync.RWMutex
-	opsErrorLogStopping  bool
-	opsErrorLogQueueLen  atomic.Int64
-	opsErrorLogEnqueued  atomic.Int64
-	opsErrorLogDropped   atomic.Int64
-	opsErrorLogProcessed atomic.Int64
-	opsErrorLogSanitized atomic.Int64
+	opsErrorLogStopOnce   sync.Once
+	opsErrorLogWorkersWg  sync.WaitGroup
+	opsErrorLogMu         sync.RWMutex
+	opsErrorLogStopping   bool
+	opsErrorLogQueueLen   atomic.Int64
+	opsErrorLogQueueBytes atomic.Int64
+	opsErrorLogEnqueued   atomic.Int64
+	opsErrorLogDropped    atomic.Int64
+	opsErrorLogProcessed  atomic.Int64
+	opsErrorLogSanitized  atomic.Int64
 
 	opsErrorLogLastDropLogAt atomic.Int64
 
@@ -154,6 +123,7 @@ func startOpsErrorLogWorkers() {
 	workerCount, queueSize := opsErrorLogConfig()
 	opsErrorLogQueue = make(chan opsErrorLogJob, queueSize)
 	opsErrorLogQueueLen.Store(0)
+	opsErrorLogQueueBytes.Store(0)
 
 	opsErrorLogWorkersWg.Add(workerCount)
 	for i := 0; i < workerCount; i++ {
@@ -165,6 +135,7 @@ func startOpsErrorLogWorkers() {
 					return
 				}
 				opsErrorLogQueueLen.Add(-1)
+				opsErrorLogQueueBytes.Add(-job.queuedBytes)
 				batch := make([]opsErrorLogJob, 0, opsErrorLogBatchSize)
 				batch = append(batch, job)
 
@@ -184,6 +155,7 @@ func startOpsErrorLogWorkers() {
 							return
 						}
 						opsErrorLogQueueLen.Add(-1)
+						opsErrorLogQueueBytes.Add(-nextJob.queuedBytes)
 						batch = append(batch, nextJob)
 					case <-timer.C:
 						break batchLoop
@@ -239,6 +211,20 @@ func enqueueOpsErrorLog(ops *service.OpsService, entry *service.OpsInsertErrorLo
 	if ops == nil || entry == nil {
 		return
 	}
+	entry.UserAgent = normalizeOpsPersistentUserAgent(entry.UserAgent)
+	if entry.ErrorBody != "" {
+		originalBody := entry.ErrorBody
+		body, truncated := service.SanitizeOpsErrorBodyForQueue(originalBody)
+		entry.ErrorBody = body
+		if truncated || body != originalBody {
+			opsErrorLogSanitized.Add(1)
+		}
+	}
+	if err := service.SanitizeOpsUpstreamErrorsForQueue(entry); err != nil {
+		opsErrorLogDropped.Add(1)
+		maybeLogOpsErrorLogDrop()
+		return
+	}
 	select {
 	case <-opsErrorLogShutdownCh:
 		return
@@ -259,18 +245,29 @@ func enqueueOpsErrorLog(ops *service.OpsService, entry *service.OpsInsertErrorLo
 	if opsErrorLogStopping || opsErrorLogQueue == nil {
 		return
 	}
+	queuedBytes := estimateOpsErrorLogJobBytes(entry)
+	if !reserveOpsErrorLogQueueBytes(queuedBytes) {
+		opsErrorLogDropped.Add(1)
+		maybeLogOpsErrorLogDrop()
+		return
+	}
 
 	select {
-	case opsErrorLogQueue <- opsErrorLogJob{ops: ops, entry: entry}:
-		opsErrorLogQueueLen.Add(1)
+	case opsErrorLogQueue <- opsErrorLogJob{ops: ops, entry: entry, queuedBytes: queuedBytes}:
 		opsErrorLogEnqueued.Add(1)
 	default:
+		opsErrorLogQueueLen.Add(-1)
+		opsErrorLogQueueBytes.Add(-queuedBytes)
 		// Queue is full; drop to avoid blocking request handling.
 		opsErrorLogDropped.Add(1)
 		maybeLogOpsErrorLogDrop()
 	}
 }
 
+func normalizeOpsPersistentUserAgent(value string) string {
+	return truncateString(strings.TrimSpace(strings.ToValidUTF8(value, "")), opsErrorLogMaxUserAgentBytes)
+}
+
 func StopOpsErrorLogWorkers() bool {
 	opsErrorLogStopOnce.Do(func() {
 		opsErrorLogShutdownOnce.Do(func() {
@@ -293,6 +290,7 @@ func stopOpsErrorLogWorkers() bool {
 
 	if ch == nil {
 		opsErrorLogQueueLen.Store(0)
+		opsErrorLogQueueBytes.Store(0)
 		return true
 	}
 
@@ -305,6 +303,7 @@ func stopOpsErrorLogWorkers() bool {
 	select {
 	case <-done:
 		opsErrorLogQueueLen.Store(0)
+		opsErrorLogQueueBytes.Store(0)
 		return true
 	case <-time.After(opsErrorLogDrainTimeout):
 		return false
@@ -315,6 +314,14 @@ func OpsErrorLogQueueLength() int64 {
 	return opsErrorLogQueueLen.Load()
 }
 
+func OpsErrorLogQueueBytes() int64 {
+	return opsErrorLogQueueBytes.Load()
+}
+
+func OpsErrorLogQueueBytesCapacity() int64 {
+	return opsErrorLogMaxQueueBytes
+}
+
 func OpsErrorLogQueueCapacity() int {
 	opsErrorLogMu.RLock()
 	ch := opsErrorLogQueue
@@ -355,12 +362,15 @@ func maybeLogOpsErrorLogDrop() {
 	}
 
 	queued := opsErrorLogQueueLen.Load()
+	queuedBytes := opsErrorLogQueueBytes.Load()
 	queueCap := OpsErrorLogQueueCapacity()
 
 	log.Printf(
-		"[OpsErrorLogger] queue is full; dropping logs (queued=%d cap=%d enqueued_total=%d dropped_total=%d processed_total=%d sanitized_total=%d)",
+		"[OpsErrorLogger] queue is full; dropping logs (queued=%d cap=%d queued_bytes=%d bytes_cap=%d enqueued_total=%d dropped_total=%d processed_total=%d sanitized_total=%d)",
 		queued,
 		queueCap,
+		queuedBytes,
+		opsErrorLogMaxQueueBytes,
 		opsErrorLogEnqueued.Load(),
 		opsErrorLogDropped.Load(),
 		opsErrorLogProcessed.Load(),
@@ -368,6 +378,46 @@ func maybeLogOpsErrorLogDrop() {
 	)
 }
 
+func reserveOpsErrorLogQueueBytes(size int64) bool {
+	if size < 1 {
+		size = 1
+	}
+	for {
+		current := opsErrorLogQueueBytes.Load()
+		if current > opsErrorLogMaxQueueBytes-size {
+			return false
+		}
+		if opsErrorLogQueueBytes.CompareAndSwap(current, current+size) {
+			opsErrorLogQueueLen.Add(1)
+			return true
+		}
+	}
+}
+
+func estimateOpsErrorLogJobBytes(entry *service.OpsInsertErrorLogInput) int64 {
+	if entry == nil {
+		return 1
+	}
+	const fixedOverhead = 512
+	size := fixedOverhead + len(entry.RequestID) + len(entry.ClientRequestID) +
+		len(entry.Platform) + len(entry.Model) + len(entry.RequestPath) +
+		len(entry.InboundEndpoint) + len(entry.UpstreamEndpoint) +
+		len(entry.RequestedModel) + len(entry.UpstreamModel) + len(entry.UserAgent) +
+		len(entry.ErrorPhase) + len(entry.ErrorType) + len(entry.Severity) +
+		len(entry.ErrorMessage) + len(entry.ErrorBody) + len(entry.ErrorSource) +
+		len(entry.ErrorOwner) + len(entry.APIKeyPrefix)
+	if entry.UpstreamErrorMessage != nil {
+		size += len(*entry.UpstreamErrorMessage)
+	}
+	if entry.UpstreamErrorDetail != nil {
+		size += len(*entry.UpstreamErrorDetail)
+	}
+	if entry.UpstreamErrorsJSON != nil {
+		size += len(*entry.UpstreamErrorsJSON)
+	}
+	return int64(size)
+}
+
 func opsErrorLogConfig() (workerCount int, queueSize int) {
 	workerCount = runtime.GOMAXPROCS(0) * 2
 	if workerCount < opsErrorLogMinWorkerCount {
@@ -470,9 +520,12 @@ type opsCaptureWriter struct {
 	gin.ResponseWriter
 	limit int
 	buf   bytes.Buffer
+	ctx   *gin.Context
 }
 
-const opsCaptureWriterLimit = 64 * 1024
+const opsCaptureWriterLimit = service.OpsErrorLogQueueBodyMaxBytes
+
+const opsCaptureWriterPoolMaxRetainedCapacity = service.OpsErrorLogQueueBodyMaxBytes
 
 var opsCaptureWriterPool = sync.Pool{
 	New: func() any {
@@ -496,11 +549,19 @@ func releaseOpsCaptureWriter(w *opsCaptureWriter) {
 		return
 	}
 	w.ResponseWriter = nil
+	w.ctx = nil
 	w.limit = opsCaptureWriterLimit
+	if !shouldPoolOpsCaptureWriter(w) {
+		return
+	}
 	w.buf.Reset()
 	opsCaptureWriterPool.Put(w)
 }
 
+func shouldPoolOpsCaptureWriter(w *opsCaptureWriter) bool {
+	return w != nil && w.buf.Cap() <= opsCaptureWriterPoolMaxRetainedCapacity
+}
+
 func (w *opsCaptureWriter) Status() int {
 	if w.ResponseWriter == nil {
 		return 0
@@ -577,7 +638,7 @@ func (w *opsCaptureWriter) Write(b []byte) (int, error) {
 	if w.ResponseWriter == nil {
 		return 0, nil
 	}
-	if w.Status() >= 400 && w.limit > 0 && w.buf.Len() < w.limit {
+	if w.shouldCapture() && w.Status() >= 400 && w.limit > 0 && w.buf.Len() < w.limit {
 		remaining := w.limit - w.buf.Len()
 		if len(b) > remaining {
 			_, _ = w.buf.Write(b[:remaining])
@@ -592,7 +653,7 @@ func (w *opsCaptureWriter) WriteString(s string) (int, error) {
 	if w.ResponseWriter == nil {
 		return 0, nil
 	}
-	if w.Status() >= 400 && w.limit > 0 && w.buf.Len() < w.limit {
+	if w.shouldCapture() && w.Status() >= 400 && w.limit > 0 && w.buf.Len() < w.limit {
 		remaining := w.limit - w.buf.Len()
 		if len(s) > remaining {
 			_, _ = w.buf.WriteString(s[:remaining])
@@ -603,6 +664,14 @@ func (w *opsCaptureWriter) WriteString(s string) (int, error) {
 	return w.ResponseWriter.WriteString(s)
 }
 
+func (w *opsCaptureWriter) shouldCapture() bool {
+	if w.ctx == nil {
+		return true
+	}
+	_, rejected := middleware2.GetIngressRejectReason(w.ctx)
+	return !rejected
+}
+
 // OpsErrorLoggerMiddleware records error responses (status >= 400) into ops_error_logs.
 //
 // Notes:
@@ -612,6 +681,7 @@ func OpsErrorLoggerMiddleware(ops *service.OpsService) gin.HandlerFunc {
 	return func(c *gin.Context) {
 		originalWriter := c.Writer
 		w := acquireOpsCaptureWriter(originalWriter)
+		w.ctx = c
 		defer func() {
 			// Restore the original writer before returning so outer middlewares
 			// don't observe a pooled wrapper that has been released.
@@ -623,6 +693,10 @@ func OpsErrorLoggerMiddleware(ops *service.OpsService) gin.HandlerFunc {
 		c.Writer = w
 		c.Next()
 
+		if _, rejected := middleware2.GetIngressRejectReason(c); rejected {
+			return
+		}
+
 		if ops == nil {
 			return
 		}
@@ -989,7 +1063,7 @@ func OpsErrorLoggerMiddleware(ops *service.OpsService) gin.HandlerFunc {
 			IsCountTokens:     isCountTokensRequest(c),
 
 			ErrorMessage: parsed.Message,
-			// Keep the full captured error body (capture is already capped at 64KB) so the
+			// Keep the captured error body (already capped at the queue-safe limit) so the
 			// service layer can sanitize JSON before truncating for storage.
 			ErrorBody:   string(body),
 			ErrorSource: errorSource,
@@ -1002,7 +1076,7 @@ func OpsErrorLoggerMiddleware(ops *service.OpsService) gin.HandlerFunc {
 
 		if apiKey != nil {
 			entry.APIKeyID = &apiKey.ID
-			// 有效(未删除)key 报错时快照前缀,key 之后被删也保留;与 INVALID_API_KEY 的 attempted_key_prefix 互斥。
+			// 有效 key 报错时快照前缀，key 之后被删也保留。
 			entry.APIKeyPrefix = keyPrefix(apiKey.Key, 8)
 			if apiKey.User != nil {
 				entry.UserID = &apiKey.User.ID
@@ -1022,22 +1096,6 @@ func OpsErrorLoggerMiddleware(ops *service.OpsService) gin.HandlerFunc {
 			entry.ClientIP = &clientIP
 		}
 
-		// 已删除 key 归因:仅 INVALID_API_KEY 才尝试。响应已写出,此处不阻塞客户端。
-		if parsed.Code == opsCodeInvalidAPIKey {
-			if attemptedKey := extractAttemptedKey(c); attemptedKey != "" {
-				entry.AttemptedKeyPrefix = keyPrefix(attemptedKey, 8)
-				if looksLikeSystemKey(attemptedKey) {
-					if res, lookupErr := ops.LookupDeletedKeyAudit(c.Request.Context(), attemptedKey); lookupErr != nil {
-						log.Printf("[OpsErrorLogger] LookupDeletedKeyAudit failed: %v", lookupErr)
-					} else if res != nil {
-						owner := res.UserID
-						entry.DeletedKeyOwnerUserID = &owner
-						entry.DeletedKeyName = res.KeyName
-					}
-				}
-			}
-		}
-
 		enqueueOpsErrorLog(ops, entry)
 	}
 }
@@ -1706,11 +1764,8 @@ func shouldSkipOpsErrorLog(ctx context.Context, ops *service.OpsService, message
 	}
 
 	// Get advanced settings to check filter configuration
-	settings, err := ops.GetOpsAdvancedSettings(ctx)
-	if err != nil || settings == nil {
-		// If we can't get settings, don't skip (fail open)
-		return false
-	}
+	_ = ctx
+	settings := ops.OpsAdvancedSettingsSnapshot()
 
 	msgLower := strings.ToLower(message)
 	bodyLower := strings.ToLower(body)

backend/internal/handler/ops_error_logger_attribution_test.go

@@ -1,39 +1,9 @@
 package handler
 
 import (
-	"net/http"
-	"net/http/httptest"
 	"testing"
-
-	"github.com/gin-gonic/gin"
 )
 
-func TestLooksLikeSystemKey(t *testing.T) {
-	cases := []struct {
-		in   string
-		want bool
-	}{
-		{"sk-abcdef0123456789", true},
-		{"ABCdef_-0123456789", true},
-		{"short", false},
-		{"with space xxxxxxxxxx", false},
-		{"汉字key1234567890", false},
-		{"", false},
-	}
-	for _, c := range cases {
-		if got := looksLikeSystemKey(c.in); got != c.want {
-			t.Errorf("looksLikeSystemKey(%q)=%v want %v", c.in, got, c.want)
-		}
-	}
-	long := make([]byte, 129)
-	for i := range long {
-		long[i] = 'a'
-	}
-	if looksLikeSystemKey(string(long)) {
-		t.Errorf("129-char key should be rejected")
-	}
-}
-
 func TestKeyPrefix(t *testing.T) {
 	if got := keyPrefix("sk-3f2a9c7e", 8); got != "sk-3f2a9" {
 		t.Errorf("keyPrefix=%q want %q", got, "sk-3f2a9")
@@ -42,77 +12,3 @@ func TestKeyPrefix(t *testing.T) {
 		t.Errorf("short key should be returned as-is, got %q", got)
 	}
 }
-
-func TestExtractAttemptedKey(t *testing.T) {
-	gin.SetMode(gin.TestMode)
-
-	cases := []struct {
-		name    string
-		headers map[string]string
-		want    string
-	}{
-		{
-			name:    "Bearer in Authorization",
-			headers: map[string]string{"Authorization": "Bearer sk-testkey0123456789"},
-			want:    "sk-testkey0123456789",
-		},
-		{
-			name:    "Bearer case-insensitive",
-			headers: map[string]string{"Authorization": "BEARER sk-testkey0123456789"},
-			want:    "sk-testkey0123456789",
-		},
-		{
-			name:    "x-api-key header",
-			headers: map[string]string{"x-api-key": "sk-xapikey0123456789"},
-			want:    "sk-xapikey0123456789",
-		},
-		{
-			name:    "x-goog-api-key header",
-			headers: map[string]string{"x-goog-api-key": "sk-goog0123456789"},
-			want:    "sk-goog0123456789",
-		},
-		{
-			name:    "Authorization takes priority over x-api-key",
-			headers: map[string]string{"Authorization": "Bearer sk-auth0123456789", "x-api-key": "sk-xapi0123456789"},
-			want:    "sk-auth0123456789",
-		},
-		{
-			name:    "x-api-key takes priority over x-goog-api-key",
-			headers: map[string]string{"x-api-key": "sk-xapi0123456789", "x-goog-api-key": "sk-goog0123456789"},
-			want:    "sk-xapi0123456789",
-		},
-		{
-			name:    "no key headers",
-			headers: map[string]string{},
-			want:    "",
-		},
-		{
-			name:    "Bearer with leading/trailing spaces trimmed",
-			headers: map[string]string{"Authorization": "Bearer   sk-trimmed0123456789  "},
-			want:    "sk-trimmed0123456789",
-		},
-		{
-			// 非 Bearer Authorization 应被忽略,继续 fall-through 到 x-api-key(与认证中间件一致)
-			name:    "non-Bearer Authorization falls through to x-api-key",
-			headers: map[string]string{"Authorization": "junk-not-bearer", "x-api-key": "sk-realkey1234567"},
-			want:    "sk-realkey1234567",
-		},
-	}
-
-	for _, tc := range cases {
-		t.Run(tc.name, func(t *testing.T) {
-			rec := httptest.NewRecorder()
-			c, _ := gin.CreateTestContext(rec)
-			req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", nil)
-			for k, v := range tc.headers {
-				req.Header.Set(k, v)
-			}
-			c.Request = req
-
-			got := extractAttemptedKey(c)
-			if got != tc.want {
-				t.Errorf("extractAttemptedKey(%v) = %q, want %q", tc.headers, got, tc.want)
-			}
-		})
-	}
-}

backend/internal/handler/ops_error_logger_test.go

@@ -1,10 +1,13 @@
 package handler
 
 import (
+	"context"
 	"net/http"
 	"net/http/httptest"
+	"strings"
 	"sync"
 	"testing"
+	"unicode/utf8"
 
 	middleware2 "github.com/Wei-Shaw/sub2api/internal/server/middleware"
 	"github.com/Wei-Shaw/sub2api/internal/service"
@@ -12,6 +15,82 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
+type ingressRejectSettingRepo struct {
+	service.SettingRepository
+	getValueCalls int
+}
+
+func (r *ingressRejectSettingRepo) GetValue(context.Context, string) (string, error) {
+	r.getValueCalls++
+	return "", service.ErrSettingNotFound
+}
+
+func (r *ingressRejectSettingRepo) GetMultiple(context.Context, []string) (map[string]string, error) {
+	r.getValueCalls++
+	return map[string]string{}, nil
+}
+
+func (r *ingressRejectSettingRepo) Set(context.Context, string, string) error {
+	return nil
+}
+
+type ingressRejectOpsRepo struct {
+	service.OpsRepository
+	insertCalls int
+}
+
+func (r *ingressRejectOpsRepo) InsertErrorLog(context.Context, *service.OpsInsertErrorLogInput) (int64, error) {
+	r.insertCalls++
+	return 0, nil
+}
+
+func (r *ingressRejectOpsRepo) BatchInsertErrorLogs(context.Context, []*service.OpsInsertErrorLogInput) (int64, error) {
+	r.insertCalls++
+	return 0, nil
+}
+
+func TestOpsErrorLogQueueByteBudget(t *testing.T) {
+	previousBytes := opsErrorLogQueueBytes.Load()
+	previousLen := opsErrorLogQueueLen.Load()
+	opsErrorLogQueueBytes.Store(0)
+	opsErrorLogQueueLen.Store(0)
+	t.Cleanup(func() {
+		opsErrorLogQueueBytes.Store(previousBytes)
+		opsErrorLogQueueLen.Store(previousLen)
+	})
+
+	if !reserveOpsErrorLogQueueBytes(opsErrorLogMaxQueueBytes - 1) {
+		t.Fatal("first reservation within byte budget should succeed")
+	}
+	if reserveOpsErrorLogQueueBytes(2) {
+		t.Fatal("reservation beyond byte budget should be rejected")
+	}
+	if got := OpsErrorLogQueueBytes(); got != opsErrorLogMaxQueueBytes-1 {
+		t.Fatalf("queued bytes = %d, want %d", got, opsErrorLogMaxQueueBytes-1)
+	}
+	if got := OpsErrorLogQueueLength(); got != 1 {
+		t.Fatalf("queue length = %d, want 1", got)
+	}
+}
+
+func TestEstimateOpsErrorLogJobBytesIncludesVariablePayloads(t *testing.T) {
+	base := estimateOpsErrorLogJobBytes(&service.OpsInsertErrorLogInput{})
+	message := "upstream message"
+	detail := "upstream detail"
+	events := `[{"error":"x"}]`
+	entry := &service.OpsInsertErrorLogInput{
+		ErrorBody:            strings.Repeat("x", 1024),
+		ErrorMessage:         "client error",
+		UserAgent:            "test-agent",
+		UpstreamErrorMessage: &message,
+		UpstreamErrorDetail:  &detail,
+		UpstreamErrorsJSON:   &events,
+	}
+	if got := estimateOpsErrorLogJobBytes(entry); got <= base+1024 {
+		t.Fatalf("estimated bytes = %d, expected variable payloads above %d", got, base+1024)
+	}
+}
+
 func resetOpsErrorLoggerStateForTest(t *testing.T) {
 	t.Helper()
 
@@ -119,6 +198,29 @@ func TestOpsCaptureWriterPool_ResetOnRelease(t *testing.T) {
 	require.Zero(t, reused.buf.Len(), "writer should be reset before reuse")
 }
 
+func TestOpsCaptureWriterPool_DropsLargeBuffers(t *testing.T) {
+	w := &opsCaptureWriter{}
+	w.buf.Grow(opsCaptureWriterPoolMaxRetainedCapacity + 1)
+	require.False(t, shouldPoolOpsCaptureWriter(w))
+}
+
+func TestEnqueueOpsErrorLog_SanitizesAndBoundsBodyBeforeQueue(t *testing.T) {
+	setupOpsErrorLogTestQueue(t, 1)
+	ops := service.NewOpsService(nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil)
+	secret := strings.Repeat("s", service.OpsErrorLogQueueBodyMaxBytes)
+	entry := &service.OpsInsertErrorLogInput{
+		ErrorPhase: "request",
+		ErrorType:  "api_error",
+		ErrorBody:  `{"authorization":"Bearer ` + secret + `","message":"failed"}`,
+	}
+
+	enqueueOpsErrorLog(ops, entry)
+	job := <-opsErrorLogQueue
+	require.LessOrEqual(t, len(job.entry.ErrorBody), service.OpsErrorLogQueueBodyMaxBytes)
+	require.NotContains(t, job.entry.ErrorBody, secret)
+	require.Equal(t, int64(1), OpsErrorLogSanitizedTotal())
+}
+
 func TestOpsErrorLoggerMiddleware_DoesNotBreakOuterMiddlewares(t *testing.T) {
 	gin.SetMode(gin.TestMode)
 
@@ -149,6 +251,42 @@ func setupOpsErrorLogTestQueue(t *testing.T, size int) {
 	opsErrorLogMu.Unlock()
 }
 
+func TestOpsErrorLoggerMiddleware_HardSkipsIngressRejection(t *testing.T) {
+	setupOpsErrorLogTestQueue(t, 4)
+	gin.SetMode(gin.TestMode)
+
+	settings := &ingressRejectSettingRepo{}
+	repo := &ingressRejectOpsRepo{}
+	ops := service.NewOpsService(repo, settings, nil, nil, nil, nil, nil, nil, nil, nil, nil)
+	// Construction may read unrelated runtime settings; only request-path reads matter here.
+	settings.getValueCalls = 0
+
+	router := gin.New()
+	router.Use(OpsErrorLoggerMiddleware(ops))
+	router.GET("/v1/messages", func(c *gin.Context) {
+		middleware2.MarkIngressRejected(c, middleware2.IngressRejectInvalidAPIKey)
+		c.JSON(http.StatusUnauthorized, gin.H{"code": "INVALID_API_KEY", "message": "Invalid API key"})
+	})
+
+	w := httptest.NewRecorder()
+	req := httptest.NewRequest(http.MethodGet, "/v1/messages", nil)
+	router.ServeHTTP(w, req)
+
+	require.Equal(t, http.StatusUnauthorized, w.Code)
+	require.JSONEq(t, `{"code":"INVALID_API_KEY","message":"Invalid API key"}`, w.Body.String())
+	require.Zero(t, settings.getValueCalls, "ingress rejection must bypass monitoring settings reads")
+	require.Zero(t, repo.insertCalls, "ingress rejection must bypass inserts")
+	require.Zero(t, OpsErrorLogEnqueuedTotal(), "ingress rejection must not enter the error queue")
+}
+
+func TestNormalizeOpsPersistentUserAgentBoundsAndPreservesUTF8(t *testing.T) {
+	value := strings.Repeat("a", opsErrorLogMaxUserAgentBytes-1) + "你" + strings.Repeat("b", 32)
+	got := normalizeOpsPersistentUserAgent("  " + value + "  ")
+	require.LessOrEqual(t, len(got), opsErrorLogMaxUserAgentBytes)
+	require.True(t, utf8.ValidString(got))
+	require.NotContains(t, got, "b")
+}
+
 // 就地(in-band) SSE 错误挂在已固化的 HTTP 200 流上：wire 状态码为 200，
 // 常规 status>=400 采集路径不会触发。logOpsStreamError 必须据 MarkOpsStreamError
 // 补记一条错误日志，且用 IntendedStatus(429) 分级、StatusCode 仍记 wire 的 200。

backend/internal/handler/ops_ingress_reject_capture_test.go

@@ -0,0 +1,25 @@
+package handler
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	middleware2 "github.com/Wei-Shaw/sub2api/internal/server/middleware"
+	"github.com/gin-gonic/gin"
+	"github.com/stretchr/testify/require"
+)
+
+func TestOpsCaptureWriterDoesNotCopyIngressRejectBody(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	context, _ := gin.CreateTestContext(httptest.NewRecorder())
+	writer := acquireOpsCaptureWriter(context.Writer)
+	defer releaseOpsCaptureWriter(writer)
+	writer.ctx = context
+	context.Writer = writer
+	middleware2.MarkIngressRejected(context, middleware2.IngressRejectInvalidAPIKey)
+	context.Status(http.StatusUnauthorized)
+	_, err := context.Writer.WriteString(`{"code":"INVALID_API_KEY","message":"Invalid API key"}`)
+	require.NoError(t, err)
+	require.Zero(t, writer.buf.Len())
+}