lzh/picoclaw
1
0
Fork 0
mirror of https://github.com/sipeed/picoclaw.git synced 2026-06-12 18:08:54 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(security): harden unauthenticated tool-exec paths (#1360)

Browse Source 
* fix(security): harden unauthenticated tool-exec paths (GHSA-pv8c-p6jf-3fpp)

- Exec tool: channel-based access control (default deny remote)
- Cron tool: command scheduling restricted to internal channels
- Web fetch: SSRF defense-in-depth (pre-flight + dial-time + redirect checks)
- File permissions: session/state dirs 0700, files 0600
- Registry: inject __channel/__chat_id into tool args (replaces racy SetContext)

28 new security regression tests.

(cherry picked from commit 191446ae19021604d3d5b0d9376b9655ab749105)

* fix(exec): revalidate working_dir before command start

* test(web): allow local oversized payload fixture

---------

Co-authored-by: xj <gh-xj@users.noreply.github.com>
This commit is contained in:
wenjie
2026-03-11 19:22:20 +08:00
committed by GitHub
parent dea06c391c
commit 8c2a9332c6
14 changed files with 622 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/tools/web_test.go
+210
View File
@@ -5,6 +5,7 @@ import (
"context"
"encoding/json"
"fmt"
"net"
"net/http"
"net/http/httptest"
"strings"
@@ -18,6 +19,8 @@ const testFetchLimit = int64(10 * 1024 * 1024)
// TestWebTool_WebFetch_Success verifies successful URL fetching
func TestWebTool_WebFetch_Success(t *testing.T) {
withPrivateWebFetchHostsAllowed(t)
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "text/html")
w.WriteHeader(http.StatusOK)
@@ -55,6 +58,8 @@ func TestWebTool_WebFetch_Success(t *testing.T) {
// TestWebTool_WebFetch_JSON verifies JSON content handling
func TestWebTool_WebFetch_JSON(t *testing.T) {
withPrivateWebFetchHostsAllowed(t)
testData := map[string]string{"key": "value", "number": "123"}
expectedJSON, _ := json.MarshalIndent(testData, "", " ")
@@ -163,6 +168,8 @@ func TestWebTool_WebFetch_MissingURL(t *testing.T) {
// TestWebTool_WebFetch_Truncation verifies content truncation
func TestWebTool_WebFetch_Truncation(t *testing.T) {
withPrivateWebFetchHostsAllowed(t)
longContent := strings.Repeat("x", 20000)
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
@@ -205,6 +212,8 @@ func TestWebTool_WebFetch_Truncation(t *testing.T) {
}
func TestWebFetchTool_PayloadTooLarge(t *testing.T) {
withPrivateWebFetchHostsAllowed(t)
// Create a mock HTTP server
ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "text/html")
@@ -290,6 +299,8 @@ func TestWebTool_WebSearch_MissingQuery(t *testing.T) {
// TestWebTool_WebFetch_HTMLExtraction verifies HTML text extraction
func TestWebTool_WebFetch_HTMLExtraction(t *testing.T) {
withPrivateWebFetchHostsAllowed(t)
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "text/html")
w.WriteHeader(http.StatusOK)
@@ -404,6 +415,205 @@ func TestWebFetchTool_extractText(t *testing.T) {
}
}
func withPrivateWebFetchHostsAllowed(t *testing.T) {
t.Helper()
previous := allowPrivateWebFetchHosts.Load()
allowPrivateWebFetchHosts.Store(true)
t.Cleanup(func() {
allowPrivateWebFetchHosts.Store(previous)
})
}
func TestWebTool_WebFetch_PrivateHostBlocked(t *testing.T) {
tool, err := NewWebFetchTool(50000, testFetchLimit)
if err != nil {
t.Fatalf("Failed to create web fetch tool: %v", err)
}
result := tool.Execute(context.Background(), map[string]any{
"url": "http://127.0.0.1:0",
})
if !result.IsError {
t.Errorf("expected error for private host URL, got success")
}
if !strings.Contains(result.ForLLM, "private or local network") &&
!strings.Contains(result.ForUser, "private or local network") {
t.Errorf("expected private host block message, got %q", result.ForLLM)
}
}
func TestWebTool_WebFetch_PrivateHostAllowedForTests(t *testing.T) {
withPrivateWebFetchHostsAllowed(t)
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "text/plain")
w.WriteHeader(http.StatusOK)
w.Write([]byte("ok"))
}))
defer server.Close()
tool, err := NewWebFetchTool(50000, testFetchLimit)
if err != nil {
t.Fatalf("Failed to create web fetch tool: %v", err)
}
result := tool.Execute(context.Background(), map[string]any{
"url": server.URL,
})
if result.IsError {
t.Errorf("expected success when private host access is allowed in tests, got %q", result.ForLLM)
}
}
// TestWebFetch_BlocksIPv4MappedIPv6Loopback verifies ::ffff:127.0.0.1 is blocked
func TestWebFetch_BlocksIPv4MappedIPv6Loopback(t *testing.T) {
tool, err := NewWebFetchTool(50000, testFetchLimit)
if err != nil {
t.Fatalf("Failed to create web fetch tool: %v", err)
}
result := tool.Execute(context.Background(), map[string]any{
"url": "http://[::ffff:127.0.0.1]:0",
})
if !result.IsError {
t.Error("expected error for IPv4-mapped IPv6 loopback URL, got success")
}
}
// TestWebFetch_BlocksMetadataIP verifies 169.254.169.254 is blocked
func TestWebFetch_BlocksMetadataIP(t *testing.T) {
tool, err := NewWebFetchTool(50000, testFetchLimit)
if err != nil {
t.Fatalf("Failed to create web fetch tool: %v", err)
}
result := tool.Execute(context.Background(), map[string]any{
"url": "http://169.254.169.254/latest/meta-data",
})
if !result.IsError {
t.Error("expected error for cloud metadata IP, got success")
}
}
// TestWebFetch_BlocksIPv6UniqueLocal verifies fc00::/7 addresses are blocked
func TestWebFetch_BlocksIPv6UniqueLocal(t *testing.T) {
tool, err := NewWebFetchTool(50000, testFetchLimit)
if err != nil {
t.Fatalf("Failed to create web fetch tool: %v", err)
}
result := tool.Execute(context.Background(), map[string]any{
"url": "http://[fd00::1]:0",
})
if !result.IsError {
t.Error("expected error for IPv6 unique local address, got success")
}
}
// TestWebFetch_Blocks6to4WithPrivateEmbed verifies 6to4 with private embedded IPv4 is blocked
func TestWebFetch_Blocks6to4WithPrivateEmbed(t *testing.T) {
tool, err := NewWebFetchTool(50000, testFetchLimit)
if err != nil {
t.Fatalf("Failed to create web fetch tool: %v", err)
}
// 2002:7f00:0001::1 embeds 127.0.0.1
result := tool.Execute(context.Background(), map[string]any{
"url": "http://[2002:7f00:0001::1]:0",
})
if !result.IsError {
t.Error("expected error for 6to4 with private embedded IPv4, got success")
}
}
// TestWebFetch_Allows6to4WithPublicEmbed verifies 6to4 with public embedded IPv4 is NOT blocked
func TestWebFetch_Allows6to4WithPublicEmbed(t *testing.T) {
tool, err := NewWebFetchTool(50000, testFetchLimit)
if err != nil {
t.Fatalf("Failed to create web fetch tool: %v", err)
}
// 2002:0801:0101::1 embeds 8.1.1.1 (public) — pre-flight should pass,
// connection will fail (no listener) but that's after the SSRF check.
result := tool.Execute(context.Background(), map[string]any{
"url": "http://[2002:0801:0101::1]:0",
})
// Should NOT be blocked by SSRF check — error should be connection failure, not "private"
if result.IsError && strings.Contains(result.ForLLM, "private") {
t.Error("6to4 with public embedded IPv4 should not be blocked as private")
}
}
// TestWebFetch_RedirectToPrivateBlocked verifies redirects to private IPs are blocked
func TestWebFetch_RedirectToPrivateBlocked(t *testing.T) {
withPrivateWebFetchHostsAllowed(t)
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
// Redirect to a private IP
http.Redirect(w, r, "http://10.0.0.1/secret", http.StatusFound)
}))
defer server.Close()
// Temporarily disable private host allowance for the redirect check
allowPrivateWebFetchHosts.Store(false)
defer allowPrivateWebFetchHosts.Store(true)
tool, err := NewWebFetchTool(50000, testFetchLimit)
if err != nil {
t.Fatalf("Failed to create web fetch tool: %v", err)
}
result := tool.Execute(context.Background(), map[string]any{
"url": server.URL,
})
if !result.IsError {
t.Error("expected error when redirecting to private IP, got success")
}
}
// TestIsPrivateOrRestrictedIP_Table tests IP classification logic
func TestIsPrivateOrRestrictedIP_Table(t *testing.T) {
tests := []struct {
ip string
blocked bool
desc string
}{
{"127.0.0.1", true, "IPv4 loopback"},
{"10.0.0.1", true, "IPv4 private class A"},
{"172.16.0.1", true, "IPv4 private class B"},
{"192.168.1.1", true, "IPv4 private class C"},
{"169.254.169.254", true, "link-local / cloud metadata"},
{"100.64.0.1", true, "carrier-grade NAT"},
{"0.0.0.0", true, "unspecified"},
{"8.8.8.8", false, "public DNS"},
{"1.1.1.1", false, "public DNS"},
{"::1", true, "IPv6 loopback"},
{"::ffff:127.0.0.1", true, "IPv4-mapped IPv6 loopback"},
{"::ffff:10.0.0.1", true, "IPv4-mapped IPv6 private"},
{"fc00::1", true, "IPv6 unique local"},
{"fd00::1", true, "IPv6 unique local"},
{"2002:7f00:0001::1", true, "6to4 with embedded 127.x (private)"},
{"2002:0a00:0001::1", true, "6to4 with embedded 10.0.0.1 (private)"},
{"2002:0801:0101::1", false, "6to4 with embedded 8.1.1.1 (public)"},
{"2001:0000:4136:e378:8000:63bf:f5ff:fffe", true, "Teredo with client 10.0.0.1 (private)"},
{"2001:0000:4136:e378:8000:63bf:f7f6:fefe", false, "Teredo with client 8.9.1.1 (public)"},
{"2607:f8b0:4004:800::200e", false, "public IPv6 (Google)"},
}
for _, tt := range tests {
t.Run(tt.desc, func(t *testing.T) {
ip := net.ParseIP(tt.ip)
if ip == nil {
t.Fatalf("failed to parse IP: %s", tt.ip)
}
got := isPrivateOrRestrictedIP(ip)
if got != tt.blocked {
t.Errorf("isPrivateOrRestrictedIP(%s) = %v, want %v", tt.ip, got, tt.blocked)
}
})
}
}
// TestWebTool_WebFetch_MissingDomain verifies error handling for URL without domain
func TestWebTool_WebFetch_MissingDomain(t *testing.T) {
tool, err := NewWebFetchTool(50000, testFetchLimit)