From e30cd6d93f8971087b1699b0ef8c66b1549bcb37 Mon Sep 17 00:00:00 2001 From: shunli Date: Wed, 21 Jan 2026 15:47:01 +0800 Subject: [PATCH] feat: phase 6 webbridge + generic workflow (resource slots, security, bellhop/kraken/ram assets) --- data/algorithms.json | 286 +++++++++++++++++++++++++++++++++++++++++ main.py | 24 +++- src/bridges.py | 287 +++++++++++++++++++++++++++++++++++++----- src/resource_slots.py | 63 ++++++++++ src/security.py | 107 ++++++++++++++++ 5 files changed, 734 insertions(+), 33 deletions(-) create mode 100644 src/resource_slots.py create mode 100644 src/security.py diff --git a/data/algorithms.json b/data/algorithms.json index f51be88..cc64743 100644 --- a/data/algorithms.json +++ b/data/algorithms.json @@ -7,6 +7,7 @@ "type": "python", "module": "src.sample_algorithms", "function": "dc_removal_and_bandpass", + "sha256": "062aa2ec0c5f5354d75c6e9eb9e7c6adc344e5bbcc3a3e6b171ae1a08e527283", "inputs": { "raw_data": { "field": "raw_data", @@ -35,6 +36,7 @@ "type": "python", "module": "src.sample_algorithms", "function": "stft_magnitude", + "sha256": "062aa2ec0c5f5354d75c6e9eb9e7c6adc344e5bbcc3a3e6b171ae1a08e527283", "inputs": { "clean_data": { "field": "clean_data", @@ -58,6 +60,7 @@ "type": "python", "module": "src.sample_algorithms", "function": "snr_estimation", + "sha256": "062aa2ec0c5f5354d75c6e9eb9e7c6adc344e5bbcc3a3e6b171ae1a08e527283", "inputs": { "raw_data": { "field": "raw_data", @@ -81,6 +84,7 @@ "type": "python", "module": "src.sample_algorithms", "function": "quality_check", + "sha256": "062aa2ec0c5f5354d75c6e9eb9e7c6adc344e5bbcc3a3e6b171ae1a08e527283", "inputs": { "raw_data": { "field": "raw_data", @@ -104,6 +108,8 @@ "type": "cpp", "lib_path": "src/sample_algos_cpp/simple_ops.so", "function": "multiply_by_factor_cpp", + "sha256": "96dac08510cbc464c66b20eec8a3d69f5fcb5c7c5d3ea9c5765eb88d7f5f5d31", + "isolate": true, "inputs": { "in_data": { "field": "clean_data", @@ -127,6 +133,7 @@ "type": "matlab", "m_path": "src/sample_algos_matlab", "function": "bandpass_filter_m", + "sha256": "c3a88dad53c37d4d83b0974c798e87e3790d329e7842053928b2867e44890249", "inputs": { "in_data": { "field": "raw_data", @@ -146,5 +153,284 @@ "description": "滤波后数据" } } + }, + { + "name": "Bellhop 传播损失", + "version": "2.0-woa18", + "author": "AcousticFastAPI", + "description": "射线追踪法传播损失计算（自动加载 WOA18 环境数据）", + "type": "web", + "endpoint": "http://acoustic-api:8000/tl", + "method": "POST", + "slot_key": "acoustic-api-compute", + "timeout_sec": 600, + "fixed_params": { + "modelType": "bellhop" + }, + "inputs": { + "freq": { + "field": "sea_env.modeling.freq_hz", + "type": "float", + "description": "建模频率 (Hz)" + }, + "sourceDepth": { + "field": "sea_env.modeling.source_depth_m", + "type": "int", + "description": "源深度 (m)" + }, + "receiveDepth": { + "field": "sea_env.modeling.receive_depth_m", + "type": "int", + "description": "接收深度 (m)" + }, + "sLat": { + "field": "sea_env.modeling.source_lat", + "type": "float", + "description": "源点纬度" + }, + "sLon": { + "field": "sea_env.modeling.source_lon", + "type": "float", + "description": "源点经度" + }, + "rLat": { + "field": "sea_env.modeling.receive_lat", + "type": "float", + "description": "接收点纬度" + }, + "rLon": { + "field": "sea_env.modeling.receive_lon", + "type": "float", + "description": "接收点经度" + }, + "timeType": { + "field": "sea_env.modeling.time_type", + "type": "str", + "description": "时间段" + } + }, + "outputs": { + "transmission_loss": { + "field": "vis_data.tl_result_path", + "type": "str", + "description": "TL 数据文件路径" + }, + "bathymetry": { + "field": "vis_data.tl_bathym_path", + "type": "str", + "description": "地形数据文件路径" + }, + "picture": { + "field": "vis_data.tl_picture_path", + "type": "str", + "description": "TL 图片路径" + }, + "summary": { + "field": "core_index.tl_summary", + "type": "dict", + "description": "TL 统计摘要" + } + } + }, + { + "name": "Kraken 传播损失", + "version": "2.0-woa18", + "author": "AcousticFastAPI", + "description": "简正波法传播损失计算（适用于低频深海场景）", + "type": "web", + "endpoint": "http://acoustic-api:8000/tl", + "method": "POST", + "slot_key": "acoustic-api-compute", + "timeout_sec": 600, + "fixed_params": { + "modelType": "kraken" + }, + "inputs": { + "freq": { + "field": "sea_env.modeling.freq_hz", + "type": "float", + "description": "建模频率 (Hz)" + }, + "sourceDepth": { + "field": "sea_env.modeling.source_depth_m", + "type": "int", + "description": "源深度 (m)" + }, + "receiveDepth": { + "field": "sea_env.modeling.receive_depth_m", + "type": "int", + "description": "接收深度 (m)" + }, + "sLat": { + "field": "sea_env.modeling.source_lat", + "type": "float", + "description": "源点纬度" + }, + "sLon": { + "field": "sea_env.modeling.source_lon", + "type": "float", + "description": "源点经度" + }, + "rLat": { + "field": "sea_env.modeling.receive_lat", + "type": "float", + "description": "接收点纬度" + }, + "rLon": { + "field": "sea_env.modeling.receive_lon", + "type": "float", + "description": "接收点经度" + }, + "timeType": { + "field": "sea_env.modeling.time_type", + "type": "str", + "description": "时间段" + } + }, + "outputs": { + "transmission_loss": { + "field": "vis_data.tl_result_path", + "type": "str", + "description": "TL 数据文件路径" + }, + "bathymetry": { + "field": "vis_data.tl_bathym_path", + "type": "str", + "description": "地形数据文件路径" + }, + "picture": { + "field": "vis_data.tl_picture_path", + "type": "str", + "description": "TL 图片路径" + }, + "summary": { + "field": "core_index.tl_summary", + "type": "dict", + "description": "TL 统计摘要" + } + } + }, + { + "name": "RAM 传播损失", + "version": "2.0-woa18", + "author": "AcousticFastAPI", + "description": "抛物方程法传播损失计算（适用于浅海复杂地形）", + "type": "web", + "endpoint": "http://acoustic-api:8000/tl", + "method": "POST", + "slot_key": "acoustic-api-compute", + "timeout_sec": 600, + "fixed_params": { + "modelType": "ram" + }, + "inputs": { + "freq": { + "field": "sea_env.modeling.freq_hz", + "type": "float", + "description": "建模频率 (Hz)" + }, + "sourceDepth": { + "field": "sea_env.modeling.source_depth_m", + "type": "int", + "description": "源深度 (m)" + }, + "receiveDepth": { + "field": "sea_env.modeling.receive_depth_m", + "type": "int", + "description": "接收深度 (m)" + }, + "sLat": { + "field": "sea_env.modeling.source_lat", + "type": "float", + "description": "源点纬度" + }, + "sLon": { + "field": "sea_env.modeling.source_lon", + "type": "float", + "description": "源点经度" + }, + "rLat": { + "field": "sea_env.modeling.receive_lat", + "type": "float", + "description": "接收点纬度" + }, + "rLon": { + "field": "sea_env.modeling.receive_lon", + "type": "float", + "description": "接收点经度" + }, + "timeType": { + "field": "sea_env.modeling.time_type", + "type": "str", + "description": "时间段" + } + }, + "outputs": { + "transmission_loss": { + "field": "vis_data.tl_result_path", + "type": "str", + "description": "TL 数据文件路径" + }, + "bathymetry": { + "field": "vis_data.tl_bathym_path", + "type": "str", + "description": "地形数据文件路径" + }, + "picture": { + "field": "vis_data.tl_picture_path", + "type": "str", + "description": "TL 图片路径" + }, + "summary": { + "field": "core_index.tl_summary", + "type": "dict", + "description": "TL 统计摘要" + } + } + }, + { + "name": "声速剖面查询", + "version": "2.0-woa18", + "author": "AcousticFastAPI", + "description": "根据经纬度和时间段从 WOA18 和 ETOPO1 查询声速剖面", + "type": "web", + "endpoint": "http://acoustic-api:8000/profile", + "method": "POST", + "slot_key": "acoustic-api-compute", + "timeout_sec": 120, + "inputs": { + "lat": { + "field": "sea_env.modeling.source_lat", + "type": "float", + "description": "纬度" + }, + "lon": { + "field": "sea_env.modeling.source_lon", + "type": "float", + "description": "经度" + }, + "timeType": { + "field": "sea_env.modeling.time_type", + "type": "str", + "description": "时间段" + } + }, + "outputs": { + "profile": { + "field": "vis_data.profile_path", + "type": "str", + "description": "剖面数据文件路径" + }, + "picture": { + "field": "vis_data.profile_picture_path", + "type": "str", + "description": "剖面图片路径" + }, + "summary": { + "field": "core_index.profile_summary", + "type": "dict", + "description": "声道特征摘要" + } + } } ] \ No newline at end of file diff --git a/main.py b/main.py index 5ecb414..1e912ca 100644 --- a/main.py +++ b/main.py @@ -181,7 +181,7 @@ def show_algo_library(): col1, col2, col3 = st.columns(3) col1.metric("Python", "✅ 可用") col2.metric("MATLAB/Octave", "✅ 可用" if "matlab" in available_types else "❌ 不可用") - col3.metric("C++ 编译器", "✅ 可用" if "cpp" in available_types else "❌ 不可用") + col3.metric("C++ 运行时", "✅ 可用" if "cpp" in available_types else "❌ 不可用") st.json(env) st.subheader("已注册算法") @@ -406,7 +406,11 @@ def show_scheduling(): if algo_type == "python": try: - algo_func = load_algo_func(algo.get("module", ""), algo.get("function", "")) + algo_func = load_algo_func( + algo.get("module", ""), + algo.get("function", ""), + expected_sha256=algo.get("sha256") or None, + ) except Exception as e: st.error(f"算法加载失败：{algo.get('name', '')} - {e}") st.stop() @@ -430,6 +434,8 @@ def show_scheduling(): preprocessing=preprocessing, requirements=algo.get("inputs") or {}, # 包含 argtypes/restype 如果在 JSON 中定义 output_specs=algo.get("outputs") or {}, + expected_sha256=algo.get("sha256") or None, + isolate=bool(algo.get("isolate", False)), ) elif algo_type == "matlab": from src.bridges import MatlabBridge @@ -442,6 +448,20 @@ def show_scheduling(): preprocessing=preprocessing, requirements=algo.get("inputs") or {}, output_specs=algo.get("outputs") or {}, + expected_sha256=algo.get("sha256") or None, + ) + elif algo_type == "web": + from src.bridges import WebBridge + bridge = WebBridge( + algo["name"], + algo.get("endpoint", ""), + in_map, + out_map, + method=algo.get("method", "POST"), + timeout_sec=algo.get("timeout_sec", 300), + slot_key=algo.get("slot_key"), + fixed_params=algo.get("fixed_params"), + sanitize_output=algo.get("sanitize_output", True), ) else: st.warning(f"暂不支持的算法类型: {algo_type}") diff --git a/src/bridges.py b/src/bridges.py index 954354f..58a3556 100644 --- a/src/bridges.py +++ b/src/bridges.py @@ -108,6 +108,8 @@ class CppBridge(AlgorithmBridge): preprocessing: Optional[List[Dict[str, Any]]] = None, requirements: Optional[Dict[str, Any]] = None, output_specs: Optional[Dict[str, Any]] = None, + expected_sha256: Optional[str] = None, + isolate: bool = False, is_core: bool = False): super().__init__(name, is_core) self.lib_path = lib_path @@ -120,26 +122,45 @@ class CppBridge(AlgorithmBridge): self.algo_type = "cpp" self._lib = None self._func = None + self.expected_sha256 = expected_sha256 + self.isolate = isolate - def _validate_path(self, path: str): + def _validate_path(self, path: str) -> 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")): + base_dir = os.path.realpath(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) # project root + abs_path = os.path.realpath(path if os.path.isabs(path) else os.path.join(base_dir, path)) + allowed_dirs = [ + os.path.join(base_dir, "src"), + os.path.join(base_dir, "data"), + ] + if not any(os.path.commonpath([abs_path, d]) == d for d in allowed_dirs): raise PermissionError(f"Security Block: Path {path} is outside allowed directories (src/data).") + return abs_path 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}") + abs_lib_path = self._validate_path(self.lib_path) + if not os.path.exists(abs_lib_path): + raise FileNotFoundError(f"C++ library not found: {abs_lib_path}") + + if self.expected_sha256: + from src.integrity import ensure_file_sha256 + ensure_file_sha256(abs_lib_path, self.expected_sha256, label=f"cpp library {self.name}") - self._lib = ctypes.CDLL(self.lib_path) + self._lib = ctypes.CDLL(abs_lib_path) self._func = getattr(self._lib, self.function_name) # Phase 5: ABI Hardening + # 默认签名（统一约定）：void (double*, int, double*, int) + self._func.argtypes = [ + ctypes.POINTER(ctypes.c_double), + ctypes.c_int, + ctypes.POINTER(ctypes.c_double), + ctypes.c_int, + ] + self._func.restype = None + # 如果 requirements 中定义了签名，则应用它 if "argtypes" in self.requirements: # 简单映射: 目前支持 "double_ptr", "int" @@ -149,10 +170,55 @@ class CppBridge(AlgorithmBridge): mapping = {"int": ctypes.c_int, "void": None} self._func.restype = mapping.get(self.requirements["restype"], None) + def _call_isolated(self, in_data: np.ndarray) -> np.ndarray: + import json + import subprocess + import sys + + abs_lib_path = self._validate_path(self.lib_path) + if self.expected_sha256: + from src.integrity import ensure_file_sha256 + ensure_file_sha256(abs_lib_path, self.expected_sha256, label=f"cpp library {self.name}") + + cfg = { + "lib_path": abs_lib_path, + "function_name": self.function_name, + "expected_sha256": self.expected_sha256, + } + if isinstance(self.requirements, dict): + if isinstance(self.requirements.get("argtypes"), list): + cfg["argtypes"] = self.requirements.get("argtypes") + if isinstance(self.requirements.get("restype"), str): + cfg["restype"] = self.requirements.get("restype") + + base_dir = os.path.realpath(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + env = os.environ.copy() + env["PYTHONPATH"] = base_dir + os.pathsep + env.get("PYTHONPATH", "") + + with tempfile.TemporaryDirectory(prefix="cpp_bridge_") as tmp_dir: + cfg_path = os.path.join(tmp_dir, "config.json") + in_path = os.path.join(tmp_dir, "in.npy") + out_path = os.path.join(tmp_dir, "out.npy") + + with open(cfg_path, "w", encoding="utf-8") as f: + json.dump(cfg, f, ensure_ascii=False) + np.save(in_path, np.ascontiguousarray(in_data, dtype=np.float64), allow_pickle=False) + + cmd = [sys.executable, "-m", "src.cpp_isolated_runner", "--config", cfg_path, "--input", in_path, "--output", out_path] + proc = subprocess.run(cmd, cwd=base_dir, env=env, capture_output=True, text=True) + if proc.returncode != 0: + stderr = (proc.stderr or "").strip() + stdout = (proc.stdout or "").strip() + details = stderr or stdout or f"returncode={proc.returncode}" + raise RuntimeError(f"isolated C++ call failed: {details}") + + if not os.path.exists(out_path): + raise RuntimeError("isolated C++ call produced no output file") + return np.load(out_path, allow_pickle=False) + def call_algo(self, water_data: WaterSoundData) -> WaterSoundData: try: water_data.add_log(self.name, "Start C++ Execution") - self._load_lib() # 1. 准备输入 # 为了简化 Phase 3.3，我们假设目前只支持 1进1出 (double array) @@ -165,18 +231,22 @@ class CppBridge(AlgorithmBridge): # 确保数据连续且为 float64 (double) in_data = np.ascontiguousarray(in_data, dtype=np.float64) - in_ptr = in_data.ctypes.data_as(ctypes.POINTER(ctypes.c_double)) - in_len = ctypes.c_int(in_data.size) + if self.isolate: + out_data = self._call_isolated(in_data) + else: + self._load_lib() + in_ptr = in_data.ctypes.data_as(ctypes.POINTER(ctypes.c_double)) + in_len = ctypes.c_int(in_data.size) - # 2. 准备输出空间 - # 同样简化为与输入等长 - out_data = np.zeros_like(in_data) - out_ptr = out_data.ctypes.data_as(ctypes.POINTER(ctypes.c_double)) - out_len = ctypes.c_int(out_data.size) + # 2. 准备输出空间 + # 同样简化为与输入等长 + out_data = np.zeros_like(in_data) + out_ptr = out_data.ctypes.data_as(ctypes.POINTER(ctypes.c_double)) + out_len = ctypes.c_int(out_data.size) - # 3. 调用 C++ 函数 - # 默认签名: void (double*, int, double*, int) - self._func(in_ptr, in_len, out_ptr, out_len) + # 3. 调用 C++ 函数 + # 默认签名: void (double*, int, double*, int) + self._func(in_ptr, in_len, out_ptr, out_len) # 4. 映射结果回 standardization 数据 out_attr = list(self.output_mapping.values())[0] @@ -198,6 +268,7 @@ class MatlabBridge(AlgorithmBridge): preprocessing: Optional[List[Dict[str, Any]]] = None, requirements: Optional[Dict[str, Any]] = None, output_specs: Optional[Dict[str, Any]] = None, + expected_sha256: Optional[str] = None, is_core: bool = False): super().__init__(name, is_core) self.m_path = m_path # .m 文件所在目录 @@ -209,22 +280,40 @@ class MatlabBridge(AlgorithmBridge): self.output_specs = output_specs or {} self.algo_type = "matlab" self._engine = None + self.expected_sha256 = expected_sha256 - def _validate_path(self, path: str): + def _validate_path(self, path: str) -> 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")): + base_dir = os.path.realpath(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) # project root + abs_path = os.path.realpath(path if os.path.isabs(path) else os.path.join(base_dir, path)) + allowed_dirs = [ + os.path.join(base_dir, "src"), + os.path.join(base_dir, "data"), + ] + if not any(os.path.commonpath([abs_path, d]) == d for d in allowed_dirs): raise PermissionError(f"Security Block: MATLAB path {path} is outside allowed directories.") + return abs_path + + def _verify_entrypoint_integrity(self): + if not self.expected_sha256: + return + + m_dir = self._validate_path(self.m_path) + entry_file = self._validate_path(os.path.join(m_dir, f"{self.function_name}.m")) + if not os.path.exists(entry_file): + raise FileNotFoundError(f"MATLAB entry file not found: {entry_file}") + + from src.integrity import ensure_file_sha256 + ensure_file_sha256(entry_file, self.expected_sha256, label=f"matlab entry {self.name}") def _call_via_engine(self, inputs: Dict[str, Any]) -> Any: """方式 1: 使用官方 matlab.engine""" import matlab.engine - self._validate_path(self.m_path) + import matlab + m_dir = self._validate_path(self.m_path) if self._engine is None: self._engine = matlab.engine.start_matlab() - self._engine.addpath(self.m_path) + self._engine.addpath(m_dir) func = getattr(self._engine, self.function_name) # 将 numpy 数组转为 matlab 类型 (这里简化处理) @@ -234,8 +323,8 @@ 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) + m_dir = self._validate_path(self.m_path) + octave.addpath(m_dir) func = getattr(octave, self.function_name) return func(*inputs.values()) @@ -243,7 +332,7 @@ class MatlabBridge(AlgorithmBridge): """方式 3: 文件兜底 (使用 .mat 文件交换)""" from scipy.io import savemat, loadmat import subprocess - self._validate_path(self.m_path) + m_dir = 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") @@ -251,7 +340,7 @@ class MatlabBridge(AlgorithmBridge): savemat(in_file, inputs) # 构造执行脚本：加入 m_path，加载输入，调用函数，保存输出 - m_path = os.path.abspath(self.m_path).replace("'", "''") + m_path = os.path.abspath(m_dir).replace("'", "''") in_file_escaped = in_file.replace("'", "''") out_file_escaped = out_file.replace("'", "''") arglist = ",".join(inputs.keys()) @@ -280,6 +369,7 @@ class MatlabBridge(AlgorithmBridge): def call_algo(self, water_data: WaterSoundData) -> WaterSoundData: try: water_data.add_log(self.name, "Start MATLAB Execution") + self._verify_entrypoint_integrity() # 准备输入 algo_inputs = {} @@ -313,3 +403,138 @@ class MatlabBridge(AlgorithmBridge): water_data.set_error(f"MATLAB Algorithm {self.name} failed: {str(e)}") return water_data + + +import requests + +class WebBridge(AlgorithmBridge): + """HTTP/HTTPS 远程服务桥接（通用工作流组件）""" + + def __init__(self, name: str, endpoint_url: str, + input_mapping: Dict[str, str], # WSD 字段路径 -> HTTP 请求字段 + output_mapping: Dict[str, str], # HTTP 响应字段 -> WSD 字段路径 + method: str = "POST", + timeout_sec: int = 300, + slot_key: Optional[str] = None, # 资源槽位 Key（同 Key 串行执行） + fixed_params: Optional[Dict[str, Any]] = None, # 固定参数（如 modelType） + sanitize_output: bool = True, # 是否清洗输出参数 + is_core: bool = False): + super().__init__(name, is_core) + self.endpoint_url = endpoint_url + self.input_mapping = input_mapping + self.output_mapping = output_mapping + self.method = method.upper() + self.timeout_sec = timeout_sec + self.slot_key = slot_key + self.fixed_params = fixed_params or {} + self.sanitize_output = sanitize_output + self.algo_type = "web" + + def _get_nested(self, obj: Any, path: str) -> Any: + """支持 'sea_env.modeling.freq_hz' 这种嵌套取值""" + for key in path.split('.'): + if obj is None: + return None + if isinstance(obj, dict): + obj = obj.get(key) + else: + obj = getattr(obj, key, None) + return obj + + def _set_nested(self, obj: Any, path: str, value: Any) -> None: + """支持 'vis_data.tl_result_path' 这种嵌套写入""" + keys = path.split('.') + for key in keys[:-1]: + if isinstance(obj, dict): + if key not in obj: + obj[key] = {} + obj = obj[key] + else: + if not hasattr(obj, key): + setattr(obj, key, {}) + next_obj = getattr(obj, key) + if not isinstance(next_obj, dict): + setattr(obj, key, {}) + next_obj = getattr(obj, key) + obj = next_obj + + final_key = keys[-1] + if isinstance(obj, dict): + obj[final_key] = value + else: + setattr(obj, final_key, value) + + def call_algo(self, water_data: WaterSoundData) -> WaterSoundData: + from src.resource_slots import slot_context + from src.security import ( + validate_endpoint_host, sanitize_output_params, + generate_run_id, version_tracker + ) + + try: + water_data.add_log(self.name, "Start Web Execution", f"Endpoint: {self.endpoint_url}") + + # 1. 安全检查：Host 白名单 + validate_endpoint_host(self.endpoint_url) + + # 2. 构建请求参数 + request_params = dict(self.fixed_params) + for http_field, wsd_path in self.input_mapping.items(): + value = self._get_nested(water_data, wsd_path) + if value is not None: + request_params[http_field] = value + + # 3. 清洗输出参数（生成唯一 run_id） + run_id = generate_run_id() + if self.sanitize_output: + request_params = sanitize_output_params(request_params, run_id) + + water_data.add_log(self.name, "Request Prepared", f"run_id={run_id}") + + # 4. 获取资源槽位锁并发起请求 + with slot_context(self.slot_key): + if self.method == "POST": + response = requests.post( + self.endpoint_url, + json=request_params, + timeout=self.timeout_sec + ) + elif self.method == "GET": + response = requests.get( + self.endpoint_url, + params=request_params, + timeout=self.timeout_sec + ) + else: + raise ValueError(f"Unsupported HTTP method: {self.method}") + + response.raise_for_status() + result = response.json() + + # 5. 记录服务版本 + if "version" in result: + version_tracker.record(self.endpoint_url, result["version"]) + water_data.add_log(self.name, "Service Version", result["version"]) + + # 6. 检查执行结果 + if not result.get("success", True): + error_msg = result.get("error", result.get("message", "Unknown error")) + raise RuntimeError(f"Remote service returned failure: {error_msg}") + + # 7. 映射响应到 WSD + response_data = result.get("data", result) + for http_field, wsd_path in self.output_mapping.items(): + value = self._get_nested(response_data, http_field) + if value is not None: + self._set_nested(water_data, wsd_path, value) + + water_data.add_log(self.name, "Finished Web Execution") + + except requests.exceptions.Timeout: + water_data.set_error(f"WebBridge {self.name} timeout after {self.timeout_sec}s") + except requests.exceptions.RequestException as e: + water_data.set_error(f"WebBridge {self.name} request failed: {str(e)}") + except Exception as e: + water_data.set_error(f"WebBridge {self.name} failed: {str(e)}") + + return water_data diff --git a/src/resource_slots.py b/src/resource_slots.py new file mode 100644 index 0000000..b864d51 --- /dev/null +++ b/src/resource_slots.py @@ -0,0 +1,63 @@ +""" +资源槽位管理器 - 用于控制特定资源类型的并发访问 + +某些算法后端（如 AcousticFastAPI）底层使用固定文件名，不支持并发。 +通过 slot_key 机制，调度器可以确保同一 slot_key 的算法串行执行。 +""" +from __future__ import annotations + +import threading +from collections import defaultdict +from contextlib import contextmanager +from typing import Dict, Optional + + +class ResourceSlotManager: + """线程安全的资源槽位管理器""" + + _instance: Optional["ResourceSlotManager"] = None + _lock = threading.Lock() + + def __new__(cls): + if cls._instance is None: + with cls._lock: + if cls._instance is None: + cls._instance = super().__new__(cls) + cls._instance._slots: Dict[str, threading.Lock] = defaultdict(threading.Lock) + return cls._instance + + def acquire(self, slot_key: str) -> None: + """获取指定槽位的锁""" + self._slots[slot_key].acquire() + + def release(self, slot_key: str) -> None: + """释放指定槽位的锁""" + if slot_key in self._slots: + try: + self._slots[slot_key].release() + except RuntimeError: + pass # 锁未被持有 + + @contextmanager + def slot(self, slot_key: Optional[str]): + """上下文管理器：自动获取和释放槽位锁""" + if slot_key: + self.acquire(slot_key) + try: + yield + finally: + if slot_key: + self.release(slot_key) + + +# 全局单例 +_manager = ResourceSlotManager() + +def acquire_slot(slot_key: str) -> None: + _manager.acquire(slot_key) + +def release_slot(slot_key: str) -> None: + _manager.release(slot_key) + +def slot_context(slot_key: Optional[str]): + return _manager.slot(slot_key) diff --git a/src/security.py b/src/security.py new file mode 100644 index 0000000..f2480f9 --- /dev/null +++ b/src/security.py @@ -0,0 +1,107 @@ +""" +平台级安全策略模块 + +提供 Host 白名单、路径清洗、服务版本追溯等通用安全能力。 +""" +from __future__ import annotations + +import os +import re +import uuid +from typing import Dict, List, Optional, Set +from urllib.parse import urlparse + + +# 默认允许的远程服务主机（可通过环境变量扩展） +DEFAULT_ALLOWED_HOSTS: Set[str] = { + "localhost", + "127.0.0.1", + "acoustic-api", +} + +def _load_allowed_hosts() -> Set[str]: + """从环境变量加载额外的允许主机""" + hosts = set(DEFAULT_ALLOWED_HOSTS) + extra = os.environ.get("HYD_ALLOWED_HOSTS", "") + if extra: + hosts.update(h.strip() for h in extra.split(",") if h.strip()) + return hosts + +ALLOWED_HOSTS = _load_allowed_hosts() + + +def validate_endpoint_host(url: str) -> str: + """ + 校验 URL 的主机是否在白名单内（防 SSRF） + + Returns: + 解析后的主机名 + Raises: + PermissionError: 主机不在白名单内 + """ + parsed = urlparse(url) + host = parsed.hostname or "" + + if not host: + raise ValueError(f"无法解析 URL 主机: {url}") + + if host not in ALLOWED_HOSTS: + raise PermissionError( + f"SSRF Block: Host '{host}' not in allowlist. " + f"Allowed: {', '.join(sorted(ALLOWED_HOSTS))}" + ) + return host + + +def generate_run_id() -> str: + """生成唯一的运行 ID，用于输出文件命名""" + return uuid.uuid4().hex[:12] + + +def sanitize_output_params(params: Dict, run_id: Optional[str] = None) -> Dict: + """ + 清洗输出参数，防止路径穿越攻击 + + - 强制覆盖 outFileName 为调度器生成的 run_id + - 固定 outPath 为安全目录 + """ + if run_id is None: + run_id = generate_run_id() + + # 强制覆盖，不允许用户自定义 + params = dict(params) + params["outFileName"] = run_id + + # outPath 只允许预定义的安全值 + allowed_out_paths = {"tl", "profile", "output"} + out_path = params.get("outPath", "output") + if out_path not in allowed_out_paths: + params["outPath"] = "output" + + return params + + +def sanitize_filename(filename: str) -> str: + """清洗文件名，移除危险字符""" + # 只保留字母、数字、下划线、连字符、点 + return re.sub(r'[^\w\-.]', '_', filename) + + +class ServiceVersionTracker: + """服务版本追踪器，用于审计与回滚""" + + def __init__(self): + self._versions: Dict[str, str] = {} + + def record(self, endpoint: str, version: str) -> None: + self._versions[endpoint] = version + + def get(self, endpoint: str) -> Optional[str]: + return self._versions.get(endpoint) + + def all(self) -> Dict[str, str]: + return dict(self._versions) + + +# 全局版本追踪器 +version_tracker = ServiceVersionTracker()