127 lines
3.5 KiB
Go
127 lines
3.5 KiB
Go
package proxy
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"net/url"
|
|
"slices"
|
|
"strings"
|
|
|
|
"github.com/centrifugal/centrifugo/v6/internal/clientcontext"
|
|
"github.com/centrifugal/centrifugo/v6/internal/middleware"
|
|
|
|
"google.golang.org/grpc"
|
|
"google.golang.org/grpc/credentials"
|
|
"google.golang.org/grpc/credentials/insecure"
|
|
"google.golang.org/grpc/encoding/gzip"
|
|
"google.golang.org/grpc/metadata"
|
|
)
|
|
|
|
type rpcCredentials struct {
|
|
key string
|
|
value string
|
|
}
|
|
|
|
func (t rpcCredentials) GetRequestMetadata(_ context.Context, _ ...string) (map[string]string, error) {
|
|
return map[string]string{
|
|
t.key: t.value,
|
|
}, nil
|
|
}
|
|
|
|
func (t rpcCredentials) RequireTransportSecurity() bool {
|
|
return false
|
|
}
|
|
|
|
func getGrpcHost(endpoint string) (string, error) {
|
|
var host string
|
|
if strings.HasPrefix(endpoint, "grpc://") {
|
|
u, err := url.Parse(endpoint)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
host = u.Host
|
|
} else {
|
|
host = endpoint
|
|
}
|
|
return host, nil
|
|
}
|
|
|
|
func getDialOpts(name string, p Config) ([]grpc.DialOption, error) {
|
|
var dialOpts []grpc.DialOption
|
|
if p.GRPC.CredentialsKey != "" {
|
|
dialOpts = append(dialOpts, grpc.WithPerRPCCredentials(&rpcCredentials{
|
|
key: p.GRPC.CredentialsKey,
|
|
value: p.GRPC.CredentialsValue,
|
|
}))
|
|
}
|
|
if p.GRPC.TLS.Enabled {
|
|
tlsConfig, err := p.GRPC.TLS.ToGoTLSConfig("proxy_grpc:" + name)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed to create TLS config %v", err)
|
|
}
|
|
dialOpts = append(dialOpts, grpc.WithTransportCredentials(credentials.NewTLS(tlsConfig)))
|
|
} else {
|
|
dialOpts = append(dialOpts, grpc.WithTransportCredentials(insecure.NewCredentials()))
|
|
}
|
|
if p.GRPC.Compression {
|
|
dialOpts = append(dialOpts, grpc.WithDefaultCallOptions(grpc.UseCompressor(gzip.Name)))
|
|
}
|
|
|
|
if p.TestGrpcDialer != nil {
|
|
dialOpts = append(dialOpts, grpc.WithContextDialer(p.TestGrpcDialer))
|
|
}
|
|
|
|
return dialOpts, nil
|
|
}
|
|
|
|
func grpcRequestContext(ctx context.Context, proxy Config) context.Context {
|
|
md := requestMetadata(ctx, proxy.HttpHeaders, proxy.GrpcMetadata, emulatedHeaderAllowList(proxy), proxy.GRPC.StaticMetadata)
|
|
return metadata.NewOutgoingContext(ctx, md)
|
|
}
|
|
|
|
func requestMetadata(ctx context.Context, allowedHeaders, allowedMetaKeys, allowedEmulatedHeaders []string, staticMetadata map[string]string) metadata.MD {
|
|
requestMD := metadata.MD{}
|
|
|
|
// Set static metadata first, so that dynamic metadata can override it.
|
|
for k, v := range staticMetadata {
|
|
requestMD.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) {
|
|
requestMD.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)) {
|
|
requestMD.Set(k, vv...)
|
|
}
|
|
}
|
|
|
|
if !hasHTTPHeaders {
|
|
md, _ := metadata.FromIncomingContext(ctx)
|
|
for k, vv := range md {
|
|
if slices.Contains(allowedMetaKeys, k) {
|
|
requestMD[k] = vv
|
|
}
|
|
}
|
|
}
|
|
|
|
return requestMD
|
|
}
|