部署与对接

在配置后台 → 拨测节点新建节点，会得到一条完整命令，直接粘到目标机器执行即可。 也可以在后台用「远程安装」填 SSH 信息一键装好。

手动执行的话：

# 目标机器上执行，需要 root
curl -fsSL HUBURL/static/install-node.sh -o install-node.sh
bash install-node.sh复制

脚本会先做环境检测（Python 3.7+、标准库完整性、systemd、出网、磁盘）， 再验证中枢连通性和 token，全部通过才开始写文件。 token 错的话连安装目录都不会创建——不留半拉子状态。

批量铺节点（非交互）

bash install-node.sh \
  --hub HUBURL \
  --token nt_xxxxxxxx \
  --node 广东电信-深圳 \
  --province 广东 --carrier 电信 --net-type broadband \
  --interval 300 --yes复制

其它动作

bash install-node.sh --status      # 运行状态和最近日志
bash install-node.sh --test       # 跑一次探测，不改配置
bash install-node.sh --upgrade    # 升级代码，保留配置，失败自动回滚
bash install-node.sh --uninstall  # 卸载复制

省份和运营商留空会自动识别。固网机器建议显式指定（IP 库对机房 IP 的判断经常不准）， 手机和动态 IP 的机器则应该留空自动识别。

容器、Alpine、或者不想用安装脚本时：

# 只需要两个文件，纯标准库，不用 pip 装任何东西
mkdir -p /opt/gfw-probe && cd /opt/gfw-probe
curl -fsSL HUBURL/static/gfw_probe.py -o gfw_probe.py
curl -fsSL HUBURL/static/agent.py -o agent.py

export PROBE_HUB=HUBURL
export PROBE_AGENT_TOKEN=nt_xxxxxxxx
python3 agent.py --node 节点名 --interval 300复制

零依赖是刻意的设计——节点机器往往很干净，装不了东西，也不该为了监控去污染它。

不带中枢，纯本地排查

python3 gfw_probe.py -d your-domain.com --ip 203.0.113.10 --rounds 5复制

蜂窝网络的封锁策略和固网经常不一样，而你的用户绝大多数在手机上。 只在机房放固网节点，等于用一个看不见问题的视角去监控一个正在出问题的链路。

安装后打开 App，填三样（都在配置后台 → 拨测节点里）：中枢地址、 节点 Token、节点标识。然后必须点「申请电池优化豁免」—— 跳过这步的话息屏几分钟系统就冻结进程了，节点表现为"时有时无地上报"， 比完全不上报更难排查。国产 ROM 还要去自启动管理里单独允许。

自己编译

工程源码在 android/，推到 GitHub 后 Actions 页面手动触发 Build APK，产物在 Artifacts 里。不配签名密钥也能出包（debug 签名，一样能装）。

Termux 方案（备选）

pkg install python
curl -fsSL HUBURL/static/install-node.sh | bash -s -- \
  --hub HUBURL --token nt_xxx --node 广东移动-手机1复制

原生 App 比 Termux 强在运营商识别直接问系统（TelephonyManager）， 不靠 IP 地理库猜——IP 库对蜂窝网络经常判错，移动的出口 IP 可能落在别的省。

覆盖面 >> 数量。封锁策略是按省份 + 运营商下发的， 同机房堆十台的信息量不如三个省各放一台。优先级：

1. 三大运营商 × 蜂窝（移动/联通/电信各一台安卓机）—— 用户主力在这
2. 境外对照节点 —— 必需，用来区分「IP 被封」和「源站故障」
3. 重点省份固网（业务量最大的 3–5 个省）
4. 其余省份按业务量补

探测台的覆盖矩阵会直接列出还缺哪些「省份 × 运营商」格子。

规范串（顺序和分隔符必须完全一致）：

canonical = METHOD + "\n"
          + PATH + "\n"
          + SORTED_QUERY + "\n"      // k=v 按字符串排序，& 连接
          + TIMESTAMP + "\n"          // unix 秒
          + NONCE + "\n"              // ≥16 字符随机
          + SHA256_HEX(BODY)          // 无 body 则为空串的哈希

signature = HMAC_SHA256(secret, canonical) → hex复制

请求头四个：X-Api-Key、X-Timestamp、 X-Nonce、X-Signature。

• 时间戳窗口 ±300 秒，超出直接 401。三方服务器要校准 ntp
• nonce 600 秒内不可复用，防重放
• query 排序后参与签名，调换参数顺序会失败
• body 以 sha256 参与，篡改会失败
• 签名比对用常数时间，避免时序侧信道

