fix(security): resolve path-injection, B701, B101, B112 Bandit alerts

- Path injection (B614): centralized safe_resolve_path in core/utils.py,
  refactored 6 UI modules to use it with safe_root validation
- B701: added explicit autoescape=select_autoescape() to Jinja2
  Environment() calls in 3 files
- B101: replaced assert statements with proper if/raise patterns in
  12+ files (partial)
- B112: added logger.warning() to bare except:continue blocks in
  5 files
This commit is contained in:
TPTBusiness
2026-05-01 13:42:02 +02:00
parent bf36f54159
commit 64ed6b0cce
28 changed files with 126 additions and 148 deletions
@@ -54,7 +54,8 @@ def get_ds_env(
ValueError: If the env_type is not recognized.
"""
conf = DSCoderCoSTEERSettings()
assert conf_type in ["kaggle", "mlebench"], f"Unknown conf_type: {conf_type}"
if conf_type not in ["kaggle", "mlebench"]:
raise ValueError(f"Unknown conf_type: {conf_type}")
if conf.env_type == "docker":
env_conf = DSDockerConf() if conf_type == "kaggle" else MLEBDockerConf()
@@ -79,7 +80,8 @@ def get_clear_ws_cmd(stage: Literal["before_training", "before_inference"] = "be
"""
Clean the files in workspace to a specific stage
"""
assert stage in ["before_training", "before_inference"], f"Unknown stage: {stage}"
if stage not in ["before_training", "before_inference"]:
raise ValueError(f"Unknown stage: {stage}")
if DS_RD_SETTING.enable_model_dump and stage == "before_training":
cmd = "rm -r submission.csv scores.csv models trace.log"
else:
@@ -13,7 +13,7 @@ File structure
from pathlib import Path
from jinja2 import Environment, StrictUndefined
from jinja2 import Environment, StrictUndefined, select_autoescape
from rdagent.app.data_science.conf import DS_RD_SETTING
from rdagent.components.coder.CoSTEER.evaluators import (
@@ -88,7 +88,7 @@ class EnsembleMultiProcessEvolvingStrategy(MultiProcessEvolvingStrategy):
code_spec = workspace.file_dict["spec/ensemble.md"]
else:
test_code = (
Environment(undefined=StrictUndefined)
Environment(undefined=StrictUndefined, autoescape=select_autoescape())
.from_string((DIRNAME / "eval_tests" / "ensemble_test.txt").read_text())
.render(
model_names=[
@@ -2,7 +2,7 @@ import json
import re
from pathlib import Path
from jinja2 import Environment, StrictUndefined
from jinja2 import Environment, StrictUndefined, select_autoescape
from rdagent.app.data_science.conf import DS_RD_SETTING
from rdagent.components.coder.CoSTEER.evaluators import (
@@ -55,7 +55,7 @@ class EnsembleCoSTEEREvaluator(CoSTEEREvaluator):
fname = "test/ensemble_test.txt"
test_code = (DIRNAME / "eval_tests" / "ensemble_test.txt").read_text()
test_code = (
Environment(undefined=StrictUndefined)
Environment(undefined=StrictUndefined, autoescape=select_autoescape())
.from_string(test_code)
.render(
model_names=[
@@ -41,7 +41,8 @@ class ModelCoSTEEREvaluator(CoSTEEREvaluator):
final_feedback="This task has failed too many times, skip implementation.",
final_decision=False,
)
assert isinstance(target_task, ModelTask)
if not isinstance(target_task, ModelTask):
raise TypeError(f"Expected ModelTask, got {type(target_task)}")
# NOTE: Use fixed input to test the model to avoid randomness
batch_size = 8
@@ -50,7 +51,8 @@ class ModelCoSTEEREvaluator(CoSTEEREvaluator):
input_value = 0.4
param_init_value = 0.6
assert isinstance(implementation, ModelFBWorkspace)
if not isinstance(implementation, ModelFBWorkspace):
raise TypeError(f"Expected ModelFBWorkspace, got {type(implementation)}")
model_execution_feedback, gen_np_array = implementation.execute(
batch_size=batch_size,
num_features=num_features,
@@ -59,7 +61,8 @@ class ModelCoSTEEREvaluator(CoSTEEREvaluator):
param_init_value=param_init_value,
)
if gt_implementation is not None:
assert isinstance(gt_implementation, ModelFBWorkspace)
if not isinstance(gt_implementation, ModelFBWorkspace):
raise TypeError(f"Expected ModelFBWorkspace, got {type(gt_implementation)}")
_, gt_np_array = gt_implementation.execute(
batch_size=batch_size,
num_features=num_features,