modify:网关配置控制amqtt broker
This commit is contained in:
@@ -12,7 +12,7 @@
|
|||||||
| --------------------- | -------------------------------------------------------------------- |
|
| --------------------- | -------------------------------------------------------------------- |
|
||||||
| **FastAPI + Uvicorn** | 高性能异步 Web 框架,自动生成 OpenAPI 文档 |
|
| **FastAPI + Uvicorn** | 高性能异步 Web 框架,自动生成 OpenAPI 文档 |
|
||||||
| **Zenoh 服务发现** | 基于 liveliness 的 Worker 注册与 RPC 调用,负载均衡 + 超时重试 |
|
| **Zenoh 服务发现** | 基于 liveliness 的 Worker 注册与 RPC 调用,负载均衡 + 超时重试 |
|
||||||
| **MQTT 集成** | 内嵌 amqtt Broker,MQTT ↔ Worker 双向桥接 |
|
| **MQTT 集成** | amqtt Broker(内嵌或外部),MQTT ↔ Worker 双向桥接 |
|
||||||
| **动态路由生成** | 扫描 `config/workers/*/*.toml` 自动注册 HTTP 路由 |
|
| **动态路由生成** | 扫描 `config/workers/*/*.toml` 自动注册 HTTP 路由 |
|
||||||
| **Worker 框架** | `@rpc_method` 装饰器 + `Plugin` 基类统一生命周期 + Pydantic 模型绑定 |
|
| **Worker 框架** | `@rpc_method` 装饰器 + `Plugin` 基类统一生命周期 + Pydantic 模型绑定 |
|
||||||
| **启动契约校验** | TOML 路由 ↔ `@rpc_method` ↔ 生成模型三方一致性校验,fail fast |
|
| **启动契约校验** | TOML 路由 ↔ `@rpc_method` ↔ 生成模型三方一致性校验,fail fast |
|
||||||
@@ -205,7 +205,8 @@ graph TB
|
|||||||
uv run serve → FastAPI lifespan:
|
uv run serve → FastAPI lifespan:
|
||||||
1. setup_storage() — 本地文件系统存储
|
1. setup_storage() — 本地文件系统存储
|
||||||
2. setup_zenoh() — Zenoh peer 会话
|
2. setup_zenoh() — Zenoh peer 会话
|
||||||
3. setup_mqtt_broker() — 内嵌 Broker(子进程 + 文件锁单例)
|
3. setup_mqtt_broker() — 内嵌 amqtt Broker(子进程 + 文件锁单例;
|
||||||
|
`use_embedded_broker=false` 时跳过,改用外部 Broker)
|
||||||
4. setup_worker_manager() — WorkerManager + 定期 liveliness 刷新
|
4. setup_worker_manager() — WorkerManager + 定期 liveliness 刷新
|
||||||
5. load_workers() — 按插件 [workers].load_mode 分派:
|
5. load_workers() — 按插件 [workers].load_mode 分派:
|
||||||
inline → 进程内 import 插件 → 启动 Worker 协程
|
inline → 进程内 import 插件 → 启动 Worker 协程
|
||||||
@@ -337,6 +338,39 @@ start.sh 容器入口:uv sync + worker_serve --type <nam
|
|||||||
| 请求 | `chongming/<type>/<op>` | MQTT → Worker |
|
| 请求 | `chongming/<type>/<op>` | MQTT → Worker |
|
||||||
| 响应 | `chongming/<type>/<op>/response` | Worker → MQTT(`None` 时跳过) |
|
| 响应 | `chongming/<type>/<op>/response` | Worker → MQTT(`None` 时跳过) |
|
||||||
|
|
||||||
|
### Broker 来源(内嵌 vs 外部)
|
||||||
|
|
||||||
|
网关默认启动内嵌 amqtt Broker(子进程 + 文件锁单例),`[<env>.mqtt].use_embedded_broker` 控制 Broker 来源:
|
||||||
|
|
||||||
|
```toml
|
||||||
|
[development.mqtt]
|
||||||
|
enabled = true
|
||||||
|
host = "0.0.0.0" # true 时为内嵌 Broker 监听地址;false 时为外部 Broker 地址
|
||||||
|
port = 1883
|
||||||
|
use_embedded_broker = true # true(默认)网关启动 amqtt Broker;false 连接外部 Broker
|
||||||
|
```
|
||||||
|
|
||||||
|
- `use_embedded_broker = true`(默认):网关启动内嵌 amqtt Broker 子进程,MQTT Bridge 作为客户端连回自身。
|
||||||
|
- `use_embedded_broker = false`:网关**不启动** Broker,MQTT Bridge 直接连接 `host:port` 指向的外部 MQTT Broker(如独立部署的 EMQX/Mosquitto)。`host` 需改为外部 Broker 地址,`enabled = false` 时整个 MQTT 通道关闭。
|
||||||
|
|
||||||
|
### Broker 认证(用户名/密码)
|
||||||
|
|
||||||
|
`[<env>.mqtt].username` / `password` 配置 Broker 认证(可选,默认匿名):
|
||||||
|
|
||||||
|
```toml
|
||||||
|
[development.mqtt]
|
||||||
|
enabled = true
|
||||||
|
host = "0.0.0.0"
|
||||||
|
port = 1883
|
||||||
|
username = "chongming" # 可选;内嵌/外部 Broker 均生效
|
||||||
|
password = "change-me"
|
||||||
|
```
|
||||||
|
|
||||||
|
- **内嵌 Broker**(`use_embedded_broker=true`):配置后启用认证(amqtt FileAuthPlugin + argon2 哈希),所有客户端(含网关 Bridge)必须携带该账号密码连接;未配置则保持匿名。
|
||||||
|
- **外部 Broker**:网关 Bridge 以该账号密码连接,broker 侧需自行创建对应账号。
|
||||||
|
- 客户端连接示例:`mqtt://chongming:change-me@localhost:1883/`(密码含特殊字符时需百分号编码,如 `@` → `%40`;网关 Bridge 内部自动处理)。
|
||||||
|
- 注意:密码明文存放于配置,生产环境建议使用外部 Broker + 环境变量注入。
|
||||||
|
|
||||||
### 自定义订阅主题
|
### 自定义订阅主题
|
||||||
|
|
||||||
除内置的 `chongming/<type>/<op>` 接口外,可将外部 MQTT 主题桥接到任意 Worker。**推荐在插件自己的配置** `config/workers/<type>/config.toml` 中通过 `[[workers.mqtt_topics]]` 声明(目标 worker 即该插件本身):
|
除内置的 `chongming/<type>/<op>` 接口外,可将外部 MQTT 主题桥接到任意 Worker。**推荐在插件自己的配置** `config/workers/<type>/config.toml` 中通过 `[[workers.mqtt_topics]]` 声明(目标 worker 即该插件本身):
|
||||||
|
|||||||
+12
@@ -53,6 +53,18 @@ allow_origins = ["*"]
|
|||||||
enabled = true
|
enabled = true
|
||||||
host = "0.0.0.0"
|
host = "0.0.0.0"
|
||||||
port = 1883
|
port = 1883
|
||||||
|
# 是否由网关启动内嵌 amqtt Broker 子进程:
|
||||||
|
# true (默认)= 网关自带 amqtt Broker,host/port 为 Broker 监听地址
|
||||||
|
# false = 使用外部 MQTT Broker,网关仅启动 Bridge 客户端连接
|
||||||
|
# host/port 需指向外部 Broker 地址(如 "192.168.1.10")
|
||||||
|
use_embedded_broker = true
|
||||||
|
# MQTT Broker 认证(可选,默认匿名):
|
||||||
|
# - 内嵌 Broker(use_embedded_broker=true):配置后要求所有客户端(含网关 Bridge)
|
||||||
|
# 用该账号密码连接,未配置则保持匿名
|
||||||
|
# - 外部 Broker:网关 Bridge 用该账号密码连接(broker 侧自行配置对应账号)
|
||||||
|
# 注意:密码为明文存储,生产环境建议使用外部 Broker + 环境变量注入
|
||||||
|
# username = "chongming"
|
||||||
|
# password = "change-me"
|
||||||
|
|
||||||
# MQTT Bridge 额外订阅主题推荐声明在各插件自己的配置里:
|
# MQTT Bridge 额外订阅主题推荐声明在各插件自己的配置里:
|
||||||
# config/workers/<type>/config.toml → [[workers.mqtt_topics]]
|
# config/workers/<type>/config.toml → [[workers.mqtt_topics]]
|
||||||
|
|||||||
@@ -52,6 +52,18 @@ allow_origins = ["*"]
|
|||||||
enabled = true
|
enabled = true
|
||||||
host = "0.0.0.0"
|
host = "0.0.0.0"
|
||||||
port = 1883
|
port = 1883
|
||||||
|
# 是否由网关启动内嵌 amqtt Broker 子进程:
|
||||||
|
# true (默认)= 网关自带 amqtt Broker,host/port 为 Broker 监听地址
|
||||||
|
# false = 使用外部 MQTT Broker,网关仅启动 Bridge 客户端连接
|
||||||
|
# host/port 需指向外部 Broker 地址(如 "192.168.1.10")
|
||||||
|
use_embedded_broker = true
|
||||||
|
# MQTT Broker 认证(可选,默认匿名):
|
||||||
|
# - 内嵌 Broker(use_embedded_broker=true):配置后要求所有客户端(含网关 Bridge)
|
||||||
|
# 用该账号密码连接,未配置则保持匿名
|
||||||
|
# - 外部 Broker:网关 Bridge 用该账号密码连接(broker 侧自行配置对应账号)
|
||||||
|
# 注意:密码为明文存储,生产环境建议使用外部 Broker + 环境变量注入
|
||||||
|
# username = "chongming"
|
||||||
|
# password = "change-me"
|
||||||
|
|
||||||
# MQTT Bridge 额外订阅主题推荐声明在各插件自己的配置里:
|
# MQTT Bridge 额外订阅主题推荐声明在各插件自己的配置里:
|
||||||
# config/workers/<type>/config.toml → [[workers.mqtt_topics]]
|
# config/workers/<type>/config.toml → [[workers.mqtt_topics]]
|
||||||
|
|||||||
@@ -77,8 +77,9 @@ def _release_singleton_lock(name: str):
|
|||||||
pass
|
pass
|
||||||
|
|
||||||
|
|
||||||
def _run_mqtt_broker(host: str, port: int):
|
def _run_mqtt_broker(host: str, port: int, username: str | None = None,
|
||||||
"""子进程入口:启动 amqtt Broker"""
|
password: str | None = None):
|
||||||
|
"""子进程入口:启动 amqtt Broker(配置了账号密码时启用认证)"""
|
||||||
import asyncio as _asyncio
|
import asyncio as _asyncio
|
||||||
from amqtt.broker import Broker
|
from amqtt.broker import Broker
|
||||||
|
|
||||||
@@ -155,6 +156,33 @@ def _run_mqtt_broker(host: str, port: int):
|
|||||||
_Broker._initialize_client_session = _patched_init_client
|
_Broker._initialize_client_session = _patched_init_client
|
||||||
|
|
||||||
async def _serve():
|
async def _serve():
|
||||||
|
# 认证插件:配置了账号密码 → FileAuthPlugin(argon2 密码文件,仅此账号可连);
|
||||||
|
# 未配置 → 匿名(默认行为)
|
||||||
|
auth_plugins = {}
|
||||||
|
passwd_file = None
|
||||||
|
if username and password:
|
||||||
|
import tempfile
|
||||||
|
from pwdlib import PasswordHash
|
||||||
|
|
||||||
|
try:
|
||||||
|
fd, passwd_file = tempfile.mkstemp(prefix="chongming_mqtt_", suffix=".passwd")
|
||||||
|
with os.fdopen(fd, "w") as f:
|
||||||
|
f.write(f"{username}:{PasswordHash.recommended().hash(password)}\n")
|
||||||
|
auth_plugins = {
|
||||||
|
"amqtt.plugins.authentication.FileAuthPlugin": {
|
||||||
|
"password_file": passwd_file,
|
||||||
|
},
|
||||||
|
}
|
||||||
|
except Exception as e:
|
||||||
|
logger.warning(f"MQTT 认证初始化失败,回退匿名模式: {e}")
|
||||||
|
if passwd_file:
|
||||||
|
os.unlink(passwd_file)
|
||||||
|
passwd_file = None
|
||||||
|
else:
|
||||||
|
auth_plugins = {
|
||||||
|
"amqtt.plugins.authentication.AnonymousAuthPlugin": {"allow_anonymous": True},
|
||||||
|
}
|
||||||
|
|
||||||
cfg = {
|
cfg = {
|
||||||
"listeners": {
|
"listeners": {
|
||||||
"default": {
|
"default": {
|
||||||
@@ -163,7 +191,7 @@ def _run_mqtt_broker(host: str, port: int):
|
|||||||
},
|
},
|
||||||
"timeout_disconnect_delay": 2,
|
"timeout_disconnect_delay": 2,
|
||||||
"plugins": {
|
"plugins": {
|
||||||
"amqtt.plugins.authentication.AnonymousAuthPlugin": {"allow_anonymous": True},
|
**auth_plugins,
|
||||||
"amqtt.plugins.sys.broker.BrokerSysPlugin": {"sys_interval": 0},
|
"amqtt.plugins.sys.broker.BrokerSysPlugin": {"sys_interval": 0},
|
||||||
},
|
},
|
||||||
}
|
}
|
||||||
@@ -173,7 +201,15 @@ def _run_mqtt_broker(host: str, port: int):
|
|||||||
except Exception:
|
except Exception:
|
||||||
# 端口已被其他 worker 的 Broker 占用,静默退出
|
# 端口已被其他 worker 的 Broker 占用,静默退出
|
||||||
return
|
return
|
||||||
print(f"[MQTT Broker] 已启动 @ {host}:{port}")
|
finally:
|
||||||
|
# 密码文件仅在启动时读取,启动完成后即可删除
|
||||||
|
if passwd_file:
|
||||||
|
try:
|
||||||
|
os.unlink(passwd_file)
|
||||||
|
except OSError:
|
||||||
|
pass
|
||||||
|
auth_note = f" (认证用户: {username})" if username else " (匿名)"
|
||||||
|
print(f"[MQTT Broker] 已启动 @ {host}:{port}{auth_note}")
|
||||||
try:
|
try:
|
||||||
while True:
|
while True:
|
||||||
await _asyncio.sleep(1)
|
await _asyncio.sleep(1)
|
||||||
@@ -465,15 +501,31 @@ def validate_contracts(app: FastAPI | None = None):
|
|||||||
_validate(inline_plugins)
|
_validate(inline_plugins)
|
||||||
|
|
||||||
def setup_mqtt_broker(app: FastAPI):
|
def setup_mqtt_broker(app: FastAPI):
|
||||||
"""启动 MQTT Broker 子进程(原子锁确保多 worker 仅启动一个)"""
|
"""
|
||||||
|
启动内嵌 amqtt Broker 子进程(原子锁确保多 worker 仅启动一个)。
|
||||||
|
|
||||||
|
由配置 [<env>.mqtt].use_embedded_broker 控制:
|
||||||
|
true (默认)→ 网关启动 amqtt Broker 子进程,host/port 为监听地址
|
||||||
|
false → 使用外部 MQTT Broker,网关仅运行 Bridge 客户端,跳过本步
|
||||||
|
配置了 username/password 时 Broker 启用认证,否则保持匿名。
|
||||||
|
"""
|
||||||
env_config = config.get(env, {})
|
env_config = config.get(env, {})
|
||||||
mqtt_config = env_config.get("mqtt", {})
|
mqtt_config = env_config.get("mqtt", {})
|
||||||
if not mqtt_config.get("enabled", True):
|
if not mqtt_config.get("enabled", True):
|
||||||
logger.info("MQTT 已禁用")
|
logger.info("MQTT 已禁用")
|
||||||
return None
|
return None
|
||||||
|
if not mqtt_config.get("use_embedded_broker", True):
|
||||||
|
logger.info(
|
||||||
|
f"use_embedded_broker=false:使用外部 MQTT Broker "
|
||||||
|
f"({mqtt_config.get('host', 'localhost')}:{mqtt_config.get('port', 1883)}),"
|
||||||
|
f"跳过内嵌 amqtt Broker 启动"
|
||||||
|
)
|
||||||
|
return None
|
||||||
|
|
||||||
host = mqtt_config.get("host", "localhost")
|
host = mqtt_config.get("host", "localhost")
|
||||||
port = mqtt_config.get("port", 1883)
|
port = mqtt_config.get("port", 1883)
|
||||||
|
username = mqtt_config.get("username") or None
|
||||||
|
password = mqtt_config.get("password") or None
|
||||||
if not _acquire_singleton_lock("mqtt_broker"):
|
if not _acquire_singleton_lock("mqtt_broker"):
|
||||||
for _ in range(50):
|
for _ in range(50):
|
||||||
if is_port_in_use(port, host):
|
if is_port_in_use(port, host):
|
||||||
@@ -491,11 +543,12 @@ def setup_mqtt_broker(app: FastAPI):
|
|||||||
app.state.mqtt_broker_lock_acquired = False
|
app.state.mqtt_broker_lock_acquired = False
|
||||||
return
|
return
|
||||||
proc = multiprocessing.Process(
|
proc = multiprocessing.Process(
|
||||||
target=_run_mqtt_broker, args=(host, port), daemon=False
|
target=_run_mqtt_broker, args=(host, port, username, password), daemon=False
|
||||||
)
|
)
|
||||||
proc.start()
|
proc.start()
|
||||||
app.state.mqtt_broker_process = proc
|
app.state.mqtt_broker_process = proc
|
||||||
logger.info(f"MQTT Broker 子进程已启动 (PID: {proc.pid})")
|
auth_note = f",认证用户 {username}" if username else "(匿名)"
|
||||||
|
logger.info(f"MQTT Broker 子进程已启动 (PID: {proc.pid}){auth_note}")
|
||||||
for _ in range(50):
|
for _ in range(50):
|
||||||
if is_port_in_use(port, host):
|
if is_port_in_use(port, host):
|
||||||
logger.info(f"MQTT Broker 已监听 {port} 端口")
|
logger.info(f"MQTT Broker 已监听 {port} 端口")
|
||||||
@@ -521,9 +574,12 @@ async def shutdown_mqtt_broker(app: FastAPI):
|
|||||||
|
|
||||||
async def setup_mqtt(app: FastAPI, worker_manager: WorkerManager) -> MQTTBridge | None:
|
async def setup_mqtt(app: FastAPI, worker_manager: WorkerManager) -> MQTTBridge | None:
|
||||||
"""
|
"""
|
||||||
启动 MQTT Bridge(客户端模式,Broker 由子进程管理)。
|
启动 MQTT Bridge(纯客户端模式)。
|
||||||
|
|
||||||
从 config.toml 的环境段读取 host/port/enabled;若 enabled 为 false 则跳过。
|
从 config.toml 的环境段读取 host/port/enabled;若 enabled 为 false 则跳过。
|
||||||
|
Broker 来源由 use_embedded_broker 控制:
|
||||||
|
true (默认)= 内嵌 amqtt Broker(lifespan 第 3 步已启动子进程)
|
||||||
|
false = 外部 MQTT Broker(host/port 指向外部地址,网关不启动 Broker)
|
||||||
订阅主题 = 全局 [<env>.mqtt].custom_topics(可选兜底)
|
订阅主题 = 全局 [<env>.mqtt].custom_topics(可选兜底)
|
||||||
+ 各插件在其 config/workers/<type>/config.toml 中声明的 [[workers.mqtt_topics]]。
|
+ 各插件在其 config/workers/<type>/config.toml 中声明的 [[workers.mqtt_topics]]。
|
||||||
使用文件锁确保多个 gunicorn worker 中只有一个启动 Bridge。
|
使用文件锁确保多个 gunicorn worker 中只有一个启动 Bridge。
|
||||||
@@ -536,10 +592,13 @@ async def setup_mqtt(app: FastAPI, worker_manager: WorkerManager) -> MQTTBridge
|
|||||||
|
|
||||||
host = mqtt_config.get("host", "localhost")
|
host = mqtt_config.get("host", "localhost")
|
||||||
port = mqtt_config.get("port", 1883)
|
port = mqtt_config.get("port", 1883)
|
||||||
|
username = mqtt_config.get("username") or None
|
||||||
|
password = mqtt_config.get("password") or None
|
||||||
custom_topics = list(mqtt_config.get("custom_topics", []) or [])
|
custom_topics = list(mqtt_config.get("custom_topics", []) or [])
|
||||||
custom_topics += list(getattr(app.state, "plugin_mqtt_topics", []) or [])
|
custom_topics += list(getattr(app.state, "plugin_mqtt_topics", []) or [])
|
||||||
|
|
||||||
# Broker 子进程已在 lifespan 第 3 步启动(setup_mqtt_broker + 文件锁单例)
|
# Broker 来源:use_embedded_broker=true 时内嵌 amqtt Broker 子进程已在
|
||||||
|
# lifespan 第 3 步启动(setup_mqtt_broker + 文件锁单例);false 时连接外部 Broker。
|
||||||
|
|
||||||
# 原子锁确保只有一个 worker 启动 MQTT Bridge
|
# 原子锁确保只有一个 worker 启动 MQTT Bridge
|
||||||
if not _acquire_singleton_lock("mqtt_bridge"):
|
if not _acquire_singleton_lock("mqtt_bridge"):
|
||||||
@@ -553,6 +612,8 @@ async def setup_mqtt(app: FastAPI, worker_manager: WorkerManager) -> MQTTBridge
|
|||||||
host=host,
|
host=host,
|
||||||
port=port,
|
port=port,
|
||||||
custom_topics=custom_topics,
|
custom_topics=custom_topics,
|
||||||
|
username=username,
|
||||||
|
password=password,
|
||||||
)
|
)
|
||||||
app.state.mqtt_bridge = bridge
|
app.state.mqtt_bridge = bridge
|
||||||
await bridge.start()
|
await bridge.start()
|
||||||
@@ -560,7 +621,9 @@ async def setup_mqtt(app: FastAPI, worker_manager: WorkerManager) -> MQTTBridge
|
|||||||
registry = getattr(app.state, "route_registry", None)
|
registry = getattr(app.state, "route_registry", None)
|
||||||
if registry is not None:
|
if registry is not None:
|
||||||
registry.attach_bridge(bridge)
|
registry.attach_bridge(bridge)
|
||||||
logger.info(f"MQTT Bridge 已启动 ({host}:{port})")
|
broker_mode = "内嵌 amqtt" if mqtt_config.get("use_embedded_broker", True) else "外部"
|
||||||
|
auth_note = f",认证用户 {username}" if username else "(匿名)"
|
||||||
|
logger.info(f"MQTT Bridge 已启动({broker_mode} Broker {host}:{port}{auth_note})")
|
||||||
return bridge
|
return bridge
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -3,6 +3,7 @@ import asyncio
|
|||||||
import json
|
import json
|
||||||
import re
|
import re
|
||||||
from typing import Optional
|
from typing import Optional
|
||||||
|
from urllib.parse import quote
|
||||||
|
|
||||||
from amqtt.client import MQTTClient
|
from amqtt.client import MQTTClient
|
||||||
from amqtt.mqtt.constants import QOS_1
|
from amqtt.mqtt.constants import QOS_1
|
||||||
@@ -40,10 +41,13 @@ def _compile_topic(topic: str) -> re.Pattern:
|
|||||||
|
|
||||||
|
|
||||||
class MQTTBridge:
|
class MQTTBridge:
|
||||||
def __init__(self, worker_manager, host="localhost", port=1883, custom_topics=None):
|
def __init__(self, worker_manager, host="localhost", port=1883, custom_topics=None,
|
||||||
|
username=None, password=None):
|
||||||
self._wm = worker_manager
|
self._wm = worker_manager
|
||||||
self._host = host
|
self._host = host
|
||||||
self._port = port
|
self._port = port
|
||||||
|
self._username = username
|
||||||
|
self._password = password
|
||||||
self._client: Optional[MQTTClient] = None
|
self._client: Optional[MQTTClient] = None
|
||||||
self._lt: Optional[asyncio.Task] = None
|
self._lt: Optional[asyncio.Task] = None
|
||||||
self._running = False
|
self._running = False
|
||||||
@@ -82,9 +86,17 @@ class MQTTBridge:
|
|||||||
self._running = False
|
self._running = False
|
||||||
await self._sc_stop()
|
await self._sc_stop()
|
||||||
|
|
||||||
|
def _broker_uri(self) -> str:
|
||||||
|
"""构建 Broker URI;凭据经百分号编码(amqtt 连接时自动 unquote)。"""
|
||||||
|
if self._username:
|
||||||
|
user = quote(self._username, safe="")
|
||||||
|
pwd = quote(self._password or "", safe="")
|
||||||
|
return f"mqtt://{user}:{pwd}@{self._host}:{self._port}/"
|
||||||
|
return f"mqtt://{self._host}:{self._port}/"
|
||||||
|
|
||||||
async def _sc(self):
|
async def _sc(self):
|
||||||
self._client = MQTTClient(config={"auto_reconnect": False})
|
self._client = MQTTClient(config={"auto_reconnect": False})
|
||||||
await self._client.connect(f"mqtt://{self._host}:{self._port}/")
|
await self._client.connect(self._broker_uri())
|
||||||
topics = [(MQTT_SUBSCRIBE_TOPIC, QOS_1)]
|
topics = [(MQTT_SUBSCRIBE_TOPIC, QOS_1)]
|
||||||
topics += [(ct["topic"], QOS_1) for ct, _ in self._compiled]
|
topics += [(ct["topic"], QOS_1) for ct, _ in self._compiled]
|
||||||
await self._client.subscribe(topics)
|
await self._client.subscribe(topics)
|
||||||
|
|||||||
@@ -0,0 +1,170 @@
|
|||||||
|
"""
|
||||||
|
MQTT Broker 认证集成测试
|
||||||
|
|
||||||
|
验证 [<env>.mqtt].username/password 配置生效:
|
||||||
|
A. 内嵌 Broker 启用认证(FileAuthPlugin):
|
||||||
|
1. 正确凭据 → 连接/订阅/发布正常
|
||||||
|
2. 错误密码 → 连接被拒(ConnectError)
|
||||||
|
3. 匿名连接 → 连接被拒
|
||||||
|
B. 未配置凭据(默认)→ 匿名连接正常(原行为)
|
||||||
|
C. MQTTBridge 客户端凭据传递(含特殊字符密码的 URI 编码)
|
||||||
|
"""
|
||||||
|
import asyncio
|
||||||
|
import json
|
||||||
|
import multiprocessing
|
||||||
|
import socket
|
||||||
|
import sys
|
||||||
|
import time
|
||||||
|
from pathlib import Path
|
||||||
|
from urllib.parse import quote
|
||||||
|
|
||||||
|
sys.path.insert(0, str(Path(__file__).parent))
|
||||||
|
|
||||||
|
from amqtt.client import MQTTClient
|
||||||
|
from amqtt.errors import ConnectError
|
||||||
|
|
||||||
|
from src.chongming.app.core.lifespan_setup import _run_mqtt_broker
|
||||||
|
from src.chongming.app.core.mqtt_bridge import MQTTBridge
|
||||||
|
|
||||||
|
HOST = "127.0.0.1"
|
||||||
|
USERNAME = "cm_user"
|
||||||
|
PASSWORD = "p@ss w0rd!%#"
|
||||||
|
|
||||||
|
|
||||||
|
def cred_uri(port: int, user: str | None = None, pwd: str | None = None) -> str:
|
||||||
|
"""构建带凭据的 Broker URI(百分号编码,与 MQTTBridge._broker_uri 同语义)"""
|
||||||
|
if user:
|
||||||
|
return f"mqtt://{quote(user, safe='')}:{quote(pwd or '', safe='')}@{HOST}:{port}/"
|
||||||
|
return f"mqtt://{HOST}:{port}/"
|
||||||
|
|
||||||
|
|
||||||
|
def free_port() -> int:
|
||||||
|
with socket.socket() as s:
|
||||||
|
s.bind((HOST, 0))
|
||||||
|
return s.getsockname()[1]
|
||||||
|
|
||||||
|
|
||||||
|
def wait_port(port: int, timeout: float = 20.0) -> bool:
|
||||||
|
deadline = time.time() + timeout
|
||||||
|
while time.time() < deadline:
|
||||||
|
with socket.socket() as s:
|
||||||
|
s.settimeout(0.5)
|
||||||
|
if s.connect_ex((HOST, port)) == 0:
|
||||||
|
return True
|
||||||
|
time.sleep(0.1)
|
||||||
|
return False
|
||||||
|
|
||||||
|
|
||||||
|
def start_broker(port: int, username=None, password=None):
|
||||||
|
proc = multiprocessing.Process(
|
||||||
|
target=_run_mqtt_broker, args=(HOST, port, username, password), daemon=True
|
||||||
|
)
|
||||||
|
proc.start()
|
||||||
|
assert wait_port(port), f"Broker 未在 {port} 端口就绪"
|
||||||
|
return proc
|
||||||
|
|
||||||
|
|
||||||
|
async def try_connect(uri: str, expect_success: bool, label: str):
|
||||||
|
client = MQTTClient(config={"auto_reconnect": False})
|
||||||
|
try:
|
||||||
|
await client.connect(uri)
|
||||||
|
except ConnectError as e:
|
||||||
|
assert not expect_success, f"{label}: 应连接成功,实际被拒 ({e})"
|
||||||
|
return None
|
||||||
|
assert expect_success, f"{label}: 应被拒绝,实际连接成功"
|
||||||
|
return client
|
||||||
|
|
||||||
|
|
||||||
|
async def scenario_auth(port: int):
|
||||||
|
"""配置了账号密码 → 正确凭据通过,错误/匿名被拒"""
|
||||||
|
proc = start_broker(port, USERNAME, PASSWORD)
|
||||||
|
try:
|
||||||
|
# 1. 正确凭据
|
||||||
|
c = await try_connect(cred_uri(port, USERNAME, PASSWORD), True, "正确凭据")
|
||||||
|
await c.subscribe([("auth/t/#", 1)])
|
||||||
|
await c.publish("auth/t/1", b"hello", 1)
|
||||||
|
m = await asyncio.wait_for(c.deliver_message(), timeout=10)
|
||||||
|
assert m.data == b"hello", f"收到数据异常: {m.data}"
|
||||||
|
await c.disconnect()
|
||||||
|
print("PASS [A1] 正确凭据: 连接/订阅/发布/接收正常")
|
||||||
|
|
||||||
|
# 2. 错误密码
|
||||||
|
await try_connect(cred_uri(port, USERNAME, "wrong"), False, "错误密码")
|
||||||
|
print("PASS [A2] 错误密码: 连接被拒")
|
||||||
|
|
||||||
|
# 3. 匿名
|
||||||
|
await try_connect(cred_uri(port), False, "匿名连接")
|
||||||
|
print("PASS [A3] 匿名连接: 连接被拒")
|
||||||
|
finally:
|
||||||
|
proc.terminate()
|
||||||
|
proc.join(timeout=5)
|
||||||
|
|
||||||
|
|
||||||
|
async def scenario_anonymous(port: int):
|
||||||
|
"""未配置凭据(默认)→ 匿名可用"""
|
||||||
|
proc = start_broker(port)
|
||||||
|
try:
|
||||||
|
c = await try_connect(cred_uri(port), True, "匿名连接(默认)")
|
||||||
|
await c.disconnect()
|
||||||
|
print("PASS [B] 默认匿名模式: 匿名连接正常")
|
||||||
|
finally:
|
||||||
|
proc.terminate()
|
||||||
|
proc.join(timeout=5)
|
||||||
|
|
||||||
|
|
||||||
|
class _FakeWorkerManager:
|
||||||
|
async def call(self, *a, **kw):
|
||||||
|
return {"ok": True}
|
||||||
|
|
||||||
|
|
||||||
|
async def scenario_bridge(port: int):
|
||||||
|
"""MQTTBridge 客户端携带凭据(特殊字符密码经 URI 百分号编码)"""
|
||||||
|
proc = start_broker(port, USERNAME, PASSWORD)
|
||||||
|
sub = None
|
||||||
|
try:
|
||||||
|
sub = MQTTClient(config={"auto_reconnect": False})
|
||||||
|
await sub.connect(cred_uri(port, USERNAME, PASSWORD))
|
||||||
|
await sub.subscribe([("chongming/#", 1)])
|
||||||
|
|
||||||
|
bridge = MQTTBridge(worker_manager=_FakeWorkerManager(),
|
||||||
|
host=HOST, port=port,
|
||||||
|
username=USERNAME, password=PASSWORD)
|
||||||
|
await bridge.start()
|
||||||
|
assert bridge.is_running, "Bridge 未连接"
|
||||||
|
|
||||||
|
# 发布请求 → Bridge 路由到 worker → 响应回发(验证完整链路带认证)
|
||||||
|
pub = MQTTClient(config={"auto_reconnect": False})
|
||||||
|
await pub.connect(cred_uri(port, USERNAME, PASSWORD))
|
||||||
|
await pub.subscribe([("chongming/demo/op/response", 1)])
|
||||||
|
await pub.publish(
|
||||||
|
"chongming/demo/op",
|
||||||
|
b'{"meta": {"timeout": 10}, "data": {"x": 1}}', 1)
|
||||||
|
m = await asyncio.wait_for(sub.deliver_message(), timeout=10)
|
||||||
|
assert m.topic == "chongming/demo/op", f"未收到请求: {m.topic}"
|
||||||
|
m2 = await asyncio.wait_for(sub.deliver_message(), timeout=10)
|
||||||
|
assert m2.topic == "chongming/demo/op/response", f"未收到响应: {m2.topic}"
|
||||||
|
assert json.loads(m2.data) == {"ok": True}, f"响应内容异常: {m2.data}"
|
||||||
|
await bridge.stop()
|
||||||
|
await pub.disconnect()
|
||||||
|
print("PASS [C] MQTTBridge 凭据传递: 连接/路由/响应链路正常(含特殊字符密码)")
|
||||||
|
finally:
|
||||||
|
if sub and sub.session is not None:
|
||||||
|
try:
|
||||||
|
await sub.disconnect()
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
proc.terminate()
|
||||||
|
proc.join(timeout=5)
|
||||||
|
|
||||||
|
|
||||||
|
async def main():
|
||||||
|
p1, p2, p3 = free_port(), free_port(), free_port()
|
||||||
|
await scenario_auth(p1)
|
||||||
|
await scenario_anonymous(p2)
|
||||||
|
await scenario_bridge(p3)
|
||||||
|
print("\n全部通过 ✔")
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
multiprocessing.set_start_method("fork", force=True)
|
||||||
|
asyncio.run(main())
|
||||||
@@ -425,7 +425,7 @@ dependencies = [
|
|||||||
[[package]]
|
[[package]]
|
||||||
name = "chongming-cli"
|
name = "chongming-cli"
|
||||||
version = "0.1.0"
|
version = "0.1.0"
|
||||||
source = { git = "http://124.71.68.6:3000/chongming/chongming-utils-cli.git#0bf60021cbeb32610dce232eaf0e25be5d8d8b29" }
|
source = { git = "http://124.71.68.6:3000/chongming/chongming-utils-cli.git#e11d69d1ef49b5bec84f9dd7efdbc2259dac37d1" }
|
||||||
dependencies = [
|
dependencies = [
|
||||||
{ name = "chongming-config" },
|
{ name = "chongming-config" },
|
||||||
{ name = "module-bank" },
|
{ name = "module-bank" },
|
||||||
@@ -661,15 +661,6 @@ wheels = [
|
|||||||
{ url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/3f/27/4570e78fc0bf5ea0ca45eb1de3818a23787af9b390c0b0a0033a1b8236f9/diskcache-5.6.3-py3-none-any.whl", hash = "sha256:5e31b2d5fbad117cc363ebaf6b689474db18a1f6438bc82358b024abd4c2ca19" },
|
{ url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/3f/27/4570e78fc0bf5ea0ca45eb1de3818a23787af9b390c0b0a0033a1b8236f9/diskcache-5.6.3-py3-none-any.whl", hash = "sha256:5e31b2d5fbad117cc363ebaf6b689474db18a1f6438bc82358b024abd4c2ca19" },
|
||||||
]
|
]
|
||||||
|
|
||||||
[[package]]
|
|
||||||
name = "distro"
|
|
||||||
version = "1.9.0"
|
|
||||||
source = { registry = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/simple" }
|
|
||||||
sdist = { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/fc/f8/98eea607f65de6527f8a2e8885fc8015d3e6f5775df186e443e0964a11c3/distro-1.9.0.tar.gz", hash = "sha256:2fa77c6fd8940f116ee1d6b94a2f90b13b5ea8d019b98bc8bafdcabcdd9bdbed", size = 60722, upload-time = "2023-12-24T09:54:32.31Z" }
|
|
||||||
wheels = [
|
|
||||||
{ url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/12/b3/231ffd4ab1fc9d679809f356cebee130ac7daa00d6d6f3206dd4fd137e9e/distro-1.9.0-py3-none-any.whl", hash = "sha256:7bffd925d65168f85027d8da9af6bddab658135b840670a223589bc0c8ef02b2", size = 20277, upload-time = "2023-12-24T09:54:30.421Z" },
|
|
||||||
]
|
|
||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "eclipse-zenoh"
|
name = "eclipse-zenoh"
|
||||||
version = "1.10.0"
|
version = "1.10.0"
|
||||||
@@ -998,20 +989,19 @@ sdist = { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/3f/d8/bd
|
|||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "openai"
|
name = "openai"
|
||||||
version = "3.3.0"
|
version = "3.3.1"
|
||||||
source = { registry = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/simple" }
|
source = { registry = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/simple" }
|
||||||
dependencies = [
|
dependencies = [
|
||||||
{ name = "anyio" },
|
{ name = "anyio" },
|
||||||
{ name = "distro" },
|
|
||||||
{ name = "httpx2" },
|
{ name = "httpx2" },
|
||||||
{ name = "jiter" },
|
{ name = "jiter" },
|
||||||
{ name = "pydantic" },
|
{ name = "pydantic" },
|
||||||
{ name = "sniffio" },
|
{ name = "sniffio" },
|
||||||
{ name = "typing-extensions" },
|
{ name = "typing-extensions" },
|
||||||
]
|
]
|
||||||
sdist = { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/e6/6c/670271205379061112bf5ed900b78a68e21bf11733de829772c5322f92aa/openai-3.3.0.tar.gz", hash = "sha256:28f904a9fbff15288e9dd67bc0f7020d4f576d37786383a480db02fe5d8139d3", size = 1166178, upload-time = "2026-08-18T21:17:53.419Z" }
|
sdist = { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/7d/9c/ba0c292b4032ede74c249ca314ad64eb1bb5a03a843f6e01facb02f80cd8/openai-3.3.1.tar.gz", hash = "sha256:6f22807de1a976c932cecda620e8172a8c3fdbaeed29c7f21564e0c2410edf56", size = 1282113, upload-time = "2026-08-19T16:31:35.006Z" }
|
||||||
wheels = [
|
wheels = [
|
||||||
{ url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/82/83/d44980712bf51dddaca090cd772b3102e75377ddac877eaa4b2cb58d886a/openai-3.3.0-py3-none-any.whl", hash = "sha256:ded6b2112e6d299c7a2573ff6f165dc92fb64ceaa4d7daa42345f091157bd373", size = 1690265, upload-time = "2026-08-18T21:17:51.351Z" },
|
{ url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/6a/db/2b7a1b3de659bb82aef979116c74e809982b13e42c057759767552b5155f/openai-3.3.1-py3-none-any.whl", hash = "sha256:9652df7fdf8ee6f5bd58e0a12f2b1d414a18e0f06bb7a9a57c8643a5f5469bd3", size = 1690337, upload-time = "2026-08-19T16:31:32.812Z" },
|
||||||
]
|
]
|
||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
|
|||||||
Reference in New Issue
Block a user