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..501a3253 100644 --- a/core/taskengine/engine.go +++ b/core/taskengine/engine.go @@ -4603,6 +4603,19 @@ 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) + } + } + } 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) 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.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 abb2febc..6152ea40 100644 --- a/core/taskengine/vm_runner_customcode.go +++ b/core/taskengine/vm_runner_customcode.go @@ -1,10 +1,12 @@ package taskengine import ( + "encoding/json" "fmt" "math/big" "reflect" "regexp" + "sort" "strings" "time" @@ -16,6 +18,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) { + if vm == nil { + return + } + obj := jsvm.NewObject() + + 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 == "" { + return goja.Undefined() + } + var raw []byte + if scratchOnly { + raw = vm.scratchGet(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 { + vm.scratchSet(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 { + var out []string + if scratchOnly { + 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)) + } + } + // 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 + }) + + jsvm.Set("state", obj) +} + type JSProcessor struct { *CommonProcessor jsvm *goja.Runtime @@ -73,6 +159,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 +220,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..5a0ee09d 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,21 @@ 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 := goplusTokenProvider(); 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) + } + } + // Only apply JSON preprocessing if content type is JSON contentType := "" for key, value := range processedHeaders { @@ -862,6 +882,120 @@ 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)) +} + +// 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. 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 +} + +// 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 != "" && goplusTokenCache.appKey == appKey && 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 for this key). + if goplusTokenCache.token != "" && goplusTokenCache.appKey == appKey && 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, + }) + 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 { + 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 "" + } + // 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 token +} + // 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..0b9266e2 --- /dev/null +++ b/core/taskengine/vm_workflow_state_test.go @@ -0,0 +1,194 @@ +package taskengine + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "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)) +} + +// 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) +}