sm-golem-plugin/classes/security/ssrf_guard_test.go

package security

import (
	"context"
	"errors"
	"net"
	"net/netip"
	"os"
	"path/filepath"
	"strings"
	"testing"

	"git.golem15.com/golem15/summercms/modules/compass"
	"git.golem15.com/golem15/summercms/modules/fetchguard"
)

// fakeResolver answers from a table; a missing host does not resolve.
type fakeResolver map[string][]string

func (f fakeResolver) LookupIPAddr(_ context.Context, host string) ([]net.IPAddr, error) {
	ips, ok := f[strings.ToLower(host)]
	if !ok {
		return nil, &net.DNSError{Err: "no such host", Name: host, IsNotFound: true}
	}
	out := make([]net.IPAddr, len(ips))
	for i, s := range ips {
		out[i] = net.IPAddr{IP: net.ParseIP(s)}
	}
	return out, nil
}

func withResolver(t *testing.T, r Resolver) {
	t.Helper()
	prev := DefaultResolver
	DefaultResolver = r
	t.Cleanup(func() { DefaultResolver = prev })
}

func message(err error) string {
	var u *UnsafeURLError
	if !errors.As(err, &u) {
		return "<not UnsafeURLError: " + strings.TrimSpace(func() string {
			if err == nil {
				return "nil"
			}
			return err.Error()
		}()) + ">"
	}
	return u.Message
}

