fix(skill): confine ROS bridge filesystem capabilities

Author: guima-why

skills/alicloud-ros-agent/scripts/ros_agent.py

@@ -138,49 +138,53 @@ def _json_bytes(value: Any) -> bytes:
     return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8")
 
 
+def _resolve_user_owned_path(raw_path: str, code: str, label: str) -> pathlib.Path:
+    """Resolve a local path and confine it to the user's home or temp tree."""
+
+    expanded = os.path.expandvars(os.path.expanduser(raw_path))
+    normalized = os.path.normcase(os.path.realpath(expanded))
+    allowed_roots = (
+        os.path.normcase(os.path.realpath(str(pathlib.Path.home()))),
+        os.path.normcase(os.path.realpath(tempfile.gettempdir())),
+    )
+    for allowed_root in allowed_roots:
+        prefix = allowed_root.rstrip(os.sep) + os.sep
+        if normalized.startswith(prefix):
+            return pathlib.Path(normalized)
+    raise BridgeError(code, "{} must be inside the current user's home or temporary directory.".format(label))
+
+
 def _state_root() -> pathlib.Path:
     configured = os.environ.get(STATE_DIR_ENV)
     if configured:
-        return pathlib.Path(os.path.expandvars(os.path.expanduser(configured))).resolve()
+        return _resolve_user_owned_path(configured, "invalid_config", "The ROS Agent state directory")
     return pathlib.Path(os.path.expanduser("~/.cache/alicloud-ros-agent")).resolve()
 
 
 def _secure_directory(path: pathlib.Path) -> None:
-    # The bridge only calls this with its process-local state root or a child
-    # path derived from a validated job identifier.
-    # codeql[py/path-injection]
     path.mkdir(parents=True, exist_ok=True)
     if os.name != "nt":
-        # codeql[py/path-injection]
         os.chmod(str(path), 0o700)
 
 
 def _atomic_json(path: pathlib.Path, value: Dict[str, Any], mode: int = 0o600) -> None:
     _secure_directory(path.parent)
