从AI助手控制空调看物联网安全 ☠️

又名:我只是想给小智加个计算器,怎么就黑进了自家空调?


起因:小智想要个计算器 🧮

一切的开端非常单纯——我想让我的 AI 助手「小智」能算数。

GitHub 上有个项目 78/mcp-calculator,提供了 MCP(Model Context Protocol)计算器的示例代码。原理很简单:

用户对小智说"帮我算个东西"
    → 小智通过 WebSocket 调用你的 MCP Server
    → Server 执行 eval() 返回结果
    → 小智把答案告诉你

踩坑第一步:搞清楚架构

这个项目有两部分:

  • calculator.py — 真正的 MCP 工具实现,通过 stdio(标准输入输出)通信
  • mcp_pipe.py — WebSocket 桥接器,把小智平台的 WebSocket 数据和本地 stdio 串起来

架构长这样:

小智平台 ←──WebSocket──→ mcp_pipe.py ←──stdio──→ calculator_server.py

踩坑第二步:MCP 工具命名的学问

官方示例只有几行代码,但实际用起来有几个坑。我根据实际经验总结了几条铁律:

⚠️ MCP 工具设计的 6 条注意事项

  1. 工具名和参数名必须让大模型一看就懂,不要用缩写。比如 calculator 而非 calcpython_expression 而非 expr
  2. """...""" 文档注释引导大模型何时使用该工具,可以列出具体使用场景
  3. MCP Server 的 stdio 被用来传数据,不能用 print(),必须用 logger 输出调试信息
  4. 返回值控制在 1024 字节内,放在 JSON 的 result 字段里
  5. 工具列表报文有 token 数上限(后续在小智配置页显示)
  6. 连接数有上限,尽量精简工具数量

第一版计算器代码

# calculator_server.py
"""
MCP Calculator Server — 提供一个安全的数学计算工具给大模型使用。
基于 78/mcp-calculator (https://github.com/78/mcp-calculator) 改进:
- 工具和参数命名清晰,不使用缩写
- 完善的文档注释引导大模型何时使用该工具
- 使用 logger 而非 print 输出调试信息
- 返回结构化的 JSON,控制在 1024 字节内
"""

from fastmcp import FastMCP
import sys
import logging
import math
import random

# ── 日志配置 ────────────────────────────────────────────
# 注意:MCP Server 的 stdio (stdin/stdout) 用于数据传输,
# 因此不能使用 print(),必须通过 logger 输出调试信息。
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    stream=sys.stderr  # 输出到 stderr,不干扰 stdio 数据传输
)
logger = logging.getLogger('Calculator')

# 修复 Windows 控制台 UTF-8 编码问题
if sys.platform == 'win32':
    sys.stderr.reconfigure(encoding='utf-8')
    sys.stdout.reconfigure(encoding='utf-8')

mcp = FastMCP("Calculator")

@mcp.tool()
def calculator(python_expression: str) -> dict:
    """
    执行数学计算。当你需要进行数学运算、数值计算、公式求值时,
    请使用这个工具。它接受一个 Python 表达式作为输入并返回计算结果。

    参数 python_expression 要求:一个合法的 Python 数学表达式字符串。
    你可以直接在表达式中使用 'math' 和 'random' 这两个库中的函数,
    无需 import。例如:
      - 基本运算: "2 + 3 * 4"
      - 使用 math 库: "math.sqrt(16)", "math.sin(math.pi / 2)"
      - 使用 random 库: "random.randint(1, 100)"
    """
    try:
        result = eval(python_expression, {"math": math, "random": random})  # 在受限命名空间中执行
        logger.info(f"Calculating formula: {python_expression}, result: {result}")
        return {"success": True, "result": result}  # 返回值控制在1024字节内
    except Exception as error:
        logger.error(f"Calculation failed for '{python_expression}': {error}")
        return {"success": False, "result": None, "error": f"计算失败: {str(error)}"}

if __name__ == "__main__":
    logger.info("Starting MCP Calculator Server...")
    mcp.run(transport="stdio")

小智的配置(mcp_config.json):

{
  "mcpServers": {
    "calculator": {
      "type": "stdio",
      "command": "python",
      "args": ["calculator_server.py"]
    }
  }
}

设置好 .env 中的 MCP_ENDPOINT(小智平台给的 WebSocket 地址),然后:

python mcp_pipe.py

小智就学会算数了!🎉


膨胀:一个MCP可以接多个工具吗?🤔

