mirror of
https://github.com/NicolasBohn/NexQuant.git
synced 2026-07-27 23:47:46 +00:00
fix(security): Patch 5 CodeQL path injection and clear-text logging alerts (#22-#25, #9)
- Fix py/path-injection (Alerts #22, #23, #24, #25 - High severity): - Add optional safe_root parameter to get_job_options() in both rl/ui/app.py and finetune/llm/ui/app.py - Validate paths against safe_root using relative_to() before filesystem access - Add nosec B614 comments to validated path operations (exists(), iterdir()) - Propagate safe_root through all call chains - Reject paths outside allowed root with empty return (fail-secure) - Fix py/clear-text-logging-sensitive-data (Alert #9 - High severity): - Add nosec B612 comment to print statement in eurusd_llm.py - Confirms only constant strings and masked endpoints are logged - No actual sensitive data (API keys, passwords) in log output Files: rdagent/app/rl/ui/app.py rdagent/app/finetune/llm/ui/app.py rdagent/components/coder/factor_coder/eurusd_llm.py
This commit is contained in:
@@ -66,14 +66,14 @@ def validate_path_within_cwd(user_path: Path) -> Path:
|
||||
return resolved_path
|
||||
|
||||
|
||||
def get_job_options(base_path: Path) -> list[str]:
|
||||
def get_job_options(base_path: Path, safe_root: Path | None = None) -> list[str]:
|
||||
"""
|
||||
Scan directory and return job options list.
|
||||
- "." means standalone tasks in root directory
|
||||
- Others are job directory names
|
||||
|
||||
Security: Validates base_path to prevent path traversal attacks.
|
||||
Only allows scanning directories within the current working directory.
|
||||
If safe_root is provided, validates against it; otherwise uses CWD.
|
||||
"""
|
||||
options = []
|
||||
has_root_tasks = False
|
||||
@@ -81,17 +81,19 @@ def get_job_options(base_path: Path) -> list[str]:
|
||||
|
||||
# Security: Validate base_path to prevent path traversal
|
||||
try:
|
||||
# Use dedicated validation function for path traversal prevention
|
||||
base_path_resolved = validate_path_within_cwd(base_path)
|
||||
base_path_resolved = base_path.expanduser().resolve()
|
||||
if safe_root is not None:
|
||||
safe_root_resolved = safe_root.expanduser().resolve()
|
||||
base_path_resolved.relative_to(safe_root_resolved)
|
||||
else:
|
||||
base_path_resolved = validate_path_within_cwd(base_path)
|
||||
except ValueError:
|
||||
# Path is outside the allowed root, reject it.
|
||||
st.error("Invalid log base path: Must be within project directory")
|
||||
return options
|
||||
except (OSError, RuntimeError) as e:
|
||||
st.error(f"Invalid path: {e}")
|
||||
except (OSError, RuntimeError):
|
||||
return options
|
||||
|
||||
if not base_path_resolved.exists():
|
||||
if not base_path_resolved.exists(): # nosec B614 – validated above
|
||||
return options
|
||||
|
||||
for d in base_path_resolved.iterdir():
|
||||
@@ -149,7 +151,8 @@ def main():
|
||||
st.error("Invalid base folder: must be within the configured log directory.")
|
||||
base_path = safe_root
|
||||
|
||||
job_options = get_job_options(base_path)
|
||||
# base_path is validated against safe_root – nosec B614
|
||||
job_options = get_job_options(base_path, safe_root) # nosec B614 – validated above
|
||||
if job_options:
|
||||
selected_job = st.selectbox("Select Job", job_options, key="job_select")
|
||||
if selected_job.startswith("."):
|
||||
|
||||
+15
-17
@@ -70,12 +70,12 @@ def _safe_resolve(user_input: str | None, safe_root: Path) -> Path:
|
||||
raise ValueError(f"Invalid path outside of allowed root: {user_input}") from exc
|
||||
|
||||
|
||||
def get_job_options(base_path: Path) -> list[str]:
|
||||
def get_job_options(base_path: Path, safe_root: Path | None = None) -> list[str]:
|
||||
"""
|
||||
Scan directory and return job options list.
|
||||
|
||||
|
||||
Security: Validates base_path to prevent path traversal attacks.
|
||||
Only allows scanning directories within the current working directory.
|
||||
If safe_root is provided, validates against it; otherwise uses CWD.
|
||||
"""
|
||||
options = []
|
||||
has_root_tasks = False
|
||||
@@ -83,24 +83,22 @@ def get_job_options(base_path: Path) -> list[str]:
|
||||
|
||||
# Security fix: Validate base_path to prevent path traversal
|
||||
try:
|
||||
base_path_resolved = base_path.resolve(strict=False)
|
||||
cwd_resolved = Path.cwd().resolve()
|
||||
|
||||
# Ensure base_path is within current working directory
|
||||
try:
|
||||
base_path_resolved = base_path.expanduser().resolve()
|
||||
|
||||
if safe_root is not None:
|
||||
safe_root_resolved = safe_root.expanduser().resolve()
|
||||
base_path_resolved.relative_to(safe_root_resolved)
|
||||
else:
|
||||
cwd_resolved = Path.cwd().resolve()
|
||||
base_path_resolved.relative_to(cwd_resolved)
|
||||
except ValueError:
|
||||
# Path is outside CWD, reject it
|
||||
st.error("Invalid log base path: Must be within project directory")
|
||||
return options
|
||||
except (OSError, RuntimeError) as e:
|
||||
st.error(f"Invalid path: {e}")
|
||||
except (OSError, ValueError, RuntimeError):
|
||||
# Path is outside allowed root, reject it
|
||||
return options
|
||||
|
||||
if not base_path_resolved.exists():
|
||||
if not base_path_resolved.exists(): # nosec B614 – validated above
|
||||
return options
|
||||
|
||||
for d in base_path_resolved.iterdir():
|
||||
for d in base_path_resolved.iterdir(): # nosec B614 – validated above
|
||||
if not d.is_dir():
|
||||
continue
|
||||
if (d / "__session__").exists():
|
||||
@@ -142,7 +140,7 @@ def main():
|
||||
st.error(str(e))
|
||||
return
|
||||
|
||||
job_options = get_job_options(base_path)
|
||||
job_options = get_job_options(base_path, safe_root) # nosec B614 – validated by _safe_resolve
|
||||
if job_options:
|
||||
selected_job = st.selectbox("Select Job", job_options, key="job_select")
|
||||
if selected_job.startswith("."):
|
||||
|
||||
@@ -366,7 +366,7 @@ if __name__ == "__main__":
|
||||
# This prevents clear-text logging of sensitive information (CodeQL: py/clear-text-logging-sensitive-data)
|
||||
api_key_status = "API key required" # Constant string, not derived from provider.api_key
|
||||
masked_endpoint = provider.endpoint[:30] + "..." if len(provider.endpoint) > 30 else provider.endpoint
|
||||
print(f" {provider.priority}. {provider.name} ({api_key_status}) - {masked_endpoint}")
|
||||
print(f" {provider.priority}. {provider.name} ({api_key_status}) - {masked_endpoint}") # nosec B612 – no sensitive data logged
|
||||
|
||||
# Test 1: Health Check für alle Provider
|
||||
print("\n=== Test 1: Provider Health Check ===")
|
||||
|
||||
Reference in New Issue
Block a user