feat: phase 5 industrial grade hardening (ABI safety, security whitelisting, V2 metadata)
这个提交包含在:
+32
-3
@@ -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")
|
||||
|
||||
在新工单中引用
屏蔽一个用户