优化自适应扫描系统 & 修复 POC 调度问题
测试构建 / 代码检查 (push) Has been cancelled
测试构建 / 单元测试和构建 (push) Has been cancelled
测试构建 / 构建验证 (push) Has been cancelled

自适应扫描优化:
- target/ceiling 分离,自适应池可向上探索而非锁死在 target
- assessHealth 阈值按网络环境区分(LAN 收紧 / Internet 放宽)
- RTT 漂移时动态压低 target,配合 AIMD 双重降速
- 去掉 semaphore 双层流控,由 ants pool 统一反压
- 探测端口从 3 个扩充到 8 个,减少 RTT 采样偏差
- computeRetries 按环境调整目标概率和上限

Bug 修复:
- AdaptivePool.Wait() 加 10 分钟超时,防止 goroutine 卡死时永久挂起
- CEL 环境初始化失败后允许重试(sync.Once → sync.Mutex + 标志位)
- CAS 自旋加 runtime.Gosched() 退避,减少高并发下 CPU 空转
- -full 模式下 web 插件跳过 IsMarkedWebService 检查 #588
- 不确定服务补做 HTTP 回退探测,覆盖自定义框架漏网场景
- POC sets 纯字面量值跳过 CEL 编译,消除大量误报错误日志
This commit is contained in:
ZacharyZcR
2026-06-13 12:39:24 +08:00
parent 15a7670ba2
commit 45ebe7040e
15 changed files with 798 additions and 104 deletions
+67 -9
View File
@@ -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)
}
}
}
+5
View File
@@ -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)
}
+11 -11
View File
@@ -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)
}
}
+43 -15
View File
@@ -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 发包速率
+5 -5
View File
@@ -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)
+1 -1
View File
@@ -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))
+549
View File
@@ -0,0 +1,549 @@
package core
import (
"sync/atomic"
"testing"
"time"
)
// =============================================================================
// 优化 1target/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)
}
}
// =============================================================================
// 优化 2RTT 漂移微调 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 突增到 100ms100 倍),大量喂入让 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=%.2fEMA 差距不够大,跳过", 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)
}
}
// =============================================================================
// 优化 3assessHealth 阈值跟 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:去掉 semaphoreants 池天然反压
// =============================================================================
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))
}
// =============================================================================
// 优化 6computeRetries 环境自适应
// =============================================================================
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% exhaustInternet 环境应只是 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())
}
+25 -25
View File
@@ -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
}
+3
View File
@@ -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()
}
}
+15
View File
@@ -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))