ragflow/internal/agent/tool/http_helper.go

//
//  Copyright 2026 The InfiniFlow Authors. All Rights Reserved.
//
//  Licensed under the Apache License, Version 2.0 (the "License");
//  you may not use this file except in compliance with the License.
//  You may obtain a copy of the License at
//
//      http://www.apache.org/licenses/LICENSE-2.0
//
//  Unless required by applicable law or agreed to in writing, software
//  distributed under the License is distributed on an "AS IS" BASIS,
//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
//  See the License for the specific language governing permissions and
//  limitations under the License.
//

// Package tool implements RAGFlow agent canvas tool adapters in Go.
//
// All tools implement eino's tool.InvokableTool interface and are
// intended to be registered with the agent canvas via a factory function
// (see .claude/plans/agent-go-port.md §2.11.4 "Tool 关键统一模式").
package tool

import (
	"bytes"
	"context"
	"errors"
	"fmt"
	"io"
	"math/rand/v2"
	"net"
	"net/http"
	neturl "net/url"
	"time"

	"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
)

// RetryConfig controls the HTTPHelper retry policy. The defaults
// (3 attempts, 200ms base, 3s max backoff) apply when the
// corresponding field is zero. HTTPHelper{} and NewHTTPHelper()
// are interchangeable.
type RetryConfig struct {
	// MaxAttempts is the total number of attempts (including the first one).
	// Values < 1 are clamped to 1 (no retry). Default 3.
	MaxAttempts int
	// BaseBackoff is the initial backoff between attempts. Default 200ms.
	BaseBackoff time.Duration
	// MaxBackoff caps the exponential backoff. Default 3s.
	MaxBackoff time.Duration
}

func (c RetryConfig) withDefaults() RetryConfig {
	if c.MaxAttempts < 1 {
		c.MaxAttempts = 3
	}
	if c.BaseBackoff <= 0 {
		c.BaseBackoff = 200 * time.Millisecond
	}
	if c.MaxBackoff <= 0 {
		c.MaxBackoff = 3 * time.Second
	}
	return c
}

// HTTPHelper is a context-aware HTTP client shared by the HTTP
// tools. It wraps http.Client with otelhttp.NewTransport for
// OTel span propagation, enforces a 30s default timeout, and
// retries idempotent failures (5xx + network errors) per the
// RetryConfig.
//
// HTTPHelper is safe for concurrent use by multiple goroutines.
type HTTPHelper struct {
	// baseTransport is the source-of-truth *http.Transport. Do wraps it
	// with otelhttp and stores the result in h.client. DoPinned clones it
	// and installs a pinned dialer (see pinnedDialer) so the connect
	// goes to a known IP regardless of what the request's URL says.
	// Tests may mutate baseTransport directly to inject a custom
	// TLSClientConfig (e.g. an in-memory RootCAs pool) — the change
	// applies to both Do and DoPinned.
	baseTransport *http.Transport

	// client is the RoundTripper used by Do. It is an OTel-wrapped view
	// of baseTransport. WithClient replaces this field; DoPinned always
	// derives its client from baseTransport regardless, so a custom
	// client installed via WithClient does NOT participate in
	// DNS-rebinding pinning. Pinning needs the dialer to be replaced at
	// the transport layer and that is only possible if we own the
	// transport — hence the baseTransport/client split.
	client *http.Client

	retry RetryConfig
}

// NewHTTPHelper returns a HTTPHelper with default retry/timeout settings.
func NewHTTPHelper() *HTTPHelper {
	base := &http.Transport{
		Proxy: http.ProxyFromEnvironment,
		DialContext: (&net.Dialer{
			Timeout:   30 * time.Second,
			KeepAlive: 30 * time.Second,
		}).DialContext,
		ForceAttemptHTTP2:     true,
		MaxIdleConns:          100,
		IdleConnTimeout:       90 * time.Second,
		TLSHandshakeTimeout:   10 * time.Second,
		ExpectContinueTimeout: 1 * time.Second,
	}
	return &HTTPHelper{
		baseTransport: base,
		client: &http.Client{
			Timeout:   30 * time.Second,
			Transport: otelhttp.NewTransport(base),
		},
		retry: RetryConfig{}.withDefaults(),
	}
}

// NewHTTPHelperWithRetry returns a HTTPHelper with a custom retry policy.
// Pass RetryConfig{} to use defaults.
func NewHTTPHelperWithRetry(rc RetryConfig) *HTTPHelper {
	h := NewHTTPHelper()
	h.retry = rc.withDefaults()
	return h
}

// WithClient replaces the underlying http.Client used by Do. Useful for
// tests that supply a pre-configured transport (e.g. for OTel recording).
//
// Note: WithClient affects Do calls only. DoPinned always derives its
// client from the helper's internal baseTransport, so a custom client
// installed here does NOT participate in DNS-rebinding pinning. To
// customise pinning behaviour (RootCAs, etc.) mutate baseTransport
// directly — typically only tests do this.
func (h *HTTPHelper) WithClient(c *http.Client) *HTTPHelper {
	if c != nil {
		h.client = c
	}
	return h
}

