From e7dbcf3d2c881f30969c71c31204127e547e8e02 Mon Sep 17 00:00:00 2001 From: dev_agent Date: Mon, 5 Oct 2026 11:51:49 -0400 Subject: [PATCH 1/4] Add SSRF-safe recipe fetcher internal/recipes.Fetcher fetches https:// URLs only. Its DialContext resolves the host once, refuses unless every answer is public (loopback, RFC 1918, link-local, CGNAT 100.64/10, multicast, unspecified, reserved, IPv6 ULA/link-local/multicast and IPv4-mapped or -embedding forms are blocked) and dials the address it checked, so a rebinding DNS answer cannot swap in a private IP. Keep-alives and environment proxies are off, so every request and redirect hop dials and checks afresh. At most 3 redirects, each re-checked for scheme and address; bodies over 2 MB are an error rather than truncated; 10 s for the whole fetch; a descriptive User-Agent; requests to one host at least 1 s apart; 2xx pages cached by URL for an hour. Tests use a stub resolver and httptest TLS servers. Maestro task: f4414e41-bd99-498d-a613-a90053963249 Co-Authored-By: Claude Opus 5.5 --- docs/ARCHITECTURE.md | 12 +- internal/recipes/fetcher.go | 373 ++++++++++++++++++++++ internal/recipes/fetcher_test.go | 515 +++++++++++++++++++++++++++++++ 3 files changed, 899 insertions(+), 1 deletion(-) create mode 100644 internal/recipes/fetcher.go create mode 100644 internal/recipes/fetcher_test.go diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 0d77b3d..632ecc9 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -208,7 +208,17 @@ automatic** (D-021). - Dinners are grounded in real recipe pages found through recipe search (`searxng` or `brave`, per `recipe_search_settings`). -- Fetching is **SSRF-safe**. +- Fetching is **SSRF-safe** (`recipes.Fetcher`): https:// only, at every + redirect hop. Its dialer resolves the host once and refuses unless every + answer is public — loopback, RFC 1918, link-local, carrier-grade NAT + (100.64/10), multicast, unspecified, reserved, IPv6 unique-local, + link-local and multicast, and IPv4-mapped or IPv4-embedding (NAT64, 6to4, + Teredo) forms are refused — then dials the address it checked, so a + rebinding DNS answer cannot swap in a private one. Keep-alives and + environment proxies are off, so every request dials and checks afresh. At + most 3 redirects; a body over 2 MB is an error, never truncated; 10 s for + the whole fetch; a descriptive `User-Agent`; requests to one host at least + 1 s apart; 2xx pages cached by URL for an hour (64 pages, oldest evicted). - Recipe data is parsed from schema.org `Recipe` JSON-LD. - Photos are cached locally (`photos`) and processed with `golang.org/x/image`. diff --git a/internal/recipes/fetcher.go b/internal/recipes/fetcher.go new file mode 100644 index 0000000..58e18dd --- /dev/null +++ b/internal/recipes/fetcher.go @@ -0,0 +1,373 @@ +// Package recipes fetches and holds real recipe pages that ground dinners. +package recipes + +import ( + "context" + "crypto/tls" + "errors" + "fmt" + "io" + "net" + "net/http" + "net/netip" + "net/url" + "strings" + "sync" + "time" +) + +const ( + // UserAgent identifies MealQuest to the recipe sites it fetches. + UserAgent = "MealQuest/1.0 (household meal planner; recipe fetcher)" + // MaxBodyBytes caps a fetched body. A larger page is an error, never + // silently truncated: a cut-off recipe page would parse as a wrong one. + MaxBodyBytes = 2 << 20 + // MaxRedirects is how many redirects a fetch follows; each hop is + // re-checked for scheme and address. + MaxRedirects = 3 + // Timeout bounds a whole fetch: rate-limit waits, every redirect hop + // and reading the body. + Timeout = 10 * time.Second + // HostInterval is the least time between two requests to one host. + HostInterval = time.Second + // CacheTTL is how long a fetched page is served from the cache. + CacheTTL = time.Hour + // CacheSize is how many pages the cache holds; the oldest is evicted + // first. + CacheSize = 64 +) + +var ( + // ErrNotHTTPS is returned for a URL, or a redirect, that is not + // https://. + ErrNotHTTPS = errors.New("recipes: only https:// URLs are fetched") + // ErrBlockedAddress is returned when a host resolves to an address + // that is not public, such as loopback or a private network. + ErrBlockedAddress = errors.New("recipes: address is not public") + // ErrTooManyRedirects is returned after MaxRedirects redirects. + ErrTooManyRedirects = errors.New("recipes: too many redirects") + // ErrTooLarge is returned for a body over MaxBodyBytes. + ErrTooLarge = errors.New("recipes: body exceeds size cap") + // ErrTimeout is returned when a fetch runs past Timeout. + ErrTimeout = errors.New("recipes: fetch timed out") +) + +// StatusError is returned for a final response whose status is not 2xx. +type StatusError struct { + URL string + Status int +} + +func (e *StatusError) Error() string { + return fmt.Sprintf("recipes: GET %s: HTTP %d", e.URL, e.Status) +} + +// Page is a fetched page. +type Page struct { + // URL is the final URL, after redirects. + URL string + ContentType string + Body []byte +} + +// Resolver looks up a host's addresses; *net.Resolver satisfies it. +type Resolver interface { + LookupIPAddr(ctx context.Context, host string) ([]net.IPAddr, error) +} + +// dialFunc dials an address that has already passed the public-address +// check. +type dialFunc func(ctx context.Context, network, address string) (net.Conn, error) + +// Fetcher retrieves recipe pages without reaching private networks. Its +// dialer resolves each host itself, refuses the connection unless every +// answer is a public address, and dials the checked address, so a second DNS +// answer cannot swap in a private one. Keep-alives are off: every request, +// redirect hops included, dials and checks afresh. Proxies from the +// environment are not used, since a proxy would dial on the Fetcher's +// behalf. A Fetcher is safe for concurrent use. +type Fetcher struct { + client *http.Client + resolver Resolver + dial dialFunc + // timeout bounds a fetch; tests shorten it. + timeout time.Duration + // now is the clock; tests replace it. + now func() time.Time + // sleep waits d or until ctx is done; tests replace it. + sleep func(ctx context.Context, d time.Duration) error + + mu sync.Mutex + nextSlot map[string]time.Time + cache map[string]cacheEntry +} + +type cacheEntry struct { + page Page + fetched time.Time +} + +// NewFetcher returns a Fetcher that resolves through the system resolver. +func NewFetcher() *Fetcher { + return newFetcher(net.DefaultResolver, (&net.Dialer{}).DialContext, nil) +} + +// newFetcher returns a Fetcher over resolver and dial; tlsConfig, when not +// nil, replaces the default TLS settings so tests can trust their servers. +func newFetcher(resolver Resolver, dial dialFunc, tlsConfig *tls.Config) *Fetcher { + f := &Fetcher{ + resolver: resolver, + dial: dial, + timeout: Timeout, + now: time.Now, + sleep: sleep, + nextSlot: map[string]time.Time{}, + cache: map[string]cacheEntry{}, + } + transport := &http.Transport{ + Proxy: nil, + DialContext: f.dialPublic, + TLSClientConfig: tlsConfig, + DisableKeepAlives: true, + ForceAttemptHTTP2: true, + TLSHandshakeTimeout: Timeout, + } + f.client = &http.Client{ + Transport: &limitedTransport{f: f, next: transport}, + CheckRedirect: checkRedirect, + } + return f +} + +// Fetch returns the page at rawURL, from the cache when it was fetched +// within CacheTTL. Only 2xx responses are returned and cached. +func (f *Fetcher) Fetch(ctx context.Context, rawURL string) (Page, error) { + u, err := url.Parse(rawURL) + if err != nil { + return Page{}, fmt.Errorf("recipes: parse URL: %w", err) + } + if err := checkURL(u); err != nil { + return Page{}, err + } + key := u.String() + if page, ok := f.cached(key); ok { + return page, nil + } + + fetchCtx, cancel := context.WithTimeout(ctx, f.timeout) + defer cancel() + page, err := f.get(fetchCtx, key) + if err != nil { + if ctx.Err() == nil && errors.Is(fetchCtx.Err(), context.DeadlineExceeded) { + return Page{}, fmt.Errorf("%w after %s: %s", ErrTimeout, f.timeout, key) + } + return Page{}, err + } + f.store(key, page) + return page, nil +} + +func (f *Fetcher) get(ctx context.Context, rawURL string) (Page, error) { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, rawURL, nil) + if err != nil { + return Page{}, fmt.Errorf("recipes: build request: %w", err) + } + req.Header.Set("User-Agent", UserAgent) + req.Header.Set("Accept", "text/html,application/xhtml+xml") + resp, err := f.client.Do(req) + if err != nil { + return Page{}, err + } + defer resp.Body.Close() + final := resp.Request.URL.String() + if resp.StatusCode < 200 || resp.StatusCode > 299 { + return Page{}, &StatusError{URL: final, Status: resp.StatusCode} + } + if resp.ContentLength > MaxBodyBytes { + return Page{}, fmt.Errorf("%w: %s declares %d bytes", ErrTooLarge, final, resp.ContentLength) + } + body, err := io.ReadAll(io.LimitReader(resp.Body, MaxBodyBytes+1)) + if err != nil { + return Page{}, fmt.Errorf("recipes: read %s: %w", final, err) + } + if len(body) > MaxBodyBytes { + return Page{}, fmt.Errorf("%w: %s is over %d bytes", ErrTooLarge, final, MaxBodyBytes) + } + return Page{URL: final, ContentType: resp.Header.Get("Content-Type"), Body: body}, nil +} + +// cached returns the page cached under key if it is younger than CacheTTL. +func (f *Fetcher) cached(key string) (Page, bool) { + f.mu.Lock() + defer f.mu.Unlock() + e, ok := f.cache[key] + if !ok { + return Page{}, false + } + if f.now().Sub(e.fetched) >= CacheTTL { + delete(f.cache, key) + return Page{}, false + } + return e.page, true +} + +// store caches page under key, evicting the oldest entry when full. +func (f *Fetcher) store(key string, page Page) { + f.mu.Lock() + defer f.mu.Unlock() + if _, ok := f.cache[key]; !ok && len(f.cache) >= CacheSize { + var oldest string + var oldestAt time.Time + for k, e := range f.cache { + if oldest == "" || e.fetched.Before(oldestAt) { + oldest, oldestAt = k, e.fetched + } + } + delete(f.cache, oldest) + } + f.cache[key] = cacheEntry{page: page, fetched: f.now()} +} + +// waitTurn blocks until host's next request slot, keeping requests to one +// host at least HostInterval apart. +func (f *Fetcher) waitTurn(ctx context.Context, host string) error { + f.mu.Lock() + now := f.now() + slot := f.nextSlot[host] + if slot.Before(now) { + slot = now + } + f.nextSlot[host] = slot.Add(HostInterval) + f.mu.Unlock() + if wait := slot.Sub(now); wait > 0 { + return f.sleep(ctx, wait) + } + return nil +} + +// dialPublic resolves address's host once, refuses unless every answer is +// public, then dials the checked addresses in turn. +func (f *Fetcher) dialPublic(ctx context.Context, network, address string) (net.Conn, error) { + host, port, err := net.SplitHostPort(address) + if err != nil { + return nil, fmt.Errorf("recipes: dial address %q: %w", address, err) + } + var addrs []netip.Addr + if ip, err := netip.ParseAddr(host); err == nil { + addrs = []netip.Addr{ip} + } else { + answers, err := f.resolver.LookupIPAddr(ctx, host) + if err != nil { + return nil, fmt.Errorf("recipes: resolve %s: %w", host, err) + } + for _, a := range answers { + ip, ok := netip.AddrFromSlice(a.IP) + if !ok { + return nil, fmt.Errorf("recipes: resolve %s: malformed address %v", host, a.IP) + } + addrs = append(addrs, ip) + } + } + if len(addrs) == 0 { + return nil, fmt.Errorf("recipes: resolve %s: no addresses", host) + } + for _, ip := range addrs { + if !IsPublic(ip) { + return nil, fmt.Errorf("%w: %s resolves to %s", ErrBlockedAddress, host, ip) + } + } + var errs []error + for _, ip := range addrs { + conn, err := f.dial(ctx, network, net.JoinHostPort(ip.Unmap().WithZone("").String(), port)) + if err == nil { + return conn, nil + } + errs = append(errs, err) + } + return nil, errors.Join(errs...) +} + +// limitedTransport waits for each request's per-host rate-limit slot, +// redirect hops included, before sending it. +type limitedTransport struct { + f *Fetcher + next http.RoundTripper +} + +func (t *limitedTransport) RoundTrip(req *http.Request) (*http.Response, error) { + if err := t.f.waitTurn(req.Context(), strings.ToLower(req.URL.Hostname())); err != nil { + return nil, err + } + return t.next.RoundTrip(req) +} + +// checkURL refuses anything but an absolute https:// URL with a host. +func checkURL(u *url.URL) error { + if u.Scheme != "https" { + return fmt.Errorf("%w: %q", ErrNotHTTPS, u.Redacted()) + } + if u.Hostname() == "" { + return fmt.Errorf("recipes: URL %q has no host", u.Redacted()) + } + return nil +} + +// checkRedirect applies checkURL to every hop and stops after +// MaxRedirects. The hop's address is checked when it is dialed. +func checkRedirect(req *http.Request, via []*http.Request) error { + if len(via) > MaxRedirects { + return fmt.Errorf("%w: more than %d", ErrTooManyRedirects, MaxRedirects) + } + return checkURL(req.URL) +} + +// blockedPrefixes are the non-public ranges. IPv4-mapped IPv6 addresses +// are unmapped before the check, so the IPv4 entries cover them too. +var blockedPrefixes = []netip.Prefix{ + netip.MustParsePrefix("0.0.0.0/8"), // unspecified, "this network" + netip.MustParsePrefix("10.0.0.0/8"), // RFC 1918 + netip.MustParsePrefix("100.64.0.0/10"), // carrier-grade NAT + netip.MustParsePrefix("127.0.0.0/8"), // loopback + netip.MustParsePrefix("169.254.0.0/16"), // link-local + netip.MustParsePrefix("172.16.0.0/12"), // RFC 1918 + netip.MustParsePrefix("192.168.0.0/16"), // RFC 1918 + netip.MustParsePrefix("224.0.0.0/4"), // multicast + netip.MustParsePrefix("240.0.0.0/4"), // reserved, broadcast + netip.MustParsePrefix("::/96"), // unspecified, loopback, IPv4-compatible + netip.MustParsePrefix("fc00::/7"), // unique local + netip.MustParsePrefix("fe80::/10"), // link-local + netip.MustParsePrefix("fec0::/10"), // site-local (deprecated) + netip.MustParsePrefix("ff00::/8"), // multicast + netip.MustParsePrefix("64:ff9b::/96"), // NAT64, embeds any IPv4 + netip.MustParsePrefix("64:ff9b:1::/48"), // local-use NAT64 + netip.MustParsePrefix("2002::/16"), // 6to4, embeds any IPv4 + netip.MustParsePrefix("2001::/32"), // Teredo, embeds any IPv4 + netip.MustParsePrefix("::ffff:0:0:0/96"), // IPv4-translated +} + +// IsPublic reports whether ip is a public unicast address the Fetcher may +// dial. +func IsPublic(ip netip.Addr) bool { + if !ip.IsValid() { + return false + } + ip = ip.Unmap().WithZone("") + for _, p := range blockedPrefixes { + if p.Contains(ip) { + return false + } + } + return true +} + +// sleep waits d or until ctx is done, returning ctx's error if it is. +func sleep(ctx context.Context, d time.Duration) error { + t := time.NewTimer(d) + defer t.Stop() + select { + case <-t.C: + return nil + case <-ctx.Done(): + return ctx.Err() + } +} diff --git a/internal/recipes/fetcher_test.go b/internal/recipes/fetcher_test.go new file mode 100644 index 0000000..f1c2f53 --- /dev/null +++ b/internal/recipes/fetcher_test.go @@ -0,0 +1,515 @@ +package recipes + +import ( + "context" + "crypto/tls" + "crypto/x509" + "errors" + "fmt" + "net" + "net/http" + "net/http/httptest" + "net/netip" + "strings" + "sync" + "testing" + "time" +) + +// stubResolver answers each host from a list of answers, one per lookup; +// the last answer repeats. +type stubResolver struct { + mu sync.Mutex + answers map[string][][]string + calls map[string]int +} + +func (r *stubResolver) LookupIPAddr(_ context.Context, host string) ([]net.IPAddr, error) { + r.mu.Lock() + defer r.mu.Unlock() + seq, ok := r.answers[host] + if !ok { + return nil, fmt.Errorf("stub: no such host %s", host) + } + n := r.calls[host] + r.calls[host]++ + if n >= len(seq) { + n = len(seq) - 1 + } + var out []net.IPAddr + for _, s := range seq[n] { + out = append(out, net.IPAddr{IP: net.ParseIP(s)}) + } + return out, nil +} + +func (r *stubResolver) lookups(host string) int { + r.mu.Lock() + defer r.mu.Unlock() + return r.calls[host] +} + +// testEnv is a Fetcher whose public test addresses all reach one httptest +// TLS server, and which records every address it dials. +type testEnv struct { + f *Fetcher + resolver *stubResolver + server *httptest.Server + + mu sync.Mutex + dialed []string +} + +// publicIP is a documentation address the tests treat as a public host. +const publicIP = "203.0.113.10" + +func newTestEnv(t *testing.T, handler http.Handler, answers map[string][][]string) *testEnv { + t.Helper() + server := httptest.NewTLSServer(handler) + t.Cleanup(server.Close) + pool := x509.NewCertPool() + pool.AddCert(server.Certificate()) + env := &testEnv{ + resolver: &stubResolver{answers: answers, calls: map[string]int{}}, + server: server, + } + dial := func(ctx context.Context, network, address string) (net.Conn, error) { + env.mu.Lock() + env.dialed = append(env.dialed, address) + env.mu.Unlock() + host, _, err := net.SplitHostPort(address) + if err != nil { + return nil, err + } + if strings.HasPrefix(host, "203.0.113.") { + var d net.Dialer + return d.DialContext(ctx, network, server.Listener.Addr().String()) + } + return nil, fmt.Errorf("test dial: no route to %s", address) + } + env.f = newFetcher(env.resolver, dial, &tls.Config{RootCAs: pool}) + // Real requests in these tests are not rate-limited unless a test + // installs a fake clock. + env.f.sleep = func(context.Context, time.Duration) error { return nil } + return env +} + +func (e *testEnv) dials() []string { + e.mu.Lock() + defer e.mu.Unlock() + return append([]string(nil), e.dialed...) +} + +func publicAnswers() map[string][][]string { + return map[string][][]string{"example.com": {{publicIP}}} +} + +func TestFetchReturnsPage(t *testing.T) { + t.Parallel() + var gotUA string + env := newTestEnv(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + gotUA = r.Header.Get("User-Agent") + w.Header().Set("Content-Type", "text/html") + fmt.Fprint(w, "tacos") + }), publicAnswers()) + + page, err := env.f.Fetch(t.Context(), "https://example.com/tacos") + if err != nil { + t.Fatalf("Fetch: %v", err) + } + if string(page.Body) != "tacos" || page.ContentType != "text/html" || page.URL != "https://example.com/tacos" { + t.Errorf("page = %+v", page) + } + if gotUA != UserAgent { + t.Errorf("User-Agent = %q, want %q", gotUA, UserAgent) + } + if got := env.dials(); len(got) != 1 || got[0] != publicIP+":443" { + t.Errorf("dialed %v, want [%s:443]", got, publicIP) + } +} + +func TestFetchRejectsNonHTTPS(t *testing.T) { + t.Parallel() + env := newTestEnv(t, http.NotFoundHandler(), publicAnswers()) + for _, raw := range []string{"http://example.com/tacos", "ftp://example.com/", "example.com/tacos", "https:///tacos"} { + _, err := env.f.Fetch(t.Context(), raw) + if err == nil { + t.Errorf("Fetch(%q) succeeded", raw) + } + if strings.HasPrefix(raw, "http://") && !errors.Is(err, ErrNotHTTPS) { + t.Errorf("Fetch(%q) error = %v, want ErrNotHTTPS", raw, err) + } + } + if got := env.dials(); len(got) != 0 { + t.Errorf("dialed %v, want nothing", got) + } + if n := env.resolver.lookups("example.com"); n != 0 { + t.Errorf("resolved example.com %d times, want 0", n) + } +} + +// blockedIPs has an address in every refused range, including IPv4-mapped +// forms of the IPv4 ranges. +var blockedIPs = []string{ + "127.0.0.1", "127.255.255.254", "::1", "::ffff:127.0.0.1", // loopback + "10.0.0.1", "172.16.0.1", "172.31.255.255", "192.168.1.1", // RFC 1918 + "::ffff:10.1.2.3", "::ffff:172.20.0.1", "::ffff:192.168.0.10", + "169.254.169.254", "::ffff:169.254.169.254", "fe80::1", // link-local + "100.64.0.1", "100.127.255.254", "::ffff:100.100.0.1", // carrier-grade NAT + "224.0.0.1", "239.255.255.250", "ff02::1", "::ffff:224.0.0.251", // multicast + "0.0.0.0", "0.1.2.3", "::", "::ffff:0.0.0.0", // unspecified + "fc00::1", "fd12:3456:789a::1", // unique local + "255.255.255.255", "240.0.0.1", "fec0::1", + "64:ff9b::a00:1", "2002:a00:1::1", "2001:0:a00:1::1", "::a00:1", +} + +func TestIsPublic(t *testing.T) { + t.Parallel() + for _, s := range blockedIPs { + if IsPublic(netip.MustParseAddr(s)) { + t.Errorf("IsPublic(%s) = true, want false", s) + } + } + for _, s := range []string{"8.8.8.8", "203.0.113.10", "100.63.255.255", "100.128.0.1", "172.32.0.1", "2606:4700::1111", "::ffff:8.8.8.8"} { + if !IsPublic(netip.MustParseAddr(s)) { + t.Errorf("IsPublic(%s) = false, want true", s) + } + } + if IsPublic(netip.Addr{}) { + t.Error("IsPublic(zero Addr) = true, want false") + } +} + +func TestFetchRefusesEachBlockedRange(t *testing.T) { + t.Parallel() + for _, ip := range blockedIPs { + t.Run(ip, func(t *testing.T) { + t.Parallel() + env := newTestEnv(t, http.NotFoundHandler(), map[string][][]string{"example.com": {{ip}}}) + _, err := env.f.Fetch(t.Context(), "https://example.com/") + if !errors.Is(err, ErrBlockedAddress) { + t.Fatalf("Fetch error = %v, want ErrBlockedAddress", err) + } + if got := env.dials(); len(got) != 0 { + t.Errorf("dialed %v, want nothing", got) + } + }) + } +} + +func TestFetchRefusesBlockedLiteralHost(t *testing.T) { + t.Parallel() + env := newTestEnv(t, http.NotFoundHandler(), nil) + for _, raw := range []string{"https://127.0.0.1/", "https://[::1]/", "https://[::ffff:10.0.0.1]/", "https://169.254.169.254/latest/meta-data/", "https://[fe80::1%25eth0]/"} { + if _, err := env.f.Fetch(t.Context(), raw); !errors.Is(err, ErrBlockedAddress) { + t.Errorf("Fetch(%q) error = %v, want ErrBlockedAddress", raw, err) + } + } + if got := env.dials(); len(got) != 0 { + t.Errorf("dialed %v, want nothing", got) + } +} + +func TestFetchRefusesMixedAnswer(t *testing.T) { + t.Parallel() + env := newTestEnv(t, http.NotFoundHandler(), map[string][][]string{"example.com": {{publicIP, "10.0.0.1"}}}) + if _, err := env.f.Fetch(t.Context(), "https://example.com/"); !errors.Is(err, ErrBlockedAddress) { + t.Fatalf("Fetch error = %v, want ErrBlockedAddress", err) + } + if got := env.dials(); len(got) != 0 { + t.Errorf("dialed %v, want nothing", got) + } +} + +func TestFetchRefusesRedirectToPrivateHost(t *testing.T) { + t.Parallel() + answers := map[string][][]string{ + "example.com": {{publicIP}}, + "evil.example.com": {{"10.0.0.5"}}, + } + for _, target := range []string{"https://evil.example.com/admin", "https://127.0.0.1/admin", "https://[::ffff:192.168.0.1]/"} { + t.Run(target, func(t *testing.T) { + t.Parallel() + env := newTestEnv(t, http.RedirectHandler(target, http.StatusFound), answers) + _, err := env.f.Fetch(t.Context(), "https://example.com/tacos") + if !errors.Is(err, ErrBlockedAddress) { + t.Fatalf("Fetch error = %v, want ErrBlockedAddress", err) + } + if got := env.dials(); len(got) != 1 || got[0] != publicIP+":443" { + t.Errorf("dialed %v, want only the first hop", got) + } + }) + } +} + +func TestFetchRefusesRedirectToHTTP(t *testing.T) { + t.Parallel() + env := newTestEnv(t, http.RedirectHandler("http://example.com/plain", http.StatusMovedPermanently), publicAnswers()) + if _, err := env.f.Fetch(t.Context(), "https://example.com/"); !errors.Is(err, ErrNotHTTPS) { + t.Fatalf("Fetch error = %v, want ErrNotHTTPS", err) + } +} + +// chainHandler redirects /hop/N to /hop/N-1 until /hop/0, which serves a +// page. +var chainHandler = http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var n int + if _, err := fmt.Sscanf(r.URL.Path, "/hop/%d", &n); err != nil { + http.NotFound(w, r) + return + } + if n == 0 { + fmt.Fprint(w, "arrived") + return + } + http.Redirect(w, r, fmt.Sprintf("https://recipes.example.com/hop/%d", n-1), http.StatusFound) +}) + +func TestFetchFollowsAtMostThreeRedirects(t *testing.T) { + t.Parallel() + answers := map[string][][]string{ + "example.com": {{publicIP}}, + "recipes.example.com": {{"203.0.113.20"}}, + } + env := newTestEnv(t, chainHandler, answers) + page, err := env.f.Fetch(t.Context(), "https://example.com/hop/3") + if err != nil { + t.Fatalf("Fetch with 3 redirects: %v", err) + } + if page.URL != "https://recipes.example.com/hop/0" || string(page.Body) != "arrived" { + t.Errorf("page = %+v", page) + } + // Every hop is dialed, so every hop's address is checked. + if got := env.dials(); len(got) != 4 || got[0] != publicIP+":443" || got[3] != "203.0.113.20:443" { + t.Errorf("dialed %v, want 4 checked hops", got) + } + + if _, err := env.f.Fetch(t.Context(), "https://example.com/hop/4"); !errors.Is(err, ErrTooManyRedirects) { + t.Fatalf("Fetch with 4 redirects error = %v, want ErrTooManyRedirects", err) + } +} + +// TestFetchRefusesDNSSwitchToPrivate covers DNS rebinding: the host's first +// answer is public and every later one private. The first fetch dials the +// address it checked; the next lookup's private answer is refused, and the +// private address is never dialed. +func TestFetchRefusesDNSSwitchToPrivate(t *testing.T) { + t.Parallel() + answers := map[string][][]string{"example.com": {{publicIP}, {"127.0.0.1"}}} + env := newTestEnv(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + fmt.Fprint(w, "ok") + }), answers) + + if _, err := env.f.Fetch(t.Context(), "https://example.com/one"); err != nil { + t.Fatalf("first Fetch: %v", err) + } + if n := env.resolver.lookups("example.com"); n != 1 { + t.Fatalf("first Fetch resolved %d times, want 1 (checked IP must be dialed IP)", n) + } + _, err := env.f.Fetch(t.Context(), "https://example.com/two") + if !errors.Is(err, ErrBlockedAddress) { + t.Fatalf("second Fetch error = %v, want ErrBlockedAddress", err) + } + if got := env.dials(); len(got) != 1 || got[0] != publicIP+":443" { + t.Errorf("dialed %v, want only %s:443", got, publicIP) + } +} + +func TestFetchSizeCap(t *testing.T) { + t.Parallel() + env := newTestEnv(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + size := MaxBodyBytes + if r.URL.Path != "/exact" { + size++ + } + if r.URL.Path == "/declared" { + w.Header().Set("Content-Length", fmt.Sprint(size)) + } + w.Write([]byte(strings.Repeat("a", size))) + }), publicAnswers()) + + page, err := env.f.Fetch(t.Context(), "https://example.com/exact") + if err != nil { + t.Fatalf("Fetch at the cap: %v", err) + } + if len(page.Body) != MaxBodyBytes { + t.Errorf("body is %d bytes, want %d", len(page.Body), MaxBodyBytes) + } + for _, path := range []string{"/declared", "/chunked"} { + if _, err := env.f.Fetch(t.Context(), "https://example.com"+path); !errors.Is(err, ErrTooLarge) { + t.Errorf("Fetch %s error = %v, want ErrTooLarge", path, err) + } + } +} + +func TestFetchTimeout(t *testing.T) { + t.Parallel() + env := newTestEnv(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/slow-body" { + w.Write([]byte("start")) + w.(http.Flusher).Flush() + } + <-r.Context().Done() + }), publicAnswers()) + env.f.timeout = 100 * time.Millisecond + + for _, path := range []string{"/slow-headers", "/slow-body"} { + start := time.Now() + _, err := env.f.Fetch(t.Context(), "https://example.com"+path) + if !errors.Is(err, ErrTimeout) { + t.Errorf("Fetch %s error = %v, want ErrTimeout", path, err) + } + if elapsed := time.Since(start); elapsed > 5*time.Second { + t.Errorf("Fetch %s took %s", path, elapsed) + } + } + if Timeout != 10*time.Second { + t.Errorf("Timeout = %s, want 10s", Timeout) + } +} + +func TestFetchCallerCancelIsNotTimeout(t *testing.T) { + t.Parallel() + env := newTestEnv(t, http.NotFoundHandler(), publicAnswers()) + ctx, cancel := context.WithCancel(t.Context()) + cancel() + _, err := env.f.Fetch(ctx, "https://example.com/") + if !errors.Is(err, context.Canceled) || errors.Is(err, ErrTimeout) { + t.Fatalf("Fetch error = %v, want context.Canceled", err) + } +} + +// fakeClock is a clock whose sleep advances it and records the wait. +type fakeClock struct { + mu sync.Mutex + t time.Time + sleeps []time.Duration +} + +func (c *fakeClock) now() time.Time { + c.mu.Lock() + defer c.mu.Unlock() + return c.t +} + +func (c *fakeClock) sleep(_ context.Context, d time.Duration) error { + c.mu.Lock() + defer c.mu.Unlock() + c.sleeps = append(c.sleeps, d) + c.t = c.t.Add(d) + return nil +} + +func (c *fakeClock) advance(d time.Duration) { + c.mu.Lock() + defer c.mu.Unlock() + c.t = c.t.Add(d) +} + +func (c *fakeClock) waits() []time.Duration { + c.mu.Lock() + defer c.mu.Unlock() + return append([]time.Duration(nil), c.sleeps...) +} + +func installClock(f *Fetcher) *fakeClock { + c := &fakeClock{t: time.Date(2026, 10, 5, 18, 0, 0, 0, time.UTC)} + f.now = c.now + f.sleep = c.sleep + return c +} + +func TestFetchCachesByURL(t *testing.T) { + t.Parallel() + var mu sync.Mutex + hits := map[string]int{} + env := newTestEnv(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + mu.Lock() + hits[r.URL.Path]++ + n := hits[r.URL.Path] + mu.Unlock() + if r.URL.Path == "/missing" { + http.NotFound(w, r) + return + } + fmt.Fprintf(w, "%s #%d", r.URL.Path, n) + }), publicAnswers()) + clock := installClock(env.f) + + fetch := func(path string) string { + t.Helper() + page, err := env.f.Fetch(t.Context(), "https://example.com"+path) + if err != nil { + t.Fatalf("Fetch %s: %v", path, err) + } + return string(page.Body) + } + if got := fetch("/a"); got != "/a #1" { + t.Errorf("first /a = %q", got) + } + if got := fetch("/a"); got != "/a #1" { + t.Errorf("cached /a = %q, want the first response", got) + } + if got := fetch("/b"); got != "/b #1" { + t.Errorf("/b = %q, want its own response", got) + } + clock.advance(CacheTTL) + if got := fetch("/a"); got != "/a #2" { + t.Errorf("/a after CacheTTL = %q, want a refetch", got) + } + + for range 2 { + var se *StatusError + if _, err := env.f.Fetch(t.Context(), "https://example.com/missing"); !errors.As(err, &se) || se.Status != http.StatusNotFound { + t.Fatalf("Fetch /missing error = %v, want StatusError 404", err) + } + } + mu.Lock() + defer mu.Unlock() + if hits["/missing"] != 2 { + t.Errorf("/missing was requested %d times, want 2 (errors are not cached)", hits["/missing"]) + } +} + +func TestCacheEvictsOldest(t *testing.T) { + t.Parallel() + f := newFetcher(&stubResolver{}, nil, nil) + clock := installClock(f) + for i := range CacheSize + 1 { + f.store(fmt.Sprintf("https://example.com/%d", i), Page{}) + clock.advance(time.Second) + } + if len(f.cache) != CacheSize { + t.Errorf("cache holds %d pages, want %d", len(f.cache), CacheSize) + } + if _, ok := f.cached("https://example.com/0"); ok { + t.Error("oldest page was not evicted") + } + if _, ok := f.cached(fmt.Sprintf("https://example.com/%d", CacheSize)); !ok { + t.Error("newest page is missing") + } +} + +func TestFetchRateLimitsPerHost(t *testing.T) { + t.Parallel() + answers := map[string][][]string{ + "example.com": {{publicIP}}, + "recipes.example.com": {{"203.0.113.20"}}, + } + env := newTestEnv(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + fmt.Fprint(w, "ok") + }), answers) + clock := installClock(env.f) + + for _, raw := range []string{"https://example.com/1", "https://recipes.example.com/1", "https://example.com/2", "https://example.com/3"} { + if _, err := env.f.Fetch(t.Context(), raw); err != nil { + t.Fatalf("Fetch %s: %v", raw, err) + } + } + // example.com/2 waits a full interval; example.com/3 waits another. + // recipes.example.com has its own slot and never waits. + got := clock.waits() + if len(got) != 2 || got[0] != HostInterval || got[1] != HostInterval { + t.Errorf("waits = %v, want [%s %s]", got, HostInterval, HostInterval) + } +} -- 2.55.0 From 177b2dd2e78125706fd9a1822835bed18e16e164 Mon Sep 17 00:00:00 2001 From: dev_agent Date: Mon, 5 Oct 2026 12:01:12 -0400 Subject: [PATCH 2/4] Harden recipe fetcher: public-only check, cache copies, rate-limit slots - IsPublic now refuses every non-globally-reachable IPv4 special-purpose range (adds 192.0.0.0/24, TEST-NET-1/2/3, 192.88.99.0/24, benchmarking 198.18.0.0/15) and admits IPv6 only from global unicast 2000::/3, minus its documentation, 6to4, Teredo and IETF protocol-assignment ranges. Test fixtures move off TEST-NET-3 to public addresses; Fetch-level tests cover the new blocked ranges and the public IPv6 path. - The cache stores and returns copies of Page.Body, so callers cannot alter cached bytes or race with other readers. - The per-host rate limiter advances only when a request is let through; a wait that ends with its context holds no slot. Co-Authored-By: Claude Opus 5.5 --- docs/ARCHITECTURE.md | 14 +++-- internal/recipes/fetcher.go | 103 ++++++++++++++++++++----------- internal/recipes/fetcher_test.go | 97 ++++++++++++++++++++++++++--- 3 files changed, 165 insertions(+), 49 deletions(-) diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 632ecc9..b71272f 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -210,15 +210,19 @@ automatic** (D-021). (`searxng` or `brave`, per `recipe_search_settings`). - Fetching is **SSRF-safe** (`recipes.Fetcher`): https:// only, at every redirect hop. Its dialer resolves the host once and refuses unless every - answer is public — loopback, RFC 1918, link-local, carrier-grade NAT - (100.64/10), multicast, unspecified, reserved, IPv6 unique-local, - link-local and multicast, and IPv4-mapped or IPv4-embedding (NAT64, 6to4, - Teredo) forms are refused — then dials the address it checked, so a + answer is public — every non-globally-reachable IPv4 special-purpose + range (loopback, RFC 1918, link-local, carrier-grade NAT, documentation + TEST-NETs, benchmarking 198.18/15, multicast, reserved and more) is + refused, IPv4-mapped forms included, and IPv6 must be global unicast + (2000::/3) outside documentation, 6to4, Teredo and other special ranges — + then dials the address it checked, so a rebinding DNS answer cannot swap in a private one. Keep-alives and environment proxies are off, so every request dials and checks afresh. At most 3 redirects; a body over 2 MB is an error, never truncated; 10 s for the whole fetch; a descriptive `User-Agent`; requests to one host at least - 1 s apart; 2xx pages cached by URL for an hour (64 pages, oldest evicted). + 1 s apart, counted from requests actually sent (an abandoned wait holds no + slot); 2xx pages cached by URL for an hour (64 pages, oldest evicted), and + callers get their own copy of a cached body. - Recipe data is parsed from schema.org `Recipe` JSON-LD. - Photos are cached locally (`photos`) and processed with `golang.org/x/image`. diff --git a/internal/recipes/fetcher.go b/internal/recipes/fetcher.go index 58e18dd..3c680a5 100644 --- a/internal/recipes/fetcher.go +++ b/internal/recipes/fetcher.go @@ -2,6 +2,7 @@ package recipes import ( + "bytes" "context" "crypto/tls" "errors" @@ -97,8 +98,10 @@ type Fetcher struct { // sleep waits d or until ctx is done; tests replace it. sleep func(ctx context.Context, d time.Duration) error - mu sync.Mutex - nextSlot map[string]time.Time + mu sync.Mutex + // lastSent is when each host's latest request was let through; only + // requests actually sent advance it. + lastSent map[string]time.Time cache map[string]cacheEntry } @@ -121,7 +124,7 @@ func newFetcher(resolver Resolver, dial dialFunc, tlsConfig *tls.Config) *Fetche timeout: Timeout, now: time.Now, sleep: sleep, - nextSlot: map[string]time.Time{}, + lastSent: map[string]time.Time{}, cache: map[string]cacheEntry{}, } transport := &http.Transport{ @@ -140,7 +143,8 @@ func newFetcher(resolver Resolver, dial dialFunc, tlsConfig *tls.Config) *Fetche } // Fetch returns the page at rawURL, from the cache when it was fetched -// within CacheTTL. Only 2xx responses are returned and cached. +// within CacheTTL. Only 2xx responses are returned and cached. The +// returned Body is the caller's own copy. func (f *Fetcher) Fetch(ctx context.Context, rawURL string) (Page, error) { u, err := url.Parse(rawURL) if err != nil { @@ -208,11 +212,15 @@ func (f *Fetcher) cached(key string) (Page, bool) { delete(f.cache, key) return Page{}, false } - return e.page, true + page := e.page + page.Body = bytes.Clone(page.Body) + return page, true } -// store caches page under key, evicting the oldest entry when full. +// store caches a copy of page under key, evicting the oldest entry when +// full. func (f *Fetcher) store(key string, page Page) { + page.Body = bytes.Clone(page.Body) f.mu.Lock() defer f.mu.Unlock() if _, ok := f.cache[key]; !ok && len(f.cache) >= CacheSize { @@ -228,21 +236,26 @@ func (f *Fetcher) store(key string, page Page) { f.cache[key] = cacheEntry{page: page, fetched: f.now()} } -// waitTurn blocks until host's next request slot, keeping requests to one -// host at least HostInterval apart. +// waitTurn blocks until a request to host may be sent, keeping requests to +// one host at least HostInterval apart. A wait holds no reservation: the +// host's slot advances only when a request is let through, so a wait that +// ends with ctx does not delay later requests. func (f *Fetcher) waitTurn(ctx context.Context, host string) error { - f.mu.Lock() - now := f.now() - slot := f.nextSlot[host] - if slot.Before(now) { - slot = now + for { + f.mu.Lock() + now := f.now() + last, sent := f.lastSent[host] + wait := last.Add(HostInterval).Sub(now) + if !sent || wait <= 0 { + f.lastSent[host] = now + f.mu.Unlock() + return nil + } + f.mu.Unlock() + if err := f.sleep(ctx, wait); err != nil { + return err + } } - f.nextSlot[host] = slot.Add(HostInterval) - f.mu.Unlock() - if wait := slot.Sub(now); wait > 0 { - return f.sleep(ctx, wait) - } - return nil } // dialPublic resolves address's host once, refuses unless every answer is @@ -321,38 +334,58 @@ func checkRedirect(req *http.Request, via []*http.Request) error { return checkURL(req.URL) } -// blockedPrefixes are the non-public ranges. IPv4-mapped IPv6 addresses -// are unmapped before the check, so the IPv4 entries cover them too. -var blockedPrefixes = []netip.Prefix{ - netip.MustParsePrefix("0.0.0.0/8"), // unspecified, "this network" +// blockedIPv4 are the IPv4 special-purpose ranges that are not globally +// reachable (IANA IPv4 Special-Purpose Address Registry). IPv4-mapped IPv6 +// addresses are unmapped before the check, so these cover them too. +var blockedIPv4 = []netip.Prefix{ + netip.MustParsePrefix("0.0.0.0/8"), // "this network" netip.MustParsePrefix("10.0.0.0/8"), // RFC 1918 netip.MustParsePrefix("100.64.0.0/10"), // carrier-grade NAT netip.MustParsePrefix("127.0.0.0/8"), // loopback netip.MustParsePrefix("169.254.0.0/16"), // link-local netip.MustParsePrefix("172.16.0.0/12"), // RFC 1918 + netip.MustParsePrefix("192.0.0.0/24"), // IETF protocol assignments + netip.MustParsePrefix("192.0.2.0/24"), // documentation, TEST-NET-1 + netip.MustParsePrefix("192.88.99.0/24"), // 6to4 relay anycast (deprecated) netip.MustParsePrefix("192.168.0.0/16"), // RFC 1918 + netip.MustParsePrefix("198.18.0.0/15"), // benchmarking + netip.MustParsePrefix("198.51.100.0/24"), // documentation, TEST-NET-2 + netip.MustParsePrefix("203.0.113.0/24"), // documentation, TEST-NET-3 netip.MustParsePrefix("224.0.0.0/4"), // multicast netip.MustParsePrefix("240.0.0.0/4"), // reserved, broadcast - netip.MustParsePrefix("::/96"), // unspecified, loopback, IPv4-compatible - netip.MustParsePrefix("fc00::/7"), // unique local - netip.MustParsePrefix("fe80::/10"), // link-local - netip.MustParsePrefix("fec0::/10"), // site-local (deprecated) - netip.MustParsePrefix("ff00::/8"), // multicast - netip.MustParsePrefix("64:ff9b::/96"), // NAT64, embeds any IPv4 - netip.MustParsePrefix("64:ff9b:1::/48"), // local-use NAT64 - netip.MustParsePrefix("2002::/16"), // 6to4, embeds any IPv4 - netip.MustParsePrefix("2001::/32"), // Teredo, embeds any IPv4 - netip.MustParsePrefix("::ffff:0:0:0/96"), // IPv4-translated +} + +// globalIPv6 is the only IPv6 range dialed: global unicast. Everything +// outside it is refused, including ::, ::1, IPv4-compatible and +// IPv4-translated forms, NAT64 (64:ff9b::/96), discard-only (100::/64), +// unique local (fc00::/7), link-local (fe80::/10) and multicast (ff00::/8). +var globalIPv6 = netip.MustParsePrefix("2000::/3") + +// blockedIPv6 are the special-purpose ranges inside global unicast that are +// not public or embed an arbitrary IPv4 address. +var blockedIPv6 = []netip.Prefix{ + netip.MustParsePrefix("2001::/23"), // IETF protocol assignments, Teredo, benchmarking + netip.MustParsePrefix("2001:db8::/32"), // documentation + netip.MustParsePrefix("2002::/16"), // 6to4, embeds any IPv4 + netip.MustParsePrefix("3fff::/20"), // documentation } // IsPublic reports whether ip is a public unicast address the Fetcher may -// dial. +// dial. IPv4 addresses are public unless in a special-purpose range; IPv6 +// addresses must be global unicast and outside its special-purpose ranges. func IsPublic(ip netip.Addr) bool { if !ip.IsValid() { return false } ip = ip.Unmap().WithZone("") - for _, p := range blockedPrefixes { + blocked := blockedIPv4 + if ip.Is6() { + if !globalIPv6.Contains(ip) { + return false + } + blocked = blockedIPv6 + } + for _, p := range blocked { if p.Contains(ip) { return false } diff --git a/internal/recipes/fetcher_test.go b/internal/recipes/fetcher_test.go index f1c2f53..de72215 100644 --- a/internal/recipes/fetcher_test.go +++ b/internal/recipes/fetcher_test.go @@ -60,8 +60,13 @@ type testEnv struct { dialed []string } -// publicIP is a documentation address the tests treat as a public host. -const publicIP = "203.0.113.10" +// publicIP, publicIP2 and publicIPv6 are public addresses; the test dialer +// routes them to the test server and never reaches the real hosts. +const ( + publicIP = "93.184.215.14" + publicIP2 = "93.184.215.20" + publicIPv6 = "2606:2800:21f::10" +) func newTestEnv(t *testing.T, handler http.Handler, answers map[string][][]string) *testEnv { t.Helper() @@ -81,7 +86,7 @@ func newTestEnv(t *testing.T, handler http.Handler, answers map[string][][]strin if err != nil { return nil, err } - if strings.HasPrefix(host, "203.0.113.") { + if host == publicIP || host == publicIP2 || host == publicIPv6 { var d net.Dialer return d.DialContext(ctx, network, server.Listener.Addr().String()) } @@ -128,6 +133,19 @@ func TestFetchReturnsPage(t *testing.T) { } } +func TestFetchDialsPublicIPv6(t *testing.T) { + t.Parallel() + env := newTestEnv(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + fmt.Fprint(w, "ok") + }), map[string][][]string{"example.com": {{publicIPv6}}}) + if _, err := env.f.Fetch(t.Context(), "https://example.com/"); err != nil { + t.Fatalf("Fetch: %v", err) + } + if got := env.dials(); len(got) != 1 || got[0] != "["+publicIPv6+"]:443" { + t.Errorf("dialed %v, want [[%s]:443]", got, publicIPv6) + } +} + func TestFetchRejectsNonHTTPS(t *testing.T) { t.Parallel() env := newTestEnv(t, http.NotFoundHandler(), publicAnswers()) @@ -159,8 +177,15 @@ var blockedIPs = []string{ "224.0.0.1", "239.255.255.250", "ff02::1", "::ffff:224.0.0.251", // multicast "0.0.0.0", "0.1.2.3", "::", "::ffff:0.0.0.0", // unspecified "fc00::1", "fd12:3456:789a::1", // unique local - "255.255.255.255", "240.0.0.1", "fec0::1", - "64:ff9b::a00:1", "2002:a00:1::1", "2001:0:a00:1::1", "::a00:1", + "255.255.255.255", "240.0.0.1", "fec0::1", // reserved, site-local + "192.0.0.1", "192.0.0.255", // IETF protocol assignments + "192.0.2.1", "198.51.100.7", "203.0.113.10", "::ffff:203.0.113.10", // documentation + "192.88.99.1", // 6to4 relay anycast + "198.18.0.1", "198.19.255.255", "::ffff:198.18.0.1", // benchmarking + "2001:db8::1", "3fff::1", // IPv6 documentation + "2001:2::1", "100::1", // IPv6 benchmarking, discard-only + "::ffff:0:a00:1", // IPv4-translated + "64:ff9b::a00:1", "64:ff9b:1::a00:1", "2002:a00:1::1", "2001:0:a00:1::1", "::a00:1", } func TestIsPublic(t *testing.T) { @@ -170,7 +195,12 @@ func TestIsPublic(t *testing.T) { t.Errorf("IsPublic(%s) = true, want false", s) } } - for _, s := range []string{"8.8.8.8", "203.0.113.10", "100.63.255.255", "100.128.0.1", "172.32.0.1", "2606:4700::1111", "::ffff:8.8.8.8"} { + for _, s := range []string{ + "8.8.8.8", publicIP, "::ffff:8.8.8.8", "100.63.255.255", "100.128.0.1", "172.32.0.1", + "192.0.1.255", "192.0.3.1", "192.88.98.255", "198.17.255.255", "198.20.0.1", + "198.51.99.255", "203.0.112.255", "203.0.114.1", "223.255.255.254", + "2606:4700::1111", publicIPv6, "2001:4860:4860::8888", "2a00:1450::1", + } { if !IsPublic(netip.MustParseAddr(s)) { t.Errorf("IsPublic(%s) = false, want true", s) } @@ -269,7 +299,7 @@ func TestFetchFollowsAtMostThreeRedirects(t *testing.T) { t.Parallel() answers := map[string][][]string{ "example.com": {{publicIP}}, - "recipes.example.com": {{"203.0.113.20"}}, + "recipes.example.com": {{publicIP2}}, } env := newTestEnv(t, chainHandler, answers) page, err := env.f.Fetch(t.Context(), "https://example.com/hop/3") @@ -280,7 +310,7 @@ func TestFetchFollowsAtMostThreeRedirects(t *testing.T) { t.Errorf("page = %+v", page) } // Every hop is dialed, so every hop's address is checked. - if got := env.dials(); len(got) != 4 || got[0] != publicIP+":443" || got[3] != "203.0.113.20:443" { + if got := env.dials(); len(got) != 4 || got[0] != publicIP+":443" || got[3] != publicIP2+":443" { t.Errorf("dialed %v, want 4 checked hops", got) } @@ -494,7 +524,7 @@ func TestFetchRateLimitsPerHost(t *testing.T) { t.Parallel() answers := map[string][][]string{ "example.com": {{publicIP}}, - "recipes.example.com": {{"203.0.113.20"}}, + "recipes.example.com": {{publicIP2}}, } env := newTestEnv(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { fmt.Fprint(w, "ok") @@ -513,3 +543,52 @@ func TestFetchRateLimitsPerHost(t *testing.T) { t.Errorf("waits = %v, want [%s %s]", got, HostInterval, HostInterval) } } + +func TestFetchCacheIsNotSharedWithCallers(t *testing.T) { + t.Parallel() + env := newTestEnv(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + fmt.Fprint(w, "tacos") + }), publicAnswers()) + + for range 2 { + page, err := env.f.Fetch(t.Context(), "https://example.com/tacos") + if err != nil { + t.Fatalf("Fetch: %v", err) + } + if string(page.Body) != "tacos" { + t.Fatalf("body = %q, want the cached page unchanged", page.Body) + } + // Overwrite the returned body: neither the fetched nor the cached + // copy may reach the cache. + copy(page.Body, "XXXXX") + } +} + +// TestFetchCanceledWaitHoldsNoSlot covers a burst of fetches whose +// rate-limit waits end with their context: they must not push back the +// next request that is allowed through. +func TestFetchCanceledWaitHoldsNoSlot(t *testing.T) { + t.Parallel() + env := newTestEnv(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + fmt.Fprint(w, "ok") + }), publicAnswers()) + clock := installClock(env.f) + + if _, err := env.f.Fetch(t.Context(), "https://example.com/first"); err != nil { + t.Fatalf("first Fetch: %v", err) + } + env.f.sleep = func(context.Context, time.Duration) error { return context.DeadlineExceeded } + for i := range 5 { + if _, err := env.f.Fetch(t.Context(), fmt.Sprintf("https://example.com/burst/%d", i)); !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("burst Fetch %d error = %v, want context.DeadlineExceeded", i, err) + } + } + env.f.sleep = clock.sleep + clock.advance(HostInterval) + if _, err := env.f.Fetch(t.Context(), "https://example.com/after"); err != nil { + t.Fatalf("Fetch after the burst: %v", err) + } + if got := clock.waits(); len(got) != 0 { + t.Errorf("waits = %v, want none: canceled waits held slots", got) + } +} -- 2.55.0 From b013b542e3669e76366fabe83a3f605100b75422 Mon Sep 17 00:00:00 2001 From: dev_agent Date: Mon, 5 Oct 2026 12:12:37 -0400 Subject: [PATCH 3/4] Hand back rate-limit slots of requests that are never sent waitTurn now checks the context before taking a host's slot, so a request canceled before Fetch takes none. The request let through takes the slot and gets a release; limitedTransport calls it when the round trip fails before the request headers are written (traced with httptrace), restoring the host's previous send time and waking requests waiting for their turn. Adds TestFetchUnsentRequestHoldsNoSlot (canceled before Fetch and failed dial on a host with no previous request; the next request does not wait) and TestReleaseWakesWaiter. Co-Authored-By: Claude Opus 5.5 --- internal/recipes/fetcher.go | 82 +++++++++++++++++++++++++++----- internal/recipes/fetcher_test.go | 61 ++++++++++++++++++++++++ 2 files changed, 131 insertions(+), 12 deletions(-) diff --git a/internal/recipes/fetcher.go b/internal/recipes/fetcher.go index 3c680a5..1b1cc11 100644 --- a/internal/recipes/fetcher.go +++ b/internal/recipes/fetcher.go @@ -10,10 +10,12 @@ import ( "io" "net" "net/http" + "net/http/httptrace" "net/netip" "net/url" "strings" "sync" + "sync/atomic" "time" ) @@ -99,9 +101,12 @@ type Fetcher struct { sleep func(ctx context.Context, d time.Duration) error mu sync.Mutex - // lastSent is when each host's latest request was let through; only - // requests actually sent advance it. + // lastSent is when each host's latest request was let through; a + // request that is never sent hands its slot back. lastSent map[string]time.Time + // released is closed, then replaced, whenever a slot is handed back, + // waking requests waiting for their turn. + released chan struct{} cache map[string]cacheEntry } @@ -125,6 +130,7 @@ func newFetcher(resolver Resolver, dial dialFunc, tlsConfig *tls.Config) *Fetche now: time.Now, sleep: sleep, lastSent: map[string]time.Time{}, + released: make(chan struct{}), cache: map[string]cacheEntry{}, } transport := &http.Transport{ @@ -237,11 +243,15 @@ func (f *Fetcher) store(key string, page Page) { } // waitTurn blocks until a request to host may be sent, keeping requests to -// one host at least HostInterval apart. A wait holds no reservation: the -// host's slot advances only when a request is let through, so a wait that -// ends with ctx does not delay later requests. -func (f *Fetcher) waitTurn(ctx context.Context, host string) error { +// one host at least HostInterval apart. A wait holds no reservation, and a +// request ending with ctx before its turn takes no slot. The request let +// through takes the host's slot and must call release if it is not sent, +// so that it delays no other request. +func (f *Fetcher) waitTurn(ctx context.Context, host string) (release func(), err error) { for { + if err := ctx.Err(); err != nil { + return nil, err + } f.mu.Lock() now := f.now() last, sent := f.lastSent[host] @@ -249,15 +259,55 @@ func (f *Fetcher) waitTurn(ctx context.Context, host string) error { if !sent || wait <= 0 { f.lastSent[host] = now f.mu.Unlock() - return nil + return func() { f.release(host, now, last, sent) }, nil } + released := f.released f.mu.Unlock() - if err := f.sleep(ctx, wait); err != nil { - return err + if err := f.waitOrRelease(ctx, wait, released); err != nil { + return nil, err } } } +// release hands back the slot taken at taken for a request that was not +// sent, restoring the host's previous send time, and wakes waiting requests. +func (f *Fetcher) release(host string, taken, prev time.Time, hadPrev bool) { + f.mu.Lock() + defer f.mu.Unlock() + if !f.lastSent[host].Equal(taken) { + // A later request already holds the slot. + return + } + if hadPrev { + f.lastSent[host] = prev + } else { + delete(f.lastSent, host) + } + close(f.released) + f.released = make(chan struct{}) +} + +// waitOrRelease sleeps for d, returning early without error when a slot is +// handed back so the caller can look again. +func (f *Fetcher) waitOrRelease(ctx context.Context, d time.Duration, released <-chan struct{}) error { + sleepCtx, cancel := context.WithCancel(ctx) + defer cancel() + go func() { + select { + case <-released: + cancel() + case <-sleepCtx.Done(): + } + }() + err := f.sleep(sleepCtx, d) + select { + case <-released: + return nil + default: + return err + } +} + // dialPublic resolves address's host once, refuses unless every answer is // public, then dials the checked addresses in turn. func (f *Fetcher) dialPublic(ctx context.Context, network, address string) (net.Conn, error) { @@ -301,17 +351,25 @@ func (f *Fetcher) dialPublic(ctx context.Context, network, address string) (net. } // limitedTransport waits for each request's per-host rate-limit slot, -// redirect hops included, before sending it. +// redirect hops included, before sending it. A request that fails before +// its headers are written was not sent and hands its slot back. type limitedTransport struct { f *Fetcher next http.RoundTripper } func (t *limitedTransport) RoundTrip(req *http.Request) (*http.Response, error) { - if err := t.f.waitTurn(req.Context(), strings.ToLower(req.URL.Hostname())); err != nil { + release, err := t.f.waitTurn(req.Context(), strings.ToLower(req.URL.Hostname())) + if err != nil { return nil, err } - return t.next.RoundTrip(req) + var sent atomic.Bool + trace := &httptrace.ClientTrace{WroteHeaders: func() { sent.Store(true) }} + resp, err := t.next.RoundTrip(req.WithContext(httptrace.WithClientTrace(req.Context(), trace))) + if err != nil && !sent.Load() { + release() + } + return resp, err } // checkURL refuses anything but an absolute https:// URL with a host. diff --git a/internal/recipes/fetcher_test.go b/internal/recipes/fetcher_test.go index de72215..180943d 100644 --- a/internal/recipes/fetcher_test.go +++ b/internal/recipes/fetcher_test.go @@ -592,3 +592,64 @@ func TestFetchCanceledWaitHoldsNoSlot(t *testing.T) { t.Errorf("waits = %v, want none: canceled waits held slots", got) } } + +// TestFetchUnsentRequestHoldsNoSlot covers requests to a host with no +// previous request that are never sent — canceled before Fetch, or failing +// to connect — and checks that the next request is not delayed. +func TestFetchUnsentRequestHoldsNoSlot(t *testing.T) { + t.Parallel() + const unroutableIP = "93.184.215.99" + answers := map[string][][]string{"example.com": {{unroutableIP}, {publicIP}}} + env := newTestEnv(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + fmt.Fprint(w, "ok") + }), answers) + clock := installClock(env.f) + + ctx, cancel := context.WithCancel(t.Context()) + cancel() + if _, err := env.f.Fetch(ctx, "https://example.com/canceled"); !errors.Is(err, context.Canceled) { + t.Fatalf("canceled Fetch error = %v, want context.Canceled", err) + } + if _, err := env.f.Fetch(t.Context(), "https://example.com/unroutable"); err == nil || !strings.Contains(err.Error(), "no route") { + t.Fatalf("unroutable Fetch error = %v, want a dial failure", err) + } + if _, err := env.f.Fetch(t.Context(), "https://example.com/next"); err != nil { + t.Fatalf("next Fetch: %v", err) + } + if got := clock.waits(); len(got) != 0 { + t.Errorf("waits = %v, want none: unsent requests held slots", got) + } + if got := env.dials(); len(got) != 2 { + t.Errorf("dials = %v, want the unroutable and the next request only", got) + } +} + +// TestReleaseWakesWaiter checks that a request waiting behind a slot that +// is handed back goes ahead at once instead of sleeping out the interval. +func TestReleaseWakesWaiter(t *testing.T) { + t.Parallel() + f := newFetcher(&stubResolver{}, nil, nil) + release, err := f.waitTurn(t.Context(), "example.com") + if err != nil { + t.Fatalf("first waitTurn: %v", err) + } + done := make(chan error, 1) + start := time.Now() + go func() { + _, err := f.waitTurn(t.Context(), "example.com") + done <- err + }() + time.Sleep(50 * time.Millisecond) + release() + select { + case err := <-done: + if err != nil { + t.Fatalf("waiting waitTurn: %v", err) + } + if waited := time.Since(start); waited >= HostInterval { + t.Errorf("waiter took %s, want less than %s", waited, HostInterval) + } + case <-time.After(HostInterval / 2): + t.Fatal("waiter still waiting after the slot was handed back") + } +} -- 2.55.0 From 317e69c6ce2121f75a9026c9a74e93851c702974 Mon Sep 17 00:00:00 2001 From: dev_agent Date: Mon, 5 Oct 2026 12:28:04 -0400 Subject: [PATCH 4/4] Rate-limit recipe fetches when the request is written The per-host slot was taken in RoundTrip, before the transport resolved, dialed and completed TLS, so a request stalled in setup could be sent less than HostInterval after a later request to the same host. The transport now dials through dialTLS, which resolves, dials and handshakes under the request's context and returns a turnConn. The turnConn's first Write, which is the start of the request, waits for the host's slot under the Fetcher's mutex; a first Write that sends nothing hands the slot back. ALPN offers only HTTP/1.1 and keep-alives stay off, so each connection carries exactly one request. TestFetchRateLimitsSendsNotSetup stalls one request in DNS while a second is sent, and checks the sends are a full interval apart on a fake clock. Co-Authored-By: Claude Opus 5.5 --- internal/recipes/fetcher.go | 133 ++++++++++++++++++++++--------- internal/recipes/fetcher_test.go | 66 +++++++++++++++ 2 files changed, 162 insertions(+), 37 deletions(-) diff --git a/internal/recipes/fetcher.go b/internal/recipes/fetcher.go index 1b1cc11..5f7791a 100644 --- a/internal/recipes/fetcher.go +++ b/internal/recipes/fetcher.go @@ -10,12 +10,10 @@ import ( "io" "net" "net/http" - "net/http/httptrace" "net/netip" "net/url" "strings" "sync" - "sync/atomic" "time" ) @@ -88,11 +86,14 @@ type dialFunc func(ctx context.Context, network, address string) (net.Conn, erro // answer cannot swap in a private one. Keep-alives are off: every request, // redirect hops included, dials and checks afresh. Proxies from the // environment are not used, since a proxy would dial on the Fetcher's -// behalf. A Fetcher is safe for concurrent use. +// behalf. Requests are HTTP/1.1, so each connection carries exactly one +// request and its first write is that request. A Fetcher is safe for +// concurrent use. type Fetcher struct { - client *http.Client - resolver Resolver - dial dialFunc + client *http.Client + resolver Resolver + dial dialFunc + tlsConfig *tls.Config // timeout bounds a fetch; tests shorten it. timeout time.Duration // now is the clock; tests replace it. @@ -101,8 +102,9 @@ type Fetcher struct { sleep func(ctx context.Context, d time.Duration) error mu sync.Mutex - // lastSent is when each host's latest request was let through; a - // request that is never sent hands its slot back. + // lastSent is when each host's latest request began to be written to + // its connection; a request whose write fails before any byte is sent + // hands its slot back. lastSent map[string]time.Time // released is closed, then replaced, whenever a slot is handed back, // waking requests waiting for their turn. @@ -124,25 +126,23 @@ func NewFetcher() *Fetcher { // nil, replaces the default TLS settings so tests can trust their servers. func newFetcher(resolver Resolver, dial dialFunc, tlsConfig *tls.Config) *Fetcher { f := &Fetcher{ - resolver: resolver, - dial: dial, - timeout: Timeout, - now: time.Now, - sleep: sleep, - lastSent: map[string]time.Time{}, - released: make(chan struct{}), - cache: map[string]cacheEntry{}, + resolver: resolver, + dial: dial, + tlsConfig: tlsConfig, + timeout: Timeout, + now: time.Now, + sleep: sleep, + lastSent: map[string]time.Time{}, + released: make(chan struct{}), + cache: map[string]cacheEntry{}, } transport := &http.Transport{ - Proxy: nil, - DialContext: f.dialPublic, - TLSClientConfig: tlsConfig, - DisableKeepAlives: true, - ForceAttemptHTTP2: true, - TLSHandshakeTimeout: Timeout, + Proxy: nil, + DialTLSContext: f.dialTLS, + DisableKeepAlives: true, } f.client = &http.Client{ - Transport: &limitedTransport{f: f, next: transport}, + Transport: &requestTransport{next: transport}, CheckRedirect: checkRedirect, } return f @@ -245,8 +245,10 @@ func (f *Fetcher) store(key string, page Page) { // waitTurn blocks until a request to host may be sent, keeping requests to // one host at least HostInterval apart. A wait holds no reservation, and a // request ending with ctx before its turn takes no slot. The request let -// through takes the host's slot and must call release if it is not sent, -// so that it delays no other request. +// through takes the host's slot and must be sent at once, or call release +// if sending fails, so that it delays no other request. Fetch calls it from +// turnConn's first Write, so the slot is taken after DNS, dialing and the +// TLS handshake, as the request is written. func (f *Fetcher) waitTurn(ctx context.Context, host string) (release func(), err error) { for { if err := ctx.Err(); err != nil { @@ -350,26 +352,83 @@ func (f *Fetcher) dialPublic(ctx context.Context, network, address string) (net. return nil, errors.Join(errs...) } -// limitedTransport waits for each request's per-host rate-limit slot, -// redirect hops included, before sending it. A request that fails before -// its headers are written was not sent and hands its slot back. -type limitedTransport struct { - f *Fetcher +// requestCtxKey carries a request's own context into its dial. The +// transport dials with a context that keeps the request's values but not +// its cancellation; the connection and the rate-limit wait must end with +// the request. +type requestCtxKey struct{} + +// requestTransport hands each request's context, redirect hops included, +// to dialTLS. +type requestTransport struct { next http.RoundTripper } -func (t *limitedTransport) RoundTrip(req *http.Request) (*http.Response, error) { - release, err := t.f.waitTurn(req.Context(), strings.ToLower(req.URL.Hostname())) +func (t *requestTransport) RoundTrip(req *http.Request) (*http.Response, error) { + ctx := req.Context() + return t.next.RoundTrip(req.WithContext(context.WithValue(ctx, requestCtxKey{}, ctx))) +} + +// dialTLS connects to address through dialPublic and completes the TLS +// handshake, all under the request's context, and returns a turnConn whose +// first write waits for the host's rate-limit slot. ALPN offers only +// HTTP/1.1, so the connection carries this one request. +func (f *Fetcher) dialTLS(dialCtx context.Context, network, address string) (net.Conn, error) { + ctx, ok := dialCtx.Value(requestCtxKey{}).(context.Context) + if !ok { + return nil, fmt.Errorf("recipes: dial %s outside a request", address) + } + host, _, err := net.SplitHostPort(address) + if err != nil { + return nil, fmt.Errorf("recipes: dial address %q: %w", address, err) + } + conn, err := f.dialPublic(ctx, network, address) if err != nil { return nil, err } - var sent atomic.Bool - trace := &httptrace.ClientTrace{WroteHeaders: func() { sent.Store(true) }} - resp, err := t.next.RoundTrip(req.WithContext(httptrace.WithClientTrace(req.Context(), trace))) - if err != nil && !sent.Load() { + cfg := &tls.Config{} + if f.tlsConfig != nil { + cfg = f.tlsConfig.Clone() + } + cfg.ServerName = host + cfg.NextProtos = []string{"http/1.1"} + tc := tls.Client(conn, cfg) + if err := tc.HandshakeContext(ctx); err != nil { + conn.Close() + return nil, fmt.Errorf("recipes: TLS handshake with %s: %w", host, err) + } + return &turnConn{Conn: tc, f: f, ctx: ctx, host: strings.ToLower(host)}, nil +} + +// turnConn is a connection ready to carry one request. Its first Write — +// the start of the request — waits for the host's rate-limit slot, so two +// requests to one host are written at least HostInterval apart however +// long each spent resolving, dialing and in the TLS handshake. A first +// Write that sends nothing hands the slot back. +type turnConn struct { + net.Conn + f *Fetcher + ctx context.Context + host string + // started is set by the first Write; the transport writes a request + // from one goroutine, so it needs no lock. + started bool +} + +func (c *turnConn) Write(p []byte) (int, error) { + if c.started { + return c.Conn.Write(p) + } + release, err := c.f.waitTurn(c.ctx, c.host) + if err != nil { + return 0, err + } + c.started = true + n, err := c.Conn.Write(p) + if n == 0 && err != nil { release() } - return resp, err + return n, err } // checkURL refuses anything but an absolute https:// URL with a host. diff --git a/internal/recipes/fetcher_test.go b/internal/recipes/fetcher_test.go index 180943d..a210fc8 100644 --- a/internal/recipes/fetcher_test.go +++ b/internal/recipes/fetcher_test.go @@ -653,3 +653,69 @@ func TestReleaseWakesWaiter(t *testing.T) { t.Fatal("waiter still waiting after the slot was handed back") } } + +// stallingResolver holds its first lookup until unstall is closed, then +// answers from the stub. +type stallingResolver struct { + *stubResolver + once sync.Once + stalled chan struct{} + unstall chan struct{} +} + +func (r *stallingResolver) LookupIPAddr(ctx context.Context, host string) ([]net.IPAddr, error) { + first := false + r.once.Do(func() { first = true }) + if first { + close(r.stalled) + <-r.unstall + } + return r.stubResolver.LookupIPAddr(ctx, host) +} + +// TestFetchRateLimitsSendsNotSetup covers a request that stalls in DNS +// while a later request to the same host is sent: the stalled request must +// still be sent a full HostInterval after the other one, because the slot +// is taken when the request is written, not when it starts. +func TestFetchRateLimitsSendsNotSetup(t *testing.T) { + t.Parallel() + var clock *fakeClock + var mu sync.Mutex + sentAt := map[string]time.Time{} + env := newTestEnv(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + mu.Lock() + sentAt[r.URL.Path] = clock.now() + mu.Unlock() + fmt.Fprint(w, "ok") + }), publicAnswers()) + resolver := &stallingResolver{ + stubResolver: env.resolver, + stalled: make(chan struct{}), + unstall: make(chan struct{}), + } + env.f.resolver = resolver + clock = installClock(env.f) + + slow := make(chan error, 1) + go func() { + _, err := env.f.Fetch(t.Context(), "https://example.com/slow") + slow <- err + }() + <-resolver.stalled + if _, err := env.f.Fetch(t.Context(), "https://example.com/fast"); err != nil { + t.Fatalf("fast Fetch: %v", err) + } + close(resolver.unstall) + if err := <-slow; err != nil { + t.Fatalf("slow Fetch: %v", err) + } + + mu.Lock() + defer mu.Unlock() + if gap := sentAt["/slow"].Sub(sentAt["/fast"]); gap < HostInterval { + t.Errorf("slow request sent %s after fast, want at least %s", gap, HostInterval) + } + if got := clock.waits(); len(got) != 1 || got[0] != HostInterval { + t.Errorf("waits = %v, want [%s]: only the stalled request waits", got, HostInterval) + } +} -- 2.55.0