纯标准库，不需要装任何东西。

# probe_client.py
import hashlib, hmac, json, secrets, time
import urllib.request, urllib.error, urllib.parse


class ProbeClient:
    def __init__(self, base_url, api_key, secret, timeout=10):
        self.base = base_url.rstrip("/")
        self.key, self.secret, self.timeout = api_key, secret, timeout

    def _canonical(self, method, path, query, ts, nonce, body):
        pairs = sorted("%s=%s" % (urllib.parse.quote(str(k), safe=""),
                                    urllib.parse.quote(str(v), safe=""))
                       for k, v in (query or {}).items())
        return "\n".join([method.upper(), path, "&".join(pairs), str(ts), nonce,
                          hashlib.sha256(body or b"").hexdigest()])

    def _request(self, method, path, query=None, payload=None):
        body = json.dumps(payload).encode() if payload is not None else b""
        ts, nonce = int(time.time()), secrets.token_hex(16)
        sig = hmac.new(self.secret.encode(),
                       self._canonical(method, path, query, ts, nonce, body).encode(),
                       hashlib.sha256).hexdigest()

        url = self.base + path
        if query:
            # 发送顺序必须和签名时一致
            url += "?" + "&".join(sorted(
                "%s=%s" % (urllib.parse.quote(str(k), safe=""),
                            urllib.parse.quote(str(v), safe=""))
                for k, v in query.items()))

        req = urllib.request.Request(url, data=body or None, method=method.upper(),
            headers={"X-Api-Key": self.key, "X-Timestamp": str(ts),
                     "X-Nonce": nonce, "X-Signature": sig,
                     "Content-Type": "application/json"})
        try:
            with urllib.request.urlopen(req, timeout=self.timeout) as r:
                return json.loads(r.read().decode())
        except urllib.error.HTTPError as e:
            detail = e.read().decode("utf-8", "replace")
            if e.code == 401 and "时间戳" in detail:
                detail += "（服务器时钟不同步，检查 ntp）"
            raise RuntimeError("HTTP %d: %s" % (e.code, detail))

    # ---- 接口 ----
    def check(self, domains, window=3600):
        """批量查询，最多 50 个。优先用这个，别循环调单域名"""
        return self._request("POST", "/api/v3p/check",
                             {"window": window}, {"domains": list(domains)})

    def domain(self, name, detail=False, window=3600):
        return self._request("GET", "/api/v3p/domains/" + urllib.parse.quote(name),
                             {"window": window, "detail": str(detail).lower()})

    def entry(self, window=3600):
        return self._request("GET", "/api/v3p/entry", {"window": window})

    def domains(self, window=3600):
        return self._request("GET", "/api/v3p/domains", {"window": window})复制

推荐用法：服务端定时拉取，落静态快照

import json, os, tempfile

c = ProbeClient(os.environ["PROBE_HUB"],
                os.environ["PROBE_KEY"], os.environ["PROBE_SECRET"])
try:
    pool = c.entry()
    # 原子写：先写临时文件再 rename，避免页面读到写了一半的文件
    fd, tmp = tempfile.mkstemp(dir="/var/www/static")
    with os.fdopen(fd, "w") as f:
        json.dump(pool, f, ensure_ascii=False)
    os.replace(tmp, "/var/www/static/entry.json")
except Exception as e:
    # 拉取失败保留上一份快照。宁可给落地页旧数据，也不能给空池子
    print("probe entry 拉取失败:", e)复制

PHP 7.2+，只用到 curl 和 hash 扩展。完整版在 sdk/php/ProbeClient.php。

<?php
class ProbeClient
{
    private $base, $apiKey, $secret, $timeout;

    public function __construct($baseUrl, $apiKey, $secret, $timeout = 10) {
        $this->base    = rtrim($baseUrl, '/');
        $this->apiKey  = $apiKey;
        $this->secret  = $secret;
        $this->timeout = $timeout;
    }

    private function canonical($method, $path, array $query, $ts, $nonce, $body) {
        $pairs = [];
        foreach ($query as $k => $v) {
            $pairs[] = rawurlencode($k) . '=' . rawurlencode((string)$v);
        }
        sort($pairs, SORT_STRING);          // 服务端也按字符串排序
        return implode("\n", [
            strtoupper($method), $path, implode('&', $pairs),
            (string)$ts, $nonce, hash('sha256', $body === null ? '' : $body),
        ]);
    }

