restore changes

Author: RCFans

app/scenario_planning.py

@@ -33,21 +33,8 @@ def _normalize_pack_id(value: Any) -> str:
 def _resolve_workspace_path(candidate: Path) -> Path | None:
     try:
         workspace_root = WORKSPACE_ROOT.resolve()
-        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)
+        raw_path = candidate if candidate.is_absolute() else (workspace_root / candidate)
+        resolved = raw_path.resolve(strict=False)
         resolved.relative_to(workspace_root)
         return resolved
     except (OSError, RuntimeError, ValueError):
@@ -85,8 +72,10 @@ def _read_json_file(*, base_dir: Path, pack_id: str, filename: str) -> dict[str,
         if not safe_pack_id:
             return None
 
-        resolved_path = _resolve_workspace_path(base_path / safe_pack_id / filename)
-        if resolved_path is None:
+        resolved_path = (base_path / safe_pack_id / filename).resolve(strict=False)
+        try:
+            resolved_path.relative_to(base_path)
+        except ValueError:
             return None
 
         if not resolved_path.exists() or not resolved_path.is_file():
@@ -614,7 +603,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,7 +3,6 @@
 from __future__ import annotations
 
 import json
-import os
 from pathlib import Path
 from typing import Any
 
@@ -18,17 +17,6 @@ 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",
@@ -68,13 +56,9 @@ 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] | None:
+) -> dict[str, Any]:
     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")
@@ -85,8 +69,6 @@ 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")