4b6817381b
CI (OpenClaw E2E) / openclaw test (push) Has been cancelled
CI / coverage-report (push) Has been cancelled
CI / test-kubernetes (push) Has been cancelled
CI / should-run-thorough (push) Has been cancelled
CI / test-thorough (cloudwatch-demo) (push) Has been cancelled
CI / test-thorough (flink-ecs) (push) Has been cancelled
CI / test-thorough (upstream-lambda) (push) Has been cancelled
CI / test-thorough (prefect-ecs-fargate) (push) Has been cancelled
Release / build-binaries (zip, opensre.exe, onefile, windows-latest, windows-x64) (push) Has been cancelled
Benchmark image — build + push to ECR (any adapter) / build + push (push) Has been cancelled
CI / quality (ubuntu-latest) (push) Has been cancelled
CI / test (tools-runtime) (push) Has been cancelled
CI / test (e2e-general) (push) Has been cancelled
CI / test (cli-runtime) (push) Has been cancelled
CI / test (e2e-provider-and-openclaw) (push) Has been cancelled
CI / test (integrations-and-misc) (push) Has been cancelled
Release / verify (push) Has been cancelled
Release / build-python-dist (push) Has been cancelled
Release / build-binaries (tar.gz, opensre, onedir, macos-15-intel, darwin-x64) (push) Has been cancelled
Release / build-binaries (tar.gz, opensre, onedir, macos-latest, darwin-arm64) (push) Has been cancelled
Release / build-binaries (tar.gz, opensre, onedir, ubuntu-22.04, linux-x64) (push) Has been cancelled
Release / publish-release (push) Has been cancelled
Release / publish-main-release (push) Has been cancelled
Interactive Shell Live (PR + post-merge) / turn-checks (no-LLM) (push) Has been cancelled
CodeQL / Analyze (python) (push) Has been cancelled
Interactive Shell Live (PR + post-merge) / turn-live shard ${{ matrix.shard_index }} (push) Has been cancelled
Release / prepare (push) Has been cancelled
Release / build-binaries (tar.gz, opensre, onedir, ubuntu-22.04-arm, linux-arm64) (push) Has been cancelled
Synthetic Deterministic Tests / Synthetic offline (deterministic) (push) Has been cancelled
160 lines
5.4 KiB
Python
160 lines
5.4 KiB
Python
"""Shell command parsing for the interactive REPL.
|
|
|
|
Alpha mode: no command-safety policy
|
|
------------------------------------
|
|
While OpenSRE is in alpha we run **every** command the user or action agent
|
|
asks for. There is intentionally no allowlist, no read-only / mutating /
|
|
restricted classification, and no deny floor — guardrails are deliberately
|
|
omitted to keep developer velocity high. See
|
|
``docs/interactive-shell-action-policy.md`` for the rationale.
|
|
|
|
This module's only job is to turn command text into a shape the runner can
|
|
execute:
|
|
|
|
* explicit ``!`` passthrough → run the remainder through a shell,
|
|
* commands using shell operators / substitution / heredocs → run through a shell,
|
|
* anything that fails to tokenize → hand the raw string to a shell,
|
|
* everything else → split into ``argv`` and run without a shell (which also lets
|
|
the runner detect the ``cd`` / ``pwd`` REPL builtins so the working directory
|
|
persists across turns).
|
|
|
|
The only non-execution outcome is a ``parse_error`` for genuinely empty input
|
|
(e.g. a bare ``!``). That is input validation, not a safety guardrail.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
import shlex
|
|
from dataclasses import dataclass
|
|
|
|
_EXPLICIT_SHELL_PREFIX = "!"
|
|
_SHELL_OPERATOR_RE = re.compile(r"(^|\s)(\|\||&&|[|;<>]|>>|<<|2>)(\s|$)")
|
|
_INLINE_SUBSHELL_RE = re.compile(r"`|\$\(")
|
|
# Heredoc starts such as ``<<'PY'`` or ``<<EOF`` — ``<<`` alone is already covered
|
|
# by ``_SHELL_OPERATOR_RE`` only when followed by whitespace; quoted/unquoted
|
|
# delimiters need an explicit match so ``python3 - <<'PY'`` is not tokenized.
|
|
_HEREDOC_START_RE = re.compile(r"(^|\s)<<-?\s*(?:'[^'\n]+'|\"[^\"\n]+\"|[^\s\\|;&<>]+)")
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ParsedShellCommand:
|
|
"""Structured command parsing result.
|
|
|
|
``use_shell`` is True when the command must run through a real shell (explicit
|
|
``!`` passthrough, shell operators / substitution, or input that could not be
|
|
tokenized). ``passthrough`` records only the explicit ``!`` prefix so the
|
|
runner can surface the "shell passthrough" hint for it.
|
|
"""
|
|
|
|
command: str
|
|
argv: list[str] | None
|
|
passthrough: bool
|
|
use_shell: bool
|
|
parse_error: str | None = None
|
|
|
|
|
|
def _split_argv(command: str, *, is_windows: bool) -> list[str] | None:
|
|
try:
|
|
return shlex.split(command, posix=not is_windows)
|
|
except ValueError:
|
|
try:
|
|
return shlex.split(command, posix=False)
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
def parse_shell_command(command: str, *, is_windows: bool) -> ParsedShellCommand:
|
|
"""Parse command text into an executable shape (no safety policy applied)."""
|
|
stripped = command.strip()
|
|
|
|
if stripped.startswith(_EXPLICIT_SHELL_PREFIX):
|
|
passthrough_command = stripped[len(_EXPLICIT_SHELL_PREFIX) :].strip()
|
|
if not passthrough_command:
|
|
return ParsedShellCommand(
|
|
command="",
|
|
argv=None,
|
|
passthrough=True,
|
|
use_shell=True,
|
|
parse_error="missing command after passthrough prefix (!).",
|
|
)
|
|
return ParsedShellCommand(
|
|
command=passthrough_command,
|
|
argv=None,
|
|
passthrough=True,
|
|
use_shell=True,
|
|
)
|
|
|
|
if (
|
|
_SHELL_OPERATOR_RE.search(stripped) is not None
|
|
or _INLINE_SUBSHELL_RE.search(stripped) is not None
|
|
or _HEREDOC_START_RE.search(stripped) is not None
|
|
):
|
|
# Operators / substitution need a real shell; alpha mode runs them.
|
|
return ParsedShellCommand(
|
|
command=stripped,
|
|
argv=None,
|
|
passthrough=False,
|
|
use_shell=True,
|
|
)
|
|
|
|
argv = _split_argv(stripped, is_windows=is_windows)
|
|
if argv is None:
|
|
# Could not tokenize (e.g. unbalanced quotes). Hand the raw string to the
|
|
# shell instead of blocking it.
|
|
return ParsedShellCommand(
|
|
command=stripped,
|
|
argv=None,
|
|
passthrough=False,
|
|
use_shell=True,
|
|
)
|
|
|
|
if not argv:
|
|
return ParsedShellCommand(
|
|
command=stripped,
|
|
argv=None,
|
|
passthrough=False,
|
|
use_shell=False,
|
|
parse_error="empty command.",
|
|
)
|
|
|
|
if is_windows:
|
|
|
|
def _strip_outer_quotes(value: str) -> str:
|
|
if len(value) >= 2 and value[0] == value[-1] and value[0] in {"'", '"'}:
|
|
return value[1:-1]
|
|
return value
|
|
|
|
argv = [_strip_outer_quotes(token) for token in argv]
|
|
|
|
return ParsedShellCommand(
|
|
command=stripped,
|
|
argv=argv,
|
|
passthrough=False,
|
|
use_shell=False,
|
|
)
|
|
|
|
|
|
def argv_for_repl_builtin_detection(
|
|
*, parsed: ParsedShellCommand, is_windows: bool
|
|
) -> list[str] | None:
|
|
"""Argv tokens for detecting ``cd`` / ``pwd`` REPL builtins.
|
|
|
|
Only the plain ``argv`` path and explicit ``!`` passthrough opt into builtin
|
|
detection. Operator / substitution commands run wholesale through the shell,
|
|
so they intentionally return ``None`` here (a leading ``cd`` in
|
|
``cd /tmp && ls`` must not be hijacked by the builtin handler).
|
|
"""
|
|
if parsed.argv is not None:
|
|
return parsed.argv
|
|
if not parsed.passthrough or not parsed.command.strip():
|
|
return None
|
|
return _split_argv(parsed.command.strip(), is_windows=is_windows)
|
|
|
|
|
|
__all__ = [
|
|
"ParsedShellCommand",
|
|
"argv_for_repl_builtin_detection",
|
|
"parse_shell_command",
|
|
]
|