diff --git a/pyproject.toml b/pyproject.toml index bdfecd610..6909b838f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -642,9 +642,8 @@ session = [ # Sh Module - Shell utilities # Use: pip install scitex[sh] -sh = [ - "matplotlib", -] +# Real implementation lives in the standalone scitex-sh package. +sh = ["scitex-sh>=0.1.0"] # Social Module - Social media integration # Use: pip install scitex[social] diff --git a/src/scitex/sh/README.md b/src/scitex/sh/README.md deleted file mode 100644 index a23e2ca8e..000000000 --- a/src/scitex/sh/README.md +++ /dev/null @@ -1,58 +0,0 @@ - - -# scitex.sh - Shell Command Execution Module - -Safe shell command execution with injection protection. - -## Structure - -- `__init__.py` - Main interface (sh, sh_run functions) -- `_execute.py` - Core execution logic -- `_security.py` - Security validation and quoting -- `_types.py` - Type definitions -- `test_sh_simple.py` - Test file - -## Usage - -```python -import scitex - -# List format (ONLY format allowed - safe by design) -result = scitex.sh(["ls", "-la", "/home"]) - -# With user input (safe - arguments are literal) -user_input = "../malicious; rm -rf /" -result = scitex.sh(["cat", user_input]) - -# For filtering (instead of pipes) -result = scitex.sh(["ls", "-la"]) -py_files = [l for l in result['stdout'].split('\n') if '.py' in l] - -# sh_run always returns dict -result = scitex.sh_run(["echo", "test"]) -if result['success']: - print(result['stdout']) -``` - -## Security Features - -1. List format only - String commands are rejected with TypeError -2. shell=False always - No shell interpretation of special characters -3. Null byte validation - Blocks common injection vector -4. Arguments as literals - ;, |, & are treated as literal text - -## Backward Compatibility - -Old imports still work: - -```python -from scitex.sh import sh, sh_run, quote -``` - -# EOF - - diff --git a/src/scitex/sh/__init__.py b/src/scitex/sh/__init__.py index 4d8dcd91b..fde3d615d 100755 --- a/src/scitex/sh/__init__.py +++ b/src/scitex/sh/__init__.py @@ -1,96 +1,20 @@ -#!/usr/bin/env python3 -from __future__ import annotations +"""SciTeX sh — thin compatibility shim for scitex-sh. -import os +Aliases ``scitex.sh`` to the standalone ``scitex_sh`` package via ``sys.modules``. +``scitex.sh is scitex_sh``. -__FILE__ = __file__ -__DIR__ = os.path.dirname(__FILE__) +Install: ``pip install scitex[sh]`` (or ``pip install scitex-sh``). +See: https://github.com/ywatanabe1989/scitex-sh +""" -from typing import Union +import sys as _sys -from ._execute import execute -from ._security import quote, validate_command -from ._types import CommandInput, ReturnFormat, ShellResult +try: + import scitex_sh as _real +except ImportError as _e: # pragma: no cover + raise ImportError( + "scitex.sh requires the 'scitex-sh' package. " + "Install with: pip install scitex[sh] (or: pip install scitex-sh)" + ) from _e - -def sh( - command_str_or_list: CommandInput, - verbose: bool = True, - return_as: ReturnFormat = "dict", - timeout: int = None, - stream_output: bool = False, -) -> Union[str, ShellResult]: - """ - Executes a shell command safely (list format only). - - Parameters: - - command_str_or_list: Command to execute (MUST be list format) - - verbose: Whether to print command and output - - return_as: Return format ("dict" or "str") - - timeout: Timeout in seconds (None for no timeout) - - stream_output: Whether to stream output in real-time (default: False) - - Returns: - - If return_as="str": output string - - If return_as="dict": ShellResult dict - - Security Notes: - - Only list format is allowed to prevent shell injection - - Each argument is treated as a literal string - - For pipes/redirects, use Python subprocess chaining - - Examples: - -------- - >>> from scitex.sh import sh - >>> sh(["ls", "-la", "/home"]) - >>> sh(["git", "status"]) - >>> sh(["sleep", "10"], timeout=5) # Will timeout after 5 seconds - >>> sh(["./compile.sh"], stream_output=True) # Stream output in real-time - >>> - >>> # For grep-like filtering, use Python: - >>> result = sh(["ls", "-la"]) - >>> filtered = [l for l in result['stdout'].split('\\n') if '.py' in l] - """ - result = execute( - command_str_or_list, - verbose=verbose, - timeout=timeout, - stream_output=stream_output, - ) - - if return_as == "dict": - return result - else: - if result["success"]: - return result["stdout"] - else: - return result["stderr"] - - -def sh_run(command: CommandInput, verbose: bool = True) -> ShellResult: - """ - Executes a shell command and returns detailed results. - - Parameters: - - command: Command to execute (MUST be list format) - - verbose: Whether to print command and output - - Returns: - - ShellResult dict with stdout, stderr, exit_code, success - - Examples: - -------- - >>> from scitex.sh import sh_run - >>> result = sh_run(["ls", "-la"]) - >>> if result['success']: - ... print(result['stdout']) - """ - return execute(command, verbose=verbose) - - -# Legacy functions moved from gen module -from ._shell_legacy import run_shellcommand, run_shellscript - -__all__ = ["sh", "sh_run", "quote", "run_shellcommand", "run_shellscript"] - -# EOF +_sys.modules[__name__] = _real diff --git a/src/scitex/sh/_execute.py b/src/scitex/sh/_execute.py deleted file mode 100755 index 2d3b6b5f6..000000000 --- a/src/scitex/sh/_execute.py +++ /dev/null @@ -1,225 +0,0 @@ -#!/usr/bin/env python3 -# -*- coding: utf-8 -*- -# Timestamp: "2025-10-29 07:23:56 (ywatanabe)" -# File: /home/ywatanabe/proj/scitex-code/src/scitex/sh/_execute.py -# ---------------------------------------- -from __future__ import annotations - -import os - -__FILE__ = "./src/scitex/sh/_execute.py" -__DIR__ = os.path.dirname(__FILE__) -# ---------------------------------------- - -__FILE__ = __file__ - -import select -import subprocess -import sys -import time - -import scitex - -from ._security import validate_command -from ._types import CommandInput, ShellResult - - -def execute( - command_str_or_list: CommandInput, - verbose: bool = True, - timeout: int = None, - stream_output: bool = False, -) -> ShellResult: - """ - Executes a shell command safely (list format only). - - Parameters: - - command_str_or_list: Command to execute (must be list format) - - verbose: Whether to print command and output - - timeout: Timeout in seconds (None for no timeout) - - stream_output: Whether to stream output in real-time (default: False) - When True, prints output as it's generated instead of waiting - for command completion - - Returns: - - ShellResult dict with stdout, stderr, exit_code, success - - Raises: - - TypeError: If command is a string (not allowed for security) - - subprocess.TimeoutExpired: If command exceeds timeout - - Examples: - - sh(['ls', '-la']) - - sh(['git', 'status']) - - sh(['pdflatex', '-interaction=nonstopmode', 'file.tex'], stream_output=True) - """ - validate_command(command_str_or_list) - - if verbose: - cmd_display = " ".join(command_str_or_list) - print(scitex.str.color_text(f"{cmd_display}", "yellow")) - - if stream_output: - # Use real-time streaming mode - return _execute_with_streaming(command_str_or_list, verbose, timeout) - else: - # Use buffered mode (original behavior) - return _execute_buffered(command_str_or_list, verbose, timeout) - - -def _execute_buffered( - command_str_or_list: CommandInput, verbose: bool, timeout: int -) -> ShellResult: - """Execute command with buffered output (original behavior).""" - process = subprocess.Popen( - command_str_or_list, - shell=False, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - ) - - try: - stdout_bytes, stderr_bytes = process.communicate(timeout=timeout) - except subprocess.TimeoutExpired: - process.kill() - stdout_bytes, stderr_bytes = process.communicate() - timeout_msg = f"Command timed out after {timeout} seconds" - stderr_bytes = stderr_bytes + b"\n" + timeout_msg.encode("utf-8") - - stdout = stdout_bytes.decode("utf-8").strip() - stderr = stderr_bytes.decode("utf-8").strip() - exit_code = process.returncode - - result: ShellResult = { - "stdout": stdout, - "stderr": stderr, - "exit_code": exit_code, - "success": exit_code == 0, - } - - if verbose: - if stdout: - print(stdout) - if stderr: - print(scitex.str.color_text(stderr, "red")) - - return result - - -def _execute_with_streaming( - command_str_or_list: CommandInput, verbose: bool, timeout: int -) -> ShellResult: - """Execute command with real-time output streaming using select.""" - import io - - # Set PYTHONUNBUFFERED for Python scripts and unbuffered mode for shell - env = os.environ.copy() - env["PYTHONUNBUFFERED"] = "1" - - process = subprocess.Popen( - command_str_or_list, - shell=False, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - bufsize=0, # Unbuffered - env=env, - ) - - stdout_data = [] - stderr_data = [] - start_time = time.time() - - # Use non-blocking reads - import fcntl - - def make_non_blocking(fd): - flags = fcntl.fcntl(fd, fcntl.F_GETFL) - fcntl.fcntl(fd, fcntl.F_SETFL, flags | os.O_NONBLOCK) - - make_non_blocking(process.stdout) - make_non_blocking(process.stderr) - - try: - while True: - # Check timeout - if timeout and (time.time() - start_time) > timeout: - process.kill() - timeout_msg = f"Command timed out after {timeout} seconds" - if verbose: - print(scitex.str.color_text(timeout_msg, "red"), flush=True) - stderr_data.append(timeout_msg.encode()) - break - - # Check if process has finished - poll_result = process.poll() - - # Read available data from stdout - try: - chunk = process.stdout.read() - if chunk: - stdout_data.append(chunk) - if verbose: - text = chunk.decode("utf-8", errors="replace") - print(text, end="", flush=True) - except (IOError, BlockingIOError): - pass - - # Read available data from stderr - try: - chunk = process.stderr.read() - if chunk: - stderr_data.append(chunk) - if verbose: - text = chunk.decode("utf-8", errors="replace") - print(scitex.str.color_text(text, "red"), end="", flush=True) - except (IOError, BlockingIOError): - pass - - # If process finished, do final read and break - if poll_result is not None: - # Final read to catch any remaining buffered output - try: - chunk = process.stdout.read() - if chunk: - stdout_data.append(chunk) - if verbose: - text = chunk.decode("utf-8", errors="replace") - print(text, end="", flush=True) - except (IOError, BlockingIOError): - pass - - try: - chunk = process.stderr.read() - if chunk: - stderr_data.append(chunk) - if verbose: - text = chunk.decode("utf-8", errors="replace") - print( - scitex.str.color_text(text, "red"), end="", flush=True - ) - except (IOError, BlockingIOError): - pass - break - - # Small sleep to prevent CPU spinning - time.sleep(0.05) - - except Exception as e: - process.kill() - raise - - stdout = b"".join(stdout_data).decode("utf-8", errors="replace").strip() - stderr = b"".join(stderr_data).decode("utf-8", errors="replace").strip() - exit_code = process.returncode - - result: ShellResult = { - "stdout": stdout, - "stderr": stderr, - "exit_code": exit_code, - "success": exit_code == 0, - } - - return result - - -# EOF diff --git a/src/scitex/sh/_security.py b/src/scitex/sh/_security.py deleted file mode 100755 index e0b1a0ba8..000000000 --- a/src/scitex/sh/_security.py +++ /dev/null @@ -1,66 +0,0 @@ -#!/usr/bin/env python3 -# -*- coding: utf-8 -*- -# Timestamp: "2025-10-29 07:23:58 (ywatanabe)" -# File: /home/ywatanabe/proj/scitex-code/src/scitex/sh/_security.py -# ---------------------------------------- -from __future__ import annotations - -import os - -__FILE__ = "./src/scitex/sh/_security.py" -__DIR__ = os.path.dirname(__FILE__) -# ---------------------------------------- - -__FILE__ = __file__ - -import shlex -from typing import List, Union - -DANGEROUS_CHARS = [";", "|", "&", "$", "`", "\n", ">", "<", "(", ")", "{", "}"] - - -def validate_command(command_str_or_list: Union[str, List[str]]) -> None: - """ - Validates command for security issues. - - Parameters: - - command_str_or_list: Command string or list to validate - - Raises: - - TypeError: If command is a string (not allowed for security) - - ValueError: If command contains dangerous characters - """ - if isinstance(command_str_or_list, str): - raise TypeError( - "String commands are not allowed for security reasons. " - "Use list format: ['command', 'arg1', 'arg2']. " - "For pipes and redirects, use Python subprocess chaining instead." - ) - - for arg in command_str_or_list: - if "\0" in str(arg): - raise ValueError( - "Command argument contains null byte - potential shell injection attempt" - ) - - -def quote(arg: str) -> str: - """ - Safely quotes a string for use in shell commands. - - Parameters: - - arg: The argument to quote - - Returns: - - str: Safely quoted string - - Examples: - -------- - >>> filename = "file; rm -rf /" - >>> from scitex.sh import sh, quote - >>> sh(f"cat {quote(filename)}") - """ - return shlex.quote(arg) - - -# EOF diff --git a/src/scitex/sh/_shell_legacy.py b/src/scitex/sh/_shell_legacy.py deleted file mode 100755 index 44c252a45..000000000 --- a/src/scitex/sh/_shell_legacy.py +++ /dev/null @@ -1,48 +0,0 @@ -#!/usr/bin/env python3 -# -*- coding: utf-8 -*- -# Time-stamp: "2024-01-29 07:36:39 (ywatanabe)" - -import os -import subprocess - - -def run_shellscript(lpath_sh, *args): - # Check if the script is executable, if not, make it executable - if not os.access(lpath_sh, os.X_OK): - subprocess.run(["chmod", "+x", lpath_sh]) - - # Prepare the command with script path and arguments - command = [lpath_sh] + list(args) - - # Run the shell script with arguments using run_shellcommand - return run_shellcommand(*command) - # return stdout, stderr, exit_code - - -def run_shellcommand(command, *args): - # Prepare the command with additional arguments - full_command = [command] + list(args) - - # Run the command - result = subprocess.run( - full_command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True - ) - - # Get the standard output and error - stdout = result.stdout - stderr = result.stderr - exit_code = result.returncode - - # Check if the command ran successfully - if exit_code == 0: - print("Command executed successfully") - print("Output:", stdout) - else: - print("Command failed with error code:", exit_code) - print("Error:", stderr) - - return { - "stdout": stdout, - "stderr": stderr, - "exit_code": exit_code, - } diff --git a/src/scitex/sh/_skills/SKILL.md b/src/scitex/sh/_skills/SKILL.md deleted file mode 100644 index 4c9a6ceec..000000000 --- a/src/scitex/sh/_skills/SKILL.md +++ /dev/null @@ -1,70 +0,0 @@ ---- -name: stx.sh -description: Safe shell command execution. Enforces list-format commands to prevent shell injection. Supports timeout, real-time streaming, and flexible return formats. -user-invocable: false ---- - -# stx.sh — Shell Command Execution - -`stx.sh` executes subprocesses safely. String commands are rejected by design; only list format is accepted. All execution goes through `subprocess.Popen(shell=False)`. - -## Quick Reference - -```python -import scitex as stx - -# Basic — returns ShellResult dict -result = stx.sh.sh(["ls", "-la"]) -print(result["stdout"]) -print(result["exit_code"]) # 0 on success -print(result["success"]) # True/False - -# Return stdout as plain string -output = stx.sh.sh(["echo", "hello"], return_as="str") - -# Timeout after 5 seconds -result = stx.sh.sh(["sleep", "30"], timeout=5) - -# Stream output live (e.g., long builds) -result = stx.sh.sh(["make", "all"], stream_output=True) - -# Silent execution -result = stx.sh.sh(["git", "status"], verbose=False) - -# Always-dict convenience wrapper -result = stx.sh.sh_run(["git", "log", "--oneline", "-5"]) -``` - -## Sub-skills - -### Execution -- [execution.md](execution.md) — `sh()` and `sh_run()`: parameters, return values, buffered vs streaming modes, timeout behavior - -### Security -- [security.md](security.md) — injection model, `validate_command()`, `quote()`: why strings are rejected, what is validated, when to use quoting - -### Types -- [types.md](types.md) — `ShellResult` TypedDict, `CommandInput`, `ReturnFormat`: field meanings and edge cases - -### Legacy -- [legacy.md](legacy.md) — `run_shellcommand()`, `run_shellscript()`: backward-compat functions from the gen module, differences from `sh()`, migration guide - -## Public API - -| Symbol | Source | Description | -|--------|--------|-------------| -| `sh` | `__init__` | Execute command; returns dict or str | -| `sh_run` | `__init__` | Execute command; always returns `ShellResult` dict | -| `quote` | `_security` | Shell-quote a single argument via `shlex.quote` | -| `validate_command` | `_security` | Pre-flight security check (also called internally) | -| `run_shellcommand` | `_shell_legacy` | Legacy: run command + args | -| `run_shellscript` | `_shell_legacy` | Legacy: run a shell script, auto-chmod if needed | -| `ShellResult` | `_types` | TypedDict with stdout, stderr, exit_code, success | -| `CommandInput` | `_types` | Type alias for `List[str]` | -| `ReturnFormat` | `_types` | `Literal["dict", "str"]` | - -## Key Constraints - -- **No string commands.** `sh("ls -la")` raises `TypeError`. Pass `["ls", "-la"]`. -- **No pipes/redirects in command list.** Filter in Python instead: `[l for l in result["stdout"].split("\n") if ".py" in l]` -- **`exit_code`, not `returncode`.** The result key is `exit_code`. (`returncode` does not exist on `ShellResult`.) diff --git a/src/scitex/sh/_skills/execution.md b/src/scitex/sh/_skills/execution.md deleted file mode 100644 index 4674c6a4e..000000000 --- a/src/scitex/sh/_skills/execution.md +++ /dev/null @@ -1,142 +0,0 @@ ---- -description: Core shell command execution — sh() and sh_run() — including buffered and streaming modes, timeout, and return format control. ---- - -# Shell Command Execution - -## sh - -Execute a shell command and return either a dict or a plain string. - -```python -sh( - command_str_or_list: List[str], - verbose: bool = True, - return_as: Literal["dict", "str"] = "dict", - timeout: int = None, - stream_output: bool = False, -) -> Union[ShellResult, str] -``` - -**Parameters** - -| Parameter | Type | Default | Description | -|-----------|------|---------|-------------| -| `command_str_or_list` | `List[str]` | required | Command as a list of strings. String input raises `TypeError`. | -| `verbose` | `bool` | `True` | Print the command (yellow) before running. Print stdout/stderr after. | -| `return_as` | `"dict"` or `"str"` | `"dict"` | `"dict"` returns a `ShellResult` dict. `"str"` returns stdout on success, stderr on failure. | -| `timeout` | `int` or `None` | `None` | Kill the process and append a timeout message to stderr after N seconds. | -| `stream_output` | `bool` | `False` | When `True`, print output line-by-line as it is produced (real-time). When `False`, buffer and print after completion. | - -**Returns** - -- `return_as="dict"` — a `ShellResult` TypedDict: `{"stdout": str, "stderr": str, "exit_code": int, "success": bool}` -- `return_as="str"` — stdout string on success; stderr string on failure - -**Raises** - -- `TypeError` — if `command_str_or_list` is a plain string -- `ValueError` — if any argument contains a null byte (`\0`) - -**Examples** - -```python -import scitex as stx - -# Basic usage -result = stx.sh.sh(["ls", "-la", "/home"]) -print(result["stdout"]) -print(result["exit_code"]) # 0 on success - -# Check success before using output -result = stx.sh.sh(["git", "status"]) -if result["success"]: - print(result["stdout"]) - -# Return stdout as a plain string -output = stx.sh.sh(["echo", "hello"], return_as="str") -# output == "hello" - -# Kill after 5 seconds if not done -result = stx.sh.sh(["sleep", "30"], timeout=5) -# result["success"] == False -# result["stderr"] contains "Command timed out after 5 seconds" - -# Stream long-running process output live (e.g., pdflatex) -result = stx.sh.sh(["pdflatex", "-interaction=nonstopmode", "paper.tex"], - stream_output=True) - -# Silent execution (no prints) -result = stx.sh.sh(["cat", "/etc/hostname"], verbose=False) -``` - ---- - -## sh_run - -Convenience wrapper that always returns a `ShellResult` dict. Identical to `sh(..., return_as="dict")`. - -```python -sh_run( - command: List[str], - verbose: bool = True, -) -> ShellResult -``` - -**Parameters** - -| Parameter | Type | Default | Description | -|-----------|------|---------|-------------| -| `command` | `List[str]` | required | Command as a list of strings. | -| `verbose` | `bool` | `True` | Print command and output. | - -**Returns** `ShellResult` dict: `{"stdout": str, "stderr": str, "exit_code": int, "success": bool}` - -**Examples** - -```python -from scitex.sh import sh_run - -result = sh_run(["ls", "-la"]) -if result["success"]: - print(result["stdout"]) -else: - print("Failed:", result["stderr"]) - -# Suppress output -result = sh_run(["cat", "/nonexistent/file"], verbose=False) -print(result["success"]) # False -print(result["exit_code"]) # non-zero -print(result["stderr"]) # error message from cat -``` - ---- - -## Output Modes: Buffered vs Streaming - -By default (`stream_output=False`), output is **buffered**: the subprocess runs to completion, then stdout/stderr are decoded and printed together. - -With `stream_output=True`, the module uses non-blocking file descriptors and polls the process every 50 ms. Each chunk of output is printed immediately as it arrives. This is useful for long-running commands where you want live feedback. - -**Streaming sets `PYTHONUNBUFFERED=1`** in the child environment so Python scripts invoked as subprocesses also emit output without internal buffering. - -```python -# Buffered: waits for process to finish, then prints everything -result = stx.sh.sh(["make", "all"]) - -# Streaming: prints each line as produced -result = stx.sh.sh(["make", "all"], stream_output=True) -``` - -Both modes return the same `ShellResult` dict. - ---- - -## Filtering Output in Python (instead of pipes) - -Because string commands with pipes (`|`) are rejected, use Python to filter: - -```python -result = stx.sh.sh(["ls", "-la"]) -py_files = [line for line in result["stdout"].split("\n") if ".py" in line] -``` diff --git a/src/scitex/sh/_skills/legacy.md b/src/scitex/sh/_skills/legacy.md deleted file mode 100644 index 5868f005a..000000000 --- a/src/scitex/sh/_skills/legacy.md +++ /dev/null @@ -1,102 +0,0 @@ ---- -description: Legacy shell helpers (run_shellcommand, run_shellscript) carried over from the gen module for backward compatibility. ---- - -# Legacy Functions - -These functions were moved from the `gen` module and are preserved for backward compatibility. New code should prefer `sh()` or `sh_run()`. - ---- - -## run_shellcommand - -Run a command with positional arguments. Uses `subprocess.run` directly. - -```python -run_shellcommand(command: str, *args: str) -> dict -``` - -**Parameters** - -| Parameter | Type | Description | -|-----------|------|-------------| -| `command` | `str` | The executable name or path | -| `*args` | `str` | Additional positional arguments | - -**Returns** a plain dict (not `ShellResult`): - -```python -{ - "stdout": str, # raw stdout text (NOT stripped) - "stderr": str, # raw stderr text - "exit_code": int, -} -``` - -**Differences from sh()** - -- Does NOT validate against string injection (no `validate_command` call) -- Always prints success/failure message to stdout (not suppressible via `verbose`) -- Returns a plain `dict` — no `success` key -- stdout/stderr are NOT stripped - -**Example** - -```python -from scitex.sh import run_shellcommand - -result = run_shellcommand("ls", "-la") -print(result["stdout"]) -``` - ---- - -## run_shellscript - -Execute a shell script file, auto-granting execute permission if needed. - -```python -run_shellscript(lpath_sh: str, *args: str) -> dict -``` - -**Parameters** - -| Parameter | Type | Description | -|-----------|------|-------------| -| `lpath_sh` | `str` | Path to the shell script file | -| `*args` | `str` | Arguments forwarded to the script | - -**Behavior** - -1. Checks if the file is executable with `os.access(lpath_sh, os.X_OK)`. -2. If not executable, runs `chmod +x ` via `subprocess.run`. -3. Runs `[lpath_sh] + list(args)` via `run_shellcommand`. - -**Returns** the same dict as `run_shellcommand`. - -**Example** - -```python -from scitex.sh import run_shellscript - -# Script does not need to be pre-chmod'd -result = run_shellscript("./scripts/build.sh", "--release") -``` - ---- - -## Migration to sh() - -```python -# Legacy -run_shellcommand("git", "status") - -# Modern equivalent -stx.sh.sh(["git", "status"]) - -# Legacy -run_shellscript("./build.sh", "--release") - -# Modern equivalent -stx.sh.sh(["./build.sh", "--release"]) -``` diff --git a/src/scitex/sh/_skills/security.md b/src/scitex/sh/_skills/security.md deleted file mode 100644 index 8f6cdda1c..000000000 --- a/src/scitex/sh/_skills/security.md +++ /dev/null @@ -1,102 +0,0 @@ ---- -description: Shell injection prevention model — why string commands are rejected, what validate_command checks, and how quote() safely escapes arguments. ---- - -# Security Model - -## Design Principle - -All commands must be passed as a **list of strings**, never as a plain string. This mirrors `subprocess.Popen(shell=False)` directly: each list element is passed as a literal argument to `execvp`, so shell metacharacters (`;`, `|`, `&`, `$`, backticks, redirects) are never interpreted. - -```python -# WRONG — raises TypeError immediately -stx.sh.sh("ls -la | grep .py") - -# CORRECT — each token is a separate element -stx.sh.sh(["ls", "-la"]) -filtered = [l for l in result["stdout"].split("\n") if ".py" in l] -``` - ---- - -## validate_command - -Called automatically before every execution. Also callable directly for pre-flight checks. - -```python -validate_command(command_str_or_list: Union[str, List[str]]) -> None -``` - -**Checks performed** - -| Check | Raises | Reason | -|-------|--------|--------| -| Input is a `str` | `TypeError` | String commands pass through the shell; list format does not | -| Any argument contains `\0` | `ValueError` | Null bytes are a common shell injection vector | - -**Examples** - -```python -from scitex.sh import validate_command - -# Passes silently -validate_command(["git", "commit", "-m", "Add feature"]) - -# Raises TypeError -try: - validate_command("git commit -m 'Add feature'") -except TypeError as e: - print(e) -# String commands are not allowed for security reasons. -# Use list format: ['command', 'arg1', 'arg2']. - -# Raises ValueError -try: - validate_command(["echo", "test\0injected"]) -except ValueError as e: - print(e) -# Command argument contains null byte - potential shell injection attempt -``` - -**Note on dangerous characters:** Characters like `;`, `|`, `&`, `$` are NOT explicitly blocked in list arguments because they are harmless when passed as literals to a non-shell process. The real protection is `shell=False` in the underlying `subprocess.Popen`. - ---- - -## quote - -Safely shell-quote a single string so it can be embedded in a command argument. - -```python -quote(arg: str) -> str -``` - -Thin wrapper over `shlex.quote`. Wraps the string in single quotes and escapes any embedded single quotes, producing a string safe to pass through a POSIX shell. - -**Parameters** - -| Parameter | Type | Description | -|-----------|------|-------------| -| `arg` | `str` | The raw argument string to quote | - -**Returns** `str` — quoted string (e.g., `'value'` or `'value'"'"'s'`) - -**Examples** - -```python -from scitex.sh import quote - -# Spaces in filenames -safe = quote("my file.txt") -# "'my file.txt'" - -# Shell metacharacters neutralised -safe = quote("file; rm -rf /") -# "'file; rm -rf /'" - -# Use when building args that will pass through a shell (e.g., ssh) -host = "server.example.com" -remote_path = "/data/my project/" -stx.sh.sh(["ssh", host, f"ls {quote(remote_path)}"]) -``` - -**When to use:** `quote` is mainly needed when constructing an argument that will itself be interpreted by a subordinate shell (e.g., the command string for `ssh`, `bash -c`). For direct subprocess list calls with `shell=False`, quoting is unnecessary — just pass raw strings as list elements. diff --git a/src/scitex/sh/_skills/types.md b/src/scitex/sh/_skills/types.md deleted file mode 100644 index 683332a25..000000000 --- a/src/scitex/sh/_skills/types.md +++ /dev/null @@ -1,64 +0,0 @@ ---- -description: Type definitions used by stx.sh — ShellResult TypedDict, CommandInput, and ReturnFormat. ---- - -# Type Definitions - -All types are defined in `scitex.sh._types` and re-exported from `scitex.sh`. - ---- - -## ShellResult - -A `TypedDict` returned by every execution function. - -```python -class ShellResult(TypedDict): - stdout: str # Decoded, stripped stdout from the process - stderr: str # Decoded, stripped stderr from the process - exit_code: int # Process return code (0 = success) - success: bool # True when exit_code == 0 -``` - -**Field notes** - -- `stdout` and `stderr` are decoded as UTF-8 and stripped of leading/trailing whitespace. -- `success` is exactly `exit_code == 0`. -- On timeout, the timeout message is appended to `stderr` and `success` is `False`. - -**Example usage** - -```python -result = stx.sh.sh(["git", "log", "--oneline", "-5"]) - -# Type-safe field access -stdout: str = result["stdout"] -stderr: str = result["stderr"] -code: int = result["exit_code"] -ok: bool = result["success"] -``` - ---- - -## CommandInput - -```python -CommandInput = List[str] -``` - -The accepted type for all command arguments. A plain `str` is explicitly rejected — pass a list of strings. - ---- - -## ReturnFormat - -```python -ReturnFormat = Literal["dict", "str"] -``` - -Controls what `sh()` returns: - -| Value | Return type | Content | -|-------|-------------|---------| -| `"dict"` | `ShellResult` | Full result with stdout, stderr, exit_code, success | -| `"str"` | `str` | stdout on success; stderr on failure | diff --git a/src/scitex/sh/_types.py b/src/scitex/sh/_types.py deleted file mode 100755 index 232794976..000000000 --- a/src/scitex/sh/_types.py +++ /dev/null @@ -1,29 +0,0 @@ -#!/usr/bin/env python3 -# -*- coding: utf-8 -*- -# Timestamp: "2025-10-29 07:24:01 (ywatanabe)" -# File: /home/ywatanabe/proj/scitex-code/src/scitex/sh/_types.py -# ---------------------------------------- -from __future__ import annotations - -import os - -__FILE__ = "./src/scitex/sh/_types.py" -__DIR__ = os.path.dirname(__FILE__) -# ---------------------------------------- - -__FILE__ = __file__ - -from typing import List, Literal, TypedDict - - -class ShellResult(TypedDict): - stdout: str - stderr: str - exit_code: int - success: bool - - -CommandInput = List[str] -ReturnFormat = Literal["dict", "str"] - -# EOF diff --git a/src/scitex/sh/test_sh.py b/src/scitex/sh/test_sh.py deleted file mode 100755 index 36fae0970..000000000 --- a/src/scitex/sh/test_sh.py +++ /dev/null @@ -1,68 +0,0 @@ -#!/usr/bin/env python3 -# -*- coding: utf-8 -*- -# Timestamp: "2025-10-29 07:23:59 (ywatanabe)" -# File: /home/ywatanabe/proj/scitex-code/src/scitex/sh/test_sh.py -# ---------------------------------------- -from __future__ import annotations - -import os - -__FILE__ = "./src/scitex/sh/test_sh.py" -__DIR__ = os.path.dirname(__FILE__) -# ---------------------------------------- - -__FILE__ = __file__ - -import sys - -import matplotlib.pyplot as plt - -import scitex - -CONFIG, sys.stdout, sys.stderr, plt, CC = scitex.session.start(sys, plt, verbose=False) - -# Test 1: Basic list command -print("Test 1: Basic list command") -result = scitex.sh(["echo", "Hello World"], verbose=True) -print(f"Result: {result}\n") - -# Test 2: String command should be rejected -print("Test 2: String command rejection") -try: - result = scitex.sh("echo 'Hello'") - print("ERROR: Should have raised TypeError") -except TypeError as ee: - print(f"Correctly rejected: {ee}\n") - -# Test 3: sh_run convenience function -print("Test 3: sh_run convenience function") -result = scitex.sh_run(["ls", "-la"]) -print(f"Success: {result['success']}, Exit code: {result['exit_code']}\n") - -# Test 4: Error handling -print("Test 4: Error handling") -result = scitex.sh_run(["cat", "/nonexistent/file"], verbose=False) -print(f"Success: {result['success']}") -print(f"Exit code: {result['exit_code']}") -print(f"Stderr: {result['stderr']}\n") - -# Test 5: Security - quote function -print("Test 5: Security - quote function") -from scitex.sh import quote - -dangerous_input = "file; rm -rf /" -safe_quoted = quote(dangerous_input) -print(f"Original: {dangerous_input}") -print(f"Quoted: {safe_quoted}\n") - -# Test 6: Security - null byte rejection -print("Test 6: Security - null byte rejection") -try: - scitex.sh(["echo", "test\0malicious"]) - print("ERROR: Should have raised ValueError") -except ValueError as ee: - print(f"Correctly rejected: {ee}\n") - -scitex.session.close(CONFIG, verbose=False, notify=False) - -# EOF diff --git a/src/scitex/sh/test_sh_simple.py b/src/scitex/sh/test_sh_simple.py deleted file mode 100755 index 4eb99d060..000000000 --- a/src/scitex/sh/test_sh_simple.py +++ /dev/null @@ -1,59 +0,0 @@ -#!/usr/bin/env python3 -# -*- coding: utf-8 -*- -# Timestamp: "2025-10-29 07:24:00 (ywatanabe)" -# File: /home/ywatanabe/proj/scitex-code/src/scitex/sh/test_sh_simple.py -# ---------------------------------------- -from __future__ import annotations - -import os - -__FILE__ = "./src/scitex/sh/test_sh_simple.py" -__DIR__ = os.path.dirname(__FILE__) -# ---------------------------------------- - -__FILE__ = __file__ - -import sys - -sys.path.insert(0, "/home/ywatanabe/proj/scitex_repo/src") - -from scitex.sh import quote, sh, sh_run - -# Test 1: Basic list command -print("Test 1: Basic list command") -result = sh(["echo", "Hello World"], verbose=True) -print(f"Result: {result}\n") - -# Test 2: String command should be rejected -print("Test 2: String command rejection") -try: - result = sh_run("echo test") - print("ERROR: Should have raised TypeError") -except TypeError as ee: - print(f"Correctly rejected: {ee}\n") - -# Test 3: Error handling -print("Test 3: Error handling") -result = sh_run(["cat", "/nonexistent/file"], verbose=False) -print(f"Success: {result['success']}") -print(f"Exit code: {result['exit_code']}") -print(f"Stderr: {result['stderr'][:50]}\n") - -# Test 4: Security - quote function -print("Test 4: Security - quote function") -dangerous_input = "file; rm -rf /" -safe_quoted = quote(dangerous_input) -print(f"Original: {dangerous_input}") -print(f"Quoted: {safe_quoted}\n") - -# Test 5: Security - null byte rejection -print("Test 5: Security - null byte rejection") -try: - sh(["echo", "test\0malicious"]) - print("ERROR: Should have raised ValueError") -except ValueError as ee: - print(f"Correctly rejected: {ee}\n") - -print("All tests passed!") - -# EOF diff --git a/tests/scitex/sh/TODO.md b/tests/scitex/sh/TODO.md deleted file mode 100644 index bf4090303..000000000 --- a/tests/scitex/sh/TODO.md +++ /dev/null @@ -1,23 +0,0 @@ - - -Implement test codes here. - -The source codes are: - /home/ywatanabe/proj/scitex-code/src/scitex/sh: - drwxr-xr-x 5 ywatanabe ywatanabe 4.0K Oct 29 07:23 . - drwxr-xr-x 47 ywatanabe ywatanabe 4.0K Oct 29 09:07 .. - -rwxr-xr-x 1 ywatanabe ywatanabe 2.0K Oct 29 07:23 _execute.py - -rw-r--r-- 1 ywatanabe ywatanabe 2.1K Oct 29 07:07 __init__.py - drwxr-xr-x 2 ywatanabe ywatanabe 4.0K Oct 29 07:25 __pycache__ - -rw-r--r-- 1 ywatanabe ywatanabe 1.4K Oct 29 07:23 README.md - -rwxr-xr-x 1 ywatanabe ywatanabe 1.8K Oct 29 07:23 _security.py - drwxr-xr-x 3 ywatanabe ywatanabe 4.0K Oct 29 06:59 test_sh_out - -rwxr-xr-x 1 ywatanabe ywatanabe 2.0K Oct 29 07:23 test_sh.py - -rwxr-xr-x 1 ywatanabe ywatanabe 1.7K Oct 29 07:24 test_sh_simple.py - -rwxr-xr-x 1 ywatanabe ywatanabe 612 Oct 29 07:24 _types.py - - diff --git a/tests/scitex/sh/run_test.sh b/tests/scitex/sh/run_test.sh deleted file mode 100755 index 93aec496a..000000000 --- a/tests/scitex/sh/run_test.sh +++ /dev/null @@ -1,33 +0,0 @@ -#!/bin/bash -# -*- coding: utf-8 -*- -# Timestamp: "2025-10-29 09:22:27 (ywatanabe)" -# File: ./tests/scitex/git/run_test.sh - -ORIG_DIR="$(pwd)" -THIS_DIR="$(cd $(dirname ${BASH_SOURCE[0]}) && pwd)" -LOG_PATH="$THIS_DIR/.$(basename $0).log" -echo > "$LOG_PATH" - -GIT_ROOT="$(git rev-parse --show-toplevel 2>/dev/null)" - -GRAY='\033[0;90m' -GREEN='\033[0;32m' -YELLOW='\033[0;33m' -RED='\033[0;31m' -NC='\033[0m' # No Color - -echo_info() { echo -e "${GRAY}INFO: $1${NC}"; } -echo_success() { echo -e "${GREEN}SUCC: $1${NC}"; } -echo_warning() { echo -e "${YELLOW}WARN: $1${NC}"; } -echo_error() { echo -e "${RED}ERRO: $1${NC}"; } -echo_header() { echo_info "=== $1 ==="; } -# --------------------------------------- - -echo_info "Running $0..." | tee $LOG_PATH - -cd $GIT_ROOT -pytest "$THIS_DIR" | tee $LOG_PATH - -echo_info "See full log: $LOG_PATH" | tee $LOG_PATH - -# EOF diff --git a/tests/scitex/sh/test___init__.py b/tests/scitex/sh/test___init__.py deleted file mode 100755 index c710031cf..000000000 --- a/tests/scitex/sh/test___init__.py +++ /dev/null @@ -1,424 +0,0 @@ -#!/usr/bin/env python3 -# Timestamp: "2026-01-05 (ywatanabe)" -# File: tests/scitex/sh/test___init__.py - -"""Tests for sh module public API. - -This module tests the public interface of the sh module: -- sh: Main shell execution function -- sh_run: Convenience function for command execution -- quote: Re-exported from _security -""" - -import pytest - - -class TestShFunctionBasic: - """Basic tests for sh function.""" - - def test_sh_import_from_module(self): - """Test sh can be imported from scitex.sh.""" - from scitex.sh import sh - - assert sh is not None - assert callable(sh) - - def test_sh_import_from_scitex(self): - """Test sh can be imported from scitex.sh module.""" - import scitex.sh as sh_module - - assert hasattr(sh_module, "sh") - assert callable(sh_module.sh) - - def test_sh_basic_command(self): - """Test sh executes basic command.""" - from scitex.sh import sh - - result = sh(["echo", "hello"], verbose=False) - - assert isinstance(result, dict) - assert result["success"] is True - - def test_sh_returns_dict_by_default(self): - """Test sh returns dict by default.""" - from scitex.sh import sh - - result = sh(["echo", "test"], verbose=False) - - assert isinstance(result, dict) - assert "stdout" in result - assert "stderr" in result - assert "exit_code" in result - assert "success" in result - - def test_sh_return_as_dict(self): - """Test sh returns dict when return_as='dict'.""" - from scitex.sh import sh - - result = sh(["echo", "test"], verbose=False, return_as="dict") - - assert isinstance(result, dict) - assert result["stdout"] == "test" - - def test_sh_return_as_str_success(self): - """Test sh returns stdout string when return_as='str' on success.""" - from scitex.sh import sh - - result = sh(["echo", "hello world"], verbose=False, return_as="str") - - assert isinstance(result, str) - assert result == "hello world" - - def test_sh_return_as_str_failure(self): - """Test sh returns stderr string when return_as='str' on failure.""" - from scitex.sh import sh - - result = sh(["cat", "/nonexistent/file"], verbose=False, return_as="str") - - assert isinstance(result, str) - # stderr should contain error message - assert len(result) > 0 - - -class TestShFunctionOptions: - """Tests for sh function options.""" - - def test_sh_verbose_false(self): - """Test sh with verbose=False.""" - from scitex.sh import sh - - # Should not raise and should work silently - result = sh(["echo", "quiet"], verbose=False) - assert result["success"] is True - - def test_sh_verbose_true(self, capsys): - """Test sh with verbose=True prints output.""" - from scitex.sh import sh - - sh(["echo", "loud"], verbose=True) - captured = capsys.readouterr() - - # Should print command and/or output - assert "loud" in captured.out or "echo" in captured.out - - def test_sh_timeout_success(self): - """Test sh with timeout that succeeds.""" - from scitex.sh import sh - - result = sh(["echo", "fast"], verbose=False, timeout=10) - - assert result["success"] is True - - def test_sh_timeout_exceeded(self): - """Test sh with timeout that is exceeded.""" - from scitex.sh import sh - - result = sh(["sleep", "5"], verbose=False, timeout=1) - - assert result["success"] is False - - def test_sh_stream_output(self): - """Test sh with stream_output enabled.""" - from scitex.sh import sh - - result = sh( - ["echo", "streamed"], - verbose=False, - stream_output=True, - ) - - assert result["success"] is True - assert "streamed" in result["stdout"] - - -class TestShFunctionSecurity: - """Security tests for sh function.""" - - def test_sh_rejects_string_command(self): - """Test sh rejects string commands.""" - from scitex.sh import sh - - with pytest.raises(TypeError) as exc_info: - sh("echo hello", verbose=False) - - assert "String commands are not allowed" in str(exc_info.value) - - def test_sh_rejects_null_byte(self): - """Test sh rejects null byte in arguments.""" - from scitex.sh import sh - - with pytest.raises(ValueError): - sh(["echo", "test\0malicious"], verbose=False) - - def test_sh_safe_with_special_chars_in_args(self): - """Test sh safely handles special chars in arguments.""" - from scitex.sh import sh - - # These are dangerous as shell metacharacters but safe in list format - result = sh(["echo", "test; rm -rf /"], verbose=False) - - assert result["success"] is True - assert "test; rm -rf /" in result["stdout"] - - -class TestShRunFunction: - """Tests for sh_run function.""" - - def test_sh_run_import(self): - """Test sh_run can be imported.""" - from scitex.sh import sh_run - - assert sh_run is not None - assert callable(sh_run) - - def test_sh_run_basic(self): - """Test sh_run executes basic command.""" - from scitex.sh import sh_run - - result = sh_run(["echo", "hello"], verbose=False) - - assert isinstance(result, dict) - assert result["success"] is True - assert result["stdout"] == "hello" - - def test_sh_run_returns_shell_result(self): - """Test sh_run returns ShellResult structure.""" - from scitex.sh import sh_run - - result = sh_run(["pwd"], verbose=False) - - assert "stdout" in result - assert "stderr" in result - assert "exit_code" in result - assert "success" in result - - def test_sh_run_success(self): - """Test sh_run with successful command.""" - from scitex.sh import sh_run - - result = sh_run(["true"], verbose=False) - - assert result["success"] is True - assert result["exit_code"] == 0 - - def test_sh_run_failure(self): - """Test sh_run with failed command.""" - from scitex.sh import sh_run - - result = sh_run(["false"], verbose=False) - - assert result["success"] is False - assert result["exit_code"] == 1 - - def test_sh_run_captures_stderr(self): - """Test sh_run captures stderr.""" - from scitex.sh import sh_run - - result = sh_run(["bash", "-c", "echo error >&2"], verbose=False) - - assert "error" in result["stderr"] - - def test_sh_run_verbose_option(self, capsys): - """Test sh_run verbose option.""" - from scitex.sh import sh_run - - sh_run(["echo", "visible"], verbose=True) - captured = capsys.readouterr() - - assert "visible" in captured.out or "echo" in captured.out - - -class TestQuoteExport: - """Tests for quote function export.""" - - def test_quote_import_from_sh(self): - """Test quote can be imported from scitex.sh.""" - from scitex.sh import quote - - assert quote is not None - assert callable(quote) - - def test_quote_basic(self): - """Test quote basic functionality.""" - from scitex.sh import quote - - result = quote("test string") - assert "test string" in result - - def test_quote_dangerous_string(self): - """Test quote safely handles dangerous strings.""" - from scitex.sh import quote - - result = quote("test; rm -rf /") - assert "'" in result - - -class TestModuleExports: - """Tests for module exports and __all__.""" - - def test_all_exports(self): - """Test __all__ contains expected exports.""" - import scitex.sh as sh_module - - assert hasattr(sh_module, "__all__") - assert "sh" in sh_module.__all__ - assert "sh_run" in sh_module.__all__ - assert "quote" in sh_module.__all__ - - def test_sh_accessible_from_scitex(self): - """Test sh is accessible from scitex namespace.""" - import scitex - - assert hasattr(scitex, "sh") - - def test_sh_run_accessible_from_scitex_sh(self): - """Test sh_run is accessible from scitex.sh module.""" - import scitex.sh as sh_module - - assert hasattr(sh_module, "sh_run") - assert callable(sh_module.sh_run) - - -class TestIntegration: - """Integration tests for sh module.""" - - def test_sh_and_sh_run_produce_same_result(self): - """Test sh and sh_run produce equivalent results.""" - from scitex.sh import sh, sh_run - - sh_result = sh(["echo", "test"], verbose=False, return_as="dict") - sh_run_result = sh_run(["echo", "test"], verbose=False) - - assert sh_result["stdout"] == sh_run_result["stdout"] - assert sh_result["exit_code"] == sh_run_result["exit_code"] - assert sh_result["success"] == sh_run_result["success"] - - def test_complete_workflow(self, tmp_path): - """Test complete workflow with file operations.""" - from scitex.sh import sh, sh_run - - # Create a file - test_file = tmp_path / "test.txt" - sh_run(["touch", str(test_file)], verbose=False) - assert test_file.exists() - - # Write content (using bash redirect in list format won't work, - # so use echo and redirect manually) - test_file.write_text("hello world") - - # Read file - result = sh(["cat", str(test_file)], verbose=False) - assert result["stdout"] == "hello world" - - # Count words - result = sh_run(["wc", "-w", str(test_file)], verbose=False) - assert result["success"] is True - assert "2" in result["stdout"] - - def test_error_handling_workflow(self): - """Test error handling workflow.""" - from scitex.sh import sh - - # Try to read nonexistent file - result = sh( - ["cat", "/nonexistent/file"], - verbose=False, - return_as="dict", - ) - - assert result["success"] is False - assert result["exit_code"] != 0 - assert len(result["stderr"]) > 0 - - -class TestShFunctionEdgeCases: - """Edge case tests for sh function.""" - - def test_sh_empty_output(self): - """Test sh with command that produces no output.""" - from scitex.sh import sh - - result = sh(["true"], verbose=False) - - assert result["success"] is True - assert result["stdout"] == "" - - def test_sh_multiline_output(self): - """Test sh with multiline output.""" - from scitex.sh import sh - - result = sh(["printf", "line1\nline2\nline3"], verbose=False) - - assert result["success"] is True - lines = result["stdout"].split("\n") - assert len(lines) == 3 - - def test_sh_unicode_output(self): - """Test sh with unicode output.""" - from scitex.sh import sh - - result = sh(["echo", "日本語"], verbose=False) - - assert result["success"] is True - assert "日本語" in result["stdout"] - - def test_sh_return_str_with_multiline(self): - """Test sh return_as='str' with multiline output.""" - from scitex.sh import sh - - result = sh( - ["printf", "a\nb\nc"], - verbose=False, - return_as="str", - ) - - assert isinstance(result, str) - assert "a" in result - assert "b" in result - assert "c" in result - - def test_sh_with_many_arguments(self): - """Test sh with many arguments.""" - from scitex.sh import sh - - args = ["arg" + str(i) for i in range(100)] - result = sh(["echo"] + args, verbose=False) - - assert result["success"] is True - for arg in args: - assert arg in result["stdout"] - - -class TestShRunEdgeCases: - """Edge case tests for sh_run function.""" - - def test_sh_run_empty_output(self): - """Test sh_run with command that produces no output.""" - from scitex.sh import sh_run - - result = sh_run(["true"], verbose=False) - - assert result["success"] is True - assert result["stdout"] == "" - assert result["exit_code"] == 0 - - def test_sh_run_with_path_argument(self): - """Test sh_run with file path argument.""" - from scitex.sh import sh_run - - result = sh_run(["ls", "/tmp"], verbose=False) - - assert result["success"] is True - - def test_sh_run_rejects_string(self): - """Test sh_run rejects string commands.""" - from scitex.sh import sh_run - - with pytest.raises(TypeError): - sh_run("ls -la", verbose=False) - - -if __name__ == "__main__": - import os - - pytest.main([os.path.abspath(__file__), "-v"]) diff --git a/tests/scitex/sh/test__execute.py b/tests/scitex/sh/test__execute.py deleted file mode 100644 index 4d0b8e201..000000000 --- a/tests/scitex/sh/test__execute.py +++ /dev/null @@ -1,725 +0,0 @@ -#!/usr/bin/env python3 -# Timestamp: "2026-01-05 (ywatanabe)" -# File: tests/scitex/sh/test__execute.py - -"""Tests for shell command execution functions. - -This module tests the execution functionality of the sh module: -- execute: Main execution function with options -- _execute_buffered: Buffered output execution -- _execute_with_streaming: Real-time streaming execution -""" - -import os -import tempfile - -import pytest - - -class TestExecuteBasic: - """Basic tests for execute function.""" - - def test_execute_import(self): - """Test execute can be imported.""" - from scitex.sh._execute import execute - - assert execute is not None - assert callable(execute) - - def test_execute_simple_command(self): - """Test execute with simple command.""" - from scitex.sh._execute import execute - - result = execute(["echo", "hello"], verbose=False) - - assert isinstance(result, dict) - assert "stdout" in result - assert "stderr" in result - assert "exit_code" in result - assert "success" in result - - def test_execute_returns_shell_result_structure(self): - """Test execute returns proper ShellResult structure.""" - from scitex.sh._execute import execute - - result = execute(["echo", "test"], verbose=False) - - assert result["stdout"] == "test" - assert result["stderr"] == "" - assert result["exit_code"] == 0 - assert result["success"] is True - - def test_execute_echo_command(self): - """Test execute with echo command.""" - from scitex.sh._execute import execute - - result = execute(["echo", "Hello World"], verbose=False) - - assert result["stdout"] == "Hello World" - assert result["success"] is True - - def test_execute_pwd_command(self): - """Test execute with pwd command.""" - from scitex.sh._execute import execute - - result = execute(["pwd"], verbose=False) - - assert result["success"] is True - assert len(result["stdout"]) > 0 - assert "/" in result["stdout"] # Unix path - - def test_execute_ls_command(self): - """Test execute with ls command.""" - from scitex.sh._execute import execute - - result = execute(["ls", "-la"], verbose=False) - - assert result["success"] is True - assert len(result["stdout"]) > 0 - - -class TestExecuteErrorHandling: - """Tests for error handling in execute.""" - - def test_execute_nonexistent_file(self): - """Test execute with nonexistent file.""" - from scitex.sh._execute import execute - - result = execute(["cat", "/nonexistent/path/file.txt"], verbose=False) - - assert result["success"] is False - assert result["exit_code"] != 0 - assert len(result["stderr"]) > 0 - - def test_execute_invalid_command(self): - """Test execute with invalid command.""" - from scitex.sh._execute import execute - - # This should raise FileNotFoundError since the command doesn't exist - with pytest.raises(FileNotFoundError): - execute(["nonexistent_command_12345"], verbose=False) - - def test_execute_command_with_error_exit(self): - """Test execute with command that exits with error.""" - from scitex.sh._execute import execute - - result = execute(["false"], verbose=False) - - assert result["success"] is False - assert result["exit_code"] == 1 - - def test_execute_command_with_success_exit(self): - """Test execute with command that exits successfully.""" - from scitex.sh._execute import execute - - result = execute(["true"], verbose=False) - - assert result["success"] is True - assert result["exit_code"] == 0 - - def test_execute_rejects_string_command(self): - """Test execute rejects string commands.""" - from scitex.sh._execute import execute - - with pytest.raises(TypeError): - execute("echo hello", verbose=False) - - -class TestExecuteTimeout: - """Tests for timeout functionality.""" - - def test_execute_with_timeout_success(self): - """Test execute with timeout that completes in time.""" - from scitex.sh._execute import execute - - result = execute(["echo", "quick"], verbose=False, timeout=10) - - assert result["success"] is True - assert result["stdout"] == "quick" - - def test_execute_timeout_exceeded(self): - """Test execute when timeout is exceeded.""" - from scitex.sh._execute import execute - - result = execute(["sleep", "5"], verbose=False, timeout=1) - - # Command should be killed due to timeout - assert result["success"] is False - assert "timeout" in result["stderr"].lower() or result["exit_code"] != 0 - - def test_execute_no_timeout(self): - """Test execute without timeout.""" - from scitex.sh._execute import execute - - result = execute(["echo", "test"], verbose=False, timeout=None) - - assert result["success"] is True - - -class TestExecuteOutput: - """Tests for output handling.""" - - def test_execute_multiline_output(self): - """Test execute with multiline output.""" - from scitex.sh._execute import execute - - result = execute(["printf", "line1\nline2\nline3"], verbose=False) - - assert result["success"] is True - lines = result["stdout"].split("\n") - assert len(lines) == 3 - assert lines[0] == "line1" - assert lines[1] == "line2" - assert lines[2] == "line3" - - def test_execute_unicode_output(self): - """Test execute with unicode output.""" - from scitex.sh._execute import execute - - result = execute(["echo", "日本語テスト"], verbose=False) - - assert result["success"] is True - assert "日本語テスト" in result["stdout"] - - def test_execute_empty_output(self): - """Test execute with empty output.""" - from scitex.sh._execute import execute - - result = execute(["true"], verbose=False) - - assert result["success"] is True - assert result["stdout"] == "" - - def test_execute_stderr_output(self): - """Test execute captures stderr.""" - from scitex.sh._execute import execute - - # Use bash to redirect to stderr - result = execute(["bash", "-c", "echo error >&2"], verbose=False) - - assert "error" in result["stderr"] - - def test_execute_both_stdout_and_stderr(self): - """Test execute captures both stdout and stderr.""" - from scitex.sh._execute import execute - - result = execute( - ["bash", "-c", "echo stdout; echo stderr >&2"], - verbose=False, - ) - - assert "stdout" in result["stdout"] - assert "stderr" in result["stderr"] - - -class TestExecuteStreaming: - """Tests for streaming output mode.""" - - def test_execute_streaming_mode(self): - """Test execute with streaming output enabled.""" - from scitex.sh._execute import execute - - result = execute( - ["echo", "streaming test"], - verbose=False, - stream_output=True, - ) - - assert result["success"] is True - assert "streaming test" in result["stdout"] - - def test_execute_streaming_multiline(self): - """Test streaming with multiline output.""" - from scitex.sh._execute import execute - - result = execute( - ["printf", "line1\nline2\nline3"], - verbose=False, - stream_output=True, - ) - - assert result["success"] is True - assert "line1" in result["stdout"] - assert "line2" in result["stdout"] - assert "line3" in result["stdout"] - - def test_execute_streaming_with_timeout(self): - """Test streaming mode with timeout.""" - from scitex.sh._execute import execute - - result = execute( - ["sleep", "5"], - verbose=False, - stream_output=True, - timeout=1, - ) - - # Should timeout - assert result["success"] is False - - -class TestExecuteVerbose: - """Tests for verbose mode.""" - - def test_execute_verbose_false(self, capsys): - """Test execute with verbose=False suppresses output.""" - from scitex.sh._execute import execute - - execute(["echo", "silent"], verbose=False) - captured = capsys.readouterr() - - # With verbose=False, there should be minimal or no output - # Note: The actual behavior depends on implementation - - def test_execute_verbose_true(self, capsys): - """Test execute with verbose=True shows output.""" - from scitex.sh._execute import execute - - execute(["echo", "visible"], verbose=True) - captured = capsys.readouterr() - - # With verbose=True, output should be printed - assert "visible" in captured.out or "echo" in captured.out - - -class TestExecuteWithFiles: - """Tests for execute with file operations.""" - - def test_execute_cat_file(self, tmp_path): - """Test execute cat with temporary file.""" - from scitex.sh._execute import execute - - # Create temp file - test_file = tmp_path / "test.txt" - test_file.write_text("file content") - - result = execute(["cat", str(test_file)], verbose=False) - - assert result["success"] is True - assert result["stdout"] == "file content" - - def test_execute_wc_file(self, tmp_path): - """Test execute wc with temporary file.""" - from scitex.sh._execute import execute - - # Create temp file with known content - test_file = tmp_path / "test.txt" - test_file.write_text("one two three\nfour five") - - result = execute(["wc", "-w", str(test_file)], verbose=False) - - assert result["success"] is True - assert "5" in result["stdout"] - - def test_execute_touch_creates_file(self, tmp_path): - """Test execute touch creates file.""" - from scitex.sh._execute import execute - - new_file = tmp_path / "newfile.txt" - assert not new_file.exists() - - result = execute(["touch", str(new_file)], verbose=False) - - assert result["success"] is True - assert new_file.exists() - - -class TestExecuteSpecialCharacters: - """Tests for handling special characters.""" - - def test_execute_with_spaces_in_argument(self): - """Test execute with spaces in argument.""" - from scitex.sh._execute import execute - - result = execute(["echo", "hello world"], verbose=False) - - assert result["success"] is True - assert result["stdout"] == "hello world" - - def test_execute_with_special_chars_in_argument(self): - """Test execute with special chars in argument.""" - from scitex.sh._execute import execute - - # These special chars are safe in list format - result = execute(["echo", "test; echo foo"], verbose=False) - - assert result["success"] is True - # The argument should be treated literally - assert "test; echo foo" in result["stdout"] - - def test_execute_with_quotes_in_argument(self): - """Test execute with quotes in argument.""" - from scitex.sh._execute import execute - - result = execute(["echo", 'say "hello"'], verbose=False) - - assert result["success"] is True - assert '"hello"' in result["stdout"] - - -class TestExecuteEnvironment: - """Tests for environment handling.""" - - def test_execute_inherits_environment(self): - """Test execute inherits environment variables.""" - from scitex.sh._execute import execute - - # HOME should be inherited - result = execute(["bash", "-c", "echo $HOME"], verbose=False) - - assert result["success"] is True - assert len(result["stdout"]) > 0 - - def test_execute_path_available(self): - """Test execute has PATH available.""" - from scitex.sh._execute import execute - - result = execute(["which", "ls"], verbose=False) - - assert result["success"] is True - assert "ls" in result["stdout"] - - -class TestExecuteExitCodes: - """Tests for exit code handling.""" - - def test_execute_exit_code_zero(self): - """Test execute with exit code 0.""" - from scitex.sh._execute import execute - - result = execute(["bash", "-c", "exit 0"], verbose=False) - - assert result["exit_code"] == 0 - assert result["success"] is True - - def test_execute_exit_code_one(self): - """Test execute with exit code 1.""" - from scitex.sh._execute import execute - - result = execute(["bash", "-c", "exit 1"], verbose=False) - - assert result["exit_code"] == 1 - assert result["success"] is False - - def test_execute_exit_code_custom(self): - """Test execute with custom exit code.""" - from scitex.sh._execute import execute - - result = execute(["bash", "-c", "exit 42"], verbose=False) - - assert result["exit_code"] == 42 - assert result["success"] is False - - -class TestBufferedExecution: - """Tests for _execute_buffered function.""" - - def test_execute_buffered_import(self): - """Test _execute_buffered can be imported.""" - from scitex.sh._execute import _execute_buffered - - assert _execute_buffered is not None - assert callable(_execute_buffered) - - def test_execute_buffered_basic(self): - """Test _execute_buffered basic functionality.""" - from scitex.sh._execute import _execute_buffered - - result = _execute_buffered(["echo", "test"], verbose=False, timeout=None) - - assert result["success"] is True - assert result["stdout"] == "test" - - -class TestStreamingExecution: - """Tests for _execute_with_streaming function.""" - - def test_execute_streaming_import(self): - """Test _execute_with_streaming can be imported.""" - from scitex.sh._execute import _execute_with_streaming - - assert _execute_with_streaming is not None - assert callable(_execute_with_streaming) - - def test_execute_streaming_basic(self): - """Test _execute_with_streaming basic functionality.""" - from scitex.sh._execute import _execute_with_streaming - - result = _execute_with_streaming( - ["echo", "streaming"], verbose=False, timeout=None - ) - - assert result["success"] is True - assert "streaming" in result["stdout"] - - -class TestExecuteRobustness: - """Robustness and edge case tests.""" - - def test_execute_long_output(self): - """Test execute with long output.""" - from scitex.sh._execute import execute - - # Generate long output - result = execute(["seq", "1", "1000"], verbose=False) - - assert result["success"] is True - lines = result["stdout"].split("\n") - assert len(lines) == 1000 - - def test_execute_rapid_succession(self): - """Test execute in rapid succession.""" - from scitex.sh._execute import execute - - for i in range(10): - result = execute(["echo", str(i)], verbose=False) - assert result["success"] is True - assert result["stdout"] == str(i) - - def test_execute_empty_argument(self): - """Test execute with empty argument.""" - from scitex.sh._execute import execute - - result = execute(["echo", ""], verbose=False) - - assert result["success"] is True - - -if __name__ == "__main__": - import os - - import pytest - - pytest.main([os.path.abspath(__file__)]) - -# -------------------------------------------------------------------------------- -# Start of Source Code from: /home/ywatanabe/proj/scitex-code/src/scitex/sh/_execute.py -# -------------------------------------------------------------------------------- -# #!/usr/bin/env python3 -# # -*- coding: utf-8 -*- -# # Timestamp: "2025-10-29 07:23:56 (ywatanabe)" -# # File: /home/ywatanabe/proj/scitex-code/src/scitex/sh/_execute.py -# # ---------------------------------------- -# from __future__ import annotations -# import os -# -# __FILE__ = "./src/scitex/sh/_execute.py" -# __DIR__ = os.path.dirname(__FILE__) -# # ---------------------------------------- -# -# __FILE__ = __file__ -# -# import subprocess -# import sys -# import select -# import time -# -# import scitex -# from ._types import CommandInput -# from ._types import ShellResult -# from ._security import validate_command -# -# -# def execute( -# command_str_or_list: CommandInput, -# verbose: bool = True, -# timeout: int = None, -# stream_output: bool = False, -# ) -> ShellResult: -# """ -# Executes a shell command safely (list format only). -# -# Parameters: -# - command_str_or_list: Command to execute (must be list format) -# - verbose: Whether to print command and output -# - timeout: Timeout in seconds (None for no timeout) -# - stream_output: Whether to stream output in real-time (default: False) -# When True, prints output as it's generated instead of waiting -# for command completion -# -# Returns: -# - ShellResult dict with stdout, stderr, exit_code, success -# -# Raises: -# - TypeError: If command is a string (not allowed for security) -# - subprocess.TimeoutExpired: If command exceeds timeout -# -# Examples: -# - sh(['ls', '-la']) -# - sh(['git', 'status']) -# - sh(['pdflatex', '-interaction=nonstopmode', 'file.tex'], stream_output=True) -# """ -# validate_command(command_str_or_list) -# -# if verbose: -# cmd_display = " ".join(command_str_or_list) -# print(scitex.str.color_text(f"{cmd_display}", "yellow")) -# -# if stream_output: -# # Use real-time streaming mode -# return _execute_with_streaming(command_str_or_list, verbose, timeout) -# else: -# # Use buffered mode (original behavior) -# return _execute_buffered(command_str_or_list, verbose, timeout) -# -# -# def _execute_buffered( -# command_str_or_list: CommandInput, verbose: bool, timeout: int -# ) -> ShellResult: -# """Execute command with buffered output (original behavior).""" -# process = subprocess.Popen( -# command_str_or_list, -# shell=False, -# stdout=subprocess.PIPE, -# stderr=subprocess.PIPE, -# ) -# -# try: -# stdout_bytes, stderr_bytes = process.communicate(timeout=timeout) -# except subprocess.TimeoutExpired: -# process.kill() -# stdout_bytes, stderr_bytes = process.communicate() -# timeout_msg = f"Command timed out after {timeout} seconds" -# stderr_bytes = stderr_bytes + b"\n" + timeout_msg.encode("utf-8") -# -# stdout = stdout_bytes.decode("utf-8").strip() -# stderr = stderr_bytes.decode("utf-8").strip() -# exit_code = process.returncode -# -# result: ShellResult = { -# "stdout": stdout, -# "stderr": stderr, -# "exit_code": exit_code, -# "success": exit_code == 0, -# } -# -# if verbose: -# if stdout: -# print(stdout) -# if stderr: -# print(scitex.str.color_text(stderr, "red")) -# -# return result -# -# -# def _execute_with_streaming( -# command_str_or_list: CommandInput, verbose: bool, timeout: int -# ) -> ShellResult: -# """Execute command with real-time output streaming using select.""" -# import io -# -# # Set PYTHONUNBUFFERED for Python scripts and unbuffered mode for shell -# env = os.environ.copy() -# env["PYTHONUNBUFFERED"] = "1" -# -# process = subprocess.Popen( -# command_str_or_list, -# shell=False, -# stdout=subprocess.PIPE, -# stderr=subprocess.PIPE, -# bufsize=0, # Unbuffered -# env=env, -# ) -# -# stdout_data = [] -# stderr_data = [] -# start_time = time.time() -# -# # Use non-blocking reads -# import fcntl -# -# def make_non_blocking(fd): -# flags = fcntl.fcntl(fd, fcntl.F_GETFL) -# fcntl.fcntl(fd, fcntl.F_SETFL, flags | os.O_NONBLOCK) -# -# make_non_blocking(process.stdout) -# make_non_blocking(process.stderr) -# -# try: -# while True: -# # Check timeout -# if timeout and (time.time() - start_time) > timeout: -# process.kill() -# timeout_msg = f"Command timed out after {timeout} seconds" -# if verbose: -# print(scitex.str.color_text(timeout_msg, "red"), flush=True) -# stderr_data.append(timeout_msg.encode()) -# break -# -# # Check if process has finished -# poll_result = process.poll() -# -# # Read available data from stdout -# try: -# chunk = process.stdout.read() -# if chunk: -# stdout_data.append(chunk) -# if verbose: -# text = chunk.decode("utf-8", errors="replace") -# print(text, end="", flush=True) -# except (IOError, BlockingIOError): -# pass -# -# # Read available data from stderr -# try: -# chunk = process.stderr.read() -# if chunk: -# stderr_data.append(chunk) -# if verbose: -# text = chunk.decode("utf-8", errors="replace") -# print(scitex.str.color_text(text, "red"), end="", flush=True) -# except (IOError, BlockingIOError): -# pass -# -# # If process finished, do final read and break -# if poll_result is not None: -# # Final read to catch any remaining buffered output -# try: -# chunk = process.stdout.read() -# if chunk: -# stdout_data.append(chunk) -# if verbose: -# text = chunk.decode("utf-8", errors="replace") -# print(text, end="", flush=True) -# except (IOError, BlockingIOError): -# pass -# -# try: -# chunk = process.stderr.read() -# if chunk: -# stderr_data.append(chunk) -# if verbose: -# text = chunk.decode("utf-8", errors="replace") -# print( -# scitex.str.color_text(text, "red"), end="", flush=True -# ) -# except (IOError, BlockingIOError): -# pass -# break -# -# # Small sleep to prevent CPU spinning -# time.sleep(0.05) -# -# except Exception as e: -# process.kill() -# raise -# -# stdout = b"".join(stdout_data).decode("utf-8", errors="replace").strip() -# stderr = b"".join(stderr_data).decode("utf-8", errors="replace").strip() -# exit_code = process.returncode -# -# result: ShellResult = { -# "stdout": stdout, -# "stderr": stderr, -# "exit_code": exit_code, -# "success": exit_code == 0, -# } -# -# return result -# -# -# # EOF - -# -------------------------------------------------------------------------------- -# End of Source Code from: /home/ywatanabe/proj/scitex-code/src/scitex/sh/_execute.py -# -------------------------------------------------------------------------------- diff --git a/tests/scitex/sh/test__security.py b/tests/scitex/sh/test__security.py deleted file mode 100644 index f6ee285b4..000000000 --- a/tests/scitex/sh/test__security.py +++ /dev/null @@ -1,511 +0,0 @@ -#!/usr/bin/env python3 -# Timestamp: "2026-01-05 (ywatanabe)" -# File: tests/scitex/sh/test__security.py - -"""Tests for shell command security functions. - -This module tests the security functionality of the sh module: -- validate_command: Validates commands for security issues -- quote: Safely quotes arguments for shell use -- DANGEROUS_CHARS: List of dangerous shell characters -""" - -import pytest - - -class TestValidateCommandBasic: - """Basic tests for validate_command function.""" - - def test_validate_command_import(self): - """Test validate_command can be imported.""" - from scitex.sh._security import validate_command - - assert validate_command is not None - assert callable(validate_command) - - def test_validate_command_accepts_list(self): - """Test validate_command accepts list format.""" - from scitex.sh._security import validate_command - - # Should not raise any exception - validate_command(["ls", "-la"]) - validate_command(["echo", "hello"]) - validate_command(["pwd"]) - - def test_validate_command_rejects_string(self): - """Test validate_command rejects string format.""" - from scitex.sh._security import validate_command - - with pytest.raises(TypeError) as exc_info: - validate_command("ls -la") - - assert "String commands are not allowed" in str(exc_info.value) - assert "security reasons" in str(exc_info.value) - - def test_validate_command_string_with_pipes_rejected(self): - """Test string with pipes is rejected.""" - from scitex.sh._security import validate_command - - with pytest.raises(TypeError): - validate_command("ls | grep test") - - def test_validate_command_string_with_semicolon_rejected(self): - """Test string with semicolon is rejected.""" - from scitex.sh._security import validate_command - - with pytest.raises(TypeError): - validate_command("ls; rm -rf /") - - -class TestValidateCommandNullByte: - """Tests for null byte detection in validate_command.""" - - def test_null_byte_in_argument_rejected(self): - """Test null byte in argument is rejected.""" - from scitex.sh._security import validate_command - - with pytest.raises(ValueError) as exc_info: - validate_command(["echo", "test\0malicious"]) - - assert "null byte" in str(exc_info.value) - assert "shell injection" in str(exc_info.value) - - def test_null_byte_in_command_rejected(self): - """Test null byte in command is rejected.""" - from scitex.sh._security import validate_command - - with pytest.raises(ValueError): - validate_command(["ls\0", "-la"]) - - def test_null_byte_at_start_rejected(self): - """Test null byte at start of argument is rejected.""" - from scitex.sh._security import validate_command - - with pytest.raises(ValueError): - validate_command(["echo", "\0malicious"]) - - def test_null_byte_at_end_rejected(self): - """Test null byte at end of argument is rejected.""" - from scitex.sh._security import validate_command - - with pytest.raises(ValueError): - validate_command(["echo", "test\0"]) - - def test_multiple_null_bytes_rejected(self): - """Test multiple null bytes are rejected.""" - from scitex.sh._security import validate_command - - with pytest.raises(ValueError): - validate_command(["echo", "a\0b\0c"]) - - -class TestValidateCommandValidCases: - """Test validate_command with valid inputs.""" - - def test_simple_command(self): - """Test simple command passes validation.""" - from scitex.sh._security import validate_command - - # Should not raise - validate_command(["ls"]) - - def test_command_with_flags(self): - """Test command with flags passes validation.""" - from scitex.sh._security import validate_command - - validate_command(["ls", "-la", "--color=auto"]) - - def test_command_with_path(self): - """Test command with path argument passes validation.""" - from scitex.sh._security import validate_command - - validate_command(["cat", "/etc/passwd"]) - validate_command(["ls", "/home/user/Documents"]) - - def test_command_with_spaces_in_argument(self): - """Test command with spaces in argument passes validation.""" - from scitex.sh._security import validate_command - - validate_command(["echo", "Hello World"]) - validate_command(["touch", "file with spaces.txt"]) - - def test_command_with_special_chars_in_argument(self): - """Test command with special chars as data passes validation.""" - from scitex.sh._security import validate_command - - # These are dangerous as shell metacharacters but safe in list format - validate_command(["echo", "test; rm -rf /"]) - validate_command(["echo", "test | grep foo"]) - validate_command(["echo", "test && echo bar"]) - - def test_command_with_unicode(self): - """Test command with unicode characters passes validation.""" - from scitex.sh._security import validate_command - - validate_command(["echo", "日本語"]) - validate_command(["echo", "Привет мир"]) - - def test_empty_argument_list(self): - """Test empty argument is valid.""" - from scitex.sh._security import validate_command - - validate_command(["echo", ""]) - - -class TestQuoteFunction: - """Tests for the quote function.""" - - def test_quote_import(self): - """Test quote can be imported.""" - from scitex.sh._security import quote - - assert quote is not None - assert callable(quote) - - def test_quote_simple_string(self): - """Test quoting simple string.""" - from scitex.sh._security import quote - - result = quote("hello") - assert result == "hello" or result == "'hello'" - - def test_quote_string_with_spaces(self): - """Test quoting string with spaces.""" - from scitex.sh._security import quote - - result = quote("hello world") - # shlex.quote wraps in single quotes - assert result == "'hello world'" - - def test_quote_dangerous_string(self): - """Test quoting dangerous string.""" - from scitex.sh._security import quote - - dangerous = "file; rm -rf /" - result = quote(dangerous) - # Should be safely quoted - assert "'" in result or result.startswith("'") - # The semicolon should be inside quotes, not as a command separator - assert result != dangerous - - def test_quote_string_with_semicolon(self): - """Test quoting string with semicolon.""" - from scitex.sh._security import quote - - result = quote("test;command") - assert "'" in result - - def test_quote_string_with_pipe(self): - """Test quoting string with pipe character.""" - from scitex.sh._security import quote - - result = quote("test|command") - assert "'" in result - - def test_quote_string_with_ampersand(self): - """Test quoting string with ampersand.""" - from scitex.sh._security import quote - - result = quote("test&command") - assert "'" in result - - def test_quote_string_with_dollar(self): - """Test quoting string with dollar sign.""" - from scitex.sh._security import quote - - result = quote("$HOME") - assert "'" in result - - def test_quote_string_with_backtick(self): - """Test quoting string with backtick.""" - from scitex.sh._security import quote - - result = quote("`whoami`") - # Backticks should be quoted - assert "'" in result - - def test_quote_empty_string(self): - """Test quoting empty string.""" - from scitex.sh._security import quote - - result = quote("") - assert result == "''" - - def test_quote_string_with_single_quotes(self): - """Test quoting string containing single quotes.""" - from scitex.sh._security import quote - - result = quote("it's a test") - # Should handle single quotes properly - assert "it" in result - assert "a test" in result - - def test_quote_string_with_double_quotes(self): - """Test quoting string containing double quotes.""" - from scitex.sh._security import quote - - result = quote('say "hello"') - assert "hello" in result - - def test_quote_preserves_content(self): - """Test that quote preserves the original content.""" - import shlex - - from scitex.sh._security import quote - - original = "test data 123" - quoted = quote(original) - # When unquoted via shell, should get original back - # shlex.split can unquote for us - unquoted = shlex.split(quoted)[0] - assert unquoted == original - - def test_quote_newline(self): - """Test quoting string with newline.""" - from scitex.sh._security import quote - - result = quote("line1\nline2") - # Newline should be inside quotes - assert "\n" in result or "'" in result - - def test_quote_tab(self): - """Test quoting string with tab.""" - from scitex.sh._security import quote - - result = quote("col1\tcol2") - assert "\t" in result or "'" in result - - -class TestDangerousChars: - """Tests for DANGEROUS_CHARS constant.""" - - def test_dangerous_chars_import(self): - """Test DANGEROUS_CHARS can be imported.""" - from scitex.sh._security import DANGEROUS_CHARS - - assert DANGEROUS_CHARS is not None - assert isinstance(DANGEROUS_CHARS, list) - - def test_dangerous_chars_contains_semicolon(self): - """Test DANGEROUS_CHARS contains semicolon.""" - from scitex.sh._security import DANGEROUS_CHARS - - assert ";" in DANGEROUS_CHARS - - def test_dangerous_chars_contains_pipe(self): - """Test DANGEROUS_CHARS contains pipe.""" - from scitex.sh._security import DANGEROUS_CHARS - - assert "|" in DANGEROUS_CHARS - - def test_dangerous_chars_contains_ampersand(self): - """Test DANGEROUS_CHARS contains ampersand.""" - from scitex.sh._security import DANGEROUS_CHARS - - assert "&" in DANGEROUS_CHARS - - def test_dangerous_chars_contains_dollar(self): - """Test DANGEROUS_CHARS contains dollar sign.""" - from scitex.sh._security import DANGEROUS_CHARS - - assert "$" in DANGEROUS_CHARS - - def test_dangerous_chars_contains_backtick(self): - """Test DANGEROUS_CHARS contains backtick.""" - from scitex.sh._security import DANGEROUS_CHARS - - assert "`" in DANGEROUS_CHARS - - def test_dangerous_chars_contains_newline(self): - """Test DANGEROUS_CHARS contains newline.""" - from scitex.sh._security import DANGEROUS_CHARS - - assert "\n" in DANGEROUS_CHARS - - def test_dangerous_chars_contains_redirects(self): - """Test DANGEROUS_CHARS contains redirect characters.""" - from scitex.sh._security import DANGEROUS_CHARS - - assert ">" in DANGEROUS_CHARS - assert "<" in DANGEROUS_CHARS - - def test_dangerous_chars_contains_parentheses(self): - """Test DANGEROUS_CHARS contains parentheses.""" - from scitex.sh._security import DANGEROUS_CHARS - - assert "(" in DANGEROUS_CHARS - assert ")" in DANGEROUS_CHARS - - def test_dangerous_chars_contains_braces(self): - """Test DANGEROUS_CHARS contains braces.""" - from scitex.sh._security import DANGEROUS_CHARS - - assert "{" in DANGEROUS_CHARS - assert "}" in DANGEROUS_CHARS - - -class TestSecurityIntegration: - """Integration tests for security functions.""" - - def test_quote_makes_dangerous_string_safe(self): - """Test that quote makes dangerous strings safe.""" - from scitex.sh._security import DANGEROUS_CHARS, quote - - for char in DANGEROUS_CHARS: - dangerous_str = f"test{char}malicious" - quoted = quote(dangerous_str) - # The quoted string should be wrapped in quotes - assert "'" in quoted or quoted.startswith("'") - - def test_validate_accepts_quoted_dangerous_args(self): - """Test validate_command accepts safely quoted dangerous content.""" - from scitex.sh._security import quote, validate_command - - # These are safe because they're in list format - the content is data - validate_command(["echo", "test; rm -rf /"]) - validate_command(["echo", "$(whoami)"]) - validate_command(["cat", "file`id`"]) - - def test_error_messages_are_informative(self): - """Test that error messages provide guidance.""" - from scitex.sh._security import validate_command - - with pytest.raises(TypeError) as exc_info: - validate_command("ls -la") - - error_msg = str(exc_info.value) - # Should mention list format - assert "list" in error_msg.lower() - # Should mention security - assert "security" in error_msg.lower() - - -class TestEdgeCases: - """Test edge cases and boundary conditions.""" - - def test_validate_very_long_argument_list(self): - """Test validation of very long argument list.""" - from scitex.sh._security import validate_command - - # Create a command with many arguments - cmd = ["echo"] + [f"arg{i}" for i in range(1000)] - validate_command(cmd) # Should not raise - - def test_validate_empty_list(self): - """Test validation of empty list.""" - from scitex.sh._security import validate_command - - # Empty list might be valid (no command) - # Implementation behavior may vary - try: - validate_command([]) - except (ValueError, IndexError): - pass # Some implementations may reject empty lists - - def test_quote_very_long_string(self): - """Test quoting very long string.""" - from scitex.sh._security import quote - - long_str = "a" * 10000 - result = quote(long_str) - assert len(result) >= len(long_str) - - def test_quote_unicode_special_chars(self): - """Test quoting unicode special characters.""" - from scitex.sh._security import quote - - # Various unicode that might cause issues - unicode_strings = [ - "日本語", - "Ñoño", - "émoji 🎉", - "αβγ", - "中文测试", - ] - - for s in unicode_strings: - result = quote(s) - assert s in result or "'" in result - - -if __name__ == "__main__": - import os - - import pytest - - pytest.main([os.path.abspath(__file__)]) - -# -------------------------------------------------------------------------------- -# Start of Source Code from: /home/ywatanabe/proj/scitex-code/src/scitex/sh/_security.py -# -------------------------------------------------------------------------------- -# #!/usr/bin/env python3 -# # -*- coding: utf-8 -*- -# # Timestamp: "2025-10-29 07:23:58 (ywatanabe)" -# # File: /home/ywatanabe/proj/scitex-code/src/scitex/sh/_security.py -# # ---------------------------------------- -# from __future__ import annotations -# import os -# -# __FILE__ = "./src/scitex/sh/_security.py" -# __DIR__ = os.path.dirname(__FILE__) -# # ---------------------------------------- -# -# __FILE__ = __file__ -# -# import shlex -# from typing import Union -# from typing import List -# -# -# DANGEROUS_CHARS = [";", "|", "&", "$", "`", "\n", ">", "<", "(", ")", "{", "}"] -# -# -# def validate_command(command_str_or_list: Union[str, List[str]]) -> None: -# """ -# Validates command for security issues. -# -# Parameters: -# - command_str_or_list: Command string or list to validate -# -# Raises: -# - TypeError: If command is a string (not allowed for security) -# - ValueError: If command contains dangerous characters -# """ -# if isinstance(command_str_or_list, str): -# raise TypeError( -# "String commands are not allowed for security reasons. " -# "Use list format: ['command', 'arg1', 'arg2']. " -# "For pipes and redirects, use Python subprocess chaining instead." -# ) -# -# for arg in command_str_or_list: -# if "\0" in str(arg): -# raise ValueError( -# "Command argument contains null byte - potential shell injection attempt" -# ) -# -# -# def quote(arg: str) -> str: -# """ -# Safely quotes a string for use in shell commands. -# -# Parameters: -# - arg: The argument to quote -# -# Returns: -# - str: Safely quoted string -# -# Examples: -# -------- -# >>> filename = "file; rm -rf /" -# >>> from scitex.sh import sh, quote -# >>> sh(f"cat {quote(filename)}") -# """ -# return shlex.quote(arg) -# -# -# # EOF - -# -------------------------------------------------------------------------------- -# End of Source Code from: /home/ywatanabe/proj/scitex-code/src/scitex/sh/_security.py -# -------------------------------------------------------------------------------- diff --git a/tests/scitex/sh/test__types.py b/tests/scitex/sh/test__types.py deleted file mode 100644 index 0673c63b6..000000000 --- a/tests/scitex/sh/test__types.py +++ /dev/null @@ -1,286 +0,0 @@ -#!/usr/bin/env python3 -# Timestamp: "2026-01-05 (ywatanabe)" -# File: tests/scitex/sh/test__types.py - -"""Tests for shell command type definitions. - -This module tests the type definitions used in the sh module: -- ShellResult TypedDict for command execution results -- CommandInput type alias for command arguments -- ReturnFormat literal type for return format specification -""" - -from typing import get_args, get_origin, get_type_hints - -import pytest - - -class TestShellResult: - """Test ShellResult TypedDict structure.""" - - def test_shell_result_import(self): - """Test ShellResult can be imported.""" - from scitex.sh._types import ShellResult - - assert ShellResult is not None - - def test_shell_result_is_typed_dict(self): - """Test ShellResult is a TypedDict.""" - from typing import TypedDict - - from scitex.sh._types import ShellResult - - # TypedDict classes have __annotations__ - assert hasattr(ShellResult, "__annotations__") - # Check it has the expected structure - annotations = ShellResult.__annotations__ - assert "stdout" in annotations - assert "stderr" in annotations - assert "exit_code" in annotations - assert "success" in annotations - - def test_shell_result_field_types(self): - """Test ShellResult has correct field types.""" - from scitex.sh._types import ShellResult - - # Use get_type_hints to resolve forward references from __future__ annotations - hints = get_type_hints(ShellResult) - assert hints["stdout"] == str - assert hints["stderr"] == str - assert hints["exit_code"] == int - assert hints["success"] == bool - - def test_shell_result_can_be_instantiated(self): - """Test ShellResult dict can be created with correct structure.""" - from scitex.sh._types import ShellResult - - result: ShellResult = { - "stdout": "output text", - "stderr": "", - "exit_code": 0, - "success": True, - } - - assert result["stdout"] == "output text" - assert result["stderr"] == "" - assert result["exit_code"] == 0 - assert result["success"] is True - - def test_shell_result_success_case(self): - """Test ShellResult for successful command.""" - from scitex.sh._types import ShellResult - - result: ShellResult = { - "stdout": "Hello World", - "stderr": "", - "exit_code": 0, - "success": True, - } - - assert result["success"] is True - assert result["exit_code"] == 0 - - def test_shell_result_failure_case(self): - """Test ShellResult for failed command.""" - from scitex.sh._types import ShellResult - - result: ShellResult = { - "stdout": "", - "stderr": "File not found", - "exit_code": 1, - "success": False, - } - - assert result["success"] is False - assert result["exit_code"] == 1 - assert "File not found" in result["stderr"] - - def test_shell_result_multiline_output(self): - """Test ShellResult with multiline output.""" - from scitex.sh._types import ShellResult - - multiline_output = "line1\nline2\nline3" - result: ShellResult = { - "stdout": multiline_output, - "stderr": "", - "exit_code": 0, - "success": True, - } - - assert result["stdout"].count("\n") == 2 - - def test_shell_result_unicode_content(self): - """Test ShellResult with Unicode content.""" - from scitex.sh._types import ShellResult - - unicode_output = "日本語テスト 中文测试 한국어테스트" - result: ShellResult = { - "stdout": unicode_output, - "stderr": "", - "exit_code": 0, - "success": True, - } - - assert result["stdout"] == unicode_output - - -class TestCommandInput: - """Test CommandInput type alias.""" - - def test_command_input_import(self): - """Test CommandInput can be imported.""" - from scitex.sh._types import CommandInput - - assert CommandInput is not None - - def test_command_input_is_list_of_strings(self): - """Test CommandInput is List[str].""" - from typing import List - - from scitex.sh._types import CommandInput - - # CommandInput should be List[str] - assert get_origin(CommandInput) == list - args = get_args(CommandInput) - assert args == (str,) - - def test_command_input_valid_examples(self): - """Test valid CommandInput examples.""" - from scitex.sh._types import CommandInput - - # These are valid CommandInput values - cmd1: CommandInput = ["ls", "-la"] - cmd2: CommandInput = ["echo", "Hello World"] - cmd3: CommandInput = ["git", "commit", "-m", "message"] - cmd4: CommandInput = ["python", "-c", 'print("test")'] - - assert isinstance(cmd1, list) - assert all(isinstance(arg, str) for arg in cmd1) - assert isinstance(cmd2, list) - assert isinstance(cmd3, list) - assert isinstance(cmd4, list) - - def test_command_input_single_command(self): - """Test CommandInput with single command.""" - from scitex.sh._types import CommandInput - - cmd: CommandInput = ["pwd"] - assert len(cmd) == 1 - assert cmd[0] == "pwd" - - -class TestReturnFormat: - """Test ReturnFormat literal type.""" - - def test_return_format_import(self): - """Test ReturnFormat can be imported.""" - from scitex.sh._types import ReturnFormat - - assert ReturnFormat is not None - - def test_return_format_is_literal(self): - """Test ReturnFormat is a Literal type.""" - from typing import Literal - - from scitex.sh._types import ReturnFormat - - # ReturnFormat should be Literal["dict", "str"] - assert get_origin(ReturnFormat) == Literal - - def test_return_format_allowed_values(self): - """Test ReturnFormat allowed values.""" - from scitex.sh._types import ReturnFormat - - allowed = get_args(ReturnFormat) - assert "dict" in allowed - assert "str" in allowed - assert len(allowed) == 2 - - def test_return_format_dict_value(self): - """Test 'dict' is valid ReturnFormat.""" - from scitex.sh._types import ReturnFormat - - format_type: ReturnFormat = "dict" - assert format_type == "dict" - - def test_return_format_str_value(self): - """Test 'str' is valid ReturnFormat.""" - from scitex.sh._types import ReturnFormat - - format_type: ReturnFormat = "str" - assert format_type == "str" - - -class TestModuleExports: - """Test module exports and accessibility.""" - - def test_all_types_importable_from_module(self): - """Test all types can be imported from _types module.""" - from scitex.sh._types import CommandInput, ReturnFormat, ShellResult - - assert ShellResult is not None - assert CommandInput is not None - assert ReturnFormat is not None - - def test_types_consistency(self): - """Test types are consistent with usage patterns.""" - from scitex.sh._types import CommandInput, ReturnFormat, ShellResult - - # Create a mock scenario - command: CommandInput = ["echo", "test"] - return_format: ReturnFormat = "dict" - result: ShellResult = { - "stdout": "test", - "stderr": "", - "exit_code": 0, - "success": True, - } - - # All should work without type errors - assert isinstance(command, list) - assert return_format in ("dict", "str") - assert isinstance(result, dict) - - -if __name__ == "__main__": - import os - - import pytest - - pytest.main([os.path.abspath(__file__)]) - -# -------------------------------------------------------------------------------- -# Start of Source Code from: /home/ywatanabe/proj/scitex-code/src/scitex/sh/_types.py -# -------------------------------------------------------------------------------- -# #!/usr/bin/env python3 -# # -*- coding: utf-8 -*- -# # Timestamp: "2025-10-29 07:24:01 (ywatanabe)" -# # File: /home/ywatanabe/proj/scitex-code/src/scitex/sh/_types.py -# # ---------------------------------------- -# from __future__ import annotations -# import os -# -# __FILE__ = "./src/scitex/sh/_types.py" -# __DIR__ = os.path.dirname(__FILE__) -# # ---------------------------------------- -# -# __FILE__ = __file__ -# -# from typing import List, Literal, TypedDict -# -# -# class ShellResult(TypedDict): -# stdout: str -# stderr: str -# exit_code: int -# success: bool -# -# -# CommandInput = List[str] -# ReturnFormat = Literal["dict", "str"] -# -# # EOF - -# -------------------------------------------------------------------------------- -# End of Source Code from: /home/ywatanabe/proj/scitex-code/src/scitex/sh/_types.py -# -------------------------------------------------------------------------------- diff --git a/tests/scitex/sh/test_test_sh.py b/tests/scitex/sh/test_test_sh.py deleted file mode 100644 index 3ef7a3a4a..000000000 --- a/tests/scitex/sh/test_test_sh.py +++ /dev/null @@ -1,88 +0,0 @@ -# Add your tests here - -if __name__ == "__main__": - import os - - import pytest - - pytest.main([os.path.abspath(__file__)]) - -# -------------------------------------------------------------------------------- -# Start of Source Code from: /home/ywatanabe/proj/scitex-code/src/scitex/sh/test_sh.py -# -------------------------------------------------------------------------------- -# #!/usr/bin/env python3 -# # -*- coding: utf-8 -*- -# # Timestamp: "2025-10-29 07:23:59 (ywatanabe)" -# # File: /home/ywatanabe/proj/scitex-code/src/scitex/sh/test_sh.py -# # ---------------------------------------- -# from __future__ import annotations -# -# import os -# -# __FILE__ = ( -# "./src/scitex/sh/test_sh.py" -# ) -# __DIR__ = os.path.dirname(__FILE__) -# # ---------------------------------------- -# -# __FILE__ = __file__ -# -# import sys -# -# import matplotlib.pyplot as plt -# -# import scitex -# -# CONFIG, sys.stdout, sys.stderr, plt, CC = scitex.session.start( -# sys, plt, verbose=False -# ) -# -# # Test 1: Basic list command -# print("Test 1: Basic list command") -# result = scitex.sh(["echo", "Hello World"], verbose=True) -# print(f"Result: {result}\n") -# -# # Test 2: String command should be rejected -# print("Test 2: String command rejection") -# try: -# result = scitex.sh("echo 'Hello'") -# print("ERROR: Should have raised TypeError") -# except TypeError as ee: -# print(f"Correctly rejected: {ee}\n") -# -# # Test 3: sh_run convenience function -# print("Test 3: sh_run convenience function") -# result = scitex.sh_run(["ls", "-la"]) -# print(f"Success: {result['success']}, Exit code: {result['exit_code']}\n") -# -# # Test 4: Error handling -# print("Test 4: Error handling") -# result = scitex.sh_run(["cat", "/nonexistent/file"], verbose=False) -# print(f"Success: {result['success']}") -# print(f"Exit code: {result['exit_code']}") -# print(f"Stderr: {result['stderr']}\n") -# -# # Test 5: Security - quote function -# print("Test 5: Security - quote function") -# from scitex.sh import quote -# -# dangerous_input = "file; rm -rf /" -# safe_quoted = quote(dangerous_input) -# print(f"Original: {dangerous_input}") -# print(f"Quoted: {safe_quoted}\n") -# -# # Test 6: Security - null byte rejection -# print("Test 6: Security - null byte rejection") -# try: -# scitex.sh(["echo", "test\0malicious"]) -# print("ERROR: Should have raised ValueError") -# except ValueError as ee: -# print(f"Correctly rejected: {ee}\n") -# -# scitex.session.close(CONFIG, verbose=False, notify=False) -# -# # EOF - -# -------------------------------------------------------------------------------- -# End of Source Code from: /home/ywatanabe/proj/scitex-code/src/scitex/sh/test_sh.py -# -------------------------------------------------------------------------------- diff --git a/tests/scitex/sh/test_test_sh_simple.py b/tests/scitex/sh/test_test_sh_simple.py deleted file mode 100644 index 6d0c2dca1..000000000 --- a/tests/scitex/sh/test_test_sh_simple.py +++ /dev/null @@ -1,77 +0,0 @@ -# Add your tests here - -if __name__ == "__main__": - import os - - import pytest - - pytest.main([os.path.abspath(__file__)]) - -# -------------------------------------------------------------------------------- -# Start of Source Code from: /home/ywatanabe/proj/scitex-code/src/scitex/sh/test_sh_simple.py -# -------------------------------------------------------------------------------- -# #!/usr/bin/env python3 -# # -*- coding: utf-8 -*- -# # Timestamp: "2025-10-29 07:24:00 (ywatanabe)" -# # File: /home/ywatanabe/proj/scitex-code/src/scitex/sh/test_sh_simple.py -# # ---------------------------------------- -# from __future__ import annotations -# -# import os -# -# __FILE__ = ( -# "./src/scitex/sh/test_sh_simple.py" -# ) -# __DIR__ = os.path.dirname(__FILE__) -# # ---------------------------------------- -# -# __FILE__ = __file__ -# -# import sys -# -# sys.path.insert(0, "/home/ywatanabe/proj/scitex_repo/src") -# -# from scitex.sh import quote, sh, sh_run -# -# # Test 1: Basic list command -# print("Test 1: Basic list command") -# result = sh(["echo", "Hello World"], verbose=True) -# print(f"Result: {result}\n") -# -# # Test 2: String command should be rejected -# print("Test 2: String command rejection") -# try: -# result = sh_run("echo test") -# print("ERROR: Should have raised TypeError") -# except TypeError as ee: -# print(f"Correctly rejected: {ee}\n") -# -# # Test 3: Error handling -# print("Test 3: Error handling") -# result = sh_run(["cat", "/nonexistent/file"], verbose=False) -# print(f"Success: {result['success']}") -# print(f"Exit code: {result['exit_code']}") -# print(f"Stderr: {result['stderr'][:50]}\n") -# -# # Test 4: Security - quote function -# print("Test 4: Security - quote function") -# dangerous_input = "file; rm -rf /" -# safe_quoted = quote(dangerous_input) -# print(f"Original: {dangerous_input}") -# print(f"Quoted: {safe_quoted}\n") -# -# # Test 5: Security - null byte rejection -# print("Test 5: Security - null byte rejection") -# try: -# sh(["echo", "test\0malicious"]) -# print("ERROR: Should have raised ValueError") -# except ValueError as ee: -# print(f"Correctly rejected: {ee}\n") -# -# print("All tests passed!") -# -# # EOF - -# -------------------------------------------------------------------------------- -# End of Source Code from: /home/ywatanabe/proj/scitex-code/src/scitex/sh/test_sh_simple.py -# --------------------------------------------------------------------------------