背景

家庭宽带处于 NAT 环境下，无法直接获得固定公网端口。使用 OpenWrt 上的 NATMap 插件可以实现 NAT 穿透，动态获取公网 IP 和端口。每当公网端口变化时，NATMap 支持调用自定义脚本（通知脚本）来自动完成防火墙规则更新、服务配置同步等任务。

本文记录了两个通知脚本和一个辅助 Python 脚本的实现思路与内容。

脚本参数说明

NATMap 调用通知脚本时，会按顺序传入以下参数：

参数位置 变量 含义
$1 PUBLIC_IP 公网 IP 地址
$2 PUBLIC_PORT 公网端口
$3 （内部 IP） 一般不使用
$4 LOCAL_PORT 本地绑定端口（用作规则唯一标识）

使用 $4（本地端口）而不是 $2（公网端口）作为防火墙规则 ID，是为了保证每次端口变化时能精准定位并覆盖旧规则，避免规则堆积。

脚本一：openport_openlist.sh

用途

用于 OpenList 网盘服务的端口映射场景。每当公网端口变化时：

  1. 更新 OpenWrt 防火墙的 IPv4 DNAT 转发规则，将公网端口流量转发到内网 OpenList 服务（固定端口 5244）。
  2. 调用 Python 脚本，将 OpenList 管理后台中对应存储的 URL 同步更新为新的公网地址。

脚本内容

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
#!/bin/sh

# --- 获取参数 ---
PUBLIC_IP=$1
PUBLIC_PORT=$2

LOCAL_IP="192.168.7.108"  # 手动设置转发目标
LOCAL_PORT=$4  # 核心：用本地绑定端口作为唯一标识

# --- 生成唯一的防火墙规则 ID ---
V4_RULE_ID="natmap_v4_${LOCAL_PORT}"

echo "正在为本地端口 $LOCAL_PORT 处理映射: $PUBLIC_PORT"

# --- 动态更新 IPv4 端口转发 (DNAT) ---
uci -q delete firewall.${V4_RULE_ID}
uci set firewall.${V4_RULE_ID}=redirect
uci set firewall.${V4_RULE_ID}.name="NATMap_V4_${LOCAL_PORT}_Ext_${PUBLIC_PORT}"
uci set firewall.${V4_RULE_ID}.src='wan'
uci set firewall.${V4_RULE_ID}.src_dport="$LOCAL_PORT"
uci set firewall.${V4_RULE_ID}.dest='lan'
uci set firewall.${V4_RULE_ID}.dest_ip="$LOCAL_IP"
uci set firewall.${V4_RULE_ID}.dest_port="5244"  # OpenList 固定端口
uci set firewall.${V4_RULE_ID}.proto='tcp'
uci set firewall.${V4_RULE_ID}.target='DNAT'

# 应用防火墙配置
uci commit firewall
/etc/init.d/firewall restart

# === 同步远端 URL 到 OpenList 存储 ===
python3 /usr/lib/natmap/update_openlist.py --path /NEC7 --url https://your-domain.example.com:$PUBLIC_PORT

关键点

  • dest_port 写死为 5244，是 OpenList 默认监听端口，与公网端口无关。
  • IPv6 放行规则部分已注释掉，按需启用。
  • 最后一行调用辅助 Python 脚本，将新的公网 URL 同步回 OpenList 管理端。

脚本二：openport_qb.sh

用途

用于 qBittorrent 的端口映射场景。每当公网端口变化时：

  1. 更新 OpenWrt 防火墙的 IPv4 DNAT 转发规则。
  2. 更新 IPv6 放行规则（允许 IPv6 直连）。
  3. 通过 qBittorrent Web API 将监听端口同步为当前公网端口，保证 BT 做种可连通。

脚本内容

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
#!/bin/sh

# --- 获取参数 ---
PUBLIC_IP=$1
PUBLIC_PORT=$2

LOCAL_IP="192.168.7.108"
LOCAL_PORT=$4

V4_RULE_ID="natmap_v4_${LOCAL_PORT}"
V6_RULE_ID="natmap_v6_${LOCAL_PORT}"

echo "正在为本地端口 $LOCAL_PORT 处理映射: $PUBLIC_PORT"

