From 9dc2dde59f95c314050e84634471c789b4848848 Mon Sep 17 00:00:00 2001 From: chakcy <947105045@qq.com> Date: Thu, 17 Sep 2026 11:23:34 +0800 Subject: [PATCH] =?UTF-8?q?modify=EF=BC=9A=20=E6=B6=88=E6=81=AF=E6=A8=A1?= =?UTF-8?q?=E5=9D=97?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 3 + deploy/mqtt/.gitignore | 3 + deploy/mqtt/config/acl | 50 +++ deploy/mqtt/config/mosquitto.conf | 62 +++ deploy/mqtt/docker-compose.yml | 31 ++ docs/mqtt-guide.md | 703 ++++++++++++++++++++++++++++++ 6 files changed, 852 insertions(+) create mode 100644 deploy/mqtt/.gitignore create mode 100644 deploy/mqtt/config/acl create mode 100644 deploy/mqtt/config/mosquitto.conf create mode 100644 deploy/mqtt/docker-compose.yml create mode 100644 docs/mqtt-guide.md diff --git a/README.md b/README.md index bd60050..cbb9fe6 100644 --- a/README.md +++ b/README.md @@ -602,6 +602,9 @@ svfs.messages.subscribe((message) => { * 断线重连交给 paho（指数退避 1→30 秒），重连成功后自动重新订阅； * 连接状态、每个通道的收发与丢弃计数都进诊断日志（含 `forbidden`/`dropped`）。 +生产落地（Docker 起 broker、认证/ACL/TLS 怎么选、服务端发布代码、离线补齐接口的契约） +见 [`docs/mqtt-guide.md`](docs/mqtt-guide.md)；可直接跑的 broker 配置在 `deploy/mqtt/`。 + ### 发布 → 安装的完整链路 发布端用现成的 `svfs publish`，客户端用 manifest 安装（对象逐个 sha256 校验）： diff --git a/deploy/mqtt/.gitignore b/deploy/mqtt/.gitignore new file mode 100644 index 0000000..729c05a --- /dev/null +++ b/deploy/mqtt/.gitignore @@ -0,0 +1,3 @@ +# 密码文件与证书都是本机生成的，不进仓库 +config/passwd +config/certs/ diff --git a/deploy/mqtt/config/acl b/deploy/mqtt/config/acl new file mode 100644 index 0000000..53cfe81 --- /dev/null +++ b/deploy/mqtt/config/acl @@ -0,0 +1,50 @@ +# svfs 消息模块的 ACL 示例（最小权限）。默认**未启用**——要用就在 +# mosquitto.conf 里解开 allow_anonymous false / password_file / acl_file 三行， +# 否则这份文件不会生效。 +# +# 规则先记两条，踩坑都出在这儿： +# +# * 这里的 <用户名> 是 **MQTT 连接的 username**。客户端默认用 svfs 登录名 +# （settings.mqtt_config：mqtt_username 留空 = 登录名），所以通常两者一致； +# 在设置里覆盖过 broker 账号时，以覆盖后的为准。 +# * 组织标识（下面的 default）要**换成你的 SVFS_ORG**，且不能写成通配—— +# 多租户隔离靠的就是这一段。用户名里有特殊字符时按百分号编码填 +# （拿 svfs_client.topics.topic_segment 算一下，例如「张三」→ %E5%BC%A0%E4%B8%89）。 + +# ---- 匿名段：第一个 user 之前的规则属于匿名连接 ---------------------- +# +# 开了 allow_anonymous false 后这一段就没用了（示例服务端届时要用下面的 +# svfs-server 账号发布）。留着是为了「允许匿名发布、但读取仍受控」的折中场景。 +topic write svfs/default/user/+/message +topic write svfs/default/user/+/plugin/+ +topic write svfs/default/broadcast/# + +# ---- 服务端发布账号 -------------------------------------------------- +# +# 服务端要能给任意用户写消息；它只需要写，不需要读。 +user svfs-server +topic write svfs/default/user/+/message +topic write svfs/default/user/+/plugin/+ +topic write svfs/default/broadcast/# +topic write svfs/default/broadcast/plugin/+ + +# ---- 客户端：每人只读自己那两条订阅主题 ------------------------------ +# +# 客户端登录后订阅的是 svfs//user/<自己>/# 与 svfs//broadcast/#， +# 所以只给这些就够（读不到别人的消息、也没有写权限）。 +user alice +topic read svfs/default/user/alice/# +topic read svfs/default/broadcast/# + +user bob +topic read svfs/default/user/bob/# +topic read svfs/default/broadcast/# + +# ---- 等价的通配写法（可选） ------------------------------------------ +# +# 用户多了不想一个个写，可以用 pattern：%u 会被替换成连接用的用户名。 +# 前提是 %u 独占一个主题层级（不能写成 svfs/default/user/%u-2/# 这种）。 +# 注意 pattern 对所有用户生效（包括匿名与 svfs-server），所以这里只给读。 +# +# pattern read svfs/default/user/%u/# +# pattern read svfs/default/broadcast/# diff --git a/deploy/mqtt/config/mosquitto.conf b/deploy/mqtt/config/mosquitto.conf new file mode 100644 index 0000000..999c80a --- /dev/null +++ b/deploy/mqtt/config/mosquitto.conf @@ -0,0 +1,62 @@ +# svfs 消息模块用的 mosquitto 配置（挂到容器的 /mosquitto/config/mosquitto.conf）。 +# +# 默认这份是「内网匿名」：能立刻跑起来，但任何能连到 1883 的人都能收发消息， +# 只适合内网/开发。下面另有「认证 + ACL」与「TLS」两个注释段，要用时解开， +# 完整说明（含客户端怎么配）见 docs/mqtt-guide.md 的「二、Docker 部署 broker」。 + +# ---- 监听与放行 ------------------------------------------------------ +# +# mosquitto 2.x 的规矩：一旦定义了 listener，allow_anonymous 的默认值就是 +# false——不显式写这一行，客户端会被直接拒连（"能连上端口却连不上 broker" +# 的典型来源）。同理，不定义 listener 时只接受 loopback 连接。 +listener 1883 +allow_anonymous true + +# ---- 数据与日志 ------------------------------------------------------ +# +# 本模块实际用的是 QoS 0 + clean session（见 docs/mqtt-guide.md），持久化本身 +# 帮不上什么忙；留着它只是为了以后调 QoS、用 retained 消息时不用回头改配置。 +persistence true +persistence_location /mosquitto/data/ + +# 同时往 stdout 和文件写：compose logs 看实时，./log 留档。 +log_dest stdout +log_dest file /mosquitto/log/mosquitto.log +log_type all + +# ---- 认证 + ACL（可选，生产该开） ------------------------------------ +# +# 要开认证，按顺序做三件事： +# 1) 造密码文件（用户名就用 svfs 的登录名，客户端默认拿它当 broker 用户名）： +# docker compose exec mqtt mosquitto_passwd -c -b /mosquitto/config/passwd alice 密码 +# docker compose exec mqtt mosquitto_passwd -b /mosquitto/config/passwd svfs-server 服务端密码 +# （-c 会覆盖已有文件，之后加用户不要再带 -c；Linux 上宿主的 config/ 目录 +# 属主不是 uid 1883 时会写不进去，先 sudo chown -R 1883:1883 config） +# 2) 客户端侧准备 broker 密码：它的默认值是**登录 token**，和这里的静态密码 +# 对不上，连接会被拒（CONNACK code=4/5）。两条路： +# * 在客户端 settings.json 里写 mqtt_password（见 docs/mqtt-guide.md 三）； +# * 或由服务端在登录响应里下发 mqtt.password（推荐，用户端零配置）。 +# 3) 解开下面三行，重启容器。 +# +# allow_anonymous false +# password_file /mosquitto/config/passwd +# acl_file /mosquitto/config/acl +# +# 注意：开了认证之后示例服务端（examples/plugin_server.py）也会连不上—— +# 它的 _mqtt_publish 不带 broker 账号，得给它加 SVFS_MQTT_USER/SVFS_MQTT_PASS +# 的几行补丁，或让它用独立的发布账号（docs/mqtt-guide.md 有现成代码）。 + +# ---- TLS（可选） ----------------------------------------------------- +# +# 解开这段起 8883，同时把 compose 里的 8883 端口映射打开。 +# 证书放进 config/certs/（.gitignore 已挡住），容器内路径就是 /mosquitto/config/certs/…。 +# +# 重要：客户端只做「系统信任库 + CERT_REQUIRED」校验，没有单独的 CA 配置项 +# （设置里的 ca_bundle 只作用于 HTTP）。所以自签证书必须把 CA 装进**每台 +# 客户端的系统信任库**，否则握手一定失败——内网先用明文 1883 往往更省事。 +# +# listener 8883 +# allow_anonymous false +# cafile /mosquitto/config/certs/ca.crt +# certfile /mosquitto/config/certs/server.crt +# keyfile /mosquitto/config/certs/server.key diff --git a/deploy/mqtt/docker-compose.yml b/deploy/mqtt/docker-compose.yml new file mode 100644 index 0000000..e30bab8 --- /dev/null +++ b/deploy/mqtt/docker-compose.yml @@ -0,0 +1,31 @@ +# MQTT broker（mosquitto 2.x）：svfs 消息模块的实时推送通道。 +# +# docker compose up -d # 起 +# docker compose logs -f mqtt # 看日志（配置里 log_dest stdout） +# docker compose down # 停 +# +# 默认配置是「内网匿名」——能跑起来，但任何能连到 1883 的人都收发得到消息， +# 只适合内网/开发。上认证、ACL、TLS 见 config/mosquitto.conf 里的三个注释段， +# 完整说明见 docs/mqtt-guide.md 的「二、Docker 部署 broker」。 + +services: + mqtt: + image: eclipse-mosquitto:2 + container_name: svfs-mqtt + restart: unless-stopped + ports: + - "1883:1883" # 明文：内网/开发（默认只开这个） + # - "8883:8883" # TLS：解 config/mosquitto.conf 的 TLS 段、放好证书再开这个 + volumes: + # 配置目录整份挂进去（conf / acl / passwd / certs 都在这里）。 + # 挂目录而不是单个文件：mosquitto_passwd 要往这里写密码文件， + # 只挂文件的话写进去的东西会落在容器层，重启就没了。 + - ./config:/mosquitto/config + # 数据与日志用具名卷：容器里的 mosquitto 用户是 uid 1883，用宿主目录 + # 在 Linux 上常撞权限错误；具名卷由 Docker 按镜像里的属主创建，开箱即用。 + - mqtt-data:/mosquitto/data + - mqtt-log:/mosquitto/log + +volumes: + mqtt-data: + mqtt-log: diff --git a/docs/mqtt-guide.md b/docs/mqtt-guide.md new file mode 100644 index 0000000..13d8589 --- /dev/null +++ b/docs/mqtt-guide.md @@ -0,0 +1,703 @@ +# 消息模块说明：broker 部署 / 客户端配置 / 服务端开发 + +svfs 的消息模块让**服务端能主动把消息推给某个用户或某个插件**：给用户看的进客户端 +「消息中心」，给插件的（仪表盘数据、行情、位号刷新这类业务载荷）只投给那个插件。 + +整套东西只需要三样：一个 MQTT broker、客户端里配好连接、服务端把消息发出去。 +本文按这三块展开—— + +| 你要做的事 | 看哪一节 | +|---|---| +| 把 broker 跑起来（Docker） | 二、Docker 部署 broker | +| 让客户端连上、看到消息 | 三、前端配置（客户端） | +| 在自己的服务端里发消息 | 四、服务端开发 | + +可直接运行的部署资产在 `deploy/mqtt/`：`docker compose up -d` 起一个 mosquitto， +配置里预留了认证、ACL、TLS 三段（默认全关，注释形式）。 + +> 关于本文的验证程度：主题方案、封套解析、去重、配置优先级这些都对着源码逐条核过 +> （各节标注了出处），仓库里的测试也覆盖了这些路径；但 `deploy/mqtt/` 的 +> compose / mosquitto 配置是按 mosquitto 2.x 的官方语义写的，**开发机上没有 Docker， +> 没有实际起过容器**。第一次落地前先在内网按第二节跑一遍，有问题再回来对。 + +--- + +## 一、模块速览 + +### 两条通道 + +| 通道 | 时机 | 说明 | +|---|---|---| +| MQTT 推送 | 实时 | 服务端主动推；broker 地址由登录响应下发（也可在设置里手配） | +| HTTP 补齐 | **登录成功后先做一次**，也可点「立即同步」 | `GET /api/messages?since=…` 把离线期间攒下的消息补回来 | + +登录后的顺序是**先补齐、再交给 MQTT**（`shell.py` 的 `_start_messaging`）：MQTT 只能保证 +"从现在起"的消息，离线那批只能靠补齐拿回来。两条路的消息都喂进同一个总线、按 id 去重， +所以推送与补齐重叠不会重复显示。 + +### 消息怎么走 + +```text +服务端 ──publish──> MQTT broker ──> MqttTransport（paho 网络线程） + │ on_message（回调还在网络线程） + ▼ + MqttService（Qt 排队信号 → 主线程） + │ + ┌──────────────────┘ + ▼ + MessageBus：按主题分类 → 归属校验 → 去重 → 分发 + │ + ┌───────────┴────────────┐ + ▼ ▼ + 消息中心（✉ 未读徽标） 插件队列（页面打开后投递） +``` + +### 实现分层 + +| 模块 | 层 | 职责 | +|---|---|---| +| `topics.py` | 纯标准库 | 主题拼装 / 解析 / 层级转义 | +| `messages.py` | 纯标准库 | 消息封套解析 + 消息日志（环形缓冲、未读、去重） | +| `bus.py` | 纯标准库 | 总线：分类、归属校验、去重、分发 | +| `mqtt.py` | 纯标准库（paho 可选） | 连接、重连、订阅、收包 | +| `mqtt_service.py` | Qt | 把网络线程回调转成主线程信号 | +| `messages_page.py` | Qt | 消息中心界面 | +| `shell.py` | Qt | 生命周期编排（登录后先补齐再连） | + +纯层不 import Qt，可以在没有 PySide6 的环境里单独测；`mqtt.py` 里的 paho 是可选依赖， +缺了只报 `missing` 状态，不影响其它功能。 + +### 四条语义（先记住，后面都用得上） + +* **只有登录后才有消息**：没有身份时总线一律不收（`bus._allowed`），退出登录即断开并清空； +* **分类只看主题，不看载荷内容**：所以服务端不会"一不小心把仪表盘数据推进用户的消息列表"； +* **主机消息有封套、进消息中心；插件消息原样投给插件**，两类泾渭分明； +* **实时通道是 QoS 0**（丢了不补，靠登录后补齐兜底），去重按消息 id。 + +客户端侧依赖： + +```bash +pip install "sqlite-vfs[mqtt]" # paho-mqtt，实时推送通道需要 +``` + +没装不会坏事：状态显示「未安装 paho-mqtt，实时推送不可用」，补齐照常把消息带回来。 +开发环境 `uv sync` 已经带了 paho（dev 依赖里有）；打包客户端时，构建环境里装了它就会一并打进去。 + +--- + +## 二、Docker 部署 broker（mosquitto） + +### 五分钟起步 + +```bash +cd deploy/mqtt +docker compose up -d +docker compose logs -f mqtt # 看到 "Opening ipv4 listen socket on port 1883" 就是好了 +``` + +目录长这样： + +```text +deploy/mqtt/ +├── docker-compose.yml # eclipse-mosquitto:2，映射 1883（8883 注释着） +├── .gitignore # 密码文件与证书不进仓库 +└── config/ # 整份挂到容器的 /mosquitto/config + ├── mosquitto.conf # 默认匿名；认证/ACL/TLS 三段注释着 + └── acl # 最小权限 ACL 示例（要用才在 conf 里启用） +``` + +`config/` 是整目录挂载（不是单文件）：`mosquitto_passwd` 要往这里写密码文件， +只挂单个文件的话写进去的东西落在容器层，重启就没了。数据与日志用具名卷 +（`mqtt-data` / `mqtt-log`），免得撞上 Linux 上的目录属主问题（容器里的 +mosquitto 是 uid 1883）。 + +### 先确认 broker 真的在收发 + +镜像里自带 `mosquitto_sub` / `mosquitto_pub`，不用另装客户端： + +```bash +# 终端 1：订阅所有 svfs 主题（-v 顺带打出主题） +docker compose exec mqtt mosquitto_sub -h localhost -t 'svfs/#' -v + +# 终端 2：模拟服务端推给 alice 的一条主机消息 +docker compose exec mqtt mosquitto_pub -h localhost \ + -t svfs/default/user/alice/message \ + -m '{"id":"t1","title":"你好","body":"来自 mosquitto_pub","level":"info"}' +``` + +终端 1 应当原样打印出主题与载荷。这一对通了，说明 broker 与主题方案没问题， +剩下的就是服务端和客户端的事。 + +### 配置逐行看 + +| 配置 | 作用 | 注意 | +|---|---|---| +| `listener 1883` | 开监听口 | mosquitto 2.x 不写 listener 时只接受 loopback 连接（容器里等于没人连得上） | +| `allow_anonymous true` | 放行无凭据连接 | 定义了 listener 之后它**默认是 false**，必须显式写——"端口通、连不上"多半是这里 | +| `persistence true` | 落盘会话/保留消息 | 本模块 QoS 0 + clean session，其实用不上；留着是免得以后调 QoS 时回头改 | +| `log_dest stdout` | 日志进 `docker compose logs` | 另一条 `log_dest file` 是留档，用 `docker compose exec mqtt cat /mosquitto/log/mosquitto.log` 看 | + +**客户端带用户名密码连匿名 broker 会被拒吗？** 不会。客户端总是带上用户名 +（默认是登录名）与密码（默认是登录 token），而 `allow_anonymous true` 且没有 +`password_file` 时，broker 没有可校验的东西，直接放行。所以内网匿名是最省事的组合： +客户端零配置，broker 地址由登录响应下发。 + +### 档 A：内网匿名（默认，先用它跑通） + +适合内网、开发、演示。风险是**任何能连到 1883 的人都能订阅所有主题**，包括别人的消息。 +真要收紧就把它绑到内网地址、只对服务端所在网段放行，或者直接进档 B。 + +### 档 B：认证 + 最小权限 ACL（生产该用） + +`config/mosquitto.conf` 的「认证 + ACL」段里写了三步（造密码文件、准备客户端密码、 +解开三行重启）。这里补三处最容易踩的： + +**1）客户端的密码默认是登录 token，不是你在 passwd 里设的那个密码。** + +取值顺序是 `mqtt_password` 设置 → 服务端下发的 `mqtt.password` → **登录 token** +（`settings.py` 的 `mqtt_config`）。所以开了 `password_file` 之后二选一： + +* 服务端在登录响应的 `mqtt` 段里多下发一个 `password`（客户端零配置，推荐； + 示例服务端只下发 host/port/tls/prefix/org，需要就自己加上这两个键）； +* 或在客户端 `settings.json` 里写 `mqtt_password`（明文落盘，见第三节的安全提示）。 + +**2）服务端自己也连不上了。** + +示例服务端的 `_mqtt_publish` 不带 broker 账号，`allow_anonymous false` 之后每条推送 +都会失败（`/api/push` 会如实返回 `error: …`，不会假装成功）。给它加账号： + +```python +# examples/plugin_server.py 的 _mqtt_publish 里，mqtt.Client(...) 之后、connect 之前 +user = os.environ.get("SVFS_MQTT_USER", "").strip() +if user: + client.username_pw_set(user, os.environ.get("SVFS_MQTT_PASS") or None) +``` + +然后带 `SVFS_MQTT_USER=svfs-server SVFS_MQTT_PASS=…` 启动，并在 ACL 里给它写权限 +（`config/acl` 的 `svfs-server` 段就是现成的）。更正规的做法是服务端用一条常驻连接 +发布，见第四节的发布代码。 + +**3）ACL 里的用户名是 broker 用户名，组织段要写死。** + +ACL 的 `<用户名>` 默认就等于 svfs 登录名（`mqtt_username` 留空时）；组织段（`acl` 里的 +`default`）要换成你的 `SVFS_ORG` 并保持固定——多租户隔离靠的就是这一段，写成 `+` +等于没隔离。用 `pattern` 时 `%u` 必须独占一个主题层级（`svfs/x/user/%u/#` 可以， +`svfs/x/user/%u-2/#` 不行）。 + +### 档 C：TLS（8883） + +解开 `config/mosquitto.conf` 的 TLS 段、把证书放进 `config/certs/`、打开 compose 里的 +8883 端口映射。证书自签即可： + +```bash +cd deploy/mqtt/config && mkdir -p certs && cd certs +openssl req -x509 -newkey rsa:2048 -nodes -keyout ca.key -out ca.crt -days 3650 \ + -subj "/CN=svfs-dev-ca" +openssl req -newkey rsa:2048 -nodes -keyout server.key -out server.csr \ + -subj "/CN=mqtt.example" +openssl x509 -req -in server.csr -CA ca.crt -CAkey ca.key -CAcreateserial \ + -out server.crt -days 3650 -extfile <(printf "subjectAltName=DNS:mqtt.example,IP:10.0.0.5") +``` + +**关键约束：客户端没有 CA 配置项。** 它只做「系统信任库 + `CERT_REQUIRED`」校验 +（`mqtt.py` 的 `_configure`：`tls_set(cert_reqs=ssl.CERT_REQUIRED)`）， +设置里的 `ca_bundle` 只管 HTTP（`netutil`），没有接到 MQTT 上。所以自签证书要么把 +CA 装进**每台客户端**的系统信任库（Windows：导入「受信任的根证书颁发机构」； +macOS：钥匙串；Linux：`/usr/local/share/ca-certificates` + `update-ca-certificates`）， +要么内网先用明文 1883 + 网络层隔离。**证书里的 CN/SAN 必须和客户端填的 host 一致**， +用 IP 连就写 IP。 + +自测（在客户端机器上跑，验证信任链）： + +```bash +openssl s_client -connect 10.0.0.5:8883 -CAfile ca.crt { + // message.text 是原文；能解析成对象时另有 message.json + render(message.json ?? message.text); + }); +} +``` + +消息对象是 `{text: string, json?: object, time: string}`——`time` 目前恒为空串 +（投递路径里没带时间），要时间戳就自己把它放在载荷里。 + +**不用怕"消息比页面先到"**，这条链路上有三层缓冲： + +1. 插件页面是懒加载的，可能还没打开； +2. 总线按插件缓存最近 50 条（超了丢最旧的并计 `dropped`），去重记忆 200 条； +3. 页面打开时宿主把攒下的投进去；页面里的 `window.__svfsPush` 在 DocumentCreation 阶段 + 就立好了，订阅之前到的先排队、`subscribe` 时补发。 + +所以 `subscribe` 只写"收到消息怎么办"就够了，不需要自己去拉历史。调试时可以用客户端 +「设置 → 桥调试台 → 安装调试插件」里的 messages 面板对照（`sample.py` 里的实现）。 + +--- + +## 四、服务端开发 + +### 要做的两件事 + +1. **发布**：把消息 publish 到 broker 的对应主题； +2. **下发**：在登录响应里带上 broker 信息，让客户端自动连上（否则用户得自己填地址）。 + +### 4.1 拼主题别手写 + +```python +from svfs_client.topics import ( + user_message_topic, user_plugin_topic, + broadcast_message_topic, broadcast_plugin_topic, +) +``` + +装 `pip install sqlite-vfs` 就能复用（纯标准库，不需要 PySide6）。**为什么必须复用**： +用户名与插件 id 要占一个主题层级，得先过一层可逆的百分号编码（安全字符 +`[A-Za-z0-9._@-]` 原样保留，其余编码），两端规则不一致就会"发得出去、收不到"， +而且 `a b` 与 `a_b` 会撞成同一个主题、串了租户。示例服务端就是直接 import 这个模块的。 + +### 4.2 主题与通道 + +| 主题 | 谁收 | 进消息中心 | +|---|---|---| +| `//user//message` | 该用户 | 是 | +| `//user//plugin/` | 该用户的该插件 | 否 | +| `//broadcast/message` | 整个组织 | 是（带广播标记） | +| `//broadcast/plugin/` | 整个组织的该插件 | 否 | + +客户端登录后只订阅两条（最小权限，不是整个组织的 `#`）： + +```text +//user//# +//broadcast/# +``` + +即使 broker 的 ACL 配错、把别人的消息推了过来，总线还会按当前组织/用户再挡一道 +（记 `forbidden` 计数 + 诊断日志）。 + +### 4.3 发布代码 + +示例服务端（`examples/plugin_server.py` 的 `_mqtt_publish`）为了演示简单，是 +"连上 → 发一条 → 断开"，每条消息一次握手；生产上换成常驻连接更合适： + +```python +import json +import os +import threading +from datetime import datetime + +import paho.mqtt.client as mqtt +from svfs_client.topics import user_message_topic, user_plugin_topic + +MQTT_HOST = os.environ.get("SVFS_MQTT_HOST", "") +MQTT_PORT = int(os.environ.get("SVFS_MQTT_PORT", "") or 1883) +MQTT_PREFIX = os.environ.get("SVFS_MQTT_PREFIX", "svfs") +ORG = os.environ.get("SVFS_ORG", "").strip() or "default" + +_lock = threading.Lock() +_client = None + + +def _connection(): + """进程内共用一条连接（paho 自己管心跳与重连）。""" + global _client + with _lock: + if _client is None: + client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, + client_id="svfs-server") + user = os.environ.get("SVFS_MQTT_USER", "").strip() + if user: # broker 开了认证就必须给账号 + client.username_pw_set(user, os.environ.get("SVFS_MQTT_PASS") or None) + if os.environ.get("SVFS_MQTT_TLS", "").lower() in ("1", "true", "yes"): + client.tls_set() # 用系统信任库；自签证书见第二节「档 C：TLS」 + client.connect_async(MQTT_HOST, MQTT_PORT, 30) + client.loop_start() + _client = client + return _client + + +def publish(topic, payload, qos=0): + """发一条；没配 broker / 没连上都返回说明字符串（不抛，消息通道不能拖垮业务）。""" + if not MQTT_HOST: + return "skipped: 未配置 SVFS_MQTT_HOST" + client = _connection() + if not client.is_connected(): + # connect_async 不阻塞：进程刚起或断线重连中就是 False。如实报这一条失败 + # （paho 会自己连回来），别把业务请求堵在握手上 + return "error: 还没连上 broker（重连中）" + try: + info = client.publish(topic, payload, qos=qos) + if info.rc != mqtt.MQTT_ERR_SUCCESS: + return f"error: {mqtt.error_string(info.rc)}" + info.wait_for_publish(timeout=5) # QoS 0 一般立刻返回 + except Exception as exc: + return f"error: {exc}" + return "published" + + +def push_message(user, title, body, level="info", msg_id=""): + message = { + "id": msg_id, # 建议给：客户端按它去重 + "title": title, + "body": body, + "level": level, # info | warn | error + "time": datetime.now().isoformat(timespec="seconds"), + "source": "my-service", + } + topic = user_message_topic(MQTT_PREFIX, ORG, user) + return topic, publish(topic, json.dumps(message, ensure_ascii=False)) +``` + +上面这段有两个刻意的选择，都是踩过才知道的： + +* `connect_async` 不阻塞，所以进程刚起的那一小会儿 `is_connected()` 还是 False—— + 这期间的推送会返回 `error: 还没连上 broker（重连中）`。**宁可如实报错误，也不要让业务 + 请求卡在 TCP 握手上**；要"启动即确认连上"，就把 `connect_async` 换成同步的 `connect` + （连不上会抛，自己接住），或者在连上之前先不对外提供服务； +* 返回字符串而不是抛异常：`/api/push` 这类接口即使发不出去也要能正常返回——消息本来 + 就会进待收列表，客户端登录后还能补齐。 + +给插件发数据就是换主题、载荷随便（原样投递）： + +```python +topic = user_plugin_topic(MQTT_PREFIX, ORG, user, "com.example.dash") +publish(topic, json.dumps({"cpu": 0.42})) +``` + +### 4.4 登录响应里下发 broker + +客户端会读登录响应里的 `mqtt` 段，据此自动连接： + +```json +{ + "ok": true, + "token": "…", + "username": "alice", + "org": "acme", + "mqtt": { + "host": "10.0.0.5", + "port": 1883, + "tls": false, + "prefix": "svfs", + "org": "acme" + } +} +``` + +| 键 | 必填 | 说明 | +|---|---|---| +| `host` | 是 | 客户端能访问到的 broker 地址（不是 broker 眼里的自己） | +| `port` | 否 | 不给就按 `tls` 推 1883 / 8883 | +| `tls` | 否 | true 走 mqtts | +| `prefix` | 否 | 缺省 `svfs`，两端必须一致 | +| `org` | 否 | 缺省 `default`；也是客户端订阅时的租户段 | +| `username` / `password` | 否 | 给客户端一套独立的 broker 账号时用（客户端优先用它，`settings.py` 的 `mqtt_config`） | + +`GET /api/me`（启动时静默校验 token）也会带上同样的段，所以客户端重启后不用重新登录 +也能拿到 broker 信息。**没配 broker 就别带 `mqtt` 键**（示例服务端的 +`_mqtt_announcement` 就返回 None）——客户端会走"只补齐、不实时"的降级，这是正常状态。 + +### 4.5 主机消息的封套 + +```json +{ + "id": "b7c1…", + "title": "备份完成", + "body": "3 个包已同步", + "level": "info", + "time": "2026-09-16T10:00:00", + "source": "backup" +} +``` + +| 字段 | 类型 | 上限 | 缺省 | 说明 | +|---|---|---|---|---| +| `id` | 字符串 | 120 | 空 | 去重键，**强烈建议给**（给了就不会重复显示） | +| `title` | 字符串 | 200 | 空 | 标题 | +| `body` | 字符串 | 4000 | 空 | 正文 | +| `level` | 字符串 | — | `info` | `info` / `warn` / `error`，认不出来的当 `info` | +| `time` | 字符串 | 40 | 收到的时间 | ISO8601；要能被**字符串比较**（增量补齐靠它） | +| `source` | 字符串 | 120 | 空 | 来源标记，界面上显示 | + +解析很宽松，下面这些也能用： + +* 载荷是**纯字符串**或 JSON 标量（数字、带引号的串）→ 整段当正文； +* 不是合法 JSON 的字符串 → 整段当正文； +* `title` 与 `body` **都空**的（空串、`{}`）→ 整条丢弃，不往消息中心塞空气。 + +**去重规则**：有 `id` 就按 `id`；没有就按 `时间|标题|正文`。所以同一秒里发两条内容 +完全相同、又都没带 `id` 的消息，第二条会被当成重投丢掉——给 `id` 最省事。 + +广播的封套一样，只是走 `broadcast` 主题；建议把 `source` 写成能看出是谁发的 +（示例是 `broadcast@<用户名>`）。 + +### 4.6 插件消息 + +* 载荷**原样投递**，宿主不解析、不加封套（插件自己的协议自己定）； +* 不进消息中心、不进待收列表——它是实时业务载荷，补历史没有意义； +* 去重是"内容 + 秒"级别（同一插件、同一正文、同一秒视为重复），这是为了挡 QoS 1 重投 + 与推送/补齐的重叠。**所以：要分帧或累积的数据请自己带序号**，否则同一秒里两帧一模一样的 + 数据会被当成重投丢掉。 + +### 4.7 HTTP 补齐接口（离线消息） + +客户端登录后（以及点「立即同步」时）会拉一次： + +| 项 | 约定 | +|---|---| +| 方法与路径 | `GET /api/messages` | +| 请求头 | `Accept: application/json`、`Authorization: Bearer ` | +| 参数 | `since`（可选）：本地最新一条消息的 `time`，只要这之后的消息 | +| 响应 | `{"ok": true, "messages": [封套, …]}`；没有新消息就是空数组 | +| 没这个接口 | 返回 404，客户端当作"服务端不支持"，静默降级（不报错、不刷状态） | +| 客户端限制 | 15 秒超时、失败重试 1 次、最多读 2MB、最多取 500 条 | + +服务端侧要注意的： + +* **只返回该 token 对应用户的消息**（示例服务端按 username 分桶存）； +* `since` 用字符串比较 `>` 就够了（ISO8601 定长、字典序即时间序）； +* 返回的封套与 MQTT 推送用的是同一套，客户端按 id 去重——所以你**可以放心把发过的消息 + 也留在待收列表里**，推送与补齐重叠不会显示两条； +* 示例的待收列表是内存里每人 100 条、重启即清（`MESSAGE_BACKLOG`），生产该入库； +* 广播不进待收列表（示例如此）：**组织广播只有在线的人看得到**，要保证人人可见就得自己 + 把广播也存下来并在补齐接口里带上。 + +### 4.8 认证怎么落地 + +| 方案 | 客户端怎么连 | broker 侧 | 适合 | +|---|---|---|---| +| 内网匿名 | 用户名=登录名、密码=token，broker 不校验 | `allow_anonymous true` | 内网/开发/演示 | +| 静态账号 + 独立密码 | 服务端下发 `mqtt.password`，或客户端写 `mqtt_password` | `password_file` + `acl_file` | 多数生产场景 | +| broker 校验登录 token | **默认行为**（密码就是 token） | broker 的认证回调打到你的服务端：EMQX 5.x 的 HTTP 认证、mosquitto 的动态安全插件或 mosquitto-go-auth 这类外部插件 | 想"零配置 + 和登录体系联动" | + +第三种最贴合本模块的现成设计（客户端本来就把登录 token 当 broker 密码发），代价是 +broker 侧要接一个认证回调：收到 `(username, password)` 就调你的 `/api/verify` 校验这个 +token 是否有效、属于谁，顺便还能把 ACL 一起下发（"你只能读写 `svfs//user/<你>/#`"）。 + +### 4.9 端到端跑一遍 + +```bash +# 0) broker +cd deploy/mqtt && docker compose up -d + +# 1) 服务端：下发 broker 地址（客户端登录后自动连上） +SVFS_MQTT_HOST=127.0.0.1 SVFS_MQTT_PORT=1883 SVFS_ORG=acme \ + uv run python examples/plugin_server.py + +# 2) 登录取 token +token=$(curl -s -X POST http://127.0.0.1:8000/api/login \ + -H 'Content-Type: application/json' \ + -d '{"username":"admin","password":"admin123"}' \ + | python -c "import json,sys; print(json.load(sys.stdin)['token'])") + +# 3) 主机消息 → 消息中心 +curl -s -X POST http://127.0.0.1:8000/api/push -H "Authorization: Bearer $token" \ + -H 'Content-Type: application/json' \ + -d '{"kind":"message","title":"备份完成","body":"3 个包已同步","level":"info"}' + +# 4) 插件数据 → 只给 com.svfs.demo 这个插件 +curl -s -X POST http://127.0.0.1:8000/api/push -H "Authorization: Bearer $token" \ + -H 'Content-Type: application/json' \ + -d '{"kind":"plugin","plugin":"com.svfs.demo","data":{"cpu":0.42}}' + +# 5) 组织广播 → 组织内所有在线客户端 +curl -s -X POST http://127.0.0.1:8000/api/broadcast -H "Authorization: Bearer $token" \ + -H 'Content-Type: application/json' \ + -d '{"title":"停机维护","body":"今晚 22:00 开始","level":"warn"}' +``` + +（curl 的中文参数在 Git Bash 下没问题；PowerShell 里更稳的做法是把 JSON 写进文件再用 `-d @body.json`。） + +注意示例服务端把 uvicorn 绑在 `127.0.0.1`（源码里写死的），所以这套跑法只适用于 +"客户端和服务端在同一台机器"；要让别的机器连，改 `examples/plugin_server.py` 结尾那句 +`uvicorn.run(...)` 的 host（同时记得 `SVFS_MQTT_HOST` 要下发**客户端能访问到的**地址， +不是服务端眼里的 `127.0.0.1`）。 + +响应里看 `"mqtt"` 字段就知道有没有真的发出去：`"published"` 是发出去了， +`"skipped: …"` 是没配 broker 或服务端没装 paho，`"error: …"` 是连不上/被拒。 +即使 `skipped`，消息也已经进了待收列表，客户端登录后能补齐看到。 + +### 4.10 边界与降级 + +* 没配 `SVFS_MQTT_HOST`：`/api/push` 只写待收列表，响应 `mqtt=skipped`，客户端靠补齐看到； +* 服务端没装 paho：同上（`skipped: 服务端没装 paho-mqtt`），不影响接口本身； +* 推送只对**已登录**的客户端有效；退出登录即断开（客户端侧行为，见第三节）； +* QoS 0 不保证送达：要"必达"就靠待收列表 + 补齐 + `id` 去重，别指望 broker 重投； +* 广播不落库（示例如此）：离线的人看不到组织广播，除非你自己也存一份。 + +--- + +## 五、速查 + +### 5.1 示例服务端的环境变量 + +| 变量 | 默认 | 作用 | +|---|---|---| +| `SVFS_MQTT_HOST` | 空 | broker 地址；**空 = 不启用 MQTT**（推送只进待收列表） | +| `SVFS_MQTT_PORT` | 空 | broker 端口；空则按 tls 推 1883 / 8883 | +| `SVFS_MQTT_TLS` | 空 | `1` / `true` / `yes` 走 TLS | +| `SVFS_MQTT_PREFIX` | `svfs` | 主题前缀 | +| `SVFS_ORG` | `default` | 组织/租户标识 | +| `SVFS_USERS` | `admin:admin123` | 登录账号表，`user:pass,user2:pass2` | +| `SVFS_PORT` | 8000 | HTTP 端口 | +| `SVFS_PLUGINS_DIR` | `./svfs-plugins-root` | 插件包根目录 | +| `SVFS_UPLOAD_TOKEN` | 空 | 上传插件时的 `X-SVFS-Token` | +| `SVFS_CLIENT_VERSION` / `_URL` / `_NOTES` | 空 | 客户端自更新清单（不设则 404） | + +### 5.2 客户端 settings.json 的 MQTT 相关键 + +| 键 | 默认 | 说明 | +|---|---|---| +| `mqtt_host` | `""` | 空 = 不启用推送（但服务端下发的仍会用） | +| `mqtt_port` | `1883` | | +| `mqtt_tls` | `false` | | +| `mqtt_prefix` | `"svfs"` | | +| `mqtt_username` | `""` | 空 = 登录用户名 | +| `mqtt_password` | `""` | 空 = 登录 token（加密存储的那份） | +| `org` | `""` | 空 = 服务端下发 → `default` | + +### 5.3 总线处理结果与计数 + +`deliver()` 返回的 `action`：`host`（进消息中心）/ `plugin`（投给插件）/ `duplicate` +（重复，丢弃）/ `unknown`（主题不认识）/ `forbidden`（不是发给当前身份的）/ `empty` +（解析不出正文）。这些动作各自都有计数，`MessageBus.stats()` 能拿到快照（`received`、 +`host`、`plugin`、`duplicate`、`unknown`、`forbidden`、`empty`、`dropped`）——所以 +`forbidden` / `dropped` 有两条线索可查：诊断日志里的那两行文字，以及这个计数快照。 + +### 5.4 客户端连接参数（写死在代码里，没有配置项） + +| 参数 | 值 | 出处 | +|---|---|---| +| 客户端 id | `svfs---<6 位随机>` | 随机后缀是为了"同一人开两个客户端不互相踢下线" | +| clean session | 是 | 不指望 broker 记会话 | +| keepalive | 30 秒 | | +| 重连退避 | 1 → 30 秒（指数） | 交给 paho，重连成功后自动重新订阅 | +| 发布 / 订阅 QoS | 0 | 服务端示例也是 0 | + +--- + +## 六、测试 + +```bash +uv run python tests/run_tests.py -p "test_client_messages.py" # 主题/封套/总线（纯层） +uv run python tests/run_tests.py -p "test_client_mqtt.py" # MQTT 通道（含自建 broker 端到端） +uv run python tests/run_tests.py -p "test_plugin_server.py" # 服务端 /api/push → broker → 客户端 +uv run python tests/run_tests.py -p "test_client_ui.py" # 设置界面与消息中心 +``` + +`test_client_mqtt.py` 里的 `_MiniBroker` 是测试里手写的 MQTT 3.1.1 broker（QoS 0， +认 CONNECT / SUBSCRIBE / PUBLISH / PINGREQ / DISCONNECT 这几种报文）， +所以**这些测试不需要 Docker、也不需要真 broker**，CI 上直接跑。覆盖到的点： +主题拼装/解析与转义、封套解析与长度上限、 +去重（id 与"内容+秒"两种）、越权消息丢弃、订阅两条主题、重连后重新订阅、拒连报错、 +TLS 开关、以及"服务端 push → broker → 客户端总线"的全链路。