diff --git a/common/config_struct.go b/common/config_struct.go index 10a1f24..9f2183f 100644 --- a/common/config_struct.go +++ b/common/config_struct.go @@ -25,6 +25,7 @@ type Config struct { Timeout time.Duration // 通用超时 TimeoutExplicit bool // 用户显式指定了 -time ThreadNum int // 主线程数 + ThreadCeiling int // 线程数上限（自适应池允许的最大值） ThreadNumExplicit bool // 用户显式指定了 -t ModuleThreadNum int // 模块线程数 ModuleThreadNumExplicit bool // 用户显式指定了 -mt @@ -39,6 +40,7 @@ type Config struct { AliveOnly bool // 仅存活检测 MaxRetries int // 最大重试次数 MaxRetriesExplicit bool // 用户显式指定了 -retry + DetectedNetworkEnv int // 探测到的网络环境（来自 core.NetworkEnv） // 高级功能（从AdvancedConfig合并） Shellcode string // Shellcode @@ -180,6 +182,7 @@ func NewConfig() *Config { // 高频字段 - 使用默认常量 Timeout: time.Duration(DefaultTimeout) * time.Second, ThreadNum: DefaultThreadNum, + ThreadCeiling: DefaultThreadNum, ModuleThreadNum: 10, DisableBrute: false, DisablePing: false, diff --git a/common/i18n/locales/en.yaml b/common/i18n/locales/en.yaml index e9a4007..bbdc267 100644 --- a/common/i18n/locales/en.yaml +++ b/common/i18n/locales/en.yaml @@ -547,6 +547,8 @@ adaptive_pool_increase: other: "Concurrency adjusted: {{.Arg1}} -> {{.Arg2}} (network healthy)" adaptive_pool_slowstart_exit: other: "Slow start exit: current {{.Arg1}} (congestion detected)" +adaptive_pool_wait_timeout: + other: "Thread pool wait timed out (10 minutes), forcing exit" net_probe_result: other: "Network probe: {{.Arg1}}, RTT {{.Arg2}}ms, loss {{.Arg3}}%, concurrency {{.Arg4}}/{{.Arg5}}" net_env_lan: diff --git a/common/i18n/locales/zh.yaml b/common/i18n/locales/zh.yaml index 4fd75a2..b4b1b33 100644 --- a/common/i18n/locales/zh.yaml +++ b/common/i18n/locales/zh.yaml @@ -547,6 +547,8 @@ adaptive_pool_increase: other: "并发调整: {{.Arg1}} -> {{.Arg2}} (网络健康)" adaptive_pool_slowstart_exit: other: "慢启动退出: 当前 {{.Arg1}} (检测到拥塞)" +adaptive_pool_wait_timeout: + other: "线程池等待超时（10分钟），强制退出" net_probe_result: other: "网络探测: {{.Arg1}}, RTT {{.Arg2}}ms, 丢包 {{.Arg3}}%, 并发 {{.Arg4}}/{{.Arg5}}" net_env_lan: diff --git a/core/adaptive_pool.go b/core/adaptive_pool.go index a4a7933..8bec02f 100644 --- a/core/adaptive_pool.go +++ b/core/adaptive_pool.go @@ -35,6 +35,9 @@ type AdaptivePool struct { pool *ants.PoolWithFunc metrics *ScanMetrics + // 网络环境（影响健康评估阈值） + networkEnv NetworkEnv + // 并发控制 target int32 // 探测推荐的目标值 ceiling int32 // 绝对上限（用户指定或探测推荐） @@ -57,7 +60,7 @@ type AdaptivePool struct { // target: 目标并发数（来自 NetworkProfile.RecommendConcurrency） // ceiling: 最大并发上限 // metrics: 共享的扫描度量（scanSinglePort 写入，pool 读取） -func NewAdaptivePool(target, ceiling int, fn func(interface{}), metrics *ScanMetrics) (*AdaptivePool, error) { +func NewAdaptivePool(target, ceiling int, fn func(interface{}), metrics *ScanMetrics, env ...NetworkEnv) (*AdaptivePool, error) { // 慢启动初始值：target 的 25%，但不低于 10 initial := target / 4 if initial < 10 { @@ -72,9 +75,15 @@ func NewAdaptivePool(target, ceiling int, fn func(interface{}), metrics *ScanMet return nil, err } + netEnv := EnvWAN + if len(env) > 0 { + netEnv = env[0] + } + return &AdaptivePool{ pool: pool, metrics: metrics, + networkEnv: netEnv, target: int32(target), ceiling: int32(ceiling), currentSize: int32(initial), @@ -110,6 +119,10 @@ func (ap *AdaptivePool) adjust() { return } + // RTT 漂移微调：fast EMA 远高于 slow EMA 说明延迟持续恶化 + // 压低 target 让 AIMD 的天花板跟着降，而不是只靠乘性减 + ap.maybeReduceTarget() + current := int(atomic.LoadInt32(&ap.currentSize)) target := int(atomic.LoadInt32(&ap.target)) ceiling := int(atomic.LoadInt32(&ap.ceiling)) @@ -215,23 +228,61 @@ func (ap *AdaptivePool) assessHealth() HealthSignal { exhaustRate := float64(deltaExhausted) / float64(deltaTotal) rttRatio := ap.metrics.RTTRatio() - // 多信号综合判断 + // 阈值根据网络环境调整：内网收紧，公网放宽 + var congestExhaust, stressExhaust, congestRTT, stressRTT, goodRTT float64 + switch ap.networkEnv { + case EnvLAN: + congestExhaust, stressExhaust = 0.08, 0.03 + congestRTT, stressRTT, goodRTT = 1.8, 1.4, 1.15 + case EnvWAN: + congestExhaust, stressExhaust = 0.15, 0.05 + congestRTT, stressRTT, goodRTT = 2.5, 1.8, 1.3 + default: // Internet / Slow + congestExhaust, stressExhaust = 0.25, 0.10 + congestRTT, stressRTT, goodRTT = 3.5, 2.5, 1.5 + } + switch { - case exhaustRate > 0.15: + case exhaustRate > congestExhaust: return HealthCongested - case rttRatio > 2.5: + case rttRatio > congestRTT: return HealthCongested - case exhaustRate > 0.05: + case exhaustRate > stressExhaust: return HealthStressed - case rttRatio > 1.8: + case rttRatio > stressRTT: return HealthStressed - case exhaustRate < 0.01 && rttRatio < 1.3: + case exhaustRate < 0.01 && rttRatio < goodRTT: return HealthGood default: return HealthOK } } +// maybeReduceTarget 当 RTT 持续恶化时压低 target +// 不低于 ceiling 的 20%，避免过度收缩 +func (ap *AdaptivePool) maybeReduceTarget() { + rttRatio := ap.metrics.RTTRatio() + if rttRatio <= 3.0 { + return + } + + target := atomic.LoadInt32(&ap.target) + ceiling := atomic.LoadInt32(&ap.ceiling) + minTarget := ceiling / 5 + if minTarget < 10 { + minTarget = 10 + } + + // 压低 10% + newTarget := int32(float64(target) * 0.9) + if newTarget < minTarget { + newTarget = minTarget + } + if newTarget < target { + atomic.StoreInt32(&ap.target, newTarget) + } +} + func (ap *AdaptivePool) tune(newSize int) { ap.pool.Tune(newSize) atomic.StoreInt32(&ap.currentSize, int32(newSize)) @@ -246,9 +297,16 @@ func (ap *AdaptivePool) Cap() int { return int(atomic.LoadInt32(&ap.currentSize) // Release 释放线程池 func (ap *AdaptivePool) Release() { ap.pool.Release() } -// Wait 等待所有任务完成 +// Wait 等待所有任务完成（最多等待 10 分钟） func (ap *AdaptivePool) Wait() { + deadline := time.After(10 * time.Minute) for ap.pool.Running() > 0 { - time.Sleep(10 * time.Millisecond) + select { + case <-deadline: + common.LogError(i18n.Tr("adaptive_pool_wait_timeout")) + return + default: + time.Sleep(10 * time.Millisecond) + } } } diff --git a/core/base_scan_strategy.go b/core/base_scan_strategy.go index c553f4a..9cd1ddc 100644 --- a/core/base_scan_strategy.go +++ b/core/base_scan_strategy.go @@ -86,6 +86,11 @@ func (b *BaseScanStrategy) IsPluginApplicableByName(pluginName string, targetHos return b.isPluginPassesFilterType(pluginName, isCustomMode, config) } + // -full 模式下，web 插件对所有开放端口生效（跳过 IsMarkedWebService 检查） + if config.POC.Full && b.isWebPlugin(pluginName) { + return b.isPluginPassesFilterType(pluginName, isCustomMode, config) + } + // 检查端口匹配和过滤器类型 return b.isPluginApplicableToPortWithHost(pluginName, targetHost, targetPort) && b.isPluginPassesFilterType(pluginName, isCustomMode, config) } diff --git a/core/edge_cases_test.go b/core/edge_cases_test.go index 12983fa..3c60bb7 100644 --- a/core/edge_cases_test.go +++ b/core/edge_cases_test.go @@ -23,24 +23,24 @@ func TestComputeRetries_EdgeCases(t *testing.T) { {0.0, 1, 1, "精确零"}, {0.001, 1, 1, "精确边界 0.001"}, {0.0009, 1, 1, "低于 0.001 边界"}, - {0.0011, 1, 6, "高于 0.001 边界"}, - {0.95, 6, 6, "精确边界 0.95"}, - {0.949, 1, 6, "低于 0.95 边界"}, - {0.951, 6, 6, "高于 0.95 边界"}, - {1.0, 6, 6, "精确 1.0"}, - {1.5, 6, 6, "超过 1.0"}, - {100.0, 6, 6, "极大值"}, + {0.0011, 1, 5, "高于 0.001 边界"}, + {0.95, 5, 5, "精确边界 0.95"}, + {0.949, 1, 5, "低于 0.95 边界"}, + {0.951, 5, 5, "高于 0.95 边界"}, + {1.0, 5, 5, "精确 1.0"}, + {1.5, 5, 5, "超过 1.0"}, + {100.0, 5, 5, "极大值"}, {math.SmallestNonzeroFloat64, 1, 1, "最小正浮点数"}, } for _, tt := range tests { t.Run(tt.desc, func(t *testing.T) { - got := computeRetries(tt.lossRate) + got := computeRetries(tt.lossRate, EnvWAN) if got < tt.wantMin || got > tt.wantMax { t.Errorf("computeRetries(%v) = %d, want [%d, %d]", tt.lossRate, got, tt.wantMin, tt.wantMax) } - if got < 1 || got > 6 { + if got < 1 || got > 5 { t.Errorf("computeRetries(%v) = %d, 超出 [1,6] 范围", tt.lossRate, got) } }) @@ -50,8 +50,8 @@ func TestComputeRetries_EdgeCases(t *testing.T) { func TestComputeRetries_NaN_Inf(t *testing.T) { // 确保不 panic for _, v := range []float64{math.NaN(), math.Inf(1), math.Inf(-1)} { - got := computeRetries(v) - if got < 1 || got > 6 { + got := computeRetries(v, EnvWAN) + if got < 1 || got > 5 { t.Errorf("computeRetries(%v) = %d, 超出 [1,6] 范围", v, got) } } diff --git a/core/env_profiler.go b/core/env_profiler.go index 40adfe3..26130d9 100644 --- a/core/env_profiler.go +++ b/core/env_profiler.go @@ -38,8 +38,19 @@ func (ep *EnvironmentProfile) TuneConfig(config *common.Config, session *common. net := &ep.Net sys := &ep.System - // ---------- ThreadNum ---------- - // 已在 AdaptivePool 层处理（ProbeNetwork + AIMD），这里不重复 + // ---------- NetworkEnv ---------- + config.DetectedNetworkEnv = int(net.Env) + + // ---------- ThreadNum / ThreadCeiling ---------- + if !isExplicit(config, "t") { + target, ceiling := net.RecommendConcurrency(config.ThreadNum, config.ThreadNumExplicit) + old := config.ThreadNum + config.ThreadNum = target + config.ThreadCeiling = ceiling + session.LogDebug(fmt.Sprintf("ThreadNum: %d -> %d, Ceiling: %d (env=%s)", old, target, ceiling, net.Env)) + } else { + config.ThreadCeiling = config.ThreadNum + } // ---------- Timeout ---------- // 公式: median_rtt + 4 * stddev，下限 1s，上限 10s @@ -65,8 +76,7 @@ func (ep *EnvironmentProfile) TuneConfig(config *common.Config, session *common. // 单个服务的连接能力远低于 TCP SYN 扫描 // 公网服务通常有限流（MaxStartups 等），并发过高适得其反 if !isExplicit(config, "mt") { - target, _ := net.RecommendConcurrency(config.ThreadNum, config.ThreadNumExplicit) - computed := target / 30 + computed := config.ThreadNum / 30 computed = clampInt(computed, 5, 50) // 高丢包环境进一步压低，避免大量连接被丢弃浪费 @@ -79,7 +89,7 @@ func (ep *EnvironmentProfile) TuneConfig(config *common.Config, session *common. old := config.ModuleThreadNum config.ModuleThreadNum = computed - session.LogDebug(fmt.Sprintf("ModuleThreadNum: %d -> %d (target_concurrency=%d)", old, computed, target)) + session.LogDebug(fmt.Sprintf("ModuleThreadNum: %d -> %d (threadNum=%d)", old, computed, config.ThreadNum)) } // ---------- MaxRetries ---------- @@ -88,7 +98,7 @@ func (ep *EnvironmentProfile) TuneConfig(config *common.Config, session *common. // 例: 丢包率 5% → N=2, 丢包率 20% → N=3, 丢包率 50% → N=7 // 下限 1（零丢包也至少试一次），上限 6（避免对不可达目标死磕） if !isExplicit(config, "retry") && net.Samples > 0 { - computed := computeRetries(net.LossRate) + computed := computeRetries(net.LossRate, net.Env) old := config.MaxRetries config.MaxRetries = computed session.LogDebug(fmt.Sprintf("MaxRetries: %d -> %d (loss_rate=%.2f%%)", old, computed, net.LossRate*100)) @@ -135,22 +145,40 @@ func (ep *EnvironmentProfile) TuneConfig(config *common.Config, session *common. session.LogInfo(i18n.Tr("env_fd_limit", config.ThreadNum, maxConcurrency, sys.FDLimit)) config.ThreadNum = maxConcurrency } + if config.ThreadCeiling > maxConcurrency { + config.ThreadCeiling = maxConcurrency + } } } -// computeRetries 基于丢包率计算重试次数 -// 目标：重试 N 次后仍全部失败的概率 < 1% -func computeRetries(lossRate float64) int { +// computeRetries 基于丢包率和网络环境计算重试次数 +// 内网丢包异常，用更严格的目标概率（0.5%）和更低上限 +// 公网/慢速丢包常见，放宽目标概率（2%）和更高上限 +func computeRetries(lossRate float64, env NetworkEnv) int { if lossRate <= 0.001 { - return 1 // 几乎无丢包 + return 1 } + + var targetProb float64 + var maxRetries int + switch env { + case EnvLAN: + targetProb = 0.005 + maxRetries = 4 + case EnvWAN: + targetProb = 0.01 + maxRetries = 5 + default: + targetProb = 0.02 + maxRetries = 6 + } + if lossRate >= 0.95 { - return 6 // 上限 + return maxRetries } - // P(N次全失败) = lossRate^N < 0.01 - // N > log(0.01) / log(lossRate) - n := math.Ceil(math.Log(0.01) / math.Log(lossRate)) - return clampInt(int(n), 1, 6) + // P(N次全失败) = lossRate^N < targetProb + n := math.Ceil(math.Log(targetProb) / math.Log(lossRate)) + return clampInt(int(n), 1, maxRetries) } // computeICMPRate 基于环境计算 ICMP 发包速率 diff --git a/core/env_profiler_test.go b/core/env_profiler_test.go index f0e197e..477563e 100644 --- a/core/env_profiler_test.go +++ b/core/env_profiler_test.go @@ -25,15 +25,15 @@ func TestComputeRetries(t *testing.T) { {0.10, 2, 3, "10% 丢包: ceil(log(0.01)/log(0.1))=2, 但边界取 ceil 可能是 3"}, {0.20, 3, 3, "20% 丢包: 0.2^3=0.008 < 0.01"}, {0.30, 3, 4, "30% 丢包"}, - {0.50, 6, 6, "50% 丢包: ceil(log(0.01)/log(0.5))=7 但上限 6"}, - {0.80, 6, 6, "80% 丢包: 需要很多次但上限 6"}, - {0.95, 6, 6, "95% 丢包: 触顶"}, - {1.0, 6, 6, "100% 丢包: 触顶"}, + {0.50, 5, 5, "50% 丢包: ceil(log(0.01)/log(0.5))=7 但上限 5"}, + {0.80, 5, 5, "80% 丢包: 需要很多次但上限 5"}, + {0.95, 5, 5, "95% 丢包: 触顶"}, + {1.0, 5, 5, "100% 丢包: 触顶"}, } for _, tt := range tests { t.Run(tt.desc, func(t *testing.T) { - got := computeRetries(tt.lossRate) + got := computeRetries(tt.lossRate, EnvWAN) if got < tt.wantMin || got > tt.wantMax { t.Errorf("computeRetries(%.2f) = %d, want [%d, %d]", tt.lossRate, got, tt.wantMin, tt.wantMax) diff --git a/core/network_profiler.go b/core/network_profiler.go index 5aba343..228343a 100644 --- a/core/network_profiler.go +++ b/core/network_profiler.go @@ -95,7 +95,7 @@ func (p *NetworkProfile) RecommendConcurrency(userThreadNum int, explicit bool) } // probePorts 探测用的端口列表（高响应率的常见端口） -var probePorts = []int{80, 443, 22} +var probePorts = []int{80, 443, 22, 445, 8080, 3389, 21, 8443} func networkProbeAddress(host string, port int) string { return net.JoinHostPort(host, strconv.Itoa(port)) diff --git a/core/optimization_integration_test.go b/core/optimization_integration_test.go new file mode 100644 index 0000000..b97af6b --- /dev/null +++ b/core/optimization_integration_test.go @@ -0,0 +1,549 @@ +package core + +import ( + "sync/atomic" + "testing" + "time" +) + +// ============================================================================= +// 优化 1：target/ceiling 分离 +// ============================================================================= + +func TestOpt1_TargetCeilingSeparation_TuneConfig(t *testing.T) { + config := makeDefaultConfig() + session := makeTestSession(config) + + ep := &EnvironmentProfile{ + Net: NetworkProfile{ + Env: EnvLAN, + RTTMedian: 1 * time.Millisecond, + RTTStddev: 500 * time.Microsecond, + LossRate: 0.0, + Samples: 30, + }, + System: SystemProfile{FDLimit: 65536, NumCPU: 8}, + } + + ep.TuneConfig(config, session) + + if config.ThreadCeiling <= 0 { + t.Fatalf("ThreadCeiling 未被设置: %d", config.ThreadCeiling) + } + + // 内网 factor=1.5，非显式 → target=ceiling=recommended + // 但 ceiling 应该 >= target + if config.ThreadCeiling < config.ThreadNum { + t.Errorf("Ceiling(%d) < ThreadNum(%d)", config.ThreadCeiling, config.ThreadNum) + } + + t.Logf("target=%d, ceiling=%d", config.ThreadNum, config.ThreadCeiling) +} + +func TestOpt1_TargetCeilingSeparation_ExplicitT(t *testing.T) { + config := makeDefaultConfig() + config.ThreadNum = 200 + config.ThreadNumExplicit = true + session := makeTestSession(config) + + ep := &EnvironmentProfile{ + Net: NetworkProfile{ + Env: EnvInternet, + RTTMedian: 100 * time.Millisecond, + RTTStddev: 30 * time.Millisecond, + LossRate: 0.0, + Samples: 20, + }, + System: SystemProfile{FDLimit: 65536, NumCPU: 8}, + } + + ep.TuneConfig(config, session) + + // 用户显式指定 -t → ceiling = threadNum = 200 + if config.ThreadCeiling != 200 { + t.Errorf("显式 -t 200: ceiling=%d, want 200", config.ThreadCeiling) + } + if config.ThreadNum != 200 { + t.Errorf("显式 -t 200: threadNum=%d, want 200", config.ThreadNum) + } +} + +func TestOpt1_PoolUsesCeiling(t *testing.T) { + metrics := &ScanMetrics{} + target, ceiling := 50, 200 + + pool, err := NewAdaptivePool(target, ceiling, func(interface{}) {}, metrics) + if err != nil { + t.Fatalf("创建池失败: %v", err) + } + defer pool.Release() + + pool.inSlowStart = false + pool.tune(target) + + // 注入健康 metrics 让池增长 + for i := 0; i < 200; i++ { + metrics.RecordConnect(time.Millisecond) + } + + // 多次 adjust，池应能增长超过 target 但不超过 ceiling + for i := 0; i < 30; i++ { + pool.lastCheck.Store(0) + pool.adjust() + } + + finalCap := pool.Cap() + if finalCap <= target { + t.Errorf("池应能超过 target(%d): cap=%d", target, finalCap) + } + if finalCap > ceiling { + t.Errorf("池不应超过 ceiling(%d): cap=%d", ceiling, finalCap) + } + + t.Logf("target=%d, ceiling=%d, finalCap=%d", target, ceiling, finalCap) +} + +func TestOpt1_FDLimitConstraintsBothFields(t *testing.T) { + config := makeDefaultConfig() + config.ThreadNum = 1000 + session := makeTestSession(config) + + ep := &EnvironmentProfile{ + Net: NetworkProfile{ + Env: EnvLAN, + RTTMedian: 1 * time.Millisecond, + RTTStddev: 500 * time.Microsecond, + LossRate: 0.0, + Samples: 30, + }, + System: SystemProfile{FDLimit: 256, NumCPU: 4}, + } + + ep.TuneConfig(config, session) + + maxFD := 256 * 6 / 10 + if config.ThreadNum > maxFD { + t.Errorf("ThreadNum(%d) 超过 fd 限制(%d)", config.ThreadNum, maxFD) + } + if config.ThreadCeiling > maxFD { + t.Errorf("ThreadCeiling(%d) 超过 fd 限制(%d)", config.ThreadCeiling, maxFD) + } +} + +// ============================================================================= +// 优化 2：RTT 漂移微调 target +// ============================================================================= + +func TestOpt2_RTTDriftReducesTarget(t *testing.T) { + metrics := &ScanMetrics{} + pool, err := NewAdaptivePool(200, 400, func(interface{}) {}, metrics) + if err != nil { + t.Fatalf("创建池失败: %v", err) + } + defer pool.Release() + + pool.inSlowStart = false + pool.tune(200) + + // 建立基线：slow EMA 锚定在 1ms 附近 + for i := 0; i < 500; i++ { + metrics.RecordConnect(1 * time.Millisecond) + } + + origTarget := atomic.LoadInt32(&pool.target) + + // RTT 突增到 100ms（100 倍），大量喂入让 fast EMA 拉开差距 + for i := 0; i < 1000; i++ { + metrics.RecordConnect(100 * time.Millisecond) + } + + ratio := metrics.RTTRatio() + t.Logf("RTT ratio after spike: %.2f", ratio) + + if ratio <= 3.0 { + t.Skipf("RTT ratio=%.2f，EMA 差距不够大，跳过", ratio) + } + + // 需要足够的新 metrics 让 assessHealth 的 deltaTotal >= 30 + for i := 0; i < 50; i++ { + metrics.RecordConnect(100 * time.Millisecond) + } + + // 多次 adjust 触发 maybeReduceTarget + for i := 0; i < 10; i++ { + pool.lastCheck.Store(0) + pool.prevSnapshot = MetricsSnapshot{} // 重置快照让 delta 足够 + pool.adjust() + } + + newTarget := atomic.LoadInt32(&pool.target) + if newTarget >= origTarget { + t.Errorf("RTT 漂移后 target 应降低: %d -> %d (ratio=%.2f)", origTarget, newTarget, ratio) + } + + // 不应低于 ceiling/5 + minTarget := atomic.LoadInt32(&pool.ceiling) / 5 + if newTarget < minTarget { + t.Errorf("target(%d) 低于下限(%d)", newTarget, minTarget) + } + + t.Logf("RTT drift: ratio=%.2f, target %d -> %d (min=%d)", ratio, origTarget, newTarget, minTarget) +} + +func TestOpt2_NoReductionWhenStable(t *testing.T) { + metrics := &ScanMetrics{} + pool, err := NewAdaptivePool(200, 400, func(interface{}) {}, metrics) + if err != nil { + t.Fatalf("创建池失败: %v", err) + } + defer pool.Release() + + pool.inSlowStart = false + pool.tune(200) + + // 稳定 RTT + for i := 0; i < 200; i++ { + metrics.RecordConnect(10 * time.Millisecond) + } + + origTarget := atomic.LoadInt32(&pool.target) + + for i := 0; i < 10; i++ { + pool.lastCheck.Store(0) + pool.adjust() + } + + newTarget := atomic.LoadInt32(&pool.target) + if newTarget != origTarget { + t.Errorf("稳定 RTT 不应改变 target: %d -> %d", origTarget, newTarget) + } +} + +// ============================================================================= +// 优化 3：assessHealth 阈值跟 NetworkEnv 关联 +// ============================================================================= + +func TestOpt3_LANTighterThresholds(t *testing.T) { + metrics := &ScanMetrics{} + pool, err := NewAdaptivePool(100, 100, func(interface{}) {}, metrics, EnvLAN) + if err != nil { + t.Fatalf("创建池失败: %v", err) + } + defer pool.Release() + + pool.inSlowStart = false + pool.tune(100) + + // 10% exhaust rate — 对 LAN 来说应该是 Congested（阈值 8%） + for i := 0; i < 100; i++ { + if i < 10 { + metrics.RecordExhausted() + } else { + metrics.RecordConnect(time.Millisecond) + } + } + + pool.lastCheck.Store(0) + pool.adjust() + + if pool.Cap() >= 100 { + t.Errorf("LAN 10%% exhaust 应触发降速: cap=%d", pool.Cap()) + } + + t.Logf("LAN tight threshold: cap=%d (from 100)", pool.Cap()) +} + +func TestOpt3_InternetLooseThresholds(t *testing.T) { + metrics := &ScanMetrics{} + pool, err := NewAdaptivePool(100, 100, func(interface{}) {}, metrics, EnvInternet) + if err != nil { + t.Fatalf("创建池失败: %v", err) + } + defer pool.Release() + + pool.inSlowStart = false + pool.tune(100) + + // 10% exhaust rate — 对 Internet 来说不算 Congested（阈值 25%），应是 Stressed + for i := 0; i < 100; i++ { + if i < 10 { + metrics.RecordExhausted() + } else { + metrics.RecordConnect(10 * time.Millisecond) + } + } + + pool.lastCheck.Store(0) + pool.adjust() + capAfter := pool.Cap() + + // Internet 对 10% exhaust 只是 Stressed（×0.85），不是 Congested（×0.5） + if capAfter < 80 { + t.Errorf("Internet 10%% exhaust 不应大幅降速: cap=%d", capAfter) + } + + t.Logf("Internet loose threshold: cap=%d (from 100)", capAfter) +} + +func TestOpt3_EnvAffectsHealthDecision(t *testing.T) { + envs := []struct { + env NetworkEnv + name string + }{ + {EnvLAN, "LAN"}, + {EnvWAN, "WAN"}, + {EnvInternet, "Internet"}, + } + + var caps []int + + for _, e := range envs { + metrics := &ScanMetrics{} + pool, err := NewAdaptivePool(100, 100, func(interface{}) {}, metrics, e.env) + if err != nil { + t.Fatalf("创建池失败: %v", err) + } + + pool.inSlowStart = false + pool.tune(100) + + // 相同的 12% exhaust rate + for i := 0; i < 100; i++ { + if i < 12 { + metrics.RecordExhausted() + } else { + metrics.RecordConnect(time.Millisecond) + } + } + + pool.lastCheck.Store(0) + pool.adjust() + caps = append(caps, pool.Cap()) + pool.Release() + + t.Logf("%s: cap=%d (12%% exhaust)", e.name, caps[len(caps)-1]) + } + + // LAN 反应最激烈（cap 最低），Internet 最宽容（cap 最高） + if caps[0] >= caps[2] { + t.Errorf("LAN cap(%d) 应 < Internet cap(%d) for same exhaust rate", caps[0], caps[2]) + } +} + +// ============================================================================= +// 优化 4：去掉 semaphore，ants 池天然反压 +// ============================================================================= + +func TestOpt4_SemaphoreRemoved(t *testing.T) { + // 验证 portScanTask 结构体不再有 semaphore 字段 + // 如果 semaphore 被加回来，这段代码编译就会报 "unknown field" + _ = portScanTask{ + host: "127.0.0.1", + port: 80, + addr: "127.0.0.1:80", + } + t.Log("portScanTask 无 semaphore 字段，反压由 ants pool 统一管理") +} + +// ============================================================================= +// 优化 5：扩充探测端口 +// ============================================================================= + +func TestOpt5_ProbePortsExpanded(t *testing.T) { + if len(probePorts) < 5 { + t.Errorf("probePorts 只有 %d 个，应该扩充到至少 5 个", len(probePorts)) + } + + // 验证包含关键端口 + required := map[int]bool{80: false, 443: false, 22: false} + for _, p := range probePorts { + if _, ok := required[p]; ok { + required[p] = true + } + } + for port, found := range required { + if !found { + t.Errorf("probePorts 缺少关键端口 %d", port) + } + } + + // 验证没有重复 + seen := make(map[int]bool) + for _, p := range probePorts { + if seen[p] { + t.Errorf("probePorts 有重复端口 %d", p) + } + seen[p] = true + } + + t.Logf("probePorts = %v (%d 个)", probePorts, len(probePorts)) +} + +// ============================================================================= +// 优化 6：computeRetries 环境自适应 +// ============================================================================= + +func TestOpt6_RetriesEnvAware(t *testing.T) { + lossRate := 0.3 // 30% 丢包 + + lanRetry := computeRetries(lossRate, EnvLAN) + wanRetry := computeRetries(lossRate, EnvWAN) + inetRetry := computeRetries(lossRate, EnvInternet) + + // LAN 目标概率更严格(0.5%)，应该重试更多；但上限更低(4) + // Internet 目标概率更宽松(2%)，应该重试更少；但上限更高(6) + t.Logf("30%% loss: LAN=%d, WAN=%d, Internet=%d", lanRetry, wanRetry, inetRetry) + + if lanRetry < 1 || lanRetry > 4 { + t.Errorf("LAN retry=%d, 应在 [1,4]", lanRetry) + } + if wanRetry < 1 || wanRetry > 5 { + t.Errorf("WAN retry=%d, 应在 [1,5]", wanRetry) + } + if inetRetry < 1 || inetRetry > 6 { + t.Errorf("Internet retry=%d, 应在 [1,6]", inetRetry) + } +} + +func TestOpt6_RetriesMaxByEnv(t *testing.T) { + // 高丢包率，各环境应返回各自上限 + lanMax := computeRetries(0.99, EnvLAN) + wanMax := computeRetries(0.99, EnvWAN) + inetMax := computeRetries(0.99, EnvInternet) + + if lanMax != 4 { + t.Errorf("LAN max retry=%d, want 4", lanMax) + } + if wanMax != 5 { + t.Errorf("WAN max retry=%d, want 5", wanMax) + } + if inetMax != 6 { + t.Errorf("Internet max retry=%d, want 6", inetMax) + } +} + +func TestOpt6_RetriesMathCorrectness(t *testing.T) { + envs := []struct { + env NetworkEnv + targetProb float64 + name string + }{ + {EnvLAN, 0.005, "LAN"}, + {EnvWAN, 0.01, "WAN"}, + {EnvInternet, 0.02, "Internet"}, + } + + for _, e := range envs { + for _, loss := range []float64{0.05, 0.10, 0.20, 0.30} { + retries := computeRetries(loss, e.env) + prob := 1.0 + for i := 0; i < retries; i++ { + prob *= loss + } + // 重试后全失败概率应 < targetProb（除非被 clamp 了） + if prob >= e.targetProb && retries < 4 { + t.Errorf("%s loss=%.0f%% retries=%d: P=%.6f >= %.3f", + e.name, loss*100, retries, prob, e.targetProb) + } + } + } +} + +// ============================================================================= +// 端到端集成：全链路验证 +// ============================================================================= + +func TestOptAll_EndToEnd_LANToPool(t *testing.T) { + // 模拟内网探测 → TuneConfig → 创建池 → 池根据 env 自适应 + profile := classifyNetwork( + makeDurations([]int{1, 1, 2, 2, 2, 3, 3, 3, 4, 5}), + 0, 10, + ) + + config := makeDefaultConfig() + session := makeTestSession(config) + ep := &EnvironmentProfile{Net: *profile, System: SystemProfile{FDLimit: 65536, NumCPU: 8}} + ep.TuneConfig(config, session) + + // 验证 env 被存储 + if config.DetectedNetworkEnv != int(EnvLAN) { + t.Errorf("DetectedNetworkEnv=%d, want %d(LAN)", config.DetectedNetworkEnv, int(EnvLAN)) + } + + // 验证 ceiling 合理 + if config.ThreadCeiling < config.ThreadNum { + t.Errorf("ceiling(%d) < target(%d)", config.ThreadCeiling, config.ThreadNum) + } + + // 创建池并验证 env 传递 + netEnv := NetworkEnv(config.DetectedNetworkEnv) + metrics := &ScanMetrics{} + pool, err := NewAdaptivePool(config.ThreadNum, config.ThreadCeiling, func(interface{}) {}, metrics, netEnv) + if err != nil { + t.Fatalf("创建池失败: %v", err) + } + defer pool.Release() + + if pool.networkEnv != EnvLAN { + t.Errorf("池的 networkEnv=%v, want LAN", pool.networkEnv) + } + + t.Logf("端到端 LAN: target=%d ceiling=%d env=%v maxRetry=%d", + config.ThreadNum, config.ThreadCeiling, netEnv, config.MaxRetries) +} + +func TestOptAll_EndToEnd_InternetToPool(t *testing.T) { + profile := classifyNetwork( + makeDurations([]int{60, 70, 80, 90, 100, 110, 120, 130, 140, 150}), + 0, 10, + ) + + config := makeDefaultConfig() + session := makeTestSession(config) + ep := &EnvironmentProfile{Net: *profile, System: SystemProfile{FDLimit: 4096, NumCPU: 4}} + ep.TuneConfig(config, session) + + if config.DetectedNetworkEnv != int(EnvInternet) { + t.Errorf("DetectedNetworkEnv=%d, want %d(Internet)", config.DetectedNetworkEnv, int(EnvInternet)) + } + + // 公网 target 应明显低于默认 600 + if config.ThreadNum >= 600 { + t.Errorf("公网 threadNum=%d, 应 < 600", config.ThreadNum) + } + + // ceiling 应 == target（非显式模式） + if config.ThreadCeiling != config.ThreadNum { + t.Errorf("非显式模式 ceiling(%d) != target(%d)", config.ThreadCeiling, config.ThreadNum) + } + + netEnv := NetworkEnv(config.DetectedNetworkEnv) + metrics := &ScanMetrics{} + pool, err := NewAdaptivePool(config.ThreadNum, config.ThreadCeiling, func(interface{}) {}, metrics, netEnv) + if err != nil { + t.Fatalf("创建池失败: %v", err) + } + defer pool.Release() + + // 注入 12% exhaust，Internet 环境应只是 Stressed 而不是 Congested + pool.inSlowStart = false + pool.tune(config.ThreadNum) + for i := 0; i < 100; i++ { + if i < 12 { + metrics.RecordExhausted() + } else { + metrics.RecordConnect(80 * time.Millisecond) + } + } + pool.lastCheck.Store(0) + pool.adjust() + + // cap 不应被砍到一半以下（Stressed 只降 15%） + if pool.Cap() < config.ThreadNum*7/10 { + t.Errorf("Internet 12%% exhaust 降速过猛: %d -> %d", config.ThreadNum, pool.Cap()) + } + + t.Logf("端到端 Internet: target=%d ceiling=%d cap_after_stress=%d", + config.ThreadNum, config.ThreadCeiling, pool.Cap()) +} + diff --git a/core/port_scan.go b/core/port_scan.go index 6726b04..6f71fe4 100644 --- a/core/port_scan.go +++ b/core/port_scan.go @@ -101,10 +101,9 @@ func (c *resultCollector) GetAll() []string { // portScanTask 端口扫描任务（轻量级，用于滑动窗口调度） type portScanTask struct { - host string - port int - addr string // 预格式化的 host:port，避免 fmt.Sprintf 热路径分配 - semaphore chan struct{} // 完成时释放窗口槽位 + host string + port int + addr string // 预格式化的 host:port，避免 fmt.Sprintf 热路径分配 } // failedPortInfo 失败端口信息 @@ -216,20 +215,22 @@ func EnhancedPortScan(ctx context.Context, hosts []string, ports string, timeout failedCollector := &failedPortCollector{} var wg sync.WaitGroup + ceiling := config.ThreadCeiling + if ceiling < threadNum { + ceiling = threadNum + } + netEnv := NetworkEnv(config.DetectedNetworkEnv) session.LogDebug(i18n.Tr("port_scan_debug_pool_create", threadNum)) - pool, err := NewAdaptivePool(threadNum, threadNum, func(task interface{}) { + pool, err := NewAdaptivePool(threadNum, ceiling, func(task interface{}) { taskInfo, ok := task.(portScanTask) if !ok { return } - defer func() { - <-taskInfo.semaphore // 释放窗口槽位 - wg.Done() - }() + defer wg.Done() scanSinglePort(ctx, taskInfo.host, taskInfo.port, taskInfo.addr, adaptiveTO, metrics, &count, collector, failedCollector, session) common.UpdateProgressBar(1) - }, metrics) + }, metrics, netEnv) if err != nil { session.LogError(i18n.Tr("thread_pool_create_failed", err)) if stream != nil { @@ -242,7 +243,7 @@ func EnhancedPortScan(ctx context.Context, hosts []string, ports string, timeout session.LogDebug(i18n.GetText("port_scan_debug_schedule_start")) // 滑动窗口调度 - slidingWindowSchedule(iter, pool, &wg, threadNum) + slidingWindowSchedule(iter, pool, &wg) session.LogDebug(i18n.GetText("port_scan_debug_schedule_done")) // 收集结果 @@ -287,30 +288,21 @@ func EnhancedPortScan(ctx context.Context, hosts []string, ports string, timeout } // slidingWindowSchedule 滑动窗口调度器 -// 核心思想：维护固定数量的"飞行中"任务，一个完成立即补充新的 -// 优势：避免任务队列堆积，内存使用恒定 -func slidingWindowSchedule(iter *SocketIterator, pool *AdaptivePool, wg *sync.WaitGroup, windowSize int) { - // 使用信号量控制窗口大小 - semaphore := make(chan struct{}, windowSize) - +// ants.PoolWithFunc.Invoke 在池满时阻塞，天然提供反压，无需额外 semaphore +func slidingWindowSchedule(iter *SocketIterator, pool *AdaptivePool, wg *sync.WaitGroup) { for { host, port, ok := iter.Next() if !ok { break } - // 获取窗口槽位（阻塞直到有空位） - semaphore <- struct{}{} - wg.Add(1) task := portScanTask{ - host: host, - port: port, - addr: net.JoinHostPort(host, fmtPort(port)), - semaphore: semaphore, + host: host, + port: port, + addr: net.JoinHostPort(host, fmtPort(port)), } if err := pool.Invoke(task); err != nil { - <-semaphore wg.Done() } } @@ -725,6 +717,14 @@ func processServiceResult(ctx context.Context, host string, port int, addr strin details := buildServiceDetails(port, serviceInfo) isWeb := IsWebServiceByFingerprint(serviceInfo) + // 指纹既不匹配 webKeywords 也不匹配 nonWebKeywords（不确定区间） + // 补做一次 HTTP 探测，覆盖自定义 HTTP 框架等漏网场景 + if !isWeb && !isDefinitelyNonWeb(serviceInfo) { + if tryHTTPFallbackDetection(ctx, host, port, addr, config, session) { + isWeb = true + } + } + if isWeb { details["is_web"] = true } diff --git a/core/scan_metrics.go b/core/scan_metrics.go index 557e1b3..9a502a7 100644 --- a/core/scan_metrics.go +++ b/core/scan_metrics.go @@ -1,6 +1,7 @@ package core import ( + "runtime" "sync/atomic" "time" ) @@ -55,12 +56,14 @@ func updateEMA(target *atomic.Int64, sample int64, divisor int64) { if target.CompareAndSwap(0, sample) { return } + runtime.Gosched() continue } next := old + (sample-old)/divisor if target.CompareAndSwap(old, next) { return } + runtime.Gosched() } } diff --git a/core/web_scanner.go b/core/web_scanner.go index d2b4dab..d26437e 100644 --- a/core/web_scanner.go +++ b/core/web_scanner.go @@ -264,6 +264,21 @@ func IsWebServiceByFingerprint(serviceInfo *ServiceInfo) bool { return false } +// isDefinitelyNonWeb 判断服务是否明确不是 Web 服务 +// 只检查 nonWebKeywords，不在里面 = 不确定 = 值得做 HTTP 探测 +func isDefinitelyNonWeb(serviceInfo *ServiceInfo) bool { + if serviceInfo == nil || serviceInfo.Name == "" { + return false + } + serviceName := strings.ToLower(serviceInfo.Name) + for _, keyword := range nonWebKeywords { + if strings.Contains(serviceName, keyword) { + return true + } + } + return false +} + // CacheServiceInfo 缓存识别到的服务信息 func CacheServiceInfo(host string, port int, serviceInfo *ServiceInfo) { cacheKey := net.JoinHostPort(host, strconv.Itoa(port)) diff --git a/webscan/lib/Eval.go b/webscan/lib/Eval.go index 6eeae8e..f84dd0f 100644 --- a/webscan/lib/Eval.go +++ b/webscan/lib/Eval.go @@ -28,7 +28,8 @@ import ( // 基础CEL环境缓存（避免重复创建，减少内存分配） var ( - baseEnvOnce sync.Once + baseEnvMu sync.Mutex + baseEnvInited bool baseEnv *cel.Env baseProgramOpt []cel.ProgramOption ) @@ -52,47 +53,49 @@ func NewEnv(c *CustomLib) (*cel.Env, error) { return cachedCELEnv, cachedCELEnvErr } -// initBaseEnv 初始化基础CEL环境（只执行一次） +// initBaseEnv 初始化基础CEL环境（失败后允许重试） func initBaseEnv() { - baseEnvOnce.Do(func() { - // 收集所有函数声明 - var allDeclarations []*exprpb.Decl - allDeclarations = append(allDeclarations, registerStringDeclarations()...) - allDeclarations = append(allDeclarations, registerEncodingDeclarations()...) - allDeclarations = append(allDeclarations, registerCryptoDeclarations()...) - allDeclarations = append(allDeclarations, registerRandomDeclarations()...) - allDeclarations = append(allDeclarations, registerMiscDeclarations()...) + baseEnvMu.Lock() + defer baseEnvMu.Unlock() + if baseEnvInited { + return + } - // 收集所有函数实现 - var allImplementations []*functions.Overload - allImplementations = append(allImplementations, registerStringImplementations()...) - allImplementations = append(allImplementations, registerEncodingImplementations()...) - allImplementations = append(allImplementations, registerCryptoImplementations()...) - allImplementations = append(allImplementations, registerRandomImplementations()...) - allImplementations = append(allImplementations, registerMiscImplementations()...) + var allDeclarations []*exprpb.Decl + allDeclarations = append(allDeclarations, registerStringDeclarations()...) + allDeclarations = append(allDeclarations, registerEncodingDeclarations()...) + allDeclarations = append(allDeclarations, registerCryptoDeclarations()...) + allDeclarations = append(allDeclarations, registerRandomDeclarations()...) + allDeclarations = append(allDeclarations, registerMiscDeclarations()...) - // 保存程序选项供后续使用 - //nolint:staticcheck // SA1019: cel.Functions已废弃但CEL库尚未提供替代方案 - baseProgramOpt = []cel.ProgramOption{cel.Functions(allImplementations...)} + var allImplementations []*functions.Overload + allImplementations = append(allImplementations, registerStringImplementations()...) + allImplementations = append(allImplementations, registerEncodingImplementations()...) + allImplementations = append(allImplementations, registerCryptoImplementations()...) + allImplementations = append(allImplementations, registerRandomImplementations()...) + allImplementations = append(allImplementations, registerMiscImplementations()...) - // 创建基础环境 - var err error - baseEnv, err = cel.NewEnv( - cel.Container("lib"), - cel.Types(&UrlType{}, &Request{}, &Response{}, &Reverse{}), - //nolint:staticcheck // SA1019: cel.Declarations已废弃但CEL库尚未提供替代方案 - cel.Declarations( - decls.NewIdent("request", decls.NewObjectType("lib.Request"), nil), - decls.NewIdent("response", decls.NewObjectType("lib.Response"), nil), - decls.NewIdent("reverse", decls.NewObjectType("lib.Reverse"), nil), - ), - //nolint:staticcheck // SA1019: cel.Declarations已废弃但CEL库尚未提供替代方案 - cel.Declarations(allDeclarations...), - ) - if err != nil { - common.LogError(i18n.Tr("webscan_cel_init_failed", err)) - } - }) + //nolint:staticcheck // SA1019: cel.Functions已废弃但CEL库尚未提供替代方案 + baseProgramOpt = []cel.ProgramOption{cel.Functions(allImplementations...)} + + var err error + baseEnv, err = cel.NewEnv( + cel.Container("lib"), + cel.Types(&UrlType{}, &Request{}, &Response{}, &Reverse{}), + //nolint:staticcheck // SA1019: cel.Declarations已废弃但CEL库尚未提供替代方案 + cel.Declarations( + decls.NewIdent("request", decls.NewObjectType("lib.Request"), nil), + decls.NewIdent("response", decls.NewObjectType("lib.Response"), nil), + decls.NewIdent("reverse", decls.NewObjectType("lib.Reverse"), nil), + ), + //nolint:staticcheck // SA1019: cel.Declarations已废弃但CEL库尚未提供替代方案 + cel.Declarations(allDeclarations...), + ) + if err != nil { + common.LogError(i18n.Tr("webscan_cel_init_failed", err)) + return + } + baseEnvInited = true } // GetBaseEnv 获取基础CEL环境 diff --git a/webscan/lib/poc_executor.go b/webscan/lib/poc_executor.go index 30c66b7..bc6be89 100644 --- a/webscan/lib/poc_executor.go +++ b/webscan/lib/poc_executor.go @@ -889,6 +889,12 @@ func evalset(env *cel.Env, variableMap map[string]interface{}, k string, express // evalset1 执行CEL表达式的简化版本 func evalset1(env *cel.Env, variableMap map[string]interface{}, k string, expression string) (string, error) { + // 纯字面量字符串（无函数调用、运算符、变量引用）直接当值用，跳过 CEL 编译 + // 避免 sets 中的 "sql"、"database" 等被当成 CEL 变量引用产生大量错误日志 + if isPlainLiteral(expression, variableMap) { + variableMap[k] = expression + return expression, nil + } out, err := Evaluate(env, expression, variableMap) if err != nil { variableMap[k] = expression @@ -898,6 +904,26 @@ func evalset1(env *cel.Env, variableMap map[string]interface{}, k string, expres return fmt.Sprintf("%v", variableMap[k]), err } +// isPlainLiteral 判断表达式是否是纯字面量值（不需要 CEL 求值） +// 排除法：含 CEL 语法特征（括号、运算符、引号）的需要走 CEL，其余当字面量 +func isPlainLiteral(expr string, variableMap map[string]interface{}) bool { + if expr == "" { + return false + } + // 如果是已声明的变量引用，必须走 CEL 求值 + if _, exists := variableMap[expr]; exists { + return false + } + // 含 CEL 语法特征的需要走 CEL 编译 + for _, c := range expr { + switch c { + case '(', ')', '[', ']', '+', '*', '%', '=', '!', '<', '>', '&', '|', '"', '\'', '?', ':': + return false + } + } + return true +} + // CheckInfoPoc 检查POC信息并返回别名 func CheckInfoPoc(infostr string) string { for _, poc := range fingerprint.PocDatas {