feat(queue): verify QStash webhooks before dispatch

Implements Upstash signature verification in Go (HS256 JWT, current +
next signing keys for rotation, exp/nbf/body/sub checks). The
dashboard's Node `Receiver.verify()` from `@upstash/qstash` was the
reference — we mirror the same claim set and header layout so both
services accept the same signed deliveries.

Wires the verifier into `POST /internal/jobs/{kind}`: reads the raw
body, verifies `Upstash-Signature` against the configured keys and the
canonical webhook URL, then dispatches through the existing
`QStash.Dispatch` handler registry. Accepts both full `Job` envelopes
(what the engine's own Enqueue emits) and bare payloads (what the
dashboard's `publishJSON` emits).

Config now surfaces `current_signing_key` / `next_signing_key` under
`queue.qstash`, with env-var overrides that accept both VLE_-prefixed
and bare names (QSTASH_CURRENT_SIGNING_KEY etc.) so ops can share env
with the dashboard.

Unit tests cover happy path, next-key rotation, wrong key, tampered
body, expired exp, wrong sub, malformed JWT, and URL-check-skipped.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

Author: hallelx2

cmd/engine/main.go

@@ -180,8 +180,10 @@ func buildQueue(c config.QueueConfig, dbURL string) (queue.Queue, error) {
 	switch c.Driver {
 	case "qstash":
 		return queue.NewQStash(queue.QStashConfig{
-			Token:          c.QStash.Token,
-			WebhookBaseURL: c.QStash.WebhookBaseURL,
+			Token:             c.QStash.Token,
+			WebhookBaseURL:    c.QStash.WebhookBaseURL,
+			CurrentSigningKey: c.QStash.CurrentSigningKey,
+			NextSigningKey:    c.QStash.NextSigningKey,
 		})
 	case "river":
 		return queue.NewRiver(queue.RiverConfig{

internal/api/server.go

@@ -419,8 +419,90 @@ func (d Deps) handleQuery(w http.ResponseWriter, r *http.Request) {
 
 // --- internal queue webhook ---
 
+// handleQueueWebhook is the endpoint QStash POSTs to. It verifies the
+// Upstash-Signature header, then dispatches the decoded payload into the
+// queue handler registered for {kind}.
+//
+// Only wired up when the configured queue is *queue.QStash; with other
+// drivers (River, Asynq) the route is present but returns 404-ish: there
+// is no webhook consumer to run.
 func (d Deps) handleQueueWebhook(w http.ResponseWriter, r *http.Request) {
-	// TODO(phase-1): verify QStash signature, decode body, dispatch.
+	qq, ok := d.Queue.(*queue.QStash)
+	if !ok {
+		writeErr(w, http.StatusNotFound, "webhook not enabled: queue driver is not qstash")
+		return
+	}
+
+	kind := queue.JobKind(chi.URLParam(r, "kind"))
+	if kind == "" {
+		writeErr(w, http.StatusBadRequest, "missing kind")
+		return
+	}
+
+	// Read the full body up front — verification hashes it, and the
+	// handler needs the raw bytes too.
+	body, err := io.ReadAll(r.Body)
+	if err != nil {
+		writeErr(w, http.StatusBadRequest, "read body: "+err.Error())
+		return
+	}
+	_ = r.Body.Close()
+
+	// Signature check. If no verifier is configured we refuse to proceed:
+	// an unauthenticated webhook endpoint that executes jobs is an open
+	// door to the worker. Local-only dev can set VLE_QSTASH_CURRENT_SIGNING_KEY
+	// to any string and sign test requests with it.
+	v := qq.Verifier()
+	if v == nil {
+		writeErr(w, http.StatusUnauthorized, "qstash signing key not configured")
+		return
+	}
+
+	// Reconstruct the URL QStash signed against. We prefer the configured
+	// WebhookBaseURL over r.Host — behind TLS terminators r.TLS and
+	// r.Host are unreliable, and the operator already told us the
+	// canonical external URL at boot.
+	expectedURL := strings.TrimRight(qq.WebhookBaseURL(), "/") + r.URL.Path
+
+	sig := r.Header.Get("Upstash-Signature")
+	if err := v.Verify(sig, body, expectedURL); err != nil {
+		if d.Logger != nil {
+			d.Logger.Warn("qstash verify failed", "err", err, "kind", kind)
+		}
+		writeErr(w, http.StatusUnauthorized, "invalid qstash signature")
+		return
+	}
+
+	// Body shape: either a full queue.Job (has `kind` + `payload`) or the
+	// bare payload (`kind` is already in the URL). Accept both — the
+	// dashboard publishes bare payloads today; the engine's own Enqueue
+	// publishes full Jobs.
+	payload := body
+	var maybe struct {
+		Kind    queue.JobKind   `json:"kind"`
+		Payload json.RawMessage `json:"payload"`
+	}
+	if err := json.Unmarshal(body, &maybe); err == nil && maybe.Kind != "" && len(maybe.Payload) > 0 {
+		if maybe.Kind != kind {
+			writeErr(w, http.StatusBadRequest, "kind in body does not match URL")
+			return
+		}
+		payload = maybe.Payload
+	}
+
+	if err := qq.Dispatch(r.Context(), kind, payload); err != nil {
+		if errors.Is(err, queue.ErrUnknownKind) {
+			writeErr(w, http.StatusNotFound, err.Error())
+			return
+		}
+		if d.Logger != nil {
+			d.Logger.Error("qstash dispatch failed", "err", err, "kind", kind)
+		}
+		// 5xx → QStash will retry per the publish-time Upstash-Retries header.
+		writeErr(w, http.StatusInternalServerError, err.Error())
+		return
+	}
+
 	w.WriteHeader(http.StatusNoContent)
 }
 

pkg/config/config.go

@@ -92,9 +92,16 @@ type QueueConfig struct {
 }
 
 // QStashBlock configures QStash.
+//
+// Token is the publish token (used to enqueue). CurrentSigningKey and
+// NextSigningKey are used to verify inbound webhooks; both are surfaced
+// on the Upstash console under "Signing Keys". NextSigningKey is only
+// populated while rotating.
 type QStashBlock struct {
-	Token          string `yaml:"token"`
-	WebhookBaseURL string `yaml:"webhook_base_url"`
+	Token             string `yaml:"token"`
+	WebhookBaseURL    string `yaml:"webhook_base_url"`
+	CurrentSigningKey string `yaml:"current_signing_key"`
+	NextSigningKey    string `yaml:"next_signing_key"`
 }
 
 // RiverBlock configures River.
@@ -209,6 +216,17 @@ func Load(path string) (Config, error) {
 	return cfg, nil
 }
 
+// firstEnv returns the first non-empty environment variable value from
+// the supplied names, checked in order.
+func firstEnv(names ...string) string {
+	for _, n := range names {
+		if v := os.Getenv(n); v != "" {
+			return v
+		}
+	}
+	return ""
+}
+
 func applyEnvOverrides(c *Config) {
 	// Minimal, deliberately shallow env handling — production-heavy values
 	// that are typically rotated live here. Extend as needed.
@@ -230,12 +248,22 @@ func applyEnvOverrides(c *Config) {
 	if v := os.Getenv("VLE_GEMINI_API_KEY"); v != "" {
 		c.LLM.Gemini.APIKey = v
 	}
-	if v := os.Getenv("VLE_QSTASH_TOKEN"); v != "" {
+	// Accept both VLE_-prefixed and bare QSTASH_* env vars. The bare
+	// names match what the Upstash console documents and what the
+	// dashboard already uses, so ops can set them once for both
+	// services. VLE_-prefixed wins if both are set.
+	if v := firstEnv("VLE_QSTASH_TOKEN", "QSTASH_TOKEN"); v != "" {
 		c.Queue.QStash.Token = v
 	}
-	if v := os.Getenv("VLE_QSTASH_WEBHOOK_BASE_URL"); v != "" {
+	if v := firstEnv("VLE_QSTASH_WEBHOOK_BASE_URL", "QSTASH_WEBHOOK_BASE_URL"); v != "" {
 		c.Queue.QStash.WebhookBaseURL = v
 	}
+	if v := firstEnv("VLE_QSTASH_CURRENT_SIGNING_KEY", "QSTASH_CURRENT_SIGNING_KEY"); v != "" {
+		c.Queue.QStash.CurrentSigningKey = v
+	}
+	if v := firstEnv("VLE_QSTASH_NEXT_SIGNING_KEY", "QSTASH_NEXT_SIGNING_KEY"); v != "" {
+		c.Queue.QStash.NextSigningKey = v
+	}
 	if v := os.Getenv("VLE_STORAGE_DRIVER"); v != "" {
 		c.Storage.Driver = v
 	}
@@ -273,6 +301,12 @@ func (c Config) Validate() error {
 		if c.Queue.QStash.Token == "" {
 			return errors.New("queue.qstash.token is required when driver=qstash")
 		}
+		if c.Queue.QStash.WebhookBaseURL == "" {
+			return errors.New("queue.qstash.webhook_base_url is required when driver=qstash")
+		}
+		if c.Queue.QStash.CurrentSigningKey == "" {
+			return errors.New("queue.qstash.current_signing_key is required when driver=qstash")
+		}
 	case "river":
 		if c.Database.URL == "" {
 			return errors.New("database.url is required when queue.driver=river")

pkg/queue/qstash.go

@@ -21,6 +21,14 @@ type QStashConfig struct {
 	// will POST jobs to WebhookBaseURL + "/internal/jobs/:kind".
 	WebhookBaseURL string
 
+	// CurrentSigningKey / NextSigningKey are the QStash webhook signing
+	// keys used to verify inbound deliveries. Both are surfaced on the
+	// Upstash console; the "next" key is only populated during key
+	// rotation. Required when the engine is the webhook consumer; unused
+	// by Enqueue.
+	CurrentSigningKey string
+	NextSigningKey    string
+
 	// HTTPClient is optional; if nil, http.DefaultClient is used.
 	HTTPClient *http.Client
 }
@@ -33,11 +41,17 @@ type QStashConfig struct {
 type QStash struct {
 	cfg      QStashConfig
 	client   *http.Client
+	verifier *Verifier
 	mu       sync.RWMutex
 	handlers map[JobKind]Handler
 }
 
 // NewQStash constructs a new QStash-backed Queue.
+//
+// If CurrentSigningKey is set, a Verifier is constructed and made
+// available via Verifier(); callers should apply it to the webhook
+// route. Without a signing key the engine will still accept callbacks
+// (useful for local dev) but log a loud warning elsewhere.
 func NewQStash(cfg QStashConfig) (*QStash, error) {
 	if cfg.Token == "" {
 		return nil, errors.New("qstash: token is required")
@@ -49,13 +63,33 @@ func NewQStash(cfg QStashConfig) (*QStash, error) {
 	if client == nil {
 		client = &http.Client{Timeout: 15 * time.Second}
 	}
-	return &QStash{
+	q := &QStash{
 		cfg:      cfg,
 		client:   client,
 		handlers: map[JobKind]Handler{},
-	}, nil
+	}
+	if cfg.CurrentSigningKey != "" {
+		v, err := NewVerifier(VerifierConfig{
+			CurrentSigningKey: cfg.CurrentSigningKey,
+			NextSigningKey:    cfg.NextSigningKey,
+		})
+		if err != nil {
+			return nil, err
+		}
+		q.verifier = v
+	}
+	return q, nil
 }
 
+// Verifier returns the configured signature verifier, or nil if no
+// signing key was supplied. The webhook handler uses this to reject
+// unsigned or forged deliveries.
+func (q *QStash) Verifier() *Verifier { return q.verifier }
+
+// WebhookBaseURL exposes the configured base URL so the webhook handler
+// can reconstruct the `sub` claim it expects.
+func (q *QStash) WebhookBaseURL() string { return q.cfg.WebhookBaseURL }
+
 func (q *QStash) Register(kind JobKind, h Handler) {
 	q.mu.Lock()
 	defer q.mu.Unlock()