// Do issues an HTTP request and returns the response. The caller is
// responsible for closing the returned response body.
//
// body and contentType may be empty (e.g. for GET). When body is non-empty
// and contentType is empty, "application/octet-stream" is assumed so servers
// that sniff the body still behave sensibly.
//
// Retry policy:
//   - 5xx responses: retried
//   - network errors (connection refused, DNS, etc.): retried
//   - 4xx responses: NOT retried (caller error, won't help to retry)
//   - 2xx / 3xx: returned as-is
//
// The context is honored on every attempt; cancellation aborts the loop.
func (h *HTTPHelper) Do(
	ctx context.Context,
	method, url, body, contentType string,
	headers map[string]string,
) (*http.Response, error) {
	return h.doRawWithClient(ctx, h.client, method, url, body, contentType, headers)
}

// DoPinned is identical to Do but dials pinnedIP for the host in url,
// while leaving the request URL host untouched. Use immediately after a
// successful ResolveAndValidate (ssrf.go) to defeat DNS-rebinding: an
// attacker cannot swap a public A record for a private one between the
// validation lookup and the connect, because the connect goes to the IP
// we resolved here and the URL host is never re-resolved by the
// transport (Go's *http.Transport only re-resolves when the URL's host
// is used in the dial, and even then the pinned dialer intercepts).
//
// The TLS handshake ServerName is derived from the URL host
// (Go's http.Transport populates tls.Config.ServerName from
// req.URL.Host before calling DialContext), so SNI sends originalHost
// and the cert is verified against originalHost — not against the IP.
// This is critical: rewriting u.Host to the IP (the previous version of
// this fix) sent the IP as SNI and required an IP SAN in the cert,
// which is not what real HTTPS sites have. Pinning must happen at the
// transport layer, not by mutating the request URL.
//
// originalHost and pinnedIP are required; an empty value for either
// returns an error so misuse is loud rather than silently falling back
// to a non-pinned connection. The URL host must also equal originalHost
// (defense in depth — a caller that passes a mismatched URL would
// produce broken TLS and we refuse rather than silently deliver it).
//
// Note on proxies: the pinned transport has Proxy disabled, regardless
// of baseTransport.Proxy (e.g. HTTP_PROXY/HTTPS_PROXY env). A proxy
// would re-resolve the hostname, re-opening the rebinding window the
// SSRF guard just closed — and the pinned dialer would otherwise
// rewrite the proxy's own dial target to pinnedIP:proxyPort, breaking
// the connection. Operators who want proxied outbound should use Do.
func (h *HTTPHelper) DoPinned(
	ctx context.Context,
	method, url, body, contentType string,
	headers map[string]string,
	originalHost string,
	pinnedIP net.IP,
) (*http.Response, error) {
	if originalHost == "" || pinnedIP == nil {
		return nil, errors.New("http_helper: DoPinned requires originalHost and pinnedIP")
	}
	u, err := neturl.Parse(url)
	if err != nil {
		return nil, fmt.Errorf("http_helper: parse url: %w", err)
	}
	if u.Hostname() != originalHost {
		return nil, fmt.Errorf(
			"http_helper: DoPinned url host %q != originalHost %q "+
				"(refusing to pin — would break TLS SNI / cert check)",
			u.Hostname(), originalHost)
	}
	return h.doRawWithClient(ctx, h.pinnedClient(pinnedIP), method, url, body, contentType, headers)
}

// pinnedClient builds a one-shot *http.Client whose transport dials
// pinnedIP:port instead of originalHost:port. The transport is cloned
// from baseTransport so the rest of the connection behaviour (TLS
// config, idle pool) is identical to a non-pinned call.
//
// The proxy setting is explicitly disabled. Two reasons:
//
//  1. *http.Transport dials the proxy first, and pinnedDialer would
//     rewrite the proxy's own address to pinnedIP:proxyPort — that
//     would redirect the proxy dial to a port on the pinned IP and
//     fail connection establishment in any proxied deployment.
//
//  2. Even if the dialer were proxy-aware, the proxy itself receives
//     the request URL (with the original hostname) and re-resolves
//     it, which re-opens the DNS-rebinding window the SSRF guard
//     just closed. The pinned dialer must be the only thing that
//     decides where the TCP connection goes.
//
// Operators who want proxied outbound for pinned traffic should use Do
// (without pinning) and validate upstream of the proxy themselves.
//
// Design note: "pinned + proxy" is intentionally not supported today.
// A proxy that re-resolves the original hostname would defeat the
// pinning; only a transparent proxy (one that forwards by IP and does
// not re-resolve) could coexist with SSRF-pinned traffic, and the
// Go stdlib's http.Transport.Proxy contract does not give us a
// reliable way to express that. If a future deployment needs
// pinned traffic through a specific egress proxy, add a dedicated
// DoPinnedWithProxy (or an HTTPHelper config option that takes an
// already-pinned IP) — do not silently re-enable Proxy here without
// also re-deriving the SSRF guarantee for that path.
func (h *HTTPHelper) pinnedClient(pinnedIP net.IP) *http.Client {
	base := h.baseTransport.Clone()
	base.Proxy = nil
	base.DialContext = (&pinnedDialer{
		pinnedIP: pinnedIP,
		base: &net.Dialer{
			Timeout:   30 * time.Second,
			KeepAlive: 30 * time.Second,
		},
	}).DialContext
	return &http.Client{
		Timeout:   h.client.Timeout,
		Transport: otelhttp.NewTransport(base),
	}
}

