diff --git a/firewall/ipset_windows.go b/firewall/ipset_windows.go index 6ac7e95..d7b7b2f 100644 --- a/firewall/ipset_windows.go +++ b/firewall/ipset_windows.go @@ -3,9 +3,12 @@ package firewall import ( + "SamWaf/common/zlog" + "context" "fmt" "os/exec" "strings" + "time" ) // 说明：Windows 无 ipset 概念，但 netsh 单条规则的 remoteip= 支持逗号分隔多个 IP/CIDR。 @@ -18,8 +21,15 @@ const ( setRulePrefix = "SamWAF_Set_" // maxRemoteIPChars 单条规则 remoteip= 值的字符上限(留足命令行余量) maxRemoteIPChars = 3000 - // maxShardScan DestroyIPSet 时扫描删除分片规则的序号上限(兜底防死循环) - maxShardScan = 200000 + // maxShardScan DestroyIPSet 时扫描删除分片规则的序号上限(兜底防死循环)。 + // 单分片 maxRemoteIPChars=3000 字符 ≈ 200 条 IP，4096 个分片 ≈ 80 万条，远超实际订阅规模。 + maxShardScan = 4096 + // destroyMissStreak 连续多少个序号"未命中"就认为分片已删完 + destroyMissStreak = 3 + // destroyBudget DestroyIPSet 整体时间预算，超了就放弃剩余扫描(防止拖住调用方持有的锁) + destroyBudget = 30 * time.Second + // netshCmdTimeout 单条 netsh 命令的超时，防个别调用挂死 + netshCmdTimeout = 5 * time.Second ) // SupportsIPSet 报告当前环境是否可用批量封禁(Windows 下等价于系统防火墙可用，带 30s 缓存) @@ -68,21 +78,66 @@ func (fw *FireWallEngine) FlushIPSet(setName string) error { return fw.DestroyIPSet(setName) } -// DestroyIPSet 逐序号删除该集合的所有分片规则，直到某序号无匹配为止 +// DestroyIPSet 逐序号删除该集合的所有分片规则，直到连续若干个序号都删不动为止。 +// +// 停止条件刻意不再依赖 netsh 的本地化提示文案：老实现只认英文 +// "No rules match the specified criteria" 和简中"没有与指定标准相匹配的规则"， +// 一旦 netsh 输出别的内容(非中英文系统、非管理员运行时的"请求的操作需要提升"、 +// 防火墙服务异常等)，break 就永远不会触发，会 fork 二十万次 netsh 空转数小时， +// 并把调用方持有的同步锁一起焊死。现在改为：**任何一次"没有删成功"都算未命中**， +// 连续 destroyMissStreak 次未命中即停，另加整体时间预算兜底。宁可少删，绝不空转。 func (fw *FireWallEngine) DestroyIPSet(setName string) error { + // 系统防火墙不可用时一次 fork 都不做(netsh 缺失 / 防火墙未启用) + if !fw.SupportsIPSet() { + return nil + } + + start := time.Now() + deleted, miss := 0, 0 for seq := 0; seq < maxShardScan; seq++ { - ruleName := setShardRuleName(setName, seq) - cmd := exec.Command("netsh", "advfirewall", "firewall", "delete", "rule", "name="+ruleName) - _, output := fw.executeCommand(cmd) - if strings.Contains(output, "No rules match the specified criteria") || - strings.Contains(output, "没有与指定标准相匹配的规则") { - // 连续遇到无匹配即认为分片已删完 + if time.Since(start) > destroyBudget { + zlog.Warn("清理防火墙分片规则超出时间预算，已提前结束", + "set", setName, "scanned", seq, "deleted", deleted, + "elapsed", time.Since(start).Round(time.Millisecond).String()) + break + } + if fw.deleteShardRule(setName, seq) { + deleted++ + miss = 0 + continue + } + miss++ + if miss >= destroyMissStreak { break } } + if deleted > 0 { + zlog.Debug("清理防火墙分片规则完成", "set", setName, "deleted", deleted, + "elapsed", time.Since(start).Round(time.Millisecond).String()) + } return nil } +// deleteShardRule 删除单个分片规则，返回是否确实删掉了。 +// 只有 netsh 明确报告删除成功才算命中；出错、超时、输出看不懂一律当作"没有这个分片"。 +func (fw *FireWallEngine) deleteShardRule(setName string, seq int) bool { + ctx, cancel := context.WithTimeout(context.Background(), netshCmdTimeout) + defer cancel() + + ruleName := setShardRuleName(setName, seq) + cmd := exec.CommandContext(ctx, "netsh", "advfirewall", "firewall", "delete", "rule", "name="+ruleName) + err, output := fw.executeCommand(cmd) + if err != nil { + return false + } + // 已知的"没有这条规则"文案优先判定：部分环境下 netsh 会在该提示后仍打印"确定。" + if strings.Contains(output, "No rules match the specified criteria") || + strings.Contains(output, "没有与指定标准相匹配的规则") { + return false + } + return strings.Contains(output, "已删除") || strings.Contains(output, "Deleted") +} + // shardIPsByLen 把 IP/CIDR 列表按逗号拼接、累计长度不超过 maxLen 切分为多个分片字符串 func shardIPsByLen(ips []string, maxLen int) []string { var shards []string diff --git a/model/threat_ip_channel.go b/model/threat_ip_channel.go index fc8ec9b..96148ee 100644 --- a/model/threat_ip_channel.go +++ b/model/threat_ip_channel.go @@ -21,6 +21,11 @@ type ThreatIPChannel struct { LastCount int `json:"last_count"` // 上次快照收录条数 LastStatus string `gorm:"size:255" json:"last_status"` // 上次同步结果：ok 或错误摘要 Remarks string `gorm:"size:500" json:"remarks"` // 备注 + + // 以下为运行时内存态字段(gorm:"-" 不落库，无需迁移)，仅供列表接口回显"同步中"， + // 让前端能在后台拉取期间给出反馈并自动轮询，而不是点完什么都看不到。 + Syncing bool `gorm:"-" json:"syncing"` // 该渠道当前是否有同步在进行 + SyncStartedAt int64 `gorm:"-" json:"sync_started_at"` // 本次同步开始时间戳(秒)，Syncing 为 false 时无意义 } // TableName 表名 diff --git a/service/waf_service/waf_threat_ip_service.go b/service/waf_service/waf_threat_ip_service.go index f873d45..d356933 100644 --- a/service/waf_service/waf_threat_ip_service.go +++ b/service/waf_service/waf_threat_ip_service.go @@ -1,9 +1,11 @@ package waf_service import ( + "SamWaf/common/tasklog" "SamWaf/common/uuid" "SamWaf/common/zlog" "SamWaf/customtype" + "SamWaf/enums" "SamWaf/firewall" "SamWaf/global" "SamWaf/innerbean" @@ -14,8 +16,10 @@ import ( "SamWaf/waftask/threatip" "errors" "fmt" + "runtime/debug" "strings" "sync" + "sync/atomic" "time" ) @@ -23,6 +27,67 @@ import ( type WafThreatIPService struct { fw firewall.FireWallEngine syncMu sync.Mutex // 串行化同步，避免多渠道并发拉取造成内核/内存压力 + + // inflight 记录"当前正在处理(含排队等锁)的渠道 Code -> 开始时间戳(秒)"。 + // 纯内存态(重启即清空)，只用于列表接口回显"同步中"，让前端能轮询到结果。 + inflight sync.Map + // lockHolder 当前持有 syncMu 的操作描述，用于在别人拿不到锁时说清楚"是谁占着" + lockHolder atomic.Value +} + +// lockWaitOnDemand 等待 syncMu 的上限。 +// 超过就明确报错返回，而不是像以前那样无限期静默排队(用户会看到"提示已开始"却永远没结果)。 +// 取 30s：既能容纳"连着点两个渠道同步"这种正常排队，又不会让失败无限期悬着。 +const lockWaitOnDemand = 30 * time.Second + +// syncTriggerManual / syncTriggerSchedule 同步触发方式，进日志与 last_status 便于区分 +const ( + syncTriggerManual = "手动" + syncTriggerSchedule = "定时" +) + +// tryLockSync 在 wait 时间内尝试获取同步锁，holder 描述本次操作(用于让后来者知道是谁占着)。 +// 用轮询式 TryLock 而非阻塞 Lock：这条链路上任何一次卡死都会把后续所有同步永久堵住， +// 宁可明确失败并留下日志，也不要无限期静默排队。 +func (r *WafThreatIPService) tryLockSync(wait time.Duration, holder string) bool { + deadline := time.Now().Add(wait) + for { + if r.syncMu.TryLock() { + r.lockHolder.Store(holder) + return true + } + if time.Now().After(deadline) { + return false + } + time.Sleep(100 * time.Millisecond) + } +} + +// unlockSync 释放同步锁并清掉持有者标记 +func (r *WafThreatIPService) unlockSync() { + r.lockHolder.Store("") + r.syncMu.Unlock() +} + +// busyHint 描述当前占着同步锁的操作，用于错误信息与日志 +func (r *WafThreatIPService) busyHint() string { + if h, _ := r.lockHolder.Load().(string); h != "" { + return h + } + return "有其它订阅落地操作正在进行" +} + +// safeGo 起一个带 panic 兜底的后台 goroutine。 +// 以前这里是裸 go func()：一次 panic 会直接崩掉整个 SamWaf 进程，且什么线索都不留。 +func safeGo(name string, fn func()) { + go func() { + defer func() { + if e := recover(); e != nil { + zlog.Error(fmt.Sprintf("威胁情报订阅后台任务[%s]异常: %v 调试信息:%s", name, e, debug.Stack())) + } + }() + fn() + }() } var WafThreatIPServiceApp = new(WafThreatIPService) @@ -100,21 +165,28 @@ func (r *WafThreatIPService) ModifyApi(req request.WafThreatIPChannelEditReq) er code := bean.Code enable := req.Enable name := req.Name - go func() { - r.applyEnableState(code, landTarget, enable) - r.notifyOpResult("威胁情报订阅保存生效", name, nil) - }() + safeGo("保存生效:"+code, func() { + err := r.applyEnableState(code, landTarget, enable) + r.notifyOpResult("威胁情报订阅保存生效", name, err) + }) return nil } // applyEnableState 依据启用状态把某渠道落地到位：系统集合的增删 + 重建 WAF 应用层并集。 // 供编辑保存后异步调用；串行化(syncMu)避免与同步任务并发改动内核/内存态。 -func (r *WafThreatIPService) applyEnableState(code, landTarget string, enable int) { - r.syncMu.Lock() - defer r.syncMu.Unlock() +func (r *WafThreatIPService) applyEnableState(code, landTarget string, enable int) error { + start := time.Now() + if !r.tryLockSync(lockWaitOnDemand, fmt.Sprintf("渠道[%s]保存生效", code)) { + msg := "订阅落地生效跳过：" + r.busyHint() + zlog.Warn("威胁情报订阅落地生效未能获取同步锁", "channel", code, "detail", r.busyHint()) + return errors.New(msg) + } + defer r.unlockSync() + + zlog.Info("威胁情报订阅落地生效开始", "channel", code, "land", landTarget, "enable", enable) if enable == 0 { // 停用：清理系统集合(WAF 并集在下方重建时自动排除该渠道) - r.fw.DestroyIPSet(setNameForChannel(code)) + r.destroySystemSet(code) } else if landTarget == model.ThreatLandSystem || landTarget == model.ThreatLandBoth { // 启用且落地含系统层：把快照重新灌回系统集合(重启/停用会丢，需按需回灌) if r.fw.SupportsIPSet() { @@ -123,12 +195,29 @@ func (r *WafThreatIPService) applyEnableState(code, landTarget string, enable in zlog.Warn("启用渠道系统层落地失败", "channel", code, "error", err.Error()) } } + } else { + zlog.Warn("当前环境不支持 ipset 批量封禁，跳过系统层落地", "channel", code) } } else { // 启用但落地不含系统层：清理可能残留的系统集合 - r.fw.DestroyIPSet(setNameForChannel(code)) + r.destroySystemSet(code) } r.RebuildWAFUnion() + zlog.Info("威胁情报订阅落地生效完成", "channel", code, "elapsed", time.Since(start).Round(time.Millisecond).String()) + return nil +} + +// destroySystemSet 清理某渠道在系统防火墙上的集合。 +// 只在系统防火墙确实可用时才动手：纯 WAF 落地的渠道从没往系统层写过东西， +// 无条件调用只会白白 fork 一堆 iptables/netsh(Windows 上还可能长时间空转并占着同步锁)。 +func (r *WafThreatIPService) destroySystemSet(code string) { + if !r.fw.SupportsIPSet() { + zlog.Debug("系统防火墙不可用，跳过系统层集合清理", "channel", code) + return + } + if err := r.fw.DestroyIPSet(setNameForChannel(code)); err != nil { + zlog.Warn("清理系统层集合失败", "channel", code, "error", err.Error()) + } } // notifyOpResult 把某异步操作结果推送到管理端通知中心(WebSocket 实时)。 @@ -164,13 +253,18 @@ func (r *WafThreatIPService) DelApi(id string) error { return err } // 异步落地：撤系统集合 + 重建并集 - go func() { - r.syncMu.Lock() - r.fw.DestroyIPSet(setNameForChannel(code)) + safeGo("删除生效:"+code, func() { + if !r.tryLockSync(lockWaitOnDemand, fmt.Sprintf("渠道[%s]删除生效", code)) { + zlog.Warn("威胁情报订阅删除生效未能获取同步锁", "channel", code, "detail", r.busyHint()) + r.notifyOpResult("威胁情报订阅删除生效", name, errors.New("落地生效跳过："+r.busyHint()+"；下次同步或重启后会自动对齐")) + return + } + defer r.unlockSync() // 用 defer：以前是手动 Unlock，中途 panic 会永久泄漏锁，之后所有同步全卡死 + r.destroySystemSet(code) r.RebuildWAFUnion() - r.syncMu.Unlock() + zlog.Info("威胁情报订阅删除生效完成", "channel", code) r.notifyOpResult("威胁情报订阅删除生效", name, nil) - }() + }) return nil } @@ -189,10 +283,21 @@ func (r *WafThreatIPService) GetListApi(req request.WafThreatIPChannelSearchReq) if c := r.snapshotCount(list[i].Code); c >= 0 { list[i].LastCount = c } + // 回填内存态"同步中"，前端据此显示进行中并自动轮询 + list[i].Syncing, list[i].SyncStartedAt = r.syncingOf(list[i].Code) } return list, total, err } +// syncingOf 查某渠道当前是否正在同步，以及本次同步的开始时间戳 +func (r *WafThreatIPService) syncingOf(code string) (bool, int64) { + if v, ok := r.inflight.Load(code); ok { + startedAt, _ := v.(int64) + return true, startedAt + } + return false, 0 +} + // ---------------- 订阅落地只读浏览(方案三：原页"订阅来源"Tab) ---------------- // LandedChannelSummary 某渠道在指定层的落地汇总 @@ -201,10 +306,12 @@ type LandedChannelSummary struct { Name string `json:"name"` LandTarget string `json:"land_target"` Enable int `json:"enable"` - Count int `json:"count"` // 实际生效(已落地)条数：启用=快照条数，停用=0(已从防火墙/WAF移除) - SnapshotCount int `json:"snapshot_count"` // 快照收录条数(不论启用与否，供停用时提示"再启用可回灌") - LastSyncAt int64 `json:"last_sync_at"` // 上次同步时间戳(秒) - LastStatus string `json:"last_status"` // 上次同步结果 + Count int `json:"count"` // 实际生效(已落地)条数：启用=快照条数，停用=0(已从防火墙/WAF移除) + SnapshotCount int `json:"snapshot_count"` // 快照收录条数(不论启用与否，供停用时提示"再启用可回灌") + LastSyncAt int64 `json:"last_sync_at"` // 上次同步时间戳(秒) + LastStatus string `json:"last_status"` // 上次同步结果 + Syncing bool `json:"syncing"` // 当前是否有同步在进行(内存态) + SyncStartedAt int64 `json:"sync_started_at"` // 本次同步开始时间戳(秒) } // GetLandedSummary 返回落地到指定层的各渠道汇总。 @@ -226,6 +333,7 @@ func (r *WafThreatIPService) GetLandedSummary(land string) []LandedChannelSummar if ch.Enable == 0 { landed = 0 } + syncing, syncStartedAt := r.syncingOf(ch.Code) out = append(out, LandedChannelSummary{ Code: ch.Code, Name: ch.Name, @@ -235,6 +343,8 @@ func (r *WafThreatIPService) GetLandedSummary(land string) []LandedChannelSummar SnapshotCount: snapCnt, LastSyncAt: ch.LastSyncAt, LastStatus: ch.LastStatus, + Syncing: syncing, + SyncStartedAt: syncStartedAt, }) } return out @@ -308,47 +418,100 @@ func (r *WafThreatIPService) SyncByIdApi(id string) error { if err := global.GWAF_LOCAL_DB.Where("id = ?", id).First(&bean).Error; err != nil { return errors.New("记录不存在") } - go func() { - err := r.SyncChannel(bean) + zlog.Info("威胁情报订阅收到手动同步请求", "channel", bean.Code, "name", bean.Name, + "url", bean.URL, "parser", bean.ParserType, "land", bean.LandTarget) + safeGo("手动同步:"+bean.Code, func() { + err := r.syncChannelWithTrigger(bean, syncTriggerManual) r.notifyOpResult("威胁情报订阅同步", bean.Name, err) - }() + }) return nil } // SyncChannel 拉取→解析→与上次快照比对→(突变保护)→保存快照→落地(系统 ipset + WAF 并集)。 func (r *WafThreatIPService) SyncChannel(ch model.ThreatIPChannel) error { - r.syncMu.Lock() - defer r.syncMu.Unlock() + return r.syncChannelWithTrigger(ch, syncTriggerSchedule) +} - raw, err := threatip.Fetch(ch.URL) +// syncChannelWithTrigger 同 SyncChannel，额外带上触发方式(手动/定时)用于日志与状态回显。 +// +// 这条链路以前失败时是完全静默的(四个失败分支都只写 DB、不打日志，失败通知又只走一次性 WebSocket)， +// 用户点完"立即同步"只能看到"已开始"然后什么都没有。所以这里每个阶段都留了日志与耗时， +// 并把手动触发也挂进任务日志，方便在"任务日志"页直接看到全过程。 +func (r *WafThreatIPService) syncChannelWithTrigger(ch model.ThreatIPChannel, trigger string) error { + // 让本 goroutine 的 zlog 输出同时落到任务日志文件(与定时任务同一个任务名)。 + // 定时路径下 ExecuteTask 已经设过任务上下文，这里就不能重复设——否则同步完第一个渠道时 + // 的 defer Clear 会把整个 SyncAllDue 剩余过程的任务日志上下文一起清掉。 + if _, ok := tasklog.GetCurrentTask(); !ok { + tasklog.SetCurrentTask(enums.TASK_THREAT_IP_SYNC) + defer tasklog.ClearCurrentTask() + } + + start := time.Now() + + // 先标记"同步中"再去抢锁：排队等锁的这段时间前端也该看到进行中， + // 否则轮询会以为已经结束而提前收工。无论从哪个分支返回都会清掉。 + r.inflight.Store(ch.Code, start.Unix()) + defer r.inflight.Delete(ch.Code) + + if !r.tryLockSync(lockWaitOnDemand, fmt.Sprintf("渠道[%s]%s同步", ch.Code, trigger)) { + msg := fmt.Sprintf("已跳过(%s触发)：等待 %s 仍未轮到，%s", trigger, lockWaitOnDemand, r.busyHint()) + zlog.Warn("威胁情报订阅同步未能获取同步锁", "channel", ch.Code, "trigger", trigger, "detail", r.busyHint()) + r.markSyncFail(ch.Id, msg) + return errors.New(msg) + } + defer r.unlockSync() + + zlog.Info("威胁情报订阅同步开始", "channel", ch.Code, "trigger", trigger, "url", ch.URL, "parser", ch.ParserType, + "waited", time.Since(start).Round(time.Millisecond).String()) + + raw, stat, err := threatip.FetchWithStat(ch.URL) if err != nil { - r.markSyncStatus(ch.Id, "拉取失败: "+err.Error(), 0) + zlog.Error(fmt.Sprintf("威胁情报订阅拉取失败 channel=%s trigger=%s error=%s", ch.Code, trigger, err.Error())) + r.markSyncFail(ch.Id, "拉取失败: "+err.Error()) return err } + zlog.Info("威胁情报订阅拉取完成", "channel", ch.Code, "status", stat.StatusCode, + "bytes", stat.Bytes, "elapsed", stat.Elapsed.Round(time.Millisecond).String()) + + parseStart := time.Now() parseRes, err := threatip.ParseByType(ch.ParserType, strings.NewReader(string(raw)), ch.Threshold) if err != nil { - r.markSyncStatus(ch.Id, "解析失败: "+err.Error(), 0) + zlog.Error(fmt.Sprintf("威胁情报订阅解析失败 channel=%s parser=%s error=%s", ch.Code, ch.ParserType, err.Error())) + r.markSyncFail(ch.Id, "解析失败: "+err.Error()) return err } newIPs := parseRes.IPs + zlog.Info("威胁情报订阅解析完成", "channel", ch.Code, "valid", len(newIPs), "dropped", parseRes.Dropped, + "elapsed", time.Since(parseStart).Round(time.Millisecond).String()) + + // 源返回 200 但一条有效数据都没有(常见于返回了验证页/错误页)，按失败处理，避免把空快照当成功 + if len(newIPs) == 0 { + msg := fmt.Sprintf("解析结果为空(丢弃非法行 %d，响应 %d 字节)，可能订阅源返回了非预期内容或解析格式选错", parseRes.Dropped, stat.Bytes) + zlog.Error(fmt.Sprintf("威胁情报订阅解析结果为空 channel=%s detail=%s", ch.Code, msg)) + r.markSyncFail(ch.Id, msg) + return errors.New(msg) + } // 载入上次快照做突变保护与 sha 比对 oldIPs, oldSha := r.loadSnapshot(ch.Code) if len(oldIPs) > shrinkGuardMin && len(newIPs) < len(oldIPs)/shrinkGuardRatio { msg := fmt.Sprintf("疑似源异常：新快照 %d 条 远少于上次 %d 条，已跳过应用(丢弃非法行 %d)", len(newIPs), len(oldIPs), parseRes.Dropped) zlog.Warn("威胁情报订阅突变保护", "channel", ch.Code, "detail", msg) - r.markSyncStatus(ch.Id, msg, len(oldIPs)) + r.markSyncFail(ch.Id, msg) return errors.New(msg) } payload, sha, count, err := threatip.EncodeSnapshot(newIPs) if err != nil { - r.markSyncStatus(ch.Id, "快照编码失败: "+err.Error(), 0) + zlog.Error(fmt.Sprintf("威胁情报订阅快照编码失败 channel=%s error=%s", ch.Code, err.Error())) + r.markSyncFail(ch.Id, "快照编码失败: "+err.Error()) return err } if sha == oldSha && oldSha != "" { // 无变化：仅更新同步时间 - r.markSyncStatus(ch.Id, "ok(无变化)", count) + zlog.Info("威胁情报订阅内容无变化，跳过落地", "channel", ch.Code, "count", count, + "elapsed", time.Since(start).Round(time.Millisecond).String()) + r.markSyncOK(ch.Id, fmt.Sprintf("ok(无变化，%s触发，耗时%s)", trigger, time.Since(start).Round(time.Millisecond)), count) return nil } @@ -356,29 +519,36 @@ func (r *WafThreatIPService) SyncChannel(ch model.ThreatIPChannel) error { // 保存新快照(替换该渠道旧快照) if err := r.saveSnapshot(ch.Code, payload, sha, count); err != nil { - r.markSyncStatus(ch.Id, "保存快照失败: "+err.Error(), 0) + zlog.Error(fmt.Sprintf("威胁情报订阅保存快照失败 channel=%s error=%s", ch.Code, err.Error())) + r.markSyncFail(ch.Id, "保存快照失败: "+err.Error()) return err } // 落地系统防火墙(该渠道私有集合，全量重建) if ch.LandTarget == model.ThreatLandSystem || ch.LandTarget == model.ThreatLandBoth { if r.fw.SupportsIPSet() { + landStart := time.Now() if err := r.fw.RestoreIPSet(setNameForChannel(ch.Code), newIPs); err != nil { zlog.Warn("威胁情报系统层落地失败", "channel", ch.Code, "error", err.Error()) + } else { + zlog.Info("威胁情报系统层落地完成", "channel", ch.Code, "count", count, + "elapsed", time.Since(landStart).Round(time.Millisecond).String()) } } else { zlog.Warn("当前环境不支持 ipset 批量封禁，跳过系统层落地", "channel", ch.Code) } } else { - // 落地目标不含系统层：清理可能残留的系统集合 - r.fw.DestroyIPSet(setNameForChannel(ch.Code)) + // 落地目标不含系统层：清理可能残留的系统集合(系统防火墙不可用时内部会直接跳过) + r.destroySystemSet(ch.Code) } // 落地 WAF 应用层(重建全局并集) r.RebuildWAFUnion() - r.markSyncStatus(ch.Id, fmt.Sprintf("ok(+%d/-%d，丢弃%d)", len(added), len(removed), parseRes.Dropped), count) - zlog.Info("威胁情报订阅同步完成", "channel", ch.Code, "count", count, "added", len(added), "removed", len(removed)) + elapsed := time.Since(start).Round(time.Millisecond) + r.markSyncOK(ch.Id, fmt.Sprintf("ok(+%d/-%d，丢弃%d，%s触发，耗时%s)", len(added), len(removed), parseRes.Dropped, trigger, elapsed), count) + zlog.Info("威胁情报订阅同步完成", "channel", ch.Code, "trigger", trigger, "count", count, + "added", len(added), "removed", len(removed), "elapsed", elapsed.String()) return nil } @@ -405,8 +575,10 @@ func (r *WafThreatIPService) RebuildWAFUnion() { for ip := range uniq { items = append(items, ip) } + buildStart := time.Now() ipset.SetGlobalThreatMatcher(ipset.BuildMatchSet(items)) - zlog.Info("威胁情报 WAF 并集已重建", "total", len(items)) + zlog.Info("威胁情报 WAF 并集已重建", "total", len(items), + "elapsed", time.Since(buildStart).Round(time.Millisecond).String()) } // RestoreAllOnStartup 进程启动时重放：把各启用渠道的快照重新灌入系统 ipset(内存态，重启会丢)， @@ -436,15 +608,18 @@ func (r *WafThreatIPService) SyncAllDue() { var channels []model.ThreatIPChannel global.GWAF_LOCAL_DB.Where("enable = 1").Find(&channels) now := time.Now().Unix() + due := 0 for _, ch := range channels { interval := int64(defaultInt(ch.IntervalHour, 24)) * 3600 if ch.LastSyncAt > 0 && now-ch.LastSyncAt < interval { continue // 未到期 } - if err := r.SyncChannel(ch); err != nil { + due++ + if err := r.syncChannelWithTrigger(ch, syncTriggerSchedule); err != nil { zlog.Warn("威胁情报订阅同步失败", "channel", ch.Code, "error", err.Error()) } } + zlog.Info("威胁情报订阅定时检查完成", "enabled", len(channels), "due", due) } // ---------------- 内部辅助 ---------------- @@ -484,14 +659,32 @@ func (r *WafThreatIPService) saveSnapshot(code string, payload []byte, sha strin return global.GWAF_LOCAL_DB.Create(snap).Error } -// markSyncStatus 更新渠道同步状态字段 -func (r *WafThreatIPService) markSyncStatus(id, status string, count int) { - if err := global.GWAF_LOCAL_DB.Model(&model.ThreatIPChannel{}).Where("id = ?", id).Updates(map[string]interface{}{ - "LastSyncAt": time.Now().Unix(), - "LastCount": count, - "LastStatus": truncateStatus(status), - "UPDATE_TIME": customtype.JsonTime(time.Now()), - }).Error; err != nil { +// markSyncOK 同步成功后回写：刷新同步时间、收录条数与状态 +func (r *WafThreatIPService) markSyncOK(id, status string, count int) { + r.updateSyncFields(id, map[string]interface{}{ + "LastSyncAt": time.Now().Unix(), + "LastCount": count, + "LastStatus": truncateStatus(status), + }) +} + +// markSyncFail 同步失败后回写：只写状态原因。 +// +// 刻意不动 LastSyncAt、LastCount： +// - 以前失败也把 LastSyncAt 刷成当前时间，SyncAllDue 的到期判断就把失败当成了成功， +// 导致整整一个周期(默认 24h)不再重试——这正是用户说的"后面再也没成功"； +// 不刷新之后，下一次每小时的定时任务会自然重试。 +// - 以前失败还会把 LastCount 抹成 0，把上一次的真实条数丢掉。 +func (r *WafThreatIPService) markSyncFail(id, status string) { + r.updateSyncFields(id, map[string]interface{}{ + "LastStatus": truncateStatus(status), + }) +} + +// updateSyncFields 统一回写同步相关字段(自动补 UPDATE_TIME) +func (r *WafThreatIPService) updateSyncFields(id string, fields map[string]interface{}) { + fields["UPDATE_TIME"] = customtype.JsonTime(time.Now()) + if err := global.GWAF_LOCAL_DB.Model(&model.ThreatIPChannel{}).Where("id = ?", id).Updates(fields).Error; err != nil { zlog.Warn("威胁情报订阅状态回写失败", "id", id, "error", err.Error()) } } diff --git a/waftask/threatip/fetch.go b/waftask/threatip/fetch.go index 1f422e9..c251c92 100644 --- a/waftask/threatip/fetch.go +++ b/waftask/threatip/fetch.go @@ -4,6 +4,7 @@ import ( "fmt" "io" "net/http" + "strings" "time" ) @@ -13,25 +14,66 @@ var fetchClient = &http.Client{Timeout: 2 * time.Minute} // maxFetchBytes 单次拉取的响应体上限(防超大/投毒响应耗尽内存)：64MB const maxFetchBytes = 64 * 1024 * 1024 +// bodyPeekLen 非 200 时截取响应正文做摘要的长度(用于区分"真的挂了"与"返回了HTML错误页/验证页") +const bodyPeekLen = 200 + +// FetchStat 一次拉取的可观测信息，供调用方打日志/回写状态。 +type FetchStat struct { + StatusCode int // HTTP 状态码，0 表示连接阶段就失败 + Bytes int // 实际读取到的字节数 + Elapsed time.Duration // 整体耗时(含连接、TLS、读完 body) +} + // Fetch 拉取订阅源原始内容。限制最大读取字节数，返回原始 body 字节。 // 注意：调用方需确保 url 已登记进对外访问白名单(见 CLAUDE.md)，且订阅默认关闭、由用户显式启用。 func Fetch(url string) ([]byte, error) { + data, _, err := FetchWithStat(url) + return data, err +} + +// FetchWithStat 与 Fetch 相同，但额外返回本次拉取的状态码/字节数/耗时。 +// 失败时错误信息里会带上 URL、耗时与(非 200 时的)响应正文摘要——这条链路以前失败是完全静默的， +// 用户只能看到"没反应"，所以这里刻意把上下文塞足，便于日志与"上次状态"里直接看出原因。 +func FetchWithStat(url string) ([]byte, FetchStat, error) { + start := time.Now() + stat := FetchStat{} + req, err := http.NewRequest(http.MethodGet, url, nil) if err != nil { - return nil, err + stat.Elapsed = time.Since(start) + return nil, stat, fmt.Errorf("构造请求失败(url=%s): %v", url, err) } req.Header.Set("User-Agent", "SamWaf-ThreatIP") + resp, err := fetchClient.Do(req) if err != nil { - return nil, err + stat.Elapsed = time.Since(start) + return nil, stat, fmt.Errorf("请求失败(url=%s, 耗时=%s): %v", url, stat.Elapsed.Round(time.Millisecond), err) } defer resp.Body.Close() + stat.StatusCode = resp.StatusCode + if resp.StatusCode != http.StatusOK { - return nil, fmt.Errorf("拉取失败，HTTP 状态码: %d", resp.StatusCode) + peek, _ := io.ReadAll(io.LimitReader(resp.Body, bodyPeekLen)) + stat.Elapsed = time.Since(start) + return nil, stat, fmt.Errorf("HTTP 状态码 %d(url=%s, 耗时=%s, 正文摘要=%q)", + resp.StatusCode, url, stat.Elapsed.Round(time.Millisecond), summarize(peek)) } + data, err := io.ReadAll(io.LimitReader(resp.Body, maxFetchBytes)) + stat.Bytes = len(data) + stat.Elapsed = time.Since(start) if err != nil { - return nil, err + return nil, stat, fmt.Errorf("读取响应失败(url=%s, 已读=%d字节, 耗时=%s): %v", + url, stat.Bytes, stat.Elapsed.Round(time.Millisecond), err) } - return data, nil + return data, stat, nil +} + +// summarize 把响应片段压成单行摘要(去换行、压空白)，避免污染日志与 last_status +func summarize(b []byte) string { + s := strings.TrimSpace(string(b)) + s = strings.ReplaceAll(s, "\r", " ") + s = strings.ReplaceAll(s, "\n", " ") + return strings.Join(strings.Fields(s), " ") }