// TestSSRFGuard ports SSRFGuardTest.php and adds the allowlist sources.
func TestSSRFGuard(t *testing.T) {
	ctx := context.Background()
	withResolver(t, fakeResolver{
		"oaidalleapiprodscus.blob.core.windows.net": {"20.60.40.1"},
		"api.openai.com":       {"162.159.140.245", "2606:4700::6812:1"},
		"rebind.openai.com":    {"162.159.140.245", "10.0.0.5"},
		"metadata.example.com": {"169.254.169.254"},
	})
	def := DefaultAllowedHosts

	cases := []struct {
		name    string
		url     string
		allowed []string
		want    string
	}{
		{"private ipv4 127", "https://127.0.0.1/x", []string{"127.0.0.1"}, "Resolves to private/loopback IP: 127.0.0.1"},
		{"metadata 169.254", "https://169.254.169.254/latest/meta-data/", []string{"169.254.169.254"}, "Resolves to private/loopback IP: 169.254.169.254"},
		{"ipv6 loopback", "https://[::1]/x", []string{"::1"}, "Resolves to private/loopback IP: ::1"},
		{"non-https scheme", "ftp://oaidalleapiprodscus.blob.core.windows.net/x", def, "Only https:// scheme allowed"},
		{"http scheme", "http://api.openai.com/v1", def, "Only https:// scheme allowed"},
		{"non-allowlisted host", "https://evil.example.com/x", def, "Host not in allowlist: evil.example.com"},
		{"groq is not allowlisted", "https://api.groq.com/openai/v1", def, "Host not in allowlist: api.groq.com"},
		{"anthropic is not allowlisted by default", "https://api.anthropic.com/v1", def, "Host not in allowlist: api.anthropic.com"},
		{"suffix entry needs the dot", "https://evilopenai.com/v1", def, "Host not in allowlist: evilopenai.com"},
		{"no scheme", "api.openai.com/v1", def, "Invalid URL"},
		{"no host", "https:///v1", def, "Invalid URL"},
		{"unresolvable", "https://nxdomain.openai.com/v1", def, "Cannot resolve host: nxdomain.openai.com"},
		{"one private answer is enough", "https://rebind.openai.com/v1", def, "Resolves to private/loopback IP: 10.0.0.5"},
		{"resolves to metadata", "https://metadata.example.com/", []string{"metadata.example.com"}, "Resolves to private/loopback IP: 169.254.169.254"},
	}
	for _, c := range cases {
		t.Run(c.name, func(t *testing.T) {
			if got := message(AssertSafeURL(ctx, c.url, c.allowed)); got != c.want {
				t.Fatalf("AssertSafeURL(%s) = %q, want %q", c.url, got, c.want)
			}
		})
	}

	t.Run("accepts allowlisted public hosts", func(t *testing.T) {
		for _, u := range []string{
			"https://oaidalleapiprodscus.blob.core.windows.net/file.png",
			"https://api.openai.com/v1",
			"HTTPS://API.OPENAI.COM/v1",
		} {
			if err := AssertSafeURL(ctx, u, def); err != nil {
				t.Fatalf("%s: %v", u, err)
			}
		}
		if err := AssertSafeURL(ctx, "https://API.openai.com/v1", []string{".OPENAI.COM"}); err != nil {
			t.Fatalf("case-insensitive suffix: %v", err)
		}
	})

	t.Run("documented private ranges", func(t *testing.T) {
		for _, ip := range []string{"127.0.0.0", "127.0.0.1", "127.255.255.255", "10.0.0.0", "10.255.255.255", "172.16.0.0", "172.31.255.255", "192.168.0.0", "192.168.255.255", "169.254.0.0", "169.254.169.254", "169.254.255.255", "0.0.0.0", "::1", "fe80::1", "fe80::abcd:1234", "fc00::1", "fd12:3456:789a::1", "240.0.0.1"} {
			if err := AssertSafeURL(ctx, "https://"+bracket(ip)+"/", []string{ip}); message(err) != "Resolves to private/loopback IP: "+ip {
				t.Errorf("%s: %v", ip, err)
			}
		}
		for _, ip := range []string{"8.8.8.8", "1.1.1.1", "172.32.0.1", "2001:4860:4860::8888"} {
			if err := AssertSafeURL(ctx, "https://"+bracket(ip)+"/", []string{ip}); err != nil {
				t.Errorf("public %s refused: %v", ip, err)
			}
		}
	})

	t.Run("allowlist sources", func(t *testing.T) {
		dir := t.TempDir()
		if err := os.WriteFile(filepath.Join(dir, "app.yaml"), []byte("name: ssrf\n"), 0o644); err != nil {
			t.Fatal(err)
		}
		cfg, err := compass.Open(compass.Options{Dir: dir, Environ: []string{"SUMMER_ENV=development"}})
		if err != nil {
			t.Fatal(err)
		}
		t.Setenv(EnvAllowedHosts, "")
		if got := AllowedHosts(nil); strings.Join(got, ",") != "oaidalleapiprodscus.blob.core.windows.net,.openai.com" {
			t.Fatalf("default = %v", got)
		}
		if err := cfg.Set(ConfigAllowedHosts, []any{".anthropic.com"}); err != nil {
			t.Fatal(err)
		}
		if got := AllowedHosts(cfg); strings.Join(got, ",") != ".anthropic.com" {
			t.Fatalf("config = %v", got)
		}
		t.Setenv(EnvAllowedHosts, " api.groq.com, ,.anthropic.com ")
		if got := AllowedHosts(cfg); strings.Join(got, ",") != "api.groq.com,.anthropic.com" {
			t.Fatalf("env override = %v", got)
		}
		withResolver(t, fakeResolver{"api.groq.com": {"104.18.2.161"}})
		if err := AssertSafeURL(ctx, "https://api.groq.com/openai/v1", AllowedHosts(cfg)); err != nil {
			t.Fatalf("env-allowlisted host refused: %v", err)
		}
	})

	t.Run("same classification as the dial guard", func(t *testing.T) {
		if !fetchguard.IsPrivateAddr(mustAddr("100.64.0.1")) {
			t.Fatal("carrier-grade NAT is private")
		}
		if message(AssertSafeURL(ctx, "https://100.64.0.1/", []string{"100.64.0.1"})) != "Resolves to private/loopback IP: 100.64.0.1" {
			t.Fatal("the guard must refuse what the dial guard refuses")
		}
	})
}

func bracket(ip string) string {
	if strings.Contains(ip, ":") {
		return "[" + ip + "]"
	}
	return ip
}

func mustAddr(s string) netip.Addr { return netip.MustParseAddr(s) }