fix CodeQL alert

Author: RCFans

app/scenario_planning.py

@@ -33,8 +33,21 @@ def _normalize_pack_id(value: Any) -> str:
 def _resolve_workspace_path(candidate: Path) -> Path | None:
     try:
         workspace_root = WORKSPACE_ROOT.resolve()
-        raw_path = candidate if candidate.is_absolute() else (workspace_root / candidate)
-        resolved = raw_path.resolve(strict=False)
+        base_str = os.path.abspath(str(workspace_root))
+        if not base_str.endswith(os.sep):
+            base_str += os.sep
+
+        raw_str = str(candidate)
+        if os.path.isabs(raw_str):
+            target_str = os.path.abspath(raw_str)
+        else:
+            target_str = os.path.abspath(os.path.join(base_str, raw_str))
+
+        if not target_str.startswith(base_str) and target_str != base_str.rstrip(os.sep):
+            return None
+
+        # Fully contained lexically, now safe to resolve symlinks
+        resolved = Path(target_str).resolve(strict=False)
         resolved.relative_to(workspace_root)
         return resolved
     except (OSError, RuntimeError, ValueError):
@@ -72,10 +85,8 @@ def _read_json_file(*, base_dir: Path, pack_id: str, filename: str) -> dict[str,
         if not safe_pack_id:
             return None
 
-        resolved_path = (base_path / safe_pack_id / filename).resolve(strict=False)
-        try:
-            resolved_path.relative_to(base_path)
-        except ValueError:
+        resolved_path = _resolve_workspace_path(base_path / safe_pack_id / filename)
+        if resolved_path is None:
             return None
 
         if not resolved_path.exists() or not resolved_path.is_file():
@@ -603,7 +614,7 @@ def _render_flow_fallback(payloads: dict[str, Any]) -> None:
     data_root=data_root,
     output_root=output_root,
     output_pack_id=_normalize_pack_id(output_pack_override.strip()) or None,
-)
+) or {}
 paths = dict(bundle.get("paths") or {})
 payloads = dict(bundle.get("payloads") or {})
 slot_labels = _load_scenario_slot_labels()

src/omen/scenario/loader.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import json
+import os
 from pathlib import Path
 from typing import Any
 
@@ -17,6 +18,17 @@ def _read_json(path: Path) -> dict[str, Any] | None:
     return payload if isinstance(payload, dict) else None
 
 
+def _is_safe_subpath(base: Path, subpath: str) -> bool:
+    """Validate that the given subpath resolves strictly within the base directory."""
+    if not subpath:
+        return False
+    base_str = os.path.abspath(str(base))
+    if not base_str.endswith(os.sep):
+        base_str += os.sep
+    target_str = os.path.abspath(os.path.join(base_str, str(subpath)))
+    return target_str.startswith(base_str)
+
+
 def discover_spec8_pack_candidates(
     data_root: str | Path = "data/scenarios",
     output_root: str | Path = "output",
@@ -56,9 +68,13 @@ def load_spec8_flow_artifacts(
     data_root: str | Path = "data/scenarios",
     output_root: str | Path = "output",
     output_pack_id: str | None = None,
-) -> dict[str, Any]:
+) -> dict[str, Any] | None:
     data_base = Path(data_root)
     output_base = Path(output_root)
+    
+    if not _is_safe_subpath(data_base, pack_id):
+        return None
+        
     data_pack = data_base / pack_id
 
     situation = _read_json(data_pack / "situation.json")
@@ -69,6 +85,8 @@ def load_spec8_flow_artifacts(
     generation_trace = _read_json(data_pack / "generation" / "log.json")
 
     resolved_output_pack = output_pack_id or pack_id
+    if not _is_safe_subpath(output_base, resolved_output_pack):
+        return None
     result = _read_json(output_base / resolved_output_pack / "result.json")
     explanation = _read_json(output_base / resolved_output_pack / "explanation.json")