Files
graphql-monitoring-proxy/main.go
T
lukaszraczylo 02d9796adb fix(deps): update dependencies and allow loopback upstreams on fiber v3.5.0
Update all direct and indirect Go dependencies. Notable bumps are
gofiber/fiber v3.4.0 to v3.5.0, gofiber/contrib websocket v1.2.0 to
v1.2.6, fasthttp v1.72.0 to v1.74.0, gookit/goutil v0.7.6 to v0.8.0,
go-redis v9.20.1 to v9.22.0, pgx v5.10.0 to v5.11.0, OpenTelemetry
v1.44.0 to v1.46.0, grpc v1.82.1 to v1.83.2 and testify v1.11.1 to
v1.12.1. The go directive moves to 1.26.0 because websocket v1.2.6
requires it.

fiber v3.5.0 adds an SSRF dial guard to clients registered with
proxy.WithClient. The guard rejects loopback and private upstreams by
default. This proxy gets its upstream from operator configuration, and
sidecar deployments point it at localhost. Set
proxy.SecurityPolicy{AllowPrivateIPs: true} so these upstreams keep
working. This change also unblocks the nightly autoupdate workflow,
which failed on this guard.

Claude-Session: https://claude.ai/code/session_01Fu2wXqGUrfVBi5aLafKk8U
2026-09-10 15:18:31 +01:00

1166 lines
42 KiB
Go