当然可以!一个 FastMCP 实例可以注册多个 @mcp.tool()mcp_pipe.py 会把它们全部桥接给小智。

我咔咔又加了两个:

@mcp.tool()
def generate_random_number(min_value: float, max_value: float, decimal_places: int = 0) -> dict:
    """生成指定范围内的随机数,用于抽签、选号、掷骰子等场景"""
    ...

@mcp.tool()
def convert_units(value: float, from_unit: str, to_unit: str) -> dict:
    """单位换算。支持长度(m/km/cm/mm/inch/ft)、重量(kg/g/lb/oz)、温度(C/F/K)"""
    ...

三个工具加起来才几百 token,离上限远着呢。一个连接、一个进程,N 个工具。

就像这样:

mcp_config.json
  ├── calculator (3个工具: 计算/随机数/单位换算)
  └── gree-ac    (3个工具: 扫描/控制/查询)
        ↓
mcp_pipe.py → 一条 WebSocket → 小智看到 6 个工具

转折:既然能控制软件,能不能控制硬件?🔌

我问了 AI:

“我想加入一个控制格力空调的功能,有这个插件吗?”

答案是:格力的通信协议早就被逆向出来了

有个 Python 库叫 greeclimate,通过局域网 UDP 直接跟格力空调通信——和格力+ App 用的是同一套协议。

pip install greeclimate,一行命令的事。

第一个版本 —— 翻车实录 🚗💥

我照着文档写了第一版:

# gree_server.py(初版,全是bug)
@mcp.tool()
async def control_gree_ac(device_ip: str, power_on: bool = None, ...):
    """控制格力空调"""
    discovery = Discovery()
    device = None
    async for info in discovery.scan(wait_for=5):  # ❌ Bug 1: scan()不是异步迭代器!
        if info.host == device_ip:                   # ❌ Bug 2: 属性名是 .ip 不是 .host!
            device = Device(info)
            break

报错长这样:

TypeError: 'async for' requires an object with __aiter__ method, got coroutine

以及:

AttributeError: 'DeviceInfo' object has no attribute 'host'

教训:看文档不如看源码。花两分钟读一下 deviceinfo.py 就知道属性名叫 ip 不是 host;读一下 discovery.py 的签名就知道 scan() 返回 list 不是 AsyncIterator

诊断脚本 —— 步步为营 🔍

修完 bug 后跑了个诊断脚本(记得先关 Windows 防火墙,UDP 7000 端口经常被拦):

# gree_debug.py
"""格力空调诊断脚本"""
import asyncio
from greeclimate.discovery import Discovery
from greeclimate.device import Device

async def main():
    # 步骤1: 扫描设备
    print("🔍 扫描局域网格力设备...")
    discovery = Discovery()
    devices = await discovery.scan(wait_for=10)  # scan() 返回列表,不是异步迭代器
    for info in devices:
        print(f"  ✅ 发现: {info.name} @ {info.ip}:7000")  # 属性名是 .ip 不是 .host!

    # 步骤2: 绑定设备(交换加密密钥)
    device = Device(devices[0])
    await device.bind()  # 自动尝试 V1(ECB) → 失败 → V2(GCM),需要耐心等
    print(f"  ✅ 绑定成功!")

    # 步骤3: 读取状态
    await device.update_state()
    await asyncio.sleep(1)  # 等UDP异步响应回来
    print(f"  状态: 开机={device.power}, 模式={device.mode}, 温度={device.target_temperature}°C")

asyncio.run(main())

输出:

🔍 扫描局域网格力设备...
  ✅ 发现: 1e4a8b6e @ 192.168.1.104:7000 (mac: f4911e4a8b6e)  version: V1.2.1

🔐 绑定设备...
  尝试 CipherV1... ❌ timeout
  尝试 CipherV2... ✅ 绑定成功! key: 9Op2Rs5Uv8Xy1Ab4

📊 读取状态: 已关机, 温度 26°C

设备版本 V1.2.1 用的是 GCM 加密(CipherV2),greeclimate 库会自动降级尝试,10 秒内搞定。

加入小智 —— 同样的模式

mcp_config.json 扩展一下:

{
  "mcpServers": {
    "calculator": {
      "type": "stdio",
      "command": "python",
      "args": ["calculator_server.py"]
    },
    "gree-ac": {
      "type": "stdio",
      "command": "python",
      "args": ["gree_server.py"]
    }
  }
}

重启 python mcp_pipe.py,小智就有了 6 个工具。跟它说"打开空调"它就知道调 control_air_conditioner

