From 1b1b3d530c74328fab4781656dbe5db0452ec9bf Mon Sep 17 00:00:00 2001 From: Chris Li Date: Thu, 16 Jul 2026 11:54:09 -0700 Subject: [PATCH 1/3] feat: {{state.*}} per-workflow state binding + REST options.auth GoPlus provider MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two generic, reusable gateway extensions (the guardian monitor is the first consumer; see PLAN_WORKFLOW_STATE_GUARDIAN_MONITORING.md): 1. {{state.*}} — durable, cross-run, per-WORKFLOW state exposed to customCode as state.get/set/list, backed by a new wfstate:: BadgerDB namespace (scoped by taskId only — isolated per workflow, no user coupling). Writes no-op to an in-memory scratch in simulation / single-node runs / when no DB is wired, so nodes:run + workflows:simulate previews never mutate live state. Cascade-deleted on task teardown. Bound after step vars so a node named "state" can't shadow it. 2. REST options.auth — a restApi node with config.options.auth.provider: "goplus" gets a GoPlus signed session token minted + cached server-side from macros.secrets (goplus_app_key/secret; sha1(app_key+time+app_secret) -> /v1/token, refreshed ~60s early) and injected as the Authorization header. The secret never appears in the (client-visible) workflow JSON; keyless GoPlus fallback when keys are unset. Provider registry starts with GoPlus. Additive storage (new wfstate: namespace, no migration). New Go tests cover the state set/get/list roundtrip, cross-run persistence, the simulation no-op guard, and options.auth parsing. Verified end-to-end: the ava-sdk-js guardian live test verdict step now passes against a local gateway. --- config/gateway.example.yaml | 9 ++ config/test.example.yaml | 7 ++ core/taskengine/engine.go | 11 ++ core/taskengine/schema.go | 17 +++ core/taskengine/vm_runner_customcode.go | 93 ++++++++++++++ core/taskengine/vm_runner_rest.go | 103 +++++++++++++++ core/taskengine/vm_workflow_state_test.go | 145 ++++++++++++++++++++++ 7 files changed, 385 insertions(+) create mode 100644 core/taskengine/vm_workflow_state_test.go diff --git a/config/gateway.example.yaml b/config/gateway.example.yaml index 7e2248f9..354ff689 100644 --- a/config/gateway.example.yaml +++ b/config/gateway.example.yaml @@ -114,6 +114,15 @@ macros: sendgrid_key: "" thegraph_api_key: "" moralis_api_key: "" + # GoPlus signed-token auth for restApi nodes using `options.auth.provider: goplus`. + # Leave blank to fall back to keyless GoPlus (lower rate limits). + goplus_app_key: "" + goplus_app_secret: "" + # guardian_ruleset — the tunable verdict knobs the guardian customCode reads via + # {{apContext.configVars.guardian_ruleset}} (bare-injected as a JS object). MUST be + # a valid JSON object and MUST be set (an unresolved template in a customCode source + # hard-fails the node). Single-quote it so the inner double quotes survive YAML. + guardian_ruleset: '{"weak":["honeypot_related_address","blacklist_doubt"]}' # Optional: post-execution AI summarization. Disabled by default in dev. notifications: diff --git a/config/test.example.yaml b/config/test.example.yaml index eac3f623..8ad39ae8 100644 --- a/config/test.example.yaml +++ b/config/test.example.yaml @@ -91,6 +91,13 @@ macros: sendgrid_key: "" thegraph_api_key: "" moralis_api_key: "" + # GoPlus signed-token auth for restApi nodes using `options.auth.provider: goplus`. + # Leave blank to fall back to keyless GoPlus (lower rate limits). + goplus_app_key: "" + goplus_app_secret: "" + # guardian_ruleset — tunable verdict knobs read via {{apContext.configVars.guardian_ruleset}} + # (bare-injected as a JS object; must be valid JSON and must be set). + guardian_ruleset: '{"weak":["honeypot_related_address","blacklist_doubt"]}' # Optional: post-execution AI summarization. Disabled by default in tests. notifications: diff --git a/core/taskengine/engine.go b/core/taskengine/engine.go index 60a69b35..10c3dc9a 100644 --- a/core/taskengine/engine.go +++ b/core/taskengine/engine.go @@ -4603,6 +4603,17 @@ func (n *Engine) DeleteWorkflowByUser(user *model.User, taskID string) (*avsprot // checkpoint/wake — and any operator internal trigger — don't outlive the workflow. n.gcDurableStateForTask(taskID) + // Cascade-delete this workflow's durable per-run state ({{state.*}} / + // wfstate:) so it doesn't outlive the task (scoped by taskID, so this + // touches only this workflow's namespace). + if stateKeys, listErr := n.db.ListKeys(string(WorkflowStatePrefix(taskID))); listErr == nil { + for _, k := range stateKeys { + if delErr := n.db.Delete([]byte(k)); delErr != nil { + n.logger.Warn("failed to delete workflow state key", "key", k, "error", delErr) + } + } + } + n.logger.Info("📢 Starting operator notifications", "task_id", taskID) n.notifyOperatorsTaskOperation(taskID, avsproto.MessageOp_DeleteTask) n.logger.Info("✅ Delete task operation completed", "task_id", taskID) diff --git a/core/taskengine/schema.go b/core/taskengine/schema.go index 571ca384..f92c27a0 100644 --- a/core/taskengine/schema.go +++ b/core/taskengine/schema.go @@ -48,6 +48,23 @@ func WalletStorageKey(chainID int64, owner common.Address, smartWalletAddress st ) } +// WorkflowStateKey returns the key for one entry of a workflow's durable, +// cross-run mutable state — the `{{state.*}}` store exposed to customCode. +// +// Scoped ONLY by taskID (no owner, no chain): every workflow has an isolated +// namespace, so a user with several monitoring workflows never has their state +// collide, and deleting a workflow can wipe exactly its state (see +// WorkflowStatePrefix). `stateKey` is a client-defined opaque string. +func WorkflowStateKey(taskID, stateKey string) []byte { + return []byte(fmt.Sprintf("wfstate:%s:%s", taskID, stateKey)) +} + +// WorkflowStatePrefix returns the scan/cascade-delete prefix for all `{{state.*}}` +// entries of one workflow. +func WorkflowStatePrefix(taskID string) []byte { + return []byte(fmt.Sprintf("wfstate:%s:", taskID)) +} + // WalletBySaltKey returns the secondary index key that maps a // (chainID, owner, factory, salt) tuple to its current canonical wallet // address. diff --git a/core/taskengine/vm_runner_customcode.go b/core/taskengine/vm_runner_customcode.go index abb2febc..28c191bf 100644 --- a/core/taskengine/vm_runner_customcode.go +++ b/core/taskengine/vm_runner_customcode.go @@ -1,6 +1,7 @@ package taskengine import ( + "encoding/json" "fmt" "math/big" "reflect" @@ -16,6 +17,90 @@ import ( avsproto "github.com/AvaProtocol/EigenLayer-AVS/protobuf" ) +// installStateBinding exposes `state.get/set/list` to customCode JS, backed by the +// per-workflow `wfstate:` namespace (WorkflowStateKey). This is the durable, +// cross-run, client-defined state store — the `{{state.*}}` capability. +// +// - state.get(key) -> the stored JSON value, or undefined +// - state.set(key, value) -> persists arbitrary JSON under this workflow's namespace +// - state.list(prefix) -> the stateKeys (prefix included) currently stored +// +// Writes never hit the DB in simulation, for a single-node run (no task id), or when +// no DB is wired: those use an in-memory scratch map so read-after-write still works +// within one run without mutating real state. Bind this AFTER the step vars so a node +// accidentally named "state" can't shadow it. +func installStateBinding(jsvm *goja.Runtime, vm *VM) { + obj := jsvm.NewObject() + + taskID := "" + if vm != nil { + taskID = vm.GetTaskId() + } + // scratch-only when we must not (or cannot) persist. + scratchOnly := vm == nil || vm.db == nil || taskID == "" || vm.IsSimulation + scratch := map[string][]byte{} + + obj.Set("get", func(key string) interface{} { + if key == "" { + return goja.Undefined() + } + var raw []byte + if scratchOnly { + raw = scratch[key] + } else if b, err := vm.db.GetKey(WorkflowStateKey(taskID, key)); err == nil { + raw = b + } + if raw == nil { + return goja.Undefined() + } + var value interface{} + if err := json.Unmarshal(raw, &value); err != nil { + return goja.Undefined() + } + return value + }) + + obj.Set("set", func(key string, value interface{}) { + if key == "" { + return + } + encoded, err := json.Marshal(value) + if err != nil { + return + } + if scratchOnly { + scratch[key] = encoded + return + } + if setErr := vm.db.Set(WorkflowStateKey(taskID, key), encoded); setErr != nil && vm.logger != nil { + vm.logger.Warn("state.set failed", "key", key, "error", setErr) + } + }) + + obj.Set("list", func(prefix string) []string { + out := []string{} + if scratchOnly { + for k := range scratch { + if strings.HasPrefix(k, prefix) { + out = append(out, k) + } + } + return out + } + full := string(WorkflowStatePrefix(taskID)) + keys, err := vm.db.ListKeys(full + prefix) + if err != nil { + return out + } + for _, k := range keys { + out = append(out, strings.TrimPrefix(k, full)) + } + return out + }) + + jsvm.Set("state", obj) +} + type JSProcessor struct { *CommonProcessor jsvm *goja.Runtime @@ -73,6 +158,10 @@ func NewJSProcessor(vm *VM) *JSProcessor { } } + // Expose {{state.*}} — durable per-workflow state. Bound last so a step var + // accidentally named "state" cannot shadow it. + installStateBinding(r.jsvm, vm) + return &r } @@ -130,6 +219,10 @@ func NewJSProcessorWithIsolatedVars(vm *VM, isolatedVars map[string]any) *JSProc } } + // Expose {{state.*}} — durable per-workflow state. Bound last so an isolated var + // accidentally named "state" cannot shadow it. + installStateBinding(r.jsvm, vm) + return &r } diff --git a/core/taskengine/vm_runner_rest.go b/core/taskengine/vm_runner_rest.go index e8afdec1..82d6d21a 100644 --- a/core/taskengine/vm_runner_rest.go +++ b/core/taskengine/vm_runner_rest.go @@ -1,11 +1,16 @@ package taskengine import ( + "bytes" + "crypto/sha1" + "encoding/hex" "encoding/json" "fmt" "net/http" "net/http/httptest" + "strconv" "strings" + "sync" "time" avsproto "github.com/AvaProtocol/EigenLayer-AVS/protobuf" @@ -571,6 +576,15 @@ func (r *RestProcessor) Execute(stepID string, node *avsproto.RestAPINode) (*avs processedHeaders[processedKey] = processedValue } + // Server-side auth providers (options.auth): mint + attach a provider token so the + // credential never lives in the workflow JSON. GoPlus first; on "" (keys unset or + // mint failed) we send no Authorization header — GoPlus still answers keyless. + if restAuthProvider(node) == "goplus" { + if token := goplusAuthHeader(); token != "" { + processedHeaders["Authorization"] = token + } + } + // Only apply JSON preprocessing if content type is JSON contentType := "" for key, value := range processedHeaders { @@ -862,6 +876,95 @@ func detectNotificationProvider(u string) string { return "" } +// ── REST auth providers ───────────────────────────────────────────────────── +// A restApi node can request server-side credential injection with +// +// config.options.auth = { "provider": "goplus" } +// +// The gateway mints the provider's token from macros.secrets and attaches it as +// the Authorization header at execution time, so the secret never appears in the +// (client-visible) workflow JSON. GoPlus is the first provider; add more here. + +// restAuthProvider returns the lower-cased options.auth.provider (e.g. "goplus"), +// or "" when none is set. Mirrors shouldSummarize's options-bag read. +func restAuthProvider(node *avsproto.RestAPINode) string { + if node == nil || node.Config == nil || node.Config.Options == nil { + return "" + } + optsMap, ok := node.Config.Options.AsInterface().(map[string]interface{}) + if !ok { + return "" + } + auth, ok := optsMap["auth"].(map[string]interface{}) + if !ok { + return "" + } + provider, _ := auth["provider"].(string) + return strings.ToLower(strings.TrimSpace(provider)) +} + +// goplusTokenCache holds the module-cached GoPlus session token, refreshed shortly +// before expiry. The token value already includes the "Bearer " prefix GoPlus attaches. +var goplusTokenCache struct { + sync.RWMutex + token string + expires time.Time +} + +// goplusAuthHeader returns the Authorization header value for GoPlus, minting + +// caching a signed session token from macros.secrets (goplus_app_key / +// goplus_app_secret): sign = sha1_hex(app_key + time + app_secret) → POST /v1/token. +// Returns "" to fall back to keyless GoPlus access (lower rate limits) when the +// keys are unset or minting fails — the caller then sends no Authorization header. +func goplusAuthHeader() string { + appKey := GetMacroSecret("goplus_app_key") + appSecret := GetMacroSecret("goplus_app_secret") + if appKey == "" || appSecret == "" { + return "" + } + + goplusTokenCache.RLock() + if goplusTokenCache.token != "" && time.Until(goplusTokenCache.expires) > 60*time.Second { + token := goplusTokenCache.token + goplusTokenCache.RUnlock() + return token + } + goplusTokenCache.RUnlock() + + goplusTokenCache.Lock() + defer goplusTokenCache.Unlock() + // Re-check under the write lock (another goroutine may have refreshed). + if goplusTokenCache.token != "" && time.Until(goplusTokenCache.expires) > 60*time.Second { + return goplusTokenCache.token + } + + now := time.Now().Unix() + sum := sha1.Sum([]byte(appKey + strconv.FormatInt(now, 10) + appSecret)) + reqBody, _ := json.Marshal(map[string]interface{}{ + "app_key": appKey, + "sign": hex.EncodeToString(sum[:]), + "time": now, + }) + resp, err := http.Post("https://api.gopluslabs.io/api/v1/token", "application/json", bytes.NewReader(reqBody)) + if err != nil { + return "" + } + defer resp.Body.Close() + var parsed struct { + Code int `json:"code"` + Result struct { + AccessToken string `json:"access_token"` + ExpiresIn int64 `json:"expires_in"` + } `json:"result"` + } + if decodeErr := json.NewDecoder(resp.Body).Decode(&parsed); decodeErr != nil || parsed.Code != 1 || parsed.Result.AccessToken == "" { + return "" + } + goplusTokenCache.token = parsed.Result.AccessToken + goplusTokenCache.expires = time.Now().Add(time.Duration(parsed.Result.ExpiresIn) * time.Second) + return parsed.Result.AccessToken +} + // escapeJSONString properly escapes a string for use within JSON func escapeJSONString(s string) string { // Use Go's built-in JSON marshaling to properly escape the string diff --git a/core/taskengine/vm_workflow_state_test.go b/core/taskengine/vm_workflow_state_test.go new file mode 100644 index 00000000..c0afdd92 --- /dev/null +++ b/core/taskengine/vm_workflow_state_test.go @@ -0,0 +1,145 @@ +package taskengine + +import ( + "encoding/json" + "testing" + + "github.com/AvaProtocol/EigenLayer-AVS/core/testutil" + "github.com/AvaProtocol/EigenLayer-AVS/model" + avsproto "github.com/AvaProtocol/EigenLayer-AVS/protobuf" + "github.com/AvaProtocol/EigenLayer-AVS/storage" + "github.com/stretchr/testify/require" + "google.golang.org/protobuf/types/known/structpb" +) + +// runCustomCodeForState executes `source` as a customCode node inside a task with +// the given taskID, wired to db. Returns after asserting the step succeeded. +func runCustomCodeForState(t *testing.T, db storage.Storage, taskID, source string, simulation bool) { + t.Helper() + taskNode := &avsproto.TaskNode{ + Id: "code1", + Name: "code1", + TaskType: &avsproto.TaskNode_CustomCode{ + CustomCode: &avsproto.CustomCodeNode{ + Config: &avsproto.CustomCodeNode_Config{ + Lang: avsproto.Lang_LANG_JAVASCRIPT, + Source: source, + }, + }, + }, + } + trigger := &avsproto.TaskTrigger{Id: "trigger", Name: "trigger"} + vm, err := NewVMWithData(&model.Workflow{ + Task: &avsproto.Task{ + Id: taskID, + Nodes: []*avsproto.TaskNode{taskNode}, + Edges: []*avsproto.TaskEdge{{Id: "e1", Source: trigger.Id, Target: "code1"}}, + Trigger: trigger, + }, + }, nil, testutil.GetTestSmartWalletConfig(), nil) + require.NoError(t, err) + vm.WithDb(db) + if simulation { + vm.SetSimulation(true) + } + + proc := NewJSProcessor(vm) + step, err := proc.Execute("code1", taskNode.GetCustomCode()) + require.NoError(t, err) + require.True(t, step.Success, "customCode step failed: %s", step.Error) +} + +func readStateMap(t *testing.T, db storage.Storage, taskID, stateKey string) map[string]interface{} { + t.Helper() + raw, err := db.GetKey(WorkflowStateKey(taskID, stateKey)) + require.NoError(t, err, "expected wfstate key %s to exist", stateKey) + var m map[string]interface{} + require.NoError(t, json.Unmarshal(raw, &m)) + return m +} + +// TestWorkflowStateBinding_SetGetListPersist covers the {{state.*}} binding: +// set/get/list within a run, real persistence under wfstate:, and — with a +// fresh VM/processor on the same db+taskId — cross-run persistence. +func TestWorkflowStateBinding_SetGetListPersist(t *testing.T) { + db := testutil.TestMustDB() + defer db.Close() + const taskID = "state-test-task" + + // Run 1: set two keys, then probe get + list by writing the results back so we + // can assert them via the DB (avoids parsing structpb output). + runCustomCodeForState(t, db, taskID, ` + state.set("ntfy:a", { x: 1 }); + state.set("seen:b", { y: 2 }); + var got = state.get("ntfy:a"); + var keys = state.list("ntfy:"); + state.set("_probe", { gotX: got && got.x, keyCount: keys.length, firstKey: keys[0] }); + return true; + `, false) + + // Values persisted under wfstate::. + require.Equal(t, float64(1), readStateMap(t, db, taskID, "ntfy:a")["x"]) + require.Equal(t, float64(2), readStateMap(t, db, taskID, "seen:b")["y"]) + + // The probe confirms get + list worked inside the JS: list("ntfy:") matched + // exactly the one ntfy: key (not seen:b). + probe := readStateMap(t, db, taskID, "_probe") + require.Equal(t, float64(1), probe["gotX"]) + require.Equal(t, float64(1), probe["keyCount"]) + require.Equal(t, "ntfy:a", probe["firstKey"]) + + // Run 2: a FRESH VM/processor on the same db + taskID still sees run-1 state. + runCustomCodeForState(t, db, taskID, ` + var a = state.get("ntfy:a"); + state.set("_probe2", { persisted: a && a.x }); + return true; + `, false) + require.Equal(t, float64(1), readStateMap(t, db, taskID, "_probe2")["persisted"]) +} + +// TestWorkflowStateBinding_SimulationDoesNotPersist: in simulation, writes go to an +// in-memory scratch, never the real DB — so a nodes:run / simulate preview can't +// mutate a live workflow's state. +func TestWorkflowStateBinding_SimulationDoesNotPersist(t *testing.T) { + db := testutil.TestMustDB() + defer db.Close() + const taskID = "state-sim-task" + + // Read-after-write still works within the run (scratch), but nothing hits the DB. + runCustomCodeForState(t, db, taskID, ` + state.set("ntfy:x", { v: 1 }); + if (!state.get("ntfy:x")) { throw new Error("scratch read-after-write should work"); } + return true; + `, true) + + _, err := db.GetKey(WorkflowStateKey(taskID, "ntfy:x")) + require.Error(t, err, "simulation writes must not reach the DB") +} + +// TestRestAuthProvider covers the options.auth.provider parsing that gates +// server-side GoPlus token injection. +func TestRestAuthProvider(t *testing.T) { + mk := func(opts map[string]interface{}) *avsproto.RestAPINode { + var o *structpb.Value + if opts != nil { + v, err := structpb.NewValue(opts) + require.NoError(t, err) + o = v + } + return &avsproto.RestAPINode{Config: &avsproto.RestAPINode_Config{Options: o}} + } + + require.Equal(t, "goplus", restAuthProvider(mk(map[string]interface{}{ + "auth": map[string]interface{}{"provider": "goplus"}, + }))) + // Trimmed + lower-cased. + require.Equal(t, "goplus", restAuthProvider(mk(map[string]interface{}{ + "auth": map[string]interface{}{"provider": " GoPlus "}, + }))) + // No auth key. + require.Equal(t, "", restAuthProvider(mk(map[string]interface{}{"summarize": true}))) + // No options / no config. + require.Equal(t, "", restAuthProvider(mk(nil))) + require.Equal(t, "", restAuthProvider(&avsproto.RestAPINode{})) + require.Equal(t, "", restAuthProvider(nil)) +} From 0f95f6a1b01173e56be660f4e75a7373b5acc9ff Mon Sep 17 00:00:00 2001 From: Chris Li Date: Thu, 16 Jul 2026 13:11:52 -0700 Subject: [PATCH 2/3] feat: log GoPlus auth attach (Debug) / keyless fallback (Warn) on REST options.auth Confirmed live against a local gateway with real GoPlus keys in macros.secrets: options.auth survives create->store->execute, the token mints, and the Authorization header is attached (authed request). Add a Debug log on attach and a Warn on keyless fallback (via the VM logger, since the package globalLogger is nil in the aggregator path). --- core/taskengine/vm_runner_rest.go | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/core/taskengine/vm_runner_rest.go b/core/taskengine/vm_runner_rest.go index 82d6d21a..a1c41b77 100644 --- a/core/taskengine/vm_runner_rest.go +++ b/core/taskengine/vm_runner_rest.go @@ -582,6 +582,12 @@ func (r *RestProcessor) Execute(stepID string, node *avsproto.RestAPINode) (*avs if restAuthProvider(node) == "goplus" { if token := goplusAuthHeader(); token != "" { processedHeaders["Authorization"] = token + if r.vm.logger != nil { + r.vm.logger.Debug("REST: attached GoPlus session token (authed)", "stepID", stepID) + } + } else if r.vm.logger != nil { + // Keys unset or mint failed: GoPlus still answers keyless (lower limits). + r.vm.logger.Warn("REST: GoPlus auth requested but no token minted; falling back to keyless", "stepID", stepID) } } From 9ae0266a2871e42f114e1bd296d8337c75232f40 Mon Sep 17 00:00:00 2001 From: Chris Li Date: Thu, 16 Jul 2026 14:44:51 -0700 Subject: [PATCH 3/3] fix: address Copilot review on state binding + GoPlus auth - state.list: sort results in both scratch and DB paths so simulation and real runs agree on ordering (Badger iterates sorted; a Go map does not). - state scratch: move from a per-JSProcessor closure to the VM (vm.stateScratch, mutex-guarded) so all customCode steps of one simulation run share it and read-after-write stays coherent across nodes. - GoPlus token cache: key by app_key so a rotated key never reuses a stale token. - goplusAuthHeader: use an http.Client with a 15s timeout (not http.Post), check for a 2xx status, and normalize the token to a "Bearer " prefix defensively. - Add a goplusTokenProvider seam + a REST test asserting the Authorization header is attached when a token is available and left unset for the keyless fallback. - Log the ListKeys error in the wfstate cascade-delete so orphaned state is visible. --- core/taskengine/engine.go | 2 + core/taskengine/vm.go | 36 +++++++++++++++++ core/taskengine/vm_runner_customcode.go | 47 +++++++++++----------- core/taskengine/vm_runner_rest.go | 41 +++++++++++++++---- core/taskengine/vm_workflow_state_test.go | 49 +++++++++++++++++++++++ 5 files changed, 144 insertions(+), 31 deletions(-) diff --git a/core/taskengine/engine.go b/core/taskengine/engine.go index 10c3dc9a..501a3253 100644 --- a/core/taskengine/engine.go +++ b/core/taskengine/engine.go @@ -4612,6 +4612,8 @@ func (n *Engine) DeleteWorkflowByUser(user *model.User, taskID string) (*avsprot n.logger.Warn("failed to delete workflow state key", "key", k, "error", delErr) } } + } else { + n.logger.Warn("failed to list workflow state keys for cascade delete; state may be orphaned", "task_id", taskID, "error", listErr) } n.logger.Info("📢 Starting operator notifications", "task_id", taskID) diff --git a/core/taskengine/vm.go b/core/taskengine/vm.go index fbea81fb..b9bcac29 100644 --- a/core/taskengine/vm.go +++ b/core/taskengine/vm.go @@ -281,6 +281,42 @@ type VM struct { // the executor reads PendingSuspend() to checkpoint + register the wake instead // of finishing. nil for a normal run. suspend *SuspendRequest + + // stateScratch backs the {{state.*}} binding when it must NOT persist — in + // simulation, single-node runs (no task id), or when no DB is wired. It lives + // on the VM (not per-JSProcessor) so all customCode steps of one run share it, + // keeping read-after-write coherent across nodes just like the DB-backed path. + // Guarded by mu. Never flushed to storage. + stateScratch map[string][]byte +} + +// scratchGet/scratchSet/scratchList back the {{state.*}} binding's non-persistent +// mode (see stateScratch). All are mutex-guarded so parallel customCode steps are safe. +func (v *VM) scratchGet(key string) []byte { + v.mu.Lock() + defer v.mu.Unlock() + return v.stateScratch[key] +} + +func (v *VM) scratchSet(key string, value []byte) { + v.mu.Lock() + defer v.mu.Unlock() + if v.stateScratch == nil { + v.stateScratch = make(map[string][]byte) + } + v.stateScratch[key] = value +} + +func (v *VM) scratchList(prefix string) []string { + v.mu.Lock() + defer v.mu.Unlock() + out := make([]string, 0, len(v.stateScratch)) + for k := range v.stateScratch { + if strings.HasPrefix(k, prefix) { + out = append(out, k) + } + } + return out } func NewVM() *VM { diff --git a/core/taskengine/vm_runner_customcode.go b/core/taskengine/vm_runner_customcode.go index 28c191bf..6152ea40 100644 --- a/core/taskengine/vm_runner_customcode.go +++ b/core/taskengine/vm_runner_customcode.go @@ -6,6 +6,7 @@ import ( "math/big" "reflect" "regexp" + "sort" "strings" "time" @@ -30,15 +31,15 @@ import ( // within one run without mutating real state. Bind this AFTER the step vars so a node // accidentally named "state" can't shadow it. func installStateBinding(jsvm *goja.Runtime, vm *VM) { + if vm == nil { + return + } obj := jsvm.NewObject() - taskID := "" - if vm != nil { - taskID = vm.GetTaskId() - } - // scratch-only when we must not (or cannot) persist. - scratchOnly := vm == nil || vm.db == nil || taskID == "" || vm.IsSimulation - scratch := map[string][]byte{} + taskID := vm.GetTaskId() + // scratch-only when we must not (or cannot) persist. The scratch lives on the VM + // (vm.scratch*), so all customCode steps of one run share it. + scratchOnly := vm.db == nil || taskID == "" || vm.IsSimulation obj.Set("get", func(key string) interface{} { if key == "" { @@ -46,7 +47,7 @@ func installStateBinding(jsvm *goja.Runtime, vm *VM) { } var raw []byte if scratchOnly { - raw = scratch[key] + raw = vm.scratchGet(key) } else if b, err := vm.db.GetKey(WorkflowStateKey(taskID, key)); err == nil { raw = b } @@ -69,7 +70,7 @@ func installStateBinding(jsvm *goja.Runtime, vm *VM) { return } if scratchOnly { - scratch[key] = encoded + vm.scratchSet(key, encoded) return } if setErr := vm.db.Set(WorkflowStateKey(taskID, key), encoded); setErr != nil && vm.logger != nil { @@ -78,23 +79,23 @@ func installStateBinding(jsvm *goja.Runtime, vm *VM) { }) obj.Set("list", func(prefix string) []string { - out := []string{} + var out []string if scratchOnly { - for k := range scratch { - if strings.HasPrefix(k, prefix) { - out = append(out, k) - } + out = vm.scratchList(prefix) + } else { + full := string(WorkflowStatePrefix(taskID)) + keys, err := vm.db.ListKeys(full + prefix) + if err != nil { + return []string{} + } + out = make([]string, 0, len(keys)) + for _, k := range keys { + out = append(out, strings.TrimPrefix(k, full)) } - return out - } - full := string(WorkflowStatePrefix(taskID)) - keys, err := vm.db.ListKeys(full + prefix) - if err != nil { - return out - } - for _, k := range keys { - out = append(out, strings.TrimPrefix(k, full)) } + // Deterministic order regardless of backing store (Badger iterates sorted; + // the scratch map does not) so simulation and real runs agree. + sort.Strings(out) return out }) diff --git a/core/taskengine/vm_runner_rest.go b/core/taskengine/vm_runner_rest.go index a1c41b77..5a0ee09d 100644 --- a/core/taskengine/vm_runner_rest.go +++ b/core/taskengine/vm_runner_rest.go @@ -580,7 +580,7 @@ func (r *RestProcessor) Execute(stepID string, node *avsproto.RestAPINode) (*avs // credential never lives in the workflow JSON. GoPlus first; on "" (keys unset or // mint failed) we send no Authorization header — GoPlus still answers keyless. if restAuthProvider(node) == "goplus" { - if token := goplusAuthHeader(); token != "" { + if token := goplusTokenProvider(); token != "" { processedHeaders["Authorization"] = token if r.vm.logger != nil { r.vm.logger.Debug("REST: attached GoPlus session token (authed)", "stepID", stepID) @@ -909,11 +909,21 @@ func restAuthProvider(node *avsproto.RestAPINode) string { return strings.ToLower(strings.TrimSpace(provider)) } +// goplusTokenProvider is the seam the REST runner calls to obtain a GoPlus +// Authorization header value; overridable in tests. Returns "" for keyless. +var goplusTokenProvider = goplusAuthHeader + +// goplusHTTPClient bounds the token-mint call so a slow/hung GoPlus endpoint +// cannot block REST node execution indefinitely. +var goplusHTTPClient = &http.Client{Timeout: 15 * time.Second} + // goplusTokenCache holds the module-cached GoPlus session token, refreshed shortly -// before expiry. The token value already includes the "Bearer " prefix GoPlus attaches. +// before expiry. Keyed by the app_key it was minted for, so a rotated key never +// reuses a stale token. The token value already includes the "Bearer " prefix. var goplusTokenCache struct { sync.RWMutex token string + appKey string expires time.Time } @@ -930,7 +940,7 @@ func goplusAuthHeader() string { } goplusTokenCache.RLock() - if goplusTokenCache.token != "" && time.Until(goplusTokenCache.expires) > 60*time.Second { + if goplusTokenCache.token != "" && goplusTokenCache.appKey == appKey && time.Until(goplusTokenCache.expires) > 60*time.Second { token := goplusTokenCache.token goplusTokenCache.RUnlock() return token @@ -939,8 +949,8 @@ func goplusAuthHeader() string { goplusTokenCache.Lock() defer goplusTokenCache.Unlock() - // Re-check under the write lock (another goroutine may have refreshed). - if goplusTokenCache.token != "" && time.Until(goplusTokenCache.expires) > 60*time.Second { + // Re-check under the write lock (another goroutine may have refreshed for this key). + if goplusTokenCache.token != "" && goplusTokenCache.appKey == appKey && time.Until(goplusTokenCache.expires) > 60*time.Second { return goplusTokenCache.token } @@ -951,11 +961,19 @@ func goplusAuthHeader() string { "sign": hex.EncodeToString(sum[:]), "time": now, }) - resp, err := http.Post("https://api.gopluslabs.io/api/v1/token", "application/json", bytes.NewReader(reqBody)) + req, err := http.NewRequest(http.MethodPost, "https://api.gopluslabs.io/api/v1/token", bytes.NewReader(reqBody)) + if err != nil { + return "" + } + req.Header.Set("Content-Type", "application/json") + resp, err := goplusHTTPClient.Do(req) if err != nil { return "" } defer resp.Body.Close() + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return "" + } var parsed struct { Code int `json:"code"` Result struct { @@ -966,9 +984,16 @@ func goplusAuthHeader() string { if decodeErr := json.NewDecoder(resp.Body).Decode(&parsed); decodeErr != nil || parsed.Code != 1 || parsed.Result.AccessToken == "" { return "" } - goplusTokenCache.token = parsed.Result.AccessToken + // GoPlus returns the token already prefixed with "Bearer "; normalize defensively + // so a change on their side can't produce an invalid Authorization header. + token := parsed.Result.AccessToken + if !strings.HasPrefix(token, "Bearer ") { + token = "Bearer " + token + } + goplusTokenCache.token = token + goplusTokenCache.appKey = appKey goplusTokenCache.expires = time.Now().Add(time.Duration(parsed.Result.ExpiresIn) * time.Second) - return parsed.Result.AccessToken + return token } // escapeJSONString properly escapes a string for use within JSON diff --git a/core/taskengine/vm_workflow_state_test.go b/core/taskengine/vm_workflow_state_test.go index c0afdd92..0b9266e2 100644 --- a/core/taskengine/vm_workflow_state_test.go +++ b/core/taskengine/vm_workflow_state_test.go @@ -2,6 +2,8 @@ package taskengine import ( "encoding/json" + "net/http" + "net/http/httptest" "testing" "github.com/AvaProtocol/EigenLayer-AVS/core/testutil" @@ -143,3 +145,50 @@ func TestRestAuthProvider(t *testing.T) { require.Equal(t, "", restAuthProvider(&avsproto.RestAPINode{})) require.Equal(t, "", restAuthProvider(nil)) } + +// TestRestGoPlusAuthInjection verifies the Authorization header is attached when +// options.auth.provider="goplus" and a token is available, and left unset for the +// keyless fallback. The token source (goplusTokenProvider) is stubbed so the test +// never hits the real GoPlus endpoint. +func TestRestGoPlusAuthInjection(t *testing.T) { + var gotAuth string + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + gotAuth = r.Header.Get("Authorization") + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte(`{"code":1,"message":"ok","result":[]}`)) + })) + defer srv.Close() + + orig := goplusTokenProvider + defer func() { goplusTokenProvider = orig }() + + authNode := func() *avsproto.RestAPINode { + opts, err := structpb.NewValue(map[string]interface{}{"auth": map[string]interface{}{"provider": "goplus"}}) + require.NoError(t, err) + return &avsproto.RestAPINode{Config: &avsproto.RestAPINode_Config{Url: srv.URL, Method: "GET", Options: opts}} + } + run := func(node *avsproto.RestAPINode) { + gotAuth = "" + taskNode := &avsproto.TaskNode{Id: "r", Name: "r", TaskType: &avsproto.TaskNode_RestApi{RestApi: node}} + vm, err := NewVMWithData(&model.Workflow{Task: &avsproto.Task{ + Id: "auth-test", + Nodes: []*avsproto.TaskNode{taskNode}, + Edges: []*avsproto.TaskEdge{{Id: "e1", Source: "trigger", Target: "r"}}, + Trigger: &avsproto.TaskTrigger{Id: "trigger", Name: "trigger"}, + }}, nil, testutil.GetTestSmartWalletConfig(), nil) + require.NoError(t, err) + proc := NewRestProcessor(vm) + _, err = proc.Execute("r", node) + require.NoError(t, err) + } + + // A token from the provider is attached as the Authorization header. + goplusTokenProvider = func() string { return "Bearer test-token" } + run(authNode()) + require.Equal(t, "Bearer test-token", gotAuth) + + // Keyless fallback (provider returns "") sends no Authorization header. + goplusTokenProvider = func() string { return "" } + run(authNode()) + require.Equal(t, "", gotAuth) +}