// pinnedDialer is a net.Dialer-compatible DialContext that rewrites the
// destination address to pinnedIP:port. The host part of the dial
// address is discarded; only the port is preserved. The host of the
// HTTP request is not in scope here — that lives on the request, where
// it is used by the TLS layer for SNI / cert verification.
type pinnedDialer struct {
	pinnedIP net.IP
	base     *net.Dialer
}

func (d *pinnedDialer) DialContext(ctx context.Context, network, addr string) (net.Conn, error) {
	_, port, err := net.SplitHostPort(addr)
	if err != nil {
		return nil, fmt.Errorf("http_helper: pinned dial: split %q: %w", addr, err)
	}
	return d.base.DialContext(ctx, network, net.JoinHostPort(d.pinnedIP.String(), port))
}

// doRawWithClient is the shared retry loop behind Do and DoPinned. The
// client decides the transport (and therefore the dialer); Do uses the
// helper's default OTel-wrapped client, DoPinned uses a per-call client
// with a transport-level pinned dialer.
func (h *HTTPHelper) doRawWithClient(
	ctx context.Context,
	client *http.Client,
	method, url, body, contentType string,
	headers map[string]string,
) (*http.Response, error) {
	if method == "" {
		method = http.MethodGet
	}
	if body != "" && contentType == "" {
		contentType = "application/octet-stream"
	}

	var lastErr error
	for attempt := 1; attempt <= h.retry.MaxAttempts; attempt++ {
		if err := ctx.Err(); err != nil {
			return nil, err
		}

		req, err := http.NewRequestWithContext(ctx, method, url, bytes.NewReader([]byte(body)))
		if err != nil {
			return nil, fmt.Errorf("http_helper: build request: %w", err)
		}
		if contentType != "" {
			req.Header.Set("Content-Type", contentType)
		}
		for k, v := range headers {
			req.Header.Set(k, v)
		}

		resp, err := client.Do(req)
		if err != nil {
			lastErr = err
			if !isRetryableNetError(err) {
				return nil, err
			}
			if attempt == h.retry.MaxAttempts {
				return nil, fmt.Errorf("http_helper: %s %s failed after %d attempts: %w", method, SanitizeURL(url), attempt, err)
			}
			sleepCtx(ctx, backoff(h.retry.BaseBackoff, h.retry.MaxBackoff, attempt))
			continue
		}

		// 5xx is retryable, 4xx is not.
		if resp.StatusCode >= 500 {
			lastErr = fmt.Errorf("http_helper: %s %s returned %d", method, SanitizeURL(url), resp.StatusCode)
			// drain body so the connection can be reused
			_, _ = io.Copy(io.Discard, resp.Body)
			_ = resp.Body.Close()
			if attempt == h.retry.MaxAttempts {
				return nil, lastErr
			}
			sleepCtx(ctx, backoff(h.retry.BaseBackoff, h.retry.MaxBackoff, attempt))
			continue
		}

		return resp, nil
	}

	if lastErr != nil {
		return nil, lastErr
	}
	return nil, errors.New("http_helper: exhausted retries with no recorded error")
}

// backoff returns an exponentially increasing duration with full jitter,
// capped at max. attempt is 1-indexed; the first retry uses BaseBackoff.
func backoff(base, max time.Duration, attempt int) time.Duration {
	if attempt < 1 {
		attempt = 1
	}
	d := base
	for i := 1; i < attempt; i++ {
		d *= 2
		if d > max {
			d = max
			break
		}
	}
	// full jitter: randomize in [0, d] to avoid thundering herd
	return time.Duration(rand.Int64N(int64(d) + 1))
}

// sleepCtx waits for d, returning early if ctx is canceled.
func sleepCtx(ctx context.Context, d time.Duration) {
	if d <= 0 {
		return
	}
	t := time.NewTimer(d)
	defer t.Stop()
	select {
	case <-ctx.Done():
	case <-t.C:
	}
}

// isRetryableNetError reports whether a transport-level error should trigger
// a retry. Context cancellation / deadline-exceeded are NOT retryable — the
// caller explicitly asked for that.
func isRetryableNetError(err error) bool {
	if err == nil {
		return false
	}
	if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
		return false
	}
	return true
}