seewe/sub2api
1
0
Fork 0
mirror of https://github.com/Wei-Shaw/sub2api.git synced 2026-10-14 03:52:26 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix: 过滤入口拒绝日志并强化鉴权边界

Browse Source
This commit is contained in:
benjamin
2026-07-18 00:11:18 +08:00
parent 57914967cb
commit b92bbf0299
101 changed files with 5872 additions and 828 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/internal/handler/ops_error_logger.go
+130 -75
View File
@@ -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)
}
}
@@ -1694,11 +1752,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)