本章深入探讨 ROS2 的安全性框架和诊断系统，这是构建可靠、安全的机器人系统的关键基础。我们将从 SROS2（Secure ROS2）安全框架开始，学习如何通过访问控制、加密和认证机制保护机器人系统。随后深入诊断系统的设计与实现，探讨如何监控系统健康状态、检测故障并执行恢复策略。通过医疗机器人 ISO 13485 合规案例，展示如何在实际产业中实施安全和诊断机制。最后讨论功能安全认证和 SIL（Safety Integrity Level）级别，为安全关键型应用提供指导。
在分布式机器人系统中，主要面临以下安全威胁：
SROS2 建立在 DDS Security 规范之上，提供端到端的安全保护：
┌─────────────────────────────────────────────┐
│ Application Layer │
├─────────────────────────────────────────────┤
│ ROS2 Middleware │
├─────────────────────────────────────────────┤
│ DDS Security Plugins │
│ ┌──────────┬──────────┬──────────────┐ │
│ │Auth Plugin│Access │Crypto Plugin │ │
│ │ │Control │ │ │
│ └──────────┴──────────┴──────────────┘ │
├─────────────────────────────────────────────┤
│ DDS Core │
└─────────────────────────────────────────────┘
SROS2 使用 XML 格式的安全策略文件定义访问权限：
<?xml version="1.0" encoding="UTF-8"?>
<dds xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
<permissions>
<grant name="robot_controller_permissions">
<subject_name>CN=robot_controller</subject_name>
<validity>
<not_before>2024-01-01T00:00:00</not_before>
<not_after>2025-01-01T00:00:00</not_after>
</validity>
<allow_rule>
<domains>
<id>0</id>
</domains>
<publish>
<topics>
<topic>/cmd_vel</topic>
<topic>/robot_status</topic>
</topics>
</publish>
<subscribe>
<topics>
<topic>/sensors/*</topic>
</topics>
</subscribe>
</allow_rule>
</grant>
</permissions>
</dds>
SROS2 采用 PKI（Public Key Infrastructure）体系：
密钥生成和分发流程：
# 初始化安全目录
ros2 security create_keystore demo_keystore
# 生成根证书
ros2 security create_key demo_keystore /robot_controller
# 创建权限文件
ros2 security create_permission demo_keystore /robot_controller permissions.xml
# 签发证书
ros2 security create_enclave demo_keystore /robot_controller
SROS2 支持多层次的访问控制：
访问控制矩阵示例：
| 节点 | 发布权限 | 订阅权限 | 服务调用权限 |
|---|---|---|---|
| navigation | /cmd_vel, /path | /scan, /map | /get_plan |
| perception | /scan, /pointcloud | /camera/* | - |
| controller | /joint_states | /cmd_vel | /emergency_stop |
SROS2 支持多种加密算法，根据性能和安全需求选择：
对称加密（数据加密）:
- AES-128-GCM: 标准选择，平衡性能与安全
- AES-256-GCM: 高安全级别，适用于敏感数据
- ChaCha20-Poly1305: ARM 设备优化
非对称加密（密钥交换）:
- RSA-2048: 广泛支持，兼容性好
- ECDSA-P256: 更小的密钥尺寸，更高效
- EdDSA: 现代算法，抗侧信道攻击
哈希算法（完整性验证）:
- SHA-256: 标准选择
- SHA-384/512: 更高安全级别
- BLAKE2: 高性能替代方案
数据传输安全机制：
性能影响分析：
基准测试（1000字节消息，1kHz 发布频率）：
- 无安全保护: 延迟 0.5ms, CPU 5%
- 仅认证: 延迟 0.7ms, CPU 8%
- 认证+加密: 延迟 1.2ms, CPU 15%
- 完整安全套件: 延迟 1.5ms, CPU 20%
节点间安全通信的建立过程：
Node A Node B
│ │
├──────[1. Discovery]──────────>│
│ (广播节点信息和证书) │
│ │
│<─────[2. Authentication]──────┤
│ (相互验证身份证书) │
│ │
├──────[3. Key Exchange]───────>│
│ (协商会话密钥) │
│ │
│<─────[4. Access Check]────────┤
│ (验证访问权限) │
│ │
├══════[5. Secure Channel]══════┤
│ (建立加密通道) │
│ │
ROS2 诊断系统使用标准化的消息格式：
# diagnostic_msgs/msg/DiagnosticStatus.msg
byte OK=0
byte WARN=1
byte ERROR=2
byte STALE=3
byte level # 诊断级别
string name # 组件名称
string message # 人类可读的状态描述
string hardware_id # 硬件标识符
KeyValue[] values # 键值对形式的详细信息
诊断聚合器（Diagnostic Aggregator）收集和组织诊断信息：
┌──────────────────┐
│ Diagnostic │
│ Aggregator │
└────────┬─────────┘
│
┌────────────────┼────────────────┐
│ │ │
┌──────┴──────┐ ┌─────┴──────┐ ┌─────┴──────┐
│ Hardware │ │ Software │ │ Network │
│ Monitor │ │ Monitor │ │ Monitor │
└──────┬──────┘ └─────┬──────┘ └─────┬──────┘
│ │ │
┌──────┴──────┐ ┌─────┴──────┐ ┌─────┴──────┐
│ CPU Temp │ │ Node │ │ Latency │
│ Memory │ │ Frequency │ │ Packet Loss│
│ Disk Space │ │ Queue Size │ │ Bandwidth │
└─────────────┘ └────────────┘ └────────────┘
节点级诊断更新器示例：
import rclpy
from rclpy.node import Node
from diagnostic_updater import Updater, FunctionDiagnosticTask
from diagnostic_msgs.msg import DiagnosticStatus
class SensorNode(Node):
def __init__(self):
super().__init__('sensor_node')
# 初始化诊断更新器
self.updater = Updater(self)
self.updater.setHardwareID("sensor_001")
# 添加诊断任务
self.updater.add("Sensor Connection", self.check_connection)
self.updater.add("Data Rate", self.check_data_rate)
self.updater.add("Data Quality", self.check_data_quality)
# 诊断更新定时器
self.timer = self.create_timer(1.0, self.updater.update)
def check_connection(self, stat):
if self.sensor_connected:
stat.summary(DiagnosticStatus.OK, "Connected")
stat.add("Port", "/dev/ttyUSB0")
stat.add("Baud Rate", "115200")
else:
stat.summary(DiagnosticStatus.ERROR, "Disconnected")
def check_data_rate(self, stat):
rate = self.calculate_data_rate()
if rate > 90:
stat.summary(DiagnosticStatus.OK, f"{rate:.1f} Hz")
elif rate > 50:
stat.summary(DiagnosticStatus.WARN, f"{rate:.1f} Hz (Low)")
else:
stat.summary(DiagnosticStatus.ERROR, f"{rate:.1f} Hz (Critical)")
stat.add("Expected Rate", "100 Hz")
stat.add("Actual Rate", f"{rate:.1f} Hz")
配置诊断分析器以组织和解释诊断信息：
# diagnostic_aggregator.yaml
analyzers:
sensors:
type: diagnostic_aggregator/AnalyzerGroup
path: Sensors
analyzers:
lidar:
type: diagnostic_aggregator/GenericAnalyzer
path: LiDAR
contains: ['lidar']
timeout: 5.0
cameras:
type: diagnostic_aggregator/GenericAnalyzer
path: Cameras
contains: ['camera']
remove_prefix: camera_
motors:
type: diagnostic_aggregator/GenericAnalyzer
path: Motors
contains: ['motor', 'joint']
expected: ['motor_left', 'motor_right']
system:
type: diagnostic_aggregator/AnalyzerGroup
path: System
analyzers:
cpu:
type: diagnostic_aggregator/GenericAnalyzer
path: CPU
contains: ['cpu']
memory:
type: diagnostic_aggregator/GenericAnalyzer
path: Memory
contains: ['memory', 'ram']
关键系统健康指标：
监控数据流水线：
数据采集 -> 预处理 -> 异常检测 -> 告警生成 -> 响应执行
│ │ │ │ │
Metrics Filter Threshold Priority Action
Collector & Aggregate Check Assignment Trigger
实现多层次故障检测：
class FaultDetector:
def __init__(self):
self.thresholds = {
'cpu_usage': {'warn': 70, 'error': 90},
'memory_usage': {'warn': 80, 'error': 95},
'message_delay': {'warn': 50, 'error': 100}, # ms
'packet_loss': {'warn': 1, 'error': 5} # percentage
}
# 滑动窗口用于趋势分析
self.history_window = 60 # seconds
self.metric_history = defaultdict(deque)
def detect_anomalies(self, metrics):
anomalies = []
# 阈值检测
for metric, value in metrics.items():
if metric in self.thresholds:
level = self.check_threshold(metric, value)
if level:
anomalies.append((metric, level, value))
# 趋势分析
trends = self.analyze_trends(metrics)
anomalies.extend(trends)
# 相关性分析
correlations = self.check_correlations(metrics)
anomalies.extend(correlations)
return anomalies
def analyze_trends(self, metrics):
"""检测异常趋势"""
trends = []
for metric, value in metrics.items():
self.metric_history[metric].append(value)
if len(self.metric_history[metric]) > 10:
# 计算变化率
recent = list(self.metric_history[metric])[-10:]
slope = np.polyfit(range(10), recent, 1)[0]
if abs(slope) > self.get_trend_threshold(metric):
trends.append((metric, 'trend', slope))
return trends
分级故障恢复机制：
class RecoveryManager:
def __init__(self):
self.recovery_strategies = {
'node_crash': [
self.restart_node,
self.restart_with_safe_config,
self.switch_to_backup_node
],
'sensor_failure': [
self.reset_sensor,
self.use_sensor_fusion,
self.enable_degraded_mode
],
'network_issue': [
self.reset_network_interface,
self.switch_to_backup_network,
self.enable_offline_mode
]
}
def execute_recovery(self, fault_type, context):
"""执行分级恢复策略"""
strategies = self.recovery_strategies.get(fault_type, [])
for strategy in strategies:
self.get_logger().info(f"Attempting recovery: {strategy.__name__}")
try:
success = strategy(context)
if success:
self.get_logger().info(f"Recovery successful: {strategy.__name__}")
return True
except Exception as e:
self.get_logger().error(f"Recovery failed: {e}")
return False
实现优雅降级机制：
正常模式 (100% 功能)
│
├─> 性能降级模式 (80% 功能)
│ - 降低传感器采样率
│ - 简化路径规划算法
│ - 减少并发任务
│
├─> 安全模式 (50% 功能)
│ - 仅保留核心功能
│ - 限制运动速度
│ - 增加安全边界
│
└─> 紧急停止模式 (0% 功能)
- 立即停止所有运动
- 保持通信链路
- 等待人工干预
某医疗科技公司开发手术辅助机器人系统，需要满足 ISO 13485 医疗器械质量管理体系和 IEC 62304 医疗器械软件生命周期过程标准。系统包括：
根据 ISO 14971 风险管理标准，识别关键安全需求：
分层安全架构设计：
┌─────────────────────────────────────────────┐
│ 应用层（手术规划软件） │
├─────────────────────────────────────────────┤
│ 安全中间件层（SROS2） │
│ - 认证与授权 │
│ - 数据加密 │
│ - 审计日志 │
├─────────────────────────────────────────────┤
│ 实时控制层（RT-PREEMPT） │
│ - 确定性调度 │
│ - 故障检测 │
│ - 安全状态机 │
├─────────────────────────────────────────────┤
│ 硬件安全层（Safety PLC） │
│ - 紧急停止 │
│ - 位置限制 │
│ - 力矩限制 │
└─────────────────────────────────────────────┘
多级诊断系统实现：
class MedicalRobotDiagnostics:
def __init__(self):
# IEC 62304 软件安全分类
self.safety_class = "Class C" # 可能导致死亡或严重伤害
# 诊断检查项
self.diagnostic_checks = {
'critical': [ # 关键检查，失败立即停止
self.check_emergency_stop,
self.check_force_limits,
self.check_position_accuracy
],
'major': [ # 主要检查，失败进入安全模式
self.check_sensor_redundancy,
self.check_communication_integrity,
self.check_power_supply
],
'minor': [ # 次要检查，记录但继续运行
self.check_temperature,
self.check_network_latency,
self.check_log_storage
]
}
def perform_startup_diagnostics(self):
"""启动时完整诊断检查"""
results = {
'passed': True,
'critical_failures': [],
'warnings': []
}
# 执行所有级别的检查
for level, checks in self.diagnostic_checks.items():
for check in checks:
status = check()
if not status.ok:
if level == 'critical':
results['passed'] = False
results['critical_failures'].append(status)
else:
results['warnings'].append(status)
# 生成诊断报告
self.generate_diagnostic_report(results)
return results
符合 21 CFR Part 11 的审计追踪：
class AuditLogger:
def __init__(self):
self.log_encryption_key = self.load_encryption_key()
self.log_database = "audit_logs.db"
def log_event(self, event_type, user_id, action, details):
"""记录审计事件"""
event = {
'timestamp': datetime.now(timezone.utc).isoformat(),
'event_id': uuid.uuid4().hex,
'event_type': event_type,
'user_id': user_id,
'action': action,
'details': details,
'system_state': self.capture_system_state()
}
# 计算事件哈希（防篡改）
event['hash'] = self.calculate_hash(event)
# 加密敏感信息
encrypted_event = self.encrypt_event(event)
# 存储到防篡改数据库
self.store_event(encrypted_event)
# 实时同步到备份服务器
self.replicate_to_backup(encrypted_event)
系统验证测试结果：
| 测试项目 | 要求 | 实测值 | 结果 |
|---|---|---|---|
| 位置精度 | < 0.5mm | 0.3mm | ✓ |
| 力控制精度 | < 10N | 8.5N | ✓ |
| 紧急停止时间 | < 100ms | 85ms | ✓ |
| 加密延迟 | < 5ms | 3.2ms | ✓ |
| 诊断检测时间 | < 1s | 0.8s | ✓ |
| 系统启动时间 | < 60s | 45s | ✓ |
| MTBF | > 10000h | 12500h | ✓ |
机器人系统相关的功能安全标准：
IEC 61508 (基础标准)
│
├─> ISO 13849 (机械安全)
│ └─> Performance Level (PL)
│
├─> IEC 62061 (机械电气安全)
│ └─> SIL for machinery
│
├─> ISO 26262 (汽车功能安全)
│ └─> ASIL levels
│
└─> IEC 62304 (医疗软件)
└─> Software Safety Classes
安全完整性等级（Safety Integrity Level）评估：
def calculate_sil_level(severity, frequency, probability, avoidance):
"""
根据风险参数计算 SIL 等级
severity: 严重程度 (S1-S4)
frequency: 暴露频率 (F1-F2)
probability: 危险概率 (P1-P2)
avoidance: 避免可能性 (A1-A2)
"""
risk_matrix = {
('S1', 'F1', 'P1', 'A1'): 'SIL0',
('S1', 'F1', 'P1', 'A2'): 'SIL0',
('S1', 'F1', 'P2', 'A1'): 'SIL1',
('S1', 'F1', 'P2', 'A2'): 'SIL1',
('S2', 'F1', 'P1', 'A1'): 'SIL1',
('S2', 'F1', 'P1', 'A2'): 'SIL1',
('S2', 'F1', 'P2', 'A1'): 'SIL2',
('S2', 'F1', 'P2', 'A2'): 'SIL2',
('S3', 'F1', 'P1', 'A1'): 'SIL2',
('S3', 'F1', 'P1', 'A2'): 'SIL2',
('S3', 'F1', 'P2', 'A1'): 'SIL3',
('S3', 'F1', 'P2', 'A2'): 'SIL3',
('S4', 'F2', 'P2', 'A2'): 'SIL4'
}
return risk_matrix.get((severity, frequency, probability, avoidance), 'SIL3')
SIL2 级别的安全功能实现示例：
class SafetyFunction:
def __init__(self, sil_level=2):
self.sil_level = sil_level
# 根据 SIL 级别配置冗余
if sil_level >= 2:
self.redundancy = 2 # 双通道
else:
self.redundancy = 1 # 单通道
# 诊断覆盖率要求
self.diagnostic_coverage = {
1: 60, # SIL1: 60% 最小诊断覆盖率
2: 90, # SIL2: 90%
3: 99, # SIL3: 99%
4: 99 # SIL4: 99%
}[sil_level]
def execute_safety_function(self, inputs):
"""执行安全功能，带冗余和诊断"""
results = []
# 冗余执行
for channel in range(self.redundancy):
result = self.process_channel(inputs, channel)
results.append(result)
# 结果比较（2oo2 或 1oo2 投票）
if self.redundancy > 1:
if not self.compare_results(results):
return self.trigger_safe_state()
# 诊断检查
if not self.perform_diagnostics():
return self.trigger_safe_state()
return results[0]
计算危险故障概率（PFH - Probability of Dangerous Failure per Hour）：
def calculate_pfh(lambda_d, dc, beta, t_proof):
"""
计算每小时危险故障概率
lambda_d: 危险故障率
dc: 诊断覆盖率 (0-1)
beta: 共因故障比例 (0-1)
t_proof: 验证测试间隔（小时）
"""
# 单通道 PFH
pfh_single = lambda_d * (1 - dc) * t_proof / 2
# 考虑共因故障的双通道 PFH
pfh_dual = (beta * lambda_d + (1 - beta) * lambda_d**2 * t_proof) * (1 - dc)
# SIL 等级对应的 PFH 限值
sil_limits = {
1: 1e-5, # SIL1: 10^-5 到 10^-6
2: 1e-6, # SIL2: 10^-6 到 10^-7
3: 1e-7, # SIL3: 10^-7 到 10^-8
4: 1e-8 # SIL4: 10^-8 到 10^-9
}
return pfh_dual
符合 IEC 61508-3 的软件开发流程：
需求分析 ────────────────────> 验收测试
│ ↑
↓ │
系统设计 ──────────────> 系统集成测试
│ ↑
↓ │
模块设计 ────────────> 模块测试
│ ↑
↓ │
编码实现 ──────────> 单元测试
每个阶段的安全活动：
- 需求：HAZOP, FMEA
- 设计：FTA, 安全分析
- 实现：编码标准（MISRA）
- 测试：覆盖率分析，故障注入
- 验证：独立 V&V
SIL 认证所需文档清单：
认证机构评估要点：
本章系统介绍了 ROS2 的安全性和诊断系统，涵盖了从基础安全框架到高级功能安全认证的完整知识体系：
诊断覆盖率（DC）： \(DC = \frac{\lambda_{dd}}{\lambda_{d}} = \frac{\text{检测到的危险故障率}}{\text{总危险故障率}}\)
平均危险故障时间（MTTF_d）： \(MTTF_d = \frac{1}{\lambda_d \cdot (1 - DC)}\)
系统可用性（A）： \(A = \frac{MTBF}{MTBF + MTTR} = \frac{\text{平均故障间隔时间}}{\text{平均故障间隔时间} + \text{平均修复时间}}\)
冗余系统可靠性（并联）： \(R_{system} = 1 - \prod_{i=1}^{n}(1 - R_i)\)
共因故障因子（β）影响： \(\lambda_{sys} = \beta \cdot \lambda + (1-\beta) \cdot \frac{\lambda^2}{2 \cdot \mu}\)
练习 16.1：设计一个 SROS2 权限配置文件，实现以下访问控制要求：
练习 16.2：实现一个诊断更新器，监控机器人电池状态，包括电压、电流、温度和剩余容量。当电压低于 20% 时报警，低于 10% 时触发紧急模式。
练习 16.3：计算一个安全功能的 SIL 等级，给定条件：可能造成永久性伤害（S3），频繁暴露（F2），危险发生概率高（P2），几乎不可能避免（A2）。
练习 16.4：设计一个分布式故障检测系统，能够检测以下异常：
练习 16.5：为一个协作机器人设计符合 ISO 10218 标准的安全系统，包括：
练习 16.6：实现一个符合 21 CFR Part 11 的审计日志系统，要求：
练习 16.7：分析一个 SIL2 安全功能的硬件架构，给定组件故障率：
练习 16.8：设计一个机器人系统的优雅降级策略，系统包含激光雷达、相机、IMU、编码器四个传感器。定义在不同传感器失效组合下的降级模式和功能限制。