package proxy import ( "bytes" "context" "crypto/tls" "errors" "fmt" "io" "net/http" "slices" "strings" "github.com/centrifugal/centrifugo/v6/internal/clientcontext" "github.com/centrifugal/centrifugo/v6/internal/configtypes" "github.com/centrifugal/centrifugo/v6/internal/middleware" "github.com/centrifugal/centrifugo/v6/internal/proxyproto" "google.golang.org/grpc/metadata" ) type baseRequestHTTP struct { ClientID string `json:"client"` Transport string `json:"transport"` Protocol string `json:"protocol"` Encoding string `json:"encoding"` } var httpEncoder = &proxyproto.JSONEncoder{} var httpDecoder = &proxyproto.JSONDecoder{} // DefaultMaxIdleConnsPerHost is a reasonable value for all HTTP clients. const DefaultMaxIdleConnsPerHost = 255 // HTTPCaller is responsible for calling HTTP. type HTTPCaller interface { CallHTTP(context.Context, string, http.Header, []byte) ([]byte, error) } type httpCaller struct { Endpoint string HTTPClient *http.Client } // NewHTTPCaller creates new HTTPCaller. func NewHTTPCaller(httpClient *http.Client) HTTPCaller { return &httpCaller{ HTTPClient: httpClient, } } func proxyHTTPClient(p configtypes.Proxy, logTraceEntity string) (*http.Client, error) { var tlsConfig *tls.Config if p.HTTP.TLS.Enabled { var err error tlsConfig, err = p.HTTP.TLS.ToGoTLSConfig(logTraceEntity) if err != nil { return nil, fmt.Errorf("error creating TLS config: %w", err) } } return &http.Client{ Transport: &http.Transport{ MaxIdleConnsPerHost: DefaultMaxIdleConnsPerHost, TLSClientConfig: tlsConfig, }, Timeout: p.Timeout.ToDuration(), }, nil } type statusCodeError struct { Code int } func (e *statusCodeError) Error() string { return fmt.Sprintf("unexpected HTTP status code: %d", e.Code) } func (c *httpCaller) CallHTTP(ctx context.Context, endpoint string, header http.Header, reqData []byte) ([]byte, error) { req, err := http.NewRequest("POST", endpoint, bytes.NewReader(reqData)) if err != nil { return nil, fmt.Errorf("error constructing HTTP request: %w", err) } req.Header = header resp, err := c.HTTPClient.Do(req.WithContext(ctx)) if err != nil { return nil, fmt.Errorf("HTTP request error: %w", err) } defer func() { _ = resp.Body.Close() }() if resp.StatusCode != http.StatusOK { return nil, &statusCodeError{resp.StatusCode} } respData, err := io.ReadAll(resp.Body) if err != nil { return nil, fmt.Errorf("error reading HTTP body: %w", err) } return respData, nil } func transformHTTPStatusError(err error, transforms []configtypes.HttpStatusToCodeTransform) (*proxyproto.Error, *proxyproto.Disconnect) { if len(transforms) == 0 { return nil, nil } var statusErr *statusCodeError if !errors.As(err, &statusErr) { return nil, nil } for _, t := range transforms { if t.StatusCode == statusErr.Code { if t.ToError.Code > 0 { return &proxyproto.Error{ Code: t.ToError.Code, Message: t.ToError.Message, Temporary: t.ToError.Temporary, }, nil } if t.ToDisconnect.Code > 0 { return nil, &proxyproto.Disconnect{ Code: t.ToDisconnect.Code, Reason: t.ToDisconnect.Reason, } } } } return nil, nil } func httpRequestHeaders(ctx context.Context, proxy Config) http.Header { return requestHeaders(ctx, proxy.HttpHeaders, proxy.GrpcMetadata, emulatedHeaderAllowList(proxy), proxy.HTTP.StaticHeaders) } func requestHeaders(ctx context.Context, allowedHeaders, allowedMetaKeys, allowedEmulatedHeaders []string, staticHeaders map[string]string) http.Header { headers := http.Header{} // Set static headers first, so that dynamic headers can override them. for k, v := range staticHeaders { headers.Set(k, v) } // Emulated headers come from the client connect frame and are client-controlled, // so they are forwarded only when listed in emulated_headers - never via // the transport-level http_headers allow list. emulatedHeaders, _ := clientcontext.GetEmulatedHeadersFromContext(ctx) var droppedEmulated bool for k, v := range emulatedHeaders { lk := strings.ToLower(k) if slices.Contains(allowedEmulatedHeaders, lk) { headers.Set(k, v) } else if slices.Contains(allowedHeaders, lk) { // Listed in http_headers but not emulated_headers: this would // have been forwarded before v6.9.0. Flag for a throttled migration hint. droppedEmulated = true } } if droppedEmulated { logDroppedEmulatedHeadersHint(emulatedHeaders, allowedHeaders, allowedEmulatedHeaders) } httpHeaders, hasHTTPHeaders := middleware.GetHeadersFromContext(ctx) for k, vv := range httpHeaders { if slices.Contains(allowedHeaders, strings.ToLower(k)) { headers[k] = vv } } if !hasHTTPHeaders { md, _ := metadata.FromIncomingContext(ctx) for k, vv := range md { if slices.Contains(allowedMetaKeys, k) { // Canonicalize the lowercase metadata key so the transport value // lands on the same map key as emulated values written via Set // and overrides them - otherwise both would be sent as duplicate // headers with different casings. headers[http.CanonicalHeaderKey(k)] = vv } } } headers.Set("Content-Type", "application/json") return headers }