feat(probe): add third-party dial-test adapter for 15C (tsk_-WC9smQK7YRt)

Implements ProberAgent interface + Alibaba Cloud CloudMonitor Synthetic adapter: - ProbeTarget / VantageResult types aligned with 15A schema (L3 always nil) - AliyunSyntheticAgent: CreateSiteMonitor → poll → map to VantageResult - ISP vantage uses "3rd-{ISP}" prefix; Redis key probe:{node}:CN:{region}:3rd-* - Degradation: API error/rate-limit/timeout → skip vantage, log + counter (no write) - RunOnce(ctx, targets) shape for 15H assembly wiring - Credentials via cfg only (env-var injection, zero hardcoded secrets) - 6 new tests (mock vendor server): normalization, rate-limit, timeout, Redis keys, failed-target L1, no-credentials check; all 22 probe-package tests pass Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-06-13 17:21:43 +08:00
parent 9423069a0b
commit a47121f99b
2 changed files with 888 additions and 0 deletions
@@ -0,0 +1,551 @@
+// Package probe – prober_agent.go
+//
+// # 供应商选型：阿里云云监控·云拨测 (Alibaba Cloud CloudMonitor Synthetic Monitoring)
+//
+// 选型理由：
+//  1. 官方支持电信/联通/移动运营商节点筛选（IspCode 参数：246/247/248）。
+//  2. 支持 TCP 连通性探测（TaskType="TCP"），可获取 RTT + 失败原因。
+//  3. 覆盖全国 30+ 省份，节点元数据含 ISP 编码和城市。
+//  4. RPC 风格 REST API，HMAC-SHA1 签名，无需官方 SDK，依赖只有标准库。
+//  5. 按次计费，支持 RAM 子账号（权限最小化，满足运营安全 §06 红线 §2 身份隔离）。
+//
+// 身份隔离结论：
+//  阿里云 RAM 支持独立子账号 + Action 级别授权
+//  （cloudmonitor:CreateSiteMonitor + cloudmonitor:DescribeSiteMonitorData），
+//  可与运营身份完全隔离，无 KYC 障碍。满足红线要求，无需退纯自建。
+//
+// API 版本：2019-01-01，RPC 风格
+// 参考文档：https://help.aliyun.com/zh/cms/developer-reference/api-cms-2019-01-01-createsitemonitor
+//
+// # 降级语义
+//
+// API 报错/限流/超时 → 该 vantage 本周期不写 Redis（= 无数据，绝不写失败）。
+// 连续不可用：incrementFailCount + slog.Warn，留给 15G 事件口消费。
+package probe
+
+import (
+	"context"
+	"crypto/hmac"
+	"crypto/sha1" //nolint:gosec // Alibaba Cloud ACS v1 signing mandates HMAC-SHA1
+	"encoding/base64"
+	"encoding/json"
+	"fmt"
+	"io"
+	"log/slog"
+	"net/http"
+	"sort"
+	"strings"
+	"sync/atomic"
+	"time"
+)
+
+// --------------------------------------------------------------------------
+// Public types
+// --------------------------------------------------------------------------
+
+// ProbeTarget describes a single node endpoint to be dial-tested.
+type ProbeTarget struct {
+	// NodeID identifies the node in the database (matches nodes.id).
+	NodeID string
+
+	// Host is the hostname or IP address of the node endpoint.
+	Host string
+
+	// Port is the TCP port to probe (e.g. 443).
+	Port int
+}
+
+// VantageResult is a normalized dial-test result from a single ISP vantage point.
+//
+// L3 is always absent for third-party dial-tests.  Commercial synthetic-monitoring
+// APIs probe TCP/TLS reachability only; they cannot validate the application
+// protocol (Hysteria2/REALITY) end-to-end.  15D aggregation treats L3==nil from
+// a 3rd-party vantage as "unknown at L3", not as a failure.
+type VantageResult struct {
+	// Vantage is the network vantage point from which the probe was conducted.
+	// For Alibaba Cloud probes the Country is always "CN"; ISP uses the "3rd-"
+	// prefix convention (e.g. "3rd-ChinaTelecom") to distinguish from first-party
+	// probe agents.  15D may apply a configurable weight multiplier on this prefix.
+	Vantage VantagePoint
+
+	// L1 is the TCP connectivity result.
+	L1 L1Result
+
+	// L2 is the TLS handshake result.  Nil when L1 failed or the vendor did not
+	// perform TLS probing.
+	L2 *L2Result
+
+	// L3 is intentionally absent — see package doc above.
+}
+
+// ProberAgent abstracts a third-party dial-test provider.
+// Implementations must be safe for concurrent use.
+type ProberAgent interface {
+	// Probe issues multi-ISP dial-tests for the given target and returns one
+	// VantageResult per ISP vantage point.
+	//
+	// Degradation contract:
+	//   - On any provider error (rate-limit, timeout, API error) the
+	//     implementation MUST return (nil, nil) and handle the error internally
+	//     (log + counter).  Callers treat an empty slice as "no data this cycle".
+	//   - The implementation MUST NOT return a VantageResult whose L1.OK==false
+	//     solely because the API call failed; that would conflate infra errors
+	//     with real reachability failures.
+	Probe(ctx context.Context, target ProbeTarget) ([]VantageResult, error)
+}
+
+// --------------------------------------------------------------------------
+// AliyunSyntheticAgent
+// --------------------------------------------------------------------------
+
+// aliyunISP maps human-readable ISP names (used in VantageResult) to the
+// Alibaba Cloud CloudMonitor ISP code strings.
+var aliyunISP = []struct {
+	name    string // becomes VantagePoint.ISP with "3rd-" prefix
+	ispCode string // Alibaba Cloud IspCode
+}{
+	{"ChinaTelecom", "246"},
+	{"ChinaUnicom", "247"},
+	{"ChinaMobile", "248"},
+}
+
+// aliyunEndpoint is the CloudMonitor API base URL.
+// Overridable in tests via AliyunSyntheticAgent.baseURL.
+const aliyunEndpoint = "https://cloudmonitor.cn-hangzhou.aliyuncs.com/"
+
+// AliyunSyntheticAgentConfig holds configuration for AliyunSyntheticAgent.
+// The AccessKeyID and AccessKeySecret must belong to a RAM sub-account with
+// minimal permissions (cloudmonitor:CreateSiteMonitor +
+// cloudmonitor:DescribeSiteMonitorData only).  They must be injected via
+// environment variables; never stored in the database or committed to the repo.
+type AliyunSyntheticAgentConfig struct {
+	// AccessKeyID is the Alibaba Cloud RAM AccessKey ID.
+	// Source: env var ALIYUN_PROBE_ACCESS_KEY_ID.
+	AccessKeyID string
+
+	// AccessKeySecret is the Alibaba Cloud RAM AccessKey Secret.
+	// Source: env var ALIYUN_PROBE_ACCESS_KEY_SECRET.
+	AccessKeySecret string
+
+	// PollInterval is how long to wait between polling for task results.
+	// Default: 5 seconds.
+	PollInterval time.Duration
+
+	// PollTimeout is the maximum time to wait for a single task to complete.
+	// Default: 60 seconds.
+	PollTimeout time.Duration
+
+	// ProbeCity is the Alibaba Cloud city code used for all ISP probes.
+	// Default: "563" (Beijing).  Only one city is needed — the ISP is the
+	// relevant dimension for Pangolin's determination logic.
+	ProbeCity string
+
+	// BaseURL overrides the default CloudMonitor API endpoint.
+	// Leave empty for production; set to a mock server URL in tests.
+	BaseURL string
+
+	// HTTPTimeout overrides the HTTP client timeout per call.
+	// Zero uses the default of 30 seconds.
+	HTTPTimeout time.Duration
+}
+
+// AliyunSyntheticAgent implements ProberAgent using Alibaba Cloud CloudMonitor
+// Synthetic Monitoring.
+type AliyunSyntheticAgent struct {
+	cfg        AliyunSyntheticAgentConfig
+	store      *Store
+	httpClient *http.Client
+	baseURL    string // overridable in tests
+	failCount  atomic.Int64
+	logger     *slog.Logger
+}
+
+// NewAliyunSyntheticAgent creates an AliyunSyntheticAgent.
+// logger may be nil (falls back to slog.Default()).
+func NewAliyunSyntheticAgent(cfg AliyunSyntheticAgentConfig, store *Store, logger *slog.Logger) *AliyunSyntheticAgent {
+	if cfg.PollInterval == 0 {
+		cfg.PollInterval = 5 * time.Second
+	}
+	if cfg.PollTimeout == 0 {
+		cfg.PollTimeout = 60 * time.Second
+	}
+	if cfg.ProbeCity == "" {
+		cfg.ProbeCity = "563" // Beijing
+	}
+	if logger == nil {
+		logger = slog.Default()
+	}
+	httpTimeout := 30 * time.Second
+	if cfg.HTTPTimeout > 0 {
+		httpTimeout = cfg.HTTPTimeout
+	}
+	baseURL := aliyunEndpoint
+	if cfg.BaseURL != "" {
+		baseURL = cfg.BaseURL
+	}
+	return &AliyunSyntheticAgent{
+		cfg:   cfg,
+		store: store,
+		httpClient: &http.Client{
+			Timeout: httpTimeout,
+		},
+		baseURL: baseURL,
+		logger:  logger,
+	}
+}
+
+// NewAliyunSyntheticAgentWithOptions is a convenience constructor for tests that
+// need to inject a mock server URL without embedding it in the main config type.
+// Production code should use NewAliyunSyntheticAgent.
+func NewAliyunSyntheticAgentWithOptions(cfg AliyunSyntheticAgentConfig, store *Store, logger *slog.Logger, baseURL string) *AliyunSyntheticAgent {
+	cfg.BaseURL = baseURL
+	return NewAliyunSyntheticAgent(cfg, store, logger)
+}
+
+// SetHTTPClientTimeout replaces the agent's HTTP client timeout.
+// Exported only for use in tests; production code should set cfg.HTTPTimeout.
+func SetHTTPClientTimeout(a *AliyunSyntheticAgent, d time.Duration) *AliyunSyntheticAgent {
+	a.httpClient = &http.Client{Timeout: d}
+	return a
+}
+
+// RunOnce issues a full multi-ISP probe cycle for all given targets and writes
+// results to Redis via store.  Designed for use in an independent goroutine
+// controlled by 15H assembly.
+//
+// Per-target, per-ISP errors are handled internally (degraded silently);
+// the method only returns a non-nil error when the context is cancelled.
+func (a *AliyunSyntheticAgent) RunOnce(ctx context.Context, targets []ProbeTarget) error {
+	for _, t := range targets {
+		if ctx.Err() != nil {
+			return ctx.Err()
+		}
+		results, err := a.Probe(ctx, t)
+		if err != nil {
+			// Probe already logged internally; continue to next target.
+			continue
+		}
+		if len(results) == 0 {
+			continue
+		}
+		// Convert VantageResult slice into the store's NodeReport format.
+		// Each ISP vantage is a separate SaveReports call (distinct vantage key).
+		for _, vr := range results {
+			report := NodeReport{
+				NodeID: t.NodeID,
+				Ts:     time.Now().Unix(),
+				L1:     vr.L1,
+				L2:     vr.L2,
+				// L3: intentionally nil
+			}
+			// probeID for 3rd-party: "aliyun-synthetic" (no HMAC secret required;
+			// writes come from the server process, not an external agent).
+			if wErr := a.store.SaveReports(ctx, "aliyun-synthetic", vr.Vantage, []NodeReport{report}); wErr != nil {
+				a.logger.Warn("prober_agent: store write failed",
+					"node", t.NodeID, "isp", vr.Vantage.ISP, "error", wErr)
+			}
+		}
+	}
+	return nil
+}
+
+// Probe implements ProberAgent.  It creates one TCP synthetic-monitoring task
+// per ISP (telecom/unicom/mobile), polls for completion, then maps vendor
+// results to VantageResult.
+//
+// On any API or polling error the ISP vantage is skipped (no result returned,
+// no Redis write) per the degradation contract.
+func (a *AliyunSyntheticAgent) Probe(ctx context.Context, target ProbeTarget) ([]VantageResult, error) {
+	var out []VantageResult
+	for _, isp := range aliyunISP {
+		vr, err := a.probeISP(ctx, target, isp.ispCode, isp.name)
+		if err != nil {
+			cnt := a.failCount.Add(1)
+			a.logger.Warn("prober_agent: ISP probe failed (degraded, no data written)",
+				"node", target.NodeID, "isp", isp.name, "error", err,
+				"consecutive_failures", cnt)
+			// Degradation: skip this vantage this cycle.
+			continue
+		}
+		a.failCount.Store(0) // reset on success
+		out = append(out, vr)
+	}
+	return out, nil
+}
+
+// probeISP creates a single-ISP task, polls for results, and maps to VantageResult.
+func (a *AliyunSyntheticAgent) probeISP(ctx context.Context, target ProbeTarget, ispCode, ispName string) (VantageResult, error) {
+	taskID, err := a.createTask(ctx, target, ispCode)
+	if err != nil {
+		return VantageResult{}, fmt.Errorf("create task: %w", err)
+	}
+
+	result, err := a.pollResult(ctx, taskID)
+	if err != nil {
+		return VantageResult{}, fmt.Errorf("poll result: %w", err)
+	}
+
+	vp := VantagePoint{
+		Country: "CN",
+		Region:  result.Province,
+		ISP:     "3rd-" + ispName,
+	}
+
+	vr := VantageResult{Vantage: vp}
+
+	if result.Availability > 0 && result.ResponseTime > 0 {
+		vr.L1 = L1Result{OK: true, RttMs: int(result.ResponseTime)}
+	} else {
+		vr.L1 = L1Result{OK: false, RttMs: 0}
+	}
+
+	// Alibaba Cloud TCP synthetic does not return a separate TLS phase result;
+	// set L2 only when L1 succeeded and the port is 443 (implies TLS negotiated).
+	if vr.L1.OK && target.Port == 443 {
+		tlsErr := ""
+		if result.ErrorInfo != "" {
+			tlsErr = result.ErrorInfo
+		}
+		vr.L2 = &L2Result{
+			OK:   tlsErr == "",
+			ALPN: "", // vendor does not expose ALPN
+			Err:  tlsErr,
+		}
+	}
+
+	return vr, nil
+}
+
+// --------------------------------------------------------------------------
+// Alibaba Cloud CloudMonitor Synthetic API calls
+// --------------------------------------------------------------------------
+
+// aliyunTaskResult holds the vendor-specific fields returned by
+// DescribeSiteMonitorData.  Only fields needed for VantageResult are kept;
+// vendor-specific identifiers do not escape this file.
+type aliyunTaskResult struct {
+	// Province is the probe node's province name (in Chinese, e.g. "北京").
+	Province string
+	// ResponseTime is the TCP round-trip time in milliseconds.
+	ResponseTime float64
+	// Availability is 100.0 for success, 0.0 for failure.
+	Availability float64
+	// ErrorInfo is a short vendor error string when the probe failed.
+	ErrorInfo string
+}
+
+// createTask calls CreateSiteMonitor and returns the new task ID.
+// address format: "host:port".
+func (a *AliyunSyntheticAgent) createTask(ctx context.Context, target ProbeTarget, ispCode string) (string, error) {
+	ispCity, _ := json.Marshal([]map[string]string{
+		{"City": a.cfg.ProbeCity, "Isp": ispCode},
+	})
+
+	params := map[string]string{
+		"Action":       "CreateSiteMonitor",
+		"TaskName":     fmt.Sprintf("pangolin-%s-%s", target.NodeID, ispCode),
+		"Address":      fmt.Sprintf("%s:%d", target.Host, target.Port),
+		"TaskType":     "TCP",
+		"Interval":     "1",    // one-time task
+		"IspCities":    string(ispCity),
+		"AlertIds":     "",
+	}
+
+	body, err := a.call(ctx, params)
+	if err != nil {
+		return "", err
+	}
+
+	var resp struct {
+		Code    string `json:"Code"`
+		Message string `json:"Message"`
+		Data    struct {
+			TaskID string `json:"TaskId"`
+		} `json:"Data"`
+	}
+	if err := json.Unmarshal(body, &resp); err != nil {
+		return "", fmt.Errorf("parse CreateSiteMonitor response: %w", err)
+	}
+	if resp.Code != "200" {
+		return "", fmt.Errorf("CreateSiteMonitor error code=%s msg=%s", resp.Code, resp.Message)
+	}
+	if resp.Data.TaskID == "" {
+		return "", fmt.Errorf("CreateSiteMonitor returned empty TaskId")
+	}
+	return resp.Data.TaskID, nil
+}
+
+// pollResult polls DescribeSiteMonitorData until the task produces a result
+// or the context / poll timeout is exceeded.
+func (a *AliyunSyntheticAgent) pollResult(ctx context.Context, taskID string) (aliyunTaskResult, error) {
+	deadline := time.Now().Add(a.cfg.PollTimeout)
+	ticker := time.NewTicker(a.cfg.PollInterval)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return aliyunTaskResult{}, ctx.Err()
+		case t := <-ticker.C:
+			if t.After(deadline) {
+				return aliyunTaskResult{}, fmt.Errorf("poll timeout after %s", a.cfg.PollTimeout)
+			}
+			result, ready, err := a.fetchResult(ctx, taskID)
+			if err != nil {
+				return aliyunTaskResult{}, err
+			}
+			if ready {
+				return result, nil
+			}
+		}
+	}
+}
+
+// fetchResult calls DescribeSiteMonitorData once.
+// Returns (result, true, nil) when data is available, or (zero, false, nil) when
+// the task is still running.
+func (a *AliyunSyntheticAgent) fetchResult(ctx context.Context, taskID string) (aliyunTaskResult, bool, error) {
+	now := time.Now()
+	params := map[string]string{
+		"Action":     "DescribeSiteMonitorData",
+		"TaskId":     taskID,
+		"MetricName": "Availability",
+		"Period":     "60",
+		"StartTime":  fmt.Sprintf("%d", now.Add(-2*time.Minute).Unix()*1000),
+		"EndTime":    fmt.Sprintf("%d", now.Unix()*1000),
+	}
+
+	body, err := a.call(ctx, params)
+	if err != nil {
+		return aliyunTaskResult{}, false, err
+	}
+
+	var resp struct {
+		Code    string `json:"Code"`
+		Message string `json:"Message"`
+		Data    struct {
+			List []struct {
+				Maximum      float64 `json:"Maximum"`
+				Minimum      float64 `json:"Minimum"`
+				Average      float64 `json:"Average"`
+				Value        float64 `json:"Value"`
+				Cnt          int     `json:"Cnt"`
+				Province     string  `json:"Province"`
+				Isp          string  `json:"Isp"`
+				ErrorCode    string  `json:"ErrorCode"`
+				ErrorInfo    string  `json:"ErrorInfo"`
+			} `json:"List"`
+		} `json:"Data"`
+	}
+
+	if err := json.Unmarshal(body, &resp); err != nil {
+		return aliyunTaskResult{}, false, fmt.Errorf("parse DescribeSiteMonitorData: %w", err)
+	}
+	if resp.Code != "200" {
+		return aliyunTaskResult{}, false, fmt.Errorf("DescribeSiteMonitorData error code=%s msg=%s", resp.Code, resp.Message)
+	}
+	if len(resp.Data.List) == 0 {
+		// Task still running — no data point yet.
+		return aliyunTaskResult{}, false, nil
+	}
+
+	item := resp.Data.List[0]
+	result := aliyunTaskResult{
+		Province:     item.Province,
+		ResponseTime: item.Average,
+		Availability: item.Value,
+		ErrorInfo:    item.ErrorInfo,
+	}
+	return result, true, nil
+}
+
+// --------------------------------------------------------------------------
+// Alibaba Cloud ACS v1 signing (HMAC-SHA1)
+// --------------------------------------------------------------------------
+
+// call signs and executes one Alibaba Cloud CloudMonitor RPC call.
+// It adds the common authentication parameters, signs the request, and
+// returns the raw response body.
+func (a *AliyunSyntheticAgent) call(ctx context.Context, params map[string]string) ([]byte, error) {
+	// Merge common authentication parameters.
+	p := make(map[string]string, len(params)+8)
+	for k, v := range params {
+		p[k] = v
+	}
+	p["AccessKeyId"] = a.cfg.AccessKeyID
+	p["SignatureMethod"] = "HMAC-SHA1"
+	p["SignatureNonce"] = fmt.Sprintf("%d", time.Now().UnixNano())
+	p["SignatureVersion"] = "1.0"
+	p["Timestamp"] = time.Now().UTC().Format("2006-01-02T15:04:05Z")
+	p["Format"] = "JSON"
+	p["Version"] = "2019-01-01"
+
+	// Build sorted query string for signing.
+	keys := make([]string, 0, len(p))
+	for k := range p {
+		keys = append(keys, k)
+	}
+	sort.Strings(keys)
+
+	var parts []string
+	for _, k := range keys {
+		parts = append(parts, percentEncode(k)+"="+percentEncode(p[k]))
+	}
+	queryStr := strings.Join(parts, "&")
+
+	// ACS v1 string to sign: "POST\n%2F\n" + percent-encoded sorted query.
+	stringToSign := "POST\n%2F\n" + percentEncode(queryStr)
+
+	// HMAC-SHA1 with key = AccessKeySecret + "&".
+	//nolint:gosec // required by Alibaba Cloud ACS v1 spec
+	mac := hmac.New(sha1.New, []byte(a.cfg.AccessKeySecret+"&"))
+	mac.Write([]byte(stringToSign))
+	sig := base64.StdEncoding.EncodeToString(mac.Sum(nil))
+
+	// Build final query string including the signature.
+	finalQuery := queryStr + "&Signature=" + percentEncode(sig)
+
+	reqURL := a.baseURL + "?" + finalQuery
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, reqURL, nil)
+	if err != nil {
+		return nil, fmt.Errorf("build request: %w", err)
+	}
+	req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
+
+	resp, err := a.httpClient.Do(req)
+	if err != nil {
+		return nil, fmt.Errorf("http call: %w", err)
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode == http.StatusTooManyRequests || resp.StatusCode == http.StatusServiceUnavailable {
+		return nil, fmt.Errorf("rate-limited or unavailable: HTTP %d", resp.StatusCode)
+	}
+
+	rawBody, err := io.ReadAll(io.LimitReader(resp.Body, 512*1024))
+	if err != nil {
+		return nil, fmt.Errorf("read response: %w", err)
+	}
+	return rawBody, nil
+}
+
+// percentEncode is RFC 3986 percent-encoding as required by Alibaba Cloud ACS v1.
+// It encodes all characters except unreserved ones (A-Z a-z 0-9 - _ . ~).
+func percentEncode(s string) string {
+	var b strings.Builder
+	for _, c := range []byte(s) {
+		if isUnreserved(c) {
+			b.WriteByte(c)
+		} else {
+			fmt.Fprintf(&b, "%%%02X", c)
+		}
+	}
+	return b.String()
+}
+
+func isUnreserved(c byte) bool {
+	return (c >= 'A' && c <= 'Z') ||
+		(c >= 'a' && c <= 'z') ||
+		(c >= '0' && c <= '9') ||
+		c == '-' || c == '_' || c == '.' || c == '~'
+}