# --- 动态更新 IPv4 端口转发 (DNAT) ---
uci -q delete firewall.${V4_RULE_ID}
uci set firewall.${V4_RULE_ID}=redirect
uci set firewall.${V4_RULE_ID}.name="NATMap_V4_${LOCAL_PORT}_Ext_${PUBLIC_PORT}"
uci set firewall.${V4_RULE_ID}.src='wan'
uci set firewall.${V4_RULE_ID}.src_dport="$LOCAL_PORT"
uci set firewall.${V4_RULE_ID}.dest='lan'
uci set firewall.${V4_RULE_ID}.dest_ip="$LOCAL_IP"
uci set firewall.${V4_RULE_ID}.dest_port="$PUBLIC_PORT"
uci set firewall.${V4_RULE_ID}.proto='tcp'
uci set firewall.${V4_RULE_ID}.target='DNAT'

# --- 动态更新 IPv6 放行规则 ---
uci -q delete firewall.${V6_RULE_ID}
uci set firewall.${V6_RULE_ID}=rule
uci set firewall.${V6_RULE_ID}.name="NATMap_V6_${LOCAL_PORT}_Port_${PUBLIC_PORT}"
uci set firewall.${V6_RULE_ID}.src='wan'
uci set firewall.${V6_RULE_ID}.dest='lan'
uci set firewall.${V6_RULE_ID}.dest_port="$PUBLIC_PORT"
# uci set firewall.${V6_RULE_ID}.proto='tcp'
uci set firewall.${V6_RULE_ID}.family='ipv6'
uci set firewall.${V6_RULE_ID}.target='ACCEPT'

# 应用防火墙配置
uci commit firewall
/etc/init.d/firewall restart

# === qBittorrent 配置 ===
QB_IP="192.168.7.108"
QB_WEB_PORT="8080"
QB_USER="******"       # qB 登录用户名（已打码）
QB_PASS="********"     # qB 登录密码（已打码）
QB_URL="http://$QB_IP:$QB_WEB_PORT"

echo "正在尝试更新 qBittorrent 监听端口为: $PUBLIC_PORT"

# 登录获取 SID，添加 Referer 头部以绕过 CSRF 检查
LOGIN_HDR=$(curl -s -i -X POST \
    --data "username=$QB_USER&password=$QB_PASS" \
    --header "Referer: $QB_URL" \
    "$QB_URL/api/v2/auth/login")

SID=$(echo "$LOGIN_HDR" | grep -o 'SID=[^;]*' | head -n 1)

if [ -z "$SID" ]; then
    echo "错误：无法获取 SID，请检查登录信息或 qB 设置。"
    exit 1
fi

# 使用 SID 发送端口修改请求
RESULT=$(curl -s -X POST \
    -b "$SID" \
    --header "Referer: $QB_URL" \
    --data-urlencode "json={\"listen_port\": $PUBLIC_PORT}" \
    "$QB_URL/api/v2/app/setPreferences")

if [ "$RESULT" = "" ]; then
    echo "成功：qBittorrent 端口已同步为 $PUBLIC_PORT"
else
    echo "失败：qB 返回了非预期响应：$RESULT"
fi

关键点

  • qBittorrent 新版本启用了 CSRF 防护，登录和修改请求都需要携带 Referer 头部，否则会被拒绝。
  • SID 通过解析登录响应的 Cookie 头获取，全程存在变量中，不写临时文件，更安全。
  • dest_port 使用 $PUBLIC_PORT，即 qBittorrent 监听端口与对外映射端口保持一致。

辅助脚本：update_openlist.py

用途

由 openport_openlist.sh 调用，负责自动登录 OpenList 管理端，查找指定挂载路径的存储，将其 URL 字段更新为新的公网地址，完成后登出。

调用方式

1
python3 /usr/lib/natmap/update_openlist.py --path /NEC7 --url https://your-domain.example.com:$PUBLIC_PORT

脚本内容

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
import http.client
import json
import argparse
import sys

def parse_arguments():
    parser = argparse.ArgumentParser(description="自动登录并更新指定路径存储的 URL。")
    parser.add_argument('--url', type=str, required=True, help="要更新的新 URL")
    parser.add_argument('--path', type=str, required=True, help="存储的挂载路径")
    return parser.parse_args()

