File size: 5,401 Bytes
bb89bb6 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 |
from __future__ import annotations
import os
import platform
import shlex
import subprocess
from typing import Annotated
import gradio as gr
from app import _log_call_end, _log_call_start, _truncate_for_log
from ._docstrings import autodoc
from .File_System import _resolve_path, ROOT_DIR, _display_path
import shutil
def _detect_shell(prefer_powershell: bool = True) -> tuple[list[str], str]:
"""
Pick an appropriate shell for the host OS.
- Windows: use PowerShell by default, fall back to cmd.exe.
- POSIX: use /bin/bash if available, else /bin/sh.
Returns (shell_cmd_prefix, shell_name) where shell_cmd_prefix is the command list to launch the shell.
"""
system = platform.system().lower()
if system == "windows":
if prefer_powershell:
pwsh = shutil.which("pwsh")
candidates = [pwsh, shutil.which("powershell"), shutil.which("powershell.exe")]
for cand in candidates:
if cand:
return [cand, "-NoLogo", "-NoProfile", "-Command"], "powershell"
# Fallback to cmd
comspec = os.environ.get("ComSpec", r"C:\\Windows\\System32\\cmd.exe")
return [comspec, "/C"], "cmd"
# POSIX
bash = shutil.which("bash")
if bash:
return [bash, "-lc"], "bash"
sh = os.environ.get("SHELL", "/bin/sh")
return [sh, "-lc"], "sh"
# Detect shell at import time for docs/UI purposes
_DETECTED_SHELL_PREFIX, _DETECTED_SHELL_NAME = _detect_shell()
# Clarify path semantics and expose detected shell in summary
TOOL_SUMMARY = (
"Execute a shell command within a safe working directory under the tool root ('/'). "
"Paths must be relative to '/' (which maps to Nymbo-Tools/Filesystem by default); "
"absolute paths are disabled unless UNSAFE_ALLOW_ABS_PATHS=1. "
f"Detected shell: {_DETECTED_SHELL_NAME}."
)
def _run_command(command: str, cwd: str, timeout: int) -> tuple[str, str, int]:
shell_prefix, shell_name = _detect_shell()
full_cmd = shell_prefix + [command]
try:
proc = subprocess.run(
full_cmd,
cwd=cwd,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
encoding="utf-8",
errors="replace",
timeout=timeout if timeout and timeout > 0 else None,
)
return proc.stdout, proc.stderr, proc.returncode
except subprocess.TimeoutExpired as exc:
return exc.stdout or "", (exc.stderr or "") + "\n[timeout]", 124
except Exception as exc:
return "", f"Execution failed: {exc}", 1
@autodoc(summary=TOOL_SUMMARY)
def Shell_Exec(
command: Annotated[str, "Shell command to execute. Accepts multi-part pipelines as a single string."],
workdir: Annotated[str, "Working directory (relative to root unless UNSAFE_ALLOW_ABS_PATHS=1)."] = ".",
timeout: Annotated[int, "Timeout in seconds (0 = no timeout, be careful on public hosting)."] = 60,
) -> str:
_log_call_start("Shell_Exec", command=command, workdir=workdir, timeout=timeout)
if not command or not command.strip():
result = "No command provided."
_log_call_end("Shell_Exec", _truncate_for_log(result))
return result
abs_cwd, err = _resolve_path(workdir)
if err:
_log_call_end("Shell_Exec", _truncate_for_log(err))
return err
if not os.path.exists(abs_cwd):
result = f"Working directory not found: {abs_cwd}"
_log_call_end("Shell_Exec", _truncate_for_log(result))
return result
# Capture shell used for transparency
_, shell_name = _detect_shell()
stdout, stderr, code = _run_command(command, cwd=abs_cwd, timeout=timeout)
display_cwd = _display_path(abs_cwd)
header = (
f"Command: {command}\n"
f"CWD: {display_cwd}\n"
f"Root: /\n"
f"Shell: {shell_name}\n"
f"Exit code: {code}\n"
f"--- STDOUT ---\n"
)
output = header + (stdout or "<empty>") + "\n--- STDERR ---\n" + (stderr or "<empty>")
_log_call_end("Shell_Exec", _truncate_for_log(f"exit={code} stdout={len(stdout)} stderr={len(stderr)}"))
return output
def build_interface() -> gr.Interface:
return gr.Interface(
fn=Shell_Exec,
inputs=[
gr.Textbox(label="Command", placeholder="echo hello || dir", lines=2),
gr.Textbox(label="Workdir", value=".", max_lines=1),
gr.Slider(minimum=0, maximum=600, step=5, value=60, label="Timeout (seconds)"),
],
outputs=gr.Textbox(label="Output", lines=20),
title="Shell Exec",
description=(
"Run a shell command under the same safe root as File System. "
"Paths are relative to '/' (maps to Nymbo-Tools/Filesystem by default); absolute paths are disabled "
"unless UNSAFE_ALLOW_ABS_PATHS=1. Tip: set workdir to '.' to use the root. "
f"Detected shell: {_DETECTED_SHELL_NAME}. "
"<br><br><b>Examples</b>:"
"<br>- Command: <code>dir</code> (Windows) or <code>ls -la</code> (Linux)"
"<br>- Command: <code>echo hello | tee out.txt</code>"
"<br>- Workdir: <code>notes</code>, Command: <code>ls</code> to list /notes"
),
api_description=TOOL_SUMMARY,
flagging_mode="never",
submit_btn="Run",
)
__all__ = ["Shell_Exec", "build_interface"]
|