    private function request($method, $path, array $query = [], $jsonBody = null) {
        $body  = $jsonBody === null ? '' : json_encode($jsonBody, JSON_UNESCAPED_UNICODE);
        $ts    = time();
        $nonce = bin2hex(random_bytes(16));
        $sig   = hash_hmac('sha256',
            $this->canonical($method, $path, $query, $ts, $nonce, $body), $this->secret);

        $url = $this->base . $path;
        if ($query) {
            $pairs = [];
            foreach ($query as $k => $v) $pairs[] = rawurlencode($k).'='.rawurlencode((string)$v);
            sort($pairs, SORT_STRING);
            $url .= '?' . implode('&', $pairs);
        }

        $ch = curl_init($url);
        curl_setopt_array($ch, [
            CURLOPT_RETURNTRANSFER => true,
            CURLOPT_CUSTOMREQUEST  => strtoupper($method),
            CURLOPT_HTTPHEADER     => [
                'X-Api-Key: '   . $this->apiKey,
                'X-Timestamp: ' . $ts,
                'X-Nonce: '     . $nonce,
                'X-Signature: ' . $sig,
                'Content-Type: application/json',
            ],
            CURLOPT_TIMEOUT        => $this->timeout,
            CURLOPT_CONNECTTIMEOUT => 5,
        ]);
        if ($body !== '') curl_setopt($ch, CURLOPT_POSTFIELDS, $body);

        $resp = curl_exec($ch);
        $code = curl_getinfo($ch, CURLINFO_HTTP_CODE);
        $err  = curl_error($ch);
        curl_close($ch);

        if ($resp === false) throw new RuntimeException("请求失败: $err");
        $data = json_decode($resp, true);
        if ($code >= 400) {
            $d = is_array($data) && isset($data['detail']) ? $data['detail'] : $resp;
            throw new RuntimeException("HTTP $code: $d");
        }
        return $data;
    }

    public function check(array $domains, $window = 3600) {
        return $this->request('POST', '/api/v3p/check',
            ['window' => $window], ['domains' => array_values($domains)]);
    }
    public function entry($window = 3600) {
        return $this->request('GET', '/api/v3p/entry', ['window' => $window]);
    }
    public function domain($d, $detail = false, $window = 3600) {
        return $this->request('GET', '/api/v3p/domains/' . rawurlencode($d),
            ['window' => $window, 'detail' => $detail ? 'true' : 'false']);
    }
}复制

cron 拉快照

$c = new ProbeClient(getenv('PROBE_HUB'), getenv('PROBE_KEY'), getenv('PROBE_SECRET'));
try {
    $pool = $c->entry();
    $tmp = '/var/www/static/entry.json.tmp';
    file_put_contents($tmp, json_encode($pool, JSON_UNESCAPED_UNICODE), LOCK_EX);
    rename($tmp, '/var/www/static/entry.json');   // 原子替换
} catch (Exception $e) {
    error_log('probe entry 拉取失败: ' . $e->getMessage());  // 保留旧快照
}复制

Node 14+，零依赖。完整版在 sdk/node/probe-client.js。

const crypto = require('crypto');
const https = require('https');
const http = require('http');
const { URL } = require('url');

class ProbeClient {
  constructor({ baseUrl, apiKey, secret, timeout = 10000 }) {
    this.base = baseUrl.replace(/\/+$/, '');
    this.apiKey = apiKey;
    this.secret = secret;
    this.timeout = timeout;
  }

  _canonical(method, path, query, ts, nonce, body) {
    const pairs = Object.entries(query || {})
      .map(([k, v]) => `${encodeURIComponent(k)}=${encodeURIComponent(String(v))}`)
      .sort();
    const bodyHash = crypto.createHash('sha256').update(body || '').digest('hex');
    return [method.toUpperCase(), path, pairs.join('&'),
            String(ts), nonce, bodyHash].join('\n');
  }

