diff --git a/cmd/backend/main.go b/cmd/backend/main.go index ea24adbb..6c468dba 100644 --- a/cmd/backend/main.go +++ b/cmd/backend/main.go @@ -581,18 +581,23 @@ func main() { credentialsRepository, cipherService, worker_orchestrator.WorkerEnvConfig{ - AppEnv: os.Getenv("APP_ENV"), - WorkerImage: getenvDefault("WORKER_IMAGE", "ghcr.io/warmbly/worker:latest"), - KafkaBootstrap: os.Getenv("KAFKA_BOOTSTRAP_SERVERS"), - KafkaSASLUsername: os.Getenv("KAFKA_SASL_USERNAME"), - KafkaSASLPassword: os.Getenv("KAFKA_SASL_PASSWORD"), - SchemaRegistryURL: os.Getenv("SCHEMA_REGISTRY_URL"), - SchemaRegistryKey: os.Getenv("SCHEMA_REGISTRY_KEY"), - SchemaRegistrySecret: os.Getenv("SCHEMA_REGISTRY_SECRET"), - RedisURL: os.Getenv("REDIS"), - AWSRegion: os.Getenv("AWS_REGION"), - AWSAccessKeyID: os.Getenv("WORKER_AWS_ACCESS_KEY_ID"), - AWSSecretAccessKey: os.Getenv("WORKER_AWS_SECRET_ACCESS_KEY"), + AppEnv: os.Getenv("APP_ENV"), + WorkerImage: getenvDefault("WORKER_IMAGE", "ghcr.io/warmbly/worker:latest"), + KafkaBootstrap: os.Getenv("KAFKA_BOOTSTRAP_SERVERS"), + KafkaSASLUsername: os.Getenv("KAFKA_SASL_USERNAME"), + KafkaSASLPassword: os.Getenv("KAFKA_SASL_PASSWORD"), + SchemaRegistryURL: os.Getenv("SCHEMA_REGISTRY_URL"), + SchemaRegistryKey: os.Getenv("SCHEMA_REGISTRY_KEY"), + SchemaRegistrySecret: os.Getenv("SCHEMA_REGISTRY_SECRET"), + RedisURL: os.Getenv("REDIS"), + AWSRegion: os.Getenv("AWS_REGION"), + AWSAccessKeyID: os.Getenv("WORKER_AWS_ACCESS_KEY_ID"), + AWSSecretAccessKey: os.Getenv("WORKER_AWS_SECRET_ACCESS_KEY"), + EncryptedKeysBackendURL: os.Getenv("ENCRYPTED_KEYS_BACKEND_URL"), + EncryptedKeysWorkerToken: os.Getenv("INTERNAL_API_TOKEN"), + EventBusProvider: os.Getenv("EVENTBUS_PROVIDER"), + NATSURL: os.Getenv("NATS_URL"), + CodecProvider: os.Getenv("CODEC_PROVIDER"), }, getenvDefault("WORKER_INSTALLER_PATH", "/app/scripts/install-worker.sh"), ) diff --git a/cmd/worker/main.go b/cmd/worker/main.go index 0b4929e5..5dd33f27 100644 --- a/cmd/worker/main.go +++ b/cmd/worker/main.go @@ -1,10 +1,14 @@ package main import ( + "bytes" "context" + "encoding/json" "log" + "net/http" "os" "os/signal" + "strings" "syscall" "time" @@ -164,6 +168,7 @@ func main() { // the rolling 1m counters into a WorkerHealth event, publishes via the // event bus so the consumer can write a row into worker_health_samples. go workerService.Heartbeat(ctx) + go runInternalHeartbeat(ctx, workerID, bindIP) go workerService.RunHealth(ctx, 30*time.Second) // Graceful shutdown @@ -183,6 +188,60 @@ func main() { log.Println("Worker stopped") } +func runInternalHeartbeat(ctx context.Context, workerID uuid.UUID, bindIP string) { + baseURL := strings.TrimRight(os.Getenv("ENCRYPTED_KEYS_BACKEND_URL"), "/") + token := os.Getenv("ENCRYPTED_KEYS_WORKER_TOKEN") + if baseURL == "" || token == "" { + return + } + reportedIP := os.Getenv("WORKER_PUBLIC_IP") + if reportedIP == "" && bindIP != "default route" { + reportedIP = bindIP + } + if reportedIP == "" { + reportedIP = "unknown" + } + + client := &http.Client{Timeout: 10 * time.Second} + send := func() { + payload := map[string]string{ + "worker_id": workerID.String(), + "bind_ip": reportedIP, + "tier": os.Getenv("WORKER_TIER"), + "egress_kind": os.Getenv("WORKER_EGRESS_KIND"), + } + body, _ := json.Marshal(payload) + req, err := http.NewRequestWithContext(ctx, http.MethodPost, baseURL+"/api/v1/internal/worker/heartbeat", bytes.NewReader(body)) + if err != nil { + log.Println("failed to build internal heartbeat:", err) + return + } + req.Header.Set("Authorization", "Bearer "+token) + req.Header.Set("Content-Type", "application/json") + resp, err := client.Do(req) + if err != nil { + log.Println("failed internal heartbeat:", err) + return + } + defer resp.Body.Close() + if resp.StatusCode >= 300 { + log.Println("internal heartbeat returned status", resp.StatusCode) + } + } + + send() + ticker := time.NewTicker(90 * time.Second) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + send() + } + } +} + // uuidNamespaceURL is the RFC 4122 URL namespace, matching the value used by // scripts/install-worker.sh when deriving the per-IP worker ID. Keep these in // sync: the installer and the worker must agree on the derivation. diff --git a/internal/api/handler/admin_workers_ssh.go b/internal/api/handler/admin_workers_ssh.go index 0be9f65a..a4c45373 100644 --- a/internal/api/handler/admin_workers_ssh.go +++ b/internal/api/handler/admin_workers_ssh.go @@ -107,7 +107,7 @@ func (h *Handler) AdminCreateWorker(c *gin.Context) { errx.JSON(c, errx.New(errx.Internal, "failed to generate enrollment token")) return } - enrollToken = hex.EncodeToString(raw) + enrollToken = "wmenroll_" + hex.EncodeToString(raw) sum := sha256.Sum256([]byte(enrollToken)) enrollHash = hex.EncodeToString(sum[:]) exp := time.Now().Add(2 * time.Hour) diff --git a/internal/api/handler/worker_enrollment.go b/internal/api/handler/worker_enrollment.go new file mode 100644 index 00000000..17622591 --- /dev/null +++ b/internal/api/handler/worker_enrollment.go @@ -0,0 +1,87 @@ +package handler + +import ( + "crypto/sha256" + "encoding/hex" + "net/http" + "strings" + + "github.com/gin-gonic/gin" + "github.com/warmbly/warmbly/internal/errx" + "github.com/warmbly/warmbly/internal/models" +) + +type workerEnrollmentRequest struct { + Token string `json:"token"` + PublicIP string `json:"public_ip,omitempty"` +} + +// EnrollWorker exchanges a one-time enrollment token for a complete worker +// dotenv file. It is intentionally public: the high-entropy one-time token is +// the credential, and it is consumed atomically before secrets are returned. +func (h *Handler) EnrollWorker(c *gin.Context) { + if h.WorkerRepo == nil || h.WorkerOrchestrator == nil { + errx.JSON(c, errx.New(errx.ServiceUnavailable, "worker enrollment is not configured")) + return + } + + var req workerEnrollmentRequest + if err := c.ShouldBindJSON(&req); err != nil { + errx.JSON(c, errx.New(errx.BadRequest, "invalid request body")) + return + } + req.Token = strings.TrimSpace(req.Token) + if req.Token == "" { + errx.JSON(c, errx.New(errx.BadRequest, "enrollment token is required")) + return + } + + sum := sha256.Sum256([]byte(req.Token)) + worker, err := h.WorkerRepo.ConsumeEnrollmentToken(c.Request.Context(), hex.EncodeToString(sum[:])) + if err != nil { + errx.JSON(c, errx.New(errx.Internal, "failed to consume enrollment token")) + return + } + if worker == nil { + errx.JSON(c, errx.New(errx.Unauthorized, "enrollment token is invalid or expired")) + return + } + + ip := strings.TrimSpace(req.PublicIP) + if ip == "" { + ip = c.ClientIP() + } + if ip != "" { + _ = h.WorkerRepo.RecordEnrolledIP(c.Request.Context(), worker.ID, ip) + } + + envFile, _, err := h.WorkerOrchestrator.RenderEnrollmentEnv(c.Request.Context(), worker.ID) + if err != nil { + errx.JSON(c, errx.New(errx.Internal, "failed to render worker config")) + return + } + envFile += "WORKER_TIER=" + workerTierLabel(worker) + "\n" + if ip != "" { + envFile += "WORKER_PUBLIC_IP=" + ip + "\n" + } + if worker.EgressKind != "" { + envFile += "WORKER_EGRESS_KIND=" + string(worker.EgressKind) + "\n" + } + + c.Header("Content-Type", "text/plain; charset=utf-8") + c.Header("Cache-Control", "no-store") + c.String(http.StatusOK, envFile) +} + +func workerTierLabel(w *models.Worker) string { + if w == nil { + return "shared_premium" + } + if w.WorkerType == models.WorkerTypeDedicated { + return "dedicated" + } + if w.FreeTier { + return "shared_free" + } + return "shared_premium" +} diff --git a/internal/api/routes.go b/internal/api/routes.go index c3574dc7..d02db4e6 100644 --- a/internal/api/routes.go +++ b/internal/api/routes.go @@ -38,6 +38,11 @@ func Run( r.POST("/api/v1/integrations/inbound/calendly/:secret", h.InboundCalendly) r.POST("/api/v1/integrations/inbound/cal-com/:secret", h.InboundCalCom) + // Public worker enrollment. The one-time enrollment token is the + // credential; successful exchange returns a dotenv file for the installer + // and consumes the token. + r.POST("/api/v1/workers/enroll", h.EnrollWorker) + // Public OAuth-bouncer pages used by the mailbox onboarding popup. // The provider redirects here; the page postMessages the code/state // back to the SPA opener which then calls /emails/onboarding/oauth/finish. diff --git a/internal/app/worker_orchestrator/orchestrator.go b/internal/app/worker_orchestrator/orchestrator.go index b4c40f1e..0a33e3e8 100644 --- a/internal/app/worker_orchestrator/orchestrator.go +++ b/internal/app/worker_orchestrator/orchestrator.go @@ -57,6 +57,13 @@ type WorkerEnvConfig struct { AWSRegion string AWSAccessKeyID string AWSSecretAccessKey string + + EncryptedKeysBackendURL string + EncryptedKeysWorkerToken string + + EventBusProvider string + NATSURL string + CodecProvider string } type Orchestrator struct { @@ -143,6 +150,26 @@ func (o *Orchestrator) Install(ctx context.Context, workerID uuid.UUID) error { return o.repo.UpdateInstallState(ctx, workerID, models.WorkerInstallStateInstalled, "") } +// RenderEnrollmentEnv returns a complete dotenv payload for the one-command +// enrollment flow. The token exchange authenticates the caller; this method +// only renders the config the installer writes to disk. +func (o *Orchestrator) RenderEnrollmentEnv(ctx context.Context, workerID uuid.UUID) (string, string, error) { + envContent, image, err := o.renderEnvFile(ctx, workerID) + if err != nil { + return "", "", err + } + var b strings.Builder + b.WriteString("# Warmbly worker enrollment config\n") + b.WriteString("WORKER_ID=") + b.WriteString(workerID.String()) + b.WriteString("\n") + b.WriteString("WARMBLY_WORKER_IMAGE=") + b.WriteString(image) + b.WriteString("\n") + b.WriteString(envContent) + return b.String(), image, nil +} + // ApplyConfig re-writes /etc/warmbly/worker.env from the worker's current // profile + AWS creds and restarts the service. Cheaper than Install — does // not touch Docker or the installer script. Use after a credential change. @@ -578,6 +605,9 @@ func (o *Orchestrator) renderEnvFile(ctx context.Context, workerID uuid.UUID) (e write("AWS_REGION", env.AWSRegion) write("AWS_ACCESS_KEY_ID", env.AWSAccessKeyID) write("AWS_SECRET_ACCESS_KEY", env.AWSSecretAccessKey) + write("ENCRYPTED_KEYS_PROVIDER", "http") + write("ENCRYPTED_KEYS_BACKEND_URL", env.EncryptedKeysBackendURL) + write("ENCRYPTED_KEYS_WORKER_TOKEN", env.EncryptedKeysWorkerToken) write("KAFKA_BOOTSTRAP_SERVERS", env.KafkaBootstrap) write("KAFKA_SASL_USERNAME", env.KafkaSASLUsername) write("KAFKA_SASL_PASSWORD", env.KafkaSASLPassword) @@ -585,6 +615,9 @@ func (o *Orchestrator) renderEnvFile(ctx context.Context, workerID uuid.UUID) (e write("SCHEMA_REGISTRY_KEY", env.SchemaRegistryKey) write("SCHEMA_REGISTRY_SECRET", env.SchemaRegistrySecret) write("REDIS", env.RedisURL) + write("EVENTBUS_PROVIDER", env.EventBusProvider) + write("NATS_URL", env.NATSURL) + write("CODEC_PROVIDER", env.CodecProvider) return b.String(), image, nil } diff --git a/internal/repository/pg_worker_heartbeat.go b/internal/repository/pg_worker_heartbeat.go index 9aa12c45..889be2d0 100644 --- a/internal/repository/pg_worker_heartbeat.go +++ b/internal/repository/pg_worker_heartbeat.go @@ -24,6 +24,13 @@ func (r *workerRepository) UpsertOnHeartbeat(ctx context.Context, id uuid.UUID, VALUES ($1, $2, $3, TRUE, $4, $5, $6, 'healthy', 0) ON CONFLICT (id) DO UPDATE SET ip_addr = EXCLUDED.ip_addr, + active = TRUE, + install_state = CASE + WHEN workers.install_state IN ('pending', 'provisioning', 'error') THEN 'installed'::worker_install_state + ELSE workers.install_state + END, + last_seen_at = now(), + last_error = NULL, updated_at = now() ` name := "auto-registered-" + id.String()[:8] diff --git a/internal/repository/pg_worker_ssh.go b/internal/repository/pg_worker_ssh.go index 727f402b..9c1425eb 100644 --- a/internal/repository/pg_worker_ssh.go +++ b/internal/repository/pg_worker_ssh.go @@ -255,7 +255,11 @@ func (r *workerRepository) ListWorkersByProfile(ctx context.Context, profileID u func (r *workerRepository) RecordEnrolledIP(ctx context.Context, id uuid.UUID, ip string) error { _, err := r.db.Exec(ctx, ` - UPDATE workers SET ip_addr = $2, ssh_host = COALESCE(NULLIF(ssh_host,''), $2), updated_at = NOW() + UPDATE workers + SET ip_addr = $2, + ssh_host = COALESCE(NULLIF(ssh_host,''), $2), + install_state = 'provisioning', + updated_at = NOW() WHERE id = $1 `, id, ip) return err diff --git a/scripts/install-worker.sh b/scripts/install-worker.sh index ee71e420..83d6712c 100755 --- a/scripts/install-worker.sh +++ b/scripts/install-worker.sh @@ -24,6 +24,11 @@ # curl -fsSL https://get.warmbly.com/worker | sudo bash -s -- \ # --env-file /root/worker.env # +# Or with a one-time enrollment token from the dashboard: +# +# curl -fsSL https://get.warmbly.com/worker | sudo bash -s -- \ +# --enroll wmenroll_... +# # Re-running is safe: existing env values are preserved unless overridden, # and the worker ID will resolve to the same value as long as the IP is stable. @@ -43,6 +48,8 @@ CONTAINER_NAME="warmbly-worker" ACTION="install" INTERACTIVE=1 SUPPLIED_ENV_FILE="" +ENROLL_TOKEN="" +API_BASE="${WARMBLY_API_BASE:-https://api.warmbly.com}" # Comma-separated list of IPv4 addresses for multi-IP install. When non-empty, # the installer drops one templated systemd unit per IP, each bound to that IP @@ -111,6 +118,9 @@ Configuration flags: --tier Worker tier label (default: shared) --image Docker image (default: ${IMAGE}) --env-file Use this env file verbatim, skip prompts + --enroll Exchange a one-time dashboard enrollment token + for worker config and install without prompts + --api-base API base for --enroll (default: ${API_BASE}) --kafka Kafka bootstrap servers (host:port[,host:port]) --kafka-user @@ -151,6 +161,8 @@ while [[ $# -gt 0 ]]; do --tier) CFG[WORKER_TIER]="$2"; shift 2 ;; --image) IMAGE="$2"; shift 2 ;; --env-file) SUPPLIED_ENV_FILE="$2"; shift 2 ;; + --enroll) ENROLL_TOKEN="$2"; INTERACTIVE=0; shift 2 ;; + --api-base) API_BASE="$2"; shift 2 ;; --kafka) CFG[KAFKA_BOOTSTRAP_SERVERS]="$2"; shift 2 ;; --kafka-user) CFG[KAFKA_SASL_USERNAME]="$2"; shift 2 ;; @@ -315,6 +327,54 @@ merge_existing_env() { done < "$ENV_FILE" } +env_file_value() { + local file="$1" key="$2" + grep -E "^${key}=" "$file" 2>/dev/null | head -1 | cut -d= -f2- +} + +fetch_enrollment_env() { + [[ -n "$ENROLL_TOKEN" ]] || return 1 + command -v curl >/dev/null 2>&1 || die "curl is required for --enroll" + + local ip="${IP_OVERRIDE}" + if [[ -z "$ip" ]]; then + ip="$(detect_public_ip || true)" + fi + + local tmp; tmp="$(mktemp)" + local body + body="{\"token\":\"${ENROLL_TOKEN}\"" + if [[ -n "$ip" ]]; then + body+=",\"public_ip\":\"${ip}\"" + fi + body+="}" + + log "exchanging enrollment token at ${API_BASE}" + curl -fsS \ + -X POST \ + -H "Content-Type: application/json" \ + -H "Accept: text/plain" \ + --data "$body" \ + "${API_BASE%/}/api/v1/workers/enroll" > "$tmp" || { + rm -f "$tmp" + die "enrollment failed" + } + + local worker_id image + worker_id="$(env_file_value "$tmp" WORKER_ID)" + image="$(env_file_value "$tmp" WARMBLY_WORKER_IMAGE)" + [[ -n "$worker_id" ]] || die "enrollment response did not include WORKER_ID" + + install -d -m 0700 "$CONFIG_DIR" + install -m 0600 "$tmp" "$ENV_FILE" + rm -f "$tmp" + + WORKER_ID_OVERRIDE="$worker_id" + [[ -n "$image" ]] && IMAGE="$image" + ok "enrollment config installed" + return 0 +} + write_env_file() { install -d -m 0700 "$CONFIG_DIR" local tmp; tmp="$(mktemp)" @@ -322,10 +382,11 @@ write_env_file() { echo "# Warmbly worker config — managed by install-worker.sh" echo "# $(date -u +%Y-%m-%dT%H:%M:%SZ)" for key in APP_ENV AWS_CONFIG_ENABLED AWS_REGION AWS_ACCESS_KEY_ID AWS_SECRET_ACCESS_KEY \ + ENCRYPTED_KEYS_PROVIDER ENCRYPTED_KEYS_BACKEND_URL ENCRYPTED_KEYS_WORKER_TOKEN \ KAFKA_BOOTSTRAP_SERVERS KAFKA_SASL_USERNAME KAFKA_SASL_PASSWORD \ SCHEMA_REGISTRY_URL SCHEMA_REGISTRY_KEY SCHEMA_REGISTRY_SECRET \ - REDIS WORKER_TIER; do - printf '%s=%s\n' "$key" "${CFG[$key]}" + REDIS EVENTBUS_PROVIDER NATS_URL CODEC_PROVIDER WORKER_TIER WORKER_PUBLIC_IP WORKER_EGRESS_KIND; do + printf '%s=%s\n' "$key" "${CFG[$key]:-}" done for kv in "${EXTRA_ENVS[@]}"; do printf '%s\n' "$kv" @@ -499,6 +560,10 @@ list_installed_instances() { # ---------- actions ---------- prepare_common_env() { + if fetch_enrollment_env; then + return + fi + if [[ -n "$SUPPLIED_ENV_FILE" ]]; then [[ -f "$SUPPLIED_ENV_FILE" ]] || die "env file not found: $SUPPLIED_ENV_FILE" install -d -m 0700 "$CONFIG_DIR"