fix(security): Patch path injection and stack trace exposure (CodeQL #31, #27)

- Fix py/path-injection (Alert #31, High severity):
  - Add _validate_job_path() to resolve and canonicalize paths
  - Enforce job_path stays within safe_root via relative_to()
  - Update get_max_loops(), get_job_summary_df(), render_job_summary()
    to accept and validate safe_root parameter
  - Update app.py caller to pass safe_root to render_job_summary()
  - On validation failure: return empty data / show warning

- Fix py/stack-trace-exposure (Alert #27, Medium severity):
  - Remove str(e) from error response in get_live_fx_data()
  - Replace with generic message: 'Internal error while fetching live FX data'
  - Remove unused exception variable to prevent accidental leakage

Files:
  rdagent/app/rl/ui/rl_summary.py
  rdagent/app/rl/ui/app.py
  rdagent/components/coder/factor_coder/eurusd_macro.py
This commit is contained in:
TPTBusiness
2026-04-11 21:50:16 +02:00
parent 4779348d13
commit 60f10b3667
3 changed files with 35 additions and 7 deletions
+1 -1
View File
@@ -206,7 +206,7 @@ def main():
st.warning(str(e))
return
if job_path.exists():
render_job_summary(job_path, is_root=is_root_job)
render_job_summary(job_path, safe_root, is_root=is_root_job)
else:
st.warning(f"Job folder not found: {job_folder}")
return
+32 -4
View File
@@ -61,8 +61,24 @@ def get_loop_status(task_path: Path, loop_id: int) -> tuple[str, bool | None]:
return "?", None
def get_max_loops(job_path: Path) -> int:
def _validate_job_path(job_path: Path, safe_root: Path) -> Path:
"""Resolve and validate that job_path stays within safe_root."""
resolved_root = safe_root.expanduser().resolve()
resolved_job = job_path.expanduser().resolve()
try:
resolved_job.relative_to(resolved_root)
return resolved_job
except ValueError:
raise ValueError(f"Job path {resolved_job} is outside allowed root {resolved_root}")
def get_max_loops(job_path: Path, safe_root: Path | None = None) -> int:
"""Get maximum number of loops across all tasks"""
if safe_root is not None:
try:
job_path = _validate_job_path(job_path, safe_root)
except ValueError:
return 0
max_loops = 0
for task_dir in job_path.iterdir():
if is_valid_task(task_dir):
@@ -71,8 +87,14 @@ def get_max_loops(job_path: Path) -> int:
return max_loops
def get_job_summary_df(job_path: Path) -> tuple[pd.DataFrame, pd.DataFrame]:
def get_job_summary_df(job_path: Path, safe_root: Path | None = None) -> tuple[pd.DataFrame, pd.DataFrame]:
"""Generate summary DataFrame for all tasks in job"""
if safe_root is not None:
try:
job_path = _validate_job_path(job_path, safe_root)
except ValueError:
return pd.DataFrame(), pd.DataFrame()
if not job_path.exists():
return pd.DataFrame(), pd.DataFrame()
@@ -149,12 +171,18 @@ def style_df_with_decisions(df: pd.DataFrame, decisions_df: pd.DataFrame):
return df.style.apply(lambda _: styles, axis=None)
def render_job_summary(job_path: Path, is_root: bool = False) -> None:
def render_job_summary(job_path: Path, safe_root: Path, is_root: bool = False) -> None:
"""Render job summary UI"""
try:
job_path = _validate_job_path(job_path, safe_root)
except ValueError:
st.warning("Invalid job path outside allowed root")
return
title = "Standalone Tasks" if is_root else f"Job: {job_path.name}"
st.subheader(title)
df, decisions_df = get_job_summary_df(job_path)
df, decisions_df = get_job_summary_df(job_path, safe_root)
if df.empty:
st.warning("No valid tasks found in this job directory")
return
@@ -116,14 +116,14 @@ def get_live_fx_data() -> dict:
"success": True
}
except Exception as e:
except Exception:
return {
"eurusd_price": None,
"dxy_price": None,
"realized_volatility": None,
"eurusd_24h_change": None,
"success": False,
"error": str(e)
"error": "Internal error while fetching live FX data"
}