import json
|
import time
|
import base64
|
import hashlib
|
import hmac
|
import os
|
import re
|
import glob
|
from typing import Dict, List, Tuple
|
from aiohttp import web
|
|
from core.auth import AuthManager
|
from core.utils.util import get_local_ip, get_vision_url
|
from core.api.base_handler import BaseHandler
|
|
TAG = __name__
|
|
|
def _safe_basename(filename: str) -> str:
|
# Prevent directory traversal
|
return os.path.basename(filename)
|
|
|
def _parse_version(ver: str) -> Tuple[int, ...]:
|
# conservative parser: split by non-digit, keep numeric parts
|
parts = re.findall(r"\d+", ver)
|
return tuple(int(p) for p in parts) if parts else (0,)
|
|
|
def _is_higher_version(a: str, b: str) -> bool:
|
"""Return True if version string a > b (semver-like numeric compare)."""
|
ta = _parse_version(a)
|
tb = _parse_version(b)
|
# compare tuple lexicographically, but allow different lengths
|
maxlen = max(len(ta), len(tb))
|
for i in range(maxlen):
|
ai = ta[i] if i < len(ta) else 0
|
bi = tb[i] if i < len(tb) else 0
|
if ai > bi:
|
return True
|
if ai < bi:
|
return False
|
return False
|
|
|
class OTAHandler(BaseHandler):
|
def __init__(self, config: dict):
|
super().__init__(config)
|
auth_config = config["server"].get("auth", {})
|
self.auth_enable = auth_config.get("enabled", False)
|
# 设备白名单
|
self.allowed_devices = set(auth_config.get("allowed_devices", []))
|
secret_key = config["server"]["auth_key"]
|
expire_seconds = auth_config.get("expire_seconds")
|
self.auth = AuthManager(secret_key=secret_key, expire_seconds=expire_seconds)
|
|
# firmware storage
|
self.bin_dir = os.path.join(os.getcwd(), "data", "bin")
|
# cache structure: { 'updated_at': timestamp, 'ttl': seconds, 'files_by_model': { model: [(version, filename), ...] } }
|
self._bin_cache: Dict = {
|
"updated_at": 0,
|
"ttl": config.get("firmware_cache_ttl", 30),
|
"files_by_model": {},
|
}
|
|
def _refresh_bin_cache_if_needed(self):
|
now = int(time.time())
|
ttl = int(self._bin_cache.get("ttl", 30))
|
if now - int(
|
self._bin_cache.get("updated_at", 0)
|
) < ttl and self._bin_cache.get("files_by_model"):
|
return
|
|
files_by_model: Dict[str, List[Tuple[str, str]]] = {}
|
try:
|
if not os.path.isdir(self.bin_dir):
|
os.makedirs(self.bin_dir, exist_ok=True)
|
|
# match files like model_1.2.3.bin (allow dots, dashes, underscores in model and version)
|
pattern = os.path.join(self.bin_dir, "*.bin")
|
for path in glob.glob(pattern):
|
fname = os.path.basename(path)
|
# filename format: {model}_{version}.bin
|
m = re.match(r"^(.+?)_([0-9][A-Za-z0-9\.\-_]*)\.bin$", fname)
|
if not m:
|
# skip files not conforming to naming rule
|
continue
|
model = m.group(1)
|
version = m.group(2)
|
files_by_model.setdefault(model, []).append((version, fname))
|
|
# sort versions for each model descending
|
for model, items in files_by_model.items():
|
items.sort(key=lambda it: _parse_version(it[0]), reverse=True)
|
|
self._bin_cache["files_by_model"] = files_by_model
|
self._bin_cache["updated_at"] = now
|
self.logger.bind(tag=TAG).info(
|
f"Firmware cache refreshed: {len(files_by_model)} models"
|
)
|
except Exception as e:
|
self.logger.bind(tag=TAG).error(f"刷新固件缓存失败: {e}")
|
# keep previous cache if any
|
|
def generate_password_signature(self, content: str, secret_key: str) -> str:
|
"""生成MQTT密码签名
|
|
Args:
|
content: 签名内容 (clientId + '|' + username)
|
secret_key: 密钥
|
|
Returns:
|
str: Base64编码的HMAC-SHA256签名
|
"""
|
try:
|
hmac_obj = hmac.new(
|
secret_key.encode("utf-8"), content.encode("utf-8"), hashlib.sha256
|
)
|
signature = hmac_obj.digest()
|
return base64.b64encode(signature).decode("utf-8")
|
except Exception as e:
|
self.logger.bind(tag=TAG).error(f"生成MQTT密码签名失败: {e}")
|
return ""
|
|
def _get_websocket_url(self, local_ip: str, port: int) -> str:
|
"""获取websocket地址
|
|
Args:
|
local_ip: 本地IP地址
|
port: 端口号
|
|
Returns:
|
str: websocket地址
|
"""
|
server_config = self.config["server"]
|
websocket_config = server_config.get("websocket", "")
|
|
if "你的" not in websocket_config:
|
return websocket_config
|
else:
|
return f"ws://{local_ip}:{port}/xiaozhi/v1/"
|
|
async def handle_post(self, request):
|
"""处理 OTA POST 请求
|
|
This handler will:
|
- read device id/client id (as before)
|
- attempt to determine device model and current firmware version (prefer headers, fallback to body)
|
- check data/bin for newer firmware for that model
|
- if found a newer firmware, set firmware.url to the download endpoint
|
"""
|
try:
|
data = await request.text()
|
self.logger.bind(tag=TAG).debug(f"OTA请求方法: {request.method}")
|
self.logger.bind(tag=TAG).debug(f"OTA请求头: {request.headers}")
|
self.logger.bind(tag=TAG).debug(f"OTA请求数据: {data}")
|
|
device_id = request.headers.get("device-id", "")
|
if device_id:
|
self.logger.bind(tag=TAG).info(f"OTA请求设备ID: {device_id}")
|
else:
|
raise Exception("OTA请求设备ID为空")
|
|
client_id = request.headers.get("client-id", "")
|
if client_id:
|
self.logger.bind(tag=TAG).info(f"OTA请求ClientID: {client_id}")
|
else:
|
raise Exception("OTA请求ClientID为空")
|
|
data_json = {}
|
try:
|
data_json = json.loads(data) if data else {}
|
except Exception:
|
data_json = {}
|
|
server_config = self.config["server"]
|
# Distinguish ports:
|
# - websocket_port is used to construct websocket URL (server["port"])
|
# - http_port is used to construct OTA download URLs (server["http_port"])
|
websocket_port = int(server_config.get("port", 8000))
|
http_port = int(server_config.get("http_port", 8003))
|
local_ip = get_local_ip()
|
|
# Determine device model (prefer headers)
|
device_model = ""
|
# header candidates
|
for h in ("device-model", "device_model", "model"):
|
if h in request.headers:
|
device_model = request.headers.get(h, "").strip()
|
break
|
# body fallback
|
if not device_model:
|
try:
|
if "board" in data_json and isinstance(data_json["board"], dict):
|
device_model = data_json["board"].get("type", "")
|
elif "model" in data_json:
|
device_model = data_json.get("model", "")
|
except Exception:
|
device_model = ""
|
if not device_model:
|
device_model = "default"
|
|
# Determine device current version (prefer headers)
|
device_version = ""
|
for h in (
|
"device-version",
|
"device_version",
|
"firmware-version",
|
"app-version",
|
"application-version",
|
):
|
if h in request.headers:
|
device_version = request.headers.get(h, "").strip()
|
break
|
if not device_version:
|
try:
|
device_version = data_json.get("application", {}).get("version", "")
|
except Exception:
|
device_version = ""
|
if not device_version:
|
device_version = "0.0.0"
|
|
return_json = {
|
"server_time": {
|
"timestamp": int(round(time.time() * 1000)),
|
"timezone_offset": server_config.get("timezone_offset", 8) * 60,
|
},
|
"firmware": {
|
"version": device_version,
|
"url": "",
|
},
|
}
|
|
# existing mqtt/websocket logic (unchanged)
|
mqtt_gateway_endpoint = server_config.get("mqtt_gateway")
|
|
if mqtt_gateway_endpoint: # 如果配置了非空字符串
|
# 尝试从请求数据中获取设备型号（已解析 above）
|
try:
|
group_id = f"GID_{device_model}".replace(":", "_").replace(" ", "_")
|
except Exception as e:
|
self.logger.bind(tag=TAG).error(f"获取设备型号失败: {e}")
|
group_id = "GID_default"
|
|
mac_address_safe = device_id.replace(":", "_")
|
mqtt_client_id = f"{group_id}@@@{mac_address_safe}@@@{mac_address_safe}"
|
|
# 构建用户数据
|
user_data = {"ip": "unknown"}
|
try:
|
user_data_json = json.dumps(user_data)
|
username = base64.b64encode(user_data_json.encode("utf-8")).decode(
|
"utf-8"
|
)
|
except Exception as e:
|
self.logger.bind(tag=TAG).error(f"生成用户名失败: {e}")
|
username = ""
|
|
# 生成密码
|
password = ""
|
signature_key = server_config.get("mqtt_signature_key", "")
|
if signature_key:
|
password = self.generate_password_signature(
|
mqtt_client_id + "|" + username, signature_key
|
)
|
if not password:
|
password = "" # 签名失败则留空，由设备决定是否允许无密码
|
else:
|
self.logger.bind(tag=TAG).warning("缺少MQTT签名密钥，密码留空")
|
|
# 构建MQTT配置（直接使用 mqtt_gateway 字符串）
|
return_json["mqtt"] = {
|
"endpoint": mqtt_gateway_endpoint,
|
"client_id": mqtt_client_id,
|
"username": username,
|
"password": password,
|
"publish_topic": "device-server",
|
"subscribe_topic": f"devices/p2p/{mac_address_safe}",
|
}
|
self.logger.bind(tag=TAG).info(f"为设备 {device_id} 下发MQTT网关配置")
|
|
else: # 未配置 mqtt_gateway，下发 WebSocket
|
# 如果开启了认证，则进行认证校验
|
token = ""
|
if self.auth_enable:
|
if self.allowed_devices:
|
if device_id not in self.allowed_devices:
|
token = self.auth.generate_token(client_id, device_id)
|
else:
|
token = self.auth.generate_token(client_id, device_id)
|
# NOTE: use websocket_port here
|
return_json["websocket"] = {
|
"url": self._get_websocket_url(local_ip, websocket_port),
|
"token": token,
|
}
|
self.logger.bind(tag=TAG).info(
|
f"未配置MQTT网关，为设备 {device_id} 下发WebSocket配置"
|
)
|
|
# Now check firmware files for updates
|
try:
|
self._refresh_bin_cache_if_needed()
|
files_by_model = self._bin_cache.get("files_by_model", {})
|
candidates = files_by_model.get(device_model, [])
|
|
self.logger.bind(tag=TAG).info(
|
f"查找型号 {device_model} 的固件，找到 {len(candidates)} 个候选"
|
)
|
|
chosen_url = ""
|
chosen_version = device_version
|
|
# candidates are sorted descending by version
|
for ver, fname in candidates:
|
if _is_higher_version(ver, device_version):
|
# build download url (only allow download via our download endpoint)
|
chosen_version = ver
|
# Use get_vision_url to get the base URL and replace the path
|
vision_url = get_vision_url(self.config)
|
# Replace the path from "/mcp/vision/explain" to "/xiaozhi/ota/download/{fname}"
|
chosen_url = vision_url.replace(
|
"/mcp/vision/explain", f"/xiaozhi/ota/download/{fname}"
|
)
|
break
|
|
if chosen_url:
|
return_json["firmware"]["version"] = chosen_version
|
return_json["firmware"]["url"] = chosen_url
|
self.logger.bind(tag=TAG).info(
|
f"为设备 {device_id} 下发固件 {chosen_version} [如果地址前缀有误，请检查配置文件中的server.vision_explain]-> {chosen_url} "
|
)
|
else:
|
self.logger.bind(tag=TAG).info(
|
f"设备 {device_id} 固件已是最新: {device_version}"
|
)
|
|
except Exception as e:
|
self.logger.bind(tag=TAG).error(f"检查固件版本时出错: {e}")
|
|
response = web.Response(
|
text=json.dumps(return_json, separators=(",", ":")),
|
content_type="application/json",
|
)
|
except Exception as e:
|
self.logger.bind(tag=TAG).error(f"OTA POST处理异常: {e}")
|
return_json = {"success": False, "message": "request error."}
|
response = web.Response(
|
text=json.dumps(return_json, separators=(",", ":")),
|
content_type="application/json",
|
)
|
finally:
|
self._add_cors_headers(response)
|
return response
|
|
async def handle_get(self, request):
|
"""处理 OTA GET 请求"""
|
try:
|
server_config = self.config["server"]
|
local_ip = get_local_ip()
|
# use websocket port for websocket URL
|
websocket_port = int(server_config.get("port", 8000))
|
websocket_url = self._get_websocket_url(local_ip, websocket_port)
|
message = f"OTA接口运行正常，向设备发送的websocket地址是：{websocket_url}"
|
response = web.Response(text=message, content_type="text/plain")
|
except Exception as e:
|
self.logger.bind(tag=TAG).error(f"OTA GET请求异常: {e}")
|
response = web.Response(text="OTA接口异常", content_type="text/plain")
|
finally:
|
self._add_cors_headers(response)
|
return response
|
|
async def handle_download(self, request):
|
"""
|
下载固件接口
|
URL: /xiaozhi/ota/download/{filename}
|
- 只允许下载 data/bin 目录下的 .bin 文件
|
- filename 必须是 basename 且匹配安全的模式
|
"""
|
try:
|
fname = request.match_info.get("filename", "")
|
if not fname:
|
raise web.HTTPBadRequest(text="filename required")
|
|
# sanitize
|
fname = _safe_basename(fname)
|
# pattern: allow letters, numbers, dot, underscore, dash
|
if not re.match(r"^[A-Za-z0-9\.\-_]+\.bin$", fname):
|
raise web.HTTPBadRequest(text="invalid filename")
|
|
file_path = os.path.join(self.bin_dir, fname)
|
# ensure realpath is under bin_dir
|
file_real = os.path.realpath(file_path)
|
bin_dir_real = os.path.realpath(self.bin_dir)
|
if (
|
not file_real.startswith(bin_dir_real + os.sep)
|
and file_real != bin_dir_real
|
):
|
raise web.HTTPForbidden(text="forbidden")
|
|
if not os.path.isfile(file_real):
|
raise web.HTTPNotFound(text="file not found")
|
|
# use FileResponse to stream file
|
resp = web.FileResponse(path=file_real)
|
except web.HTTPError as e:
|
resp = e
|
except Exception as e:
|
self.logger.bind(tag=TAG).error(f"固件下载异常: {e}")
|
resp = web.Response(text="download error", status=500)
|
finally:
|
try:
|
self._add_cors_headers(resp)
|
except Exception:
|
pass
|
return resp
|
