chore(repo): initialize team collaboration repository

flashops/services/control-plane/flashops_control/safety/__init__.py

@@ -0,0 +1,5 @@
+"""危险动作的安全门禁。"""
+
+from .gates import PreflightRequest, PreflightResult, run_preflight
+
+__all__ = ["PreflightRequest", "PreflightResult", "run_preflight"]

flashops/services/control-plane/flashops_control/safety/gates.py

@@ -0,0 +1,90 @@
+"""安全默认拒绝的预检规则。"""
+from __future__ import annotations
+
+from dataclasses import asdict, dataclass
+from pathlib import PurePath
+from typing import Dict, List, Sequence
+
+
+@dataclass(frozen=True)
+class Check:
+    key: str
+    name: str
+    passed: bool
+    note: str
+
+
+@dataclass(frozen=True)
+class PreflightRequest:
+    expected_serial: str
+    discovered_serial: str
+    allow_destructive: bool
+    device_path: str
+    system_device_paths: Sequence[str]
+    firmware_model_allowed: bool
+    host_online: bool
+    oob_online: bool
+
+
+@dataclass(frozen=True)
+class PreflightResult:
+    passed: bool
+    checks: List[Check]
+
+    def as_dict(self) -> Dict[str, object]:
+        return {
+            "passed": self.passed,
+            "checks": [asdict(check) for check in self.checks],
+        }
+
+
+def _same_device(left: str, right: str) -> bool:
+    left_name = PurePath(left).name
+    right_name = PurePath(right).name
+    if left_name == right_name:
+        return True
+    # /dev/nvme0n1p2 属于 /dev/nvme0n1；拒绝把其任一分区当作测试盘。
+    return left_name.startswith(right_name + "p") or right_name.startswith(left_name + "p")
+
+
+def run_preflight(request: PreflightRequest) -> PreflightResult:
+    serial_ok = bool(request.expected_serial) and (
+        request.expected_serial == request.discovered_serial
+    )
+    system_disk_safe = bool(request.device_path) and not any(
+        _same_device(request.device_path, path) for path in request.system_device_paths
+    )
+    checks = [
+        Check("serial", "DUT 序列号白名单绑定", serial_ok, request.discovered_serial or "未发现"),
+        Check(
+            "destructive",
+            "破坏性写入授权",
+            request.allow_destructive,
+            "已审批" if request.allow_destructive else "DUT 未授权破坏性写入",
+        ),
+        Check(
+            "system_disk",
+            "系统盘 / 分区保护",
+            system_disk_safe,
+            "启动盘已排除" if system_disk_safe else "目标命中系统盘或其分区",
+        ),
+        Check(
+            "firmware",
+            "固件适配型号",
+            request.firmware_model_allowed,
+            "型号匹配" if request.firmware_model_allowed else "固件不适用于该型号",
+        ),
+        Check(
+            "host",
+            "测试主机与 Agent",
+            request.host_online,
+            "在线" if request.host_online else "主机离线",
+        ),
+        Check(
+            "oob",
+            "带外控制器",
+            request.oob_online,
+            "Power / Reset 可用" if request.oob_online else "带外通道离线",
+        ),
+    ]
+    return PreflightResult(all(check.passed for check in checks), checks)