转发消息至用户业务系统

{
  "version": "1.0",
  "profile": {
    "ProductId": "YOUR_PRODUCT_ID",
    "CategoryId": "YOUR_CATEGORY_ID"
  },
  "properties": [
    {
      "id": "power_switch", "name": "设备开关", "mode": "rw", "required": false,
      "define": {"type": "bool", "mapping": {"0": "关闭", "1": "开启"}}
    },
    {
      "id": "brightness", "name": "灯光亮度", "mode": "rw", "required": false,
      "define": {"type": "int", "unit": "%", "min": "0", "max": "100", "step": "1"}
    },
    {
      "id": "temperature", "name": "设备温度", "mode": "r", "required": false,
      "define": {"type": "float", "unit": "℃", "min": "-20", "max": "100", "step": "0.1"}
    }
  ],
  "events": [],
  "actions": []
}


'use strict';

const express = require('express');
const crypto = require('crypto');

const app = express();
const TOKEN = process.env.IOT_FORWARD_TOKEN;

app.use(express.json({ limit: '1mb' }));

function verifySignature(req, res, next) {
    if (!TOKEN) {
        return res.status(500).send('Token not configured');
    }

    const signature = String(
        req.headers.signature || ''
    ).trim().toLowerCase();

    const timestamp = String(req.headers.timestamp || '');
    const nonce = String(req.headers.nonce || '');

    if (
        !timestamp ||
        !nonce ||
        !/^[0-9a-f]{40}$/.test(signature)
    ) {
        return res.status(403).send('Forbidden');
    }

    const parts = [TOKEN, timestamp, nonce].sort();
    const source = parts.join('');

    const expected = crypto
        .createHash('sha1')
        .update(source, 'utf8')
        .digest('hex');

    const valid = crypto.timingSafeEqual(
        Buffer.from(signature, 'hex'),
        Buffer.from(expected, 'hex')
    );

    if (!valid) {
        return res.status(403).send('Forbidden');
    }

    next();
}

// HTTP 服务地址校验
app.get('/test', verifySignature, (req, res) => {
    const echo = req.headers.echostr;

    if (typeof echo !== 'string') {
        return res.status(400).send('Missing Echostr');
    }

    console.log('[VERIFY] 地址校验成功');

    return res.status(200)
        .type('text/plain')
        .send(echo);
});

// 设备消息接收
app.post('/test', verifySignature, (req, res) => {
    console.log(JSON.stringify({
        event: 'IOT_MESSAGE_RECEIVED',
        time: new Date().toISOString(),
        body: req.body
    }));

    return res.status(200)
        .type('text/plain')
        .send('success');
});

// SCF Web 函数监听 9000 端口
app.listen(9000, '0.0.0.0', () => {
    console.log('[SCF] HTTP Server started');
});

https://<实际 SCF 公网函数 URL>/test

curl.exe -i "https://<实际 SCF 公网函数 URL>/test"

HTTP/1.1 403 Forbidden