又翻了:小智太蠢了,听不懂人话 🤦

工具名叫 control_gree_ac,参数叫 power_on: bool,大模型根本不能从"开空调"这三个字映射过去。

改!要让工具名和描述充满触发式中文

@mcp.tool()
async def control_air_conditioner(
    action: str,           # "turn_on"=开空调, "turn_off"=关空调
    temperature: int = None,  # 目标温度,如用户说"26度"则传26
    mode: str = None,      # "cool"(制冷), "heat"(制热)
    fan_speed: str = None, # "low"(低), "medium"(中), "high"(高)
    device_ip: str = "192.168.1.104"  # 写死IP,省得每次扫描
) -> dict:
    """
    【开关空调、调节温度和模式】

    当用户说以下任何话时,必须调用此工具:
      - "开空调" "打开空调" → action="turn_on"
      - "关空调" "空调关机" → action="turn_off"
      - "空调调到26度" → action="turn_on", temperature=26
      - "制冷" "制热" "除湿" → mode参数
      - "风速大一点" → fan_speed="high"
    """

终极版就是让描述里写满用户可能说的原话,大模型不是程序员,它是模式匹配器。你写"用户说 X → 参数填 Y",它就能对上。

这一版上线后,小智终于能从"听不懂"进化到"秒懂"。🎯


顿悟:好像…不太对劲?😨

一切都太顺利了。然后我突然意识到一个恐怖的事实:

1. pip install greeclimate     ← 任何人都能装
2. python gree_debug.py        ← 一行命令扫描全网段的格力空调
3. 没有密码、没有配对确认、没有任何认证
4. 加密密钥硬编码在开源库里   ← 全世界的格力空调用同一把钥匙!

这意味着什么?

只要连上你家 WiFi,任何人都能控制你的空调。

陌生人可以在凌晨三点把你的空调开到 16°C 制冷,或者冬天开到 30°C 制热跑一晚上。更缺德的是反复开关——压缩机的寿命就这样被消耗。


为什么消费级 IoT 这么不安全?🏚️

答案比你想的更无语:厂商的设计假设是 “能连上我家 WiFi 的人都是可信的”

在 2015 年这个假设也许勉强成立。但在 2026 年:

  • 你家的 WiFi 密码可能被客人分享过
  • 装过万能钥匙的家人手机已经把密码上传到了云端
  • 路由器漏洞可以让攻击者直接进内网
  • IoT 设备的固件出厂后基本不会更新

所有品牌的消费级智能空调,只要能用手机 App 控制,安全模型大同小异:

品牌 协议 安全
格力 局域网 UDP,硬编码密钥 同网段即可控制
美的 类似,有开源库 midea-msmart 同上
小米/米家 miIO 协议,云下发 token 拿到 token 后局域网可控
大金/日立/海尔 类似架构 基本一样

所有不经过苹果 HomeKit 认证的智能家居设备,局域网安全都形同虚设。 HomeKit 至少要求硬件级安全芯片,但贵得多。


怎么自保?🛡️

你不可能把所有家电都换成 HomeKit 认证的,但可以在网络层面隔离

措施 效果 成本
VLAN 隔离 — IoT 单独网段 ✅ 治本 需要支持 VLAN 的路由器(OpenWrt/中高端)
访客 WiFi — IoT 走客人网络 ✅ 可用 大部分路由器都支持
改路由器管理员密码 ❌ 没用
隐藏 SSID ❌ 自欺欺人
MAC 白名单 ⚠️ 部分有用但可伪造

一句话:把空调、摄像头、灯泡放在一个独立的 WiFi/网段里,跟你的电脑手机隔开。这样就算主网密码泄露了,IoT 设备也碰不到。


完整的工具集 📦

经过一天的折腾,小智现在有了这些能力:

工具 功能 代码文件
calculator Python 表达式计算(含 math/random) calculator_server.py
generate_random_number 生成随机数 同上
convert_units 单位换算(长度/重量/温度) 同上
scan_for_air_conditioners 扫描局域网格力空调 gree_server.py
control_air_conditioner 开关空调/调温/设模式/调风速 同上
check_air_conditioner_status 查询空调运行状态 同上

配置文件 mcp_config.json 管理所有服务器,mcp_pipe.py 做 WebSocket 桥接,.env 存放小智平台地址。


Logo

作为“人工智能6S店”的官方数字引擎,为AI开发者与企业提供一个覆盖软硬件全栈、一站式门户。

更多推荐