feat: 统一服务缓存 + 指纹驱动插件匹配

将 webServiceCache 扩展为通用 serviceCache，所有指纹识别结果
统一缓存，插件匹配时端口不命中则回退到服务名称匹配。

删除多余的 service_cache.go，复用已有的 ServiceInfo 体系。
补充 nil 防御、Explicit 标记、大量单元/集成/回归测试。

plugins/services/cassandra.go

@@ -8,6 +8,7 @@ import (
 	"fmt"
 	"io"
 	"net"
+	"sync/atomic"
 	"time"
 
 	"github.com/shadow1ng/fscan/common"
@@ -165,20 +166,19 @@ func (p *CassandraPlugin) doCassandraAuth(ctx context.Context, info *common.Host
 
 // ── CQL wire protocol 工具 ──────────────────────────────────────
 
-var cqlStreamID int16
+var cqlStreamID uint32
+
+func nextCQLStreamID() uint16 {
+	return uint16((atomic.AddUint32(&cqlStreamID, 1) - 1) & 0x7fff)
+}
 
 func cqlSend(conn net.Conn, opcode byte, body []byte) error {
-	id := cqlStreamID
-	if cqlStreamID == 32767 {
-		cqlStreamID = 0
-	} else {
-		cqlStreamID++
-	}
+	id := nextCQLStreamID()
 
 	// frame: [1B version|flags] [2B stream] [1B opcode] [4B length] [body]
 	header := make([]byte, 8)
 	header[0] = cqlVersion
-	binary.BigEndian.PutUint16(header[1:3], uint16(id))
+	binary.BigEndian.PutUint16(header[1:3], id)
 	header[3] = opcode
 	binary.BigEndian.PutUint32(header[4:8], uint32(len(body)))
 

plugins/services/credential_tester.go

@@ -7,6 +7,7 @@ import (
 	"io"
 	"net"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/shadow1ng/fscan/common"
@@ -61,7 +62,11 @@ type AuthFunc func(ctx context.Context, cred Credential) *AuthResult
 // ErrorClassifier 错误分类函数
 type ErrorClassifier func(err error) ErrorType
 
-var authCleanupWait = 2 * time.Second
+var authCleanupWaitNanos int64 = int64(2 * time.Second)
+
+func authCleanupWait() time.Duration {
+	return time.Duration(atomic.LoadInt64(&authCleanupWaitNanos))
+}
 
 // =============================================================================
 // 单凭据测试（解决 goroutine 泄漏）
@@ -70,9 +75,36 @@ var authCleanupWait = 2 * time.Second
 // TestSingleCredential 安全地测试单个凭据
 // 正确处理 context 取消时的资源清理
 func TestSingleCredential(ctx context.Context, cred Credential, authFn AuthFunc) *AuthResult {
+	if ctx == nil {
+		ctx = context.Background()
+	}
+	if authFn == nil {
+		return &AuthResult{
+			Success:   false,
+			ErrorType: ErrorTypeUnknown,
+			Error:     fmt.Errorf("auth function is nil"),
+		}
+	}
+	if err := ctx.Err(); err != nil {
+		return &AuthResult{
+			Success:   false,
+			ErrorType: ErrorTypeNetwork,
+			Error:     err,
+		}
+	}
+
 	resultChan := make(chan *AuthResult, 1)
 
 	go func() {
+		defer func() {
+			if r := recover(); r != nil {
+				resultChan <- &AuthResult{
+					Success:   false,
+					ErrorType: ErrorTypeUnknown,
+					Error:     fmt.Errorf("auth function panic: %v", r),
+				}
+			}
+		}()
 		result := authFn(ctx, cred)
 		resultChan <- result
 	}()
@@ -83,7 +115,7 @@ func TestSingleCredential(ctx context.Context, cred Credential, authFn AuthFunc)
 	case <-ctx.Done():
 		// context 被取消后只做有界等待，避免 authFn 卡死时清理 goroutine 也永久泄漏。
 		go func() {
-			timer := time.NewTimer(authCleanupWait)
+			timer := time.NewTimer(authCleanupWait())
 			defer timer.Stop()
 
 			select {
@@ -116,15 +148,35 @@ type ConcurrentTestConfig struct {
 	UseProxy                bool          // 代理模式下跳过直连 TCP 预检
 }
 
+func normalizeConcurrentTestConfig(testConfig ConcurrentTestConfig) ConcurrentTestConfig {
+	if testConfig.Concurrency <= 0 {
+		testConfig.Concurrency = 10
+	}
+	if testConfig.MaxRetries <= 0 {
+		testConfig.MaxRetries = 3
+	}
+	if testConfig.RetryDelay <= 0 {
+		testConfig.RetryDelay = time.Second
+	}
+	if testConfig.MaxConsecutiveNetErrors <= 0 {
+		testConfig.MaxConsecutiveNetErrors = 5
+	}
+	return testConfig
+}
+
 // DefaultConcurrentTestConfig 默认配置
 func DefaultConcurrentTestConfig(config *common.Config) ConcurrentTestConfig {
 	concurrency := config.ModuleThreadNum
 	if concurrency <= 0 {
 		concurrency = 10
 	}
+	maxRetries := config.MaxRetries
+	if maxRetries <= 0 {
+		maxRetries = 3
+	}
 	return ConcurrentTestConfig{
 		Concurrency:             concurrency,
-		MaxRetries:              3,
+		MaxRetries:              maxRetries,
 		RetryDelay:              time.Second,
 		MaxConsecutiveNetErrors: 5,
 		UseProxy:                config.Network.Socks5Proxy != "" || config.Network.HTTPProxy != "",
@@ -147,6 +199,9 @@ func TestCredentialsConcurrently(
 	serviceName string,
 	testConfig ConcurrentTestConfig,
 ) *ScanResult {
+	if ctx == nil {
+		ctx = context.Background()
+	}
 	if len(credentials) == 0 {
 		return &ScanResult{
 			Success: false,
@@ -154,11 +209,16 @@ func TestCredentialsConcurrently(
 			Error:   fmt.Errorf("%s", i18n.GetText("service_no_test_creds")),
 		}
 	}
+	testConfig = normalizeConcurrentTestConfig(testConfig)
 
 	// TCP 预检：快速验证目标可达，避免对不可达目标浪费全部凭据尝试
 	// 代理模式下跳过：net.DialTimeout 直连无法到达代理后的内网目标
 	if testConfig.TargetAddr != "" && !testConfig.UseProxy {
-		preConn, err := net.DialTimeout("tcp", testConfig.TargetAddr, 3*time.Second)
+		dialCtx, dialCancel := context.WithTimeout(ctx, 3*time.Second)
+		defer dialCancel()
+
+		var dialer net.Dialer
+		preConn, err := dialer.DialContext(dialCtx, "tcp", testConfig.TargetAddr)
 		if err != nil {
 			return &ScanResult{
 				Success: false,
@@ -240,10 +300,6 @@ func workerTestCredentials(
 	testConfig ConcurrentTestConfig,
 ) {
 	consecutiveNetErrors := 0
-	maxNetErrors := testConfig.MaxConsecutiveNetErrors
-	if maxNetErrors <= 0 {
-		maxNetErrors = 5
-	}
 
 	for cred := range credChan {
 		// 检查是否应该停止
@@ -254,7 +310,7 @@ func workerTestCredentials(
 		}
 
 		// 连续网络错误达到阈值，目标可能不可达，提前退出
-		if consecutiveNetErrors >= maxNetErrors {
+		if consecutiveNetErrors >= testConfig.MaxConsecutiveNetErrors {
 			return
 		}
 
@@ -292,10 +348,18 @@ func testCredentialWithRetry(
 
 		// 测试凭据
 		result := TestSingleCredential(ctx, cred, authFn)
+		if result == nil {
+			result = &AuthResult{
+				Success:   false,
+				ErrorType: ErrorTypeUnknown,
+				Error:     fmt.Errorf("auth function returned nil result"),
+			}
+		}
 
-		if result.Success && result.Conn != nil {
-			// 成功，关闭连接并返回
-			_ = result.Conn.Close()
+		if result.Success {
+			if result.Conn != nil {
+				_ = result.Conn.Close()
+			}
 			return &ScanResult{
 				Type:     plugins.ResultTypeCredential,
 				Success:  true,

plugins/services/credential_tester_test.go

@@ -8,6 +8,8 @@ import (
 	"sync/atomic"
 	"testing"
 	"time"
+
+	"github.com/shadow1ng/fscan/common"
 )
 
 /*
@@ -187,6 +189,84 @@ func TestMatchIgnoreCase(t *testing.T) {
 // 并发测试
 // =============================================================================
 
+func setAuthCleanupWaitForTest(wait time.Duration) func() {
+	oldWait := atomic.LoadInt64(&authCleanupWaitNanos)
+	atomic.StoreInt64(&authCleanupWaitNanos, int64(wait))
+	return func() { atomic.StoreInt64(&authCleanupWaitNanos, oldWait) }
+}
+
+func TestDefaultConcurrentTestConfigUsesConfigRetries(t *testing.T) {
+	cfg := DefaultConcurrentTestConfig(&common.Config{
+		ModuleThreadNum: 7,
+		MaxRetries:      5,
+	})
+
+	if cfg.Concurrency != 7 {
+		t.Fatalf("Concurrency = %d, want 7", cfg.Concurrency)
+	}
+	if cfg.MaxRetries != 5 {
+		t.Fatalf("MaxRetries = %d, want config MaxRetries 5", cfg.MaxRetries)
+	}
+}
+
+func TestDefaultConcurrentTestConfigRetriesFallback(t *testing.T) {
+	cfg := DefaultConcurrentTestConfig(&common.Config{
+		ModuleThreadNum: 0,
+		MaxRetries:      0,
+	})
+
+	if cfg.Concurrency != 10 {
+		t.Fatalf("Concurrency = %d, want fallback 10", cfg.Concurrency)
+	}
+	if cfg.MaxRetries != 3 {
+		t.Fatalf("MaxRetries = %d, want fallback 3", cfg.MaxRetries)
+	}
+}
+
+func TestTestCredentialsConcurrently_ZeroValueConfigStillRuns(t *testing.T) {
+	var calls atomic.Int32
+	authFn := func(ctx context.Context, cred Credential) *AuthResult {
+		calls.Add(1)
+		return &AuthResult{Success: true}
+	}
+
+	result := TestCredentialsConcurrently(context.Background(), []Credential{{Username: "u", Password: "p"}}, authFn, "test", ConcurrentTestConfig{})
+	if !result.Success {
+		t.Fatalf("zero-value config should still test credentials: %v", result.Error)
+	}
+	if calls.Load() != 1 {
+		t.Fatalf("authFn calls = %d, want 1", calls.Load())
+	}
+}
+
+func TestTestCredentialsConcurrently_PrecheckHonorsCanceledContext(t *testing.T) {
+	var calls atomic.Int32
+	authFn := func(ctx context.Context, cred Credential) *AuthResult {
+		calls.Add(1)
+		return &AuthResult{Success: false}
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+
+	start := time.Now()
+	result := TestCredentialsConcurrently(ctx, []Credential{{Username: "u", Password: "p"}}, authFn, "test", ConcurrentTestConfig{
+		Concurrency: 1,
+		MaxRetries:  1,
+		TargetAddr:  "203.0.113.1:65000",
+	})
+
+	if result.Success {
+		t.Fatal("canceled context should not return success")
+	}
+	if calls.Load() != 0 {
+		t.Fatalf("authFn calls = %d, want 0 when precheck context is canceled", calls.Load())
+	}
+	if elapsed := time.Since(start); elapsed > 200*time.Millisecond {
+		t.Fatalf("precheck ignored canceled context, elapsed=%v", elapsed)
+	}
+}
+
 // mockConn 模拟连接
 type mockConn struct {
 	closed atomic.Bool
@@ -336,6 +416,60 @@ func TestTestCredentialsConcurrently_ContextCancel(t *testing.T) {
 	}
 }
 
+func TestTestCredentialsConcurrently_CancelWithStuckAuthReturnsPromptly(t *testing.T) {
+	defer setAuthCleanupWaitForTest(20 * time.Millisecond)()
+
+	credentials := make([]Credential, 10)
+	for i := range credentials {
+		credentials[i] = Credential{Username: "user", Password: "pass"}
+	}
+
+	authStarted := make(chan struct{}, len(credentials))
+	releaseAuth := make(chan struct{})
+	authFn := func(ctx context.Context, cred Credential) *AuthResult {
+		authStarted <- struct{}{}
+		<-releaseAuth
+		return &AuthResult{Success: false, ErrorType: ErrorTypeNetwork}
+	}
+
+	config := ConcurrentTestConfig{
+		Concurrency: 3,
+		MaxRetries:  1,
+		RetryDelay:  time.Millisecond,
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan *ScanResult, 1)
+	go func() {
+		done <- TestCredentialsConcurrently(ctx, credentials, authFn, "test", config)
+	}()
+
+	for i := 0; i < config.Concurrency; i++ {
+		select {
+		case <-authStarted:
+		case <-time.After(time.Second):
+			close(releaseAuth)
+			t.Fatalf("authFn started %d workers, want %d", i, config.Concurrency)
+		}
+	}
+
+	start := time.Now()
+	cancel()
+	select {
+	case result := <-done:
+		close(releaseAuth)
+		if result.Success {
+			t.Fatal("context取消后不应该返回成功")
+		}
+		if elapsed := time.Since(start); elapsed > 200*time.Millisecond {
+			t.Fatalf("取消后返回过慢: %v", elapsed)
+		}
+	case <-time.After(time.Second):
+		close(releaseAuth)
+		t.Fatal("authFn 卡住时并发测试没有及时返回")
+	}
+}
+
 // =============================================================================
 // 单凭据测试
 // =============================================================================
@@ -361,6 +495,49 @@ func TestTestSingleCredential_Success(t *testing.T) {
 	}
 }
 
+func TestTestSingleCredential_CanceledContextSkipsAuth(t *testing.T) {
+	var calls atomic.Int32
+	authFn := func(ctx context.Context, cred Credential) *AuthResult {
+		calls.Add(1)
+		return &AuthResult{Success: true}
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+
+	result := TestSingleCredential(ctx, Credential{Username: "admin", Password: "admin"}, authFn)
+	if result.Success {
+		t.Fatal("canceled context should not return success")
+	}
+	if calls.Load() != 0 {
+		t.Fatalf("authFn calls = %d, want 0", calls.Load())
+	}
+}
+
+func TestTestSingleCredential_NilAuthFunc(t *testing.T) {
+	result := TestSingleCredential(context.Background(), Credential{Username: "admin", Password: "admin"}, nil)
+	if result.Success {
+		t.Fatal("nil authFn should not return success")
+	}
+	if result.Error == nil {
+		t.Fatal("nil authFn should return an error")
+	}
+}
+
+func TestTestSingleCredential_RecoverAuthPanic(t *testing.T) {
+	authFn := func(ctx context.Context, cred Credential) *AuthResult {
+		panic("boom")
+	}
+
+	result := TestSingleCredential(context.Background(), Credential{Username: "admin", Password: "admin"}, authFn)
+	if result.Success {
+		t.Fatal("panic authFn should not return success")
+	}
+	if result.Error == nil {
+		t.Fatal("panic authFn should return an error")
+	}
+}
+
 // TestTestSingleCredential_ContextCancel 测试context取消时的资源清理
 func TestTestSingleCredential_ContextCancel(t *testing.T) {
 	conn := &mockConn{}
@@ -403,9 +580,7 @@ func TestTestSingleCredential_ContextCancel(t *testing.T) {
 }
 
 func TestTestSingleCredential_ContextCancelCleanupIsBounded(t *testing.T) {
-	oldWait := authCleanupWait
-	authCleanupWait = 20 * time.Millisecond
-	defer func() { authCleanupWait = oldWait }()
+	defer setAuthCleanupWaitForTest(20 * time.Millisecond)()
 
 	authStarted := make(chan struct{})
 	releaseAuth := make(chan struct{})
@@ -480,6 +655,45 @@ func TestRetryLogic_NetworkErrorRetries(t *testing.T) {
 	}
 }
 
+func TestRetryLogic_SuccessWithoutConn(t *testing.T) {
+	var attempts atomic.Int32
+	authFn := func(ctx context.Context, cred Credential) *AuthResult {
+		attempts.Add(1)
+		return &AuthResult{Success: true}
+	}
+
+	result := TestCredentialsConcurrently(context.Background(), []Credential{{Username: "admin", Password: "admin"}}, authFn, "test", ConcurrentTestConfig{
+		Concurrency: 1,
+		MaxRetries:  3,
+	})
+	if !result.Success {
+		t.Fatalf("success result without Conn should be accepted: %v", result.Error)
+	}
+	if attempts.Load() != 1 {
+		t.Fatalf("attempts = %d, want 1", attempts.Load())
+	}
+}
+
+func TestRetryLogic_NilAuthResultDoesNotPanic(t *testing.T) {
+	var attempts atomic.Int32
+	authFn := func(ctx context.Context, cred Credential) *AuthResult {
+		attempts.Add(1)
+		return nil
+	}
+
+	result := TestCredentialsConcurrently(context.Background(), []Credential{{Username: "admin", Password: "admin"}}, authFn, "test", ConcurrentTestConfig{
+		Concurrency: 1,
+		MaxRetries:  2,
+		RetryDelay:  time.Millisecond,
+	})
+	if result.Success {
+		t.Fatal("nil auth result should not return success")
+	}
+	if attempts.Load() != 2 {
+		t.Fatalf("attempts = %d, want 2", attempts.Load())
+	}
+}
+
 // TestRetryLogic_AuthErrorNoRetry 认证错误不应该重试
 func TestRetryLogic_AuthErrorNoRetry(t *testing.T) {
 	var attempts atomic.Int32

plugins/services/kafka.go

@@ -8,6 +8,7 @@ import (
 	"fmt"
 	"io"
 	"net"
+	"sync/atomic"
 	"time"
 
 	"github.com/shadow1ng/fscan/common"
@@ -154,9 +155,12 @@ func (p *KafkaPlugin) doKafkaAuth(ctx context.Context, info *common.HostInfo, cr
 
 var kafkaCorrelationID int32
 
+func nextKafkaCorrelationID() int32 {
+	return atomic.AddInt32(&kafkaCorrelationID, 1) - 1
+}
+
 func kafkaSend(conn net.Conn, apiKey, apiVersion int16, body []byte) error {
-	corrID := kafkaCorrelationID
-	kafkaCorrelationID++
+	corrID := nextKafkaCorrelationID()
 
 	// 请求格式: [4B len] [2B api_key] [2B api_version] [4B corr_id] [2B client_id_len] [client_id] [body]
 	clientID := "fscan"

plugins/services/mongodb.go

@@ -11,6 +11,7 @@ import (
 	"io"
 	"net"
 	"strings"
+	"sync/atomic"
 	"time"
 
 	"github.com/shadow1ng/fscan/common"
@@ -156,8 +157,7 @@ const (
 var mongoRequestID uint32
 
 func nextRequestID() uint32 {
-	mongoRequestID++
-	return mongoRequestID
+	return atomic.AddUint32(&mongoRequestID, 1)
 }
 
 // buildMongoCommand 构建 MongoDB 命令的 OP_MSG body (最小 BSON 实现)

plugins/services/protocol_ids_test.go

@@ -0,0 +1,46 @@
+package services
+
+import (
+	"sync"
+	"testing"
+)
+
+func TestProtocolIDsAreConcurrentSafe(t *testing.T) {
+	const workers = 64
+	const perWorker = 64
+
+	tests := []struct {
+		name string
+		next func() uint32
+	}{
+		{"mongodb", nextRequestID},
+		{"kafka", func() uint32 { return uint32(nextKafkaCorrelationID()) }},
+		{"cassandra", func() uint32 { return uint32(nextCQLStreamID()) }},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			var wg sync.WaitGroup
+			values := make(chan uint32, workers*perWorker)
+			for i := 0; i < workers; i++ {
+				wg.Add(1)
+				go func() {
+					defer wg.Done()
+					for j := 0; j < perWorker; j++ {
+						values <- tt.next()
+					}
+				}()
+			}
+			wg.Wait()
+			close(values)
+
+			seen := make(map[uint32]struct{}, workers*perWorker)
+			for value := range values {
+				if _, ok := seen[value]; ok {
+					t.Fatalf("duplicate protocol id %d", value)
+				}
+				seen[value] = struct{}{}
+			}
+		})
+	}
+}