python -m pip install "paho-mqtt>=2.1,<3.0"

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Tencent Cloud IoT Explorer device simulator, MQTT HMAC/PSK auth, port 1883.
WARNING: Port 1883 is unencrypted. Use test-only device credentials.
"""

import base64
import binascii
import getpass
import hashlib
import hmac
import json
import secrets
import string
import threading
import time
from datetime import datetime

try:
    import paho.mqtt.client as mqtt
except ImportError as exc:
    raise SystemExit('请先运行: python -m pip install "paho-mqtt>=2.1,<3"') from exc

SDK_APP_ID = "12010126"


def log(tag, message):
    print(f"\\n[{datetime.now():%H:%M:%S}] {tag:<10} {message}", flush=True)


def required(prompt, secret=False):
    while True:
        value = (getpass.getpass(prompt) if secret else input(prompt)).strip()
        if value:
            return value
        print("此项不能为空。")


def mqtt_credentials(product_id, device_name, device_secret):
    client_id = product_id + device_name
    connid = ''.join(secrets.choice(string.ascii_uppercase + string.digits) for _ in range(5))
    expiry = int(time.time()) + 24 * 3600
    username = f"{client_id};{SDK_APP_ID};{connid};{expiry}"
    try:
        key = base64.b64decode(device_secret, validate=True)
    except (ValueError, binascii.Error) as exc:
        raise ValueError("DeviceSecret 不是有效的 Base64 字符串") from exc
    if not key:
        raise ValueError("DeviceSecret 解码为空")
    signed = hmac.new(key, username.encode("utf-8"), hashlib.sha256).hexdigest()
    return client_id, username, f"{signed};hmacsha256"


class Simulator:
    def __init__(self, product_id, device_name, device_secret, broker):
        self.product_id = product_id
        self.device_name = device_name
        self.device_secret = device_secret
        self.broker = broker
        self.topic_up = f"$thing/up/property/{product_id}/{device_name}"
        self.topic_down = f"$thing/down/property/{product_id}/{device_name}"
        self.values = {"power_switch": 0, "brightness": 50, "temperature": 25.0}
        self.connected = threading.Event()
        self.client = None
        self.intentional_disconnect = False

    def on_connect(self, client, userdata, flags, reason_code, properties):
        if reason_code.is_failure:
            self.connected.clear()
            log("ERROR", f"MQTT 连接被拒绝: {reason_code}")
            return
        self.connected.set()
        log("CONNECTED", f"已连接到 {self.broker}:1883")
        result, mid = client.subscribe(self.topic_down, qos=1)
        if result == mqtt.MQTT_ERR_SUCCESS:
            log("SUBSCRIBE", self.topic_down)
        else:
            log("ERROR", f"订阅失败: {result}")

    def on_connect_fail(self, client, userdata):
        log("ERROR", "MQTT 网络连接失败，请检查地址、端口及网络")

    def on_disconnect(self, client, userdata, disconnect_flags, reason_code, properties):
        self.connected.clear()
        if not self.intentional_disconnect:
            log("DISCONNECT", f"异常断开: {reason_code}，客户端可能尝试重连")

    def on_message(self, client, userdata, message):
        text = message.payload.decode("utf-8", errors="replace")
        try:
            data = json.loads(text)
        except json.JSONDecodeError:
            log("RECEIVE", f"{message.topic}: {text}")
            return

        method = data.get("method", "unknown")
        if method == "report_reply":
            code = data.get("code")
            tag = "REPORT_OK" if code == 0 else "REPORT_ERR"
            log(tag, f"clientToken={data.get('clientToken')} code={code} status={data.get('status', '')}")
        elif method == "control":
            # V1仅做上报验证；云端控制不在本例处理范围内。
            log("CONTROL", "收到控制请求（V1未实现执行/回复）: " + json.dumps(data, ensure_ascii=False))
        else:
            log("RECEIVE", json.dumps(data, ensure_ascii=False))

    def connect(self):
        if self.client is not None:
            self.disconnect()
        client_id, username, password = mqtt_credentials(
            self.product_id, self.device_name, self.device_secret
        )
        client = mqtt.Client(
            callback_api_version=mqtt.CallbackAPIVersion.VERSION2,
            client_id=client_id,
            clean_session=True,
            protocol=mqtt.MQTTv311,
            reconnect_on_failure=True,
        )
        client.username_pw_set(username, password)
        client.on_connect = self.on_connect
        client.on_connect_fail = self.on_connect_fail
        client.on_disconnect = self.on_disconnect
        client.on_message = self.on_message
        self.intentional_disconnect = False
        self.client = client
        log("CONNECT", f"正在连接 {self.broker}:1883 / {self.device_name} ...")
        try:
            # MQTT 1883: 仅供临时验收，不使用 TLS；不要输入正式生产设备的密钥。
            client.connect(self.broker, 1883, keepalive=60)
            client.loop_start()
        except Exception as exc:
            self.client = None
            self.connected.clear()
            log("ERROR", f"连接启动失败: {exc}")

    def disconnect(self):
        client = self.client
        if client is None:
            log("INFO", "当前未连接")
            return
        self.intentional_disconnect = True
        self.connected.clear()
        self.client = None
        try:
            client.disconnect()
            client.loop_stop()
        finally:
            log("DISCONNECT", "已主动断开连接")

    def publish(self, fields):
        if not self.connected.is_set() or self.client is None:
            log("WARN", "设备尚未连接，按 C 连接后再发送")
            return
        token = f"demo-{int(time.time() * 1000)}-{secrets.token_hex(2)}"
        message = {
            "method": "report",
            "clientToken": token,
            "timestamp": int(time.time() * 1000),
            "params": fields,
        }
        raw = json.dumps(message, ensure_ascii=False, separators=(",", ":"))
        info = self.client.publish(self.topic_up, raw, qos=1)
        if info.rc != mqtt.MQTT_ERR_SUCCESS:
            log("PUBLISH_ERR", f"MQTT 客户端未能排队发送: {info.rc}")
            return
        log("PUBLISH", f"Topic: {self.topic_up}\\n{json.dumps(message, ensure_ascii=False, indent=2)}")
        log("INFO", "消息已交给 MQTT 客户端；请以 report_reply code=0 判断平台处理成功。")

    def show_state(self):
        log("STATE", "连接状态: " + ("在线" if self.connected.is_set() else "离线"))
        log("STATE", json.dumps(self.values, ensure_ascii=False, indent=2))


def print_menu():
    print("\\n" + "=" * 50)
    print("  Tencent Cloud IoT | 设备模拟器 V1.0")
    print("  1 开启   2 关闭   3 设置亮度   4 设置温度")
    print("  5 上报全部属性    6 查看状态")
    print("  C 连接/重连      D 断开连接")
    print("  H 帮助           Q 退出")
    print("=" * 50)


def ranged_number(label, lower, upper, integer=False):
    while True:
        try:
            raw = input(f"{label} ({lower}~{upper}): ").strip()
            value = int(raw) if integer else float(raw)
            if not (lower <= value <= upper):
                raise ValueError("超出范围")
            return value
        except ValueError:
            print("输入无效，请重新输入有效数字。")


def main():
    print_menu()
    print("请输入测试设备三元组（本程序不会保存密钥或打印密钥）。")
    product_id = required("ProductID: ")
    device_name = required("DeviceName: ")
    device_secret = required("DeviceSecret (输入隐藏): ", secret=True)
    default_broker = f"{product_id}.iotcloud.tencentdevices.com"
    broker = input(f"MQTT Broker [{default_broker}]，按 Enter 默认: ").strip() or default_broker
    if "://" in broker or "/" in broker or ":" in broker:
        raise ValueError("Broker 只能填主机名，不应包含协议、路径或端口")
    sim = Simulator(product_id, device_name, device_secret, broker)
    try:
        # 先在本地检查设备密钥是否合法，再开始连接。
        mqtt_credentials(product_id, device_name, device_secret)
        sim.connect()
        while True:
            cmd = input("\\nIoT ❯ ").strip().lower()
            if cmd == "1":
                sim.values["power_switch"] = 1
                sim.publish({"power_switch": 1})
            elif cmd == "2":
                sim.values["power_switch"] = 0
                sim.publish({"power_switch": 0})
            elif cmd == "3":
                value = ranged_number("亮度", 0, 100, integer=True)
                sim.values["brightness"] = value
                sim.publish({"brightness": value})
            elif cmd == "4":
                value = round(ranged_number("温度", -20, 100), 1)
                sim.values["temperature"] = value
                sim.publish({"temperature": value})
            elif cmd == "5":
                sim.publish(dict(sim.values))
            elif cmd == "6":
                sim.show_state()
            elif cmd == "c":
                sim.connect()
            elif cmd == "d":
                sim.disconnect()
            elif cmd in ("h", "help", "?"):
                print_menu()
            elif cmd in ("q", "quit", "exit"):
                break
            elif not cmd:
                continue
            else:
                log("WARN", "未知命令，输入 H 查看帮助")
    except (KeyboardInterrupt, EOFError):
        print("\\n用户中止运行")
    finally:
        if sim.client is not None:
            sim.disconnect()
    return 0


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except ValueError as exc:
        raise SystemExit(f"参数错误: {exc}")