-    # Atomic state files are always beneath the bridge-owned state directory.
-    # codeql[py/path-injection]
     descriptor, temporary = tempfile.mkstemp(prefix=path.name + ".", suffix=".tmp", dir=str(path.parent))
     try:
         with os.fdopen(descriptor, "w", encoding="utf-8") as handle:
             json.dump(value, handle, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
             handle.flush()
             os.fsync(handle.fileno())
         if os.name != "nt":
-            # codeql[py/path-injection]
             os.chmod(temporary, mode)
-        # codeql[py/path-injection]
         os.replace(temporary, str(path))
     finally:
         with contextlib.suppress(OSError):
-            # codeql[py/path-injection]
             os.unlink(temporary)
 
 
 def _load_state_json(path: pathlib.Path, code: str = "job_not_found") -> Dict[str, Any]:
     try:
-        # Callers pass only bridge state paths derived from validated local
-        # identifiers; remote StartChat payloads cannot select this path.
-        # codeql[py/path-injection]
         with path.open("r", encoding="utf-8") as handle:
             value = json.load(handle)
     except (OSError, ValueError) as exc:
@@ -451,24 +455,22 @@ def sanitize_text(value: Any, maximum: int = 4000, preserve_lines: bool = False)
 
 
 def _workspace(raw_path: Optional[str] = None) -> pathlib.Path:
-    # The optional value comes only from the authenticated loopback manager;
-    # existence and directory type are checked before it is used.
-    # codeql[py/path-injection]
-    path = pathlib.Path(raw_path or os.getcwd()).expanduser().resolve()
+    path = (
+        _resolve_user_owned_path(raw_path, "invalid_input", "The workspace")
+        if raw_path is not None
+        else pathlib.Path.cwd().resolve()
+    )
     if not path.is_dir():
         raise BridgeError("invalid_input", "The workspace must be an existing directory.")
     return path
 
 
 def _read_workspace_file(workspace: pathlib.Path, raw_path: str, maximum: int, label: str) -> str:
-    # The resolved path is rejected below unless it remains inside the already
-    # validated workspace. Constructing it alone performs no filesystem read.
-    # codeql[py/path-injection]
-    path = pathlib.Path(raw_path).expanduser().resolve()
-    try:
-        path.relative_to(workspace)
-    except ValueError as exc:
-        raise BridgeError("invalid_input", "{} must be inside the workspace.".format(label)) from exc
+    workspace_path = os.path.normcase(os.path.realpath(str(workspace)))
+    resolved_path = os.path.normcase(os.path.realpath(os.path.expanduser(raw_path)))
+    if not resolved_path.startswith(workspace_path.rstrip(os.sep) + os.sep):
+        raise BridgeError("invalid_input", "{} must be inside the workspace.".format(label))
+    path = pathlib.Path(resolved_path)
     try:
         data = path.read_bytes()
     except OSError as exc:
@@ -594,18 +596,10 @@ def build_permission_query(
 
 def resolve_aliyun(raw_path: str) -> str:
     expanded = os.path.expanduser(raw_path)
-    if os.path.dirname(expanded):
-        path = os.path.abspath(expanded)
-        # An explicit CLI path is local installation policy. It is checked as
-        # a regular file and later executed without a shell.
-        # codeql[py/path-injection]
-        if not os.path.isfile(path):
-            raise BridgeError("cli_not_found", "Alibaba Cloud CLI was not found at the requested path.")
-        return path
     resolved = shutil.which(expanded)
     if not resolved:
         raise BridgeError("cli_not_found", "Alibaba Cloud CLI is not installed or is not on PATH.")
-    return resolved
+    return os.path.abspath(resolved)
 
 
 def build_start_chat_parameters(
@@ -2407,11 +2401,8 @@ def _finish_job(
             if isinstance(job.get(key), str):
                 boundary[key] = job[key]
         data = _json_bytes(boundary) + b"\n"
-        # The spool belongs to a validated job under the bridge state root.
-        # codeql[py/path-injection]
         current_size = spool.stat().st_size if spool.exists() else 0
         if current_size + len(data) <= MAX_SPOOL_BYTES:
-            # codeql[py/path-injection]
             with spool.open("ab") as handle:
                 handle.write(data)
                 handle.flush()
@@ -2583,12 +2574,9 @@ def _fail_sideband_job(
 
 
 def _read_spool(spool: pathlib.Path) -> List[Dict[str, Any]]:
-    # Spool paths are produced only by _job_paths after job-id validation.
-    # codeql[py/path-injection]
     if not spool.exists():
         return []
     values = []
-    # codeql[py/path-injection]
     with spool.open("r", encoding="utf-8") as handle:
         for line in handle:
             try:
@@ -2631,18 +2619,14 @@ def _follow_timeout_result(job_id: str, start_cursor: int) -> Optional[Dict[str,
             "time": int(time.time()),
         }
         data = _json_bytes(marker) + b"\n"
-        # The spool belongs to a validated job under the bridge state root.
-        # codeql[py/path-injection]
         current_size = spool.stat().st_size if spool.exists() else 0
         if current_size + len(data) > MAX_SPOOL_BYTES:
             raise BridgeError("stream_failed", "The bounded ROS Agent event spool is full.")
-        # codeql[py/path-injection]
         with spool.open("ab") as handle:
             handle.write(data)
             handle.flush()
             os.fsync(handle.fileno())
         if os.name != "nt":
-            # codeql[py/path-injection]
             os.chmod(str(spool), 0o600)
         return _job_result(
             job_id,
@@ -3127,35 +3111,28 @@ def _stop_process(process: Any) -> None:
 
 def _spawn_worker(job_id: str, request: Dict[str, Any]) -> int:
     root, job_path, _spool = _job_paths(job_id)
-    request_path = root / ("request-{}.json".format(uuid.uuid4().hex))
+    request_token = uuid.uuid4().hex
+    request_path = root / ("request-{}.json".format(request_token))
     _atomic_json(request_path, request)
+    canonical_job_id = uuid.UUID(job_id).hex
     command = [
         sys.executable,
         str(pathlib.Path(__file__).resolve()),
         "_worker",
         "--job-id",
-        job_id,
-        "--request-file",
-        str(request_path),
+        canonical_job_id,
+        "--request-token",
+        request_token,
     ]
     log_path = root / "worker.log"
     creationflags = getattr(subprocess, "CREATE_NEW_PROCESS_GROUP", 0) if os.name == "nt" else 0
-    # Worker paths are generated beneath a validated job directory and the
-    # launched argv fixes both the interpreter and this bridge script.
-    # codeql[py/path-injection]
     if log_path.exists() and log_path.stat().st_size > MAX_DIAGNOSTIC_BYTES:
-        # codeql[py/path-injection]
         with log_path.open("wb"):
             pass
     try:
-        # codeql[py/path-injection]
         with log_path.open("ab", buffering=0) as log:
             if os.name != "nt":
-                # codeql[py/path-injection]
                 os.chmod(str(log_path), 0o600)
-            # command is a fixed interpreter/script pair plus a validated job
-            # id and a bridge-generated request filename; shell=False is used.
-            # codeql[py/command-line-injection]
             process = subprocess.Popen(
                 command,
                 stdin=subprocess.DEVNULL,
@@ -3166,7 +3143,6 @@ def _spawn_worker(job_id: str, request: Dict[str, Any]) -> int:
             )
     except OSError as exc:
         with contextlib.suppress(OSError):
-            # codeql[py/path-injection]
             request_path.unlink()
         request_seq = int(request.get("requestSeq") or 0)
         error = BridgeError("worker_start_failed", "The StartChat worker could not be started.", True)
@@ -3631,8 +3607,16 @@ def _cancel_job_local(payload: Dict[str, Any]) -> Dict[str, Any]:
     return result
 
 
-def run_worker(job_id: str, request_file: str) -> int:
-    request_path = pathlib.Path(request_file).resolve()
+def run_worker(job_id: str, request_token: str) -> int:
+    try:
+        canonical_job_id = uuid.UUID(job_id).hex
+        canonical_request_token = uuid.UUID(request_token).hex
+    except (AttributeError, ValueError) as exc:
+        raise BridgeError("invalid_input", "The worker launch capability is invalid.") from exc
+    if canonical_job_id != job_id or canonical_request_token != request_token:
+        raise BridgeError("invalid_input", "The worker launch capability is invalid.")
+    root, _job_path, _spool = _job_paths(canonical_job_id)
+    request_path = root / ("request-{}.json".format(canonical_request_token))
     request = _load_state_json(request_path, "invalid_input")
     with contextlib.suppress(OSError):
         request_path.unlink()
@@ -4262,7 +4246,7 @@ def build_parser() -> argparse.ArgumentParser:
     server.add_argument("--record-file", required=True)
     worker = subparsers.add_parser("_worker", help=argparse.SUPPRESS)
     worker.add_argument("--job-id", required=True)
-    worker.add_argument("--request-file", required=True)
+    worker.add_argument("--request-token", required=True)
     return parser
 
 
@@ -4276,7 +4260,7 @@ def main(argv: Optional[List[str]] = None) -> int:
         if args.command == "_server":
             return run_manager_server(args.record_file)
         if args.command == "_worker":
-            return run_worker(args.job_id, args.request_file)
+            return run_worker(args.job_id, args.request_token)
         apply_skill_config(args, load_skill_config())
         if args.command == "check":
             result = run_check(args)