Block IPv6 transition addresses in the Assistant fetch guard

isBlockedFetchIP had the same bypass as the audit webhook validator and the
restricted outbound transport fixed in 70d275288: every net.IP predicate it
uses reads only the literal address bytes, so 64:ff9b::a9fe:a9fe fetched
169.254.169.254 while looking like ordinary global unicast.

Reuse securityutil.EmbeddedIPv4Candidates rather than growing a second
AI-local list of transition prefixes, and hold each embedded destination to
the same policy as the outer address so PULSE_AI_ALLOW_LOOPBACK and
PULSE_AI_ALLOW_PRIVATE_IPS keep working through the wrapper.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Richard Courtman
2026-07-31 17:56:35 +01:00
committed by courtmanr@gmail.com
parent d4ce19219c
commit 68f1ba8832
3 changed files with 80 additions and 0 deletions
@@ -6839,3 +6839,20 @@ browser-title suffix, but it must not rename Pulse Assistant, alter Assistant
page context, change tool/provider identity, or enter prompt, transcript, or
Patrol evidence payloads. The AI runtime continues to consume canonical route
and resource context independently of the shell's displayed brand.
The Assistant fetch tool's SSRF guard in `internal/ai/service.go` classifies a
destination by reachability, not by the literal bytes of the address. The
`net.IP` predicates it is built from (`IsLoopback`, `IsUnspecified`,
`IsLinkLocalUnicast`, `IsPrivate`, `IsGlobalUnicast`) each inspect only the
address handed to them, so an IPv6 transition address carries an internal IPv4
destination past all of them: `64:ff9b::a9fe:a9fe` reaches 169.254.169.254
while reporting itself as ordinary global unicast. `isBlockedFetchIP` must
therefore unwrap NAT64, 6to4, Teredo, ISATAP, IPv4-compatible and
IPv4-translated encodings through the shared
`securityutil.EmbeddedIPv4Candidates` helper that the audit-webhook validator
and the restricted outbound transport already use, and must not fork a
second AI-local list of transition prefixes. Every embedded destination is held
to the same policy as the outer address, so `PULSE_AI_ALLOW_LOOPBACK` and
`PULSE_AI_ALLOW_PRIVATE_IPS` relax the embedded check exactly as they relax the
outer one, and a transition address wrapping a permitted public target stays
permitted.
+10
View File
@@ -37,6 +37,7 @@ import (
"github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources"
"github.com/rcourtman/pulse-go-rewrite/pkg/aicontracts"
pkglicensing "github.com/rcourtman/pulse-go-rewrite/pkg/licensing"
"github.com/rcourtman/pulse-go-rewrite/pkg/securityutil"
"github.com/rs/zerolog/log"
)
@@ -3834,6 +3835,15 @@ func isBlockedFetchIP(ip net.IP) bool {
if !ip.IsGlobalUnicast() {
return true
}
// SECURITY: every check above reads the literal address bytes, so an IPv6
// transition address (NAT64, 6to4, Teredo, ISATAP) tunnels to an internal
// IPv4 destination none of them can see. Hold the embedded destination to
// the same policy, including the loopback and private-IP escape hatches.
for _, embedded := range securityutil.EmbeddedIPv4Candidates(ip) {
if isBlockedFetchIP(embedded) {
return true
}
}
return false
}
+53
View File
@@ -138,6 +138,59 @@ func TestIsBlockedFetchIP(t *testing.T) {
}
}
// TestIsBlockedFetchIP_IPv6TransitionAddresses covers the SSRF bypass where a
// NAT64, 6to4, Teredo, ISATAP or IPv4-compatible address carries an internal
// IPv4 destination past every net.IP predicate.
func TestIsBlockedFetchIP_IPv6TransitionAddresses(t *testing.T) {
tests := []struct {
ip string
blocked bool
}{
{"64:ff9b::a9fe:a9fe", true}, // NAT64 -> 169.254.169.254
{"64:ff9b::7f00:1", true}, // NAT64 -> 127.0.0.1
{"64:ff9b::a00:1", true}, // NAT64 -> 10.0.0.1
{"64:ff9b:1::c0a8:1", true}, // NAT64 local-use -> 192.168.0.1
{"64:ff9b:1:a9fe:a9:fe00::", true}, // NAT64 local-use /48 -> 169.254.169.254
{"2002:ac10:1::", true}, // 6to4 -> 172.16.0.1
{"2002:a9fe:a9fe::", true}, // 6to4 -> 169.254.169.254
{"2001:0:a9fe:a9fe::", true}, // Teredo server -> 169.254.169.254
{"2001:db8::5efe:c0a8:101", true}, // ISATAP -> 192.168.1.1
{"::192.168.1.1", true}, // IPv4-compatible -> 192.168.1.1
{"::ffff:0:10.0.0.1", true}, // IPv4-translated -> 10.0.0.1
{"64:ff9b::808:808", false}, // NAT64 -> 8.8.8.8
{"2002:808:808::", false}, // 6to4 -> 8.8.8.8
{"2606:4700:4700::1111", false}, // ordinary global unicast
}
for _, tt := range tests {
ip := net.ParseIP(tt.ip)
if ip == nil {
t.Fatalf("failed to parse %q", tt.ip)
}
if got := isBlockedFetchIP(ip); got != tt.blocked {
t.Errorf("isBlockedFetchIP(%s) = %v, want %v", tt.ip, got, tt.blocked)
}
}
}
func TestIsBlockedFetchIP_IPv6TransitionHonoursEscapeHatches(t *testing.T) {
nat64Private := net.ParseIP("64:ff9b::c0a8:1") // -> 192.168.0.1
nat64Loopback := net.ParseIP("64:ff9b::7f00:1") // -> 127.0.0.1
t.Setenv("PULSE_AI_ALLOW_PRIVATE_IPS", "true")
if isBlockedFetchIP(nat64Private) {
t.Error("NAT64-wrapped private IP should be allowed when PULSE_AI_ALLOW_PRIVATE_IPS=true")
}
if !isBlockedFetchIP(nat64Loopback) {
t.Error("NAT64-wrapped loopback should stay blocked when only private IPs are allowed")
}
t.Setenv("PULSE_AI_ALLOW_LOOPBACK", "true")
if isBlockedFetchIP(nat64Loopback) {
t.Error("NAT64-wrapped loopback should be allowed when PULSE_AI_ALLOW_LOOPBACK=true")
}
}
func TestFetchURL_SizeLimit(t *testing.T) {
os.Setenv("PULSE_AI_ALLOW_LOOPBACK", "true")
defer os.Unsetenv("PULSE_AI_ALLOW_LOOPBACK")