From b69c47975ae93651563bd6dd5a1c30007c804eaf Mon Sep 17 00:00:00 2001 From: shunli Date: Wed, 21 Jan 2026 09:54:29 +0800 Subject: [PATCH] feat: phase 5 industrial grade hardening (ABI safety, security whitelisting, V2 metadata) --- main.py | 31 ++++++++++++++++++++++++++++--- src/bridges.py | 35 ++++++++++++++++++++++++++++++++--- src/env_detect.py | 38 ++++++++++++++++++++++++++++++++++++++ src/models.py | 12 +++++++++--- 4 files changed, 107 insertions(+), 9 deletions(-) diff --git a/main.py b/main.py index 56041c7..5ecb414 100644 --- a/main.py +++ b/main.py @@ -139,6 +139,17 @@ def show_home(): """, unsafe_allow_html=True) + # 环境诊断与建议 (Phase 5) + from src.env_detect import get_actionable_suggestions + suggestions = get_actionable_suggestions() + if suggestions: + st.error("⚠️ 检测到环境瓶颈") + for sug in suggestions: + with st.expander(f"🛠️ {sug['target']} - {sug['reason']}"): + st.code(sug['fix'], language="bash") + else: + st.success("✨ 系统环境全量兼容 (v2.0.4)") + col1, col2 = st.columns(2) with col1: st.subheader("🚀 核心能力") @@ -355,7 +366,14 @@ def show_scheduling(): if uploaded_file and st.session_state.pipeline: data = np.load(uploaded_file) - wsd = WaterSoundData(raw_data=data, sample_rate=float(sample_rate)) + wsd = WaterSoundData( + raw_data=data, + sample_rate=float(sample_rate), + sample_rate_unit="Hz", + channel_layout="time_first", + coordinate_system="WGS84", + collect_time=datetime.now().strftime("%Y-%m-%d %H:%M:%S") + ) # 自动映射 (Phase 2 简化版：直接使用元数据中的映射) scheduler = CoreScheduler() @@ -410,7 +428,7 @@ def show_scheduling(): in_map, out_map, preprocessing=preprocessing, - requirements=algo.get("inputs") or {}, + requirements=algo.get("inputs") or {}, # 包含 argtypes/restype 如果在 JSON 中定义 output_specs=algo.get("outputs") or {}, ) elif algo_type == "matlab": @@ -478,7 +496,14 @@ def show_sea_trial(): if uploaded_file: data = np.load(uploaded_file) - wsd = WaterSoundData(raw_data=data, sample_rate=float(sample_rate)) + wsd = WaterSoundData( + raw_data=data, + sample_rate=float(sample_rate), + sample_rate_unit="Hz", + channel_layout="time_first", + coordinate_system="WGS84", + collect_time=datetime.now().strftime("%Y-%m-%d %H:%M:%S") + ) # 预设模板执行 if st.button("开始一键巡检"): diff --git a/src/bridges.py b/src/bridges.py index 5291401..954354f 100644 --- a/src/bridges.py +++ b/src/bridges.py @@ -121,15 +121,33 @@ class CppBridge(AlgorithmBridge): self._lib = None self._func = None + def _validate_path(self, path: str): + """Security Whitelisting: Ensure path is within the project's 'src' or 'data' directory""" + abs_path = os.path.abspath(path) + base_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) # project root + if not abs_path.startswith(os.path.join(base_dir, "src")) and \ + not abs_path.startswith(os.path.join(base_dir, "data")): + raise PermissionError(f"Security Block: Path {path} is outside allowed directories (src/data).") + def _load_lib(self): - """延迟加载动态库""" + """延迟加载动态库并安全配置 ABI""" if self._lib is None: + self._validate_path(self.lib_path) if not os.path.exists(self.lib_path): raise FileNotFoundError(f"C++ library not found: {self.lib_path}") + self._lib = ctypes.CDLL(self.lib_path) self._func = getattr(self._lib, self.function_name) - # 注意：在 C 侧我们需要定义统一签名，例如: - # void func(double* in, int in_len, double* out, int out_len) + + # Phase 5: ABI Hardening + # 如果 requirements 中定义了签名，则应用它 + if "argtypes" in self.requirements: + # 简单映射: 目前支持 "double_ptr", "int" + mapping = {"double_ptr": ctypes.POINTER(ctypes.c_double), "int": ctypes.c_int} + self._func.argtypes = [mapping.get(t, ctypes.c_void_p) for t in self.requirements["argtypes"]] + if "restype" in self.requirements: + mapping = {"int": ctypes.c_int, "void": None} + self._func.restype = mapping.get(self.requirements["restype"], None) def call_algo(self, water_data: WaterSoundData) -> WaterSoundData: try: @@ -192,9 +210,18 @@ class MatlabBridge(AlgorithmBridge): self.algo_type = "matlab" self._engine = None + def _validate_path(self, path: str): + """Security Whitelisting: Ensure MATLAB path is within the project's 'src' or 'data' directory""" + abs_path = os.path.abspath(path) + base_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) # project root + if not abs_path.startswith(os.path.join(base_dir, "src")) and \ + not abs_path.startswith(os.path.join(base_dir, "data")): + raise PermissionError(f"Security Block: MATLAB path {path} is outside allowed directories.") + def _call_via_engine(self, inputs: Dict[str, Any]) -> Any: """方式 1: 使用官方 matlab.engine""" import matlab.engine + self._validate_path(self.m_path) if self._engine is None: self._engine = matlab.engine.start_matlab() self._engine.addpath(self.m_path) @@ -207,6 +234,7 @@ class MatlabBridge(AlgorithmBridge): def _call_via_oct2py(self, inputs: Dict[str, Any]) -> Any: """方式 2: 使用开源 oct2py (Octave)""" from oct2py import octave + self._validate_path(self.m_path) octave.addpath(self.m_path) func = getattr(octave, self.function_name) return func(*inputs.values()) @@ -215,6 +243,7 @@ class MatlabBridge(AlgorithmBridge): """方式 3: 文件兜底 (使用 .mat 文件交换)""" from scipy.io import savemat, loadmat import subprocess + self._validate_path(self.m_path) with tempfile.TemporaryDirectory(prefix="matlab_bridge_") as tmp_dir: in_file = os.path.join(tmp_dir, "tmp_in.mat") diff --git a/src/env_detect.py b/src/env_detect.py index 060f21c..d4e8ebc 100644 --- a/src/env_detect.py +++ b/src/env_detect.py @@ -84,3 +84,41 @@ def get_available_algo_types() -> List[str]: types.append("cpp") return types + + +def get_actionable_suggestions() -> List[Dict[str, str]]: + """针对故障环境提供具体的修复指令建议""" + env = collect_env_info() + suggestions = [] + + # MATLAB/Octave 建议 + if not (env["matlab"]["engine_available"] or env["matlab"]["oct2py_available"] or env["matlab"]["octave_cli"]): + suggestions.append({ + "target": "MATLAB/Octave", + "reason": "未检测到 MATLAB 引擎或 Octave 环境", + "fix": "建议安装 Octave: `sudo apt install octave` 或安装相关 Python 库: `pip install oct2py`" + }) + elif not env["matlab"]["engine_available"] and env["matlab"]["octave_cli"]: + suggestions.append({ + "target": "MATLAB Engine", + "reason": "当前使用 Octave 降级执行，性能可能受限", + "fix": "若需原生性能，请安装 MATLAB 并运行 `cd \"$(matlabroot)/extern/engines/python\" && python setup.py install`" + }) + + # C++ 建议 + if not env["cpp"]["any_available"]: + suggestions.append({ + "target": "C++ Compiler", + "reason": "未检测到 gcc/g++/clang 等编译器", + "fix": "建议安装基础构建工具: `sudo apt install build-essential`" + }) + + # 核心库建议 + if not env["scipy_available"]: + suggestions.append({ + "target": "SciPy", + "reason": "缺少 scipy 库，影响 MATLAB 兜底执行", + "fix": "运行 `pip install scipy`" + }) + + return suggestions diff --git a/src/models.py b/src/models.py index 873ddb6..8307fdb 100644 --- a/src/models.py +++ b/src/models.py @@ -5,12 +5,18 @@ from typing import Dict, Optional, List, Any @dataclass class WaterSoundData: - """唯一数据交换标准 (WaterSoundData)""" + """唯一数据交换标准 (WaterSoundData) v2.0""" + SCHEMA_VERSION: str = field(default="2.0", init=False) + # 1. 基础原始数据 raw_data: Optional[np.ndarray] = None # 原始声呐数据 - sample_rate: float = 0.0 # 采样率 + sample_rate: float = 0.0 # 采样率 (默认单位: Hz) + sample_rate_unit: str = "Hz" # 采样率单位 + channel_layout: str = "time_first" # 通道布局 (time_first: [N, C], channel_first: [C, N]) + coordinate_system: str = "WGS84" # 坐标参考系 collect_time: str = "" # 数据采集时间 - sea_env: Dict[str, Any] = field(default_factory=dict) # 海洋环境参数（声速、温度、盐度等） + data_source_hash: str = "" # 原始数据校验和 (SHA256) + sea_env: Dict[str, Any] = field(default_factory=dict) # 海洋环境参数 # 2. 处理过程数据 clean_data: Optional[np.ndarray] = None # 清洗后数据