package main
import (
"context"
"flag"
"fmt"
"net/http"
"net/url"
"os"
"os/signal"
"path/filepath"
"runtime"
"strconv"
"strings"
"sync"
"syscall"
"time"
// Register pprof handlers on http.DefaultServeMux. Listener is bound to
// 127.0.0.1 only and gated by PPROF_PORT — never expose publicly.
_ "net/http/pprof" //nolint:gosec // G108: handlers gated by PPROF_PORT, bound to 127.0.0.1 only
"github.com/gofiber/fiber/v3/middleware/proxy"
"github.com/gookit/goutil/envutil"
graphql "github.com/lukaszraczylo/go-simple-graphql"
libpack_cache "github.com/lukaszraczylo/graphql-monitoring-proxy/cache"
libpack_config "github.com/lukaszraczylo/graphql-monitoring-proxy/config"
libpack_logging "github.com/lukaszraczylo/graphql-monitoring-proxy/logging"
libpack_monitoring "github.com/lukaszraczylo/graphql-monitoring-proxy/monitoring"
libpack_tracing "github.com/lukaszraczylo/graphql-monitoring-proxy/tracing"
telemetry "github.com/lukaszraczylo/oss-telemetry"
// Auto-tune GOMAXPROCS from cgroup CPU quota (containerized workloads).
_ "go.uber.org/automaxprocs"
)
// appVersion is the build version. Set via ldflags during build:
//
// -X main.appVersion=v1.2.3
var appVersion = "dev"
var (
cfg *config
cfgMutex sync.RWMutex
once sync.Once
tracer *libpack_tracing.TracingSetup
shutdownManager *ShutdownManager
// pprofServer is the optional debug pprof HTTP server started by
// parseConfig when PPROF_PORT is set. It is stored here (rather than
// discarded) so main can register it with shutdownManager and Shutdown
// it cleanly instead of leaking the goroutine (see L2).
pprofServer *http.Server
)
// Shutdown phases for components registered on shutdownManager.
//
// shutdownPhaseDrainTraffic runs first: it owns the request-serving
// lifecycle (accepting connections, draining in-flight requests) and must
// finish before anything it depends on is torn down.
//
// shutdownPhaseCloseBackends runs after the drain completes. Every
// component here (connection pool, backend health manager, metrics
// aggregator, cache, tracer) is a dependency that in-flight requests read
// from during the drain - closing them earlier reproduces the redis
// "client is closed" failures the phased shutdown exists to prevent.
const (
// shutdownPhaseDrainTraffic is negative so it always sorts strictly
// before DefaultShutdownPhase (0, see shutdown.go). A plain
// RegisterComponent call must never land in the same phase as traffic
// draining, or it would be torn down concurrently with in-flight
// requests instead of after they finish draining.
shutdownPhaseDrainTraffic = -1
// shutdownPhaseCloseBackends stays after DefaultShutdownPhase too, so an
// unphased component (DefaultShutdownPhase) is guaranteed to shut down
// before the backends it may depend on.
shutdownPhaseCloseBackends = 1
)
// getDetailsFromEnv retrieves the value from the environment or returns the default.
// It first checks for a prefixed environment variable (GMP_KEY), then falls back to the unprefixed version.
func getDetailsFromEnv[T any](key string, defaultValue T) T {
prefixedKey := "GMP_" + key
switch v := any(defaultValue).(type) {
case string:
if val, ok := os.LookupEnv(prefixedKey); ok {
return any(val).(T)
}
return any(envutil.Getenv(key, v)).(T)
case int:
// An explicitly-empty value (VAR="") is treated as unset -- fall
// through to the next source instead of warning, since there is
// nothing invalid to parse, only a deliberately blank override.
if val, ok := os.LookupEnv(prefixedKey); ok && val != "" {
if intVal, err := strconv.Atoi(strings.TrimSpace(val)); err == nil {
return any(intVal).(T)
}
// C3: the prefixed var is set but unparsable - use the coded
// default. Do NOT fall through to the unprefixed name, or a
// stale/unrelated unprefixed value could silently win.
warnInvalidEnv(prefixedKey, val, v)
return any(v).(T)
}
if val, ok := os.LookupEnv(key); ok && val != "" {
if intVal, err := strconv.Atoi(strings.TrimSpace(val)); err == nil {
return any(intVal).(T)
}
// C2: envutil.GetInt silently returns 0 (not the default) when
// the value is present but non-numeric, because it discards the
// parse error. Parse it ourselves so we can warn and keep the
// coded default instead of silently accepting 0.
warnInvalidEnv(key, val, v)
return any(v).(T)
}
return any(v).(T)
case bool:
// An explicitly-empty value (VAR="") is treated as unset -- fall
// through to the next source instead of warning, since there is
// nothing invalid to parse, only a deliberately blank override.
if val, ok := os.LookupEnv(prefixedKey); ok && val != "" {
if b, valid := parseBoolEnv(val); valid {
return any(b).(T)
}
// C13: the prefixed and unprefixed bool parsers must agree on
// the same truthy/falsy token sets (true/1/on/yes,
// false/0/off/no). On anything else, warn and use the default
// rather than silently disagreeing with the unprefixed path.
warnInvalidEnv(prefixedKey, val, v)
return any(v).(T)
}
if val, ok := os.LookupEnv(key); ok && val != "" {
if b, valid := parseBoolEnv(val); valid {
return any(b).(T)
}
warnInvalidEnv(key, val, v)
return any(v).(T)
}
return any(v).(T)
case float64:
// envutil has no GetFloat; handle the GMP_ prefixed and plain keys
// directly. Previously float64 env values (CIRCUIT_FAILURE_RATIO,
// CIRCUIT_BACKOFF_MULTIPLIER, RETRY_BUDGET_TOKENS_PER_SEC) fell into
// default and were silently impossible to configure.
// An explicitly-empty value (VAR="") is treated as unset -- fall
// through to the next source instead of warning, since there is
// nothing invalid to parse, only a deliberately blank override.
if val, ok := os.LookupEnv(prefixedKey); ok && val != "" {
if f, err := strconv.ParseFloat(strings.TrimSpace(val), 64); err == nil {
return any(f).(T)
}
// C3: prefixed var set but unparsable - use the default, do not
// fall through to the unprefixed name.
warnInvalidEnv(prefixedKey, val, v)
return any(v).(T)
}
if val, ok := os.LookupEnv(key); ok && val != "" {
if f, err := strconv.ParseFloat(strings.TrimSpace(val), 64); err == nil {
return any(f).(T)
}
// C2: unprefixed var set but unparsable - warn and use default.
warnInvalidEnv(key, val, v)
return any(v).(T)
}
return defaultValue
default:
return defaultValue
}
}
// warnInvalidEnv logs that an environment variable's value could not be
// parsed as its expected type and that the coded default is used instead.
// getDetailsFromEnv runs for many fields before cfg (and its Logger) exists,
// so this prefers cfg.Logger once available and falls back to a throwaway
// libpack_logging.Logger otherwise.
func warnInvalidEnv(varName, rawValue string, defaultValue any) {
msg := &libpack_logging.LogMessage{
Message: "Invalid value for environment variable, using default",
Pairs: map[string]any{"var": varName, "value": rawValue, "default": defaultValue},
}
cfgMutex.RLock()
var logger *libpack_logging.Logger
if cfg != nil {
logger = cfg.Logger
}
cfgMutex.RUnlock()
if logger == nil {
logger = libpack_logging.New()
}
logger.Warning(msg)
}
// parseBoolEnv parses val using the same truthy/falsy token set gookit's
// envutil.GetBool accepts (true/1/on/yes and false/0/off/no), so the
// GMP_-prefixed and unprefixed bool paths in getDetailsFromEnv agree on the
// same input instead of the prefixed path only accepting "true"/"1" (C13).
// ok is false when val matches neither set.
func parseBoolEnv(val string) (parsed bool, ok bool) {
switch strings.ToLower(strings.TrimSpace(val)) {
case "true", "1", "on", "yes":
return true, true
case "false", "0", "off", "no":
return false, true
default:
return false, false
}
}
// validateJWTClaimPath validates JWT claim paths to prevent injection attacks
func validateJWTClaimPath(path string) error {
if path == "" {
return nil // Empty path is valid (feature disabled)
}
// Prevent path traversal attempts
if strings.Contains(path, "..") {
return fmt.Errorf("invalid JWT claim path (contains '..'): %s", path)
}
// Prevent absolute paths
if strings.HasPrefix(path, "/") {
return fmt.Errorf("invalid JWT claim path (absolute path not allowed): %s", path)
}
// Limit depth to prevent DoS from deeply nested claims
parts := strings.Split(path, ".")
if len(parts) > 10 {
return fmt.Errorf("invalid JWT claim path (too deep, max 10 levels): %s", path)
}
// Validate each part contains only allowed characters
for _, part := range parts {
if part == "" {
return fmt.Errorf("invalid JWT claim path (empty part): %s", path)
}
// Allow alphanumeric, underscore, and hyphen
for _, ch := range part {
if !((ch >= 'a' && ch <= 'z') || (ch >= 'A' && ch <= 'Z') ||
(ch >= '0' && ch <= '9') || ch == '_' || ch == '-') {
return fmt.Errorf("invalid JWT claim path (invalid character '%c'): %s", ch, path)
}
}
}
return nil
}
// parseConfig loads and parses the configuration.
func parseConfig() {
libpack_config.PKG_NAME = "graphql_proxy"
c := config{}
// Server configurations
c.Server.PortGraphQL = getDetailsFromEnv("PORT_GRAPHQL", 8080)
c.Server.PortMonitoring = getDetailsFromEnv("MONITORING_PORT", 9393)
// S6: bind address for the GraphQL proxy / admin API / monitoring
// listeners. Empty default preserves today's "bind all interfaces"
// behavior (":port"); consumers format as "<BindAddress>:<port>".
c.Server.BindAddress = getDetailsFromEnv("BIND_ADDRESS", "")
c.Server.HostGraphQL = getDetailsFromEnv("HOST_GRAPHQL", "http://localhost/")
c.Server.HostGraphQLReadOnly = getDetailsFromEnv("HOST_GRAPHQL_READONLY", "")
// Client configurations
c.Client.JWTUserClaimPath = getDetailsFromEnv("JWT_USER_CLAIM_PATH", "")
c.Client.JWTRoleClaimPath = getDetailsFromEnv("JWT_ROLE_CLAIM_PATH", "")
// Validate JWT claim paths for security
if err := validateJWTClaimPath(c.Client.JWTUserClaimPath); err != nil {
fmt.Fprintf(os.Stderr, "❌ CRITICAL ERROR: Invalid JWT_USER_CLAIM_PATH: %v\n", err)
os.Exit(1)
}
if err := validateJWTClaimPath(c.Client.JWTRoleClaimPath); err != nil {
fmt.Fprintf(os.Stderr, "❌ CRITICAL ERROR: Invalid JWT_ROLE_CLAIM_PATH: %v\n", err)
os.Exit(1)
}
// JWT signature verification (GHSA-9gqw-h2rw-44wv): read here, next to the
// claim-path variables, but the verifier itself is built further down
// (after c.Logger exists) so a fatal misconfiguration can be logged
// through the normal logger, matching the other fatal-config-error
// patterns in this function. Off by default -- JWT_VERIFY_SIGNATURE=false
// or unset keeps today's unverified claim decode (see
// extractClaimsFromJWTHeader in details.go) byte-for-byte unchanged.
jwtVerifySignature := getDetailsFromEnv("JWT_VERIFY_SIGNATURE", false)
jwtSecret := getDetailsFromEnv("JWT_SECRET", "")
jwtPublicKey := getDetailsFromEnv("JWT_PUBLIC_KEY", "")
jwtJWKSURL := getDetailsFromEnv("JWT_JWKS_URL", "")
jwtSigningMethods := getDetailsFromEnv("JWT_SIGNING_METHODS", "")
jwtIssuer := getDetailsFromEnv("JWT_ISSUER", "")
jwtAudience := getDetailsFromEnv("JWT_AUDIENCE", "")
jwtLeewaySeconds := getDetailsFromEnv("JWT_LEEWAY_SECONDS", 0)
c.Client.RoleFromHeader = getDetailsFromEnv("ROLE_FROM_HEADER", "")
c.Client.RoleRateLimit = getDetailsFromEnv("ROLE_RATE_LIMIT", false)
// In-memory cache
c.Cache.CacheEnable = getDetailsFromEnv("ENABLE_GLOBAL_CACHE", false)
c.Cache.CacheTTL = getDetailsFromEnv("CACHE_TTL", 60)
c.Cache.CacheMaxMemorySize = getDetailsFromEnv("CACHE_MAX_MEMORY_SIZE", 100) // Default 100MB
c.Cache.CacheMaxEntries = getDetailsFromEnv("CACHE_MAX_ENTRIES", 10000) // Default 10000 entries
c.Cache.CacheUseLRU = getDetailsFromEnv("CACHE_USE_LRU", false) // Use LRU eviction algorithm
// GraphQL query parsing cache - auto-calculate based on CPU cores if not set
c.Cache.GraphQLQueryCacheSize = getDetailsFromEnv("GRAPHQL_QUERY_CACHE_SIZE", runtime.GOMAXPROCS(0)*250)
// SECURITY: Per-user cache isolation (enabled by default for security)
// Set CACHE_PER_USER_DISABLED=true ONLY if you have a single-user application
// or understand the security implications of shared cache across users
c.Cache.PerUserCacheDisabled = getDetailsFromEnv("CACHE_PER_USER_DISABLED", false)
// Log warning if per-user caching is disabled
if c.Cache.PerUserCacheDisabled {
defer func() {
if c.Logger != nil {
c.Logger.Warning(&libpack_logging.LogMessage{
Message: "⚠️ Per-user cache isolation is DISABLED - Users may see each other's cached data!",
Pairs: map[string]any{
"security_risk": "CRITICAL - Do not use in multi-user applications",
"recommendation": "Remove CACHE_PER_USER_DISABLED or set it to false",
},
})
}
}()
}
// Redis cache
c.Cache.CacheRedisEnable = getDetailsFromEnv("ENABLE_REDIS_CACHE", false)
c.Cache.CacheRedisURL = getDetailsFromEnv("CACHE_REDIS_URL", "localhost:6379")
c.Cache.CacheRedisPassword = getDetailsFromEnv("CACHE_REDIS_PASSWORD", "")
c.Cache.CacheRedisDB = getDetailsFromEnv("CACHE_REDIS_DB", 0)
// Security configurations
c.Security.BlockIntrospection = getDetailsFromEnv("BLOCK_SCHEMA_INTROSPECTION", false)
c.Security.IntrospectionAllowed = func() []string {
urls := getDetailsFromEnv("ALLOWED_INTROSPECTION", "")
if urls == "" {
return nil
}
return strings.Split(urls, ",")
}()
c.LogLevel = strings.ToUpper(getDetailsFromEnv("LOG_LEVEL", "info"))
c.EnableAllocationTracking = getDetailsFromEnv("ENABLE_ALLOCATION_TRACKING", false)
// Logger setup
c.Logger = libpack_logging.New().SetMinLogLevel(libpack_logging.GetLogLevel(c.LogLevel)).
SetFieldName("timestamp", "ts").SetFieldName("message", "msg").SetShowCaller(false)
// C1: a non-positive CACHE_TTL crashes the cache cleanup ticker
// (time.NewTicker panics on d <= 0). Clamp to the coded default before
// this value ever reaches cache init.
if c.Cache.CacheTTL <= 0 {
c.Logger.Warning(&libpack_logging.LogMessage{
Message: "Invalid CACHE_TTL (must be positive), using default",
Pairs: map[string]any{"requested": c.Cache.CacheTTL, "default": 60},
})
c.Cache.CacheTTL = 60
}
// Health check
c.Server.HealthcheckGraphQL = getDetailsFromEnv("HEALTHCHECK_GRAPHQL_URL", "")
c.Client.GQLClient = graphql.NewConnection()
c.Client.GQLClient.SetEndpoint(c.Server.HealthcheckGraphQL)
// Server modes
c.Server.AccessLog = getDetailsFromEnv("ENABLE_ACCESS_LOG", false)
c.Server.ReadOnlyMode = getDetailsFromEnv("READ_ONLY_MODE", false)
c.Server.AllowURLs = func() []string {
urls := getDetailsFromEnv("ALLOWED_URLS", "")
if urls == "" {
return nil
}
return strings.Split(urls, ",")
}()
// Client timeout and connection configurations with bounds checking
clientTimeout := getDetailsFromEnv("PROXIED_CLIENT_TIMEOUT", 120)
if clientTimeout < 1 || clientTimeout > 3600 { // 1 second to 1 hour max
c.Logger.Warning(&libpack_logging.LogMessage{
Message: "Invalid client timeout, using default",
Pairs: map[string]any{"requested": clientTimeout, "default": 120},
})
clientTimeout = 120
}
c.Client.ClientTimeout = clientTimeout
// Configure HTTP connection pool and timeouts with sensible defaults
// MaxConnsPerHost limits parallel connections to prevent overwhelming backends
maxConns := getDetailsFromEnv("MAX_CONNS_PER_HOST", 1024)
if maxConns < 1 || maxConns > 10000 { // Reasonable bounds
c.Logger.Warning(&libpack_logging.LogMessage{
Message: "Invalid max connections per host, using default",
Pairs: map[string]any{"requested": maxConns, "default": 1024},
})
maxConns = 1024
}
c.Client.MaxConnsPerHost = maxConns
// Configure distinct timeout values for more granular control with bounds checking
readTimeout := getDetailsFromEnv("CLIENT_READ_TIMEOUT", c.Client.ClientTimeout)
if readTimeout < 1 || readTimeout > 3600 {
readTimeout = c.Client.ClientTimeout
}
c.Client.ReadTimeout = readTimeout
writeTimeout := getDetailsFromEnv("CLIENT_WRITE_TIMEOUT", c.Client.ClientTimeout)
if writeTimeout < 1 || writeTimeout > 3600 {
writeTimeout = c.Client.ClientTimeout
}
c.Client.WriteTimeout = writeTimeout
// MaxIdleConnDuration controls how long connections stay in the pool
idleDuration := getDetailsFromEnv("CLIENT_MAX_IDLE_CONN_DURATION", 300)
if idleDuration < 1 || idleDuration > 7200 { // 1 second to 2 hours max
idleDuration = 300
}
c.Client.MaxIdleConnDuration = idleDuration
// Retry backoff delays (milliseconds). Configurable so tests can set very
// short values without burning seconds per attempt.
retryBaseDelay := getDetailsFromEnv("CLIENT_RETRY_BASE_DELAY_MS", 500)
if retryBaseDelay < 1 {
retryBaseDelay = 500
}
c.Client.RetryBaseDelayMs = retryBaseDelay
retryMaxDelay := getDetailsFromEnv("CLIENT_RETRY_MAX_DELAY_MS", 10000)
if retryMaxDelay < retryBaseDelay {
retryMaxDelay = retryBaseDelay
}
c.Client.RetryMaxDelayMs = retryMaxDelay
// Secure by default: TLS verification is enabled unless explicitly disabled
c.Client.DisableTLSVerify = getDetailsFromEnv("CLIENT_DISABLE_TLS_VERIFY", false)
// Warn if TLS verification is disabled (security risk)
if c.Client.DisableTLSVerify {
// Logger might not be initialized yet, will log after logger setup
defer func() {
if c.Logger != nil {
c.Logger.Warning(&libpack_logging.LogMessage{
Message: "⚠️ TLS certificate verification is DISABLED - This is a security risk in production!",
Pairs: map[string]any{
"recommendation": "Enable TLS verification by removing CLIENT_DISABLE_TLS_VERIFY or setting it to false",
},
})
}
}()
}
// Create HTTP client with the optimized parameters
c.Client.FastProxyClient = createFasthttpClient(&c)
// fiber v3.5.0+ guards clients registered via proxy.WithClient against
// loopback/private upstreams. The upstream is operator-configured
// (HOST_GRAPHQL) and is commonly a localhost sidecar, so allow it.
proxy.WithSecurityPolicy(proxy.SecurityPolicy{AllowPrivateIPs: true})
proxy.WithClient(c.Client.FastProxyClient) // Setting the global proxy client
// API configurations
c.Server.EnableApi = getDetailsFromEnv("ENABLE_API", false)
c.Server.ApiPort = getDetailsFromEnv("API_PORT", 9090)
// Validate and sanitize banned users file path to prevent path traversal
bannedUsersFile := getDetailsFromEnv("BANNED_USERS_FILE", "/go/src/app/banned_users.json")
if validatedPath, err := validateFilePath(bannedUsersFile); err != nil {
c.Logger.Error(&libpack_logging.LogMessage{
Message: "Invalid banned users file path, using default",
Pairs: map[string]any{"requested": bannedUsersFile, "error": err.Error()},
})
c.Api.BannedUsersFile = "/go/src/app/banned_users.json"
} else {
c.Api.BannedUsersFile = validatedPath
}
c.Server.PurgeOnCrawl = getDetailsFromEnv("PURGE_METRICS_ON_CRAWL", false)
c.Server.PurgeEvery = getDetailsFromEnv("PURGE_METRICS_ON_TIMER", 1800) // Default: purge metrics every 30 minutes
// Hasura event cleaner
c.HasuraEventCleaner.Enable = getDetailsFromEnv("HASURA_EVENT_CLEANER", false)
c.HasuraEventCleaner.ClearOlderThan = getDetailsFromEnv("HASURA_EVENT_CLEANER_OLDER_THAN", 1)
c.HasuraEventCleaner.EventMetadataDb = getDetailsFromEnv("HASURA_EVENT_METADATA_DB", "")
// Tracing configuration
c.Tracing.Enable = getDetailsFromEnv("ENABLE_TRACE", false)
c.Tracing.Endpoint = getDetailsFromEnv("TRACE_ENDPOINT", "localhost:4317")
// Circuit Breaker configuration - optimized for high-traffic production environments
c.CircuitBreaker.Enable = getDetailsFromEnv("ENABLE_CIRCUIT_BREAKER", false)
c.CircuitBreaker.MaxFailures = getDetailsFromEnv("CIRCUIT_MAX_FAILURES", 10) // Higher tolerance for transient failures
c.CircuitBreaker.FailureRatio = getDetailsFromEnv("CIRCUIT_FAILURE_RATIO", 0.5) // Trip at 50% failure rate
c.CircuitBreaker.SampleSize = getDetailsFromEnv("CIRCUIT_SAMPLE_SIZE", 100) // Statistically significant sample
c.CircuitBreaker.Timeout = getDetailsFromEnv("CIRCUIT_TIMEOUT_SECONDS", 60) // Longer recovery time for stability
c.CircuitBreaker.MaxRequestsInHalfOpen = getDetailsFromEnv("CIRCUIT_MAX_HALF_OPEN_REQUESTS", 5) // More probe requests
c.CircuitBreaker.ReturnCachedOnOpen = getDetailsFromEnv("CIRCUIT_RETURN_CACHED_ON_OPEN", true)
c.CircuitBreaker.TripOnTimeouts = getDetailsFromEnv("CIRCUIT_TRIP_ON_TIMEOUTS", true)
c.CircuitBreaker.TripOn5xx = getDetailsFromEnv("CIRCUIT_TRIP_ON_5XX", true)
c.CircuitBreaker.TripOn4xx = getDetailsFromEnv("CIRCUIT_TRIP_ON_4XX", false) // 4xx are usually client errors
c.CircuitBreaker.BackoffMultiplier = getDetailsFromEnv("CIRCUIT_BACKOFF_MULTIPLIER", 1.0) // No backoff by default
c.CircuitBreaker.MaxBackoffTimeout = getDetailsFromEnv("CIRCUIT_MAX_BACKOFF_TIMEOUT", 300) // 5 minutes max
// Retry budget configuration
c.RetryBudget.Enable = getDetailsFromEnv("RETRY_BUDGET_ENABLE", true)
c.RetryBudget.TokensPerSecond = getDetailsFromEnv("RETRY_BUDGET_TOKENS_PER_SEC", 10.0)
// C14: a non-positive refill rate means the token bucket never refills,
// permanently locking out retries once the initial burst is spent.
// Clamp to the coded default (10.0/s, consistent with the default above
// and with RETRY_BUDGET_MAX_TOKENS's coded default of 100 below).
if c.RetryBudget.TokensPerSecond <= 0 {
c.Logger.Warning(&libpack_logging.LogMessage{
Message: "Invalid RETRY_BUDGET_TOKENS_PER_SEC (must be positive), using default",
Pairs: map[string]any{"requested": c.RetryBudget.TokensPerSecond, "default": 10.0},
})
c.RetryBudget.TokensPerSecond = 10.0
}
c.RetryBudget.MaxTokens = getDetailsFromEnv("RETRY_BUDGET_MAX_TOKENS", 100)
// Request coalescing configuration
c.RequestCoalescing.Enable = getDetailsFromEnv("REQUEST_COALESCING_ENABLE", true)
// WebSocket configuration
c.WebSocket.Enable = getDetailsFromEnv("WEBSOCKET_ENABLE", false)
c.WebSocket.PingInterval = getDetailsFromEnv("WEBSOCKET_PING_INTERVAL", 30)
c.WebSocket.PongTimeout = getDetailsFromEnv("WEBSOCKET_PONG_TIMEOUT", 60)
c.WebSocket.MaxMessageSize = int64(getDetailsFromEnv("WEBSOCKET_MAX_MESSAGE_SIZE", 524288)) // 512KB
// Admin dashboard configuration
c.AdminDashboard.Enable = getDetailsFromEnv("ADMIN_DASHBOARD_ENABLE", true)
// Optional debug pprof endpoint. Disabled unless PPROF_PORT is set to a
// valid integer. Bound to 127.0.0.1 ONLY — pprof must never be exposed
// publicly (it leaks memory layout, allows arbitrary CPU profiles, etc).
if pprofPortStr := getDetailsFromEnv("PPROF_PORT", ""); pprofPortStr != "" {
if pprofPort, err := strconv.Atoi(pprofPortStr); err == nil && pprofPort > 0 && pprofPort < 65536 {
addr := "127.0.0.1:" + strconv.Itoa(pprofPort)
c.Logger.Info(&libpack_logging.LogMessage{
Message: "pprof endpoint listening on " + addr,
})
// L2: keep the server so it can be registered with
// shutdownManager and Shutdown cleanly instead of leaking the
// goroutine for the life of the process.
srv := &http.Server{
Addr: addr,
Handler: nil,
ReadHeaderTimeout: 5 * time.Second,
ReadTimeout: 30 * time.Second,
WriteTimeout: 120 * time.Second,
IdleTimeout: 120 * time.Second,
}
pprofServer = srv
go func(listenAddr string) {
if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
c.Logger.Error(&libpack_logging.LogMessage{
Message: "pprof endpoint failed",
Pairs: map[string]any{"error": err.Error(), "addr": listenAddr},
})
}
}(addr)
} else {
c.Logger.Warning(&libpack_logging.LogMessage{
Message: "PPROF_PORT set but invalid; pprof endpoint disabled",
Pairs: map[string]any{"value": pprofPortStr},
})
}
}
cfgMutex.Lock()
cfg = &c
cfgMutex.Unlock()
// Initialize tracing if enabled
if cfg.Tracing.Enable {
if cfg.Tracing.Endpoint == "" {
cfg.Logger.Warning(&libpack_logging.LogMessage{
Message: "Tracing endpoint not configured, using default localhost:4317",
})
cfg.Tracing.Endpoint = "localhost:4317"
}
var err error
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
tracer, err = libpack_tracing.NewTracing(ctx, cfg.Tracing.Endpoint)
if err != nil {
cfg.Logger.Error(&libpack_logging.LogMessage{
Message: "Failed to initialize tracing",
Pairs: map[string]any{"error": err.Error()},
})
} else {
cfg.Logger.Info(&libpack_logging.LogMessage{
Message: "Tracing initialized",
Pairs: map[string]any{"endpoint": cfg.Tracing.Endpoint},
})
}
}
// Initialize metrics aggregator FIRST if Redis is enabled (even if cache is disabled)
// This allows cluster mode monitoring even when cache is off
if cfg.Cache.CacheRedisEnable {
cfg.Logger.Info(&libpack_logging.LogMessage{
Message: "Initializing metrics aggregator for cluster mode",
Pairs: map[string]any{
"redis_url": cfg.Cache.CacheRedisURL,
"redis_db": cfg.Cache.CacheRedisDB,
},
})
if err := InitializeMetricsAggregator(
cfg.Cache.CacheRedisURL,
cfg.Cache.CacheRedisPassword,
cfg.Cache.CacheRedisDB,
cfg.Logger,
); err != nil {
cfg.Logger.Error(&libpack_logging.LogMessage{
Message: "FAILED to initialize metrics aggregator - cluster mode will not work",
Pairs: map[string]any{
"error": err.Error(),
},
})
} else {
cfg.Logger.Info(&libpack_logging.LogMessage{
Message: "✓ Metrics aggregator successfully initialized",
Pairs: map[string]any{
"instance_id": GetMetricsAggregator().GetInstanceID(),
},
})
}
}
// Initialize cache if enabled
if cfg.Cache.CacheEnable || cfg.Cache.CacheRedisEnable {
cacheConfig := &libpack_cache.CacheConfig{
Logger: cfg.Logger,
TTL: cfg.Cache.CacheTTL,
PerUserCacheDisabled: cfg.Cache.PerUserCacheDisabled,
}
// Redis cache configurations
if cfg.Cache.CacheRedisEnable {
cacheConfig.Redis.Enable = true
cacheConfig.Redis.URL = cfg.Cache.CacheRedisURL
cacheConfig.Redis.Password = cfg.Cache.CacheRedisPassword
cacheConfig.Redis.DB = cfg.Cache.CacheRedisDB
} else {
// Memory cache configurations
cacheConfig.Memory.MaxMemorySize = int64(cfg.Cache.CacheMaxMemorySize) * 1024 * 1024 // Convert MB to bytes
cacheConfig.Memory.MaxEntries = int64(cfg.Cache.CacheMaxEntries)
cacheConfig.Memory.UseLRU = cfg.Cache.CacheUseLRU
cacheType := "standard"
if cfg.Cache.CacheUseLRU {
cacheType = "LRU"
}
cfg.Logger.Info(&libpack_logging.LogMessage{
Message: "Configuring memory cache with limits",
Pairs: map[string]any{
"type": cacheType,
"max_memory_mb": cfg.Cache.CacheMaxMemorySize,
"max_entries": cfg.Cache.CacheMaxEntries,
},
})
}
libpack_cache.EnableCache(cacheConfig)
// Start memory monitoring for in-memory cache if it's not Redis
// Will be started with context in main()
}
// Initialize circuit breaker if enabled
if cfg.CircuitBreaker.Enable {
initCircuitBreaker(cfg)
}
// Note: Retry budget is initialized in main() with context for graceful shutdown
// Initialize request coalescer
if cfg.RequestCoalescing.Enable {
InitializeRequestCoalescer(true, cfg.Logger, cfg.Monitoring)
}
// Initialize WebSocket proxy
if cfg.WebSocket.Enable {
wsConfig := WebSocketConfig{
Enabled: cfg.WebSocket.Enable,
PingInterval: time.Duration(cfg.WebSocket.PingInterval) * time.Second,
PongTimeout: time.Duration(cfg.WebSocket.PongTimeout) * time.Second,
MaxMessageSize: cfg.WebSocket.MaxMessageSize,
}
InitializeWebSocketProxy(cfg.Server.HostGraphQL, wsConfig, cfg.Logger, cfg.Monitoring)
}
// Initialize backend health manager
if cfg.Server.HostGraphQL != "" {
healthMgr := InitializeBackendHealth(cfg.Client.FastProxyClient, cfg.Server.HostGraphQL, cfg.Logger)
// Start health checking in background
healthMgr.StartHealthChecking()
}
// Note: RPS tracker is initialized in main() with context for graceful shutdown
// Initialize JWT signature verification (GHSA-9gqw-h2rw-44wv). The proxy
// derives per-user cache keys from JWT claims (see details.go); those
// claims were previously trusted without checking the token's signature,
// so a forged token with a guessed claim could read another user's
// cached response. A misconfigured key source here is fatal: the proxy
// must never boot half-configured, trusting claims it cannot actually
// verify.
if jwtVerifySignature {
verifier, err := newJWTVerifier(jwtVerifierConfig{
Secret: jwtSecret,
PublicKeyPEM: jwtPublicKey,
JWKSURL: jwtJWKSURL,
SigningMethods: jwtSigningMethods,
Issuer: jwtIssuer,
Audience: jwtAudience,
LeewaySeconds: jwtLeewaySeconds,
})
if err != nil {
cfg.Logger.Error(&libpack_logging.LogMessage{
Message: "Failed to initialize JWT signature verification",
Pairs: map[string]any{"error": err.Error()},
})
if ifNotInTest() {
fmt.Fprintln(os.Stderr, "❌ CRITICAL ERROR: JWT_VERIFY_SIGNATURE is enabled but misconfigured")
fmt.Fprintln(os.Stderr, err.Error())
os.Exit(1)
}
} else {
cfg.Client.JWTVerifier = verifier
cfg.Logger.Info(&libpack_logging.LogMessage{
Message: "JWT signature verification enabled",
})
}
}
// Load rate limit configuration with improved error handling
if err := loadRatelimitConfig(); err != nil {
// Log the error with clear guidance
detailedError := err.Error()
cfg.Logger.Error(&libpack_logging.LogMessage{
Message: "Failed to start service due to rate limit configuration error",
Pairs: map[string]any{
"error": detailedError,
},
})
// If we're not in a test environment, print to stderr and exit if config error
if ifNotInTest() {
fmt.Fprintln(os.Stderr, "⚠️ CRITICAL ERROR: Rate limit configuration problem detected")
fmt.Fprintln(os.Stderr, detailedError)
os.Exit(1)
}
}
// API and event cleaner will be started with context in main()
prepareQueriesAndExemptions()
// Initialize GraphQL parsing optimizations
initGraphQLParsing()
}
func main() {
telemetry.SendForModule("graphql-monitoring-proxy", "github.com/lukaszraczylo/graphql-monitoring-proxy", appVersion)
// Parse configuration
parseConfig()
// Setup graceful shutdown
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
// Initialize shutdown manager
shutdownManager = NewShutdownManager(ctx)
// Initialize RPS tracker with context for graceful shutdown
InitializeRPSTracker(ctx)
cfg.Logger.Info(&libpack_logging.LogMessage{
Message: "RPS tracker initialized",
})
// Initialize retry budget with context for graceful shutdown
if cfg.RetryBudget.Enable {
retryBudgetConfig := RetryBudgetConfig{
TokensPerSecond: cfg.RetryBudget.TokensPerSecond,
MaxTokens: cfg.RetryBudget.MaxTokens,
Enabled: cfg.RetryBudget.Enable,
}
InitializeRetryBudgetWithContext(ctx, retryBudgetConfig, cfg.Logger)
}
// Create a wait group to manage goroutines
var wg sync.WaitGroup
// Setup signal handling for graceful shutdown
sigCh := make(chan os.Signal, 1)
signal.Notify(sigCh, os.Interrupt, syscall.SIGTERM)
go func() {
<-sigCh
cfg.Logger.Info(&libpack_logging.LogMessage{
Message: "Shutdown signal received, stopping services...",
})
cancel()
}()
// Start background services with context
once.Do(func() {
// Start API server
shutdownManager.RunGoroutine("api-server", func(ctx context.Context) {
if err := enableApi(ctx); err != nil {
cfg.Logger.Error(&libpack_logging.LogMessage{
Message: "API server error",
Pairs: map[string]any{"error": err.Error()},
})
}
})
// Start event cleaner
shutdownManager.RunGoroutine("event-cleaner", func(ctx context.Context) {
if err := enableHasuraEventCleaner(ctx); err != nil {
cfg.Logger.Error(&libpack_logging.LogMessage{
Message: "Event cleaner error",
Pairs: map[string]any{"error": err.Error()},
})
}
})
// Start cache memory monitoring if not using Redis
if cfg.Cache.CacheEnable && !cfg.Cache.CacheRedisEnable {
shutdownManager.RunGoroutine("cache-memory-monitoring", startCacheMemoryMonitoring)
}
})
// Register connection pool for cleanup. It wraps the same fasthttp
// client the proxy uses per-request (cfg.Client.FastProxyClient), so it
// must outlive the traffic drain below.
shutdownManager.RegisterComponentWithPhase("http-connection-pool", shutdownPhaseCloseBackends, func(ctx context.Context) error {
if connectionPoolManager != nil {
return connectionPoolManager.Shutdown()
}
return nil
})
// Register backend health manager for cleanup. performProxyRequestWithRetries
// reads healthMgr.IsHealthy() on every proxied request, so it must
// outlive the traffic drain below.
shutdownManager.RegisterComponentWithPhase("backend-health-manager", shutdownPhaseCloseBackends, func(ctx context.Context) error {
if healthMgr := GetBackendHealthManager(); healthMgr != nil {
healthMgr.Shutdown()
}
return nil
})
// Register metrics aggregator for cleanup. Shutdown() closes its Redis
// client, so it belongs with the other backend closers, after the
// traffic drain below.
shutdownManager.RegisterComponentWithPhase("metrics-aggregator", shutdownPhaseCloseBackends, func(ctx context.Context) error {
if aggregator := GetMetricsAggregator(); aggregator != nil {
aggregator.Shutdown()
}
return nil
})
// Register HTTP proxy so in-flight requests drain gracefully on shutdown
// instead of being cut off when the process exits. This runs in the
// traffic-drain phase, before any of the backends above are closed.
shutdownManager.RegisterComponentWithPhase("http-proxy", shutdownPhaseDrainTraffic, shutdownHTTPProxy)
// L2: register the optional pprof debug server (if PPROF_PORT was set)
// so it stops cleanly instead of leaking its listener goroutine forever.
if pprofServer != nil {
shutdownManager.RegisterComponent("pprof-server", func(ctx context.Context) error {
return pprofServer.Shutdown(ctx)
})
}
// Close the Redis cache client's connection pool at shutdown (in-memory
// caches are no-ops). Runs after the traffic drain so in-flight
// CacheLookup/CacheStore calls do not hit a closed Redis client.
shutdownManager.RegisterComponentWithPhase("cache", shutdownPhaseCloseBackends, func(context.Context) error {
libpack_cache.Shutdown()
return nil
})
// Start monitoring server
cfg.Logger.Info(&libpack_logging.LogMessage{
Message: "Starting monitoring server...",
Pairs: map[string]any{"port": cfg.Server.PortMonitoring},
})
// Start monitoring server in a goroutine
wg.Add(1)
monitoringErrCh := make(chan error, 1)
go func() {
defer wg.Done()
if err := StartMonitoringServer(); err != nil {
monitoringErrCh <- err
}
}()
// Give monitoring server time to initialize
select {
case err := <-monitoringErrCh:
cfg.Logger.Critical(&libpack_logging.LogMessage{
Message: "Failed to start monitoring server",
Pairs: map[string]any{
"error": err.Error(),
"port": cfg.Server.PortMonitoring,
},
})
os.Exit(1)
case <-time.After(2 * time.Second):
// Continue if no error received within timeout
}
// Wait for GraphQL backend to be ready before starting proxy
if healthMgr := GetBackendHealthManager(); healthMgr != nil {
startupTimeout := time.Duration(getDetailsFromEnv("BACKEND_STARTUP_TIMEOUT", 300)) * time.Second
cfg.Logger.Info(&libpack_logging.LogMessage{
Message: "Waiting for GraphQL backend to be ready",
Pairs: map[string]any{
"timeout_seconds": int(startupTimeout.Seconds()),
},
})
if err := healthMgr.WaitForBackendReady(startupTimeout); err != nil {
cfg.Logger.Critical(&libpack_logging.LogMessage{
Message: "GraphQL backend did not become ready in time",
Pairs: map[string]any{
"error": err.Error(),
"timeout": startupTimeout.String(),
},
})
// Don't exit immediately, but warn that backend is not ready
cfg.Logger.Warning(&libpack_logging.LogMessage{
Message: "Starting proxy anyway - requests will fail until backend becomes available",
})
}
}
// Start HTTP proxy
cfg.Logger.Info(&libpack_logging.LogMessage{
Message: "Starting HTTP proxy server...",
Pairs: map[string]any{"port": cfg.Server.PortGraphQL},
})
// Start HTTP proxy in a goroutine
wg.Add(1)
proxyErrCh := make(chan error, 1)
go func() {
defer wg.Done()
if err := StartHTTPProxy(); err != nil {
proxyErrCh <- err
}
}()
// Block for a moment to check for immediate startup errors
select {
case err := <-proxyErrCh:
cfg.Logger.Critical(&libpack_logging.LogMessage{
Message: "Failed to start HTTP proxy server",
Pairs: map[string]any{
"error": err.Error(),
"port": cfg.Server.PortGraphQL,
},
})
os.Exit(1)
case <-time.After(1 * time.Second):
// Continue if no error received within timeout
}
// Wait for context cancellation
<-ctx.Done()
// Perform cleanup
cfg.Logger.Info(&libpack_logging.LogMessage{
Message: "Shutting down services...",
})
// Register tracer shutdown. proxy.go calls tracer.StartSpan /
// ExtractSpanContext on every proxied request, so this closes after the
// traffic drain along with the other backend dependencies.
if tracer != nil {
shutdownManager.RegisterComponentWithPhase("tracer", shutdownPhaseCloseBackends, func(ctx context.Context) error {
return tracer.Shutdown(ctx)
})
}
// Perform graceful shutdown of all components
if err := shutdownManager.Shutdown(30 * time.Second); err != nil {
cfg.Logger.Error(&libpack_logging.LogMessage{
Message: "Error during shutdown",
Pairs: map[string]any{"error": err.Error()},
})
}
// Wait for all goroutines to finish (with timeout)
waitCh := make(chan struct{})
go func() {
wg.Wait()
close(waitCh)
}()
select {
case <-waitCh:
cfg.Logger.Info(&libpack_logging.LogMessage{
Message: "All services shut down gracefully",
})
case <-time.After(10 * time.Second):
cfg.Logger.Warning(&libpack_logging.LogMessage{
Message: "Some services didn't shut down gracefully within timeout",
})
}
}
// startCacheMemoryMonitoring polls memory cache usage and updates metrics
func startCacheMemoryMonitoring(ctx context.Context) {
// Check every few seconds (more frequent than cleanup routine)
ticker := time.NewTicker(15 * time.Second)
defer ticker.Stop()
cfg.Logger.Info(&libpack_logging.LogMessage{
Message: "Starting memory cache monitoring",
})
// Use mutex to protect concurrent access to metrics registration
var metricsMutex sync.Mutex
// Create initial metrics with proper synchronization
metricsMutex.Lock()
cfg.Monitoring.RegisterMetricsGauge(libpack_monitoring.MetricsCacheMemoryLimit, nil,
float64(libpack_cache.GetCacheMaxMemorySize()))
metricsMutex.Unlock()
for {
select {
case <-ctx.Done():
cfg.Logger.Info(&libpack_logging.LogMessage{
Message: "Stopping cache memory monitoring",
})
return
case <-ticker.C:
// Skip if monitoring not initialized or cache not initialized
if cfg.Monitoring == nil || !libpack_cache.IsCacheInitialized() {
continue
}
// Get current memory usage atomically
memoryUsage := libpack_cache.GetCacheMemoryUsage()
memoryLimit := libpack_cache.GetCacheMaxMemorySize()
// Update metrics with proper synchronization
metricsMutex.Lock()
cfg.Monitoring.RegisterMetricsGauge(libpack_monitoring.MetricsCacheMemoryUsage, nil,
float64(memoryUsage))
cfg.Monitoring.RegisterMetricsGauge(libpack_monitoring.MetricsCacheMemoryLimit, nil,
float64(memoryLimit))
// Calculate percentage (protect against division by zero)
var percentUsed float64
if memoryLimit > 0 {
percentUsed = float64(memoryUsage) / float64(memoryLimit) * 100.0
}
cfg.Monitoring.RegisterMetricsGauge(libpack_monitoring.MetricsCacheMemoryPercent, nil,
percentUsed)
metricsMutex.Unlock()
// Log if memory usage is high (over 80%)
if percentUsed > 80.0 {
cfg.Logger.Warning(&libpack_logging.LogMessage{
Message: "Memory cache usage is high",
Pairs: map[string]any{
"memory_usage_bytes": memoryUsage,
"memory_limit_bytes": memoryLimit,
"percent_used": percentUsed,
},
})
}
}
}
}
// validateFilePath validates and sanitizes file paths to prevent path traversal attacks
func validateFilePath(path string) (string, error) {
if path == "" {
return "", fmt.Errorf("empty path not allowed")
}
// Reject bare current directory for security
if path == "." {
return "", fmt.Errorf("bare current directory not allowed")
}
// URL decode the path to detect encoded traversal attempts
decodedPath := path
if strings.Contains(path, "%") {
// Try to decode URL encoding (single and double)
for i := 0; i < 3; i++ { // Handle multiple levels of encoding
if decoded, err := url.QueryUnescape(decodedPath); err == nil {
decodedPath = decoded
} else {
break
}
}
}
// Check for path traversal patterns (in both original and decoded)
checkPaths := []string{path, decodedPath}
for _, checkPath := range checkPaths {
if strings.Contains(checkPath, "..") {
return "", fmt.Errorf("path traversal attempt detected")
}
}
// Check for dangerous characters
dangerousChars := []string{";", "|", "\n", "\r"}
for _, char := range dangerousChars {
if strings.Contains(path, char) {
return "", fmt.Errorf("dangerous character detected in path")
}
}
// Clean and normalize the path
cleaned := filepath.Clean(path)
// Get absolute path
absPath, err := filepath.Abs(cleaned)
if err != nil {
return "", fmt.Errorf("invalid file path: %w", err)
}
// Get working directory as base
workDir, err := os.Getwd()
if err != nil {
return "", fmt.Errorf("cannot determine working directory: %w", err)
}
// Define allowed directories
allowedDirs := []string{
workDir, // Current working directory
"/tmp", // Temporary files
"/var/tmp", // System temporary files
"/go/src/app", // Docker container default
}
// Check if the path is within any allowed directory
isAllowed := false
for _, allowedDir := range allowedDirs {
// Ensure both paths are cleaned and absolute for proper comparison
cleanedAllowed := filepath.Clean(allowedDir)
if strings.HasPrefix(absPath, cleanedAllowed+string(filepath.Separator)) || absPath == cleanedAllowed {
isAllowed = true
break
}
}
if !isAllowed {
return "", fmt.Errorf("path not in allowed directories")
}
// Additional security checks
if strings.Contains(absPath, "\x00") {
return "", fmt.Errorf("null byte in path")
}
// Return the original path if it's within the current working directory and is relative
if strings.HasPrefix(absPath, workDir) && !filepath.IsAbs(path) {
return path, nil
}
return absPath, nil
}
// ifNotInTest checks if the program is not running in a test environment.
func ifNotInTest() bool {
return flag.Lookup("test.v") == nil
}