package distworker import ( "fmt" "net/url" "os" "strconv" "strings" ) // DefaultMaxConcurrency is the default upper bound for the worker pool size // and the number of dispatch goroutines that drain the job queue. const DefaultMaxConcurrency = 32 const ( // DefaultHTTPHost is the bind address used when WORKER_HOST is unset. DefaultHTTPHost = "0.0.0.0" // DefaultHTTPPort is the bind port used when WORKER_PORT is unset. // Picked >20000 to avoid colliding with the main RSMon app (which // binds 7401 by default) when the worker is co-located on the // same host. Operators are still free to override via WORKER_PORT. DefaultHTTPPort = 27401 // EnvPublicURL is the canonical environment variable for the // publicly advertised origin. It wins over the legacy // EnvWorkerURLLegacy when both are set. EnvPublicURL = "PUBLIC_URL" // EnvWorkerURLLegacy is the legacy name for the advertised public // origin. It is accepted only for the bounded migration defined in // docs/public-endpoint-and-identity.md milestone 1 and is // deprecated: new configuration must set EnvPublicURL. EnvWorkerURLLegacy = "WORKER_URL" // schemeHTTP / schemeHTTPS are the only schemes accepted on // PUBLIC_URL. Peer workers and the main app need an http(s) origin // they can dial with Go's net/http stack. schemeHTTP = "http" schemeHTTPS = "https" // loopbackHostnames lists hostnames treated as loopback for the // http-on-non-loopback warning emitted by ValidateHTTPConfig. loopbackHostnames = "localhost,127.0.0.1,::1,0.0.0.0" ) // HTTPConfig holds the settings that govern the worker's local HTTP // listener (web app MVP in Task 3 and Raft peer connections in Task 4). // Host/Port are the bind interface. PublicURL is the publicly-advertised // origin peers and the main app use to reach the worker; it is NOT // derived from Host:Port because workers commonly sit behind a reverse // proxy / Traefik with HTTPS while listening on plain HTTP internally. // // PublicURL is sourced from PUBLIC_URL (strict origin) or, for the // bounded migration, from the deprecated WORKER_URL (legacy-tolerant // absolute URL). URLSource records which variable supplied the value so // validation and diagnostics name the correct source. // // Login/Password are basic auth credentials for the worker web app API // (Task 3). They are kept in memory only; Task 3 may hash them before // any persistent store. type HTTPConfig struct { Host string Port int PublicURL string URLSource PublicURLSource Login string Password string } // PublicURLSource identifies which environment variable supplied the // advertised public origin, so callers can log the legacy deprecation // exactly once. type PublicURLSource int const ( // PublicURLSourceNone means no advertised origin is configured. PublicURLSourceNone PublicURLSource = iota // PublicURLSourceCanonical means PUBLIC_URL was configured. PublicURLSourceCanonical // PublicURLSourceLegacy means only WORKER_URL was configured; the // value is accepted for the bounded migration. PublicURLSourceLegacy ) // PublicURLFromEnv reads the advertised public origin. PUBLIC_URL is // the canonical source; the deprecated WORKER_URL is used only when // PUBLIC_URL is unset. The returned source lets the caller warn about // the legacy fallback without re-reading the environment. func PublicURLFromEnv() (string, PublicURLSource) { if v := strings.TrimSpace(os.Getenv(EnvPublicURL)); v != "" { return v, PublicURLSourceCanonical } if v := strings.TrimSpace(os.Getenv(EnvWorkerURLLegacy)); v != "" { return v, PublicURLSourceLegacy } return "", PublicURLSourceNone } // IsAuthConfigured reports whether both WORKER_LOGIN and WORKER_PASSWORD // are set. The HTTP listener refuses to start otherwise. func (c HTTPConfig) IsAuthConfigured() bool { return c.Login != "" && c.Password != "" } // IsListenConfigured reports whether the listener should bind at all. // Currently always true while WORKER_PORT > 0; placeholder for Task 3 // to wire "skip listen" semantics. func (c HTTPConfig) IsListenConfigured() bool { return c.Port > 0 } // Config holds the local worker connection settings. // Runtime settings are delivered by the control plane over websocket. type Config struct { URL string Token string BootstrapToken string StateFile string WorkerID string // MaxConcurrency caps the worker pool size and the number of dispatcher // goroutines. A value <= 0 falls back to DefaultMaxConcurrency. // Runtime configuration delivered by the control plane via the websocket // "init" or "config" message is clamped to this value when resizing. MaxConcurrency int // HTTP holds the local web app / Raft listener settings (Task 2). HTTP HTTPConfig } // ConfigFromEnv creates a Config from environment variables. func ConfigFromEnv() Config { return Config{ URL: normalizeURL(os.Getenv("RSMON_URL")), Token: os.Getenv("RSMON_TOKEN"), BootstrapToken: os.Getenv("RSMON_BOOTSTRAP_TOKEN"), StateFile: strings.TrimSpace(os.Getenv("RSMON_STATE_FILE")), WorkerID: strings.TrimSpace(os.Getenv("RSMON_WORKER_ID")), HTTP: HTTPConfigFromEnv(), } } // HTTPConfigFromEnv reads HTTP listener settings from the environment. // Empty WORKER_HOST defaults to DefaultHTTPHost; empty WORKER_PORT defaults // to DefaultHTTPPort. A malformed WORKER_PORT falls back to the default. // The public origin is resolved from PUBLIC_URL (canonical) with the // deprecated WORKER_URL as the migration fallback; it is parsed loosely // here — ValidateHTTPConfig does the real check. func HTTPConfigFromEnv() HTTPConfig { port := DefaultHTTPPort if raw := strings.TrimSpace(os.Getenv("WORKER_PORT")); raw != "" { if v, err := strconv.Atoi(raw); err == nil && v > 0 && v <= 65535 { port = v } } host := strings.TrimSpace(os.Getenv("WORKER_HOST")) if host == "" { host = DefaultHTTPHost } publicURL, source := PublicURLFromEnv() return HTTPConfig{ Host: host, Port: port, PublicURL: publicURL, URLSource: source, Login: os.Getenv("WORKER_LOGIN"), Password: os.Getenv("WORKER_PASSWORD"), } } // ValidateHTTPConfig enforces the local HTTP-listener invariants: // // - WORKER_LOGIN and WORKER_PASSWORD must be both set or both empty. // A mixed state (XOR) is a config bug and must fail fast so an // operator notices immediately. Both empty is allowed while the HTTP // listener is not started yet. // - When the listener would actually start (WORKER_PORT > 0), both // must be set; otherwise we are going to expose an unauthenticated // endpoint. // - A configured advertised origin must validate. The canonical // PUBLIC_URL is held to the strict origin rules of ValidatePublicURL // and, in an explicitly production environment // (DEPLOY_ENV/RSMON_ENV/GO_ENV = production), plain HTTP on a // non-loopback host is rejected. A legacy WORKER_URL (URLSource is // PublicURLSourceLegacy) is held only to the tolerant // ValidateAdvertisedURL so shapes that previously ran keep running; // it never fails production startup, preserving the historical // warn-only behavior. func ValidateHTTPConfig(c HTTPConfig, willListen bool) error { loginSet := c.Login != "" passSet := c.Password != "" if loginSet != passSet { return fmt.Errorf( "WORKER_LOGIN and WORKER_PASSWORD must be both set or both empty (got login=%s password=%s)", boolStr(loginSet), boolStr(passSet)) } if willListen && !c.IsAuthConfigured() { return fmt.Errorf("HTTP listener refused to start: WORKER_LOGIN and WORKER_PASSWORD must be set when WORKER_PORT > 0") } if c.PublicURL == "" { return nil } if c.URLSource == PublicURLSourceLegacy { if err := ValidateAdvertisedURL(c.PublicURL); err != nil { return fmt.Errorf("%s: %w", EnvWorkerURLLegacy, err) } return nil } if err := ValidatePublicURL(c.PublicURL); err != nil { return err } if productionEnv() { if host, warn := WarnInsecurePublicURL(c.PublicURL); warn { return fmt.Errorf( "%s uses plain HTTP on a non-loopback host (%s), which is not permitted in production; front %s with a TLS-terminating reverse proxy", EnvPublicURL, host, EnvPublicURL, ) } } return nil } // ValidatePublicURL enforces the strict origin shape required by // docs/public-endpoint-and-identity.md ("One Worker, One Public URL") // for the canonical PUBLIC_URL configuration: an absolute http(s) URL // with a scheme and authority and nothing else. Userinfo, query, // fragment, any path other than "/", and a missing hostname (for example // "https://:27401") are rejected because every route lives on the origin // root and credentials must not leak into the advertised endpoint. func ValidatePublicURL(raw string) error { u, err := parseURL(raw, EnvPublicURL) if err != nil { return err } if u.User != nil { return fmt.Errorf("%s must not contain userinfo (got %q)", EnvPublicURL, raw) } if u.RawQuery != "" { return fmt.Errorf("%s must not contain a query (got %q)", EnvPublicURL, raw) } if u.Fragment != "" { return fmt.Errorf("%s must not contain a fragment (got %q)", EnvPublicURL, raw) } if u.Path != "" && u.Path != "/" { return fmt.Errorf("%s must not contain a path (got %q)", EnvPublicURL, raw) } return nil } // ValidateAdvertisedURL enforces the tolerant absolute-URL shape used // for the legacy WORKER_URL and for endpoints supplied by the control // plane: an http(s) URL with a scheme, a host, and a non-empty hostname. // Unlike ValidatePublicURL it does not reject a path, userinfo, query, // or fragment, because legacy configurations and old control planes may // carry such shapes and must not newly fail. It still rejects values // that could never be dialed, such as "https://:27401". func ValidateAdvertisedURL(raw string) error { _, err := parseURL(raw, EnvWorkerURLLegacy) return err } // parseURL parses an absolute http(s) advertised URL. Both validators // share the scheme/host/hostname rules; label names the source in error // messages so diagnostics point at the variable the operator set. func parseURL(raw, label string) (*url.URL, error) { u, err := url.Parse(raw) if err != nil { return nil, fmt.Errorf("%s is not a valid URL: %v", label, err) } if u.Scheme == "" || u.Host == "" { return nil, fmt.Errorf("%s must be an absolute URL with scheme and host (got %q)", label, raw) } if u.Scheme != schemeHTTP && u.Scheme != schemeHTTPS { return nil, fmt.Errorf("%s scheme must be http or https (got %q)", label, u.Scheme) } if u.Hostname() == "" { return nil, fmt.Errorf("%s must include a host (got %q)", label, raw) } return u, nil } // productionEnv reports whether the worker is explicitly configured for // production. Only an explicit literal enables the HTTPS-only PUBLIC_URL // policy; an unset variable keeps the historical warn-on-plain-HTTP // behavior so existing installations upgrade without surprise. func productionEnv() bool { for _, key := range []string{"DEPLOY_ENV", "RSMON_ENV", "GO_ENV"} { switch strings.ToLower(strings.TrimSpace(os.Getenv(key))) { case "production", "prod": return true } } return false } // WarnInsecurePublicURL logs a warning when PUBLIC_URL uses plain http // for a non-loopback host. Returns the host so the caller can log it. // A no-op for the loopback case (typical local dev) and for https URLs. func WarnInsecurePublicURL(rawURL string) (host string, shouldWarn bool) { if rawURL == "" { return "", false } u, err := url.Parse(rawURL) if err != nil { return "", false } if u.Scheme != schemeHTTP { return u.Host, false } if isLoopbackHost(u.Hostname()) { return u.Host, false } return u.Host, true } func isLoopbackHost(host string) bool { host = strings.ToLower(host) for _, h := range strings.Split(loopbackHostnames, ",") { if host == strings.TrimSpace(h) { return true } } return false } func boolStr(b bool) string { if b { return "set" } return "empty" } func normalizeURL(endpoint string) string { endpoint = strings.TrimRight(endpoint, "/") if strings.HasSuffix(endpoint, "/api/worker") { return strings.TrimSuffix(endpoint, "/api/worker") } if strings.HasSuffix(endpoint, "/worker") { return strings.TrimSuffix(endpoint, "/worker") } return endpoint }