  _request(method, path, query = {}, jsonBody = null) {
    const body = jsonBody === null ? '' : JSON.stringify(jsonBody);
    const ts = Math.floor(Date.now() / 1000);
    const nonce = crypto.randomBytes(16).toString('hex');
    const sig = crypto.createHmac('sha256', this.secret)
      .update(this._canonical(method, path, query, ts, nonce, body))
      .digest('hex');

    const qs = Object.entries(query)
      .map(([k, v]) => `${encodeURIComponent(k)}=${encodeURIComponent(String(v))}`)
      .sort().join('&');
    const url = new URL(this.base + path + (qs ? '?' + qs : ''));
    const lib = url.protocol === 'https:' ? https : http;

    return new Promise((resolve, reject) => {
      const req = lib.request(url, {
        method: method.toUpperCase(),
        headers: {
          'X-Api-Key': this.apiKey,
          'X-Timestamp': String(ts),
          'X-Nonce': nonce,
          'X-Signature': sig,
          'Content-Type': 'application/json',
          'Content-Length': Buffer.byteLength(body),
        },
        timeout: this.timeout,
      }, (res) => {
        let buf = '';
        res.on('data', d => { buf += d; });
        res.on('end', () => {
          let data;
          try { data = JSON.parse(buf); }
          catch { return reject(new Error('返回不是合法 JSON: ' + buf.slice(0, 200))); }
          if (res.statusCode >= 400) {
            let msg = data.detail || buf;
            if (res.statusCode === 401 && /时间戳/.test(msg)) msg += '（时钟不同步，检查 ntp）';
            return reject(new Error(`HTTP ${res.statusCode}: ${msg}`));
          }
          resolve(data);
        });
      });
      req.on('timeout', () => req.destroy(new Error('请求超时')));
      req.on('error', reject);
      if (body) req.write(body);
      req.end();
    });
  }

  check(domains, { window = 3600 } = {}) {
    return this._request('POST', '/api/v3p/check', { window }, { domains });
  }
  entry({ window = 3600 } = {}) {
    return this._request('GET', '/api/v3p/entry', { window });
  }
  domain(d, { detail = false, window = 3600 } = {}) {
    return this._request('GET', `/api/v3p/domains/${encodeURIComponent(d)}`,
                          { window, detail: String(detail) });
  }
}

module.exports = { ProbeClient };复制

curl（用 shell 算签名，适合排查）

#!/usr/bin/env bash
HUB="HUBURL"
KEY="pk_xxxxx"
SECRET="你的secret"
PATH_="/api/v3p/domains"
QUERY="window=3600"

TS=$(date +%s)
NONCE=$(head -c16 /dev/urandom | xxd -p)
BODY_HASH=$(printf '' | sha256sum | cut -d' ' -f1)
CANON=$(printf 'GET\n%s\n%s\n%s\n%s\n%s' "$PATH_" "$QUERY" "$TS" "$NONCE" "$BODY_HASH")
SIG=$(printf '%s' "$CANON" | openssl dgst -sha256 -hmac "$SECRET" -hex | awk '{print $NF}')

curl -s "$HUB$PATH_?$QUERY" \
  -H "X-Api-Key: $KEY" \
  -H "X-Timestamp: $TS" \
  -H "X-Nonce: $NONCE" \
  -H "X-Signature: $SIG"复制

Go

func sign(secret, method, path, query string, ts int64, nonce string, body []byte) string {
    h := sha256.Sum256(body)
    canonical := strings.Join([]string{
        strings.ToUpper(method), path, query,
        strconv.FormatInt(ts, 10), nonce, hex.EncodeToString(h[:]),
    }, "\n")
    m := hmac.New(sha256.New, []byte(secret))
    m.Write([]byte(canonical))
    return hex.EncodeToString(m.Sum(nil))
}
// query 传排序后的 "a=1&b=2"；无 body 时 body 传 []byte{}复制

判定值

对外返回只到省份/运营商粒度，不含节点名和节点 IP。

<script src="/js/entry-probe.js"></script>
<script>
EntryProbe.start({
  // 硬编码兜底池：永远有效，即使所有快照源都挂了
  fallback: ['a1.example.com', 'a2.example.net', 'a3.example.org'],
  // 快照地址，不是中枢地址。至少两个不相关的域名
  snapshots: [
    'https://cdn-a.example.com/entry.json',
    'https://cdn-b.example.net/entry.json'
  ],
  probePath: '/favicon.ico',   // 每个域名都必须存在的小文件
  timeout: 2500,
  onPicked: function (host) {
    location.replace('https://' + host + location.pathname + location.search);
  },
  onFail: function () {
    document.getElementById('tip').textContent = '当前网络暂时无法连接，请稍后重试或使用 App';
  }
});
</script>复制

已有「线路列表 + 延迟显示」的页面

EntryProbe.measureAll(['line1.example.com', 'line2.example.net'])
  .then(rows => rows.forEach(r =>
    render(r.host, r.ok ? r.ms + 'ms' : '不可用')));复制

用图片而不是 fetch 探测是有意的：不受 CORS 限制、老 webview 兼容、 被墙时表现为超时/error 正好是需要的信号。

下载脚本：HUBURL/static/entry-probe.js