(#2377) fix(shell): sanitize terminal-facing output and harden path checks

Harden terminal-facing output in the exec tool and logger by escaping control and format characters before they reach the console or LLM-facing output. Refine exec workspace path validation so normalized relative paths inside the working directory remain allowed while real escapes are blocked.

- [x] 🐞 Bug fix (non-breaking change which fixes an issue)
- [ ]  New feature (non-breaking change which adds functionality)
- [ ] 📖 Documentation update
- [ ]  Code refactoring (no functional changes, no api changes)

- [ ] 🤖 Fully AI-generated (100% AI, 0% Human)
- [x] 🛠️ Mostly AI-generated (AI draft, Human verified/modified)
- [ ] 👨‍💻 Mostly Human-written (Human lead, AI assisted or none)

N/A

- **Reference URL:** N/A
- **Reasoning:** Raw ANSI escape sequences and Unicode bidi overrides should not be able to alter operator terminal state. The previous workspace guard also rejected any path containing `../` even when the normalized path still resolved inside the working directory.

- **Hardware:** PC
- **OS:** Linux
- **Model/Provider:** GPT-5.2 / OpenAI codex
- **Channels:** internal terminal / exec tool usage

<details>
<summary>Click to view Logs/Screenshots</summary>

- Added tests for ANSI and bidi escaping in terminal-facing output
- Added tests for allowing normalized in-workspace relative traversal
- Added tests for blocking relative traversal that escapes the working directory

</details>

- [x] My code/docs follow the style of this project.
- [x] I have performed a self-review of my own changes.
- [ ] I have updated the documentation accordingly.
This commit is contained in:
Vincent Janelle 2026-04-06 08:42:54 -07:00
parent df9124b824
commit b9b4c7ac04
5 changed files with 234 additions and 5 deletions

View file

@ -10,6 +10,8 @@ import (
"strings"
"sync"
"github.com/sipeed/picoclaw/pkg/termutil"
"github.com/rs/zerolog"
"golang.org/x/term"
)
@ -88,13 +90,15 @@ func formatFieldValue(i any) string {
case []byte:
s = string(val)
default:
return fmt.Sprintf("%v", i)
s = fmt.Sprintf("%v", i)
}
if unquoted, err := strconv.Unquote(s); err == nil {
s = unquoted
}
s = termutil.EscapeControlChars(s)
if strings.Contains(s, "\n") {
return fmt.Sprintf("\n%s", s)
}

View file

@ -250,6 +250,16 @@ func TestFormatFieldValue(t *testing.T) {
input: " ",
expected: `" "`,
},
{
name: "ANSI escape is rendered safely",
input: "\x1b[31mred\x1b[0m",
expected: `\x1b[31mred\x1b[0m`,
},
{
name: "Bidi override is escaped",
input: "safe\u202edanger",
expected: `safe\u202edanger`,
},
}
for _, tt := range tests {

34
pkg/termutil/escape.go Normal file
View file

@ -0,0 +1,34 @@
package termutil
import (
"fmt"
"strings"
"unicode"
)
// EscapeControlChars preserves readable text while escaping control and format
// characters that could alter terminal state, such as ANSI escape codes and
// bidi overrides.
func EscapeControlChars(input string) string {
var sb strings.Builder
sb.Grow(len(input))
for _, r := range input {
switch {
case r == '\n' || r == '\r' || r == '\t':
sb.WriteRune(r)
case r < 0x20 || r == 0x7f:
sb.WriteString(fmt.Sprintf("\\x%02x", r))
case unicode.Is(unicode.Cf, r):
if r <= 0xffff {
sb.WriteString(fmt.Sprintf("\\u%04x", r))
} else {
sb.WriteString(fmt.Sprintf("\\U%08x", r))
}
default:
sb.WriteRune(r)
}
}
return sb.String()
}

View file

@ -21,6 +21,8 @@ import (
"github.com/sipeed/picoclaw/pkg/config"
"github.com/sipeed/picoclaw/pkg/constants"
"github.com/sipeed/picoclaw/pkg/isolation"
"github.com/sipeed/picoclaw/pkg/logger"
"github.com/sipeed/picoclaw/pkg/termutil"
)
var (
@ -100,6 +102,11 @@ var (
// absolutePathPattern matches absolute file paths in commands (Unix and Windows).
absolutePathPattern = regexp.MustCompile(`[A-Za-z]:\\[^\\\"']+|/[^\s\"']+`)
// commandTokenPattern extracts unquoted shell tokens for lightweight path inspection.
// This intentionally stays simple; it is only used to identify obvious path-like
// arguments that need workspace validation.
commandTokenPattern = regexp.MustCompile(`[^\s"'` + "`" + `]+`)
// safePaths are kernel pseudo-devices that are always safe to reference in
// commands, regardless of workspace restriction. They contain no user data
// and cannot cause destructive writes.
@ -291,6 +298,7 @@ func (t *ExecTool) executeRun(ctx context.Context, args map[string]any) *ToolRes
}
isPty := getBoolArg("pty")
isBackground := getBoolArg("background")
requestedWD, _ := args["cwd"].(string)
if isPty {
if runtime.GOOS == "windows" {
@ -303,6 +311,7 @@ func (t *ExecTool) executeRun(ctx context.Context, args map[string]any) *ToolRes
if t.restrictToWorkspace && t.workingDir != "" {
resolvedWD, err := validatePathWithAllowPaths(wd, t.workingDir, true, t.allowedPathPatterns)
if err != nil {
t.logGuardBlock(command, requestedWD, cwd, err.Error())
return ErrorResult("Command blocked by safety guard (" + err.Error() + ")")
}
cwd = resolvedWD
@ -319,6 +328,7 @@ func (t *ExecTool) executeRun(ctx context.Context, args map[string]any) *ToolRes
}
if guardError := t.guardCommand(command, cwd); guardError != "" {
t.logGuardBlock(command, requestedWD, cwd, guardError)
return ErrorResult(guardError)
}
@ -327,6 +337,7 @@ func (t *ExecTool) executeRun(ctx context.Context, args map[string]any) *ToolRes
if t.restrictToWorkspace && t.workingDir != "" && cwd != t.workingDir {
resolved, err := filepath.EvalSymlinks(cwd)
if err != nil {
t.logGuardBlock(command, requestedWD, cwd, fmt.Sprintf("path resolution failed: %v", err))
return ErrorResult(fmt.Sprintf("Command blocked by safety guard (path resolution failed: %v)", err))
}
if isAllowedPath(resolved, t.allowedPathPatterns) {
@ -339,6 +350,7 @@ func (t *ExecTool) executeRun(ctx context.Context, args map[string]any) *ToolRes
}
rel, err := filepath.Rel(wsResolved, resolved)
if err != nil || !filepath.IsLocal(rel) {
t.logGuardBlock(command, requestedWD, cwd, "working directory escaped workspace")
return ErrorResult("Command blocked by safety guard (working directory escaped workspace)")
}
cwd = resolved
@ -352,6 +364,21 @@ func (t *ExecTool) executeRun(ctx context.Context, args map[string]any) *ToolRes
return t.runSync(ctx, command, cwd)
}
func (t *ExecTool) logGuardBlock(command, requestedWD, resolvedWD, reason string) {
fields := map[string]any{
"tool": t.Name(),
"command": command,
"reason": reason,
}
if requestedWD != "" {
fields["requested_cwd"] = requestedWD
}
if resolvedWD != "" {
fields["resolved_cwd"] = resolvedWD
}
logger.WarnCF("tool", "Exec blocked by safety guard", fields)
}
func (t *ExecTool) runSync(ctx context.Context, command, cwd string) *ToolResult {
// timeout == 0 means no timeout
var cmdCtx context.Context
@ -443,6 +470,8 @@ func (t *ExecTool) runSync(ctx context.Context, command, cwd string) *ToolResult
output = "(no output)"
}
output = termutil.EscapeControlChars(output)
maxLen := 10000
if len(output) > maxLen {
output = output[:maxLen] + fmt.Sprintf("\n... (truncated, %d more chars)", len(output)-maxLen)
@ -706,6 +735,7 @@ func (t *ExecTool) executeRead(args map[string]any) *ToolResult {
}
output := session.Read()
output = termutil.EscapeControlChars(output)
resp := ExecResponse{
SessionID: sessionID,
@ -1054,10 +1084,6 @@ func (t *ExecTool) guardCommand(command, cwd string) string {
}
if t.restrictToWorkspace {
if strings.Contains(cmd, "..\\") || strings.Contains(cmd, "../") {
return "Command blocked by safety guard (path traversal detected)"
}
cwdPath, err := filepath.Abs(cwd)
if err != nil {
return ""
@ -1073,6 +1099,10 @@ func (t *ExecTool) guardCommand(command, cwd string) string {
for _, loc := range matchIndices {
raw := cmd[loc[0]:loc[1]]
if loc[0] > 0 && !isPathTokenBoundary(cmd[loc[0]-1]) {
continue
}
// Skip URL path components that look like they're from web URLs.
// When a URL like "https://github.com" is parsed, the regex captures
// "//github.com" as a match (the path portion after "https:").
@ -1115,11 +1145,86 @@ func (t *ExecTool) guardCommand(command, cwd string) string {
return "Command blocked by safety guard (path outside working dir)"
}
}
for _, raw := range extractTraversalPathCandidates(cmd) {
p := raw
if !filepath.IsAbs(p) {
p = filepath.Join(cwdPath, p)
}
resolved, err := filepath.Abs(p)
if err != nil {
continue
}
if safePaths[resolved] {
continue
}
if isAllowedPath(resolved, t.allowedPathPatterns) {
continue
}
rel, err := filepath.Rel(cwdPath, resolved)
if err != nil {
continue
}
if strings.HasPrefix(rel, "..") {
return "Command blocked by safety guard (path outside working dir)"
}
}
}
return ""
}
func extractTraversalPathCandidates(command string) []string {
tokens := commandTokenPattern.FindAllString(command, -1)
if len(tokens) == 0 {
return nil
}
candidates := make([]string, 0, len(tokens))
seen := make(map[string]struct{}, len(tokens))
for _, token := range tokens {
for _, part := range strings.FieldsFunc(token, func(r rune) bool {
return r == '=' || r == ',' || r == ';' || r == '(' || r == ')'
}) {
part = strings.Trim(part, `"'`)
if part == "" {
continue
}
hasTraversal := strings.Contains(part, "../") ||
strings.Contains(part, `..\`) ||
strings.HasSuffix(part, "/..") ||
strings.HasSuffix(part, `\..`) ||
part == ".."
if !hasTraversal {
continue
}
if _, ok := seen[part]; ok {
continue
}
seen[part] = struct{}{}
candidates = append(candidates, part)
}
}
return candidates
}
func isPathTokenBoundary(b byte) bool {
switch b {
case ' ', '\t', '\n', '\r', '"', '\'', '`', '(', ')', '[', ']', '{', '}', ',', ';', '=', ':':
return true
default:
return false
}
}
func (t *ExecTool) SetTimeout(timeout time.Duration) {
t.timeout = timeout
}

View file

@ -703,6 +703,82 @@ func TestShellTool_URLBypassPrevented(t *testing.T) {
}
}
func TestShellTool_RelativeTraversalInsideWorkingDirAllowed(t *testing.T) {
root := t.TempDir()
workspace := filepath.Join(root, "workspace")
if err := os.MkdirAll(filepath.Join(workspace, "subdir"), 0o755); err != nil {
t.Fatalf("failed to create workspace: %v", err)
}
if err := os.WriteFile(filepath.Join(workspace, "file.txt"), []byte("ok"), 0o644); err != nil {
t.Fatalf("failed to create file: %v", err)
}
tool, err := NewExecTool(workspace, true)
if err != nil {
t.Fatalf("unable to configure exec tool: %s", err)
}
result := tool.Execute(context.Background(), map[string]any{
"action": "run",
"command": "cat subdir/../file.txt",
})
if result.IsError && strings.Contains(result.ForLLM, "path outside working dir") {
t.Fatalf("normalized relative path inside working dir should be allowed: %s", result.ForLLM)
}
}
func TestShellTool_RelativeTraversalOutsideWorkingDirBlocked(t *testing.T) {
root := t.TempDir()
workspace := filepath.Join(root, "workspace")
if err := os.MkdirAll(filepath.Join(workspace, "subdir"), 0o755); err != nil {
t.Fatalf("failed to create workspace: %v", err)
}
if err := os.WriteFile(filepath.Join(workspace, "secret.txt"), []byte("secret"), 0o644); err != nil {
t.Fatalf("failed to create file: %v", err)
}
tool, err := NewExecTool(workspace, true)
if err != nil {
t.Fatalf("unable to configure exec tool: %s", err)
}
result := tool.Execute(context.Background(), map[string]any{
"action": "run",
"command": "cat ../secret.txt",
"cwd": filepath.Join(workspace, "subdir"),
})
if !result.IsError || !strings.Contains(result.ForLLM, "path outside working dir") {
t.Fatalf("relative path that escapes working dir should be blocked, got: %s", result.ForLLM)
}
}
func TestShellTool_OutputEscapesTerminalControlChars(t *testing.T) {
tool, err := NewExecTool("", false)
if err != nil {
t.Fatalf("unable to configure exec tool: %s", err)
}
result := tool.Execute(context.Background(), map[string]any{
"action": "run",
"command": "printf '\\033[31mred\\033[0m\\u202E'",
})
if result.IsError {
t.Fatalf("expected success, got error: %s", result.ForLLM)
}
if strings.Contains(result.ForLLM, "\x1b") {
t.Fatalf("expected ANSI escape to be escaped in ForLLM, got: %q", result.ForLLM)
}
if strings.ContainsRune(result.ForLLM, '\u202e') {
t.Fatalf("expected bidi control to be escaped in ForLLM, got: %q", result.ForLLM)
}
if !strings.Contains(result.ForLLM, `\x1b[31mred\x1b[0m`) || !strings.Contains(strings.ToLower(result.ForLLM), `\u202e`) {
t.Fatalf("expected escaped control sequence in output, got: %q", result.ForLLM)
}
}
func TestShellTool_Background_ReturnsImmediately(t *testing.T) {
tool, err := NewExecTool("", false)
require.NoError(t, err)