生产环境网络安全加固落地复盘：架构改造、权限收敛与踩坑记录

# safety_controller.py
import time
import logging

class SafeServoController:
    def __init__(self, raw_serial_driver):
        self.driver = raw_serial_driver
        # 硬编码物理安全界限，覆盖掉网络下发的任何极端值
        self.LIMITS = {
            1: (45.0, 135.0),  # 头部俯仰角度限位
            2: (10.0, 170.0),  # 身体旋转角度限位
        }
        self.last_cmd_time = 0.0
        self.MIN_INTERVAL = 0.02  # 限制下发频率，防止高频指令打崩串口或使舵机震颤

    def move_servo(self, servo_id: int, target_angle: float):
        current_time = time.time()
        if current_time - self.last_cmd_time < self.MIN_INTERVAL:
            logging.warning("指令频率过高，丢弃该帧")
            return False

        if servo_id not in self.LIMITS:
            logging.error(f"未授权的舵机 ID: {servo_id}")
            return False

        min_ang, max_ang = self.LIMITS[servo_id]
        # 强制截断到安全范围
        clamped_angle = max(min_ang, min(target_angle, max_ang))
        
        self.last_cmd_time = current_time
        return self.driver.write_angle(servo_id, clamped_angle)

# auth_gateway.py
import hmac
import hashlib
import json
import time
from fastapi import FastAPI, WebSocket, WebSocketDisconnect

app = FastAPI()
SHARED_SECRET = b"your_offline_provisioned_secret_key"  # 部署时烧录的预共享密钥

def verify_signature(payload: dict, client_signature: str) -> bool:
    # 检查时间戳，防止重放攻击 (窗口设为 500ms)
    ts = payload.get("timestamp", 0)
    now_ms = int(time.time() * 1000)
    if abs(now_ms - ts) > 500:
        return False

    # 重构签名串
    message = json.dumps(payload, sort_keys=True).encode("utf-8")
    expected_sig = hmac.new(SHARED_SECRET, message, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected_sig, client_signature)

@app.websocket("/ws/control")
async def control_endpoint(websocket: WebSocket):
    await websocket.accept()
    try:
        while True:
            raw_data = await websocket.receive_text()
            packet = json.loads(raw_data)
            
            payload = packet.get("payload")
            signature = packet.get("sig")

            if not payload or not signature or not verify_signature(payload, signature):
                await websocket.send_json({"error": "Unauthorized or replay attack"})
                await websocket.close()
                break

            # 鉴权通过，执行运动解算并下发
            servo_id = payload.get("id")
            angle = payload.get("angle")
            # safe_controller.move_servo(servo_id, angle)
            
    except WebSocketDisconnect:
        pass