def main():
    args = parse_arguments()
    new_url = args.url
    target_path = args.path

    conn = http.client.HTTPSConnection("pan.example.com")  # OpenList 域名（已打码）
    token = None

    # --- 登录 ---
    print(">>> 正在登录...")
    login_payload = json.dumps({
        "username": "admin",
        "password": "**********"  # OpenList 登录密码（已打码）
    }).encode('utf-8')

    try:
        conn.request("POST", "/api/auth/login", login_payload, {'Content-Type': 'application/json'})
        res = conn.getresponse()
        response_dict = json.loads(res.read().decode("utf-8"))

        if response_dict.get("code") != 200:
            print(f"❌ 登录失败：{response_dict.get('message', '未知错误')}")
            sys.exit(1)

        token = response_dict["data"]["token"]
        print(f"✅ 登录成功，Token: {token[:10]}...")

    except Exception as e:
        print(f"❌ 登录请求错误: {e}")
        sys.exit(1)

    headers = {'Authorization': token}

    try:
        # 获取存储列表
        print("\n>>> 获取存储列表...")
        conn.request("GET", "/api/admin/storage/list?page=1&per_page=10", None, headers)
        res = conn.getresponse()
        response_dict = json.loads(res.read().decode("utf-8"))

        storage_info = next(
            (item for item in response_dict["data"]["content"] if item["mount_path"] == target_path),
            None
        )

        if storage_info is not None:
            target_id = storage_info['id']
            print(f"✅ 找到路径 '{target_path}'，ID: {target_id}")

            # 修改 addition 字段中的 url
            addition_dict = json.loads(storage_info["addition"])
            print(f">>> 将 URL 从 '{addition_dict.get('url')}' 更新为 '{new_url}'")
            addition_dict["url"] = new_url
            storage_info["addition"] = json.dumps(addition_dict)

            # 发送更新请求
            print("\n>>> 正在更新存储...")
            conn.request(
                "POST", "/api/admin/storage/update",
                json.dumps(storage_info).encode('utf-8'),
                {'Authorization': token, 'Content-Type': 'application/json'}
            )
            res = conn.getresponse()
            response_dict = json.loads(res.read().decode("utf-8"))

            if response_dict.get("code") != 200:
                raise Exception(f"更新失败：{response_dict.get('message', '未知错误')}")
            else:
                print("✅ 存储 URL 更新成功。")
        else:
            print(f"⚠️ 未找到路径 '{target_path}'，跳过更新。")

    except Exception as e:
        print(f"❌ 业务错误: {e}")

    finally:
        # 登出
        print("\n>>> 正在登出...")
        try:
            conn.request("GET", "/api/auth/logout", None, headers)
            res = conn.getresponse()
            response_dict = json.loads(res.read().decode("utf-8"))
            if response_dict.get("code") == 200:
                print("✅ 成功登出。")
            else:
                print(f"⚠️ 登出失败：{response_dict.get('message', '未知错误')}")
        except Exception as e:
            print(f"⚠️ 登出请求错误: {e}")

if __name__ == "__main__":
    main()

设计要点

  • 使用标准库 http.client，无需额外安装依赖，适合 OpenWrt 精简环境。
  • 采用 try...finally 结构，确保无论业务成功与否，登录态都会被清理（登出）。
  • 通过 --path 参数精准匹配挂载路径，支持多存储场景下的定向更新。

文件部署位置

文件 路径
openport_openlist.sh /usr/lib/natmap/openport_openlist.sh
openport_qb.sh /usr/lib/natmap/openport_qb.sh
update_openlist.py /usr/lib/natmap/update_openlist.py

在 NATMap 插件的对应实例配置中，将「通知脚本」字段填写为对应脚本的完整路径即可。

小结

脚本 应用场景 主要动作
openport_openlist.sh OpenList 网盘 更新 IPv4 DNAT + 同步 OpenList URL
openport_qb.sh qBittorrent 更新 IPv4 DNAT + IPv6 放行 + 同步 qB 端口
update_openlist.py OpenList API 登录 → 查找存储 → 更新 URL → 登出

整套方案实现了公网端口变化后的全自动同步，无需人工干预。