wangsli/acoustic_algo_scheduler
1
0
派生 0
代码 工单 合并请求 工作流 软件包 项目 发布 百科 活动

feat: phase 6 webbridge + generic workflow (resource slots, security, bellhop/kraken/ram assets)

浏览代码
这个提交包含在：
wangsli
2026-01-21 15:47:01 +08:00
父节点 7477acbdfa
当前提交 e30cd6d93f
共修改 5 个文件，包含 734 行新增和 33 行删除
下载 Patch 文件 下载 Diff 文件 展开所有文件 折叠所有文件
data/algorithms.json
+286
查看文件
@@ -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": "声道特征摘要"
}
}
}
]
main.py
+22 -2
查看文件
@@ -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', '<unknown>')} - {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}")
src/bridges.py
+256 -31
查看文件
@@ -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
src/resource_slots.py
+63
查看文件
@@ -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)
src/security.py
+107
查看文件
@@ -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()