opencode/picoclaw
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(security): add SSRF protection to web_fetch tool

Browse source 
Block requests to private IP ranges (RFC1918, loopback, link-local),
cloud metadata endpoints, and IPv6 reserved addresses. Includes DNS
rebinding protection via DialContext IP verification at connection time.
This commit is contained in:
Paul De Velder 2026-02-23 14:56:41 +01:00
parent edd920a6c4
commit 574a59e1a1
2 changed files with 198 additions and 16 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

102
pkg/tools/web.go
View file

@ -6,6 +6,7 @@ import (
"encoding/json"
"fmt"
"io"
"net"
"net/http"
"net/url"
"regexp"
@ -13,6 +14,70 @@ import (
"time"
)
// privateRanges contains CIDR blocks for private/reserved IP addresses.
var privateRanges []*net.IPNet
func init() {
for _, cidr := range []string{
"10.0.0.0/8",
"172.16.0.0/12",
"192.168.0.0/16",
"127.0.0.0/8",
"169.254.0.0/16",
"::1/128",
"fc00::/7",
"fe80::/10",
} {
_, network, _ := net.ParseCIDR(cidr)
privateRanges = append(privateRanges, network)
}
}
// blockedHosts contains hostnames that should never be accessed.
var blockedHosts = map[string]bool{
"metadata.google.internal": true,
"metadata.google.internal.": true,
}
// isPrivateIP checks if an IP is in a private or reserved range.
func isPrivateIP(ip net.IP) bool {
for _, r := range privateRanges {
if r.Contains(ip) {
return true
}
}
return false
}
// validateURLSafety checks if a URL is safe to fetch (not targeting private infrastructure).
func validateURLSafety(parsedURL *url.URL) error {
host := parsedURL.Hostname()
if blockedHosts[host] {
return fmt.Errorf("blocked: metadata endpoint")
}
// Check if host is a direct IP address
if ip := net.ParseIP(host); ip != nil {
if isPrivateIP(ip) {
return fmt.Errorf("blocked: private IP address")
}
return nil
}
// Resolve hostname and check all IPs
ips, err := net.LookupIP(host)
if err != nil {
return nil // let DNS failures fail at HTTP level
}
for _, ip := range ips {
if isPrivateIP(ip) {
return fmt.Errorf("blocked: URL resolves to private IP")
}
}
return nil
}
const (
userAgent = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
)
@ -441,6 +506,7 @@ func (t *WebSearchTool) Execute(ctx context.Context, args map[string]any) *ToolR
type WebFetchTool struct {
maxChars int
skipSSRFCheck bool // only for testing with httptest servers
}
func NewWebFetchTool(maxChars int) *WebFetchTool {
@ -497,6 +563,13 @@ func (t *WebFetchTool) Execute(ctx context.Context, args map[string]any) *ToolRe
return ErrorResult("missing domain in URL")
}
// SSRF protection: block private IPs and metadata endpoints
if !t.skipSSRFCheck {
if err := validateURLSafety(parsedURL); err != nil {
return ErrorResult(err.Error())
}
}
maxChars := t.maxChars
if mc, ok := args["maxChars"].(float64); ok {
if int(mc) > 100 {
@ -511,14 +584,35 @@ func (t *WebFetchTool) Execute(ctx context.Context, args map[string]any) *ToolRe
req.Header.Set("User-Agent", userAgent)
client := &http.Client{
Timeout: 60 * time.Second,
Transport: &http.Transport{
transport := &http.Transport{
MaxIdleConns: 10,
IdleConnTimeout: 30 * time.Second,
DisableCompression: false,
TLSHandshakeTimeout: 15 * time.Second,
},
}
// DNS rebinding protection: verify resolved IPs at connection time
if !t.skipSSRFCheck {
transport.DialContext = func(dialCtx context.Context, network, addr string) (net.Conn, error) {
host, port, err := net.SplitHostPort(addr)
if err != nil {
return nil, err
}
ips, err := net.DefaultResolver.LookupIPAddr(dialCtx, host)
if err != nil {
return nil, err
}
for _, ip := range ips {
if isPrivateIP(ip.IP) {
return nil, fmt.Errorf("blocked: connection to private IP")
}
}
dialer := &net.Dialer{Timeout: 10 * time.Second}
return dialer.DialContext(dialCtx, network, net.JoinHostPort(ips[0].IP.String(), port))
}
}
client := &http.Client{
Timeout: 60 * time.Second,
Transport: transport,
CheckRedirect: func(req *http.Request, via []*http.Request) error {
if len(via) >= 5 {
return fmt.Errorf("stopped after 5 redirects")

104
pkg/tools/web_test.go
View file

@ -3,10 +3,14 @@ package tools
import (
"context"
"encoding/json"
"net"
"net/http"
"net/http/httptest"
"net/url"
"strings"
"testing"
"github.com/stretchr/testify/assert"
)
// TestWebTool_WebFetch_Success verifies successful URL fetching
@ -18,7 +22,7 @@ func TestWebTool_WebFetch_Success(t *testing.T) {
}))
defer server.Close()
tool := NewWebFetchTool(50000)
tool := &WebFetchTool{maxChars: 50000, skipSSRFCheck: true}
ctx := context.Background()
args := map[string]any{
"url": server.URL,
@ -54,7 +58,7 @@ func TestWebTool_WebFetch_JSON(t *testing.T) {
}))
defer server.Close()
tool := NewWebFetchTool(50000)
tool := &WebFetchTool{maxChars: 50000, skipSSRFCheck: true}
ctx := context.Background()
args := map[string]any{
"url": server.URL,
@ -75,7 +79,7 @@ func TestWebTool_WebFetch_JSON(t *testing.T) {
// TestWebTool_WebFetch_InvalidURL verifies error handling for invalid URL
func TestWebTool_WebFetch_InvalidURL(t *testing.T) {
tool := NewWebFetchTool(50000)
tool := &WebFetchTool{maxChars: 50000, skipSSRFCheck: true}
ctx := context.Background()
args := map[string]any{
"url": "not-a-valid-url",
@ -96,7 +100,7 @@ func TestWebTool_WebFetch_InvalidURL(t *testing.T) {
// TestWebTool_WebFetch_UnsupportedScheme verifies error handling for non-http URLs
func TestWebTool_WebFetch_UnsupportedScheme(t *testing.T) {
tool := NewWebFetchTool(50000)
tool := &WebFetchTool{maxChars: 50000, skipSSRFCheck: true}
ctx := context.Background()
args := map[string]any{
"url": "ftp://example.com/file.txt",
@ -117,7 +121,7 @@ func TestWebTool_WebFetch_UnsupportedScheme(t *testing.T) {
// TestWebTool_WebFetch_MissingURL verifies error handling for missing URL
func TestWebTool_WebFetch_MissingURL(t *testing.T) {
tool := NewWebFetchTool(50000)
tool := &WebFetchTool{maxChars: 50000, skipSSRFCheck: true}
ctx := context.Background()
args := map[string]any{}
@ -145,7 +149,7 @@ func TestWebTool_WebFetch_Truncation(t *testing.T) {
}))
defer server.Close()
tool := NewWebFetchTool(1000) // Limit to 1000 chars
tool := &WebFetchTool{maxChars: 1000, skipSSRFCheck: true} // Limit to 1000 chars
ctx := context.Background()
args := map[string]any{
"url": server.URL,
@ -214,7 +218,7 @@ func TestWebTool_WebFetch_HTMLExtraction(t *testing.T) {
}))
defer server.Close()
tool := NewWebFetchTool(50000)
tool := &WebFetchTool{maxChars: 50000, skipSSRFCheck: true}
ctx := context.Background()
args := map[string]any{
"url": server.URL,
@ -315,7 +319,7 @@ func TestWebFetchTool_extractText(t *testing.T) {
// TestWebTool_WebFetch_MissingDomain verifies error handling for URL without domain
func TestWebTool_WebFetch_MissingDomain(t *testing.T) {
tool := NewWebFetchTool(50000)
tool := &WebFetchTool{maxChars: 50000, skipSSRFCheck: true}
ctx := context.Background()
args := map[string]any{
"url": "https://",
@ -405,3 +409,87 @@ func TestWebTool_TavilySearch_Success(t *testing.T) {
t.Errorf("Expected 'via Tavily' in output, got: %s", result.ForUser)
}
}
// --- SSRF Protection Tests ---
func TestIsPrivateIP(t *testing.T) {
tests := []struct {
ip string
private bool
}{
{"127.0.0.1", true},
{"10.0.0.1", true},
{"172.16.0.1", true},
{"192.168.1.1", true},
{"169.254.169.254", true},
{"::1", true},
{"8.8.8.8", false},
{"1.1.1.1", false},
{"93.184.216.34", false},
}
for _, tt := range tests {
ip := net.ParseIP(tt.ip)
got := isPrivateIP(ip)
assert.Equal(t, tt.private, got, "isPrivateIP(%s)", tt.ip)
}
}
func TestValidateURLSafety_PrivateIPs(t *testing.T) {
tests := []struct {
urlStr string
wantError bool
}{
{"http://127.0.0.1:8080/admin", true},
{"http://10.0.0.1/internal", true},
{"http://192.168.1.1/router", true},
{"http://169.254.169.254/latest/meta-data", true},
{"http://[::1]/path", true},
}
for _, tt := range tests {
parsed, _ := url.Parse(tt.urlStr)
err := validateURLSafety(parsed)
if tt.wantError {
assert.Error(t, err, "expected block for %s", tt.urlStr)
} else {
assert.NoError(t, err, "expected allow for %s", tt.urlStr)
}
}
}
func TestValidateURLSafety_MetadataEndpoints(t *testing.T) {
parsed, _ := url.Parse("http://metadata.google.internal/computeMetadata/v1/")
err := validateURLSafety(parsed)
assert.Error(t, err)
assert.Contains(t, err.Error(), "metadata")
}
func TestWebFetch_SSRF_BlocksPrivateIP(t *testing.T) {
tool := NewWebFetchTool(50000) // SSRF check enabled (default)
ctx := context.Background()
result := tool.Execute(ctx, map[string]any{"url": "http://127.0.0.1:8080/"})
assert.True(t, result.IsError)
assert.Contains(t, result.ForLLM, "blocked")
}
func TestWebFetch_SSRF_BlocksMetadata(t *testing.T) {
tool := NewWebFetchTool(50000)
ctx := context.Background()
result := tool.Execute(ctx, map[string]any{"url": "http://169.254.169.254/latest/meta-data"})
assert.True(t, result.IsError)
assert.Contains(t, result.ForLLM, "blocked")
}
func TestWebFetch_SSRF_AllowsPublicWithSkip(t *testing.T) {
// Verify that skipSSRFCheck allows httptest servers (127.0.0.1)
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.WriteHeader(http.StatusOK)
w.Write([]byte("OK"))
}))
defer server.Close()
tool := &WebFetchTool{maxChars: 50000, skipSSRFCheck: true}
result := tool.Execute(context.Background(), map[string]any{"url": server.URL})
assert.False(t, result.IsError, "skipSSRFCheck should allow localhost: %s", result.ForLLM)
}